From 90694e69af2a364716f392e13a887e2ae25877e5 Mon Sep 17 00:00:00 2001
From: Codex <codex@openai.com>
Date: Sun, 18 Oct 2026 13:20:35 -0700
Subject: [PATCH] test: define union space-time event ownership contract

---
 .../multigrid/spacetime_union_plan.py         | 212 ++++++++++++++++++
 .../multigrid/test_spacetime_union_plan.py    | 191 ++++++++++++++++
 2 files changed, 403 insertions(+)
 create mode 100644 tests/validation/multigrid/spacetime_union_plan.py
 create mode 100644 tests/validation/multigrid/test_spacetime_union_plan.py

diff --git a/tests/validation/multigrid/spacetime_union_plan.py b/tests/validation/multigrid/spacetime_union_plan.py
new file mode 100644
index 0000000..223f857
--- /dev/null
+++ b/tests/validation/multigrid/spacetime_union_plan.py
@@ -0,0 +1,212 @@
+"""Host-only experimental union event contract; no solver or runtime admission.
+
+Consumes existing level-wide union geometry, never its legacy transfer tables.
+The frozen 707afe6 rectangle oracle remains independent and unchanged. Topology
+acceptance proves packet accounting, not interpolation accuracy or GPU support.
+"""
+from types import SimpleNamespace
+
+import numpy as np
+import spacetime_reference as ref
+
+
+class UnresolvedTopology(ValueError):
+    """An experimental coupling limitation, not a production geometry rejection."""
+    def __init__(self, code, **detail):
+        self.code, self.detail = code, detail
+        super().__init__(f"experimental space-time topology unresolved: {code}: {detail}")
+
+
+def owned_slots(layout, directions, cells):
+    """Map dense logical reads/writes to existing owner-core storage, not halos."""
+    cells = np.asarray(cells)
+    if not np.issubdtype(cells.dtype, np.integer) or np.any((cells < 0) | (cells >= np.prod(layout.shape))):
+        raise ValueError('invalid dense cell index')
+    owner = layout.owner.ravel()[cells]
+    if np.any(owner == layout.hole):
+        raise UnresolvedTopology('missing_block_owner', cells=cells[owner == layout.hole].tolist())
+    return layout.slot(owner, directions)
+
+
+def _first_entry(covered, origin, child, c):
+    """First continuous union entry of one characteristic, in exact quarter times.
+
+    Probe the cells immediately before/after each crossed grid line using integer
+    eighth-cell coordinates. Return only actual exterior faces of the UNION.
+    """
+    start4 = 4*origin + 2*child + 1
+    for t4 in (1, 3):
+        before = (2*start4 + c*(2*t4-1))//8
+        after = (2*start4 + c*(2*t4+1))//8
+        if not covered[tuple(before)] and covered[tuple(after)]:
+            axes = []
+            for j in range(2):
+                if before[j] != after[j]:
+                    neighbor = after.copy(); neighbor[j] = before[j]
+                    if not covered[tuple(neighbor)]: axes.append(j)
+            return t4, axes
+    return None
+
+
+class UnionPlan:
+    """Two-level, level-wide square geometry; all coordinates are level-local.
+
+    Assumes a complete parent domain; does not validate nested parent coverage.
+    ``pair`` is existing FixedPatchSet2D geometry, even for one rectangle. Scalar
+    window origins and nested state coverage need their own contract; they are
+    explicitly unadmitted here. No global rectangle-gap rule is imposed.
+    """
+    phase = 'arrival_f'
+    runtime_supported = False
+    interpolation_accuracy = 'unqualified_for_new_unions'
+
+    def __init__(self, pair):
+        ny, nx = pair.coarse_shape
+        self.origin = tuple(pair.fine_origin)
+        if ny != nx or pair.fine_shape != (2*ny, 2*nx) or self.origin != (0., 0.):
+            raise UnresolvedTopology('requires_level_wide_square_pair',
+                                     coarse_shape=pair.coarse_shape, fine_shape=pair.fine_shape, origin=self.origin)
+        self.n, self.nf = nx, 2*nx
+        covered = np.array(pair.covered, dtype=bool, copy=True)
+        active = np.kron(covered, np.ones((2, 2), bool))
+        if not np.array_equal(active, pair.fine_kind == 0):
+            raise UnresolvedTopology('non_pair_active_coverage')
+        self.maps = SimpleNamespace(coarse_active=~covered, coarse_covered=covered, fine_active=active)
+        self.outgoing = [[], []]
+        # Check transit FIRST: a fine exit may not be silently routed into a
+        # covered coarse cell. Returning fine packets need a separately derived rule.
+        for q in range(1, 9):
+            c = np.array((ref.cpu.CY[q], ref.cpu.CX[q]))
+            boundary = active & ~np.roll(active, tuple(-c), axis=(0, 1))
+            for y, x in zip(*np.nonzero(boundary)):
+                p = np.array((y, x))
+                for m in (0, 1):
+                    target = (p+(2-m)*c)//2
+                    if covered[tuple(target)]:
+                        raise UnresolvedTopology('fine_exit_reentry', q=q, substep=m,
+                            fine_path_yx=[(p+j*c).tolist() for j in range(3-m)], coarse_target_yx=target.tolist())
+                    self.outgoing[m].append((q, int(y*self.nf+x), int(target[0]*nx+target[1])))
+        self.outgoing = [np.array(x, dtype=np.int64).reshape(-1, 3) for x in self.outgoing]
+        self.groups = []
+        for iy, ix in zip(*np.nonzero(~covered)):
+            origin = np.array((iy, ix))
+            # Two-cell production margins make all nearby probes in bounds.
+            if not covered[max(0,iy-1):iy+2,max(0,ix-1):ix+2].any(): continue
+            for q in range(1, 9):
+                c = np.array((ref.cpu.CY[q], ref.cpu.CX[q]))
+                children = [np.array((dy, dx)) for dy in (0, 1) for dx in (0, 1)]
+                starts = [2*origin+child for child in children]
+                events = []
+                for p in starts:
+                    event = None
+                    for m in (0, 1):
+                        old, new = p+m*c, p+(m+1)*c
+                        if not active[tuple(old)] and active[tuple(new)]:
+                            event = (m, int(old[0]*self.nf+old[1])); break
+                    events.append(event)
+                if all(e is None for e in events): continue
+                entries = [_first_entry(covered, origin, child, c) for child in children]
+                axes = sorted({j for entry in entries if entry is not None for j in entry[1]})
+                if not axes:
+                    raise UnresolvedTopology('missing_exterior_support', source_yx=origin.tolist(), q=q)
+                points = []
+                for p, entry, event in zip(starts, entries, events):
+                    position = (p+.5)/2
+                    planes = origin+(c > 0)
+                    times = [(planes[j]-position[j])/c[j] for j in axes]
+                    time = max(times)
+                    if entry is not None and (4*time != entry[0] or event is None):
+                        raise UnresolvedTopology('nonconvex_entry_support', source_yx=origin.tolist(), q=q,
+                                                 quarter_time=entry[0], support_quarter_time=float(4*time))
+                    offset = position+c*time-(origin+.5+c*.5)
+                    points.append((float(offset[1]), float(offset[0]), float(time-.5)) if len(axes)==2
+                                  else (float(offset[1-axes[0]]), float(time-.5)))
+                kind = 'corner' if len(axes)==2 else 'face'; points = tuple(points)
+                samples, _, evaluation, coefficients = ref.interpolation(kind, points)
+                indices = []
+                for row in samples:
+                    shift = np.array((row[1],row[0]), int) if kind=='corner' else np.eye(2,dtype=int)[1-axes[0]]*int(row[0])
+                    pos = (origin+shift) % nx; indices.append(int(pos[0]*nx+pos[1]))
+                target = (origin+c) % nx
+                self.groups.append(ref.Group(q,int(iy*nx+ix),int(target[0]*nx+target[1]),kind,
+                    points,np.array(indices),evaluation@coefficients,tuple(events)))
+        receivers = {(int(q),int(target)) for rows in self.outgoing for q,_,target in rows}
+        receivers |= {(g.q,g.receiver) for g in self.groups if None in g.events}
+        self.receivers = np.array(sorted(receivers),dtype=np.int64).reshape(-1,2)
+        self.proof = validate_ownership(self)
+
+
+def validate_ownership(plan):
+    """Independent mask-based event coverage and four-quarter receiver accounting."""
+    n, nf, active = plan.n, plan.nf, plan.maps.fine_active
+    actual = [(m,g.q,slot) for g in plan.groups for e in g.events if e is not None for m,slot in [e]]
+    expected = {(m,q,int((y-ref.cpu.CY[q])*nf+x-ref.cpu.CX[q]))
+        for m in (0,1) for q in range(1,9) for y,x in zip(*np.nonzero(active))
+        if not active[y-ref.cpu.CY[q],x-ref.cpu.CX[q]]}
+    if len(actual)!=len(set(actual)) or set(actual)!=expected:
+        raise ValueError('incoming writer uniqueness/coverage failed')
+    packets = {tuple(map(int,row)):[] for row in plan.receivers}
+    if len(packets)!=len(plan.receivers): raise ValueError('duplicate coarse receiver')
+    anchors = set()
+    for i,g in enumerate(plan.groups):
+        if not 1<=g.q<9 or not 0<=g.source<n*n or not plan.maps.coarse_active.ravel()[g.source]:
+            raise ValueError('invalid coarse source anchor')
+        if (g.q,g.source) in anchors: raise ValueError('duplicate coarse source group')
+        anchors.add((g.q,g.source))
+        origin=np.array(divmod(g.source,n)); c=np.array((ref.cpu.CY[g.q],ref.cpu.CX[g.q]))
+        target=(origin+c)%n
+        if g.receiver!=target[0]*n+target[1]: raise ValueError('group receiver provenance failed')
+        ordered=[]; entries=[]
+        for dy,dx in ((0,0),(0,1),(1,0),(1,1)):
+            p=2*origin+(dy,dx); event=None
+            for m in (0,1):
+                old,new=p+m*c,p+(m+1)*c
+                if not active[tuple(old)] and active[tuple(new)]:
+                    event=(m,int(old[0]*nf+old[1])); break
+            ordered.append(event)
+            entries.append(_first_entry(plan.maps.coarse_covered,origin,np.array((dy,dx)),c))
+        if tuple(ordered)!=g.events or all(e is None for e in ordered):
+            raise ValueError('ordered child characteristic provenance failed')
+        axes=sorted({j for e in entries if e is not None for j in e[1]})
+        if not axes or g.kind!=('corner' if len(axes)==2 else 'face'):
+            raise ValueError('group supporting axes differ')
+        samples,_,evaluation,coefficients=ref.interpolation(g.kind,g.points)
+        indices=[]
+        for row in samples:
+            shift=np.array((row[1],row[0]),int) if g.kind=='corner' else np.eye(2,dtype=int)[1-axes[0]]*int(row[0])
+            pos=(origin+shift)%n; indices.append(int(pos[0]*n+pos[1]))
+        if not np.array_equal(g.samples,indices): raise ValueError('sample anchor provenance failed')
+        for k,(dy,dx) in enumerate(((0,0),(0,1),(1,0),(1,1))):
+            position=origin+(np.array((dy,dx))+.5)/2
+            time=max((origin[j]+int(c[j]>0)-position[j])/c[j] for j in axes)
+            offset=position+c*time-(origin+.5+c*.5)
+            point=(float(offset[1]),float(offset[0]),float(time-.5)) if g.kind=='corner' else (float(offset[1-axes[0]]),float(time-.5))
+            if g.points[k]!=point: raise ValueError('evaluation point provenance failed')
+        source=np.zeros(len(g.samples)); source[0 if g.kind=='corner' else 1]=1
+        if (not np.allclose(g.weights,evaluation@coefficients,rtol=0,atol=2e-14)
+                or not np.allclose(g.weights.mean(axis=0),source,rtol=0,atol=2e-14)):
+            raise ValueError('group coefficients/directional budget failed')
+        for k,event in enumerate(g.events):
+            if event is None:
+                if (g.q,g.receiver) not in packets: raise ValueError('missing leftover receiver')
+                packets[(g.q,g.receiver)].append(('leftover',i,k))
+    for m,rows in enumerate(plan.outgoing):
+        expected_out = {(q,int(y*nf+x),int(((y+(2-m)*ref.cpu.CY[q])//2)*n+(x+(2-m)*ref.cpu.CX[q])//2))
+            for q in range(1,9) for y,x in zip(*np.nonzero(active))
+            if not active[y+ref.cpu.CY[q],x+ref.cpu.CX[q]]}
+        actual_out = [tuple(map(int,row)) for row in rows]
+        if len(actual_out)!=len(set(actual_out)) or set(actual_out)!=expected_out:
+            raise ValueError('outgoing uniqueness/coverage failed')
+        for q,slot,target in actual_out: packets[(q,target)].append(('outgoing',m,q,slot))
+    expected_receivers = {(q,int(y*n+x)) for y,x in zip(*np.nonzero(plan.maps.coarse_active))
+        for q in range(1,9) if any(active[2*y+dy-ref.cpu.CY[q],2*x+dx-ref.cpu.CX[q]]
+            for dy in (0,1) for dx in (0,1)
+            if 0<=2*y+dy-ref.cpu.CY[q]<nf and 0<=2*x+dx-ref.cpu.CX[q]<nf)}
+    if set(packets)!=expected_receivers: raise ValueError('coarse receiver coverage failed')
+    tokens=[token for row in packets.values() for token in row]
+    if any(len(row)!=4 for row in packets.values()) or len(tokens)!=len(set(tokens)):
+        raise ValueError('receiver needs four distinct quarter-packets')
+    return dict(topology='supported',incoming_events=len(actual),outgoing_events=sum(map(len,plan.outgoing)),
+        groups=len(plan.groups),receivers=len(packets),receiver_quarter_packets=len(tokens),
+        interpolation_accuracy=plan.interpolation_accuracy,runtime_supported=False,
+        scope='pair-local ownership/coefficients; complete parent domain assumed, nested coverage unvalidated')
diff --git a/tests/validation/multigrid/test_spacetime_union_plan.py b/tests/validation/multigrid/test_spacetime_union_plan.py
new file mode 100644
index 0000000..2aab543
--- /dev/null
+++ b/tests/validation/multigrid/test_spacetime_union_plan.py
@@ -0,0 +1,191 @@
+"""Host-only ownership/geometry tests, never a GPU or accuracy admission."""
+import copy
+import json
+import sys
+import unittest
+
+import numpy as np
+from CelerisLab.config import LBMConfig
+from CelerisLab.lbm.blocks import BlockLayout2D
+from CelerisLab.lbm.refinement import FixedPatch2D, FixedPatchSet2D
+from spacetime_union_plan import UnionPlan, UnresolvedTopology, owned_slots, validate_ownership
+import spacetime_reference as ref
+import spacetime_gpu as frozen  # pack_plan only: no CUDA import/owner construction
+from spacetime_table_guard import exact_weights
+
+MATRIX=[]
+CASES={
+ 'rectangle': ([(4,4,12,12)],True),
+ 'two_separated': ([(3,3,6,6),(9,9,12,12)],True),
+ 'one_cell_face_gap': ([(3,3,6,10),(7,3,10,10)],True),
+ 'one_cell_diagonal_gap': ([(3,3,6,6),(7,7,10,10)],True),
+ 'staggered_gap': ([(3,3,6,8),(7,5,11,10)],True),
+ 'thin_rectangle': ([(4,4,5,11)],True),
+ 'concave_L': ([(4,4,5,7),(4,5,7,7)],False),
+ 'corner_contact': ([(3,3,6,6),(6,6,9,9)],False),
+ 'cross': ([(3,6,13,9),(6,3,9,13)],False),
+ 'narrow_notch': ([(3,3,6,11),(7,3,10,11),(3,9,10,11)],False),
+ 'hole': ([(3,3,13,5),(3,11,13,13),(3,5,5,11),(11,5,13,11)],False),
+}
+
+
+def pair(rects,n=16,ny=None):
+    cfg=LBMConfig(nx=n,ny=n if ny is None else ny,viscosity=.0192,velocity=0.)
+    return FixedPatchSet2D.from_config(cfg,rects)
+
+
+def assert_same(test,a,b):
+    # Includes event-child order, zero padding, integer dtype/shape and FP64 bits.
+    aa,bb=frozen.pack_plan(a),frozen.pack_plan(b)
+    test.assertEqual(aa.keys(),bb.keys())
+    for key in aa:
+        test.assertEqual((aa[key].dtype,aa[key].shape,aa[key].tobytes()),
+                         (bb[key].dtype,bb[key].shape,bb[key].tobytes()),key)
+    for x,y in zip(a.groups,b.groups):
+        test.assertEqual((x.q,x.source,x.receiver,x.kind,x.points,x.events),
+                         (y.q,y.source,y.receiver,y.kind,y.points,y.events))
+
+
+class UnionContract(unittest.TestCase):
+    def test_single_rectangle_exact_707afe6(self):
+        frozen.check_sources()
+        for n,b in ((12,(3,3,9,8)),(16,(4,5,11,12)),(16,(4,4,5,7)),(64,(8,8,56,48))):
+            with self.subTest(n=n,bounds=b):
+                assert_same(self,UnionPlan(pair([b],n)),ref.Plan(n,b))
+        self.assertNotIn('pycuda.driver',sys.modules)
+
+    def test_union_decomposition_and_permutation_invariance(self):
+        baseline=UnionPlan(pair([(4,4,12,12)]))
+        decompositions=[[(4,4,8,12),(8,4,12,12)],[(4,4,10,12),(6,4,12,12)],
+                        [(4,4,12,8),(4,8,12,12)],[(4,4,12,12),(4,4,12,12)]]
+        for rectangles in decompositions:
+            for order in (rectangles,list(reversed(rectangles))):
+                assert_same(self,baseline,UnionPlan(pair(order)))
+        two=CASES['two_separated'][0]
+        assert_same(self,UnionPlan(pair(two)),UnionPlan(pair(list(reversed(two)))))
+
+    def test_supported_unresolved_matrix(self):
+        for name,(rectangles,supported) in CASES.items():
+            geometry=pair(rectangles)  # Existing production accepts EVERY fixture.
+            row=dict(case=name,rectangles=rectangles,production_geometry_accepted=True,
+                     runtime_supported=False,interpolation_accuracy='unqualified_for_new_unions')
+            if supported:
+                plan=UnionPlan(geometry); row.update(plan.proof)
+                self.assertEqual(plan.proof['receiver_quarter_packets'],4*len(plan.receivers))
+            else:
+                with self.assertRaises(UnresolvedTopology) as caught: UnionPlan(geometry)
+                self.assertEqual(caught.exception.code,'fine_exit_reentry')
+                row.update(topology='unresolved',code=caught.exception.code,witness=caught.exception.detail)
+            MATRIX.append(row)
+
+    def test_L_witness_and_reflections(self):
+        rectangles=CASES['concave_L'][0]
+        with self.assertRaises(UnresolvedTopology) as caught: UnionPlan(pair(rectangles))
+        self.assertEqual(caught.exception.detail,dict(q=5,substep=0,
+            fine_path_yx=[[8,9],[9,10],[10,11]],coarse_target_yx=[5,5]))
+        directions=set()
+        for reflect in (False,True):
+            transformed=[(16-x1,y0,16-x0,y1) if reflect else (x0,y0,x1,y1) for x0,y0,x1,y1 in rectangles]
+            for _ in range(4):
+                with self.assertRaises(UnresolvedTopology) as caught: UnionPlan(pair(transformed))
+                detail=caught.exception.detail; directions.add(detail['q'])
+                p=np.array(detail['fine_path_yx']); geometry=pair(transformed)
+                self.assertEqual([bool(geometry.fine_kind[tuple(z)]==0) for z in p],[True,False,True])
+                transformed=[(16-y1,x0,16-y0,x1) for x0,y0,x1,y1 in transformed]
+        self.assertGreater(len(directions),1)
+
+    def test_ordered_provenance_and_packet_mutations_fail(self):
+        baseline=UnionPlan(pair(CASES['two_separated'][0]))
+        def reject(change):
+            bad=copy.deepcopy(baseline); change(bad)
+            with self.assertRaises((ValueError,KeyError)): validate_ownership(bad)
+        reject(lambda p:setattr(p.groups[0],'source',p.groups[0].source+1))
+        reject(lambda p:setattr(p.groups[0],'samples',p.groups[0].samples+1))
+        reject(lambda p:setattr(p.groups[0],'events',tuple(reversed(p.groups[0].events))))
+        reject(lambda p:setattr(p.groups[0],'receiver',p.groups[0].receiver+1))
+        reject(lambda p:p.outgoing.__setitem__(0,p.outgoing[0][1:]))
+        reject(lambda p:p.outgoing.__setitem__(1,np.vstack((p.outgoing[1],p.outgoing[1][0]))))
+        reject(lambda p:setattr(p,'receivers',p.receivers[1:]))
+        def balanced(p):
+            p.groups[0].weights[:2,:2]+=1e-5*np.array([[1,-1],[-1,1]])
+        reject(balanced)  # Preserves row/column sums, corrupts the actual operator.
+
+    def test_coefficients_and_directional_transfer_budget(self):
+        rng=np.random.default_rng(31)
+        for name in ('two_separated','one_cell_face_gap','thin_rectangle'):
+            p=UnionPlan(pair(CASES[name][0])); total=np.zeros(9)
+            receiver={tuple(row):[] for row in p.receivers}; first_leftover=None
+            for g in p.groups:
+                samples,a,e,coeff=ref.interpolation(g.kind,g.points)
+                np.testing.assert_allclose(g.weights,exact_weights(g.kind,g.points),rtol=0,atol=2e-14)
+                np.testing.assert_allclose(g.weights@np.ones(len(samples)),1,rtol=0,atol=2e-14)
+                if g.kind=='face': beta=np.array([1.,.03,.02,0.])
+                else:
+                    beta=np.array([1.,.03,.02,-ref.cpu.CX[g.q]*.03-ref.cpu.CY[g.q]*.02])
+                    self.assertLess(abs((coeff@(a@np.array([1.,.03,.02,0.])))[3]),2e-14)
+                np.testing.assert_allclose(g.weights@(a@beta),e@beta,rtol=0,atol=2e-14)
+                values=rng.normal(size=len(samples)); h=g.weights@values
+                origin=values[0 if g.kind=='corner' else 1]
+                # Fine incoming plus coarse leftovers must exactly split one
+                # coarse directional source budget, regardless of sample values.
+                incoming=sum(h[k] for k,event in enumerate(g.events) if event is not None)
+                leftover=sum(h[k] for k,event in enumerate(g.events) if event is None)
+                self.assertLess(abs(.25*(incoming+leftover)-origin),2e-14)
+                total[g.q]+=.25*incoming-origin
+                for k,event in enumerate(g.events):
+                    if event is None:
+                        receiver[(g.q,g.receiver)].append(h[k])
+                        if first_leftover is None: first_leftover=(g.q,h[k])
+            # Tagged collision-free boundary-flux ledger, not a solver trajectory:
+            # every outgoing fine debit is credited to its actual receiver.
+            for rows in p.outgoing:
+                for q,slot,target in rows:
+                    value=float(rng.normal()); total[q]-=.25*value
+                    receiver[(q,target)].append(value)
+            for (q,target),values in receiver.items():
+                self.assertEqual(len(values),4); total[q]+=.25*sum(values)
+            np.testing.assert_allclose(total,0,rtol=0,atol=2e-14)
+            damaged=total.copy(); q,value=first_leftover; damaged[q]-=.25*value
+            self.assertGreater(float(abs(damaged).max()),1e-5)
+            self.assertEqual(validate_ownership(p)['topology'],'supported')
+
+    def test_existing_block_addressing_and_sparse_holes(self):
+        geometry=pair(CASES['two_separated'][0]); p=UnionPlan(geometry)
+        layouts=(BlockLayout2D.build(16,16,4,4),geometry.fine_layout((4,4)))
+        for fine,layout in enumerate(layouts):
+            cells=layout.shape[0]*layout.shape[1]
+            dense=np.arange(9*cells,dtype=np.float64).reshape(9,*layout.shape)
+            blocked=layout.to_blocked(dense).ravel()
+            refs=[]
+            if not fine:
+                refs=[(g.q,int(k)) for g in p.groups for k in [g.source,g.receiver,*g.samples]]
+            else:
+                refs=[(g.q,slot) for g in p.groups for event in g.events if event is not None for _,slot in [event]]
+                refs += [(int(q),int(slot)) for rows in p.outgoing for q,slot,_ in rows]
+            q,k=np.array(refs).T
+            np.testing.assert_array_equal(blocked[owned_slots(layout,q,k)],dense.reshape(9,-1)[q,k])
+        holes=np.flatnonzero(~layouts[1].owned)
+        self.assertGreater(len(holes),0)
+        with self.assertRaises(UnresolvedTopology): owned_slots(layouts[1],np.array([1]),holes[:1])
+
+    def test_deferred_layouts_and_phase_are_explicit(self):
+        with self.assertRaises(UnresolvedTopology) as caught: UnionPlan(pair([(3,3,8,8)],16,12))
+        self.assertEqual(caught.exception.code,'requires_level_wide_square_pair')
+        cfg=LBMConfig(nx=16,ny=16,viscosity=.0192,velocity=0.)
+        scalar=FixedPatch2D.from_config(cfg,(4,4,8,8))
+        with self.assertRaises(UnresolvedTopology): UnionPlan(scalar)
+        self.assertEqual(UnionPlan.phase,'arrival_f'); self.assertFalse(UnionPlan.runtime_supported)
+        # Restriction and collision do not commute, even at omega=1. This is
+        # why postcollision G storage cannot be relabeled/inverted into this F contract.
+        w=ref.cpu.W; cx,cy=ref.cpu.CX,ref.cpu.CY
+        def equilibrium(u):
+            cu=cx*u;return w*(1+3*cu+4.5*cu*cu-1.5*u*u)
+        children=np.array([equilibrium(u) for u in (.03,-.03,.03,-.03)])
+        self.assertAlmostEqual(float(abs(w-children.mean(axis=0)).max()),.0006,places=14)
+
+
+if __name__=='__main__':
+    result=unittest.TextTestRunner(verbosity=2).run(unittest.defaultTestLoader.loadTestsFromTestCase(UnionContract))
+    print(json.dumps(dict(passed=result.wasSuccessful(),tests=result.testsRun,
+        scope='host-only topology/provenance/coefficient checks; no runtime or accuracy admission',matrix=MATRIX),indent=2))
+    sys.exit(not result.wasSuccessful())