test: opt in to window-ledger DFG observations with separate receipts

A (per-base read-and-sum per 80-step interval), W (window + step(80) +
one read) and W1 (window + 80 x step(1)) over equal simulated duration
from the same fresh state and warm-up, 3 interleaved repeats. Bitwise
populations and per-interval total impulses (documented engineering
tolerance) are compared outside timing; health checks untimed. Records
candidate source identity (revision, dirty flag, file hashes) and does not
use or modify frozen refined-DFG manifests/results.

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

docs/development/refinement_mrt.md

@@ -0,0 +1,34 @@
+# MRT reuse in blocked/refined stepping (candidate)
+
+Base `9805383`. Opt-in by `LBMConfig(collision="MRT")`; SRT remains the default and its
+kernel statements, including the shifted retained-delta/direct-momentum path, are unchanged
+under the alternate compile branch.
+
+## What changes
+- `blocks.cu` (`BlockStep`), `step/block_channel.cu` (`BlockChannelStep`) and `refinement.cu`
+  (`patch_collide`) gain a compile-time `COLLISION_MODEL == 2` branch that repeats the uniform
+  `collide_dispatch` sequence with zero forcing: `compute_rho_u`, omega clamp to
+  `[OMEGA_COLLISION_MIN, OMEGA_COLLISION_MAX]`, `collide_mrt_no_force`.
+- Kernel and host guards accept SRT or MRT only. TRT, LES, non-FP32 storage, non-double-buffer
+  streaming, and shifted dense/mixed refinement remain rejected.
+- Transfers, subcycling, BFL, raw exchange ledgers and channel closures are untouched. The
+  existing pressure-NEQ outlet selects its MRT branch (unknowns-only NEQ extrapolation) from
+  `COLLISION_MODEL`, so MRT channel parity must use uniform MRT as reference.
+
+## Contracts (existing operator, not redefined here)
+- Shifted storage uses the existing MRT representation (equilibrium moments from `rho - 1`),
+  not the SRT retained-delta/direct-momentum precision. Mass and momentum are conserved
+  moments (relaxation rate 0).
+- Kinetic rates are the operator's fixed `s_e = s_eps = s_q = 1.2` on every level; the shear
+  rate is the existing per-level omega. Under acoustic 2:1 scaling the physical bulk viscosity
+  therefore halves on each finer level. No rate rescaling is applied.
+- No LES, no external forcing (`d_params` force fields stay zero).
+- Shifted Jets stay SRT-only: `body/jets.py` keeps its existing guard. This extension does not
+  provide shifted MRT Jets.
+- `PeriodicBlockStepper` (uniform blocked stepper) remains SRT-only.
+
+## Evidence status
+CPU: `tests/unit/test_refinement_mrt.py` (host guards, source contract). GPU:
+`tests/validation/multigrid/test_refinement_mrt_gpu.py` (uniform-MRT channel parity, AMR MRT
+smoke) and unchanged SRT GPU tests. Strict parity may be subject to cross-kernel FMA
+contraction differences already documented for SRT. Not a DFG/benchmark validation.

docs/validation/jet_circle_cv.md

@@ -113,47 +113,3 @@ These criteria are fixed before the continuation runs. There is no new physical
 Actual earlier scale-1 pilot cost was 52.3884s launch-to-exit, 46.1414s after READY for800 steps with full-field diagnostics every base step. A naive same-diagnostic3200-step scale-1 extrapolation is about210s; the new876 field samples reduce transfer count but add ray diagnostics and compression. Use conservative planning envelopes210–500s for scale1 and900–3200s for scale2, not timing guarantees. The scale2 conservative envelope accounts for4x cells and4x root steps; sparse sampling mitigates but does not eliminate that cost. Post-READY safety caps are1200s and5400s respectively, replacing the unsuitable fixed1800s assumption. Root owns compilation/startup timeout and serial launch decisions.
 
 Device-envelope estimates remain128/512MiB for s1/s2, at least1GiB free (or twice estimate). Output estimate is under64MiB per run including five primitive snapshots and876 sample records; no all-time populations are saved. Plan/metadata and sampled files are rewritten at each100-reference-time window to retain bounded partial evidence if interrupted. READY still follows compiled warmup and state/time reset, before measured stepping begins.
-
-## Dense-root temporal diagnostic: preparation only
-
-The audited continuation did not resolve time quadrature: all48 audited stride2 RMS changes exceeded their stride1 CV residuals, by factors1.55–11.46. That historical failure and its gates remain unchanged. The following is a measurement-only experiment to test this explanation, not a solver fix, a physics pass, or permission to extend GPU work. The separate periodic global stock-minus-raw mass residual also persists; denser contour quadrature cannot by itself resolve it.
-
-Future launch, only after the computer is online and root approves a serialized bounded run:
-
-```sh
-CUDA_VISIBLE_DEVICES="$VERIFIED_IDLE_GPU" python tests/validation/run_jet_circle_cv.py --repo "$PWD" --output "$NEW_DENSE_S1" --dense-root --steps 800 --scale 1
-CUDA_VISIBLE_DEVICES="$VERIFIED_IDLE_GPU" python tests/validation/run_jet_circle_cv.py --repo "$PWD" --output "$NEW_DENSE_S2" --dense-root --steps 800 --scale 2
-python tests/validation/analyze_jet_cv_continuation.py "$NEW_DENSE_S1" "$NEW_DENSE_S2" --output "$DENSE_COMPARISON_JSON"
-```
-
-This mode fixes Re100,beta.1 and reference t=0..800 (t*≈.9965), with800/3200 root intervals and801/3201 endpoint records. It is mutually exclusive with --continuation and rejects longer runs or other scales/Re. Physical geometry, source manifest, solver, velocity commands, kernel and raw observation path are unchanged. There are no new finest-hop hooks. Existing GPU population downloads occur every root interval for CPU primitive diagnostics; those distributions are discarded rather than retained.
-
-Dense scalar records contain CV endpoint inventories, each flux and split momentum terms, density/speed, physical composite mass, exact cumulative raw partitions and cumulative absolute aperture net mass. Rays remain on the exact old reference cadence:0..100 at dt1, then104..800 at dt4,276 samples for each scale. Schema2 maps their smaller arrays through ray_sample_indices into the dense records. Full primitive snapshots occur only at reference0,100,800. Saved data is capped at32MiB per case; dense arrays are scalar diagnostics, not all-time full fields. Schema/unit metadata, file hashes, complete-schedule checks and matched root time intervals remain mandatory.
-
-The analyzer derives root strides1,2,4 and the exact old cadence from the SAME dense primitive records, always keeping window endpoints and the SAME raw ledger. At scale2 these root strides correspond to reference dt.25,.5,1; the old cadence corresponds to root gaps4/16. Its fourth quadrature result and change from dense are separately identified. Existing residual normalizations and historical files are preserved. No automatic refinement-progress flags are issued for this temporal-only mode. Root-sampled agreement cannot certify finer-than-root time convergence or exclude aliasing; an unresolved outcome stays unresolved, with no automatic longer run.
-
-Three distinct throughput quantities must remain separate:
-
-- Prescribed unsigned nominal mass at rho_ref=1:2*Q_command*window_duration.
-- Observed raw absolute NET aperture mass: sum over root intervals of abs(dm_blowing)+abs(dm_suction), after spatial reduction within each aperture. This retains reversals between root intervals that abs(window-net) loses. Within-aperture or within-root/substep reversals remain unresolved by this ledger. The legacy raw_gross_mass_over_nominal field retains its old abs(window-net)/ONE-aperture-nominal definition; it is not this new quantity.
-- Sparse reconstructed absolute Q/M: true-arc quadrature of abs(u.n) and abs(rho*u.n), using each existing linear/quadratic ray estimate, followed by old-cadence time integration. This is estimator-dependent measured-field evidence, not exact wall truth, not dense-root quadrature, and not a prescribed nominal flow.
-
-Global stock-minus-raw mass balance is reported per root interval and summed per window, in reference mass units. Raw total-minus-wall/aperture partition discrepancies are also separate. Neither is bundled into the CV temporal error, and neither is called a proven raw-extractor defect. Source/geometry/state validity, both signed and gross budgets, old failures, and physics_validation_pass=false remain intact.
-
-Planning envelope from the earlier52-second800-step pilot: s1 60–240s, s2 600–2400s, with conservative post-READY caps1200/5400s. Scale2 has4x cells and4x root observations over the same reference horizon; sparse saving reduces output, not all CPU/PCIe observation cost. These are preparation estimates, not measured timings. Root owns startup/compilation timeout and GPU serialization. No GPU/remote action was taken to prepare this option.
-
-Budget enforcement preflights the projected retained file sizes before replacing each snapshot/window report or the sample+metadata pair. If32MiB would be exceeded, staged replacements are removed, the previous verified sample/hash-metadata pair remains intact, and execution stops. Temporary compression files need additional short-lived disk space and are distinct from the32MiB retained-evidence cap. This protects budget failures; it is not a filesystem crash-recovery transaction. Both raw partition identities, total−wall−aperture and aperture−blowing−suction, are saved separately from stock/raw and CV residuals.
-
-## Optional shifted-FP32 representation control
-
-The dense-root campaign accepts `--shifted-fp32` as an explicit numerical-representation control. Without it, the historical unshifted FP32 default remains unchanged. Use a fresh output directory and append the flag to the reviewed dense command, for example:
-
-```bash
-CUDA_VISIBLE_DEVICES="$VERIFIED_IDLE_GPU" python tests/validation/run_jet_circle_cv.py --repo "$PWD" --output "$NEW_SHIFTED_DENSE_S1" --dense-root --steps 800 --scale 1 --shifted-fp32
-```
-
-Root owns the future GPU launch and scale selection. This option requires the bounded dense campaign; it does not authorize another run. Geometry, Jet profile/amplitude, initial physical seed, viscosity, time horizon, root sampling, sparse rays/fields, normalization, thresholds, budgets and prior failed gates are unchanged. The `case.ddf_shifting` flag and `case.storage_variant` receipt appear in the plan, READY, window report and sample metadata. Preserve the unshifted evidence in its original directory. Compare like-scale records at identical root endpoints, including CV mass/momentum, global stock-minus-raw balance, raw partitions and measured aperture flux. A reduced mass residual would support representation sensitivity; it would not establish physical convergence or prove that every residual is roundoff.
-
-Both production blocked steppers already support this representation: initialization takes physical populations and encodes `g=f-w` to FP32; public download returns physical `f` decoded in FP64. The runner deliberately retains its historical FP32-rounded physical equilibrium seed for BOTH variants rather than also changing initial precision. Encoding introduces a small additional rounding difference: maximum per-population change 1.035e-10 and density change 3.201e-10 at both scales for this seed. Thus the decoded starting states are not bit-identical. The residual initial density offset is therefore not removed. All existing measurement routines receive physical populations and apply no extra weight shift. Snapshots keep the existing primitive-field precision.
-
-The existing `BlockCurvedBoundary` optional-input adapter invokes the same shifted-aware BFL helper as the AMR boundary: wall density restores the unit background, raw mass uses reflected-minus-outgoing (opposite weights cancel), and reaction restores twice the direction weight. No host correction is applied to raw rows. The shifted branch selects `compute_feq_from_momentum` rather than unshifted `compute_feq`, so this control changes existing numerical algebra as well as storage representation. It cannot isolate storage precision alone or identify collision as the sole cause. Block collision preserves the density perturbation in its shifted branch; blocked halo copies and AMR transfers already support shifted storage. This option changes no solver/kernel/API or source manifest. Existing shifted-storage/Jet host tests and the new runner codec tests are CPU evidence only; numerical GPU behavior remains to be checked by the reviewed control.

docs/validation/refined_dfg_mrt.md

@@ -0,0 +1,71 @@
+# Matched MRT DFG pilot preparation
+
+Preparation only: this test-only extension reuses the existing runner and existing MRT
+operator. GPU admission awaits reviewed MRT capability results; no pilot/full execution
+is implied by CPU preflight. No production, historical SRT manifest, reference data,
+periodic gate, geometry, initialization or observation algorithm is changed here.
+
+## Selection and provenance
+
+Use `--collision MRT` for **both** uniform D80 and refined root D40/fine D80. There are
+no independently selectable collision models per case. Default remains SRT.
+
+MRT verifies `data/refined_dfg/source_manifest_mrt.json`, pinned to reviewed numerical
+source snapshot `3ea131e92c057843034f11a1c574301a2430dfce`, including the shared geometry,
+pressure/cycle helpers and official reference. Test-only commits may change HEAD without
+changing this numerical source pin. Plan, READY, case summaries and comparison separately
+record `checkout_revision` and `checkout_dirty` using existing source-provenance helper;
+the runner hash remains separate. Collision is recorded in every receipt, and full-mode
+validation rejects missing or different collision labels in plan, either summary or either
+READY receipt, in addition to existing source/runner/health/time checks.
+
+The historical SRT manifest remains byte-for-byte unchanged. SRT on the MRT candidate
+checkout deliberately fails that source pin; use a matching historical checkout via
+`--source` for SRT. No fallback or automatic repinning is permitted. Historical SRT
+failure/diagnostic evidence remains attached to its original source and runner hashes;
+it cannot authorize an MRT full run. The existing failed-observation step-reporting
+undercount is not changed in this narrowly scoped selection patch.
+
+## Frozen physical comparison
+
+All settings remain those documented in `refined_dfg.md`: U_lattice=.01, physical mean
+U=1 m/s, D=.1 m, shear ν=.001 m²/s, fixed physical domain/body/wall planes, patch
+(40,24,320,140), shifted FP32 double buffers, no LES/forcing/Jets, finest dx=.00125 m,
+finest dt=.0000125 s, unchanged half-cell frame conversion and axis-specific phase.
+Force interval accumulation, physical leaf mass, pressure interpolation, flux sampling,
+safety thresholds, t*=5 pilot and t*=300 cap are unchanged. Existing accuracy/periodicity
+criteria are imported unchanged. Both MRT cases are fresh runs; old MRT or SRT results
+are not substituted as controls.
+
+Both cases use the **existing MRT pressure-NEQ branch**: only unknown directions q2/6/8
+are extrapolated. This differs from SRT's all-nine .5-damped branch; the retained
+`srt_neq_damp=.5` configuration value is inactive under MRT. Backflow clamp remains off.
+The outlet is still noncanonical relative to exact DFG do-nothing conditions.
+
+The reused MRT operator keeps fixed kinetic rates `s_e=s_eps=s_q=1.2` on all levels;
+only the shear rate follows existing per-level viscosity. Physical bulk relaxation
+therefore differs between coarse and fine levels. Transfer operators and subcycling
+are unchanged. This is the candidate's declared behavior, not a new fully mode-scaled
+MRT transfer scheme. Do not tune rates, the patch or gates after seeing outcomes.
+Capability smoke/parity gates do not establish DFG accuracy or speedup.
+
+## CPU and admission commands
+
+From this checkout:
+
+```sh
+python tests/unit/test_refined_dfg.py
+python tests/validation/run_refined_dfg.py --collision MRT --mode plan --output /tmp/new-mrt-dfg-plan
+```
+
+After root verifies the candidate MRT root/parity/refinement capability results and admits
+an idle device, the paired pilot may use the existing bounds and external watchdog:
+
+```sh
+CUDA_VISIBLE_DEVICES=<admitted-device> timeout --signal=TERM --kill-after=30s <hard-budget>s python tests/validation/run_refined_dfg.py --collision MRT --mode pilot --gpu-gates-approved --max-wall-seconds <soft-budget> --output <new-mrt-pilot-directory>
+```
+
+The flag is caller attestation, not evidence. An unhealthy first case prevents the second.
+Existing strict pilot receipt validation and calibrated wall-time controls remain; full
+mode is not being admitted as part of this preparation. Root must review both healthy
+pilot outcomes and separately decide any full-run admission.

docs/validation/refined_dfg_timing.md

