research: VIV driver reads diagnostics by MovingTables.DIAG name; coupling-contract README

Index values are unchanged (1, 3, 7, 10, 13); the run recorded at fb507aa1
used the same columns. Research-only.

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

src/CelerisLab/body/motion.py

@@ -1,10 +1,10 @@
 # CelerisLab/body/motion.py
-"""Translation targets and device tables (dense double buffer, D2Q9 only).
+"""Host state of a translating circle (dense double buffer, D2Q9 only).
 
 Contract: docs/moving_bodies.md, sections 4a, 5 and 7. The applied geometry
 (last successfully committed offset) and an optional pending target are two
-separate facts; centres are always ``base + offset``. MovingCircle owns host
-targets, MovingTables owns device records, and Simulation commits completed steps.
+separate facts; centres are always ``base + offset``. Nothing here touches
+the device: the caller supplies upload/step/synchronize callables.
 """
 from __future__ import annotations
 
@@ -18,6 +18,10 @@ R_MIN = 4.0        # smallest radius covered by the donor enumeration
 
 Vec = Tuple[float, float]
 
+# D2Q9 non-rest directions, paired order of lbm/descriptors.
+_CX = (1, -1, 0, 0, 1, -1, 1, -1)
+_CY = (0, 0, 1, -1, 1, -1, -1, 1)
+
 
 def _vec(value, name: str) -> Vec:
     x, y = (float(v) for v in value)
@@ -94,6 +98,23 @@ class MovingCircle:
             offset, velocity = self.applied, (0.0, 0.0)
         return self.centre(self.applied), self.centre(offset), velocity
 
+    def advance(self, upload: Callable, step: Callable, synchronize: Callable) -> None:
+        """Upload the record, run one step and commit only after a successful sync.
+
+        Any exception after the upload is attempted leaves device geometry
+        unknown: the circle is marked failed, never reports applied state again,
+        and the owner's ``on_failure`` hook is called.
+        """
+        record = self.record()
+        try:
+            upload(record)
+            step()
+            synchronize()
+        except BaseException as exc:
+            self.fail(exc)
+            raise
+        self.commit()
+
     def commit(self) -> None:
         """Applied := pending offset (caller has synchronized a successful step)."""
         if self.pending is not None:
@@ -117,6 +138,43 @@ class MovingCircle:
             raise RuntimeError("moving circle state is unknown after a failed step")
 
 
+def _inside(x: int, y: int, c: Vec, r: float) -> bool:
+    return (x - c[0]) ** 2 + (y - c[1]) ** 2 < r * r
+
+
+def empty_donor_cells(c_old: Vec, c_new: Vec, radius: float):
+    """Fresh cells (inside c_old, outside c_new) whose donor set V(x) is empty."""
+    r = radius
+    xs = range(math.floor(min(c_old[0], c_new[0]) - r) - 1, math.ceil(max(c_old[0], c_new[0]) + r) + 2)
+    ys = range(math.floor(min(c_old[1], c_new[1]) - r) - 1, math.ceil(max(c_old[1], c_new[1]) + r) + 2)
+    empty = []
+    for x in xs:
+        for y in ys:
+            if not _inside(x, y, c_old, r) or _inside(x, y, c_new, r):
+                continue
+            if not any(not _inside(x + cx, y + cy, c_old, r) and not _inside(x + cx, y + cy, c_new, r)
+                       for cx, cy in zip(_CX, _CY)):
+                empty.append((x, y))
+    return empty
+
+
+def enumerate_donor_coverage(radius: float, positions: int = 8, directions: int = 16) -> int:
+    """Count empty-donor fresh cells over sub-cell centres and moves of MAX_STEP.
+
+    Evidence for the declared bounds (radius >= R_MIN, |dc|_2 <= MAX_STEP),
+    not a proof; the device keeps the ``no_donor`` counter.
+    """
+    count = 0
+    for i in range(positions):
+        for j in range(positions):
+            c_old = (i / positions, j / positions)
+            for k in range(directions):
+                a = 2.0 * math.pi * k / directions
+                c_new = (c_old[0] + MAX_STEP * math.cos(a), c_old[1] + MAX_STEP * math.sin(a))
+                count += len(empty_donor_cells(c_old, c_new, radius))
+    return count
+
+
 class MovingTables:
     """Device tables of the translating circles for ``MovingCircleCells``/``Links``.
 

