test: DFG Re20 steady rule as agreed (10 % range, 2 % slope, window-mean verdict)

Supersedes 71221837. Samples every t* 0.5 (interval-mean forces), trailing t* 50..60 window (21 samples); range <= 10 % and |slope*window| <= 2 % of each official band width. Verdict on the window mean with the distance to the nearest band edge. Our engineering rule, frozen before any run. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
test: DFG Re20 steady rule per review (1 % band, range and slope, 0.5 s samples)
2026-10-20 13:41:41 +00:00 · 2026-10-20 13:39:06 +00:00 · 2026-10-20 13:37:19 +00:00
4 changed files with 13 additions and 226 deletions
@@ -1,35 +0,0 @@
-# Compatibility and API changes
-
-Changes that can make a previously accepted setup or call behave differently.
-Each entry names the commit, the affected entry points and the observable result.
-
-## 87c3a31: cut-link ownership is validated for every body
-
-`validate_cut_link_ownership` (`body/preprocess/flag_overlay.py`) now runs for
-passive bodies as well as Jet/aperture bodies. Every cut link must have an
-active fluid receiver (FLUID set, SOLID clear), and each `(receiver, direction)`
-population may have only one writer.
-
-Rejected with `ValueError`, before flags, populations or device buffers change:
-
-1. Overlapping or touching passive solids whose cut links write the same
-   population ("Boundary requires unique cut-link writers").
-2. A body whose cut-link receiver is a non-fluid cell, e.g. a channel wall row
-   or the inlet/outlet column ("Cut links must have active fluid receivers").
-
-Both were previously accepted silently for passive bodies; the interpolated
-boundary then had undefined results. The rules apply to the cut links of all
-bodies together, so a passive body overlapping a Jet body is rejected as well.
-The check runs in `Simulation.initialize`, runtime body sync (`sync_bodies`),
-checkpoint save and checkpoint load (`common/checkpoint.py`). Load validates
-before restoring anything, so a checkpoint whose saved geometry violates either
-rule no longer restores; rebuild the geometry with separated bodies.
-
-## 87c3a31: `Simulation.read_surface_exchange` accepts Jet-driven bodies
-
-`read_surface_exchange(id)` previously required `SurfaceActuation` on the body.
-It now also accepts bodies that own Jet aperture links, including Jets with zero
-or negative amplitude (the aperture test is geometric). Bodies with neither
-`SurfaceActuation` nor aperture links still raise
-`ValueError("Body has no surface actuation")`. The value is unchanged: the latest
-step's wall/aperture exchange, not total nozzle force.
@@ -1,101 +0,0 @@
-"""Synthetic checks of the annulus mass ledger. Host-only, stdlib; not solver evidence."""
-
-import random
-import sys
-from pathlib import Path
-import unittest
-
-sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "validation"))
-import annulus_ledger as al  # noqa: E402
-
-NX = NY = 24
-# Annulus 4 <= r <= 8 around (12, 12); dyadic values keep every sum exact.
-A = {(x, y) for x in range(NX) for y in range(NY) if 16 <= (x - 12) ** 2 + (y - 12) ** 2 <= 64}
-DELTA = 0.375
-
-
-def owner(seed=1):
-    rng = random.Random(seed)
-    return [[[rng.randint(1, 1024) / 1024 for _ in range(NX)] for _ in range(NY)] for _ in range(9)]
-
-
-def collide(P, source=None):
-    """Mass-conserving stand-in: rotate populations within each cell (+ optional source)."""
-    D = [[[P[(q + 1) % 9][y][x] for x in range(NX)] for y in range(NY)] for q in range(9)]
-    if source:
-        (x, y), s = source
-        D[0][y][x] += s
-    return D
-
-
-def link(interior):
-    """(x, y, q) with x in A and x - c_q inside (seam) or outside (edge) A."""
-    for (x, y) in sorted(A):
-        for q in range(1, 9):
-            src = ((x - al.CX[q]) % NX, (y - al.CY[q]) % NY)
-            if (src in A) == interior:
-                return x, y, q
-    raise AssertionError
-
-
-class Ledger(unittest.TestCase):
-    def setUp(self):
-        self.O = owner()
-        self.P = al.pull(self.O, NX, NY)
-
-    def run_ledger(self, P, D):
-        return al.ledger(self.O, P, D, A, NX, NY)
-
-    def test_ideal_pull_has_no_h_and_closes(self):
-        r = self.run_ledger(self.P, collide(self.P))
-        self.assertEqual((r["H_seam"], r["H_edge"], r["C"], r["residual"]), (0, 0, 0, 0))
-        self.assertEqual(r["dM"], -r["Q_exact"])
-        self.assertNotEqual(r["Q_exact"], 0)
-
-    def test_collision_source_is_c(self):
-        x, y = sorted(A)[3]
-        r = self.run_ledger(self.P, collide(self.P, ((x, y), DELTA)))
-        self.assertEqual((r["C"], r["H"], r["residual"]), (DELTA, 0, 0))
-
-    def test_interior_seam_perturbation(self):
-        x, y, q = link(interior=True)
-        self.P[q][y][x] += DELTA
-        r = self.run_ledger(self.P, collide(self.P))
-        self.assertEqual((r["H_seam"], r["H_edge"], r["residual"]), (DELTA, 0, 0))
-        # Invisible to any flux through the edge of A; only H explains the mass change.
-        self.assertEqual(al.staged_flux(self.O, self.P, A, NX, NY), r["Q_exact"])
-        self.assertEqual(r["dM"] + r["Q_exact"], DELTA)
-
-    def test_edge_perturbation_and_staged_flux_double_count(self):
-        x, y, q = link(interior=False)
-        self.P[q][y][x] += DELTA
-        r = self.run_ledger(self.P, collide(self.P))
-        self.assertEqual((r["H_seam"], r["H_edge"], r["residual"]), (0, DELTA, 0))
-        staged = al.staged_flux(self.O, self.P, A, NX, NY)
-        self.assertEqual(staged, r["Q_exact"] - DELTA)
-        # Staged flux + full H counts the staging change twice.
-        self.assertEqual(r["dM"] + staged - r["C"] - r["H"], -DELTA)
-        # Staged flux alone closes only without the edge part of H.
-        self.assertEqual(r["dM"] + staged - r["C"] - r["H_seam"], 0)
-
-    def test_owner_snapshot_after_staging_hides_the_change(self):
-        # Snapshot taken after staging wrote the same slot into the owner image.
-        x, y, q = link(interior=False)
-        sx, sy = (x - al.CX[q]) % NX, (y - al.CY[q]) % NY
-        self.P[q][y][x] += DELTA
-        late = [[row[:] for row in plane] for plane in self.O]
-        late[q][sy][sx] += DELTA
-        r = al.ledger(late, self.P, collide(self.P), A, NX, NY)
-        self.assertEqual((r["H"], r["residual"]), (0, 0))
-
-    def test_wrong_pull_direction_is_flagged(self):
-        # Indexing error (push instead of pull): H is nonzero with no physical cause.
-        bad = [[[self.O[q][(y + al.CY[q]) % NY][(x + al.CX[q]) % NX] for x in range(NX)]
-                for y in range(NY)] for q in range(9)]
-        r = self.run_ledger(bad, collide(bad))
-        self.assertNotEqual(r["H"], 0)
-        self.assertEqual(r["residual"], 0)
-
-
-if __name__ == "__main__":
-    unittest.main()
@@ -1,75 +0,0 @@
-"""Per-hop mass ledger of a fluid cell set A (host-only, stdlib, D2Q9 pull scheme).
-
-Diagnostic definition only; no solver code, no GPU. Arrays are indexed [q][y][x] on a
-periodic nx x ny lattice; A is a set of (x, y) ACTIVE cells.
-
-  O  owner populations before any staging (canonical owner slots, start of the hop)
-  P  populations each cell actually pulls: P[q](x) is what x reads for direction q
-     (after halo copy, BFL/coalesce staging, ...)
-  D  populations written by the cell after collision
-
-With M0 = sum_{A,q} O and M1 = sum_{A,q} D (both owner snapshots of A):
-
-  M1 - M0 + Q_exact = C + H                                            (exact)
-
-  C       = sum_{x in A, q} (D_q(x) - P_q(x))              collision source
-  H       = sum_{x in A, q} (P_q(x) - O_q(x - c_q))        actual minus ideal owner input,
-            split into H_seam (x - c_q in A: internal pulls, e.g. tile seams) and
-            H_edge (x - c_q not in A: inputs crossing the edge of A)
-  Q_exact = sum_{y in A, y+c_q not in A} O_q(y) - sum_{y not in A, y+c_q in A} O_q(y)
-            owner-based net outward flux (outgoing minus incoming)
-
-Proof: sum_{A,q} O_q(x - c_q) - M0 = -Q_exact (relabel y = x - c_q per direction), and
-sum_{A,q} D = C + H + sum_{A,q} O_q(x - c_q). Q_exact must use O, not staged values: a
-flux built from staged inputs already contains H_edge, so adding the full H counts it
-twice (see staged_flux). The edge of A must not cross cut links, otherwise O at the solid
-end enters Q_exact.
-"""
-
-CX = (0, 1, -1, 0, 0, 1, -1, 1, -1)
-CY = (0, 0, 0, 1, -1, 1, -1, -1, 1)
-
-
-def _src(x, y, q, nx, ny):
-    return (x - CX[q]) % nx, (y - CY[q]) % ny
-
-
-def pull(O, nx, ny):
-    """Ideal pulled input: P[q](x) = O[q](x - c_q)."""
-    return [[[O[q][_src(x, y, q, nx, ny)[1]][_src(x, y, q, nx, ny)[0]]
-              for x in range(nx)] for y in range(ny)] for q in range(9)]
-
-
-def ledger(O, P, D, A, nx, ny):
-    A = set(A)
-    out = dict(dM=0.0, C=0.0, H_seam=0.0, H_edge=0.0, Q_exact=0.0)
-    for (x, y) in A:
-        for q in range(9):
-            out["dM"] += D[q][y][x] - O[q][y][x]
-            out["C"] += D[q][y][x] - P[q][y][x]
-            sx, sy = _src(x, y, q, nx, ny)
-            key = "H_seam" if (sx, sy) in A else "H_edge"
-            out[key] += P[q][y][x] - O[q][sy][sx]
-            if (sx, sy) not in A:
-                out["Q_exact"] -= O[q][sy][sx]          # incoming owner population
-            tx, ty = (x + CX[q]) % nx, (y + CY[q]) % ny
-            if (tx, ty) not in A:
-                out["Q_exact"] += O[q][y][x]            # outgoing owner population
-    out["H"] = out["H_seam"] + out["H_edge"]
-    out["residual"] = out["dM"] + out["Q_exact"] - out["C"] - out["H"]
-    return out
-
-
-def staged_flux(O, P, A, nx, ny):
-    """Outward flux with incoming terms from the staged inputs P (the wrong combination)."""
-    A = set(A)
-    q_out = 0.0
-    for (x, y) in A:
-        for q in range(9):
-            sx, sy = _src(x, y, q, nx, ny)
-            if (sx, sy) not in A:
-                q_out -= P[q][y][x]
-            tx, ty = (x + CX[q]) % nx, (y + CY[q]) % ny
-            if (tx, ty) not in A:
-                q_out += O[q][y][x]
-    return q_out
@@ -80,12 +80,11 @@ class ThreeLevelCPU(unittest.TestCase):
        f, g = run._levels[1:]
        b = g.boundary
        events, feeds = [], []