@@ -0,0 +1,58 @@
+# Refined DFG MRT timing harness (measurement only)
+
+`tests/validation/time_refined_dfg.py` times the matched MRT refined/uniform DFG Re100 setup of
+`run_refined_dfg.py` (same configuration, geometry and initial state functions). No numerical,
+physics-gate or production change; no force or accuracy result is produced.
+
+| Variant | Loop | Force reads |
+|---|---|---|
+| A | refined `step(1)` + `read_raw_exchange` per base step | runner-like force-read workload (values discarded) |
+| B | refined `step(1)` only, no host reads (`observe_raw_exchange` stays on) | timing-only |
+| C | refined single `step(N)` | timing-only; ledger keeps the final base interval only |
+| E | uniform `sim.run(<=100)` + `read_force` | runner-like force-read workload (values discarded) |
+| E' | uniform single `sim.run(2N)` | timing-only; at most a whole-window average, different summation order |
+
+- No variant retains a valid force history; A/E omit the runner's interval aggregation and
+  validation, and no variant includes macro/probe/CSV costs. Results are not full benchmark
+  throughput.
+- Status lines: `CONSTRUCTED` (construction/JIT complete) and `WARM_READY` (first warm-up and
+  health check passed) per case.
+- Equal simulated duration: N refined base steps vs 2N uniform D80 steps.
+- Every repeat starts from the same fresh state (refined `initialize`, uniform `snapshot/restore`)
+  and the same warm-up; construction/JIT, warm-up, downloads and parity checks are untimed.
+- The actual stepping stream is synchronized explicitly before and after each timed loop; the
+  wall timer (`time.perf_counter`) is the only source. No CUDA-event span is called kernel time.
+- `--max-wall-seconds` is required for GPU runs (outer timeout, checked between variants and
+  every 100 steps in per-step loops).
+- Final populations and NaN masks must be bitwise identical to the first A repeat (A/B/C) or the
+  first E repeat (E/E'); refined and uniform grids are never compared. Leaf/fluid populations
+  must be finite. Physical health (finite leaf/fluid macros, rho in [0.5, 1.5], max Mach <= 0.2,
+  the existing runner limits) is checked after warm-up and after each timed window. Matching NaN
+  is accepted only as a storage mask; any infinity or level-count mismatch fails parity. All
+  parity/health checks are untimed.
+- Cases run sequentially (uniform, then refined) in one process with startup serialized;
+  variants are interleaved within each repeat. A-B and B-C are measured comparative
+  differences under GPU variability, not exact causal decompositions.
+
+Usage (root selects one verified idle GPU):
+
+    PYTHONPATH=src python -m unittest tests/unit/test_time_refined_dfg.py
+    python tests/validation/time_refined_dfg.py --output <new> --plan
+    CUDA_VISIBLE_DEVICES=<gpu> python tests/validation/time_refined_dfg.py --output <new> --max-wall-seconds <s>
+
+Defaults: 2000 base steps, 200 warm-up base steps, 3 repeats. Profiler capture and the
+`observe_raw_exchange=False` variant are deferred until these measurements exist.
+
+## A versus observation-window ledger (`tests/validation/time_window_dfg.py`)
+
+Measurement-only comparison on the same shifted-MRT refined setup: A (per-base `step(1)` +
+`read_raw_exchange`, summed per 80-base-step interval), W (`start_exchange_window`, `step(80)`,
+one window read) and W1 (window with 80 x `step(1)`, isolating the read pattern from call
+batching). Same fresh state, warm-up and simulated duration; populations must be bitwise equal
+to the first A repeat and per-interval total impulses within the documented engineering
+tolerance, all outside timing. The candidate source is recorded (checkout revision, dirty flag,
+file hashes); frozen refined-DFG manifests/results are not used or changed.
+
+    PYTHONPATH=src python -m unittest tests/unit/test_time_window_dfg.py
+    python tests/validation/time_window_dfg.py --output <new> --plan
+    CUDA_VISIBLE_DEVICES=<gpu> python tests/validation/time_window_dfg.py --output <new> --max-wall-seconds <s>

docs/validation/refined_dfg_window.md

@@ -0,0 +1,58 @@
+# Opt-in observation-window ledger in the DFG runner
+
+This test-only integration adds `--exchange-window` to `run_refined_dfg.py`. It uses the
+existing production ledger; no collision, boundary, transfer, physical setup, reference,
+periodicity gate or acceptance threshold changes. Default remains the historical
+per-base-step observation path. Uniform stepping/force accumulation is unchanged.
+
+For the refined case, each existing observation interval opens a new ledger at completed
+base step `begin`, advances exactly the same `n` base intervals, and reads the integrated
+root-lattice impulse for `[begin, begin+n]`. The contract requires root units, `window`
+kind, `2*n` fine samples, matching endpoints, and fine dt=.5. The existing coefficient
+calculation still divides this impulse by `n` and `.5*U_lattice**2*D_root`; force timestamps
+remain `(begin+n/2)*U_lattice/D_root`. It does not divide an already averaged force again.
+Pressure/macro/physical-leaf mass measurements and safety checks remain at the original
+interval end. Pilot interval is 80 root steps; early full-run intervals remain 2000, and
+any final shortened interval uses its actual `n`.
+
+The opt-in uses one `step(n)` call and one window read per observation, rather than `n`
+Python calls/readbacks. Internal subcycling and per-base synchronization remain owned by
+the same backend. Cooperative deadline checks surround that complete interval; unlike
+the default path they cannot interrupt between base steps inside `step(n)`. The existing
+external hard watchdog is still required. The full-run cadence/cap is documented here,
+but this preparation does not admit a full run or a GPU operation.
+
+## Provenance and receipt separation
+
+`source_manifest_window.json` pins the reviewed a233c9032487a6006e950d1e327079b00466c410
+snapshot (production ledger origin d930) plus existing helper/reference files. Both SRT
+and MRT are supported by this explicit opt-in source pin; collision selection remains
+independent and shared by the two cases. Existing SRT and MRT manifests are untouched.
+Default mode continues to reject the changed candidate under those historical pins;
+use a matching historical `--source` checkout to run its default path. There is no
+silent repinning, source fallback, or reuse of historical outputs as candidate evidence.
+
+Plan, READY, summaries, timeout/comparison receipts now record `observation_mode` as
+`last_base_step` or `window`, alongside collision and actual checkout provenance. On a
+uniform receipt this labels its **paired campaign's refined observation mode**, not a
+change to uniform observation. Full-mode pilot validation rejects missing or mismatched
+mode labels in the plan, either case summary, and either READY receipt. Source/runner
+hash, collision, completeness and health checks remain. Cross-mode pilot reuse is denied.
+
+The earlier short A/W/W1 timing/capability evidence is not a full-run speedup or accuracy
+claim. Mathematical impulse integration/units are unchanged; floating-point reduction
+order can differ, so bitwise equality of force sums is not promised. Physics acceptance
+continues to use the existing frozen gates. Historical failures/results are untouched.
+
+CPU preparation commands, from the repository root:
+
+```sh
+python tests/unit/test_refined_dfg.py
+PYTHONPATH=src python -m unittest tests.unit.test_raw_exchange_window tests.unit.test_time_window_dfg -v
+python tests/validation/run_refined_dfg.py --collision MRT --exchange-window --mode plan --output /tmp/new-window-dfg-plan
+```
+
+Root must independently review and admit any execution. An admitted paired pilot uses
+the existing watchdog, device selection, `--gpu-gates-approved`, positive
+`--max-wall-seconds` and a new output directory, adding only `--exchange-window` to the
+selected collision's pilot command. No GPU or remote action was performed for this patch.

src/CelerisLab/lbm/channel.py

@@ -1,4 +1,4 @@
-"""Host contract for the blocked root's uniform-SRT channel boundary.
+"""Host contract for the blocked root's uniform SRT/MRT channel boundary.
 
 Boundary ghosts are evolved state, but are not physical fluid leaves. Halo
 ownership remains periodic to reproduce the uniform kernel's wrapped pulls.
@@ -29,7 +29,7 @@ def validate_channel_config(cfg, blocks):
 def validate_channel_transfers(pair, layout, targets):
     """Reject actual exchange/closure overlap, not a padded geometric envelope.
 
-    Explosion donors and restriction receivers must use ordinary bulk SRT
+    Explosion donors and restriction receivers must use ordinary bulk collision
     semantics. Restriction writes receiver-local source slots; these may not
     overwrite the channel's extra immutable-input reads (pressure donor and
     same-node halfway-wall corrections). The slot check deliberately respects

src/CelerisLab/lbm/kernels/blocks.cu

@@ -14,12 +14,15 @@
 #include "core/obs.cuh"
 #include "operators/macro.cuh"
 #include "operators/collision_srt.cuh"
+#if COLLISION_MODEL == 2
+#include "operators/collision_mrt.cuh"
+#endif
 #include "streaming/pull_double_buffer.cuh"
 #include "boundary/curved_boundary.cuh"
 #include "boundary/planar_velocity.cuh"
 
-#if DIM != 2 || NQ != 9 || COLLISION_MODEL != 0 || STORE_PRECISION != 0 || USE_LES || STREAMING_MODEL != 0
-#error "Blocks require FP32 D2Q9 SRT double buffers without LES"
+#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"
 #endif
 
 // Fixed-patch cell kinds, matching refinement.py ACTIVE/INNER_GHOST/OUTER_GHOST.
@@ -141,6 +144,13 @@ extern "C" __global__ void BlockStep(const fpxx* in, fpxx* out, unsigned int hal
                         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
@@ -153,6 +163,7 @@ extern "C" __global__ void BlockStep(const fpxx* in, fpxx* out, unsigned int hal
             compute_feq(rho, ux, uy, feq);
 #endif
             collide_srt_no_force(f, feq, d_params.omega);
+#endif
         }
     }
     stream_pull_store(k, f, out + base);

src/CelerisLab/lbm/kernels/refinement.cu

@@ -11,6 +11,9 @@
 #include "core/obs.cuh"
 #include "operators/macro.cuh"
 #include "operators/collision_srt.cuh"
+#if COLLISION_MODEL == 2
+#include "operators/collision_mrt.cuh"
+#endif
 #include "streaming/pull_double_buffer.cuh"
 #include "boundary/planar_velocity.cuh"
 #include "boundary/curved_boundary.cuh"
@@ -19,8 +22,8 @@ extern "C" {
 #include "step/curved_double.cu"
 }
 
-#if DIM != 2 || NQ != 9 || COLLISION_MODEL != 0 || STORE_PRECISION != 0 || USE_DDF_SHIFTING || USE_LES || STREAMING_MODEL != 0
-#error "Refinement requires unshifted FP32 D2Q9 SRT double buffers without LES"
+#if DIM != 2 || NQ != 9 || (COLLISION_MODEL != 0 && COLLISION_MODEL != 2) || STORE_PRECISION != 0 || USE_DDF_SHIFTING || USE_LES || STREAMING_MODEL != 0
+#error "Refinement requires unshifted FP32 D2Q9 SRT or MRT double buffers without LES"
 #endif
 
 // Enabled only in the validation build; no diagnostic arrays in normal builds.
@@ -33,10 +36,17 @@ __device__ unsigned int incoming_counts[NQ * TOTAL_CELLS];
 #endif
 
 __device__ __forceinline__ void patch_collide(float* f, unsigned long k) {
+#if COLLISION_MODEL == 2
+    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);
     compute_feq(rho, ux, uy, feq);
     collide_srt_no_force(f, feq, d_params.omega);
+#endif
 #ifdef CELERIS_REFINEMENT_AUDIT
     ++collision_counts[k];
 #endif

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

@@ -54,12 +54,20 @@ extern "C" __global__ void BlockChannelStep(
             f[7] = load_ddf(in + base, index_f(k, 8));
         }
         if (!wall && !outlet) {
+#if COLLISION_MODEL == 2
+            // 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] = {};
             compute_rho_u(f, rho, ux, uy);
             compute_feq(rho, ux, uy, feq);
             const float omega = fminf(OMEGA_COLLISION_MAX, fmaxf(OMEGA_COLLISION_MIN, d_params.omega));
             collide_srt(f, feq, forcing, omega);
+#endif
         }
     }
     stream_pull_store(k, f, out + base);

src/CelerisLab/lbm/raw_exchange.py

@@ -7,16 +7,26 @@ import numpy as np
 
 
 def summarize_exchange(rows, body_ids, aperture, *, body_id, level, h,
-                       time_end, window, reference_center):
+                       time_end, window, reference_center, duration=None):
     """Reduce FP64 interval integrals or convert FP32 last-substep raw rows.
 
     dt_l=h_l under acoustic 2:1 refinement. Local momentum j -> base force
     h*j -> impulse h*dt*j; torque adds one h; mass is h**2*dm.
+    window="window" reduces an FP64 observation-window ledger spanning
+    ``duration`` (integer >= 1) base intervals; rates divide by that duration.
     """
-    if window not in ("last_base_step", "last_substep"):
+    if window not in ("last_base_step", "last_substep", "window"):
         raise ValueError("unknown exchange window")
+    if window == "window":
+        if isinstance(duration, bool) or not isinstance(duration, (int, np.integer)) or duration < 1:
+            raise ValueError("window duration must be an integer number of base intervals >= 1")
+    elif duration is not None:
+        raise ValueError("duration applies only to window='window'")
     values = np.array(rows, dtype=np.float64, copy=True)
-    duration = 1.0 if window == "last_base_step" else h
+    if window == "window":
+        duration = float(duration)
+    else:
+        duration = 1.0 if window == "last_base_step" else h
     if window == "last_substep":
         values *= np.array([h*h, h*h, h*h*h, h*h])
     selected = np.asarray(body_ids) == body_id
@@ -31,6 +41,7 @@ def summarize_exchange(rows, body_ids, aperture, *, body_id, level, h,
     return dict(wall=reduce(selected & ~aperture), aperture=reduce(selected & aperture),
                 total=reduce(selected), body_id=int(body_id), level=int(level),
                 time_start=time_end-duration, time_end=time_end, dt=h,
-                sample_count=round(1/h) if window == "last_base_step" else 1,
+                sample_count=(round(duration/h) if window == "window"
+                              else round(1/h) if window == "last_base_step" else 1),
                 window=window, units="base_lattice_per_unit_depth",
                 definition="raw_boundary_exchange", reference_center=tuple(reference_center), valid=True)

src/CelerisLab/lbm/refinement.py

@@ -52,17 +52,17 @@ class FixedPatch2D:
     def from_config(cls, cfg: LBMConfig, bounds) -> "FixedPatch2D":
         """Prepare maps without changing cfg or allocating coarse-sized tables.
 
-        The initial backend contract is D2Q9/SRT/FP32/double_buffer,
+        The backend contract is D2Q9/SRT or MRT/FP32/double_buffer,
         without LES. Two coarse cells of margin keep all halo donors in bounds;
         this does NOT certify that those donors are fluid or periodic.
         """
         checked = replace(cfg)
         checked.validate()
-        if (not checked.is_d2q9 or checked.collision != "SRT"
+        if (not checked.is_d2q9 or checked.collision not in ("SRT", "MRT")
                 or checked.streaming != "double_buffer"
                 or checked.store_precision != "FP32"
                 or checked.les_enabled):
-            raise ValueError("FixedPatch2D requires D2Q9/SRT/FP32/double_buffer without LES")
+            raise ValueError("FixedPatch2D requires D2Q9/SRT or MRT/FP32/double_buffer without LES")
         bounds = tuple(bounds)
         if len(bounds) != 4 or any(isinstance(v, (bool, np.bool_))
                                    or not isinstance(v, Integral) for v in bounds):
@@ -451,7 +451,7 @@ def prepare_fine_boundary(cfg, patch, objects, *, wall_aware=False, jets=(),
     checked = replace(cfg)
     checked.validate()
     if (patch.coarse_shape != (cfg.ny, cfg.nx) or not checked.is_d2q9
-            or checked.collision != "SRT" or checked.streaming != "double_buffer"
+            or checked.collision not in ("SRT", "MRT") or checked.streaming != "double_buffer"
             or checked.store_precision != "FP32" or checked.les_enabled
             or patch.tau_coarse != 3*checked.viscosity+.5):
         raise ValueError("Fine boundary requires the matching fixed-patch configuration")

src/CelerisLab/lbm/refinement_stepper.py

@@ -172,6 +172,9 @@ class FixedPatchStepper:
     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 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.
@@ -216,7 +219,7 @@ class FixedPatchStepper:
                  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_raw_exchange=False, observe_window_exchange=False):
         if HALO != 4:
             raise RuntimeError("refinement.cu requires a four-cell fine halo")
         if coarse_boundary not in ("periodic", "slot_reservoir", "channel"):
@@ -315,6 +318,13 @@ class FixedPatchStepper:
                                       for b, layout in zip(self._boundary_hosts, self._layouts)):
             raise ValueError("raw exchange currently requires blocked body levels")
         self._observe_raw_exchange = observe_raw_exchange
+        if not isinstance(observe_window_exchange, bool):
+            raise ValueError("observe_window_exchange must be bool")
+        if observe_window_exchange and not observe_raw_exchange:
+            raise ValueError("observe_window_exchange requires observe_raw_exchange")
+        self._observe_window = observe_window_exchange
+        self._window_valid = False
+        self._window_start = None
         self._exchange_valid = False
         self._exchange_spacing = tuple(spacings)
         self._exchange_centers = tuple(tuple(tuple(o.geometry.center) for o in objects)
@@ -365,6 +375,8 @@ class FixedPatchStepper:
                             count = len(b.tables["cell"])
                             b.last_raw = self._allocate(count * 4 * 4)
                             b.interval_integral = self._allocate(count * 4 * 8)
+                            if self._observe_window:
+                                b.window_integral = self._allocate(count * 4 * 8)
                             b.accumulate = level.module.get_function("BlockAccumulateExchange")
                         b.kernel = level.module.get_function("BlockCurvedBoundary")
                 if self._level_boundaries[1] is not None:
@@ -559,6 +571,7 @@ class FixedPatchStepper:
                     cuda.memset_d8(self._fine_curved.link_exchange_gpu, 0, self._fine_curved.count * 12)
                 self._step_count = 0
                 self._exchange_valid = False
+                self._window_valid = False
                 self._status = "ready"
             except BaseException:
                 self._status = "failed"
@@ -639,7 +652,8 @@ class FixedPatchStepper:
                 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_raw_exchange=self._observe_raw_exchange,
+                observe_window_exchange=self._observe_window)
             try:
                 if self._ddf_shifting:
                     candidate._initialize_raw(_StoredPopulations(transferred.g, transferred.shifted))
@@ -771,6 +785,10 @@ class FixedPatchStepper:
             b.accumulate(b.last_raw, b.interval_integral, np.uint32(count),
                          np.float64(h), np.float64(h),
                          grid=((count + self._block[0] - 1) // self._block[0], 1), **kw)
+            if self._window_valid and b.window_integral is not None:
+                b.accumulate(b.last_raw, b.window_integral, np.uint32(count),
+                             np.float64(h), np.float64(h),
+                             grid=((count + self._block[0] - 1) // self._block[0], 1), **kw)
         flags = np.uintp(0) if level.boundary is None else level.boundary.flags
         level.kernel(src, dst, np.uint32(level.layout.halo), flags, level.kind, grid=level.grid, **kw)
 
@@ -804,9 +822,57 @@ class FixedPatchStepper:
                     self._exchange_valid = self._observe_raw_exchange
                 self._status = "ready"
             except BaseException:
+                self._window_valid = False
                 self._status = "failed"
                 raise
 
+    def start_exchange_window(self):
+        """Zero the window ledgers and open a window at the current completed step count."""
+        with self._lock, self.ctx.activate():
+            self._require("ready")
+            if not self._observe_window:
+                raise RuntimeError("window exchange observation is disabled")
+            self._window_valid = False
+            try:
+                for b in self._level_boundaries:
+                    if b is not None and b.window_integral is not None:
+                        cuda.memset_d8(b.window_integral, 0, len(b.tables["cell"]) * 32)
+            except BaseException:
+                self._status = "failed"
+                raise
+            self._window_start = self._step_count
+            self._window_valid = True
+
+    def read_exchange_window(self, body_id, *, level=1):
+        """Non-destructive read of [window start, current step count] in base-lattice units."""
+        with self._lock, self.ctx.activate():
+            self._require("ready")
+            if not self._observe_window:
+                raise RuntimeError("window exchange observation is disabled")
+            if not self._window_valid:
+                raise RuntimeError("no valid exchange window; call start_exchange_window()")
+            duration = self._step_count - self._window_start
+            if duration < 1:
+                raise RuntimeError("exchange window requires at least one completed base interval")
+            if (isinstance(level, bool) or not isinstance(level, Integral)
+                    or not 0 <= level < len(self._levels)):
+                raise ValueError("level must name an existing level")
+            b = self._level_boundaries[level]
+            if (b is None or isinstance(body_id, bool) or not isinstance(body_id, Integral)
+                    or not 0 <= body_id < len(b.host.objects)):
+                raise ValueError("body_id must name a body on this level")
+            rows = np.empty((len(b.tables["cell"]), 4), np.float64)
+            try:
+                cuda.memcpy_dtoh(rows, b.window_integral)
+            except BaseException:
+                self._status = "failed"
+                raise
+            return summarize_exchange(rows, b.tables["body_id"], b.tables["aperture"],
+                                      body_id=body_id, level=level, h=self._exchange_spacing[level],
+                                      time_end=self._step_count, window="window",
+                                      reference_center=self._exchange_centers[level][body_id],
+                                      duration=duration)
+
     def read_raw_exchange(self, body_id, *, level=1, window="last_base_step"):
         """Read raw boundary reaction about its geometric center in the root edge frame.
 
@@ -885,6 +951,7 @@ class FixedPatchStepper:
                     if b is not None:
                         b.flags = b.device = b.kernel = None
                         b.last_raw = b.interval_integral = b.accumulate = None
+                        b.window_integral = None
                 self._levels.clear()
                 self._pairs = []
                 if getattr(self, "_fine_curved", None) is not None:

src/CelerisLab/lbm/refinement_tables.py

@@ -85,6 +85,7 @@ class _BlockedBoundary:
     last_raw: object = None
     interval_integral: object = None
     accumulate: object = None
+    window_integral: object = None  # optional observation-window FP64 ledger
 
 
 def _prepare_blocked_boundary(boundary, layout, cfg, jets, *, _shifted_blocked=False):

tests/unit/test_jet_circle_cv.py

@@ -28,41 +28,6 @@ def eq(rho, u, v):
 
 
 class Host(unittest.TestCase):
-    def test_shifted_storage_receipt_preserves_physical_case(self):
-        for scale in (1, 2):
-            base = case(scale)
-            shifted = case(scale, ddf_shifting=True)
-            self.assertFalse(base.pop("ddf_shifting"))
-            self.assertTrue(shifted.pop("ddf_shifting"))
-            self.assertEqual(base.pop("storage_variant")["name"], "unshifted_fp32")
-            self.assertEqual(shifted.pop("storage_variant")["name"], "shifted_fp32")
-            self.assertEqual(base, shifted)
-
-    def test_shifted_physical_download_roundtrip(self):
-        from CelerisLab.lbm.population_storage import _encode_physical, _decode_stored
-        from run_jet_circle_cv import moments
-        from jet_cv_measurements import primitives
-
-        for speed in (0.03, 0.015):
-            seed = state((9, 48, 64), speed)
-            self.assertEqual(seed.dtype, np.float32)
-            plain = _decode_stored(_encode_physical(seed, False), False)
-            decoded = _decode_stored(_encode_physical(seed, True), True)
-            self.assertEqual(decoded.dtype, np.float64)
-            np.testing.assert_array_equal(plain, seed)
-            # Same FP32 physical seed; only encoding quantization may differ.
-            np.testing.assert_allclose(decoded, seed, rtol=0, atol=3e-10)
-            rho, momentum, velocity = moments(decoded)
-            rho2, velocity2 = primitives(decoded, CX, CY)
-            np.testing.assert_array_equal(rho, rho2)
-            np.testing.assert_array_equal(velocity, velocity2)
-            np.testing.assert_allclose(momentum, rho * velocity, atol=1e-17)
-            fluid = np.ones(seed.shape[1:], bool)
-            inv, flux = cv(decoded, fluid, (0, 0), (5, 4, 25, 20), 0.0072)
-            expected, _ = cv(seed, fluid, (0, 0), (5, 4, 25, 20), 0.0072)
-            np.testing.assert_allclose(inv, expected, rtol=0, atol=1e-6)
-            np.testing.assert_allclose(flux, 0, atol=1e-14)
-
     def test_descriptor_and_uniform_closed_flux(self):
         from CelerisLab.lbm.descriptors import D2Q9_EX, D2Q9_EY
 

tests/unit/test_jet_cv_measurements.py

@@ -31,111 +31,6 @@ from analyze_jet_cv_continuation import integrate, analyze
 
 
 class Measurements(unittest.TestCase):
-    def test_budget_failure_preserves_verified_pair(self):
-        from jet_cv_measurements import install_evidence
-
-        with tempfile.TemporaryDirectory() as tmp:
-            path = Path(tmp)
-            data = path / "samples.npz"
-            meta = path / "samples.json"
-            data.write_bytes(b"old-data")
-            meta.write_text("old-meta")
-            staged = path / ".samples.pending.npz"
-            staged_meta = path / ".meta.pending"
-            staged.write_bytes(b"x" * 100)
-            staged_meta.write_text("new-meta")
-            with self.assertRaisesRegex(RuntimeError, "last verified files preserved"):
-                install_evidence({data: staged, meta: staged_meta}, limit=20)
-            self.assertEqual(data.read_bytes(), b"old-data")
-            self.assertEqual(meta.read_text(), "old-meta")
-            self.assertEqual(
-                sorted(p.name for p in path.iterdir()), ["samples.json", "samples.npz"]
-            )
-
-    def test_dense_root_and_exact_old_cadence(self):
-        from jet_cv_measurements import old_sample_steps
-
-        for scale, count in ((1, 801), (2, 3201)):
-            c = case(scale, steps=800)
-            c["dense_root"] = True
-            dense = sample_steps(c)
-            old = old_sample_steps(c)
-            self.assertEqual(len(dense), count)
-            self.assertEqual(len(old), 276)
-            np.testing.assert_array_equal(dense, np.arange(800 * scale**2 + 1))
-            self.assertTrue(np.isin(old, dense).all())
-            np.testing.assert_array_equal(old[:101] / scale**2, np.arange(101))
-            np.testing.assert_array_equal(old[101:] / scale**2, np.arange(104, 801, 4))
-
-    def test_root_interval_absolute_net_mass_reversals(self):
-        from jet_cv_measurements import absolute_aperture_interval_mass
-
-        masks = (np.array([True, False]), np.array([False, True]))
-        rows = np.zeros((2, 4))
-        rows[0, 3] = 1
-        first = absolute_aperture_interval_mass(rows, *masks)
-        rows[0, 3] = -1
-        self.assertEqual(first + absolute_aperture_interval_mass(rows, *masks), 2)
-        self.assertEqual(abs(1 - 1), 0)
-
-    def test_dense_offline_oscillation_preserves_ledger(self):
-        import json
-        from jet_cv_measurements import old_sample_steps
-
-        for s in (1, 2):
-            c = case(s, steps=800)
-            c["dense_root"] = True
-            steps = sample_steps(c)
-            t = steps / s**2
-            n = len(t)
-            ridx = old_sample_steps(c)
-            # Nonzero steady reference fluxes cancel exact stock changes.
-            inv = np.zeros((n, 2, 2, 3))
-            inv[..., 0] = (100 - 0.001 * t[:, None, None]) * s**2
-            inv[..., 1] = -t[:, None, None] * 0.002 * s
-            flux = np.zeros_like(inv)
-            flux[..., 0] = 0.001 + 1e-4 * np.cos(np.pi / 2 * t[:, None, None])
-            flux[..., 1] = 0.002 / s
-            raw = np.zeros((n, 2, 5, 4))
-            with tempfile.TemporaryDirectory() as tmp:
-                path = Path(tmp)
-                (path / f"samples_s{s}.json").write_text(
-                    json.dumps(dict(case=c, sample_file="samples.npz"))
-                )
-                np.savez(
-                    path / "samples.npz",
-                    t_reference=t,
-                    cv_inventory=inv,
-                    cv_flux=flux,
-                    raw_cumulative=raw,
-                    ray_sample_indices=ridx,
-                    arc_estimates=np.zeros((len(ridx), 2, 2, 2, 2)),
-                    arc_absolute_estimates=np.zeros((len(ridx), 2, 2, 2, 2)),
-                    raw_absolute_aperture_mass_cumulative=np.broadcast_to(
-                        2 * t[:, None] * s**2, (n, 2)
-                    ),
-                    global_mass_reference=np.ones((n, 2)) * 100,
-                    level_stats=np.broadcast_to([1, 1, 0, 0.03, 100], (n, 3, 5)),
-                    velocity_error=np.zeros(n),
-                )
-                result = analyze(path)
-                w = result["windows"][1]
-                self.assertTrue(result["complete_schedule"])
-                np.testing.assert_allclose(
-                    np.array(w["normalized_cv_residual"])[0], 0, atol=1e-11
-                )
-                self.assertGreater(
-                    abs(np.array(w["old_cadence_change_from_dense"])[..., 0]).max(),
-                    0.005,
-                )
-                np.testing.assert_allclose(
-                    w["global_stock_raw_balance_sum_reference"], 0
-                )
-                np.testing.assert_allclose(
-                    w["raw_absolute_aperture_net_interval_mass_reference"], 200
-                )
-                np.testing.assert_allclose(w["raw_reaction_coefficient"], 0)
-
     def test_common_physical_schedule(self):
         a = sample_steps(case(1, steps=3200))
         b = sample_steps(case(2, steps=3200))
@@ -239,57 +134,6 @@ class Measurements(unittest.TestCase):
                 snap["uniform_velocity_reference"][0][fluid], 0.01, rtol=1e-7
             )
 