src/CelerisLab/lbm/stepper.py

@@ -21,7 +21,7 @@ class LBMStepper:
 
         self._esopull = (cfg.streaming == "esopull")
         # Set by the owner when a step would be wrong (translating circles
-        # without attached moving tables, a pending target, or a failed step).
+        # excluded from the packed links before the moving kernels exist).
         self.blocked: str | None = None
         self.moving = None  # MovingTables attached by the owner (translating circles)
 
@@ -39,10 +39,8 @@ class LBMStepper:
         self.apply_esopull_curved_fn = module.get_function("ApplyEsoPullCurvedKernel")
         self.sensor_fn = module.get_function("SensorKernel")
         self.force_region_fn = module.get_function("ForceRegionKernel")
-        # Moving-circle kernels exist only in D2Q9 double-buffer modules.
-        moving = not self._esopull and cfg.is_d2q9
-        self.moving_fn = module.get_function("MovingCircleLinks") if moving else None
-        self.moving_cells_fn = module.get_function("MovingCircleCells") if moving else None
+        self.moving_fn = None if self._esopull else module.get_function("MovingCircleLinks")
+        self.moving_cells_fn = None if self._esopull else module.get_function("MovingCircleCells")
 
         tpb = cfg.threads_per_block
         self.block = (tpb, 1, 1)

src/CelerisLab/simulation.py

@@ -242,8 +242,7 @@ class Simulation:
         self._require_usable()
         tables = self.stepper.moving
         if tables is None:
-            from .body.motion import MovingTables
-            return np.zeros((0, MovingTables.DIAG_FLOATS), np.float32)
+            return np.zeros((0, 6), np.float32)
         self.stream.synchronize()
         return tables.download_diagnostics()
 
@@ -786,7 +785,6 @@ class Simulation:
         (docs/moving_bodies.md 4a, 5). Any failure leaves the Simulation unusable."""
         tables = self.stepper.moving
         circles = [(obj.motion, obj.obj_id) for obj in self._moving_bodies()]
-        pending = any(motion.pending is not None for motion, _ in circles)
         try:
             advancing = tables.upload_records(circles, stream)
             self.stepper._step(steps, action_gpu=self.bodies.action_gpu,
@@ -799,12 +797,10 @@ class Simulation:
                 no_donor = tables.download_diagnostics()[:, tables.DIAG.index("no_donor")]
                 if np.any(no_donor > 0):
                     raise RuntimeError("translating circle refill found no donor cell")
-            if pending:  # a target equal to the applied offset is consumed as a hold step
                 for motion, _ in circles:
                     motion.commit()
-                if advancing:
-                    tables.upload_records(circles, stream)  # first hold: old = new = applied, u = 0
-                    stream.synchronize()
+                tables.upload_records(circles, stream)  # first hold: old = new = applied, u = 0
+                stream.synchronize()
                 self._attach_moving()
         except BaseException as exc:
             for motion, _ in circles:  # device geometry/state unknown after any failure

tests/unit/test_moving_bodies.py

@@ -1,13 +1,14 @@
-"""Translating-circle host contract (docs/moving_bodies.md 4a, 5, 7).
+"""Translating-circle host contract, S1 (docs/moving_bodies.md 4a, 5, 7).
 
-Real Simulation entrypoints with device doubles; no CUDA context or GPU work.
+Host doubles only: no kernels exist yet, so every positive-step run with a
+moving body must be rejected before any device write.
 """
-import math
 from types import SimpleNamespace
 
 import pytest
 
-from CelerisLab.body.motion import MAX_STEP, R_MIN, MovingCircle, Translation
+from CelerisLab.body.motion import (MAX_STEP, R_MIN, MovingCircle, Translation,
+                                    enumerate_donor_coverage)
 
 BAD = (float("nan"), float("inf"))
 
@@ -45,96 +46,46 @@ def test_targets_validate_against_applied_and_last_wins():
     assert c.applied == (0., 0.)
 
 
