diff --git a/tests/validation/multigrid/spacetime_gpu.cu b/tests/validation/multigrid/spacetime_gpu.cu
new file mode 100644
index 0000000..6b6de1c
--- /dev/null
+++ b/tests/validation/multigrid/spacetime_gpu.cu
@@ -0,0 +1,146 @@
+// Test-only arrival-f space-time coupling. No production scheduler changes.
+#include <stdint.h>
+#include <math.h>
+#include "config.h"
+#include "core/precision.cuh"
+#include "core/descriptors.cuh"
+#include "core/layout.cuh"
+#include "operators/macro.cuh"
+#include "operators/collision_srt.cuh"
+#include "streaming/pull_double_buffer.cuh"
+
+#if DIM != 2 || NQ != 9 || COLLISION_MODEL != 0 || STORE_PRECISION != 0 || USE_DDF_SHIFTING || USE_LES || STREAMING_MODEL != 0
+#error "Space-time experiment requires unshifted FP32 D2Q9 SRT double buffers without LES"
+#endif
+
+__device__ __forceinline__ void collide(float* f, float omega) {
+    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, omega);
+}
+
+__device__ __forceinline__ void health(float value, unsigned int* counts) {
+    if (!isfinite(value)) atomicAdd(counts + 4, 1u);
+    if (value <= 0.0f) atomicAdd(counts + 5, 1u);
+}
+
+extern "C" __global__ void CoarseStart(
+    const float* coarse, const float* fine, float* out, float omega,
+    unsigned int x0, unsigned int y0, unsigned int x1, unsigned int y1,
+    unsigned int* counts)
+{
+    unsigned int x, y; unsigned long k;
+    index_from_thread(x, y, k);
+    if (x >= NX || y >= NY) return;
+    const bool covered = x >= x0 && x < x1 && y >= y0 && y < y1;
+    float f[NQ];
+    for (unsigned int q = 0; q < NQ; ++q) {
+        const unsigned long s = q * 4ul * TOTAL_CELLS + 2ul * y * (2u * NX) + 2u * x;
+        f[q] = covered ? 0.25f * (fine[s] + fine[s+1] + fine[s+2u*NX] + fine[s+2u*NX+1])
+                       : coarse[index_f(k, q)];
+    }
+    collide(f, omega);
+    for (int q = 0; q < NQ; ++q) if (!covered) health(f[q], counts);
+    stream_pull_store(k, f, out);
+}
+
+extern "C" __global__ void Collide(
+    const float* in, float* out, const unsigned char* active, float omega,
+    unsigned int* counts)
+{
+    unsigned int x, y; unsigned long k;
+    index_from_thread(x, y, k);
+    if (x >= NX || y >= NY) return;
+    float f[NQ];
+    if (active == nullptr || active[k]) {
+        for (int q = 0; q < NQ; ++q) f[q] = in[index_f(k, q)];
+        collide(f, omega);
+        if (active != nullptr) for (int q = 0; q < NQ; ++q) health(f[q], counts);
+    } else {
+        for (int q = 0; q < NQ; ++q) f[q] = NAN;
+    }
+    stream_pull_store(k, f, out);
+}
+
+extern "C" __global__ void Pull(
+    const float* in, float* out, const unsigned char* active, unsigned int* counts)
+{
+    unsigned int x, y; unsigned long k;
+    index_from_thread(x, y, k);
+    if (x >= NX || y >= NY) return;
+    float f[NQ];
+    if (active == nullptr || active[k]) {
+        unsigned long j[NQ];
+        compute_neighbors(x, y, j);
+        stream_pull_load(k, f, in, j);
+        if (active != nullptr) for (int q = 0; q < NQ; ++q) health(f[q], counts);
+    } else {
+        for (int q = 0; q < NQ; ++q) f[q] = NAN;
+    }
+    stream_pull_store(k, f, out);
+}
+
+extern "C" __global__ void Interpolate(
+    const float* g0, const float* gp, const uint32_t* samples,
+    const uint32_t* origins, const int* sizes, const double* weights,
+    double* pre, float* h, unsigned int groups, unsigned int frozen,
+    unsigned int* counts)
+{
+    unsigned int group = blockIdx.x * blockDim.x + threadIdx.x;
+    if (group >= groups) return;
+    const int size = sizes[group];
+    for (int child = 0; child < 4; ++child) {
+        double value = frozen ? (double)g0[origins[group]] : 0.0;
+        if (!frozen) for (int j = 0; j < size; ++j) {
+            const float* source = j < size/2 ? g0 : gp;
+            value += weights[(4u*group+child)*10u+j] * (double)source[samples[10u*group+j]];
+        }
+        const float stored = __double2float_rn(value);
+        pre[4u*group+child] = value;
+        h[4u*group+child] = stored;
+        if (!isfinite(value)) atomicAdd(counts, 1u);
+        if (value <= 0.0) atomicAdd(counts+1, 1u);
+        if (!isfinite(stored)) atomicAdd(counts+2, 1u);
+        if (stored <= 0.0f) atomicAdd(counts+3, 1u);
+    }
+}
+
+extern "C" __global__ void Inject(
+    float* fine, const float* h, const uint32_t* rows, unsigned int count,
+    unsigned int omit_first)
+{
+    unsigned int row = blockIdx.x * blockDim.x + threadIdx.x;
+    if (row >= count || (omit_first && row == 0)) return;
+    fine[rows[2u*row]] = h[rows[2u*row+1]];
+}
+
+extern "C" __global__ void Register(
+    const float* predicted, const float* fine, const float* h,
+    const uint32_t* targets, const uint32_t* sources, const int* times,
+    double* registers, unsigned int count, int substep, int mutation)
+{
+    unsigned int row = blockIdx.x * blockDim.x + threadIdx.x;
+    if (row >= count) return;
+    double value = substep == 0 ? -(double)predicted[targets[row]] : registers[row];
+    for (int j = 0; j < 4; ++j) {
+        const unsigned int slot = 4u*row+j;
+        const int when = times[slot];
+        if (when != substep && !(substep == 0 && when == -1)) continue;
+        if ((mutation == 1 && when == -1) || (mutation == 2 && when == 0)
+                || (mutation == 3 && when == 1)) continue;
+        value += 0.25 * (double)(when == -1 ? h[sources[slot]] : fine[sources[slot]]);
+    }
+    registers[row] = value;
+}
+
+extern "C" __global__ void ApplyRegisters(
+    float* predicted, const double* registers, const uint32_t* targets,
+    unsigned int count, unsigned int* counts)
+{
+    unsigned int row = blockIdx.x * blockDim.x + threadIdx.x;
+    if (row >= count) return;
+    const unsigned int slot = targets[row];
+    predicted[slot] = __double2float_rn((double)predicted[slot] + registers[row]);
+    health(predicted[slot], counts);
+}
diff --git a/tests/validation/multigrid/spacetime_gpu.py b/tests/validation/multigrid/spacetime_gpu.py
new file mode 100644
index 0000000..ece4025
--- /dev/null
+++ b/tests/validation/multigrid/spacetime_gpu.py
@@ -0,0 +1,308 @@
+"""Test-only arrival-f GPU owner; CUDA compilation/execution requires admission."""
+from dataclasses import replace
+import hashlib
+from numbers import Integral
+from pathlib import Path
+
+import numpy as np
+import spacetime_reference as ref
+
+PINS = {"spacetime_reference.py": "1fe13242b9a41062a1227db4a6f9b695165aefab7f4c0b3b2adff0c9023552bf",
+        "cpu_reference.py": "bba2a06a9f6517b32079ffff15e4060d8c3b7ed35f16ca134e1877226c9e4e1e",
+        "../taylor_green/oracle.py": "83e4d7425d45538278e9fb181af450a36358bd9a5ed6be57ff611dd9ae5eb35b"}
+
+
+def check_sources():
+    for name, expected in PINS.items():
+        if hashlib.sha256((Path(__file__).parent/name).read_bytes()).hexdigest() != expected:
+            raise RuntimeError(f"reference source changed: {name}")
+
+
+def validate_config(config):
+    cfg = replace(config)
+    cfg.validate()
+    if (cfg.nx != cfg.ny or not 8 <= cfg.nx <= 64 or cfg.nz != 1
+            or cfg.collision != "SRT" or not cfg.is_d2q9
+            or cfg.streaming != "double_buffer" or cfg.store_precision != "FP32"
+            or cfg.data_type != "FP32" or cfg.ddf_shifting or cfg.les_enabled
+            or cfg.velocity != 0 or cfg.wall_slot_width or cfg.wall_normal_velocity):
+        raise ValueError("experiment requires periodic unforced square N8..64, unshifted FP32 D2Q9/SRT")
+    fine = replace(cfg, nx=2*cfg.nx, ny=2*cfg.ny, viscosity=2*cfg.viscosity)
+    fine.validate()
+    return cfg, fine
+
+
+def pack_plan(plan):
+    """Pack reviewed coordinate events; no production transfer builder imports."""
+    n, nf = plan.n, plan.nf
+    groups, receivers = plan.groups, plan.receivers
+    data = dict(samples=np.zeros((len(groups), 10), np.uint32),
+                origins=np.empty(len(groups), np.uint32), sizes=np.empty(len(groups), np.int32),
+                weights=np.zeros((len(groups), 4, 10), np.float64),
+                targets=(receivers[:, 0]*n*n+receivers[:, 1]).astype(np.uint32))
+    lookup = {tuple(row): i for i, row in enumerate(receivers)}
+    packets = [[] for _ in receivers]; incoming = [[], []]
+    for i, g in enumerate(groups):
+        size = len(g.samples)
+        if size not in (6, 10):
+            raise ValueError("unexpected interpolation stencil")
+        data['samples'][i, :size] = g.q*n*n+g.samples
+        data['origins'][i] = g.q*n*n+g.source
+        data['sizes'][i] = size
+        data['weights'][i, :, :size] = g.weights
+        for k, event in enumerate(g.events):
+            if event is None:
+                packets[lookup[(g.q, g.receiver)]].append((-1, 4*i+k))
+            else:
+                m, slot = event
+                incoming[m].append((g.q*nf*nf+slot, 4*i+k))
+    for m, outgoing in enumerate(plan.outgoing):
+        for q, source, target in outgoing:
+            packets[lookup[(int(q), int(target))]].append((m, int(q*nf*nf+source)))
+    if any(len(row) != 4 or len(set(row)) != 4 for row in packets):
+        raise ValueError("each receiver needs four distinct quarter-packets")
+    if len({packet for row in packets for packet in row}) != 4*len(receivers):
+        raise ValueError("packet has multiple receiver owners")
+    data['times'] = np.array([[t for t, _ in row] for row in packets], np.int32)
+    data['sources'] = np.array([[s for _, s in row] for row in packets], np.uint32)
+    for m, rows in enumerate(incoming):
+        data[f'incoming{m}'] = np.array(rows, np.uint32)
+        if len({slot for slot, _ in rows}) != len(rows):
+            raise ValueError("duplicate incoming writer")
+    return {key: np.ascontiguousarray(value) for key, value in data.items()}
+
+
+def table_hash(data):
+    digest = hashlib.sha256()
+    for key, value in sorted(data.items()):
+        digest.update(f"{key}:{value.dtype}:{value.shape}".encode()+value.tobytes())
+    return digest.hexdigest()
+
+
+def rounding_ledger(plan, tables, before, after, g0, predicted, fine_phases, pre, h, registers):
+    """Audit actual stored FP32 phases; never call a missing packet 'rounding'."""
+    maps, n = plan.maps, plan.n
+    f64 = lambda a: np.asarray(a, dtype=np.float64)
+    collision = (f64(g0[:, maps.coarse_active])-f64(before.coarse_f[:, maps.coarse_active])).sum(axis=1)
+    for f, g in fine_phases:
+        collision += .25*(f64(g[:, maps.fine_active])-f64(f[:, maps.fine_active])).sum(axis=1)
+    np.testing.assert_array_equal(h, pre.astype(np.float32))
+    origins = f64(g0).ravel()[tables['origins']]
+    np.testing.assert_allclose(pre.mean(axis=1), origins, rtol=0, atol=2e-14)
+    group = np.bincount(tables['origins']//(n*n), weights=f64(h).mean(axis=1)-origins, minlength=9)
+    packet = np.empty(tables['times'].shape, np.float64)
+    for when, source in ((-1, h), (0, fine_phases[0][1]), (1, fine_phases[1][1])):
+        mask = tables['times'] == when
+        packet[mask] = f64(source).ravel()[tables['sources'][mask]]
+    p = f64(predicted).ravel()[tables['targets']]
+    ideal = -p + .25*packet.sum(axis=1)
+    reg_error = registers-ideal
+    bound = 16*np.finfo(float).eps*(abs(p)+.25*abs(packet).sum(axis=1))
+    if np.any(abs(reg_error) > bound):
+        raise AssertionError("receiver packet register mismatch (not roundoff)")
+    actual = after.coarse_f.ravel()[tables['targets']]
+    np.testing.assert_array_equal(actual, (p+registers).astype(np.float32))
+    q = tables['targets']//(n*n)
+    store_error = np.bincount(q, weights=f64(actual)-(p+registers), minlength=9)
+    register_error = np.bincount(q, weights=reg_error, minlength=9)
+    # directional() must sum converted arrays, not accumulate in FP32.
+    total = lambda s: ref.directional(ref.cpu.State(f64(s.coarse_f), f64(s.fine_f)), maps)
+    raw = total(after)-total(before)-collision
+    unexplained = raw-group-store_error-register_error
+    limit = float(64*np.finfo(float).eps*total(before).sum())
+    np.testing.assert_allclose(unexplained, 0, rtol=0, atol=limit)
+    return dict(raw=raw.tolist(), physical_collision=collision.tolist(),
+                child_rounding=group.tolist(), receiver_rounding=store_error.tolist(),
+                register_rounding=register_error.tolist(), unexplained=unexplained.tolist(),
+                unexplained_bound=limit, minimum_pre=float(pre.min()), minimum_stored=float(h.min()))
+
+
+class SpaceTimeGPU:
+    """Caller-owned context, one stream, explicit synchronized arrival state."""
+    def __init__(self, config, bounds, *, context, mode='spacetime', identity=False):
+        check_sources()
+        self.configs = validate_config(config)
+        if mode not in ('spacetime', 'frozen') or type(identity) is not bool:
+            raise ValueError("unsupported interpolation or debug collision mode")
+        self.bounds = tuple(bounds)
+        if len(self.bounds) != 4 or any(isinstance(v, bool) or not isinstance(v, Integral) for v in self.bounds):
+            raise ValueError("rectangle bounds must be four integers")
+        self.plan = ref.Plan(config.nx, self.bounds)
+        self.tables = pack_plan(self.plan)
+        self.ctx, self.mode, self.identity = context, mode, identity
+        self.allocations, self.modules, self.build_metadata = [], [], []
+        self.status, self.step_count, self.device_bytes = 'new', 0, 0
+        import pycuda.driver as cuda
+        from CelerisLab.cuda import compiler_v2 as compiler
+        root = Path(__file__).resolve().parents[3]
+        if Path(compiler._KERNEL_DIR).resolve() != root/'src/CelerisLab/lbm/kernels':
+            raise RuntimeError("wrong production kernel checkout")
+        self.cuda = cuda
+        with context.activate():
+            self.stream = cuda.Stream()
+            try:
+                source = Path(__file__).with_suffix('.cu')
+                for cfg in self.configs:
+                    build = compiler.generate_config(cfg)
+                    try:
+                        Path(build, source.name).write_bytes(source.read_bytes())
+                        ptx = compiler.compile_kernel(context.sm_arch, source=source.name, build_dir=build)
+                        self.modules.append(compiler.load_module(ptx))
+                        import json
+                        metadata = json.loads(Path(ptx+'.json').read_text())
+                        metadata['test_entry_sha256'] = hashlib.sha256(source.read_bytes()).hexdigest()
+                        self.build_metadata.append(metadata)
+                    finally:
+                        compiler.cleanup_build_dir(build)
+                self.device_tables = {key: self.upload(value) for key, value in self.tables.items()}
+                self.active = self.upload(self.plan.maps.fine_active.astype(np.uint8))
+                self.coarse = [self.allocate(9*config.nx**2*4) for _ in range(4)]
+                self.fine = [self.allocate(9*(2*config.nx)**2*4) for _ in range(3)]
+                self.pre = self.allocate(len(self.plan.groups)*4*8)
+                self.h = self.allocate(len(self.plan.groups)*4*4)
+                self.registers = self.allocate(len(self.plan.receivers)*8)
+                self.counts = self.allocate(6*4)
+            except BaseException:
+                try:
+                    self.close()
+                except Exception:
+                    pass
+                raise
+
+    def allocate(self, size):
+        pointer = self.cuda.mem_alloc(size)
+        self.allocations.append(pointer); self.device_bytes += size
+        return pointer
+
+    def upload(self, array):
+        pointer = self.allocate(array.nbytes)
+        self.cuda.memcpy_htod(pointer, array)
+        return pointer
+
+    def read(self, pointer, shape, dtype=np.float32):
+        self.stream.synchronize()
+        out = np.empty(shape, dtype)
+        self.cuda.memcpy_dtoh(out, pointer)
+        return out
+
+    def launch(self, level, name, *args, count=None):
+        cfg = self.configs[level]
+        nt = cfg.threads_per_block
+        grid = ((cfg.nx+nt-1)//nt, cfg.ny) if count is None else ((count+nt-1)//nt, 1)
+        self.modules[level].get_function(name)(*args, block=(nt,1,1), grid=grid, stream=self.stream)
+
+    def initialize(self, state):
+        if (self.status not in ('new', 'ready') or state.phase != 'arrival_f'
+                or isinstance(state.tick, bool) or not isinstance(state.tick, Integral) or state.tick != 0):
+            raise ValueError("initial synchronized arrival state with tick=0 required")
+        arrays = [np.array(a, dtype=np.float32, copy=True, order='C') for a in (state.coarse_f,state.fine_f)]
+        for a, mask in zip(arrays, (self.plan.maps.coarse_active,self.plan.maps.fine_active)):
+            if a.shape != (9,*mask.shape) or not np.isfinite(a[:,mask]).all() or (a[:,mask] <= 0).any():
+                raise ValueError("positive finite leaf populations and exact shapes required")
+            a[:,~mask] = np.nan
+        with self.ctx.activate():
+            try:
+                self.cuda.memcpy_htod(self.coarse[0], arrays[0])
+                self.cuda.memcpy_htod(self.fine[0], arrays[1])
+            except BaseException:
+                self.status = 'failed'
+                raise
+        self.status, self.step_count = 'ready', 0
+
+    def download(self):
+        if self.status != 'ready':
+            raise RuntimeError(f"GPU owner is {self.status}")
+        try:
+            with self.ctx.activate():
+                arrays = [self.read(self.coarse[0], (9,self.plan.n,self.plan.n)),
+                          self.read(self.fine[0], (9,self.plan.nf,self.plan.nf))]
+        except BaseException:
+            self.status = 'failed'
+            raise
+        for a, mask in zip(arrays, (self.plan.maps.coarse_active,self.plan.maps.fine_active)):
+            a[:,~mask] = np.nan
+        return ref.cpu.State(*arrays, tick=2*self.step_count)
+
+    def step(self, *, mutation=0):
+        """One audited interval; mutation 1/2/3 drops packets, 4 omits a ghost."""
+        if self.status != 'ready' or type(mutation) is not int or mutation not in range(5):
+            raise RuntimeError("ready owner and known test mutation required")
+        before = self.download()
+        n, nf, t = self.plan.n, self.plan.nf, self.device_tables
+        ng, nr = len(self.plan.groups), len(self.plan.receivers)
+        u32, null = np.uint32, np.uintp(0)
+        omega = [np.float32(0 if self.identity else cfg.omega) for cfg in self.configs]
+        with self.ctx.activate():
+            try:
+                self.status = 'advancing'
+                self.cuda.memset_d32_async(self.counts, 0, 6, self.stream)
+                fc, g0, predicted, gp = self.coarse
+                self.launch(0, 'CoarseStart', fc,self.fine[0],g0,omega[0],
+                            *(u32(v) for v in self.bounds),self.counts)
+                self.launch(0, 'Pull', g0,predicted,null,self.counts)
+                self.launch(0, 'Collide', predicted,gp,null,omega[0],self.counts)
+                self.launch(0, 'Interpolate', g0,gp,t['samples'],t['origins'],t['sizes'],t['weights'],
+                            self.pre,self.h,u32(ng),u32(self.mode=='frozen'),self.counts,count=ng)
+                coarse_g = self.read(g0,(9,n,n)); prediction = self.read(predicted,(9,n,n))
+                fine_phases = []
+                for m in (0,1):
+                    f, out, g = self.fine
+                    old_f = self.read(f,(9,nf,nf))
+                    self.launch(1, 'Collide', f,g,self.active,omega[1],self.counts)
+                    fine_phases.append((old_f,self.read(g,(9,nf,nf))))
+                    self.launch(0, 'Register', predicted,g,self.h,t['targets'],t['sources'],t['times'],
+                                self.registers,u32(nr),np.int32(m),np.int32(mutation),count=nr)
+                    rows = self.tables[f'incoming{m}']
+                    self.launch(1, 'Inject', g,self.h,t[f'incoming{m}'],u32(len(rows)),u32(mutation==4),count=len(rows))
+                    self.launch(1, 'Pull', g,out,self.active,self.counts)
+                    self.fine[0], self.fine[1] = out, f
+                self.launch(0, 'ApplyRegisters', predicted,self.registers,t['targets'],u32(nr),self.counts,count=nr)
+                counts = self.read(self.counts,(6,),np.uint32)
+                if counts.any():
+                    raise AssertionError(f"boundary/leaf health failed: {counts.tolist()}")
+                coarse_after = self.read(predicted,(9,n,n))
+                coarse_after[:,~self.plan.maps.coarse_active] = np.nan
+                after = ref.cpu.State(coarse_after,self.read(self.fine[0],(9,nf,nf)),tick=2*(self.step_count+1))
+                ledger = rounding_ledger(self.plan,self.tables,before,after,coarse_g,prediction,fine_phases,
+                                         self.read(self.pre,(ng,4),np.float64),self.read(self.h,(ng,4)),
+                                         self.read(self.registers,(nr,),np.float64))
+                self.coarse[0], self.coarse[2] = predicted, fc
+                self.step_count += 1
+                self.status = 'ready'
+                return ledger
+            except BaseException:
+                self.status = 'failed'
+                raise
+
+    def close(self):
+        if self.status == 'closed':
+            return
+        with self.ctx.activate():
+            error = None
+            try:
+                self.stream.synchronize()
+            except Exception as exc:
+                error = exc
+            failed = []
+            for pointer in self.allocations:
+                try:
+                    pointer.free()
+                except Exception as exc:
+                    error = error or exc
+                    failed.append(pointer)
+            self.allocations = failed
+            self.status = 'cleanup_failed' if failed else 'closed'
+            if not failed:
+                self.modules.clear()
+            if error:
+                raise error
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, kind, value, traceback):
+        try:
+            self.close()
+        except Exception:
+            if kind is None:
+                raise
diff --git a/tests/validation/multigrid/test_spacetime_gpu.py b/tests/validation/multigrid/test_spacetime_gpu.py
new file mode 100644
index 0000000..3e26557
--- /dev/null
+++ b/tests/validation/multigrid/test_spacetime_gpu.py
@@ -0,0 +1,255 @@
+"""Explicit host-only checks or admitted tiny GPU audit; never a long trajectory."""
+import argparse
+from contextlib import nullcontext
+from dataclasses import replace
+import hashlib
+import json
+import os
+from pathlib import Path
+import subprocess
+import sys
+import time
+from types import SimpleNamespace
+import unittest
+from unittest.mock import Mock
+
+import numpy as np
+
+ROOT = Path(__file__).resolve().parents[3]
+sys.path.insert(0, str(ROOT/'src'))
+from CelerisLab.config import LBMConfig
+import spacetime_gpu as gpu
+ref = gpu.ref
+RECORDS = []
+REPORT = None
+OUTPUT = None
+GPU_ADMITTED = False
+
+
+def record(value):
+    RECORDS.append(value)
+    if OUTPUT is not None:
+        ref.write_receipt(OUTPUT,REPORT)
+
+
+def config(n):
+    return LBMConfig(nx=n,ny=n,viscosity=.0192,velocity=0.,ddf_shifting=False)
+
+
+def initial(plan, tagged=True):
+    arrays = []
+    for n, mask in ((plan.n,plan.maps.coarse_active),(plan.nf,plan.maps.fine_active)):
+        a = np.broadcast_to(ref.cpu.W[:,None,None], (9,n,n)).copy()
+        if tagged:
+            a *= 1+1e-3*np.random.default_rng(n).standard_normal(a.shape)
+        a[:,~mask] = np.nan
+        arrays.append(a)
+    return ref.cpu.State(*arrays)
+
+
+class Host(unittest.TestCase):
+    @classmethod
+    def setUpClass(cls):
+        cls.plan = ref.Plan(12,(3,3,9,8))
+
+    def test_pins_and_supported_config(self):
+        gpu.check_sources()
+        for n in (12,16,64):
+            c,f = gpu.validate_config(config(n))
+            self.assertAlmostEqual(c.omega,1/.5576)
+            self.assertAlmostEqual(f.omega,1/.6152)
+        for change in (dict(ddf_shifting=True),dict(collision='TRT'),dict(nx=128,ny=128),
+                       dict(ny=16),dict(les_enabled=True),dict(velocity=.03)):
+            with self.subTest(change=change), self.assertRaises(ValueError):
+                gpu.validate_config(replace(config(12),**change))
+
+    def test_packed_weights_and_packet_ownership(self):
+        for plan in (self.plan,ref.Plan(16,(4,5,11,12))):
+            t = gpu.pack_plan(plan)
+            self.assertEqual(t['sources'].shape,(len(plan.receivers),4))
+            self.assertEqual(np.count_nonzero(t['times']==-1),24)
+            self.assertEqual(gpu.table_hash(t),gpu.table_hash(gpu.pack_plan(plan)))
+            for i,g in enumerate(plan.groups):
+                np.testing.assert_array_equal(t['weights'][i,:,:len(g.samples)],g.weights)
+                np.testing.assert_allclose(t['weights'][i].sum(axis=1),1,rtol=0,atol=2e-14)
+                np.testing.assert_array_equal(t['samples'][i,:len(g.samples)],g.q*plan.n**2+g.samples)
+            for m in (0,1):
+                self.assertEqual(len(t[f'incoming{m}']),len(plan.outgoing[m]))
+
+    def test_duplicate_writer_is_rejected(self):
+        plan = self.plan
+        groups = list(plan.groups)
+        a,b = next((i,k) for i,g in enumerate(groups) for k,e in enumerate(g.events) if e is not None)
+        event = groups[a].events[b]
+        other = next(i for i,g in enumerate(groups) if i!=a and g.q==groups[a].q and event[0] in [e[0] for e in g.events if e])
+        events = list(groups[other].events)
+        k = next(k for k,e in enumerate(events) if e is not None and e[0]==event[0])
+        events[k] = event; groups[other] = replace(groups[other],events=tuple(events))
+        fake = SimpleNamespace(n=plan.n,nf=plan.nf,groups=groups,receivers=plan.receivers,outgoing=plan.outgoing)
+        with self.assertRaisesRegex(ValueError,'duplicate incoming'):
+            gpu.pack_plan(fake)
+
+    def test_rest_ledger_and_missing_packet_discrimination(self):
+        p,t = self.plan,gpu.pack_plan(self.plan)
+        state = initial(p,False)
+        state = ref.cpu.State(state.coarse_f.astype(np.float32),state.fine_f.astype(np.float32))
+        g0 = np.broadcast_to(ref.cpu.W.astype(np.float32)[:,None,None],(9,p.n,p.n)).copy()
+        pre = np.repeat(g0.ravel()[t['origins'],None].astype(float),4,axis=1)
+        h = pre.astype(np.float32); registers = np.zeros(len(p.receivers))
+        phases = [(state.fine_f,state.fine_f)]*2
+        args = (p,t,state,state,g0,g0,phases,pre,h)
+        ledger = gpu.rounding_ledger(*args,registers)
+        self.assertEqual(max(map(abs,ledger['unexplained'])),0)
+        registers[0] = .01
+        with self.assertRaisesRegex(AssertionError,'register mismatch'):
+            gpu.rounding_ledger(*args,registers)
+
+    def test_second_substep_characteristic_donor(self):
+        p = ref.Plan(16,(4,4,12,12)); t = gpu.pack_plan(p)
+        row = t['incoming1'][t['incoming1'][:,0]==5*32*32+10*32+7][0]
+        self.assertEqual(t['origins'][row[1]//4],5*16*16+4*16+3)
+
+    def test_failed_free_remains_retryable(self):
+        run=gpu.SpaceTimeGPU.__new__(gpu.SpaceTimeGPU)
+        run.ctx=SimpleNamespace(activate=nullcontext); run.stream=Mock()
+        first,second=Mock(),Mock(); first.free.side_effect=[RuntimeError('free failed'),None]
+        run.allocations=[first,second]; run.modules=[]; run.status='ready'
+        with self.assertRaisesRegex(RuntimeError,'free failed'):
+            run.close()
+        self.assertEqual(run.allocations,[first]); self.assertEqual(run.status,'cleanup_failed')
+        run.close()
+        self.assertEqual(run.status,'closed'); self.assertEqual(run.allocations,[])
+        second.free.assert_called_once()
+
+
+class Device(unittest.TestCase):
+    @classmethod
+    def setUpClass(cls):
+        if not GPU_ADMITTED:
+            raise unittest.SkipTest('CUDA compilation/execution requires explicit tiny-job admission')
+        from CelerisLab.cuda.context import CudaContext
+        cls.context = CudaContext(0)
+
+    @classmethod
+    def tearDownClass(cls):
+        cls.context.close()
+
+    def test_tiny_equivalence_budgets_and_phase(self):
+        for mode,identity in (('spacetime',False),('spacetime',True),('frozen',False),('frozen',True)):
+            with self.subTest(mode=mode,identity=identity), gpu.SpaceTimeGPU(
+                    config(12),(3,3,9,8),context=self.context,mode=mode,identity=identity) as run:
+                spec = ref.Spec(n_coarse=12,patch_lo=1,halo=2,tau_coarse=.5576,tau_fine=.6152,
+                                collision='identity' if identity else 'srt')
+                for tagged,steps in ((False,2),(True,8)):
+                    expected = initial(run.plan,tagged); run.initialize(expected)
+                    before = run.download()
+                    for k in range(1,steps+1):
+                        expected,_ = ref.advance(spec,run.plan,expected,mode)
+                        ledger = run.step(); actual = run.download()
+                        self.assertEqual(actual.tick,2*k); self.assertEqual(run.step_count,k)
+                        for a,b in zip((actual.coarse_f,actual.fine_f),(expected.coarse_f,expected.fine_f)):
+                            np.testing.assert_array_equal(np.isnan(a),np.isnan(b))
+                            np.testing.assert_allclose(a,b,rtol=0,atol=3e-6,equal_nan=True)
+                        total = lambda s: ref.cpu.MOMENTS.T @ ref.directional(
+                            ref.cpu.State(s.coarse_f.astype(float),s.fine_f.astype(float)),run.plan.maps)
+                        np.testing.assert_allclose(total(actual),total(before),rtol=0,atol=2e-4)
+                        if tagged and not identity and k==1:
+                            for a,b,tau,mask in zip((actual.coarse_f,actual.fine_f),
+                                    (expected.coarse_f,expected.fine_f),(.5576,.6152),
+                                    (run.plan.maps.coarse_active,run.plan.maps.fine_active)):
+                                wrong = ref.cpu._collide(spec,b,np.isfinite(b),tau,mask,np.zeros(mask.shape,int))
+                                self.assertGreater(float(np.nanmax(abs(a-wrong))),3e-6)
+                        if k in (1,2,8):
+                            record(dict(case='tiny',mode=mode,identity=identity,tagged=tagged,
+                                                step=k,device_bytes=run.device_bytes,ledger=ledger))
+
+    def test_omitted_packets_and_ghosts_fail(self):
+        for mutation in (1,2,3,4):
+            with self.subTest(mutation=mutation), gpu.SpaceTimeGPU(
+                    config(12),(3,3,9,8),context=self.context) as run:
+                run.initialize(initial(run.plan))
+                with self.assertRaises(AssertionError):
+                    run.step(mutation=mutation)
+                self.assertEqual(run.step_count,0)
+                self.assertEqual(run.status,'failed')
+                with self.assertRaises(RuntimeError):
+                    run.download()
+
+    def test_frozen_donor_trap(self):
+        with gpu.SpaceTimeGPU(config(16),(4,4,12,12),context=self.context,mode='frozen',identity=True) as run:
+            state = initial(run.plan,False)
+            state.coarse_f[5,4,3]=1.; state.coarse_f[5,5,3]=2.
+            run.initialize(state); run.step()
+            self.assertEqual(run.download().fine_f[5,11,8],1.)
+            # Deliberately negative boundary data are reported and remain unclipped.
+            with self.context.activate():
+                data=np.full((9,16,16),-.01,np.float32)
+                run.cuda.memcpy_htod(run.coarse[1],data); run.cuda.memcpy_htod(run.coarse[3],data)
+                run.cuda.memset_d32_async(run.counts,0,6,run.stream)
+                t=run.device_tables; ng=len(run.plan.groups)
+                run.launch(0,'Interpolate',run.coarse[1],run.coarse[3],t['samples'],t['origins'],
+                           t['sizes'],t['weights'],run.pre,run.h,np.uint32(ng),np.uint32(0),run.counts,count=ng)
+                counts=run.read(run.counts,(6,),np.uint32)
+                self.assertEqual(counts[1],4*ng); self.assertEqual(counts[3],4*ng)
+                self.assertTrue((run.read(run.h,(ng,4))<0).all())
+
+    def test_n64_one_interval_only(self):
+        sys.path.insert(0,str(Path(__file__).parents[1]/'taylor_green'))
+        import oracle as tg
+        with gpu.SpaceTimeGPU(config(64),(8,8,56,48),context=self.context) as run:
+            arrays = [tg.initial((64,64),0,64,.0192),tg.initial((128,128),1,64,.0192)]
+            for a,mask in zip(arrays,(run.plan.maps.coarse_active,run.plan.maps.fine_active)):
+                a[:,~mask]=np.nan
+            state = ref.cpu.State(*arrays); run.initialize(state)
+            spec = ref.Spec(n_coarse=64,patch_lo=8,halo=16,tau_coarse=.5576,tau_fine=.6152,collision='srt')
+            expected,_ = ref.advance(spec,run.plan,state)
+            started=time.monotonic(); ledger=run.step(); seconds=time.monotonic()-started
+            actual=run.download()
+            errors=[]
+            for a,b in zip((actual.coarse_f,actual.fine_f),(expected.coarse_f,expected.fine_f)):
+                np.testing.assert_array_equal(np.isnan(a),np.isnan(b))
+                np.testing.assert_allclose(a,b,rtol=0,atol=3e-6,equal_nan=True)
+                errors.append(float(np.nanmax(abs(a-b))))
+            record(dict(case='N64_one_interval',errors=errors,ledger=ledger,seconds=seconds,
+                                device_bytes=run.device_bytes,table_sha256=gpu.table_hash(run.tables),
+                                build_metadata=run.build_metadata,device=self.context.name))
+
+
+def main():
+    global REPORT,OUTPUT,GPU_ADMITTED
+    parser=argparse.ArgumentParser(description=__doc__)
+    mode=parser.add_mutually_exclusive_group(required=True)
+    mode.add_argument('--host-only',action='store_true')
+    mode.add_argument('--gpu-admitted',action='store_true')
+    parser.add_argument('--output',type=Path)
+    args=parser.parse_args()
+    GPU_ADMITTED=args.gpu_admitted
+    if args.gpu_admitted and (not os.environ.get('CUDA_VISIBLE_DEVICES') or args.output is None):
+        parser.error('GPU job requires explicit CUDA_VISIBLE_DEVICES and --output')
+    if args.output is not None and args.output.exists():
+        parser.error('output exists; retain evidence and choose a new path')
+    started=time.monotonic()
+    report=dict(scope='tiny GPU equivalence and rounding audit; no long trajectory',completed=False,
+                requested_mode='admitted_tiny_gpu' if args.gpu_admitted else 'host_only',
+                commit=subprocess.check_output(['git','rev-parse','HEAD'],cwd=ROOT,text=True).strip(),
+                dirty=bool(subprocess.check_output(['git','status','--porcelain'],cwd=ROOT,text=True)),
+                hashes={p.name:hashlib.sha256(p.read_bytes()).hexdigest() for p in
+                        (Path(__file__),Path(gpu.__file__),Path(gpu.__file__).with_suffix('.cu'))},
+                reference_hashes=gpu.PINS,cuda_visible_devices=os.environ.get('CUDA_VISIBLE_DEVICES'),records=RECORDS)
+    REPORT,OUTPUT=report,args.output
+    if args.output:
+        ref.write_receipt(args.output,report)
+    suite=unittest.TestLoader().loadTestsFromTestCase(Host)
+    if args.gpu_admitted:
+        suite.addTests(unittest.TestLoader().loadTestsFromTestCase(Device))
+    result=unittest.TextTestRunner(verbosity=2).run(suite)
+    report.update(completed=True,passed=result.wasSuccessful(),tests=result.testsRun,
+                  failures=[dict(test=str(test),error=error) for test,error in result.failures+result.errors],
+                  seconds=time.monotonic()-started)
+    if args.output:
+        ref.write_receipt(args.output,report)
+    return 0 if result.wasSuccessful() else 1
+
+
+if __name__=='__main__':
+    raise SystemExit(main())