-    def test_dense_recorder_retains_sparse_rays(self):
-        h = 0.5
-        y, x = np.mgrid[:64, :64]
-        x = (x + 0.5) * h
-        y = (y + 0.5) * h
-        fluid = (x - 16) ** 2 + (y - 16) ** 2 > 25
-        fine_leaf = fluid.copy()
-        fine_leaf[:, :4] = False
-        root_leaf = np.zeros((32, 32), bool)
-        root_leaf[:, :2] = True
-
-        def eq(ny, nx):
-            cu = CX[:, None, None] * 0.01
-            return np.broadcast_to(
-                W[:, None, None] * (1 + 3 * cu + 4.5 * cu * cu - 1.5 * 0.01**2),
-                (9, ny, nx),
-            ).copy()
-
-        ug = eq(64, 64)
-        levels = [eq(32, 32), ug.copy()]
-        c = dict(
-            scale=1,
-            dense_root=True,
-            steps=800,
-            center=[16, 16],
-            diameter=10,
-            u=0.01,
-            nu=0.0072,
-            contours=[[9, 9, 23, 23], [7, 7, 25, 25]],
-        )
-        data = {
-            name: [cv(ug, mask, (0, 0), r, c["nu"]) for r in c["contours"]]
-            for name, mask in [("uniform_adapter", fluid), ("amr_api", fine_leaf)]
-        }
-        with tempfile.TemporaryDirectory() as tmp:
-            rec = MeasurementRecorder(
-                c, tmp, fluid, [root_leaf, fine_leaf], (0, 0), CX, CY
-            )
-            for step in (0, 101):
-                rec.capture(step, ug, levels, data, {k: np.zeros(2) for k in data}, 0.0)
-                rec.records[-1]["raw_cumulative"] = np.zeros((2, 5, 4))
-                rec.records[-1]["raw_absolute_aperture_mass_cumulative"] = np.zeros(2)
-            rec.flush()
-            a = np.load(Path(tmp) / "samples_s1.npz")
-            np.testing.assert_array_equal(a["step"], [0, 101])
-            np.testing.assert_array_equal(a["ray_sample_indices"], [0])
-            self.assertEqual(a["arc_estimates"].shape, (1, 2, 2, 2, 2))
-            self.assertTrue(
-                np.all(a["arc_absolute_estimates"] >= abs(a["arc_estimates"]) - 1e-14)
-            )
-
     def test_offline_mass_force_normalization(self):
         import json
 

tests/unit/test_raw_exchange_window.py

@@ -0,0 +1,45 @@
+"""Host reduction contract for the optional observation-window exchange ledger."""
+import unittest
+
+import numpy as np
+
+from CelerisLab.lbm.raw_exchange import summarize_exchange
+
+
+def reduce(rows, window, duration=None, h=.5, time_end=10):
+    return summarize_exchange(rows, np.array([0, 0, 1]), np.array([False, True, False]),
+                              body_id=0, level=1, h=h, time_end=time_end, window=window,
+                              reference_center=(0., 0.), duration=duration)
+
+
+class ExchangeWindowReduction(unittest.TestCase):
+    def test_window_duration_scaling(self):
+        rows = np.array([[1., 2., 3., 4.], [.5, -1., 0., 1.], [9., 9., 9., 9.]])
+        out = reduce(rows, "window", duration=4)
+        np.testing.assert_array_equal(out["total"]["impulse"], [1.5, 1.])
+        np.testing.assert_array_equal(out["total"]["force"], [1.5 / 4, 1. / 4])
+        self.assertEqual(out["total"]["torque"], 3. / 4)
+        self.assertEqual(out["total"]["mass_rate"], 5. / 4)
+        np.testing.assert_array_equal(out["aperture"]["impulse"], [.5, -1.])
+        self.assertEqual((out["time_start"], out["time_end"]), (6., 10))
+        self.assertEqual(out["sample_count"], 8)  # duration / h fine substeps
+        self.assertEqual(out["window"], "window")
+
+    def test_existing_windows_unchanged(self):
+        rows = np.array([[1., 2., 3., 4.], [0., 0., 0., 0.], [0., 0., 0., 0.]])
+        base = reduce(rows, "last_base_step")
+        self.assertEqual((base["time_start"], base["sample_count"]), (9., 2))
+        np.testing.assert_array_equal(base["total"]["force"], [1., 2.])
+        sub = reduce(rows, "last_substep")
+        self.assertEqual(sub["sample_count"], 1)
+
+    def test_duration_validation(self):
+        rows = np.zeros((3, 4))
+        for bad in (None, 0, -1, 1.5, True):
+            with self.subTest(duration=bad), self.assertRaises(ValueError):
+                reduce(rows, "window", duration=bad)
+        with self.assertRaises(ValueError):
+            reduce(rows, "last_base_step", duration=2)
+
+
+if __name__ == "__main__": unittest.main()

tests/unit/test_refined_dfg.py

@@ -13,7 +13,15 @@ h, periodic = r.helpers(r.DEFAULT_SOURCE)
 
 
 def test_source_pin():
-    assert r.verify_source(r.DEFAULT_SOURCE)["commit"].startswith("cd3a657")
+    historical = json.loads((r.HERE / "data/refined_dfg/source_manifest.json").read_text())
+    assert historical["commit"].startswith("cd3a657")
+    mrt_history = json.loads((r.HERE / "data/refined_dfg/source_manifest_mrt.json").read_text())
+    assert mrt_history["commit"].startswith("3ea131e")
+    assert r.verify_source(r.DEFAULT_SOURCE, "MRT", "window")["commit"].startswith("a233c90")
+    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")
 
 
 def test_matched_units_and_planes():
@@ -135,16 +143,20 @@ def test_axis_specific_sampling_phase():
 
 def test_full_requires_completed_pilot():
     with unittest.TestCase().assertRaisesRegex(ValueError, "pilot-results"):
-        r.verify_pilot(None, r.verify_source(r.DEFAULT_SOURCE))
+        r.verify_pilot(None, r.verify_source(r.DEFAULT_SOURCE, "MRT", "window"))
 
 
-def pilot_fixture(path):
+def pilot_fixture(path, observation_mode="window"):
     """Synthetic contract fixture only; never a physical result or run receipt."""