-def test_record_commit_and_hold(monkeypatch):
-    sim, writes, c = routed_advancing_simulation(monkeypatch)
-    assert c.record() == ((50., 40.), (50., 40.), (0., 0.))
-    sim.set_body(0, offset=(0., .25), vy=.05)
+def test_record_commit_and_hold():
+    c = circle()
+    assert c.record() == ((50., 40.), (50., 40.), (0., 0.))  # hold: no translation velocity
+    c.set_target((0., .25), (0., .05))
     assert c.record() == ((50., 40.), (50., 40.25), (0., .05))
-    seen_at_sync = []
-    sim.run(1, stream=SimpleNamespace(synchronize=lambda: seen_at_sync.append(c.applied)))
-    assert seen_at_sync == [(0., 0.), (0., .25)]  # commit between step sync and hold sync
-    uploads = [w[1][0] for w in writes if isinstance(w, tuple) and w[0] == "upload"]
-    assert uploads == [((50., 40.), (50., 40.25), (0., .05)),
-                       ((50., 40.25), (50., 40.25), (0., 0.))]
+    uploads = []
+    c.advance(uploads.append, lambda: None, lambda: None)
+    assert uploads == [((50., 40.), (50., 40.25), (0., .05))]
     assert c.applied == (0., .25) and c.pending is None  # offset-only commit
-    sim.run(1, stream=STREAM)  # no new target: hold step
-    uploads = [w[1][0] for w in writes if isinstance(w, tuple) and w[0] == "upload"]
+    c.advance(uploads.append, lambda: None, lambda: None)  # no new target: hold step
     assert uploads[-1] == ((50., 40.25), (50., 40.25), (0., 0.)) and c.applied == (0., .25)
-    sim.set_body(0, offset=(0., 0.), vy=-.05)  # zero offset is a real target
+    c.set_target((0., 0.), (0., -.05))  # return to zero offset is a real target
     assert c.pending == ((0., 0.), (0., -.05))
-    sim.run(1, stream=STREAM)
+    c.advance(uploads.append, lambda: None, lambda: None)
     assert c.applied == (0., 0.) and c.pending is None
 
 
 @pytest.mark.parametrize("stage", ("upload", "step", "synchronize"))
-def test_failure_after_upload_attempt_is_fail_closed(monkeypatch, stage):
-    sim, writes, c = routed_advancing_simulation(monkeypatch, fail="upload" if stage == "upload" else None)
-    sim.set_body(0, offset=(0., .25), vy=.05)
+def test_failure_after_upload_attempt_is_fail_closed(stage):
+    c = circle()
+    c.set_target((0., .25), (0., .05))
 
-    def fail(*a, **k):
-        raise RuntimeError(f"{stage} failed")
-    if stage == "step":
-        sim.stepper._step = fail
-    stream = SimpleNamespace(synchronize=fail) if stage == "synchronize" else STREAM
+    def call(name):
+        def run(*_):
+            if name == stage:
+                raise RuntimeError(f"{name} failed")
+        return run
     with pytest.raises(RuntimeError, match=f"{stage} failed"):
-        sim.run(1, stream=stream)
-    assert c.failed and c.applied == (0., 0.) and c.pending == ((0., .25), (0., .05))
-    assert sim._failed is not None and sim.stepper.blocked == sim._failed
-    for later in (c.record, lambda: sim.set_body(0, offset=(0., .1), vy=0.)):
+        c.advance(call("upload"), call("step"), call("synchronize"))
+    assert c.failed and c.applied == (0., 0.)
+    for later in (c.record, lambda: c.set_target((0., .1), (0., 0.)),
+                  lambda: c.advance(*(lambda *_: None,) * 3)):
         with pytest.raises(RuntimeError, match="unknown"):
             later()
-    with pytest.raises(RuntimeError, match="Simulation failed"):
-        sim.run(1, stream=STREAM)
-    with pytest.raises(RuntimeError, match="moving circle step failed"):
-        sim.stepper.step(1, action_gpu=None, obs_gpu=None)
 
 
 def test_band_covers_offset_box_with_margin():
     assert circle().band() == (50 - 2 - 10 - 2, 50 + 2 + 10 + 2, 40 - 3 - 10 - 2, 40 + 3 + 10 + 2)
 
 