-        advance, gather = run._advance, run._gather_halo
-        def record_feed(index, buffer, **kw):  # index 2: L1 hop feeds L2
-            if index == 2:
-                self.assertIs(buffer, (f.current, f.spare)[len(feeds)])
-                feeds.append(buffer)
-            advance(index, buffer, **kw)
+        advance, gather = run._advance_child, run._gather_halo
+        def record_feed(buffer, **kw):
+            self.assertIs(buffer, (f.current, f.spare)[len(feeds)])
+            feeds.append(buffer)
+            advance(buffer, **kw)
        def record_halo(level, buffer, **kw):
            if level is g:
                events.append(("halo", buffer))
@@ -96,7 +95,7 @@ class ThreeLevelCPU(unittest.TestCase):
            self.assertIs(kind, g.kind)
            events.append(("step", src))
        g.kernel.side_effect = record_step
-        with patch.object(run, "_advance", side_effect=record_feed), \
+        with patch.object(run, "_advance_child", side_effect=record_feed), \
                patch.object(run, "_gather_halo", side_effect=record_halo):
            run.step()
        self.assertEqual(feeds, [f.current, f.spare])
@@ -106,8 +105,8 @@ class ThreeLevelCPU(unittest.TestCase):
        self.assertGreater(len(np.unique(b.tables["block"])), 1)
        # Negative control: reusing L1's old source at the midpoint must fail.
        feeds.clear()