-    manifest = r.verify_source(r.DEFAULT_SOURCE)
+    manifest = (r.verify_source(r.DEFAULT_SOURCE, "MRT", "window")
+                if observation_mode == "window" else
+                json.loads((r.HERE / "data/refined_dfg/source_manifest_mrt.json").read_text()))
     (path / "plan.json").write_text(
         json.dumps(
             dict(
                 source_manifest=manifest,
+                collision="MRT",
+                observation_mode=observation_mode,
                 bounds_root_edge=list(r.BOUNDS),
                 runner_sha256=r.runner_hash(),
             )
@@ -177,6 +189,8 @@ def pilot_fixture(path):
             json.dumps(
                 dict(
                     case=case,
+                    collision="MRT",
+                    observation_mode=observation_mode,
                     mode="pilot",
                     status="pilot_complete",
                     healthy=True,
@@ -192,6 +206,8 @@ def pilot_fixture(path):
             json.dumps(
                 dict(
                     case=case,
+                    collision="MRT",
+                    observation_mode=observation_mode,
                     mode="pilot",
                     step=interval,
                     runner_sha256=r.runner_hash(),
@@ -208,7 +224,7 @@ def test_pilot_validation_and_calibration():
     with TemporaryDirectory() as tmp:
         path = Path(tmp)
         manifest = pilot_fixture(path)
-        assert r.verify_pilot(path, manifest) == 3606.0
+        assert r.verify_pilot(path, manifest, "MRT", "window") == 3606.0
 
 
 def test_reject_malformed_pilot_receipts():
@@ -238,7 +254,7 @@ def test_reject_malformed_pilot_receipts():
             mutate(bad)
             receipt.write_text(json.dumps(bad))
             with unittest.TestCase().assertRaises(ValueError):
-                r.verify_pilot(path, manifest)
+                r.verify_pilot(path, manifest, "MRT", "window")
         # Exact reviewer regression: previously accepted NaN endpoint.
         receipt.write_text(
             json.dumps(
@@ -250,11 +266,11 @@ def test_reject_malformed_pilot_receipts():
             )
         )
         with unittest.TestCase().assertRaises(ValueError):
-            r.verify_pilot(path, manifest)
+            r.verify_pilot(path, manifest, "MRT", "window")
         receipt.write_text(json.dumps(valid))
         (path / "uniform/ready.json").write_text("{}")
         with unittest.TestCase().assertRaises(ValueError):
-            r.verify_pilot(path, manifest)
+            r.verify_pilot(path, manifest, "MRT", "window")
 
 
 def test_strict_finite_json_and_deadline():
@@ -283,6 +299,158 @@ def test_pressure_probes_actual_body_mask():
     assert h.pressure_probes(np.ones(pair.fine_shape), fluid, p)["dp"] == 0.0
 
 
+def test_matched_mrt_selection():
+    for diameter in (40, 80):
+        srt, sbody, sp = r.config(h, diameter)
+        mrt, mbody, mp = r.config(h, diameter, "MRT")
+        assert mrt["method"]["collision"] == "MRT"
+        mrt["method"]["collision"] = "SRT"
+        assert mrt == srt and mbody == sbody and mp == sp
+    srt_plan, mrt_plan = r.plan(h), r.plan(h, "MRT")
+    for key in ("root", "uniform", "bounds_root_edge", "finest_dx_m", "finest_dt_s"):
+        assert srt_plan[key] == mrt_plan[key]
+    assert "q2/6/8" in mrt_plan["outlet"] and "all nine" in srt_plan["outlet"]
+    cfg, pair, _, host = r.host_geometry(r.DEFAULT_SOURCE, h, "MRT")
+    assert cfg.collision == "MRT" and pair.fine_shape == (240, 568)
+    assert len(host.boundary_cells) > 0
+
+
+def test_reject_collision_receipts():
+    from tempfile import TemporaryDirectory
+
+    for rel in ("plan.json", "uniform/summary.json", "refined/summary.json",
+                "uniform/ready.json", "refined/ready.json"):
+        for invalid in ("SRT", None):
+            with TemporaryDirectory() as tmp:
+                path = Path(tmp)
+                manifest = pilot_fixture(path)
+                receipt = path / rel
+                record = json.loads(receipt.read_text())
+                if invalid is None:
+                    record.pop("collision")
+                else:
+                    record["collision"] = invalid
+                receipt.write_text(json.dumps(record))
+                with unittest.TestCase().assertRaises(ValueError):
+                    r.verify_pilot(path, manifest, "MRT", "window")
+
+
+def test_record_actual_checkout_separately():
+    from source_provenance import checkout_provenance
+
+    actual = checkout_provenance(r.DEFAULT_SOURCE)
+    assert len(actual["checkout_revision"]) == 40
+    assert isinstance(actual["checkout_dirty"], bool)
+    assert r.verify_source(r.DEFAULT_SOURCE, "MRT", "window")["production_origin_commit"].startswith("d930")
+
+
+def test_window_plan_preserves_physics():
+    for collision in ("SRT", "MRT"):
+        baseline = r.plan(h, collision)
+        window = r.plan(h, collision, "window")
+        assert baseline.pop("observation_mode") == "last_base_step"
+        assert window.pop("observation_mode") == "window"
+        assert baseline == window
+        assert r.verify_source(r.DEFAULT_SOURCE, collision, "window")["observation_mode"] == "window"
+    with unittest.TestCase().assertRaises(ValueError):
+        r.verify_source(r.DEFAULT_SOURCE, "MRT", "unknown")
+
+
+def test_reject_observation_mode_receipts():
+    from tempfile import TemporaryDirectory
+
+    for mode, other in (("window", "last_base_step"), ("last_base_step", "window")):
+        with TemporaryDirectory() as tmp:
+            path = Path(tmp)
+            manifest = pilot_fixture(path, mode)
+            assert r.verify_pilot(path, manifest, "MRT", mode) == 3606.0
+            with unittest.TestCase().assertRaises(ValueError):
+                r.verify_pilot(path, manifest, "MRT", other)
+        for rel in ("plan.json", "uniform/summary.json", "refined/summary.json",
+                    "uniform/ready.json", "refined/ready.json"):
+            for invalid in (other, None):
+                with TemporaryDirectory() as tmp:
+                    path = Path(tmp)
+                    manifest = pilot_fixture(path, mode)
+                    receipt = path / rel
+                    record = json.loads(receipt.read_text())
+                    if invalid is None:
+                        record.pop("observation_mode")
+                    else:
+                        record["observation_mode"] = invalid
+                    receipt.write_text(json.dumps(record))
+                    with unittest.TestCase().assertRaises(ValueError):
+                        r.verify_pilot(path, manifest, "MRT", mode)
+
+
+class ExchangeFixture:
+    """Synthetic interval ledger for orchestration tests, not a solver or GPU result."""
+    def __init__(self):
+        self.step_count = 11
+        self.events = []
+        self.begin = None
+        self.override = {}
+
+    def start_exchange_window(self):
+        self.begin = self.step_count
+        self.events.append(("start", self.step_count))
+
+    def step(self, n):
+        self.events.append(("step", n))
+        self.step_count += n
+
+    def result(self, start, window):
+        from CelerisLab.lbm.raw_exchange import summarize_exchange
+
+        total = sum(range(start + 1, self.step_count + 1))
+        result = summarize_exchange([[total, -2*total, 0., 0.]], [0], [False],
+            body_id=0, level=1, h=.5, time_end=self.step_count, window=window,
+            reference_center=(80.5, 81.),
+            **({"duration": self.step_count-start} if window == "window" else {}))
+        result.update(self.override)
+        return result
+
+    def read_exchange_window(self, body_id, *, level):
+        assert body_id == 0 and level == 1
+        self.events.append(("read_window", self.step_count))
+        return self.result(self.begin, "window")
+
+    def read_raw_exchange(self, body_id, *, level, window):
+        assert body_id == 0 and level == 1 and window == "last_base_step"
+        self.events.append(("read_raw", self.step_count))
+        return self.result(self.step_count-1, window)
+
+
+def test_window_interval_impulse_sampling_and_units():
+    base, window = ExchangeFixture(), ExchangeFixture()
+    for n in (80, 2000, 3):  # Pilot/dense, early full, and final partial interval.
+        begin = base.step_count
+        a = r.advance_refined_interval(base, n, float("inf"), "last_base_step")
+        w = r.advance_refined_interval(window, n, float("inf"), "window")
+        np.testing.assert_array_equal(w, a)
+        expected = sum(range(begin+1, begin+n+1))
+        np.testing.assert_array_equal(w, [expected, -2*expected])
+        den = .5 * r.SPEED**2 * 40
+        np.testing.assert_array_equal(w/n/den, a/n/den)
+        assert window.step_count == base.step_count == begin+n
+        assert window.events[-3:] == [("start", begin), ("step", n), ("read_window", begin+n)]
+        assert (begin+n/2)*r.SPEED/40 == (window.step_count-n/2)*r.SPEED/40
+    assert not any(event[0] == "read_raw" for event in window.events)
+
+
+def test_window_contract_rejection():
+    for key, value in (("units", "local"), ("window", "last_base_step"),
+                       ("sample_count", 2), ("time_start", 0), ("time_end", 0), ("dt", 1.)):
+        backend = ExchangeFixture()
+        backend.override[key] = value
+        with unittest.TestCase().assertRaisesRegex(RuntimeError, "window force"):
+            r.advance_refined_interval(backend, 80, float("inf"), "window")
+    backend = ExchangeFixture()
+    with unittest.TestCase().assertRaises(TimeoutError):
+        r.advance_refined_interval(backend, 80, 0, "window")
+    assert not backend.events
+
+
 if __name__ == "__main__":
     suite = unittest.TestSuite(
         unittest.FunctionTestCase(fn)

tests/unit/test_refinement.py

@@ -142,7 +142,7 @@ class FixedPatchTests(unittest.TestCase):
                 FixedPatch2D.from_config(LBMConfig(nx=nx, ny=ny, viscosity=.1), bounds)
 
     def test_unsupported_backend_and_fine_relaxation_rejected(self):
-        for kwargs in ({"collision": "TRT"}, {"collision": "MRT"}, {"streaming": "esopull"},
+        for kwargs in ({"collision": "TRT"}, {"streaming": "esopull"},
                        {"les_enabled": True},
                        {"store_precision": "FP16S"}, {"lattice_model": "D3Q19"},
                        {"data_type": "FP64"}, {"omega_min": 1.0}):

tests/unit/test_refinement_mrt.py

@@ -0,0 +1,44 @@
+"""Host guards and source contract for MRT reuse in refined/blocked stepping (no CUDA)."""
+from pathlib import Path
+import unittest
+
+from CelerisLab.config import LBMConfig
+from CelerisLab.lbm.refinement import FixedPatch2D
+
+KERNELS = Path(__file__).resolve().parents[2] / "src/CelerisLab/lbm/kernels"
+BOUNDS = (8, 6, 16, 14)
+
+
+class RefinementMRTHost(unittest.TestCase):
+    def test_patch_accepts_implemented_models(self):
+        for collision in ("SRT", "MRT"):
+            with self.subTest(collision=collision):
+                FixedPatch2D.from_config(LBMConfig(nx=24, ny=20, viscosity=.1, collision=collision), BOUNDS)
+
+    def test_patch_rejects_unimplemented(self):
+        for options in (dict(collision="TRT"), dict(collision="MRT", les_enabled=True)):
+            with self.subTest(options=options), self.assertRaises(ValueError):
+                FixedPatch2D.from_config(LBMConfig(nx=24, ny=20, viscosity=.1, **options), BOUNDS)
+
+    def test_srt_statements_and_guards_preserved(self):
+        """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()
+        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, 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):
+            self.assertIn("(COLLISION_MODEL != 0 && COLLISION_MODEL != 2)", source)
+            self.assertIn("USE_LES", source)
+            self.assertIn("STORE_PRECISION != 0", source)
+            self.assertIn('#if COLLISION_MODEL == 2\n#include "operators/collision_mrt.cuh"\n#endif', source)
+        self.assertIn("USE_DDF_SHIFTING ||", refinement)  # dense/mixed shifted stays rejected
+        for source in (blocks, channel, refinement):
+            self.assertEqual(source.count("collide_mrt_no_force("), 1)
+
+
+if __name__ == "__main__": unittest.main()

tests/unit/test_time_refined_dfg.py

@@ -0,0 +1,86 @@
+"""CPU contract of the measurement-only refined DFG timing harness (no CUDA)."""
+from pathlib import Path
+import sys
+import unittest
+
+import numpy as np
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "validation"))
+import time_refined_dfg as t  # noqa: E402
+
+
+class TimingHarnessContract(unittest.TestCase):
+    def test_equal_simulated_duration(self):
+        s = t.schedule(2000, 200)
+        self.assertEqual(s["refined"], (200, 2000))
+        self.assertEqual(s["uniform"], (400, 4000))  # one refined base step = two D80 steps
+        self.assertAlmostEqual(s["tstar_window"], 2000 * .01 / 40)
+        for bad in ((0, 1), (1, -1)):
+            with self.assertRaises(ValueError):
+                t.schedule(*bad)
+
+    def test_timing_only_labels(self):
+        for name in ("B", "C", "E'"):
+            self.assertFalse(t.VARIANTS[name]["force_history_valid"])
+            self.assertTrue(t.VARIANTS[name]["timing_only"])
+        for name in ("A", "E"):
+            self.assertFalse(t.VARIANTS[name]["force_history_valid"])  # reads discarded
+            self.assertTrue(t.VARIANTS[name]["force_reads"])
+            self.assertIn("runner-like", t.VARIANTS[name]["force_note"])
+        self.assertEqual({v["case"] for v in t.VARIANTS.values()}, {"refined", "uniform"})
+        self.assertIn("final base interval", t.VARIANTS["C"]["force_note"])
+        self.assertIn("whole-window", t.VARIANTS["E'"]["force_note"])
+
+    def test_deadline(self):
+        import time
+        with self.assertRaises(TimeoutError):
+            t.check_deadline(time.monotonic() - 1)
+        t.check_deadline(time.monotonic() + 60)
+
+    def test_parity_detects_bits_and_nan_masks(self):
+        a = np.array([[1., np.nan], [2., 3.]], np.float32)
+        self.assertIsNone(t.same_state([a], [a.copy()]))
+        b = a.copy(); b[1, 1] = np.nextafter(np.float32(3), np.float32(4))
+        self.assertIn("bits", t.same_state([a], [b]))
+        c = a.copy(); c[0, 0] = np.nan
+        self.assertIn("NaN", t.same_state([a], [c]))
+        self.assertIn("level count", t.same_state([a], [a, a]))
+        inf = a.copy(); inf[1, 0] = np.inf
+        self.assertIn("infinite", t.same_state([inf], [inf.copy()]))  # matching inf still fails
+        self.assertIn("infinite", t.same_state([a], [inf]))
+
+    def test_flat_uniform_ddf(self):
+        fluid = np.array([[True, False, True], [True, True, False]])
+        flat = np.ones(9 * fluid.size, np.float32)  # Simulation.get_ddf() layout
+        self.assertTrue(t.uniform_populations_finite(flat, fluid))
+        flat[1 * fluid.size + 1] = np.nan  # q=1, solid cell 1: ignored
+        self.assertTrue(t.uniform_populations_finite(flat, fluid))
+        flat[4 * fluid.size + 0] = np.nan  # q=4, fluid cell 0: detected
+        self.assertFalse(t.uniform_populations_finite(flat, fluid))
+        self.assertTrue(t.uniform_populations_finite(np.ones((9,) + fluid.shape), fluid))
+        with self.assertRaises(RuntimeError):
+            t.uniform_populations_finite(np.ones(9 * fluid.size + 1), fluid)
+
+    def test_physical_health_limits(self):
+        ok = (np.array([1., 1.01]), np.array([.01, .02]), np.array([0., .01]))
+        health = t.physical_health([ok])
+        self.assertAlmostEqual(health["rho_max"], 1.01)
+        for bad in ((np.array([1., np.nan]),) + ok[1:],
+                    (np.array([.4, 1.]),) + ok[1:],
+                    (ok[0], np.array([.2, 0.]), ok[2]),
+                    (np.array([]), np.array([]), np.array([]))):
+            with self.assertRaises(RuntimeError):
+                t.physical_health([bad])
+
+    def test_summary_reports_differences_without_claims(self):
+        rows = [dict(variant=v, seconds=s) for v, s in
+                (("A", 3.), ("A", 5.), ("B", 2.), ("C", 1.), ("E", 2.), ("E'", 1.5))]
+        s = t.summarize(rows, .5)
+        self.assertEqual(s["A"]["median_seconds"], 4.)
+        self.assertEqual(s["A_minus_B_seconds"], 2.)
+        self.assertEqual(s["B_minus_C_seconds"], 1.)
+        self.assertEqual(s["A_over_E"], 2.)
+        self.assertIn("not exact causal", s["interpretation"])
+
+
+if __name__ == "__main__": unittest.main()

tests/unit/test_time_window_dfg.py

@@ -0,0 +1,38 @@
+"""CPU contract of the A-versus-window timing harness (no CUDA)."""
+from pathlib import Path
+import sys
+import unittest
+
+import numpy as np
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "validation"))
+import time_window_dfg as w  # noqa: E402
+
+
+class WindowTimingContract(unittest.TestCase):
+    def test_interval_and_variants(self):
+        self.assertEqual(w.INTERVAL, 80)
+        self.assertEqual(w.VARIANTS, ("A", "W", "W1"))
+
+    def test_components(self):
+        obs = dict(total=dict(impulse=np.array([1., 2.]), angular_impulse=3., mass_into_fluid=4.))
+        np.testing.assert_array_equal(w.components(obs), [1., 2., 3., 4.])
+
+    def test_interval_comparison(self):
+        ref = np.array([[1., -1., 0., 0.]])
+        scale = np.array([[10., 10., 1., 1.]])
+        self.assertEqual(w.compare_intervals(ref, scale, ref.copy()), 0.)
+        near = ref + np.array([[5e-9, 0., 0., 0.]])
+        self.assertLessEqual(w.compare_intervals(ref, scale, near), 1.)
+        far = ref + np.array([[1e-7, 0., 0., 0.]])
+        self.assertGreater(w.compare_intervals(ref, scale, far), 1.)
+        zero = np.zeros((1, 4))
+        self.assertGreater(w.compare_intervals(zero, zero, zero + 1e-12), 1.)  # floor, no div-by-0
+
+    def test_candidate_identity_records_files(self):
+        identity = w.candidate_identity(Path(__file__).resolve().parents[2])
+        self.assertIn("checkout_revision", identity)
+        self.assertEqual(set(identity["sha256"]), set(w.CANDIDATE_FILES))
+
+
+if __name__ == "__main__": unittest.main()

tests/validation/analyze_jet_cv_continuation.py

@@ -32,7 +32,7 @@ def analyze(directory):
     t = a["t_reference"]
     if not len(t) or t[0] != 0 or np.any(np.diff(t) <= 0):
         raise ValueError("Sample reference times must increase strictly from zero")
-    from jet_cv_measurements import sample_steps, old_sample_steps
+    from jet_cv_measurements import sample_steps
 
     complete_schedule = np.array_equal(t, sample_steps(c) / s**2)
     records = []
@@ -45,15 +45,7 @@ def analyze(directory):
     )  # angular impulse reference = root/s^2 (not used in force tests)
     q0 = 2 * np.pi / 12 * (c["diameter"] / 2) * c["jet_speed"]
     f0 = 0.5 * (c["u"] * s) ** 2 * (c["diameter"] / s)
-    dense_root = c.get("dense_root", False)
-    old_times = old_sample_steps(c) / s**2
-    ray_indices = (
-        a["ray_sample_indices"] if "ray_sample_indices" in a else np.arange(len(t))
-    )
-    ray_times = t[ray_indices]
-    if dense_root and complete_schedule and not np.array_equal(ray_times, old_times):
-        raise ValueError("Sparse ray records do not match the declared old cadence")
-    for lo in range(0, c.get("steps", 3200 * s * s) // (s * s), 100):
+    for lo in range(0, 3200, 100):
         hi = lo + 100
         ids = np.flatnonzero((t >= lo) & (t <= hi))
         if len(ids) < 2 or t[ids[0]] != lo or t[ids[-1]] != hi:
@@ -62,14 +54,8 @@ def analyze(directory):
         delta = inv[last] - inv[first]
         ledger = raw[last] - raw[first]
         cv_variants = []
-        selections = [np.unique(np.r_[ids[::stride], ids[-1]]) for stride in (1, 2, 4)]
-        if dense_root:
-            selections.append(ids[np.isin(t[ids], old_times)])
-        for selected in selections:
-            if t[selected[0]] != lo or t[selected[-1]] != hi:
-                raise ValueError(
-                    "Every quadrature selection must retain both window endpoints"
-                )
+        for stride in (1, 2, 4):
+            selected = np.unique(np.r_[ids[::stride], ids[-1]])
             total = integrate(t[selected], flux[selected])
             m = delta[:, :, 0] + total[:, :, 0]
             j = -delta[:, :, 1:] - total[:, :, 1:]
@@ -82,61 +68,16 @@ def analyze(directory):
                     axis=-1,
                 )
             )
-        ray_ids = np.flatnonzero((ray_times >= lo) & (ray_times <= hi))
         arc = (
-            integrate(ray_times[ray_ids], a["arc_estimates"][ray_ids]) / 100
+            integrate(t[ids], a["arc_estimates"][ids]) / 100
         )  # Q/M rate conversion is unity
         # Each estimator, aperture, Q/M; compare against signed prescribed Q (rho_ref=1 for M only).
         arc_ratio = arc / np.array([q0, -q0])[None, None, :, None]
         global_delta = (
             a["global_mass_reference"][last] - a["global_mass_reference"][first]
         )
-        dense_details = {}
-        if dense_root:
-            absolute = (
-                a["raw_absolute_aperture_mass_cumulative"][last]
-                - a["raw_absolute_aperture_mass_cumulative"][first]
-            ) / s**2
-            stock_raw = np.diff(a["global_mass_reference"][ids], axis=0) - np.diff(
-                raw[ids, :, 2, 3], axis=0
-            )
-            partition = ledger[:, 2] - ledger[:, 0] - ledger[:, 1]
-            dense_details = dict(
-                quadrature_names=[
-                    "root_stride1",
-                    "root_stride2",
-                    "root_stride4",
-                    "exact_old_cadence",
-                ],
-                root_dt_reference=1 / s**2,
-                old_cadence_change_from_dense=(
-                    cv_variants[3] - cv_variants[0]
-                ).tolist(),
-                root_temporal_sensitivity_max=float(
-                    np.max(abs(np.array(cv_variants)[1:3] - cv_variants[0]))
-                ),
-                raw_absolute_aperture_net_interval_mass_reference=absolute.tolist(),
-                prescribed_unsigned_nominal_mass_reference=float(2 * q0 * 100),
-                legacy_raw_gross_field_definition="raw_gross_mass_over_nominal retains abs(window net blowing)+abs(window net suction), divided by ONE nominal aperture throughput; it is not root-interval absolute throughput.",
-                raw_absolute_to_prescribed_unsigned_nominal=(
-                    absolute / (2 * q0 * 100)
-                ).tolist(),
-                global_stock_raw_balance_max_abs_root_interval_reference=np.max(
-                    abs(stock_raw), axis=0
-                ).tolist(),
-                global_stock_raw_balance_sum_reference=stock_raw.sum(axis=0).tolist(),
-                raw_total_minus_wall_aperture_reference=partition.tolist(),
-                raw_aperture_minus_blowing_suction_reference=(
-                    ledger[:, 1] - ledger[:, 3] - ledger[:, 4]
-                ).tolist(),
-                measured_ray_absolute_volume_mass_reference=integrate(
-                    ray_times[ray_ids], a["arc_absolute_estimates"][ray_ids]
-                ).tolist(),
-                measured_ray_cadence="Sparse old cadence; estimator-dependent, not dense-root or exact wall throughput",
-            )
         records.append(
             dict(
-                **dense_details,
                 t_reference=[lo, hi],
                 startup=hi <= 800,
                 normalized_cv_residual=np.array(cv_variants).tolist(),
@@ -156,12 +97,6 @@ def analyze(directory):
         )
     return dict(
         scale=s,
-        dense_root=dense_root,
-        previous_temporal_failure_retained=(
-            "Prior continuation stride2 changes exceeded baseline residual in all48 audited comparisons; this diagnostic does not retroactively change those gates."
-            if dense_root
-            else None
-        ),
         complete_schedule=bool(complete_schedule),
         sample_checksum_verified=checksum_verified,
         windows=records,
@@ -187,10 +122,8 @@ def main():
         results=results,
         physics_validation_pass=False,
     )
-    if (
-        not any(x.get("dense_root", False) for x in results)
-        and sorted(x["scale"] for x in results) == [1, 2]
-        and all(x["complete_schedule"] for x in results)
+    if sorted(x["scale"] for x in results) == [1, 2] and all(
+        x["complete_schedule"] for x in results
     ):
         ordered = sorted(results, key=lambda x: x["scale"])
         summary = []

tests/validation/data/refined_dfg/source_manifest_mrt.json

@@ -0,0 +1,112 @@
+{
+  "commit": "3ea131e92c057843034f11a1c574301a2430dfce",
+  "production_origin_commit": "3ea131e92c057843034f11a1c574301a2430dfce",
+  "collision": "MRT",
+  "sha256": {
+    "src/CelerisLab/__init__.py": "21cc469bd6e970a1f9ff5970c8eea4602065e41fc881854759ee565ee55c10c6",
+    "src/CelerisLab/body/__init__.py": "e4e0e8f0769a94428d5b02fac790443997556cfc5515a5f1ed1595d2c694d69d",
+    "src/CelerisLab/body/action_smoother.py": "b1a23ee6bbd4850b172cb8738c5c3546ee465a54189e6af9d8044f023d4687dc",
+    "src/CelerisLab/body/coupling/__init__.py": "c931f7eb318d48408c3e2febcc6917c88c8dcc1697cbf4329b1f340d94c35763",
+    "src/CelerisLab/body/coupling/soa_packer.py": "8ec26fd55ef74801bbd5ab6bd3d0a7dabc750cbd498aaac2ca4c61a0108ad3c9",
+    "src/CelerisLab/body/ddf_patch.py": "60fcfb7441e73bf290bcff41ca3a83da3e7ddad765d7247464e8f02e3043b4e2",
+    "src/CelerisLab/body/geometry/__init__.py": "ada88980ab672959aca9aa3a4c4aade3219524b2ab68110cd2a2f41e3e0ca90a",
+    "src/CelerisLab/body/geometry/base.py": "cceca14361ca2f34d34a3e43730e1a79b8ca81be00f0dd62e884db32b479419f",
+    "src/CelerisLab/body/geometry/circle.py": "7605ca7c21103f1d801ac73ab8dd18dffbdac425b2dee7b8858b283628a9e436",
+    "src/CelerisLab/body/jets.py": "4104c24af4651f4bb4f8d8fd36fb6d88f41c6e4469822f36e22e74c9ede31dac",
+    "src/CelerisLab/body/manager.py": "01b653d25bd9b4333379c8bfd12383596b686d2029f2e91b1c91b618540ec6b3",
+    "src/CelerisLab/body/objects.py": "8e63cee8faa9cc6243bcff428be628dfad73a02a9a3b8ece7d78c0a63406a08d",
+    "src/CelerisLab/body/preprocess/__init__.py": "dfd9d58bc07f626ab23a70053e31b88a6d9165d2e4c76a4e9f48abe674d3d2ec",
+    "src/CelerisLab/body/preprocess/flag_overlay.py": "831fc9e9736b758847759a085ff133eb49ec91318ace4fda76f85960138ee300",
+    "src/CelerisLab/body/registry.py": "fbb7327780adbdc6da15ebdb799980daf32df22ee974af0893d7409f177bbb68",
+    "src/CelerisLab/body/sync_plan.py": "3a15bfe9ebeb91482dbfc68850761ef39c872cc12d2efc795babd0c01e392323",
+    "src/CelerisLab/common/__init__.py": "d4dbde78bc97fccc9eabac9abae8f0f15c112c10ff8e9b45bd9401d4723df1ef",
+    "src/CelerisLab/common/_types.py": "aad2ff1a9033d4fd68783ad7123afb47201ddc58785cb6dc0246947037072a5f",
+    "src/CelerisLab/common/checkpoint.py": "b2f960aecd5ac27a3a7130b809ffad4709c44d522edef0359ab952654625410b",
+    "src/CelerisLab/common/pathline.py": "a1b6efcc0a076d0f1dc8d49a595779426d7d5626b00072a59bac710ff02ad046",
+    "src/CelerisLab/common/preprocess.py": "7fc78a08857cf8e98519500b094a1d896af189b6b15658ff9b388ed51bed4515",
+    "src/CelerisLab/common/render.py": "0e055476257eb2bcd11e733766e52195de02f8fad228f570a3511a7c831bdc29",
+    "src/CelerisLab/common/streakline/__init__.py": "13ff1cfa416fdd721a08a659e76b5a5a99b8d3201add10a83b7b0bc654bc1eec",
+    "src/CelerisLab/common/streakline/_config.py": "a9185ef68e9c7b5d5a1dc0b04a28341e2ad91d13c145a7c106636a878cf0a605",
+    "src/CelerisLab/common/streakline/_integrate.py": "3fbd4368b2c8d0e7fda40dc65aac5df02b085008b4c68acd83cb7ae45d33a13b",
+    "src/CelerisLab/common/streakline/_render.py": "236cd637414ee318c4d66f9aed32110d8dc42a100d65fde06172f7cd0ba62c65",
+    "src/CelerisLab/common/streakline/_streakline.py": "1a3e08daa706b5f83cdeb337e9a945a36ddf7583c40af1d0bb9d825ac954df37",
+    "src/CelerisLab/config.py": "2f1bba7515f5bc08cb3a8d140bcafe7fa0689e01b3b06494697bb2956a71530e",
+    "src/CelerisLab/configs/CONFIG.md": "41cafdbe525366c08aa9d0416ea026ccf0026ba7123b8a9b65f968a80d928e3a",
+    "src/CelerisLab/configs/config_body.json": "055f261f0828ca1d4f2bb694bd15b72bbcabe31d552326208f28e37f761adab1",
+    "src/CelerisLab/configs/config_lbm.json": "17e4311f1f250c6ece59177a8b6ff856b69a016c7c10267e774b030a0a390865",
+    "src/CelerisLab/configs/config_lbm_flume_mrt_zouhe.json": "8f593321db05a563a42931603a228d61cbcba06dfbe113262d5cd0ad93c0fe79",
+    "src/CelerisLab/configs/config_lbm_flume_trt_regularized.json": "f4bba8e6a4339fc3c4eb14f6e2f57c4b22993c73703c773d853edfbe34736e56",
+    "src/CelerisLab/configs/config_lbm_three_cylinder_triangle.json": "eeb8e7149dbe4aa6b3a9562802735aa950750dfbe2d972a0727e788573afa817",
+    "src/CelerisLab/cuda/__init__.py": "0d01dff9f0399c8baba822aab5ca0122a135f011008ed8a499ebd9f7e5b06aea",
+    "src/CelerisLab/cuda/compiler_v2.py": "c4cc3ec4445c8d75e6a253dd87963877117a722bc73b9d4811806fc6045d08c1",
+    "src/CelerisLab/cuda/context.py": "013dd4ec6d01d2fbd11afdfb103eac6a273bd9f798d663d761b851eeb27885e9",
+    "src/CelerisLab/cuda/obs_layout.py": "5ffb2c3aeeaaaaeb64f2cd8febfbcf4ad9919e5414020ad753cfade0683b21c2",
+    "src/CelerisLab/lbm/__init__.py": "7f60fb3aaae81501b510d08c2fedff44fcb05657260eb10b6f5cc91e7204306b",
+    "src/CelerisLab/lbm/block_stepper.py": "00b313daddc6e63d8ab8423c9190ca7e92ad291de20a2f6b80eb53e3f2d85ec7",
+    "src/CelerisLab/lbm/blocks.py": "1e644c5719993f418185ad4a10b6950a30636fc912befd9eab41ff48b4cf85cd",
+    "src/CelerisLab/lbm/channel.py": "40d0354e5dc97bcdb7e7b5c1e7f9303a109b7134d8a3af88b8ff4cd75ab5a095",
+    "src/CelerisLab/lbm/curved_links.py": "b6957fa13d13f864dba4f7364e93a39905d062dbb3ee9b3f5191add81fff0732",
+    "src/CelerisLab/lbm/descriptors.py": "c9f025dfc14da43b989047b84cc22bd05d5cc92c9ec695c1da68fdfb788e3f2b",
+    "src/CelerisLab/lbm/equilibrium.py": "2fb2f2bb7ff326d23abce80fc2a4c0907b17740ea3e29fe6ae079c5044783826",
+    "src/CelerisLab/lbm/field.py": "0e727eacc3706add4d8d15c418e9b631050871f831698d3ac689dd5f2c701b9b",
+    "src/CelerisLab/lbm/hierarchy.py": "f234975074f6b6d289d573ff6ea6832feae0e27cb04f80b37fe81792b13a12e7",
+    "src/CelerisLab/lbm/initializers.py": "dacf9586963e862d6d0a98ef1a066036be6f91e9d423720661d8ef2c1fb2df6a",
+    "src/CelerisLab/lbm/kernels/blocks.cu": "2fa894b20a29f006911ac3e3aad0793cc4aaf9ccd7a67d01c6320a0836b1d8ef",
+    "src/CelerisLab/lbm/kernels/boundary/bounce_back.cuh": "aba3748c564ed91b161d268ef8b737b7c7be96a823063d201f3da684dd845233",
+    "src/CelerisLab/lbm/kernels/boundary/curved_boundary.cuh": "aeb16bfd483514fb03e15a2fed437d673c3b8b2fe68ab83edf4244ef593878ea",
+    "src/CelerisLab/lbm/kernels/boundary/esopull_curved_helpers.cuh": "b9b474bc4edaff05681c5443235472d04072d0d775fc0eb9c5967faef52b0258",
+    "src/CelerisLab/lbm/kernels/boundary/ibm_kernels.cuh": "e6909b547dc8d64cdd8a76a90660243a8f5fee535caa161b52b4dd8b7c29934b",
+    "src/CelerisLab/lbm/kernels/boundary/inlet/channel_stabilized.cuh": "efe077afcbb6b731256a987c2fd79a1dbc470c902f87070d60421ea6cb13d4e2",
+    "src/CelerisLab/lbm/kernels/boundary/inlet/common.cuh": "59b38a89439878e00de7de16b9c7b03feff443fcc5a421593f7d2536dcbb7ab8",
+    "src/CelerisLab/lbm/kernels/boundary/inlet/equilibrium.cuh": "f0964c811b2c41a5aaa1c6d99c490b06e850c634eb8749f1f05ef61f93c2de1d",
+    "src/CelerisLab/lbm/kernels/boundary/inlet/regularized.cuh": "943d4a571818bacb89c70bccfaac202691121e3e385803ce6309123a0a03ddaf",
+    "src/CelerisLab/lbm/kernels/boundary/inlet/zou_he_local.cuh": "e73e66b853b1e4e0e30d8f4185d832aa414ce670a1813ce2b5459f2887f71f89",
+    "src/CelerisLab/lbm/kernels/boundary/inlet_outlet.cuh": "de8b213a6da42c2e5e9f8966199fc66ea4541b85144a90f3f5dbad3c6ab63b97",
+    "src/CelerisLab/lbm/kernels/boundary/outlet/pressure_neq.cuh": "c8b2adc66d12fe602076ddc9693c5c208711fcaddb58aa9a31401a269f51c4c4",
+    "src/CelerisLab/lbm/kernels/boundary/planar_velocity.cuh": "501b2759d0e2e4280cb6fdd127d66a50fa89f6de48fc2cae9a7eafb11ae90bed",
+    "src/CelerisLab/lbm/kernels/config/config_grid.h": "1db3e030ebdc116cd7f02362d1fd077daac55e8f0b4329f6285948c224d3ca5d",
+    "src/CelerisLab/lbm/kernels/config/config_method.h": "4374e04a130f90f97d17cc20c40d2f9ea0ed0dd8603817d07f215d02266e247c",
+    "src/CelerisLab/lbm/kernels/config/config_objects.h": "db793af052c756561a190a3ff2cb93ee5fe28462ee4f97c6af80882b4dc8b072",
+    "src/CelerisLab/lbm/kernels/config/config_obs.h": "c11e2249359145b5c4cbaaf0487506fa77c1f4b82c7c169ddc574aa0babcd727",
+    "src/CelerisLab/lbm/kernels/config/config_physics.h": "8ac0990b168e1058588676cc05c80df8634cfd671dbc92a89246c3c372d83a4e",
+    "src/CelerisLab/lbm/kernels/config.h": "7d1124125e0b354558297934460e64c760fb68cdd7bf6d01a1238bb2eaab8c37",
+    "src/CelerisLab/lbm/kernels/core/descriptors.cuh": "8440bc14857da9007982650f828a23608eec9f7cc4bf772e8c8dc1239cb45356",
+    "src/CelerisLab/lbm/kernels/core/flags.cuh": "0025705f24710330d9d9195f713879bfb650a85a3449baef0e196b831af104ab",
+    "src/CelerisLab/lbm/kernels/core/layout.cuh": "1fbd820021c89ac3efcde24145996cb0147769e607e498e61f308ed3e0b13706",
+    "src/CelerisLab/lbm/kernels/core/obs.cuh": "b742b77c9f61e1f020fe9922c3838ae9d855e17b381154958b58f10a22938922",
+    "src/CelerisLab/lbm/kernels/core/params.cuh": "6e18bf730d936d940e3d85452f0714a2eaae31bf41002f9fd520ec5b372878dd",
+    "src/CelerisLab/lbm/kernels/core/precision.cuh": "849df14b4cce50af52399e4b8f3e997196f3ae5f1163f68e28df3e6b2507a8c4",
+    "src/CelerisLab/lbm/kernels/kernel_v2.cu": "c7bd3fb705b77261ba2441c52a24a7e6a7865c165cba088a1b75201c4d29c408",
+    "src/CelerisLab/lbm/kernels/operators/collision_mrt.cuh": "aee444eb60c47a3fd518dca975d1af05425a0818a0ec6c3fb3f6150f60c2cb44",
+    "src/CelerisLab/lbm/kernels/operators/collision_srt.cuh": "6b7ff842f392c8f853d2d1a97cc69bfe50f21283f73e659404da07cf6c95f5e1",
+    "src/CelerisLab/lbm/kernels/operators/collision_trt.cuh": "9725b9840b9ad8b4ec7be61ff5d06d3151e86d1a87585c13f2fbf934d0d2c5f6",
+    "src/CelerisLab/lbm/kernels/operators/forcing_guo.cuh": "969a430c1af03f32e3bd0df95d805e64f56334717c44dd4ba208ba503062a24b",
+    "src/CelerisLab/lbm/kernels/operators/helpers.cuh": "0b3a4d95d7d19f5fec09e5423b8f4cc99f8a226b541967425967330da5a051ec",
+    "src/CelerisLab/lbm/kernels/operators/macro.cuh": "2be20eb2b7d7f5522daaad7f4b57b1ff8cdf883efe4a67846a96783b2fe846a3",
+    "src/CelerisLab/lbm/kernels/operators/turbulence_smag.cuh": "ff7f000297ee0817e255ad9fcb6a13b647e3ddd5b206518d35f38d2895a9f501",
+    "src/CelerisLab/lbm/kernels/refinement.cu": "b45aad9c022cecb64f7e968ec3cae2d8385a3cc430ee409700c404a3f4dd8cf8",
+    "src/CelerisLab/lbm/kernels/step/aux_kernels.cu": "578a5b20a157241ee678547059abb1372dd5a589ad06bb2c15b987eb01ca1013",
+    "src/CelerisLab/lbm/kernels/step/block_channel.cu": "ee8a31398b8e5233d977c816e6b8e060c1eb2ec71320e594226dbca55593c0d0",
+    "src/CelerisLab/lbm/kernels/step/curved_double.cu": "bfa7a1df1e939020314d49755045f7a85fe4a19cc1c68dbe62653c66c5e7c2b3",
+    "src/CelerisLab/lbm/kernels/step/esopull_macro.cu": "c73116020ca217ede1e4de27c8269c31f9d91e33433ac8252a21e9b14187c50f",
+    "src/CelerisLab/lbm/kernels/step/esopull_sync.cu": "c0ff3e72900c3ab89a05dd03684e96d3b740cf1672d565727fc2af165933d087",
+    "src/CelerisLab/lbm/kernels/step/init_flow.cu": "305be3bb12fc0d039acec1701baabcf0e8a35e683c49c34f2af8ca9867d2b29c",
+    "src/CelerisLab/lbm/kernels/step/one_step_double.cu": "f9f8242b38e8e0cf8425b75e2b0215bc15363c62eaa0fb43c761b8e0931bfdc9",
+    "src/CelerisLab/lbm/kernels/step/one_step_esopull.cu": "aa223b2744ae6c9125673df311a96ce588032c1ba58bebb11fd33ecb990f8ed4",
+    "src/CelerisLab/lbm/kernels/streaming/esopull_semantic_helpers.cuh": "dd5a9091628b079efd97e53fd49e52d09dcee88529235db77a5648e02937af6e",
+    "src/CelerisLab/lbm/kernels/streaming/esopull_single_buffer.cuh": "77ed4ab9a0ebc5ff62a91a7c39066ef395391ae36e9040a69fb2bd8b1626dc79",
+    "src/CelerisLab/lbm/kernels/streaming/pull_double_buffer.cuh": "b65d741ed78b7b62f99531ba0c0ce1936663d8dc9640e7d7d16c707f1baf6549",
+    "src/CelerisLab/lbm/population_storage.py": "a6a86c29c52c892edfe4802fb773c387212e3f4cc86708d80e154986ddf88044",
+    "src/CelerisLab/lbm/raw_exchange.py": "8694202466eba848563e298656353b0f5db94e21a01fa05f51aad6dc2b9fd311",
+    "src/CelerisLab/lbm/refinement.py": "100d5e46ac894f6008b61364cdad5d37c2ca7ed949d8d71405572d6366526417",
+    "src/CelerisLab/lbm/refinement_stepper.py": "f2bb34fa90805ad21c3e7cf9f9a67d88d3ae3e249b4517d3898567b4bf9e364d",
+    "src/CelerisLab/lbm/refinement_tables.py": "fdd344024b89df3a83174dc810ca5e32db5e19eb8b1609f8f73b0821f56546b6",
+    "src/CelerisLab/lbm/stepper.py": "0dcb7cd7ae87e33e3dc67a38b6aa43042fbe673e37d9ad165342ee1d695933ce",
+    "src/CelerisLab/lbm/synchronized_remap.py": "7400fc4beaffd4649d6a9c524d38ce5fd6fa7eeb32eb5332e4030a4b6648f197",
+    "src/CelerisLab/simulation.py": "0a03904dc6de88667d7af699005a289adf8b09f0976cb0d099b041f4830a8f4c",
+    "tests/validation/re100_helpers.py": "289c6e01075ff361573f862b90cfd61e5980c9a8efcbc78aef69979b4004fd60",
+    "tests/validation/periodic.py": "a2212c0e2705126327a5019e9cafe64eeca398da1b75061ff23630c9843697a6",
+    "tests/validation/data/dfg_re100/reference_cycle.json": "073d1bda1629697efebad95667bfe80598e217d708dbffd7d5f6c6999379c55f",
+    "tests/validation/source_provenance.py": "6dfe8f2d819dba0e240ba761d31162a5033685d492d09db75eb2b8aaab3a5a4c"
+  }
+}

tests/validation/data/refined_dfg/source_manifest_window.json

@@ -0,0 +1,116 @@
+{
+  "commit": "a233c9032487a6006e950d1e327079b00466c410",
+  "production_origin_commit": "d9303234795fccf8591ba6274dfc544a84d956ff",
+  "observation_mode": "window",
+  "supported_collisions": [
+    "SRT",
+    "MRT"
+  ],
+  "sha256": {
+    "src/CelerisLab/__init__.py": "21cc469bd6e970a1f9ff5970c8eea4602065e41fc881854759ee565ee55c10c6",
+    "src/CelerisLab/body/__init__.py": "e4e0e8f0769a94428d5b02fac790443997556cfc5515a5f1ed1595d2c694d69d",
+    "src/CelerisLab/body/action_smoother.py": "b1a23ee6bbd4850b172cb8738c5c3546ee465a54189e6af9d8044f023d4687dc",
+    "src/CelerisLab/body/coupling/__init__.py": "c931f7eb318d48408c3e2febcc6917c88c8dcc1697cbf4329b1f340d94c35763",
+    "src/CelerisLab/body/coupling/soa_packer.py": "8ec26fd55ef74801bbd5ab6bd3d0a7dabc750cbd498aaac2ca4c61a0108ad3c9",
+    "src/CelerisLab/body/ddf_patch.py": "60fcfb7441e73bf290bcff41ca3a83da3e7ddad765d7247464e8f02e3043b4e2",
+    "src/CelerisLab/body/geometry/__init__.py": "ada88980ab672959aca9aa3a4c4aade3219524b2ab68110cd2a2f41e3e0ca90a",
+    "src/CelerisLab/body/geometry/base.py": "cceca14361ca2f34d34a3e43730e1a79b8ca81be00f0dd62e884db32b479419f",
+    "src/CelerisLab/body/geometry/circle.py": "7605ca7c21103f1d801ac73ab8dd18dffbdac425b2dee7b8858b283628a9e436",
+    "src/CelerisLab/body/jets.py": "4104c24af4651f4bb4f8d8fd36fb6d88f41c6e4469822f36e22e74c9ede31dac",
+    "src/CelerisLab/body/manager.py": "01b653d25bd9b4333379c8bfd12383596b686d2029f2e91b1c91b618540ec6b3",
+    "src/CelerisLab/body/objects.py": "8e63cee8faa9cc6243bcff428be628dfad73a02a9a3b8ece7d78c0a63406a08d",
+    "src/CelerisLab/body/preprocess/__init__.py": "dfd9d58bc07f626ab23a70053e31b88a6d9165d2e4c76a4e9f48abe674d3d2ec",
+    "src/CelerisLab/body/preprocess/flag_overlay.py": "831fc9e9736b758847759a085ff133eb49ec91318ace4fda76f85960138ee300",
+    "src/CelerisLab/body/registry.py": "fbb7327780adbdc6da15ebdb799980daf32df22ee974af0893d7409f177bbb68",
+    "src/CelerisLab/body/sync_plan.py": "3a15bfe9ebeb91482dbfc68850761ef39c872cc12d2efc795babd0c01e392323",
+    "src/CelerisLab/common/__init__.py": "d4dbde78bc97fccc9eabac9abae8f0f15c112c10ff8e9b45bd9401d4723df1ef",
+    "src/CelerisLab/common/_types.py": "aad2ff1a9033d4fd68783ad7123afb47201ddc58785cb6dc0246947037072a5f",
+    "src/CelerisLab/common/checkpoint.py": "b2f960aecd5ac27a3a7130b809ffad4709c44d522edef0359ab952654625410b",
+    "src/CelerisLab/common/pathline.py": "a1b6efcc0a076d0f1dc8d49a595779426d7d5626b00072a59bac710ff02ad046",
+    "src/CelerisLab/common/preprocess.py": "7fc78a08857cf8e98519500b094a1d896af189b6b15658ff9b388ed51bed4515",
+    "src/CelerisLab/common/render.py": "0e055476257eb2bcd11e733766e52195de02f8fad228f570a3511a7c831bdc29",
+    "src/CelerisLab/common/streakline/__init__.py": "13ff1cfa416fdd721a08a659e76b5a5a99b8d3201add10a83b7b0bc654bc1eec",
+    "src/CelerisLab/common/streakline/_config.py": "a9185ef68e9c7b5d5a1dc0b04a28341e2ad91d13c145a7c106636a878cf0a605",
+    "src/CelerisLab/common/streakline/_integrate.py": "3fbd4368b2c8d0e7fda40dc65aac5df02b085008b4c68acd83cb7ae45d33a13b",
+    "src/CelerisLab/common/streakline/_render.py": "236cd637414ee318c4d66f9aed32110d8dc42a100d65fde06172f7cd0ba62c65",
+    "src/CelerisLab/common/streakline/_streakline.py": "1a3e08daa706b5f83cdeb337e9a945a36ddf7583c40af1d0bb9d825ac954df37",
+    "src/CelerisLab/config.py": "2f1bba7515f5bc08cb3a8d140bcafe7fa0689e01b3b06494697bb2956a71530e",
+    "src/CelerisLab/configs/CONFIG.md": "41cafdbe525366c08aa9d0416ea026ccf0026ba7123b8a9b65f968a80d928e3a",
+    "src/CelerisLab/configs/config_body.json": "055f261f0828ca1d4f2bb694bd15b72bbcabe31d552326208f28e37f761adab1",
+    "src/CelerisLab/configs/config_lbm.json": "17e4311f1f250c6ece59177a8b6ff856b69a016c7c10267e774b030a0a390865",
+    "src/CelerisLab/configs/config_lbm_flume_mrt_zouhe.json": "8f593321db05a563a42931603a228d61cbcba06dfbe113262d5cd0ad93c0fe79",
+    "src/CelerisLab/configs/config_lbm_flume_trt_regularized.json": "f4bba8e6a4339fc3c4eb14f6e2f57c4b22993c73703c773d853edfbe34736e56",
+    "src/CelerisLab/configs/config_lbm_three_cylinder_triangle.json": "eeb8e7149dbe4aa6b3a9562802735aa950750dfbe2d972a0727e788573afa817",
+    "src/CelerisLab/cuda/__init__.py": "0d01dff9f0399c8baba822aab5ca0122a135f011008ed8a499ebd9f7e5b06aea",
+    "src/CelerisLab/cuda/compiler_v2.py": "c4cc3ec4445c8d75e6a253dd87963877117a722bc73b9d4811806fc6045d08c1",
+    "src/CelerisLab/cuda/context.py": "013dd4ec6d01d2fbd11afdfb103eac6a273bd9f798d663d761b851eeb27885e9",
+    "src/CelerisLab/cuda/obs_layout.py": "5ffb2c3aeeaaaaeb64f2cd8febfbcf4ad9919e5414020ad753cfade0683b21c2",
+    "src/CelerisLab/lbm/__init__.py": "7f60fb3aaae81501b510d08c2fedff44fcb05657260eb10b6f5cc91e7204306b",
+    "src/CelerisLab/lbm/block_stepper.py": "00b313daddc6e63d8ab8423c9190ca7e92ad291de20a2f6b80eb53e3f2d85ec7",
+    "src/CelerisLab/lbm/blocks.py": "1e644c5719993f418185ad4a10b6950a30636fc912befd9eab41ff48b4cf85cd",
+    "src/CelerisLab/lbm/channel.py": "40d0354e5dc97bcdb7e7b5c1e7f9303a109b7134d8a3af88b8ff4cd75ab5a095",
+    "src/CelerisLab/lbm/curved_links.py": "b6957fa13d13f864dba4f7364e93a39905d062dbb3ee9b3f5191add81fff0732",
+    "src/CelerisLab/lbm/descriptors.py": "c9f025dfc14da43b989047b84cc22bd05d5cc92c9ec695c1da68fdfb788e3f2b",
+    "src/CelerisLab/lbm/equilibrium.py": "2fb2f2bb7ff326d23abce80fc2a4c0907b17740ea3e29fe6ae079c5044783826",
+    "src/CelerisLab/lbm/field.py": "0e727eacc3706add4d8d15c418e9b631050871f831698d3ac689dd5f2c701b9b",
+    "src/CelerisLab/lbm/hierarchy.py": "f234975074f6b6d289d573ff6ea6832feae0e27cb04f80b37fe81792b13a12e7",
+    "src/CelerisLab/lbm/initializers.py": "dacf9586963e862d6d0a98ef1a066036be6f91e9d423720661d8ef2c1fb2df6a",
+    "src/CelerisLab/lbm/kernels/blocks.cu": "2fa894b20a29f006911ac3e3aad0793cc4aaf9ccd7a67d01c6320a0836b1d8ef",
+    "src/CelerisLab/lbm/kernels/boundary/bounce_back.cuh": "aba3748c564ed91b161d268ef8b737b7c7be96a823063d201f3da684dd845233",
+    "src/CelerisLab/lbm/kernels/boundary/curved_boundary.cuh": "aeb16bfd483514fb03e15a2fed437d673c3b8b2fe68ab83edf4244ef593878ea",
+    "src/CelerisLab/lbm/kernels/boundary/esopull_curved_helpers.cuh": "b9b474bc4edaff05681c5443235472d04072d0d775fc0eb9c5967faef52b0258",
+    "src/CelerisLab/lbm/kernels/boundary/ibm_kernels.cuh": "e6909b547dc8d64cdd8a76a90660243a8f5fee535caa161b52b4dd8b7c29934b",
+    "src/CelerisLab/lbm/kernels/boundary/inlet/channel_stabilized.cuh": "efe077afcbb6b731256a987c2fd79a1dbc470c902f87070d60421ea6cb13d4e2",
+    "src/CelerisLab/lbm/kernels/boundary/inlet/common.cuh": "59b38a89439878e00de7de16b9c7b03feff443fcc5a421593f7d2536dcbb7ab8",
+    "src/CelerisLab/lbm/kernels/boundary/inlet/equilibrium.cuh": "f0964c811b2c41a5aaa1c6d99c490b06e850c634eb8749f1f05ef61f93c2de1d",
+    "src/CelerisLab/lbm/kernels/boundary/inlet/regularized.cuh": "943d4a571818bacb89c70bccfaac202691121e3e385803ce6309123a0a03ddaf",
+    "src/CelerisLab/lbm/kernels/boundary/inlet/zou_he_local.cuh": "e73e66b853b1e4e0e30d8f4185d832aa414ce670a1813ce2b5459f2887f71f89",
+    "src/CelerisLab/lbm/kernels/boundary/inlet_outlet.cuh": "de8b213a6da42c2e5e9f8966199fc66ea4541b85144a90f3f5dbad3c6ab63b97",
+    "src/CelerisLab/lbm/kernels/boundary/outlet/pressure_neq.cuh": "c8b2adc66d12fe602076ddc9693c5c208711fcaddb58aa9a31401a269f51c4c4",
+    "src/CelerisLab/lbm/kernels/boundary/planar_velocity.cuh": "501b2759d0e2e4280cb6fdd127d66a50fa89f6de48fc2cae9a7eafb11ae90bed",
+    "src/CelerisLab/lbm/kernels/config.h": "7d1124125e0b354558297934460e64c760fb68cdd7bf6d01a1238bb2eaab8c37",
+    "src/CelerisLab/lbm/kernels/config/config_grid.h": "1db3e030ebdc116cd7f02362d1fd077daac55e8f0b4329f6285948c224d3ca5d",
+    "src/CelerisLab/lbm/kernels/config/config_method.h": "4374e04a130f90f97d17cc20c40d2f9ea0ed0dd8603817d07f215d02266e247c",
+    "src/CelerisLab/lbm/kernels/config/config_objects.h": "db793af052c756561a190a3ff2cb93ee5fe28462ee4f97c6af80882b4dc8b072",
+    "src/CelerisLab/lbm/kernels/config/config_obs.h": "c11e2249359145b5c4cbaaf0487506fa77c1f4b82c7c169ddc574aa0babcd727",
+    "src/CelerisLab/lbm/kernels/config/config_physics.h": "8ac0990b168e1058588676cc05c80df8634cfd671dbc92a89246c3c372d83a4e",
+    "src/CelerisLab/lbm/kernels/core/descriptors.cuh": "8440bc14857da9007982650f828a23608eec9f7cc4bf772e8c8dc1239cb45356",
+    "src/CelerisLab/lbm/kernels/core/flags.cuh": "0025705f24710330d9d9195f713879bfb650a85a3449baef0e196b831af104ab",
+    "src/CelerisLab/lbm/kernels/core/layout.cuh": "1fbd820021c89ac3efcde24145996cb0147769e607e498e61f308ed3e0b13706",
+    "src/CelerisLab/lbm/kernels/core/obs.cuh": "b742b77c9f61e1f020fe9922c3838ae9d855e17b381154958b58f10a22938922",
+    "src/CelerisLab/lbm/kernels/core/params.cuh": "6e18bf730d936d940e3d85452f0714a2eaae31bf41002f9fd520ec5b372878dd",
+    "src/CelerisLab/lbm/kernels/core/precision.cuh": "849df14b4cce50af52399e4b8f3e997196f3ae5f1163f68e28df3e6b2507a8c4",
+    "src/CelerisLab/lbm/kernels/kernel_v2.cu": "c7bd3fb705b77261ba2441c52a24a7e6a7865c165cba088a1b75201c4d29c408",
+    "src/CelerisLab/lbm/kernels/operators/collision_mrt.cuh": "aee444eb60c47a3fd518dca975d1af05425a0818a0ec6c3fb3f6150f60c2cb44",
+    "src/CelerisLab/lbm/kernels/operators/collision_srt.cuh": "6b7ff842f392c8f853d2d1a97cc69bfe50f21283f73e659404da07cf6c95f5e1",
+    "src/CelerisLab/lbm/kernels/operators/collision_trt.cuh": "9725b9840b9ad8b4ec7be61ff5d06d3151e86d1a87585c13f2fbf934d0d2c5f6",
+    "src/CelerisLab/lbm/kernels/operators/forcing_guo.cuh": "969a430c1af03f32e3bd0df95d805e64f56334717c44dd4ba208ba503062a24b",
+    "src/CelerisLab/lbm/kernels/operators/helpers.cuh": "0b3a4d95d7d19f5fec09e5423b8f4cc99f8a226b541967425967330da5a051ec",
+    "src/CelerisLab/lbm/kernels/operators/macro.cuh": "2be20eb2b7d7f5522daaad7f4b57b1ff8cdf883efe4a67846a96783b2fe846a3",
+    "src/CelerisLab/lbm/kernels/operators/turbulence_smag.cuh": "ff7f000297ee0817e255ad9fcb6a13b647e3ddd5b206518d35f38d2895a9f501",
+    "src/CelerisLab/lbm/kernels/refinement.cu": "b45aad9c022cecb64f7e968ec3cae2d8385a3cc430ee409700c404a3f4dd8cf8",
+    "src/CelerisLab/lbm/kernels/step/aux_kernels.cu": "578a5b20a157241ee678547059abb1372dd5a589ad06bb2c15b987eb01ca1013",
+    "src/CelerisLab/lbm/kernels/step/block_channel.cu": "ee8a31398b8e5233d977c816e6b8e060c1eb2ec71320e594226dbca55593c0d0",
+    "src/CelerisLab/lbm/kernels/step/curved_double.cu": "bfa7a1df1e939020314d49755045f7a85fe4a19cc1c68dbe62653c66c5e7c2b3",
+    "src/CelerisLab/lbm/kernels/step/esopull_macro.cu": "c73116020ca217ede1e4de27c8269c31f9d91e33433ac8252a21e9b14187c50f",
+    "src/CelerisLab/lbm/kernels/step/esopull_sync.cu": "c0ff3e72900c3ab89a05dd03684e96d3b740cf1672d565727fc2af165933d087",
+    "src/CelerisLab/lbm/kernels/step/init_flow.cu": "305be3bb12fc0d039acec1701baabcf0e8a35e683c49c34f2af8ca9867d2b29c",
+    "src/CelerisLab/lbm/kernels/step/one_step_double.cu": "f9f8242b38e8e0cf8425b75e2b0215bc15363c62eaa0fb43c761b8e0931bfdc9",
+    "src/CelerisLab/lbm/kernels/step/one_step_esopull.cu": "aa223b2744ae6c9125673df311a96ce588032c1ba58bebb11fd33ecb990f8ed4",
+    "src/CelerisLab/lbm/kernels/streaming/esopull_semantic_helpers.cuh": "dd5a9091628b079efd97e53fd49e52d09dcee88529235db77a5648e02937af6e",
+    "src/CelerisLab/lbm/kernels/streaming/esopull_single_buffer.cuh": "77ed4ab9a0ebc5ff62a91a7c39066ef395391ae36e9040a69fb2bd8b1626dc79",
+    "src/CelerisLab/lbm/kernels/streaming/pull_double_buffer.cuh": "b65d741ed78b7b62f99531ba0c0ce1936663d8dc9640e7d7d16c707f1baf6549",
+    "src/CelerisLab/lbm/population_storage.py": "a6a86c29c52c892edfe4802fb773c387212e3f4cc86708d80e154986ddf88044",
+    "src/CelerisLab/lbm/raw_exchange.py": "9031a196006a0f74dee4028e6f3bb673bc19ef7959f9f04dde9e01f93293aeb3",
+    "src/CelerisLab/lbm/refinement.py": "100d5e46ac894f6008b61364cdad5d37c2ca7ed949d8d71405572d6366526417",
+    "src/CelerisLab/lbm/refinement_stepper.py": "b6b835cf97f5aa045045bb4a47cfde34b89929d3872be410cb155727dfca2d33",
+    "src/CelerisLab/lbm/refinement_tables.py": "6eea593aaf791a2c644f1c74df0a41f6f5e62a2fce9feae76e19cf84b615607a",
+    "src/CelerisLab/lbm/stepper.py": "0dcb7cd7ae87e33e3dc67a38b6aa43042fbe673e37d9ad165342ee1d695933ce",
+    "src/CelerisLab/lbm/synchronized_remap.py": "7400fc4beaffd4649d6a9c524d38ce5fd6fa7eeb32eb5332e4030a4b6648f197",
+    "src/CelerisLab/simulation.py": "0a03904dc6de88667d7af699005a289adf8b09f0976cb0d099b041f4830a8f4c",
+    "tests/validation/data/dfg_re100/reference_cycle.json": "073d1bda1629697efebad95667bfe80598e217d708dbffd7d5f6c6999379c55f",
+    "tests/validation/periodic.py": "a2212c0e2705126327a5019e9cafe64eeca398da1b75061ff23630c9843697a6",
+    "tests/validation/re100_helpers.py": "289c6e01075ff361573f862b90cfd61e5980c9a8efcbc78aef69979b4004fd60",
+    "tests/validation/source_provenance.py": "6dfe8f2d819dba0e240ba761d31162a5033685d492d09db75eb2b8aaab3a5a4c"
+  }
+}

tests/validation/jet_cv_measurements.py

@@ -10,63 +10,13 @@ LINEAR = np.array([2.5, -1.5, 0.0])
 QUADRATIC = np.array([4.375, -5.25, 1.875])
 
 
-EVIDENCE_LIMIT = 32 * 1024**2
-
-
-def install_evidence(replacements, limit=None):
-    """Preflight retained bytes before replacing files; cap failure preserves old pair.
-
-    Staged compression bytes are temporary disk overhead, not retained evidence.
-    This guards budget failures, not a transactional crash-recovery filesystem.
-    """
-    temporary = set(replacements.values())
-    try:
-        directory = next(iter(replacements)).parent
-        retained = sum(
-            p.stat().st_size
-            for p in directory.iterdir()
-            if p.is_file() and p not in temporary
-        )
-        projected = retained - sum(p.stat().st_size for p in replacements if p.exists())
-        projected += sum(p.stat().st_size for p in temporary)
-        if limit is not None and projected > limit:
-            raise RuntimeError(
-                "Dense-root 32 MiB retained-evidence budget would be exceeded; last verified files preserved"
-            )
-        for target, staged in replacements.items():
-            staged.replace(target)
-    finally:
-        for p in temporary:
-            if p.exists():
-                p.unlink()
-
-
-def write_evidence_text(path, text, *, dense_root=False):
-    staged = path.with_name("." + path.name + ".pending")
-    staged.write_text(text)
-    install_evidence({path: staged}, EVIDENCE_LIMIT if dense_root else None)
-
-
 def sample_steps(c):
     """Same reference times at both scales; dense startup retained."""
-    if c.get("dense_root", False):
-        return np.arange(c["steps"] + 1, dtype=int)
-    return old_sample_steps(c)
-
-
-def old_sample_steps(c):
-    """Exact previous physical cadence, clipped to this campaign's endpoint."""
     s2 = c["scale"] ** 2
-    end = c.get("steps", 3200 * s2) // s2
-    reference = sorted(set(range(0, min(100, end) + 1)) | set(range(100, end + 1, 4)))
+    reference = sorted(set(range(0, 101)) | set(range(100, 3201, 4)))
     return np.array(reference, int) * s2
 
 
-def absolute_aperture_interval_mass(rows, blowing, suction):
-    """Sum absolute aperture NET masses for one root interval, not per-link flux."""
-    return float(abs(rows[blowing, 3].sum()) + abs(rows[suction, 3].sum()))
-
-
 def factors(scale):
     """Lref=one s=1 root length, Tref=one s=1 root time, rho_ref=1."""
     return dict(
@@ -155,7 +105,7 @@ def ray_geometry(fluid, origin, center, radius, h=0.5):
     )
 
 
-def reconstruct_aperture(rho, vel, geo, *, return_absolute=False):
+def reconstruct_aperture(rho, vel, geo):
     field = np.concatenate((rho[None], vel), axis=0)
     x, y, a, b = geo["ix"], geo["iy"], geo["a"], geo["b"]
     ray = (
@@ -165,24 +115,12 @@ def reconstruct_aperture(rho, vel, geo, *, return_absolute=False):
         + field[:, y + 1, x + 1] * a * b
     )
     estimates = []
-    absolute = []
     for coeff in (LINEAR, QUADRATIC):
         wall = np.einsum("cajk,k->caj", ray, coeff)
         un = np.einsum("caj,ajc->aj", wall[1:], geo["normal"])
         q = np.sum(un * geo["ds"], axis=1)
         mass = np.sum(wall[0] * un * geo["ds"], axis=1)
         estimates.append(np.stack((q, mass), axis=-1))
-        absolute.append(
-            np.stack(
-                (
-                    np.sum(abs(un) * geo["ds"], axis=1),
-                    np.sum(abs(wall[0] * un) * geo["ds"], axis=1),
-                ),
-                axis=-1,
-            )
-        )
-    if return_absolute:
-        return ray, np.array(estimates), np.array(absolute)
     return ray, np.array(estimates)
 
 
@@ -198,8 +136,6 @@ class MeasurementRecorder:
         self.cx = cx
         self.cy = cy
         self.records = []
-        self.sparse_rays = []
-        self.ray_steps = set(old_sample_steps(c).tolist())
         self.rays = [
             ray_geometry(fluid_u, (0.0, 0.0), np.array(c["center"]), c["diameter"] / 2),
             ray_geometry(leaves[1], origin_a, np.array(c["center"]), c["diameter"] / 2),
@@ -243,16 +179,8 @@ class MeasurementRecorder:
         record["cv_inventory"] = []
         record["cv_flux"] = []
         record["momentum_flux_terms"] = []
-        save_ray = not c.get("dense_root", False) or step in self.ray_steps
-        ray_record = dict(
-            sample_index=len(self.records),
-            arc_ray_values=[],
-            arc_estimates=[],
-            arc_absolute_estimates=[],
-        )
-        if not c.get("dense_root", False):
-            record["arc_ray_values"] = []
-            record["arc_estimates"] = []
+        record["arc_ray_values"] = []
+        record["arc_estimates"] = []
         record["nearest_mass_flux"] = [
             nearest_flux[k] for k in ("uniform_adapter", "amr_api")
         ]
@@ -275,18 +203,9 @@ class MeasurementRecorder:
                 rtol=2e-10,
             )
             record["momentum_flux_terms"].append(terms)
-            if save_ray:
-                ray, estimates, absolute = reconstruct_aperture(
-                    rho, vel, self.rays[idx], return_absolute=True
-                )
-                ray_record["arc_ray_values"].append(ray)
-                ray_record["arc_estimates"].append(estimates)
-                ray_record["arc_absolute_estimates"].append(absolute)
-                if not c.get("dense_root", False):
-                    record["arc_ray_values"].append(ray)
-                    record["arc_estimates"].append(estimates)
-        if c.get("dense_root", False) and save_ray:
-            self.sparse_rays.append(ray_record)
+            ray, estimates = reconstruct_aperture(rho, vel, self.rays[idx])
+            record["arc_ray_values"].append(ray)
+            record["arc_estimates"].append(estimates)
         self.records.append(record)
         if step // s**2 in (0, 100, 800, 1600, 3200) and step % s**2 == 0:
             fields = {}
@@ -304,11 +223,7 @@ class MeasurementRecorder:
                 fields[name + "_origin_reference"] = np.asarray(origin) / s
                 fields[name + "_h_reference"] = np.array(h / s)
             file = self.output / f"fields_s{s}_tref{step//s**2}.npz"
-            staged = file.with_name("." + file.name + ".pending.npz")
-            np.savez_compressed(staged, **fields)
-            install_evidence(
-                {file: staged}, EVIDENCE_LIMIT if c.get("dense_root", False) else None
-            )
+            np.savez_compressed(file, **fields)
             self.snapshots.append(
                 dict(
                     file=file.name,
@@ -321,32 +236,19 @@ class MeasurementRecorder:
     def flush(self):
         path = self.output / f"samples_s{self.c['scale']}.npz"
         arrays = {k: np.array([r[k] for r in self.records]) for k in self.records[0]}
-        if self.c.get("dense_root", False):
-            arrays["ray_sample_indices"] = np.array(
-                [r["sample_index"] for r in self.sparse_rays], dtype=int
-            )
-            for key in ("arc_ray_values", "arc_estimates", "arc_absolute_estimates"):
-                arrays[key] = np.array([r[key] for r in self.sparse_rays])
         for i, g in enumerate(self.rays):
             for k in ("normal", "ds", "positions"):
                 arrays[f"ray_{i}_{k}"] = g[k]
         arrays["ray_offsets_fine_cells"] = OFFSETS
         arrays["linear_coefficients"] = LINEAR
         arrays["quadratic_coefficients"] = QUADRATIC
-        staged = path.with_name("." + path.name + ".pending.npz")
-        np.savez_compressed(staged, **arrays)
+        np.savez_compressed(path, **arrays)
         metadata = dict(
-            schema=2 if self.c.get("dense_root", False) else 1,
+            schema=1,
             case=self.c,
-            dense_root=bool(self.c.get("dense_root", False)),
-            ray_sample_index_key=(
-                "ray_sample_indices" if self.c.get("dense_root", False) else None
-            ),
-            raw_absolute_throughput_definition="Cumulative sum of abs(blowing NET dm)+abs(suction NET dm) per root interval; unresolved within-aperture/substep reversals; not abs(window net).",
-            absolute_throughput_definition="Quadrature integral of abs(reconstructed u.n), abs(reconstructed rho*u.n); not abs(signed integral), not prescribed nominal, not exact wall truth.",
             reference_conversion=factors(self.c["scale"]),
             sample_file=path.name,
-            sample_sha256=hashlib.sha256(staged.read_bytes()).hexdigest(),
+            sample_sha256=hashlib.sha256(path.read_bytes()).hexdigest(),
             sample_count=len(self.records),
             snapshots=self.snapshots,
             level_stats_columns=[
@@ -378,10 +280,6 @@ class MeasurementRecorder:
             ],
             warning="Extrapolation and nearest-node values are estimator-dependent, not exact wall values. Fields/ray arrays use root-lattice units except explicitly named reference arrays.",
         )
-        meta_path = self.output / f"samples_s{self.c['scale']}.json"
-        meta_staged = meta_path.with_name("." + meta_path.name + ".pending")
-        meta_staged.write_text(json.dumps(metadata, indent=2) + "\n")
-        install_evidence(
-            {path: staged, meta_path: meta_staged},
-            EVIDENCE_LIMIT if self.c.get("dense_root", False) else None,
+        (self.output / f"samples_s{self.c['scale']}.json").write_text(
+            json.dumps(metadata, indent=2) + "\n"
         )

tests/validation/multigrid/test_exchange_window.py

@@ -0,0 +1,260 @@
+"""Optional observation-window exchange ledger: host contract (mocked driver) and GPU gate.
+
+Run CPU and GPU classes in separate processes; GPU callers select the device.
+ExchangeWindowGPU uses the SRT/unshifted Jet fixture; ExchangeWindowShiftedMRTGPU covers a
+shifted MRT, all-blocked, stationary-body case without Jets (no shifted-MRT Jet claim).
+GPU tolerance (conservative engineering tolerance, not a universal proof): the window ledger
+sums h*dt*raw per link over fine substeps, while the reference sums per-base interval ledgers
+over links and then over base steps. Both are FP64 reorderings of the same FP32-exact substep
+contributions, so the scale is the sum of the ACTUAL absolute substep contributions (read from
+last_raw after every fine hop), not of per-base totals that could hide substep cancellation:
+|window - reference| <= 4 * m * eps * sum|substep terms| + 1e-300, m = links * substeps.
+"""
+from contextlib import ExitStack
+import unittest
+from unittest.mock import patch
+
+import numpy as np
+import pycuda.driver as cuda
+
+from CelerisLab.lbm.refinement_stepper import FixedPatchStepper
+from test_refinement_blocked_base import mocked_backend
+from test_refinement_blocked_fine import BOUNDS, circle, config
+
+EPS = np.finfo(np.float64).eps
+
+
+def build(m, window=True, raw=True):
+    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)
+
+
+def same_bits(test, a, b):
+    """Identical NaN masks, no infinities, identical finite FP32 bits (incl. signed zero)."""
+    a, b = np.asarray(a), np.asarray(b)
+    test.assertEqual(a.dtype, b.dtype)
+    test.assertFalse(np.isinf(a).any() or np.isinf(b).any())
+    np.testing.assert_array_equal(np.isnan(a), np.isnan(b))
+    finite = np.isfinite(a)
+    np.testing.assert_array_equal(a[finite].view(np.uint32), b[finite].view(np.uint32))
+
+
+def accumulations(m):
+    return [c for c in m["calls"] if c[0] == "BlockAccumulateExchange"]
+
+
+class ExchangeWindowCPU(unittest.TestCase):
+    def test_constructor_validation(self):
+        with mocked_backend() as m:
+            with self.assertRaisesRegex(ValueError, "requires observe_raw_exchange"):
+                build(m, window=True, raw=False)
+            with self.assertRaisesRegex(ValueError, "must be bool"):
+                build(m, window=1)
+
+    def test_state_machine_and_launches(self):
+        with mocked_backend() as m, ExitStack() as stack:
+            stack.enter_context(patch.object(cuda, "memset_d8"))
+            stack.enter_context(patch.object(cuda, "memset_d8_async"))
+            stack.enter_context(patch.object(cuda, "memcpy_dtoh",
+                                             side_effect=lambda out, src: out.fill(0)))
+            backend = build(m)
+            b = backend._level_boundaries[1]
+            self.assertIsNotNone(b.window_integral)
+            with self.assertRaises(RuntimeError):
+                backend.start_exchange_window()  # not initialized
+            arrays = [np.full(level.shape, 1 / 9, np.float32) for level in backend._levels]
+            backend.initialize(*arrays)
+            with self.assertRaisesRegex(RuntimeError, "no valid exchange window"):
+                backend.read_exchange_window(0)
+            backend.step(1)
+            self.assertEqual(len(accumulations(m)), 2)  # closed window: interval ledger only
+            backend.start_exchange_window()
+            with self.assertRaisesRegex(RuntimeError, "at least one completed"):
+                backend.read_exchange_window(0)
+            m["calls"].clear()
+            backend.step(2)
+            calls = accumulations(m)
+            self.assertEqual(len(calls), 8)  # two fine hops x two ledgers x two base steps
+            self.assertEqual({id(c[1][1]) for c in calls},
+                             {id(b.interval_integral), id(b.window_integral)})
+            out = backend.read_exchange_window(0)
+            self.assertEqual((out["time_start"], out["time_end"], out["sample_count"]), (1., 3, 4))
+            self.assertEqual(backend.read_exchange_window(0)["time_end"], 3)  # non-destructive
+            backend.initialize(*arrays)  # invalidates
+            with self.assertRaisesRegex(RuntimeError, "no valid exchange window"):
+                backend.read_exchange_window(0)
+            backend.start_exchange_window()
+            with patch.object(backend, "_hop", side_effect=RuntimeError("injected")):
+                with self.assertRaises(RuntimeError):
+                    backend.step(1)
+            self.assertFalse(backend._window_valid)
+            with self.assertRaises(RuntimeError):
+                backend.read_exchange_window(0)  # failed status
+
+    def test_disabled_default_has_no_window_cost(self):
+        with mocked_backend() as m, ExitStack() as stack:
+            stack.enter_context(patch.object(cuda, "memset_d8_async"))
+            backend = build(m, window=False)
+            self.assertIsNone(backend._level_boundaries[1].window_integral)
+            backend.initialize(*[np.full(l.shape, 1 / 9, np.float32) for l in backend._levels])
+            backend.step(1)
+            self.assertEqual(len(accumulations(m)), 2)
+            with self.assertRaisesRegex(RuntimeError, "disabled"):
+                backend.start_exchange_window()
+
+
+class ExchangeWindowGPU(unittest.TestCase):
+    def test_window_matches_summed_base_reads_and_populations_unchanged(self):
+        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, fine_block=(16, 20), fine_objects=[rc()], jets=[jet()],
+                      observe_raw_exchange=True)
+            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)
+                windowed.step(2)
+                plain.step(2)
+                windowed.start_exchange_window()
+                b = windowed._level_boundaries[1]
+                count = len(b.tables["cell"])
+                factors = np.array([.25, .25, .125, .25])  # h*dt, h*h*dt, h*h for level-1 h=.5
+                scale = np.zeros(4)
+                substeps = 0
+                kernel = windowed._levels[1].kernel
+
+                def audited(src, dst, *args, **kw):
+                    nonlocal scale, substeps
+                    windowed._stream.synchronize()  # BFL and both accumulations are complete
+                    raw = np.empty((count, 4), np.float32)
+                    cuda.memcpy_dtoh(raw, b.last_raw)
+                    scale = scale + (np.abs(raw.astype(np.float64)) * factors).sum(axis=0)
+                    substeps += 1
+                    kernel(src, dst, *args, **kw)
+                windowed._levels[1].kernel = audited
+                keys = ("impulse", "angular_impulse", "mass_into_fluid")
+                summed = {part: [0., 0., 0., 0.] for part in ("total", "wall", "aperture")}
+                steps = 0
+                for speed in (0., .002, .002, -.001, 0., 0.):  # amplitude updates stay in-window
+                    windowed.set_jet_amplitude(0, speed)
+                    plain.set_jet_amplitude(0, speed)
+                    windowed.step(1)
+                    plain.step(1)
+                    steps += 1
+                    obs = windowed.read_raw_exchange(0)
+                    for part in summed:
+                        o = obs[part]
+                        summed[part] = [summed[part][0] + o["impulse"][0], summed[part][1] + o["impulse"][1],
+                                        summed[part][2] + o["angular_impulse"],
+                                        summed[part][3] + o["mass_into_fluid"]]
+                    for a, e in zip(windowed._download_raw().values, plain._download_raw().values):
+                        same_bits(self, a, e)
+                out = windowed.read_exchange_window(0)
+                self.assertEqual((out["time_end"] - out["time_start"], out["sample_count"]),
+                                 (steps, 2 * steps))
+                self.assertEqual(substeps, 2 * steps)
+                bound = 4 * (count * substeps) * EPS * scale + 1e-300
+                for part in summed:
+                    o = out[part]
+                    got = np.array([o["impulse"][0], o["impulse"][1], o["angular_impulse"],
+                                    o["mass_into_fluid"]])
+                    diff = np.abs(got - np.array(summed[part]))
+                    self.assertTrue((diff <= bound).all(), f"{part}: {diff} > {bound}")
+                    np.testing.assert_array_equal(o["force"], o["impulse"] / steps)
+                again = windowed.read_exchange_window(0)
+                np.testing.assert_array_equal(again["total"]["impulse"], out["total"]["impulse"])
+                windowed.initialize(*arrays)
+                with self.assertRaises(RuntimeError):
+                    windowed.read_exchange_window(0)
+
+
+def audit_substeps(stepper, b, count, record):
+    """Wrap the level-1 step kernel: after BFL and accumulation, add |substep terms| to record."""
+    factors = np.array([.25, .25, .125, .25])
+    kernel = stepper._levels[1].kernel
+
+    def audited(src, dst, *args, **kw):
+        stepper._stream.synchronize()
+        raw = np.empty((count, 4), np.float32)
+        cuda.memcpy_dtoh(raw, b.last_raw)
+        record["scale"] = record["scale"] + (np.abs(raw.astype(np.float64)) * factors).sum(axis=0)
+        record["substeps"] += 1
+        kernel(src, dst, *args, **kw)
+    stepper._levels[1].kernel = audited
+
+
+def components(obs):
+    o = obs["total"]
+    return np.array([o["impulse"][0], o["impulse"][1], o["angular_impulse"], o["mass_into_fluid"]])
+
+
+class ExchangeWindowShiftedMRTGPU(unittest.TestCase):
+    """Shifted MRT, all-blocked, stationary body, NO Jet (scoped separately from SRT/Jet)."""
+
+    def test_window_step0_reset_and_split_step(self):
+        from test_raw_exchange_gpu import BOUNDS as RB, circle as rc, state
+        from CelerisLab.config import LBMConfig
+        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, 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)
+                    run.step(2)
+                    if run is not plain:
+                        run.start_exchange_window()
+                b = windowed._level_boundaries[1]
+                count = len(b.tables["cell"])
+                record = dict(scale=np.zeros(4), substeps=0)
+                audit_substeps(windowed, b, count, record)
+                steps, summed = 5, np.zeros(4)
+                for _ in range(steps):
+                    windowed.step(1)
+                    plain.step(1)
+                    summed += components(windowed.read_raw_exchange(0))
+                    for a, e in zip(windowed._download_raw().values, plain._download_raw().values):
+                        same_bits(self, a, e)
+                split.step(steps)  # one call: same ledger arithmetic as five step(1) calls
+                for a, e in zip(windowed._download_raw().values, split._download_raw().values):
+                    same_bits(self, a, e)
+                rows = [np.empty((count, 4), np.float64) for _ in range(2)]
+                cuda.memcpy_dtoh(rows[0], b.window_integral)
+                cuda.memcpy_dtoh(rows[1], split._level_boundaries[1].window_integral)
+                np.testing.assert_array_equal(rows[0].view(np.uint64), rows[1].view(np.uint64))
+
+                out = windowed.read_exchange_window(0)
+                self.assertEqual(record["substeps"], 2 * steps)
+                self.assertEqual((out["time_end"] - out["time_start"], out["sample_count"]),
+                                 (steps, 2 * steps))
+                bound = 4 * (count * record["substeps"]) * EPS * record["scale"] + 1e-300
+                diff = np.abs(components(out) - summed)
+                self.assertTrue((diff <= bound).all(), f"{diff} > {bound}")
+                np.testing.assert_array_equal(split.read_exchange_window(0)["total"]["impulse"],
+                                              out["total"]["impulse"])
+
+                windowed.step(0)  # no interval: window and its duration unchanged
+                again = windowed.read_exchange_window(0)
+                self.assertEqual(again["time_end"], out["time_end"])
+                np.testing.assert_array_equal(again["total"]["impulse"], out["total"]["impulse"])
+
+                windowed.start_exchange_window()  # reset after a window
+                with self.assertRaisesRegex(RuntimeError, "at least one completed"):
+                    windowed.read_exchange_window(0)
+                record.update(scale=np.zeros(4), substeps=0)
+                windowed.step(1)
+                single = components(windowed.read_raw_exchange(0))
+                reset = windowed.read_exchange_window(0)
+                self.assertEqual(reset["time_end"] - reset["time_start"], 1)
+                bound = 4 * (count * record["substeps"]) * EPS * record["scale"] + 1e-300
+                self.assertTrue((np.abs(components(reset) - single) <= bound).all())
+
+
+if __name__ == "__main__": unittest.main()

tests/validation/multigrid/test_refinement_mrt_gpu.py

@@ -0,0 +1,95 @@
+"""MRT reuse GPU gates. Caller owns GPU admission (CUDA_VISIBLE_DEVICES).
+
+Run: python -m unittest discover -s tests/validation/multigrid -p test_refinement_mrt_gpu.py
+Parity compares the blocked MRT channel root with uniform MRT OneStep: the existing
+pressure-NEQ outlet branch differs by collision model, so SRT is never the reference.
+AMR checks are finite-evolution/split-invariance smoke tests, not benchmark validation.
+"""
+from dataclasses import replace
+import unittest
+import numpy as np
+
+from CelerisLab.config import LBMConfig
+from CelerisLab.cuda.context import CudaContext
+from CelerisLab.lbm.population_storage import _encode_physical
+from CelerisLab.lbm.refinement_stepper import FixedPatchStepper
+from test_channel_root_gpu import run_root, state
+from test_refinement_blocked_base import raw, exact
+
+
+class RefinementMRTGPU(unittest.TestCase):
+    def test_uniform_matched_mrt_root(self):
+        with CudaContext(0) as context:
+            for shifted in (False, True):
+                for profile in ("uniform", "parabolic"):
+                    cfg = LBMConfig(nx=24, ny=20, viscosity=.1, velocity=.025, collision="MRT",
+                                    ddf_shifting=shifted, inlet_profile=profile)
+                    initial = _encode_physical(state(cfg.ny, cfg.nx), shifted)
+                    expected = run_root(cfg, context, initial)
+                    self.assertTrue(np.isfinite(expected).all())
+                    for block in ((20, 24), (5, 6), (4, 1)):
+                        np.testing.assert_array_equal(run_root(cfg, context, initial, block), expected)
+
+    def test_amr_mrt_smoke(self):
+        from CelerisLab.body.geometry.circle import CircleGeometry
+        from CelerisLab.body.objects import SimObject
+        center = (24., 16.)
+        body = SimObject(0, geometry=CircleGeometry(*center, 2.15), center=center, radius=2.15)
+        with CudaContext(0) as context:
+            for boundary in ("periodic", "channel"):
+                for shifted in (False, True):
+                    cfg = LBMConfig(nx=48, ny=32, viscosity=.1, velocity=.015 if boundary == "channel" else 0.,
+                                    collision="MRT", ddf_shifting=shifted)
+                    reference = None
+                    for block in ((32, 48), (8, 8), (4, 1)):
+                        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()
+                            for a, mask in zip(raw.values, backend._state):
+                                self.assertTrue(np.isfinite(a[:, mask]).all())
+                            if reference is None: reference = raw.values
+                            else:
+                                for a, b in zip(reference, raw.values): np.testing.assert_array_equal(a, b)
+
+    def test_dense_versus_blocked_mrt_two_level(self):
+        """Unshifted MRT: all-dense PatchCoarseStep/PatchFineStep vs mixed/all-blocked storage."""
+        from test_refinement_blocked_base import BASE_BLOCKS, FINE_BLOCK
+        from test_refinement_blocked_fine import NX, NY, BOUNDS
+        import test_refinement_blocked_fine as fixture
+        cfg = LBMConfig(nx=NX, ny=NY, viscosity=.1, collision="MRT")
+        with CudaContext(0) as context:
+            results = {}
+            layouts = ((None, None), *((base, FINE_BLOCK) for base in (None, *BASE_BLOCKS)))
+            for base_block, fine_block in layouts:
+                with FixedPatchStepper(cfg, 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)
+            for layout in layouts[1:]:
+                for level, (a, b) in enumerate(zip(results[layout], results[None, None])):
+                    exact(self, a, b, f"layout={layout} level={level}")
+
+    def test_shifted_all_blocked_three_level_mrt_smoke(self):
+        """Per-level shear rate omega_l = 1/(3*2**l*nu + 1/2); finite ACTIVE leaves."""
+        cfg = LBMConfig(nx=16, ny=16, viscosity=.1, collision="MRT", ddf_shifting=True)
+        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, (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)
+                    checked.validate()  # omega is derived only by validate()
+                    self.assertEqual(checked.collision, "MRT")
+                    self.assertTrue(checked.ddf_shifting)
+                    self.assertTrue(np.isclose(checked.omega, 1 / (3 * 2**level * .1 + .5), rtol=1e-14))
+                backend.initialize(*[state(*level.shape[1:]) for level in backend._levels])
+                backend.step(4)
+                values = backend._download_raw().values
+                for a, mask in zip(values, backend._state):
+                    self.assertTrue(np.isfinite(a[:, mask]).all())
+
+
+if __name__ == "__main__": unittest.main()

tests/validation/run_jet_circle_cv.py

@@ -18,24 +18,11 @@ CY = np.array([0, 0, 0, 1, -1, 1, -1, -1, 1])
 W = np.array([4 / 9] + [1 / 9] * 4 + [1 / 36] * 4)
 
 
-def case(scale=1, re=100, steps=800, ddf_shifting=False):
+def case(scale=1, re=100, steps=800):
     u = 0.03 / scale
     d = 24.084 * scale
     return dict(
         scale=scale,
-        ddf_shifting=bool(ddf_shifting),
-        storage_variant=dict(
-            name="shifted_fp32" if ddf_shifting else "unshifted_fp32",
-            device_populations="g=f-w" if ddf_shifting else "f",
-            device_dtype="float32",
-            equilibrium_path=(
-                "compute_feq_from_momentum" if ddf_shifting else "compute_feq"
-            ),
-            initial_physical_seed="historical FP32-rounded equilibrium, identical in both variants",
-            downloaded_populations="physical f, decoded by public stepper.download()",
-            downloaded_dtype="float64" if ddf_shifting else "float32",
-            raw_exchange="physical reaction and mass; no host weight correction",
-        ),
         re_d=re,
         beta=0.1,
         nx=192 * scale,
@@ -239,7 +226,7 @@ def run_case(c, ctx, output, provenance, continuation=False):
         collision="SRT",
         streaming="double_buffer",
         store_precision="FP32",
-        ddf_shifting=c.get("ddf_shifting", False),
+        ddf_shifting=False,
         les_enabled=False,
     )
     fine_cfg = LBMConfig(
@@ -250,7 +237,7 @@ def run_case(c, ctx, output, provenance, continuation=False):
         collision="SRT",
         streaming="double_buffer",
         store_precision="FP32",
-        ddf_shifting=c.get("ddf_shifting", False),
+        ddf_shifting=False,
         les_enabled=False,
     )
     obj = SimObject(
@@ -363,10 +350,6 @@ def run_case(c, ctx, output, provenance, continuation=False):
             for name in ("uniform_adapter", "amr_api")
         }
 
-        from jet_cv_measurements import absolute_aperture_interval_mass
-
-        cumulative_absolute_aperture_mass = np.zeros(2)
-
         def capture(step):
             ug = uniform.download()
             levels = amr.download()
@@ -398,10 +381,6 @@ def run_case(c, ctx, output, provenance, continuation=False):
                         for name in ("uniform_adapter", "amr_api")
                     ]
                 )
-                if c.get("dense_root", False):
-                    recorder.records[-1][
-                        "raw_absolute_aperture_mass_cumulative"
-                    ] = cumulative_absolute_aperture_mass.copy()
             return data, error, aperture
 
         previous, _, previous_aperture = capture(0)
@@ -448,17 +427,7 @@ def run_case(c, ctx, output, provenance, continuation=False):
             np.testing.assert_allclose(
                 ar.sum(axis=0)[:2], obs["total"]["impulse"], atol=1e-13, rtol=1e-13
             )
-            np.testing.assert_allclose(
-                ar[:, 3].sum(), obs["total"]["mass_into_fluid"], atol=1e-13, rtol=1e-13
-            )
-            for backend_index, (name, rr) in enumerate(
-                [("uniform_adapter", ur), ("amr_api", ar)]
-            ):
-                cumulative_absolute_aperture_mass[
-                    backend_index
-                ] += absolute_aperture_interval_mass(
-                    rr, masks["blowing"], masks["suction"]
-                )
+            for name, rr in [("uniform_adapter", ur), ("amr_api", ar)]:
                 for k, mask in masks.items():
                     subtotal = rr[mask].sum(axis=0)
                     accumulated[name][k] += subtotal
@@ -550,12 +519,8 @@ def run_case(c, ctx, output, provenance, continuation=False):
                     physics_validation_pass=False,
                     precision="FP32 evolution, FP64 host diagnostics",
                 )
-                from jet_cv_measurements import write_evidence_text
-
-                write_evidence_text(
-                    output / f"re{c['re_d']}_s{c['scale']}.json",
-                    json.dumps(payload, indent=2) + "\n",
-                    dense_root=c.get("dense_root", False),
+                (output / f"re{c['re_d']}_s{c['scale']}.json").write_text(
+                    json.dumps(payload, indent=2) + "\n"
                 )
                 print(
                     json.dumps(
@@ -587,33 +552,16 @@ def main():
         action="store_true",
         help="Fixed measurement campaign: Re100, scale1/2, 3200 reference steps",
     )
-    p.add_argument(
-        "--dense-root",
-        action="store_true",
-        help="Bounded ref0..800 scalar diagnostics at every root interval; sparse rays/fields",
-    )
-    p.add_argument(
-        "--shifted-fp32",
-        action="store_true",
-        help="Opt-in shifted numerical-representation control for --dense-root; historical unshifted default unchanged",
-    )
     p.add_argument("--plan", action="store_true")
     a = p.parse_args()
-    if a.shifted_fp32 and not a.dense_root:
-        p.error("--shifted-fp32 requires the bounded --dense-root campaign")
     if a.steps <= 0 or a.steps > 3200:
         p.error("steps must be 1..3200")
     if a.re == 400 and not a.allow_re400:
         p.error("Re400 requires explicit conditional-campaign opt-in")
-    if a.dense_root:
-        if a.continuation or a.scale not in (1, 2) or a.re != 100 or a.steps != 800:
-            p.error(
-                "--dense-root requires --re 100 --scale 1/2 --steps 800, without --continuation"
-            )
     if a.continuation:
         if a.scale not in (1, 2) or a.re != 100 or a.steps != 3200:
             p.error("--continuation requires --re 100 --scale 1/2 --steps 3200")
-    c = case(a.scale, a.re, a.steps, ddf_shifting=a.shifted_fp32)
+    c = case(a.scale, a.re, a.steps)
     if a.continuation:
         from jet_cv_measurements import sample_steps, factors
 
@@ -627,28 +575,12 @@ def main():
             post_ready_budget_seconds=1200 if a.scale == 1 else 5400,
             runtime_estimate_seconds=[210, 500] if a.scale == 1 else [900, 3200],
         )
-    if a.dense_root:
-        from jet_cv_measurements import sample_steps, old_sample_steps, factors
-
-        c.update(
-            dense_root=True,
-            measurement_schema=2,
-            reference_conversion=factors(a.scale),
-            post_ready_budget_seconds=1200 if a.scale == 1 else 5400,
-            runtime_estimate_seconds=[60, 240] if a.scale == 1 else [600, 2400],
-        )
-        c.update(
-            sample_count=len(sample_steps(c)),
-            sparse_ray_sample_count=len(old_sample_steps(c)),
-            scalar_sample_dt_reference=1 / a.scale**2,
-            raw_interval_dt_reference=1 / a.scale**2,
-        )
     plan = dict(
         case=c,
         required_revision=PIN,
         required_free_device_mib=1024,
         estimated_peak_device_mib=128 * a.scale**2,
-        estimated_output_mib=32 if a.dense_root else (64 if a.continuation else 2),
+        estimated_output_mib=64 if a.continuation else 2,
         max_steps=c["steps"],
         gpu_launch_owner="root; this runner does not select or reserve a GPU",
     )
@@ -680,9 +612,7 @@ 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 or a.dense_root
-        )
+        run_case(c, ctx, a.output, provenance, continuation=a.continuation)
     print(
         json.dumps(
             dict(status="COMPLETE_DIAGNOSTIC_ONLY", physics_validation_pass=False)

tests/validation/run_refined_dfg.py

@@ -1,4 +1,4 @@
-"""Frozen matched SRT DFG comparison. GPU execution requires explicit root admission."""
+"""Frozen matched SRT/MRT DFG comparison. GPU execution requires explicit root admission."""
 
 import argparse
 import csv
@@ -39,20 +39,22 @@ def helpers(source):
     return h, periodic
 
 
-def config(h, diameter):
-    cfg, body, p = h.make_config(diameter, SPEED, "SRT", "neq_extrap", "zou_he_local")
+def config(h, diameter, collision="SRT"):
+    cfg, body, p = h.make_config(diameter, SPEED, collision, "neq_extrap", "zou_he_local")
     cfg["method"]["outlet"]["srt_neq_damp"] = 0.5
     return cfg, body, p
 
 
-def plan(h):
-    _, _, c = config(h, 40)
-    _, _, f = config(h, 80)
+def plan(h, collision="SRT", observation_mode="last_base_step"):
+    _, _, c = config(h, 40, collision)
+    _, _, f = config(h, 80, collision)
     x0, y0, x1, y1 = BOUNDS
     fine_active = 4 * (x1 - x0) * (y1 - y0)
     coarse_uncovered = c["nx"] * c["ny"] - fine_active // 4
     return dict(
         status="prepared_not_run",
+        collision=collision,
+        observation_mode=observation_mode,
         root_gpu_gates_required=True,
         root=c,
         uniform=f,
@@ -82,7 +84,11 @@ def plan(h):
         / (2 * f["nx"] * f["ny"]),
         cost_caveat="Geometry-only estimates include ghosts/solids; exclude block halos, transfers, observation and launch overhead. Not measured speedup.",
         sampling_offset="For origin (38,22), fine physical x=(j+75.5)*dx80 versus uniform i*dx80: half-fine-cell x offset. Fine physical y=(k+42.5)*dx80 equals uniform (l-.5)*dx80 at l=k+43: y samples align. Body, wall and source planes coincide.",
-        outlet="SRT pressure NEQ, all nine fneq damped by .5, no backflow clamp; noncanonical DFG outlet",
+        outlet=(
+            "SRT pressure NEQ, all nine fneq damped by .5, no backflow clamp; noncanonical DFG outlet"
+            if collision == "SRT" else
+            "MRT pressure NEQ, unknowns q2/6/8 extrapolated; no backflow clamp; noncanonical DFG outlet"
+        ),
         modes={"pilot_tstar": 5.0, "full_tstar": 300.0},
         observation_dense_tstar=0.02,
     )
@@ -120,7 +126,7 @@ def initial_state(shape, origin, spacing, height=164.0, speed=SPEED, root=False)
     return w * (1 + 3 * cu + 4.5 * cu * cu - 1.5 * velocity * velocity)
 
 
-def host_geometry(source, h):
+def host_geometry(source, h, collision="SRT"):
     """Production donor/transfer and complete cut-link checks, without CUDA."""
     from CelerisLab.config import load_lbm_config
     from CelerisLab.lbm.refinement import FixedPatch2D, prepare_fine_boundary
@@ -133,7 +139,7 @@ def host_geometry(source, h):
     from CelerisLab.body.geometry.circle import CircleGeometry
     from CelerisLab.body.objects import SimObject
 
-    cfgdict, _, _ = config(h, 40)
+    cfgdict, _, _ = config(h, 40, collision)
     with TemporaryDirectory() as tmp:
         p = Path(tmp) / "cfg.json"
         p.write_text(json.dumps(cfgdict))
@@ -151,14 +157,60 @@ def host_geometry(source, h):
     return cfg, pair, body, host
 
 
-def verify_source(source):
-    expected = json.loads((HERE / "data/refined_dfg/source_manifest.json").read_text())
+def verify_source(source, collision="SRT", observation_mode="last_base_step"):
+    if collision not in ("SRT", "MRT"):
+        raise ValueError("Matched collision must be SRT or MRT")
+    if observation_mode not in ("last_base_step", "window"):
+        raise ValueError("Unknown refined observation mode")
+    name = ("source_manifest_window.json" if observation_mode == "window" else
+            "source_manifest.json" if collision == "SRT" else "source_manifest_mrt.json")
+    expected = json.loads((HERE / "data/refined_dfg" / name).read_text())
     for name, digest in expected["sha256"].items():
         if hashlib.sha256((source / name).read_bytes()).hexdigest() != digest:
             raise RuntimeError("Source mismatch: " + name)
     return expected
 
 
+def advance_refined_interval(backend, n, deadline, observation_mode):
+    """Return the same root-unit interval impulse, never an already averaged force."""
+    impulse = np.zeros(2)
+    if observation_mode == "window":
+        begin = backend.step_count
+        check_deadline(deadline)
+        backend.start_exchange_window()
+        backend.step(n)
+        exchange = backend.read_exchange_window(0, level=1)
+        if (
+            exchange["units"] != "base_lattice_per_unit_depth"
+            or exchange["window"] != "window"
+            or exchange["sample_count"] != 2 * n
+            or exchange["time_start"] != begin
+            or exchange["time_end"] != begin + n
+            or exchange["dt"] != 0.5
+        ):
+            raise RuntimeError("Unexpected window force observation contract")
+        impulse += exchange["total"]["impulse"]
+        check_deadline(deadline)
+        return impulse
+    if observation_mode != "last_base_step":
+        raise ValueError("Unknown refined observation mode")
+    for _ in range(n):
+        check_deadline(deadline)
+        backend.step(1)
+        exchange = backend.read_raw_exchange(
+            0, level=1, window="last_base_step"
+        )
+        if (
+            exchange["units"] != "base_lattice_per_unit_depth"
+            or exchange["sample_count"] != 2
+            or exchange["time_end"] - exchange["time_start"] != 1
+        ):
+            raise RuntimeError("Unexpected force observation contract")
+        impulse += exchange["total"]["impulse"]
+        check_deadline(deadline)
+    return impulse
+
+
 def run_case(a, h, periodic, case):
     from CelerisLab.simulation import Simulation
     from CelerisLab.cuda.context import CudaContext
@@ -167,7 +219,7 @@ def run_case(a, h, periodic, case):
     from contextlib import ExitStack
 
     refined = case == "refined"
-    cfgdict, bodydict, p = config(h, 40 if refined else 80)
+    cfgdict, bodydict, p = config(h, 40 if refined else 80, a.collision)
     cap = 5.0 if a.mode == "pilot" else 300.0
     max_steps = math.ceil(cap * p["diameter"] / SPEED)
     rows = []
@@ -179,7 +231,7 @@ def run_case(a, h, periodic, case):
     dest.mkdir()
     with ExitStack() as stack:
         if refined:
-            cfg, pair, body, host = host_geometry(a.source, h)
+            cfg, pair, body, host = host_geometry(a.source, h, a.collision)
             context = stack.enter_context(CudaContext(0))
             backend = stack.enter_context(
                 FixedPatchStepper(
@@ -191,6 +243,7 @@ def run_case(a, h, periodic, case):
                     fine_block=FINE_BLOCK,
                     fine_objects=(body,),
                     observe_raw_exchange=True,
+                    **({"observe_window_exchange": True} if a.observation_mode == "window" else {}),
                 )
             )
             from CelerisLab.lbm.channel import channel_masks
@@ -240,20 +293,9 @@ def run_case(a, h, periodic, case):
                 impulse = np.zeros(2)
                 tick = time.monotonic()
                 if refined:
-                    for _ in range(n):
-                        check_deadline(a.deadline)
-                        backend.step(1)
-                        exchange = backend.read_raw_exchange(
-                            0, level=1, window="last_base_step"
-                        )
-                        if (
-                            exchange["units"] != "base_lattice_per_unit_depth"
-                            or exchange["sample_count"] != 2
-                            or exchange["time_end"] - exchange["time_start"] != 1
-                        ):
-                            raise RuntimeError("Unexpected force observation contract")
-                        impulse += exchange["total"]["impulse"]
-                        check_deadline(a.deadline)
+                    impulse = advance_refined_interval(
+                        backend, n, a.deadline, a.observation_mode
+                    )
                 else:
                     done = 0
                     while done < n:
@@ -336,6 +378,9 @@ def run_case(a, h, periodic, case):
                 if len(rows) == 1:
                     ready = dict(
                         case=case,
+                        collision=a.collision,
+                        observation_mode=a.observation_mode,
+                        **a.provenance,
                         mode=a.mode,
                         step=begin,
                         tstar=row["tstar"],
@@ -348,6 +393,9 @@ def run_case(a, h, periodic, case):
                     print("READY " + json.dumps(ready, allow_nan=False), flush=True)
     result = dict(
         case=case,
+        collision=a.collision,
+        observation_mode=a.observation_mode,
+        **a.provenance,
         mode=a.mode,
         history=rows,
         healthy=bool(healthy),
@@ -416,12 +464,16 @@ def check_deadline(deadline):
         )
 
 
-def verify_pilot(path, manifest):
+def verify_pilot(path, manifest, collision="SRT", observation_mode="last_base_step"):
     if path is None:
         raise ValueError(
             "Full comparison requires --pilot-results from a completed matched pilot"
         )
     previous = read_finite_json(path / "plan.json")
+    if previous.get("observation_mode") != observation_mode:
+        raise ValueError("Pilot observation mode differs")
+    if previous.get("collision") != collision:
+        raise ValueError("Pilot collision differs")
     if previous["source_manifest"] != manifest or previous["bounds_root_edge"] != list(
         BOUNDS
     ):
@@ -432,7 +484,9 @@ def verify_pilot(path, manifest):
     for case, diameter in (("uniform", 80), ("refined", 40)):
         summary = read_finite_json(path / case / "summary.json")
         if (
-            summary.get("case") != case
+            summary.get("observation_mode") != observation_mode
+            or summary.get("collision") != collision
+            or summary.get("case") != case
             or summary.get("mode") != "pilot"
             or summary.get("status") != "pilot_complete"
             or summary.get("healthy") is not True
@@ -483,7 +537,9 @@ def verify_pilot(path, manifest):
                 raise ValueError("Unhealthy pilot observations")
         ready = read_finite_json(path / case / "ready.json")
         if (
-            ready.get("case") != case
+            ready.get("observation_mode") != observation_mode
+            or ready.get("collision") != collision
+            or ready.get("case") != case
             or ready.get("mode") != "pilot"
             or ready.get("step") != interval
             or ready.get("runner_sha256") != runner_hash()
@@ -507,6 +563,9 @@ def verify_pilot(path, manifest):
 
 def main():
     parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--exchange-window", action="store_true",
+                        help="Opt in to refined window-ledger observations on the pinned candidate")
+    parser.add_argument("--collision", choices=["SRT", "MRT"], default="SRT")
     parser.add_argument("--source", type=Path, default=DEFAULT_SOURCE)
     parser.add_argument("--output", type=Path, required=True)
     parser.add_argument("--mode", choices=["plan", "pilot", "full"], default="plan")
@@ -519,9 +578,16 @@ def main():
     )
     a = parser.parse_args()
     a.source = a.source.resolve()
-    manifest = verify_source(a.source)
+    a.observation_mode = "window" if a.exchange_window else "last_base_step"
+    manifest = verify_source(a.source, a.collision, a.observation_mode)
     h, periodic = helpers(a.source)
-    calibration = verify_pilot(a.pilot_results, manifest) if a.mode == "full" else None
+    from source_provenance import checkout_provenance
+
+    a.provenance = checkout_provenance(a.source)
+    calibration = (
+        verify_pilot(a.pilot_results, manifest, a.collision, a.observation_mode)
+        if a.mode == "full" else None
+    )
     if a.mode != "plan" and (
         a.max_wall_seconds is None
         or not math.isfinite(a.max_wall_seconds)
@@ -536,16 +602,17 @@ def main():
     if a.output.exists() and any(a.output.iterdir()):
         raise ValueError("Output must be empty/new")
     a.output.mkdir(parents=True, exist_ok=True)
-    specification = plan(h)
+    specification = plan(h, a.collision, a.observation_mode)
     specification.update(
         source_manifest=manifest,
+        **a.provenance,
         gates=periodic.GATES,
         reference_gates=periodic.REFERENCE_GATES,
         runner_sha256=runner_hash(),
         wall_time_budget_seconds=budget,
         full_pilot_calibration_seconds=calibration,
     )
-    host_geometry(a.source, h)
+    host_geometry(a.source, h, a.collision)
     (a.output / "plan.json").write_text(json.dumps(specification, indent=2))
     if a.mode == "plan":
         return 0
@@ -561,6 +628,9 @@ def main():
         except TimeoutError as exc:
             failure = dict(
                 status="wall_time_limit",
+                collision=a.collision,
+                observation_mode=a.observation_mode,
+                **a.provenance,
                 case=case,
                 matched_case_pass=False,
                 error=str(exc),
@@ -574,6 +644,9 @@ def main():
         if not result["healthy"]:
             break
     comparison = dict(
+        collision=a.collision,
+        observation_mode=a.observation_mode,
+        **a.provenance,
         status="pilot_only" if a.mode == "pilot" else "incomplete",
         matched_case_pass=False,
     )

tests/validation/time_refined_dfg.py

@@ -0,0 +1,326 @@
+"""Measurement-only wall timing for the matched MRT refined DFG case. No numerical edits.
+
+Variants over EQUAL simulated duration (one refined base step = two uniform D80 steps):
+  A  refined: step(1) + read_raw_exchange per base step  -- runner-like force-read workload
+  B  refined: step(1) only, no host reads        -- timing-only, no force history
+  C  refined: one step(N) call                   -- timing-only; ledger keeps final base interval only
+  E  uniform: sim.run(<=100) + read_force                -- runner-like force-read workload
+  E' uniform: one sim.run(2N) call                -- timing-only; at most a whole-window force
+                                                    average (different summation order), no history
+A/E read forces but discard them: no retained force history, no interval aggregation/validation,
+and no macro/probe/CSV cost, so they are not full benchmark throughput.
+observe_raw_exchange stays True for A/B/C: B/C omit host reads only.
+Status lines: CONSTRUCTED = backend/simulation built (JIT done); WARM_READY = first warm-up and
+health check passed for that case.
+Parity compares A/B/C with each other and E/E' with each other, never uniform with refined.
+A-B and B-C are measured comparative differences, not exact causal decompositions.
+Each repeat starts from the same fresh initial state with the same warm-up; construction,
+compilation, parity and health checks are outside timed regions. The actual stepping stream is
+synchronized explicitly immediately before the timer starts and after the timed loop. Wall time
+(perf_counter) is the only timing source; no CUDA-event span is reported as kernel time. An outer
+--max-wall-seconds deadline is checked between variants and inside per-step loops.
+"""
+
+import argparse
+import json
+import math
+import os
+from pathlib import Path
+import statistics
+import sys
+import time
+from tempfile import TemporaryDirectory
+
+import numpy as np
+
+HERE = Path(__file__).resolve().parent
+sys.path.insert(0, str(HERE))
+import run_refined_dfg as r  # noqa: E402  (published runner: same setup functions)
+
+VARIANTS = {
+    "A": dict(case="refined", force_history_valid=False, force_reads=True,
+              force_note="runner-like force-read workload; values discarded, no aggregation"),
+    "B": dict(case="refined", force_history_valid=False, timing_only=True,
+              force_note="no host reads"),
+    "C": dict(case="refined", force_history_valid=False, timing_only=True,
+              force_note="raw ledger retains only the final base interval"),
+    "E": dict(case="uniform", force_history_valid=False, force_reads=True,
+              force_note="runner-like force-read workload; values discarded, no aggregation"),
+    "E'": dict(case="uniform", force_history_valid=False, timing_only=True,
+               force_note="at most a whole-window accumulated average; different summation order"),
+}
+
+
+def schedule(n_base, warmup_base):
+    """Steps per case for equal simulated duration (refined dt_root = 2 dt_uniform)."""
+    if n_base <= 0 or warmup_base < 0:
+        raise ValueError("n_base must be positive and warmup_base nonnegative")
+    return dict(refined=(warmup_base, n_base), uniform=(2 * warmup_base, 2 * n_base),
+                tstar_window=n_base * r.SPEED / 40.0)
+
+
+# Existing run_refined_dfg safety limits (rho in [.5, 1.5], sampled max Mach <= .2).
+RHO_LIMITS, MAX_MACH = (0.5, 1.5), 0.2
+
+
+def same_state(reference, values):
+    """Bitwise populations and identical NaN masks; returns a short description or None.
+
+    Matching NaN is accepted only as a storage mask (inactive/ghost/solid slots); physical
+    leaf health is checked separately. Any infinity in either state is a mismatch.
+    """
+    reference, values = list(reference), list(values)
+    if len(reference) != len(values):
+        return f"level count {len(reference)} != {len(values)}"
+    for level, (a, b) in enumerate(zip(reference, values)):
+        a, b = np.asarray(a), np.asarray(b)
+        if a.shape != b.shape:
+            return f"level {level}: shape {a.shape} != {b.shape}"
+        if np.isinf(a).any() or np.isinf(b).any():
+            return f"level {level}: infinite population"
+        if not np.array_equal(np.isnan(a), np.isnan(b)):
+            return f"level {level}: NaN mask differs"
+        finite = np.isfinite(a)
+        if a.dtype != b.dtype or not np.array_equal(a[finite].view(np.uint8), b[finite].view(np.uint8)):
+            return f"level {level}: finite bits differ"
+    return None
+
+
+def uniform_populations_finite(ddf, fluid):
+    """Simulation.get_ddf() may be flat (9*ny*nx); view it as (Q, cells) before masking."""
+    ddf, fluid = np.asarray(ddf), np.asarray(fluid, bool).ravel()
+    if fluid.size == 0 or ddf.size % fluid.size:
+        raise RuntimeError(f"DDF size {ddf.size} incompatible with {fluid.size} cells")
+    return bool(np.isfinite(ddf.reshape(ddf.size // fluid.size, fluid.size)[:, fluid]).all())
+
+
+def physical_health(fields):
+    """fields: (rho, ux, uy) arrays already restricted to physical leaves/fluid."""
+    if not fields or any(not np.isfinite(v).all() or v.size == 0 for f in fields for v in f):
+        raise RuntimeError("nonfinite or empty physical macroscopic field")
+    rho_min = float(min(f[0].min() for f in fields))
+    rho_max = float(max(f[0].max() for f in fields))
+    max_mach = float(max(np.sqrt(3 * (f[1] ** 2 + f[2] ** 2)).max() for f in fields))
+    if not (RHO_LIMITS[0] <= rho_min <= rho_max <= RHO_LIMITS[1] and max_mach <= MAX_MACH):
+        raise RuntimeError(f"health limits violated: rho [{rho_min}, {rho_max}], Mach {max_mach}")
+    return dict(rho_min=rho_min, rho_max=rho_max, max_mach=max_mach)
+
+
+def refined_health(backend, masks):
+    arrays = backend.download()  # physical FP64 populations; non-leaves NaN by contract
+    if any(not np.isfinite(g[:, m]).all() for g, m in zip(arrays, masks)):
+        raise RuntimeError("nonfinite physical leaf population")
+    return physical_health([tuple(v[m] for v in r.macros(g)) for g, m in zip(arrays, masks)])
+
+
+def uniform_health(sim, fluid):
+    m = sim.get_macroscopic()
+    return physical_health([tuple(np.asarray(m[k], np.float64).reshape(fluid.shape)[fluid]
+                                  for k in ("rho", "ux", "uy"))])
+
+
+def check_deadline(deadline):
+    if time.monotonic() >= deadline:
+        raise TimeoutError("timing harness exceeded --max-wall-seconds")
+
+
+def sync_refined(backend):
+    with backend.ctx.activate():
+        backend._stream.synchronize()
+
+
+def sync_uniform(sim):
+    with sim.ctx.activate():
+        sim.stream.synchronize()
+
+
+def refined_loop(backend, variant, n, deadline):
+    if variant == "C":
+        backend.step(n)
+        return
+    for i in range(n):
+        backend.step(1)
+        if variant == "A":
+            backend.read_raw_exchange(0, level=1, window="last_base_step")
+        if i % 100 == 99:
+            check_deadline(deadline)
+
+
+def uniform_loop(sim, variant, n, deadline):
+    if variant == "E'":
+        sim.run(n)
+        return
+    done = 0
+    while done < n:
+        batch = min(100, n - done)
+        sim.run(batch)
+        sim.read_force(0)
+        done += batch
+        check_deadline(deadline)
+
+
+def time_refined(a, h, sched):
+    from contextlib import ExitStack
+    from CelerisLab.cuda.context import CudaContext
+    from CelerisLab.lbm.refinement_stepper import FixedPatchStepper
+    from CelerisLab.lbm.channel import channel_masks
+    from CelerisLab.lbm.refinement import ACTIVE
+    from CelerisLab.lbm.descriptors import SOLID
+
+    warm, n = sched["refined"]
+    rows, reference = [], {}
+    with ExitStack() as stack:
+        cfg, pair, body, host = r.host_geometry(a.source, h, "MRT")
+        context = stack.enter_context(CudaContext(0))
+        backend = stack.enter_context(FixedPatchStepper(
+            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),
+                 r.initial_state(pair.fine_shape, pair.fine_origin, 0.5))
+        print(json.dumps(dict(status="CONSTRUCTED", case="refined")), flush=True)
+        for repeat in range(a.repeats):
+            for variant in ("A", "B", "C"):
+                check_deadline(a.deadline)
+                backend.initialize(*fresh)  # resets step count; same fresh state every time
+                refined_loop(backend, variant, warm, a.deadline)
+                warm_health = refined_health(backend, masks)  # untimed
+                if repeat == 0 and variant == "A":
+                    print(json.dumps(dict(status="WARM_READY", case="refined")), flush=True)
+                sync_refined(backend)
+                start = time.perf_counter()
+                refined_loop(backend, variant, n, a.deadline)
+                sync_refined(backend)
+                seconds = time.perf_counter() - start
+                end_health = refined_health(backend, masks)  # untimed
+                values = [np.array(v, copy=True) for v in backend._download_raw().values]
+                if not all(np.isfinite(v[:, m]).all() for v, m in zip(values, backend._state)):
+                    raise RuntimeError(f"refined {variant} repeat {repeat}: nonfinite leaf populations")
+                mismatch = None
+                if repeat == 0 and variant == "A":
+                    reference["A"] = values
+                else:
+                    mismatch = same_state(reference["A"], values)
+                rows.append(dict(variant=variant, repeat=repeat, seconds=seconds,
+                                 base_steps=n, step_count=backend.step_count,
+                                 health_after_warmup=warm_health, health_after_window=end_health,
+                                 parity_vs_A=mismatch or "identical", **VARIANTS[variant]))
+                if mismatch:
+                    raise RuntimeError(f"refined {variant} repeat {repeat}: {mismatch}")
+    return rows
+
+
+def time_uniform(a, h, sched):
+    from CelerisLab.simulation import Simulation
+    from CelerisLab.lbm.descriptors import FLUID
+
+    warm, n = sched["uniform"]
+    rows, reference = [], {}
+    cfgdict, bodydict, params = r.config(h, 80, "MRT")
+    with TemporaryDirectory() as tmp:
+        tmp = Path(tmp)
+        (tmp / "lbm.json").write_text(json.dumps(cfgdict))
+        (tmp / "body.json").write_text(json.dumps(bodydict))
+        with Simulation(str(tmp / "lbm.json"), str(tmp / "body.json"), device_id=0) as sim:
+            sim.initialize()
+            sim.snapshot()  # fresh state; restore keeps clock and streaming parity
+            fluid = (sim.get_flags() & FLUID) != 0
+            ny, nx = params["ny"], params["nx"]
+            fluid = fluid.reshape(ny, nx)
+            print(json.dumps(dict(status="CONSTRUCTED", case="uniform")), flush=True)
+            for repeat in range(a.repeats):
+                for variant in ("E", "E'"):
+                    check_deadline(a.deadline)
+                    sim.restore()
+                    if sim.stepper.step_count % 2:
+                        raise RuntimeError("uniform restore parity changed between variants")
+                    uniform_loop(sim, variant, warm, a.deadline)
+                    warm_health = uniform_health(sim, fluid)  # untimed
+                    if repeat == 0 and variant == "E":
+                        print(json.dumps(dict(status="WARM_READY", case="uniform")), flush=True)
+                    sync_uniform(sim)
+                    start = time.perf_counter()
+                    uniform_loop(sim, variant, n, a.deadline)
+                    sync_uniform(sim)
+                    seconds = time.perf_counter() - start
+                    end_health = uniform_health(sim, fluid)  # untimed
+                    values = [sim.get_ddf()]
+                    if not uniform_populations_finite(values[0], fluid):
+                        raise RuntimeError(f"uniform {variant} repeat {repeat}: nonfinite fluid populations")
+                    mismatch = None
+                    if repeat == 0 and variant == "E":
+                        reference["E"] = values
+                    else:
+                        mismatch = same_state(reference["E"], values)
+                    rows.append(dict(variant=variant, repeat=repeat, seconds=seconds,
+                                     uniform_steps=n, parity_vs_E=mismatch or "identical",
+                                     health_after_warmup=warm_health, health_after_window=end_health,
+                                     **VARIANTS[variant]))
+                    if mismatch:
+                        raise RuntimeError(f"uniform {variant} repeat {repeat}: {mismatch}")
+    return rows
+
+
+def summarize(rows, tstar_window):
+    summary = {}
+    for variant in VARIANTS:
+        seconds = [row["seconds"] for row in rows if row["variant"] == variant]
+        if seconds:
+            median = statistics.median(seconds)
+            summary[variant] = dict(median_seconds=median, seconds=seconds,
+                                    seconds_per_tstar=median / tstar_window,
+                                    **VARIANTS[variant])
+    if {"A", "B", "C"} <= summary.keys():
+        summary["A_minus_B_seconds"] = summary["A"]["median_seconds"] - summary["B"]["median_seconds"]
+        summary["B_minus_C_seconds"] = summary["B"]["median_seconds"] - summary["C"]["median_seconds"]
+    if {"A", "E"} <= summary.keys():
+        summary["A_over_E"] = summary["A"]["median_seconds"] / summary["E"]["median_seconds"]
+    if {"C", "E'"} <= summary.keys():
+        summary["C_over_Eprime"] = summary["C"]["median_seconds"] / summary["E'"]["median_seconds"]
+    summary["interpretation"] = ("Measured comparative differences under GPU variability; "
+                                 "not exact causal decompositions or performance claims.")
+    return summary
+
+
+def main():
+    p = argparse.ArgumentParser(description=__doc__)
+    p.add_argument("--source", type=Path, default=r.DEFAULT_SOURCE)
+    p.add_argument("--output", type=Path, required=True)
+    p.add_argument("--base-steps", type=int, default=2000)
+    p.add_argument("--warmup-base-steps", type=int, default=200)
+    p.add_argument("--repeats", type=int, default=3)
+    p.add_argument("--max-wall-seconds", type=float, help="Required outer timeout for GPU runs")
+    p.add_argument("--plan", action="store_true")
+    a = p.parse_args()
+    if a.repeats < 1 or a.base_steps > 20000:
+        p.error("repeats must be >= 1 and base-steps <= 20000")
+    a.source = a.source.resolve()
+    manifest = r.verify_source(a.source, "MRT")
+    sched = schedule(a.base_steps, a.warmup_base_steps)
+    plan = dict(collision="MRT", manifest_commit=manifest["commit"], schedule=sched,
+                repeats=a.repeats, variants=VARIANTS, timer="time.perf_counter wall",
+                excluded="construction/JIT, warm-up, state download and parity checks")
+    if a.plan:
+        print(json.dumps(plan, indent=2))
+        return 0
+    if a.max_wall_seconds is None or not math.isfinite(a.max_wall_seconds) or a.max_wall_seconds <= 0:
+        p.error("GPU timing requires a finite positive --max-wall-seconds")
+    a.deadline = time.monotonic() + a.max_wall_seconds
+    if not os.environ.get("CUDA_VISIBLE_DEVICES") or "," in os.environ["CUDA_VISIBLE_DEVICES"]:
+        p.error("Root must select exactly one verified idle GPU via CUDA_VISIBLE_DEVICES")
+    if a.output.exists() and any(a.output.iterdir()):
+        p.error("Output must be empty/new")
+    a.output.mkdir(parents=True, exist_ok=True)
+    h, _ = r.helpers(a.source)
+    rows = time_uniform(a, h, sched) + time_refined(a, h, sched)
+    result = dict(plan=plan, rows=rows, summary=summarize(rows, sched["tstar_window"]))
+    if not all(math.isfinite(row["seconds"]) for row in rows):
+        raise RuntimeError("Nonfinite timing")
+    (a.output / "timing.json").write_text(json.dumps(result, indent=2))
+    print(json.dumps(result["summary"], indent=2), flush=True)
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

tests/validation/time_window_dfg.py

@@ -0,0 +1,183 @@
+"""Measurement-only A-versus-window timing for the matched shifted-MRT refined DFG setup.
+
+Candidate source (window ledger) is recorded by checkout revision, dirty flag and file hashes;
+the frozen refined-DFG manifests/results are NOT used or modified (they pin pre-window source).
+Equal simulated duration, same fresh state and warm-up per variant, one observation interval =
+80 root base steps (the runner's dense-phase interval):
+  A   per base step: step(1) + read_raw_exchange, summed per interval (current runner workload)
+  W   per interval: start_exchange_window(); step(80); read_exchange_window()
+  W1  per interval: start_exchange_window(); 80 x step(1); read_exchange_window()
+All variants keep observe_raw_exchange=True; the backend is built once with
+observe_window_exchange=True (a closed window adds no launch). Populations (bitwise) and
+per-interval total impulses are compared OUTSIDE timing; the impulse tolerance is a
+conservative engineering tolerance (see INTERVAL_RTOL), the rigorous substep-scaled comparison
+being the capability test in tests/validation/multigrid/test_exchange_window.py.
+Wall time (perf_counter) with explicit stream synchronization is the only timing source.
+"""
+
+import argparse
+import hashlib
+import json
+import math
+import os
+from pathlib import Path
+import statistics
+import sys
+import time
+
+import numpy as np
+
+HERE = Path(__file__).resolve().parent
+sys.path.insert(0, str(HERE))
+import time_refined_dfg as t  # noqa: E402  (reuses run_refined_dfg setup via t.r)
+
+r = t.r
+INTERVAL = 80
+VARIANTS = ("A", "W", "W1")
+# Interval impulses differ only by FP64 summation order (per link over time vs per step over
+# links). Engineering tolerance: 1e-9 of the interval's summed absolute per-step totals.
+INTERVAL_RTOL = 1e-9
+CANDIDATE_FILES = ("src/CelerisLab/lbm/refinement_stepper.py", "src/CelerisLab/lbm/raw_exchange.py",
+                   "src/CelerisLab/lbm/refinement_tables.py", "src/CelerisLab/lbm/kernels/blocks.cu")
+
+
+def components(obs):
+    o = obs["total"]
+    return np.array([o["impulse"][0], o["impulse"][1], o["angular_impulse"], o["mass_into_fluid"]])
+
+
+def candidate_identity(source):
+    from source_provenance import checkout_provenance
+    return dict(checkout_provenance(source),
+                sha256={f: hashlib.sha256((source / f).read_bytes()).hexdigest()
+                        for f in CANDIDATE_FILES})
+
+
+def run_variant(backend, variant, n_intervals, deadline):
+    """Returns per-interval totals and, for A, per-interval summed |per-step totals|."""
+    totals, scales = [], []
+    for _ in range(n_intervals):
+        if variant == "A":
+            impulse, scale = np.zeros(4), np.zeros(4)
+            for _ in range(INTERVAL):
+                backend.step(1)
+                c = components(backend.read_raw_exchange(0, level=1, window="last_base_step"))
+                impulse += c
+                scale += np.abs(c)
+            totals.append(impulse)
+            scales.append(scale)
+        else:
+            backend.start_exchange_window()
+            if variant == "W":
+                backend.step(INTERVAL)
+            else:
+                for _ in range(INTERVAL):
+                    backend.step(1)
+            totals.append(components(backend.read_exchange_window(0)))
+        t.check_deadline(deadline)
+    return np.array(totals), (np.array(scales) if scales else None)
+
+
+def compare_intervals(reference, scales, totals):
+    """Max violation ratio; <= 1 passes. Scale includes a floor so zero cancels cannot divide."""
+    bound = INTERVAL_RTOL * scales + 1e-300
+    return float(np.max(np.abs(totals - reference) / bound))
+
+
+def main():
+    p = argparse.ArgumentParser(description=__doc__)
+    p.add_argument("--source", type=Path, default=r.DEFAULT_SOURCE)
+    p.add_argument("--output", type=Path, required=True)
+    p.add_argument("--base-steps", type=int, default=2000)
+    p.add_argument("--warmup-base-steps", type=int, default=200)
+    p.add_argument("--repeats", type=int, default=3)
+    p.add_argument("--max-wall-seconds", type=float)
+    p.add_argument("--plan", action="store_true")
+    a = p.parse_args()
+    if a.base_steps <= 0 or a.base_steps % INTERVAL or a.base_steps > 20000 or a.repeats < 1:
+        p.error(f"base-steps must be a positive multiple of {INTERVAL} (<= 20000); repeats >= 1")
+    a.source = a.source.resolve()
+    identity = candidate_identity(a.source)
+    plan = dict(collision="MRT", shifted_storage="as run_refined_dfg MRT config",
+                candidate_source=identity, frozen_manifests_used=False, interval_base_steps=INTERVAL,
+                base_steps=a.base_steps, warmup_base_steps=a.warmup_base_steps, repeats=a.repeats,
+                variants=VARIANTS, tstar_window=a.base_steps * r.SPEED / 40.0,
+                timer="time.perf_counter wall with explicit stream sync",
+                excluded="construction/JIT, warm-up, health, downloads, parity/impulse comparison")
+    if a.plan:
+        print(json.dumps(plan, indent=2))
+        return 0
+    if a.max_wall_seconds is None or not math.isfinite(a.max_wall_seconds) or a.max_wall_seconds <= 0:
+        p.error("GPU timing requires a finite positive --max-wall-seconds")
+    if not os.environ.get("CUDA_VISIBLE_DEVICES") or "," in os.environ["CUDA_VISIBLE_DEVICES"]:
+        p.error("Root must select exactly one verified idle GPU via CUDA_VISIBLE_DEVICES")
+    if a.output.exists() and any(a.output.iterdir()):
+        p.error("Output must be empty/new")
+    a.output.mkdir(parents=True, exist_ok=True)
+    deadline = time.monotonic() + a.max_wall_seconds
+    h, _ = r.helpers(a.source)
+
+    from contextlib import ExitStack
+    from CelerisLab.cuda.context import CudaContext
+    from CelerisLab.lbm.refinement_stepper import FixedPatchStepper
+    from CelerisLab.lbm.channel import channel_masks
+    from CelerisLab.lbm.refinement import ACTIVE
+    from CelerisLab.lbm.descriptors import SOLID
+
+    rows, ref = [], {}
+    n_intervals = a.base_steps // INTERVAL
+    with ExitStack() as stack:
+        cfg, pair, body, host = r.host_geometry(a.source, h, "MRT")
+        context = stack.enter_context(CudaContext(0))
+        backend = stack.enter_context(FixedPatchStepper(
+            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)]
+        fresh = (r.initial_state(pair.coarse_shape, (0.0, 0.0), 1.0, root=True),
+                 r.initial_state(pair.fine_shape, pair.fine_origin, 0.5))
+        print(json.dumps(dict(status="CONSTRUCTED", case="refined")), flush=True)
+        for repeat in range(a.repeats):
+            for variant in VARIANTS:
+                t.check_deadline(deadline)
+                backend.initialize(*fresh)
+                for _ in range(a.warmup_base_steps):
+                    backend.step(1)
+                warm = t.refined_health(backend, masks)
+                if repeat == 0 and variant == "A":
+                    print(json.dumps(dict(status="WARM_READY", case="refined")), flush=True)
+                t.sync_refined(backend)
+                start = time.perf_counter()
+                totals, scales = run_variant(backend, variant, n_intervals, deadline)
+                t.sync_refined(backend)
+                seconds = time.perf_counter() - start
+                end = t.refined_health(backend, masks)
+                values = [np.array(v, copy=True) for v in backend._download_raw().values]
+                if variant == "A" and repeat == 0:
+                    ref.update(values=values, totals=totals, scales=scales)
+                    parity, ratio = "identical", 0.0
+                else:
+                    parity = t.same_state(ref["values"], values) or "identical"
+                    ratio = compare_intervals(ref["totals"], ref["scales"], totals)
+                rows.append(dict(variant=variant, repeat=repeat, seconds=seconds,
+                                 intervals=n_intervals, population_parity_vs_A=parity,
+                                 interval_impulse_violation_ratio=ratio,
+                                 health_after_warmup=warm, health_after_window=end))
+                if parity != "identical" or not ratio <= 1:
+                    raise RuntimeError(f"{variant} repeat {repeat}: parity {parity}, impulse ratio {ratio}")
+    summary = {v: dict(median_seconds=statistics.median(x["seconds"] for x in rows if x["variant"] == v),
+                       seconds=[x["seconds"] for x in rows if x["variant"] == v]) for v in VARIANTS}
+    summary.update(W_over_A=summary["W"]["median_seconds"] / summary["A"]["median_seconds"],
+                   W1_over_A=summary["W1"]["median_seconds"] / summary["A"]["median_seconds"],
+                   interpretation="Measured comparative timings under GPU variability; W also "
+                                  "batches step calls, W1 isolates the read pattern. Not a "
+                                  "benchmark-throughput or causal claim.")
+    (a.output / "window_timing.json").write_text(json.dumps(dict(plan=plan, rows=rows, summary=summary),
+                                                           indent=2))
+    print(json.dumps(summary, indent=2), flush=True)
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())