-# D2Q9 non-rest directions, paired order of lbm/descriptors.
-_CX = (1, -1, 0, 0, 1, -1, 1, -1)
-_CY = (0, 0, 1, -1, 1, -1, -1, 1)
-
-
-def _inside(x: int, y: int, c, r: float) -> bool:
-    return (x - c[0]) ** 2 + (y - c[1]) ** 2 < r * r
-
-
-def empty_donor_cells(c_old, c_new, radius: float):
-    """Fresh cells (inside c_old, outside c_new) whose donor set V(x) is empty."""
-    r = radius
-    xs = range(math.floor(min(c_old[0], c_new[0]) - r) - 1, math.ceil(max(c_old[0], c_new[0]) + r) + 2)
-    ys = range(math.floor(min(c_old[1], c_new[1]) - r) - 1, math.ceil(max(c_old[1], c_new[1]) + r) + 2)
-    empty = []
-    for x in xs:
-        for y in ys:
-            if not _inside(x, y, c_old, r) or _inside(x, y, c_new, r):
-                continue
-            if not any(not _inside(x + cx, y + cy, c_old, r) and not _inside(x + cx, y + cy, c_new, r)
-                       for cx, cy in zip(_CX, _CY)):
-                empty.append((x, y))
-    return empty
-
-
-def enumerate_donor_coverage(radius: float, positions: int = 8, directions: int = 16) -> int:
-    """Count empty-donor fresh cells over sub-cell centres and moves of MAX_STEP.
-
-    Evidence for the declared bounds (radius >= R_MIN, |dc|_2 <= MAX_STEP),
-    not a proof; the device keeps the ``no_donor`` counter.
-    """
-    count = 0
-    for i in range(positions):
-        for j in range(positions):
-            c_old = (i / positions, j / positions)
-            for k in range(directions):
-                a = 2.0 * math.pi * k / directions
-                c_new = (c_old[0] + MAX_STEP * math.cos(a), c_old[1] + MAX_STEP * math.sin(a))
-                count += len(empty_donor_cells(c_old, c_new, radius))
-    return count
-
-
 @pytest.mark.parametrize("radius", (R_MIN, 4.3, 7.5, 12.0))
 def test_refill_donor_sets_nonempty_within_bounds(radius):
     # Evidence for R >= R_MIN and |dc|_2 <= MAX_STEP on a fixed sample, not a proof.
@@ -327,26 +278,6 @@ def advancing_simulation(monkeypatch, **tables):
     return sim, writes, sim.bodies.objects[0].motion
 
 
-def routed_advancing_simulation(monkeypatch, **tables):
-    """Real body registration, target routing, run and owner hook; double device I/O."""
-    sim = routed_simulation(monkeypatch)
-    bid = sim.add_body("circle", (50., 40.), 10., translation=Translation((-2., 2., -3., 3.), .1))
-    sim._initialized = True
-    sim._assert_runtime_contracts = lambda: None
-    writes = []
-    monkeypatch.setattr(sim.bodies, "_upload_action_async", lambda s: writes.append("action"))
-    monkeypatch.setattr(sim.bodies, "zero_obs_async", lambda s: writes.append("obs"))
-    monkeypatch.setattr(sim.bodies, "download_obs_full_async", lambda s: writes.append("download"))
-    sim.stepper = SimpleNamespace(
-        moving=Tables(writes, **tables), blocked=None,
-        _step=lambda n, advance=False, **k: writes.append(("step", n, advance)))
-    from CelerisLab.lbm.stepper import LBMStepper
-    sim.stepper.step = LBMStepper.step.__get__(sim.stepper)
-    sim._moving_tables = sim.stepper.moving
-    sim._attach_moving()
-    return sim, writes, sim.bodies.get(bid).motion
-
-
 def test_run_one_advances_and_commits_after_sync(monkeypatch):
     sim, writes, motion = advancing_simulation(monkeypatch)
     motion.set_target((0., .25), (0., .05))
@@ -366,44 +297,6 @@ def test_run_one_advances_and_commits_after_sync(monkeypatch):
     assert ("step", 3, False) in writes and motion.applied == (0., .25)
 
 