-        with patch.object(run, "_advance", side_effect=lambda index, buffer, **kw:
-                          record_feed(index, f.current if index == 2 else buffer, **kw)):
+        with patch.object(run, "_advance_child", side_effect=lambda buffer, **kw:
+                          record_feed(f.current, **kw)):
            with self.assertRaises(AssertionError):
                run.step()
        run.close()
@@ -173,12 +172,11 @@ class ThreeLevelGPU(unittest.TestCase):
            p1, p2 = run.patch, run.child_patch
            run.initialize(state(16, 16, 1), state(*p1.fine_shape, 2), state(*p2.fine_shape, 3))
            f = run._levels[1]
-            feeds, original = [], run._advance
-            def record(index, buffer, **kw):  # index 2: L1 hop feeds L2
-                if index == 2:
-                    feeds.append(int(buffer))
-                original(index, buffer, **kw)
-            with patch.object(run, "_advance", side_effect=record):
+            feeds, original = [], run._advance_child
+            def record(buffer, **kw):
+                feeds.append(int(buffer))
+                original(buffer, **kw)
+            with patch.object(run, "_advance_child", side_effect=record):
                cur, spare = int(f.current), int(f.spare)
                run.step(1)
            self.assertEqual(feeds, [cur, spare])  # current G, then the new G of hop 1