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>

docs/compatibility.md

@@ -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.

src/CelerisLab/body/jets.py

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

src/CelerisLab/lbm/curved_links.py

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

src/CelerisLab/lbm/refinement_tables.py

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

tests/validation/multigrid/test_refinement_three_level.py

@@ -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