-def test_zero_motion_target_is_consumed_by_run_one(monkeypatch):
-    sim, writes, motion = advancing_simulation(monkeypatch)
-    motion.set_target((0., 0.), (0., 0.))          # equal to applied: a hold target
-    sim._attach_moving()
-    assert sim.stepper.blocked == module.PENDING_BLOCKED
-    sim.run(1, stream=STREAM)
-    assert motion.pending is None and motion.applied == (0., 0.)
-    assert sim.stepper.blocked is None and ("step", 1, False) in writes
-    writes.clear()
-    sim.run(3, stream=STREAM)                      # no stale latch
-    assert ("step", 3, False) in writes
-
-
-@pytest.mark.parametrize("streaming,d2q9,bound", [("double_buffer", True, True),
-                                                  ("double_buffer", False, False),
-                                                  ("esopull", True, False)])
-def test_moving_kernels_bound_only_for_d2q9_double_buffer(monkeypatch, streaming, d2q9, bound):
-    from CelerisLab.lbm import stepper as stepper_module
-    monkeypatch.setattr(stepper_module.cuda, "mem_alloc", lambda n: object())
-    moving = {"MovingCircleLinks", "MovingCircleCells"}
-
-    def get_function(name):
-        if name in moving and not (d2q9 and streaming == "double_buffer"):
-            raise KeyError(name)                 # absent from D3Q19 / EsoPull modules
-        return lambda *a, **k: None
-    cfg = SimpleNamespace(streaming=streaming, threads_per_block=128, is_d2q9=d2q9,
-                          nx=16, ny=8, nz=4, nq=9 if d2q9 else 19)
-    stepper = stepper_module.LBMStepper(SimpleNamespace(), SimpleNamespace(get_function=get_function), cfg)
-    assert (stepper.moving_fn is not None) == bound and (stepper.moving_cells_fn is not None) == bound
-
-
-def test_moving_diagnostics_empty_shape_matches_layout(monkeypatch):
-    from CelerisLab.body.motion import MovingTables
-    sim = make_simulation(monkeypatch)
-    sim._failed, sim.stepper = None, SimpleNamespace(moving=None)
-    assert sim.moving_diagnostics().shape == (0, MovingTables.DIAG_FLOATS)
-
-
 @pytest.mark.parametrize("failure", ("upload", "first_sync", "no_donor", "hold_upload",
                                      "hold_sync"))
 def test_failed_advance_fails_simulation(monkeypatch, failure):
@@ -614,21 +507,23 @@ def test_moving_circle_excluded_from_static_links():
 
 
 def test_failed_moving_step_marks_simulation_failed(monkeypatch):
-    sim, writes, motion = routed_advancing_simulation(monkeypatch)
+    sim = routed_simulation(monkeypatch)
     sim.ctx = ClosableOwner()
-    sim.set_body(0, offset=(0., .2), vy=.05)
+    bid = sim.add_body("circle", (50., 40.), 10., translation=Translation((0., 0., -1., 1.), .1))
+    motion = sim.bodies.get(bid).motion
+    motion.set_target((0., .2), (0., .05))
 
     def fail(*_):
         raise RuntimeError("sync failed")
     with pytest.raises(RuntimeError, match="sync failed"):
-        sim.run(1, stream=SimpleNamespace(synchronize=fail))
+        motion.advance(lambda *_: None, lambda: None, fail)
     assert sim._failed is not None and motion.applied == (0., 0.)
     for call in (lambda: sim.run(0), lambda: sim.run(1), sim.step, sim.sync_bodies,
                  sim.save_checkpoint, lambda: sim.load_checkpoint("x.h5")):
         with pytest.raises(RuntimeError, match="Simulation failed"):
             call()
     with pytest.raises(RuntimeError, match="unknown"):  # no retry on the failed circle
-        sim.set_body(0, offset=(0., .1), vy=.05)
+        motion.advance(lambda *_: None, lambda: None, lambda: None)
     with sim.ctx.activate():  # a refused close keeps the failed state and resources
         with pytest.raises(RuntimeError, match="active CUDA"):
             sim.close()