Wait for setup vibrations before checking player sculk movement

tests/test_server_player_sculk_setup.py

@@ -0,0 +1,42 @@
+"""The walking harness must wait out setup vibrations before measuring movement."""
+import unittest
+
+from tools.test_server_player_sculk_parity import wait_for_quiet_sensor
+
+
+class SensorTrace:
+    def __init__(self, phases):
+        self.phases = phases
+        self.tick = 0
+
+    def wait(self, ticks):
+        self.tick += ticks
+
+    def phase(self, _host):
+        phase = next(value for start, value in reversed(self.phases) if start <= self.tick)
+        return f'minecraft:sculk_sensor[sculk_sensor_phase={phase}]'
+
+
+class QuietSensorTests(unittest.TestCase):
+    def test_waits_out_an_in_flight_vibration(self):
+        # VanillaSculkOracle's sensor_step trace: selected at 2, arrives at 5,
+        # active for 30 ticks, cooldown for 10. Tick 0 alone looks ready.
+        sensor = SensorTrace([(0, 'inactive'), (5, 'active'), (35, 'cooldown'), (45, 'inactive')])
+        wait_for_quiet_sensor(sensor, sensor.phase)
+        self.assertEqual(sensor.tick, 54)
+
+    def test_requires_a_fresh_quiet_window_after_each_vibration(self):
+        sensor = SensorTrace([(0, 'active'), (30, 'cooldown'), (40, 'inactive'),
+                              (48, 'active'), (78, 'cooldown'), (88, 'inactive')])
+        wait_for_quiet_sensor(sensor, sensor.phase)
+        self.assertEqual(sensor.tick, 97)
+
+    def test_does_not_wait_forever_for_a_noisy_setup(self):
+        sensor = SensorTrace([(0, 'active')])
+        with self.assertRaisesRegex(AssertionError, 'sensor did not settle during setup'):
+            wait_for_quiet_sensor(sensor, sensor.phase)
+        self.assertEqual(sensor.tick, 100)
+
+
+if __name__ == '__main__':
+    unittest.main()

tools/test_server_player_sculk_parity.py

@@ -10,9 +10,6 @@ import math
 import subprocess
 import threading
 
-import lab
-import lan_e2e as lan
-
 
 def until(game, observe, predicate, ticks=100):
     value = observe()
@@ -25,7 +22,24 @@ def until(game, observe, predicate, ticks=100):
     return value
 
 
+def wait_for_quiet_sensor(host, phase, ticks=100):
+    # Inactive does not mean idle: a selected vibration still has to travel to
+    # the sensor. Require a quiet window longer than its eight-block range,
+    # letting any setup vibration finish its active and cooldown phases first.
+    quiet = 0
+    state = None
+    for _ in range(ticks):
+        host.wait(1)
+        state = phase(host)
+        quiet = quiet + 1 if 'sculk_sensor_phase=inactive' in state else 0
+        if quiet == 10:
+            return
+    raise AssertionError(f'sensor did not settle during setup: {state}')
+
+
 def check_walk(host, guest, out, role):
+    import lan_e2e as lan
+
     player = host if role == 'host' else guest
     hx, y, hz = (math.floor(v) for v in lan.pos_of(host))
     x, z = hx + 6, hz + (-8 if role == 'host' else 8)
@@ -41,9 +55,7 @@ def check_walk(host, guest, out, role):
     host('command', command=f'setblock {sensor[0]} {sensor[1]} {sensor[2]} minecraft:sculk_sensor')
     phase = lambda game: game.raw('getblock', *sensor)['block']
     try:
-        until(host, lambda: phase(host), lambda s: 'sculk_sensor_phase=inactive' in s)
-        player.wait(10)
-        assert 'sculk_sensor_phase=inactive' in phase(host), 'setup activated the sensor'
+        wait_for_quiet_sensor(host, phase)
         # The sensor stays active 30 ticks. Each client samples it every tick from inside one ctl script, so a
         # runner that answers control requests slowly (frames drawn on the CPU) can't miss that window between
         # separate requests; the other client samples alongside the walk.
@@ -73,6 +85,9 @@ def check_walk(host, guest, out, role):
 
 
 def main():
+    import lab
+    import lan_e2e as lan
+
     subprocess.run(['cargo', 'build', '--locked', '--profile', 'fast', '-p', 'horizon-client', '-p', 'horizon-client-lib',
                     '--bin', 'horizon-client', '--bin', 'horizon-ctl'], cwd=lab.ROOT, check=True)
     metadata = json.loads(subprocess.check_output(