test: cross parallel tick helper thresholds with real resources

Compare deferred and in-place metadata for 2,085 varied mobs, including pack-selected effect particles and switch-off vector reuse. Exercise live climbing, snapshot order and nearest-player invalidation for 293 mobs. Build 132 random-tick section masks ahead and retain exact event, state and random comparisons through edits and reloads.

Co-Authored-By: Kacper <[email protected]>
Co-Authored-By: GPT 6.1 Sol <[email protected]>
(cherry picked from commit 41a579d428)

Cargo.lock

@@ -706,6 +706,25 @@ dependencies = [
  "cfg-if",
 ]
 
+[[package]]
+name = "crossbeam-deque"
+version = "0.8.8"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "622f3fc73690be383c7214310406f28a90e6edeadc3cea882f9d71e495b9711a"
+dependencies = [
+ "crossbeam-epoch",
+ "crossbeam-utils",
+]
+
+[[package]]
+name = "crossbeam-epoch"
+version = "0.9.21"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "dc74980687109a3b14c72fd458107bf0baa1da1a1a805e178d15501ba9b86d9d"
+dependencies = [
+ "crossbeam-utils",
+]
+
 [[package]]
 name = "crossbeam-queue"
 version = "0.3.14"
@@ -926,6 +945,12 @@ version = "1.0.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "92773504d58c093f6de2459af4af33faa518c13451eb8f2b5698ed3d36e7c813"
 
+[[package]]
+name = "either"
+version = "1.19.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "0e9c71c2167ca323c882b99918929403426e2373ea17242ff5653e0d5e1058be"
+
 [[package]]
 name = "embedded-io"
 version = "0.4.0"
@@ -1554,6 +1579,7 @@ dependencies = [
  "open",
  "png",
  "pollster",
+ "rayon",
  "rustix 1.1.5",
  "serde",
  "serde_json",
@@ -3654,6 +3680,26 @@ dependencies = [
  "objc2-quartz-core 0.3.2",
 ]
 
+[[package]]
+name = "rayon"
+version = "1.12.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "fb39b166781f92d482534ef4b4b1b2568f42613b53e5b6c160e24cfbfa30926d"
+dependencies = [
+ "either",
+ "rayon-core",
+]
+
+[[package]]
+name = "rayon-core"
+version = "1.13.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "22e18b0f0062d30d4230b2e85ff77fdfe4326feb054b9783a3460d8435c8ab91"
+dependencies = [
+ "crossbeam-deque",
+ "crossbeam-utils",
+]
+
 [[package]]
 name = "rc2"
 version = "0.8.1"

crates/horizon-client-lib/Cargo.toml

@@ -27,6 +27,9 @@ horizon-mcp = { path = "../horizon-mcp" }
 # The server thread ahead of the workers (MinecraftServer.spin's Thread.setPriority).
 horizon-optifine = { path = "../horizon-optifine" }
 horizon-threads = { path = "../horizon-threads" }
+# One pool of helper threads for the server tick's parallel phases (`parallel.rs`): threads made
+# once, not for every phase of every tick.
+rayon = "1.10"
 horizon-hdr = { path = "../horizon-hdr" }
 horizon-world = { path = "../horizon-world" }
 horizon-protocol = { path = "../horizon-protocol" }

crates/horizon-client-render/src/poi_sections.rs

@@ -7,7 +7,10 @@
 //! sections are immutable behind `Arc` and edited through `Arc::make_mut`, so
 //! a weak reference tells an edited or replaced section from the one the list
 //! was built from (see `random_tick_sections`), without keeping its blocks.
-use crate::{chunk::Section, fast_hash::FxHashMap};
+use crate::{
+    chunk::{Chunk, Section},
+    fast_hash::FxHashMap,
+};
 use std::sync::{Arc, Mutex, Weak};
 
 /// A section's POIs: the block index in the section (`y << 8 | z << 4 | x`)
@@ -18,13 +21,16 @@ struct Cached<T> {
     section: Weak<Section>,
     pois: SectionPois<T>,
 }
-/// Cached POI lists by section, `(chunk x, section index, chunk z)`.
+/// Cached POI lists by chunk column `(chunk x, chunk z)`, each by section
+/// index: a search through a column looks it up once.
 pub struct PoiSections<T> {
     sections: Mutex<Lists<T>>,
 }
 struct Lists<T> {
-    by_section: FxHashMap<(i32, i32, i32), Cached<T>>,
-    /// Size above which lists of sections no longer resident are dropped.
+    by_column: FxHashMap<(i32, i32), Vec<Option<Cached<T>>>>,
+    /// How many section lists are kept, and the count above which the lists
+    /// of sections no longer resident are dropped.
+    len: usize,
     prune_at: usize,
 }
 const MIN_PRUNE: usize = 4096;
@@ -32,27 +38,61 @@ impl<T> Default for PoiSections<T> {
     fn default() -> Self {
         Self {
             sections: Mutex::new(Lists {
-                by_section: FxHashMap::default(),
+                by_column: FxHashMap::default(),
+                len: 0,
                 prune_at: MIN_PRUNE,
             }),
         }
     }
 }
-impl<T: Copy> PoiSections<T> {
-    /// The POIs of `section`, whose block states `kind` classifies. `kind`
-    /// must give the same answer for a state on every call.
-    pub fn get(
-        &self,
-        key: (i32, i32, i32),
-        section: &Arc<Section>,
-        mut kind: impl FnMut(u32) -> Option<T>,
-    ) -> SectionPois<T> {
-        let mut lists = self.sections.lock().unwrap_or_else(|e| e.into_inner());
-        if let Some(cached) = lists.by_section.get(&key)
-            && cached.section.as_ptr() == Arc::as_ptr(section)
-        {
-            return cached.pois.clone();
+impl<T: Copy> Lists<T> {
+    /// The column's lists, for sections up to `last`, after dropping the
+    /// lists of sections no longer resident if there are too many.
+    fn column(
+        &mut self,
+        column: (i32, i32),
+        last: usize,
+    ) -> (&mut [Option<Cached<T>>], &mut usize) {
+        if self.len >= self.prune_at {
+            self.prune();
         }
+        let lists = self.by_column.entry(column).or_default();
+        if lists.len() <= last {
+            lists.resize_with(last + 1, || None);
+        }
+        (lists, &mut self.len)
+    }
+
+    /// Drops the lists of sections no longer resident.
+    fn prune(&mut self) {
+        let mut len = 0;
+        self.by_column.retain(|_, lists| {
+            for c in lists.iter_mut() {
+                if c.as_ref().is_some_and(|c| c.section.strong_count() == 0) {
+                    *c = None;
+                }
+            }
+            let kept = lists.iter().filter(|c| c.is_some()).count();
+            len += kept;
+            kept > 0
+        });
+        self.len = len;
+        self.prune_at = (len * 2).max(MIN_PRUNE);
+    }
+}
+
+/// The POIs of `section` from `slot`, listed again (and kept there) unless
+/// `slot` was listed from this very section; `len` counts the slots filled.
+fn list<'a, T: Copy>(
+    slot: &'a mut Option<Cached<T>>,
+    len: &mut usize,
+    section: &Arc<Section>,
+    kind: &mut impl FnMut(u32) -> Option<T>,
+) -> &'a SectionPois<T> {
+    let fresh = slot
+        .as_ref()
+        .is_some_and(|c| c.section.as_ptr() == Arc::as_ptr(section));
+    if !fresh {
         let pois: SectionPois<T> = if section.blocks.maybe_has(|v| kind(v).is_some()) {
             let mut states = [0u32; 4096];
             section.unpack(&mut states);
@@ -64,18 +104,56 @@ impl<T: Copy> PoiSections<T> {
         } else {
             Arc::from([])
         };
-        if lists.by_section.len() >= lists.prune_at {
-            lists.by_section.retain(|_, c| c.section.strong_count() > 0);
-            lists.prune_at = (lists.by_section.len() * 2).max(MIN_PRUNE);
+        *len += usize::from(slot.is_none());
+        *slot = Some(Cached {
+            section: Arc::downgrade(section),
+            pois,
+        });
+    }
+    &slot.as_ref().expect("just listed").pois
+}
+
+impl<T: Copy> PoiSections<T> {
+    /// The POIs of `section`, the `key.1`-th section of the column
+    /// `(key.0, key.2)`, whose block states `kind` classifies. `kind` must give
+    /// the same answer for a state on every call.
+    pub fn get(
+        &self,
+        key: (i32, i32, i32),
+        section: &Arc<Section>,
+        mut kind: impl FnMut(u32) -> Option<T>,
+    ) -> SectionPois<T> {
+        let mut lists = self.sections.lock().unwrap_or_else(|e| e.into_inner());
+        let s = usize::try_from(key.1).expect("a section index");
+        let (column, len) = lists.column((key.0, key.2), s);
+        list(&mut column[s], len, section, &mut kind).clone()
+    }
+
+    /// The POIs of `chunk` from block Y `y0` to `y1`, its first section at
+    /// `min_y`, as [`Self::get`] gives them: each section that has any, with
+    /// the Y of its lowest blocks, in order, put in `out`. One lookup of the
+    /// column.
+    pub fn column(
+        &self,
+        chunk: &Chunk,
+        min_y: i32,
+        (y0, y1): (i32, i32),
+        mut kind: impl FnMut(u32) -> Option<T>,
+        out: &mut Vec<(i32, SectionPois<T>)>,
+    ) {
+        out.clear();
+        let (first, last) = ((y0 - min_y) >> 4, (y1 - min_y) >> 4);
+        if first > last {
+            return;
+        }
+        let mut lists = self.sections.lock().unwrap_or_else(|e| e.into_inner());
+        let (slots, len) = lists.column((chunk.x, chunk.z), last as usize);
+        for s in first as usize..=last as usize {
+            let pois = list(&mut slots[s], len, &chunk.sections[s], &mut kind);
+            if !pois.is_empty() {
+                out.push((min_y + s as i32 * 16, pois.clone()));
+            }
         }
-        lists.by_section.insert(
-            key,
-            Cached {
-                section: Arc::downgrade(section),
-                pois: pois.clone(),
-            },
-        );
-        pois
     }
 }
 
@@ -143,6 +221,48 @@ mod tests {
         assert!(index.get((0, 0, 0), &chunk.sections[0], kind).is_empty());
     }
 
+    /// A column gives each section's list as `get` does, the sections
+    /// without any left out, and keeps them until a section changes.
+    #[test]
+    fn a_column_gives_what_each_section_gives() {
+        let index = PoiSections::<u8>::default();
+        let kind = |v: u32| (v == 5).then_some(5);
+        let mut chunk = Arc::new(Chunk::empty(2, -3));
+        for (x, y, z) in [(1, MIN_Y + 2, 3), (4, MIN_Y + 40, 0), (0, MIN_Y + 41, 15)] {
+            Arc::make_mut(&mut chunk).set(x, y, z, 5);
+        }
+        let mut out = Vec::new();
+        index.column(&chunk, MIN_Y, (MIN_Y, MIN_Y + 63), kind, &mut out);
+        let bases: Vec<i32> = out.iter().map(|(base, _)| *base).collect();
+        assert_eq!(bases, [MIN_Y, MIN_Y + 32]);
+        for (base, pois) in &out {
+            let s = (base - MIN_Y) / 16;
+            let one = index.get((2, s, -3), &chunk.sections[s as usize], kind);
+            assert!(Arc::ptr_eq(pois, &one));
+        }
+        // Kept: no rescan.
+        index.column(
+            &chunk,
+            MIN_Y,
+            (MIN_Y, MIN_Y + 63),
+            |_| panic!("rescanned"),
+            &mut out,
+        );
+        assert_eq!(out.len(), 2);
+        // An edit lists that section again.
+        Arc::make_mut(&mut chunk).set(4, MIN_Y + 40, 0, 1);
+        let mut rescans = 0;
+        let counted = |v: u32| {
+            rescans += 1;
+            kind(v)
+        };
+        index.column(&chunk, MIN_Y, (MIN_Y + 20, MIN_Y + 63), counted, &mut out);
+        assert!(rescans > 0);
+        assert_eq!(out.len(), 1);
+        assert_eq!(out[0].0, MIN_Y + 32);
+        assert_eq!(&*out[0].1, &[(9 << 8 | 15 << 4, 5)]);
+    }
+
     #[test]
     fn unloaded_sections_are_dropped() {
         let index = PoiSections::<u8>::default();
@@ -153,7 +273,7 @@ mod tests {
             index.get((1, i, 0), &Arc::new(Section::default()), kind);
         }
         let lists = index.sections.lock().unwrap();
-        assert!(lists.by_section.len() <= 2);
-        assert!(lists.by_section.contains_key(&(0, 0, 0)));
+        assert!(lists.len <= 2);
+        assert!(lists.by_column[&(0, 0)][0].is_some());
     }
 }

crates/horizon-client/examples/probe/mob_ai_probe.rs

@@ -1208,6 +1208,7 @@ pub fn main() -> horizon_client::Result<()> {
                     dragon: None,
                     leash_boats: &[],
                     who_lookup: None,
+                    foreign: None,
                     others_grid: $grid,
                     others_index: None,
                     others_villagers: None,

crates/horizon-client/src/client_lib.rs

@@ -8,6 +8,7 @@ pub mod animate_tick;
 pub mod arrows;
 pub mod bed_geometry;
 pub mod block_contact;
+pub mod parallel;
 pub mod powder_snow;
 pub use horizon_client_render::{
     assets, atlas_sources, audio, auth, autotune, ban_notice, bidi, biome, block_edit,

crates/horizon-client/src/entities.rs

@@ -84,7 +84,9 @@ fn slime_at(world: &World, facts: &BlockFacts, position: BlockPos) -> bool {
 mod fluids;
 mod item_fall;
 mod item_no_physics;
+mod obstacles;
 pub use fluids::{fall_reset_hit, fluid_flow, update_entity_fluids, water_clip};
+pub use obstacles::ObstacleGrid;
 
 #[cfg(test)]
 mod drop_motion_tests {
@@ -1111,110 +1113,6 @@ fn lava_cauldron_contacts(
     }
     contacts
 }
-/// The boxes that stop a dropped item (boats, minecarts, shulkers, a held-still happy ghast),
-/// binned by 8-block cell, so an item looks only at the ones near it: a farm with many drops and
-/// many vehicles checks a handful of boxes per item, not every box in the level.
-#[derive(Default)]
-pub struct ObstacleGrid {
-    cells: crate::fast_hash::FxHashMap<(i32, i32, i32), Vec<u32>>,
-    boxes: Vec<Aabb>,
-    /// Each box's entity's `getRootVehicle` when it rides something; empty
-    /// when none was given.
-    roots: Vec<Option<u32>>,
-    found: Vec<u32>,
-}
-impl ObstacleGrid {
-    const CELL: f64 = 8.0;
-    pub fn new(boxes: Vec<Aabb>) -> Self {
-        Self::with_roots(boxes, Vec::new())
-    }
-    /// [`Self::new`] with the vehicle at the bottom of what each box's entity
-    /// rides (`roots[i]` for `boxes[i]`), so [`Self::near_except`] can leave a
-    /// mover's own riders out.
-    pub fn with_roots(boxes: Vec<Aabb>, roots: Vec<Option<u32>>) -> Self {
-        let mut cells: crate::fast_hash::FxHashMap<(i32, i32, i32), Vec<u32>> = Default::default();
-        for (i, b) in boxes.iter().enumerate() {
-            let (lo, hi) = Self::span(*b);
-            for x in lo.0..=hi.0 {
-                for y in lo.1..=hi.1 {
-                    for z in lo.2..=hi.2 {
-                        cells.entry((x, y, z)).or_default().push(i as u32);
-                    }
-                }
-            }
-        }
-        Self {
-            cells,
-            boxes,
-            roots,
-            found: Vec::new(),
-        }
-    }
-    fn span(b: Aabb) -> ((i32, i32, i32), (i32, i32, i32)) {
-        let cell = |v: Vec3| {
-            (
-                (v.x / Self::CELL).floor() as i32,
-                (v.y / Self::CELL).floor() as i32,
-                (v.z / Self::CELL).floor() as i32,
-            )
-        };
-        (cell(b.min_position()), cell(b.max_position()))
-    }
-    pub fn is_empty(&self) -> bool {
-        self.boxes.is_empty()
-    }
-    /// The boxes meeting `area` (`getEntityCollisions`'s query box), in the order they were given.
-    pub fn near(&mut self, area: Aabb, out: &mut Vec<Aabb>) {
-        self.near_except(area, None, out);
-    }
-    /// [`Self::near`] without the boxes of what rides `vehicle` (given with
-    /// [`Self::with_roots`]): `Entity.canCollideWith` leaves out
-    /// `isPassengerOfSameVehicle`, so a stand's riders do not stop it.
-    pub fn near_except(&mut self, area: Aabb, vehicle: Option<u32>, out: &mut Vec<Aabb>) {
-        out.clear();
-        if self.boxes.is_empty() {
-            return;
-        }
-        let kept = |i: usize| vehicle.is_none() || self.roots.get(i).copied().flatten() != vehicle;
-        let (lo, hi) = Self::span(area);
-        let side = |lo: i32, hi: i32| (i64::from(hi) - i64::from(lo) + 1).max(0);
-        let cells = side(lo.0, hi.0)
-            .saturating_mul(side(lo.1, hi.1))
-            .saturating_mul(side(lo.2, hi.2));
-        // An area over more cells than there are boxes (a stand loaded with a
-        // huge Motion asks for one millions of blocks long): test each box
-        // once rather than walk the cells.
-        if cells > self.boxes.len() as i64 {
-            out.extend(
-                self.boxes
-                    .iter()
-                    .enumerate()
-                    .filter(|&(i, b)| kept(i) && b.intersects(area))
-                    .map(|(_, b)| *b),
-            );
-            return;
-        }
-        self.found.clear();
-        for x in lo.0..=hi.0 {
-            for y in lo.1..=hi.1 {
-                for z in lo.2..=hi.2 {
-                    if let Some(list) = self.cells.get(&(x, y, z)) {
-                        self.found.extend(list);
-                    }
-                }
-            }
-        }
-        self.found.sort_unstable();
-        self.found.dedup();
-        for &i in &self.found {
-            let b = self.boxes[i as usize];
-            if kept(i as usize) && b.intersects(area) {
-                out.push(b);
-            }
-        }
-    }
-}
-
 /// Dropped items by block cell, for `ItemEntity.mergeWithNeighbours`'s
 /// search around an item. Holds indexes into the tick's item list.
 #[derive(Default)]

crates/horizon-client/src/entities/obstacles.rs

@@ -0,0 +1,122 @@
+//! The boxes that stop a dropped item, by cell (`ObstacleGrid`).
+use crate::fast_hash::FxHashMap;
+use horizon_core::geometry::{Aabb, Vec3};
+
+/// The boxes that stop a dropped item (boats, minecarts, shulkers, a held-still happy ghast),
+/// binned by 8-block cell, so an item looks only at the ones near it: a farm with many drops and
+/// many vehicles checks a handful of boxes per item, not every box in the level.
+#[derive(Default)]
+pub struct ObstacleGrid {
+    cells: FxHashMap<(i32, i32, i32), Vec<u32>>,
+    boxes: Vec<Aabb>,
+    /// Each box's entity's `getRootVehicle` when it rides something; empty
+    /// when none was given.
+    roots: Vec<Option<u32>>,
+    found: Vec<u32>,
+}
+impl ObstacleGrid {
+    const CELL: f64 = 8.0;
+    pub fn new(boxes: Vec<Aabb>) -> Self {
+        Self::with_roots(boxes, Vec::new())
+    }
+    /// [`Self::new`] with the vehicle at the bottom of what each box's entity
+    /// rides (`roots[i]` for `boxes[i]`), so [`Self::near_except`] can leave a
+    /// mover's own riders out.
+    pub fn with_roots(boxes: Vec<Aabb>, roots: Vec<Option<u32>>) -> Self {
+        let mut cells: FxHashMap<(i32, i32, i32), Vec<u32>> = Default::default();
+        for (i, b) in boxes.iter().enumerate() {
+            let (lo, hi) = Self::span(*b);
+            for x in lo.0..=hi.0 {
+                for y in lo.1..=hi.1 {
+                    for z in lo.2..=hi.2 {
+                        cells.entry((x, y, z)).or_default().push(i as u32);
+                    }
+                }
+            }
+        }
+        Self {
+            cells,
+            boxes,
+            roots,
+            found: Vec::new(),
+        }
+    }
+    fn span(b: Aabb) -> ((i32, i32, i32), (i32, i32, i32)) {
+        let cell = |v: Vec3| {
+            (
+                (v.x / Self::CELL).floor() as i32,
+                (v.y / Self::CELL).floor() as i32,
+                (v.z / Self::CELL).floor() as i32,
+            )
+        };
+        (cell(b.min_position()), cell(b.max_position()))
+    }
+    pub fn is_empty(&self) -> bool {
+        self.boxes.is_empty()
+    }
+    /// The boxes meeting `area` (`getEntityCollisions`'s query box), in the order they were given.
+    pub fn near(&mut self, area: Aabb, out: &mut Vec<Aabb>) {
+        self.near_except(area, None, out);
+    }
+    /// [`Self::near`] without the boxes of what rides `vehicle` (given with
+    /// [`Self::with_roots`]): `Entity.canCollideWith` leaves out
+    /// `isPassengerOfSameVehicle`, so a stand's riders do not stop it.
+    pub fn near_except(&mut self, area: Aabb, vehicle: Option<u32>, out: &mut Vec<Aabb>) {
+        let mut found = std::mem::take(&mut self.found);
+        self.near_with(area, vehicle, &mut found, out);
+        self.found = found;
+    }
+    /// [`Self::near`] with the caller's scratch, so several threads can ask one grid.
+    pub fn near_shared(&self, area: Aabb, found: &mut Vec<u32>, out: &mut Vec<Aabb>) {
+        self.near_with(area, None, found, out);
+    }
+    fn near_with(
+        &self,
+        area: Aabb,
+        vehicle: Option<u32>,
+        found: &mut Vec<u32>,
+        out: &mut Vec<Aabb>,
+    ) {
+        out.clear();
+        if self.boxes.is_empty() {
+            return;
+        }
+        let kept = |i: usize| vehicle.is_none() || self.roots.get(i).copied().flatten() != vehicle;
+        let (lo, hi) = Self::span(area);
+        let side = |lo: i32, hi: i32| (i64::from(hi) - i64::from(lo) + 1).max(0);
+        let cells = side(lo.0, hi.0)
+            .saturating_mul(side(lo.1, hi.1))
+            .saturating_mul(side(lo.2, hi.2));
+        // An area over more cells than there are boxes (a stand loaded with a
+        // huge Motion asks for one millions of blocks long): test each box
+        // once rather than walk the cells.
+        if cells > self.boxes.len() as i64 {
+            out.extend(
+                self.boxes
+                    .iter()
+                    .enumerate()
+                    .filter(|&(i, b)| kept(i) && b.intersects(area))
+                    .map(|(_, b)| *b),
+            );
+            return;
+        }
+        found.clear();
+        for x in lo.0..=hi.0 {
+            for y in lo.1..=hi.1 {
+                for z in lo.2..=hi.2 {
+                    if let Some(list) = self.cells.get(&(x, y, z)) {
+                        found.extend(list);
+                    }
+                }
+            }
+        }
+        found.sort_unstable();
+        found.dedup();
+        for &i in found.iter() {
+            let b = self.boxes[i as usize];
+            if kept(i as usize) && b.intersects(area) {
+                out.push(b);
+            }
+        }
+    }
+}

crates/horizon-client/src/game.rs

@@ -1499,7 +1499,7 @@ impl LevelAccess for GameLevel<'_> {
             };
         }
         if let Some(signal) = books::lectern_signal(self.world, pos) {
-            return signal;
+            return signal * i32::from(books::has_book_state(self.registry, state));
         }
         if let Some(signal) =
             machines::shelf_signal(self.world, self.registry, state, pos, direction)

crates/horizon-client/src/game/block_ticks.rs

@@ -239,6 +239,19 @@ impl Game<'_> {
             self.tick_scheduled()?;
         }
         self.with_level(|b, level, ticks| {
+            // The masks the random ticks will read, built on helper threads first while the
+            // parallel tick is on (`HORIZON_PARALLEL_TICK`): the draws and ticks don't depend on them.
+            if !debug && horizon_client::parallel::lanes() > 1 {
+                horizon_client::spans::time("random_tick_masks", || {
+                    let chunks = &level.world.chunks;
+                    level.random_tick_sections.prepare_masks(
+                        &scratch.ticking,
+                        |key| chunks.get(&key).cloned(),
+                        level.facts,
+                        &|id| b.random_tick_candidate(BlockStateId(id)),
+                    );
+                });
+            }
             horizon_client::spans::time("random_ticks", || {
                 for (x, z) in scratch.ticking.drain(..).filter(|_| !debug) {
                     // The last answer, asked from Y `from` at a world revision:

crates/horizon-client/src/game/boats.rs

@@ -682,19 +682,14 @@ impl Game<'_> {
     /// of it (`ClipContext.Block.FALLDAMAGE_RESETTING`, `ClipContext.Fluid.WATER`)
     /// and a hit resets the fall, before the move's own drop counts.
     pub(super) fn vehicle_fall_reset(&self, fall: f64, from: Vec3, moved: Vec3) -> f64 {
-        let length = moved.length_squared();
-        if fall == 0.0 || length < 1.0 {
-            return fall;
-        }
-        let reset = &self.res.motion().fall_reset;
-        let hit = horizon_client::entities::fall_reset_hit(
+        fall_reset(
             &self.world,
-            &self.res.motion().fluids,
-            |p| reset.contains(&*self.res.facts().get(self.world.block(p.x, p.y, p.z)).name),
+            self.res.motion(),
+            self.res.facts(),
+            fall,
             from,
             moved,
-        );
-        if hit { 0.0 } else { fall }
+        )
     }
     /// `AbstractBoat.checkFallDamage` after a move that went `moved_y` (with
     /// `b` already where it landed): on the ground the fall resets; a move
@@ -2071,3 +2066,26 @@ mod tests {
     }
 }
 horizon_client::impl_joined!(Boat);
+
+/// [`Game::vehicle_fall_reset`] from the tables it reads.
+pub(super) fn fall_reset(
+    world: &horizon_client::world::World,
+    motion: &horizon_client::motion::MotionResources,
+    facts: &horizon_client::blocks::BlockFacts,
+    fall: f64,
+    from: Vec3,
+    moved: Vec3,
+) -> f64 {
+    if fall == 0.0 || moved.length_squared() < 1.0 {
+        return fall;
+    }
+    let reset = &motion.fall_reset;
+    let hit = horizon_client::entities::fall_reset_hit(
+        world,
+        &motion.fluids,
+        |p| reset.contains(&*facts.get(world.block(p.x, p.y, p.z)).name),
+        from,
+        moved,
+    );
+    if hit { 0.0 } else { fall }
+}

crates/horizon-client/src/game/books.rs

@@ -1608,6 +1608,16 @@ impl Game<'_> {
         }
     }
 }
+/// `LecternBlock.getAnalogOutputSignal`'s `HAS_BOOK` check.
+pub(super) fn has_book_state(
+    registry: &'static horizon_world::registry::BlockRegistry,
+    state: horizon_world::registry::BlockStateId,
+) -> bool {
+    registry
+        .state(state)
+        .is_some_and(|s| s.properties.get("has_book").is_some_and(|v| v == "true"))
+}
+
 /// `LecternBlockEntity.getRedstoneSignal`.
 pub(super) fn lectern_signal(world: &horizon_client::world::World, pos: BlockPos) -> Option<i32> {
     match world.block_entities.get(&(pos.x, pos.y, pos.z)) {
@@ -1721,6 +1731,7 @@ mod lectern_signal_tests {
         items::ItemStack,
     };
     use horizon_core::position::BlockPos;
+    use horizon_world::level::LevelAccess;
     use horizon_world::registry::BlockStateId;
 
     fn signal(game: &mut super::Game<'_>, book: Option<ItemStack>, page: i32) -> i32 {
@@ -1738,6 +1749,55 @@ mod lectern_signal_tests {
         lectern_signal(&game.world, BlockPos::new(key.0, key.1, key.2)).unwrap()
     }
 
+    /// `LecternBlock.getAnalogOutputSignal` reads the block entity only while
+    /// the state's `HAS_BOOK` is true.
+    #[test]
+    #[ignore = "needs prepared resources in HORIZON_TEST_RESOURCES"]
+    fn a_lectern_without_the_has_book_state_feeds_no_comparator() {
+        let mut game = server_game();
+        let key = (8, 64, 8);
+        let state = game
+            .pack
+            .assets
+            .registry
+            .default_state("minecraft:lectern")
+            .unwrap();
+        game.world.set(key.0, key.1, key.2, state).unwrap();
+        game.world.block_entities.insert(
+            key,
+            BlockEntity::Lectern(Lectern {
+                book: Some(ItemStack::new("minecraft:writable_book", 1)),
+                page: 0,
+            }),
+        );
+        let signal = |game: &mut super::Game<'_>, state| {
+            game.with_level(|_, level, _| {
+                level.block_entity_analog_signal(
+                    BlockPos::new(key.0, key.1, key.2),
+                    state,
+                    horizon_core::geometry::Direction::North,
+                )
+            })
+        };
+        assert_eq!(
+            signal(&mut game, state),
+            0,
+            "the block state says the lectern holds no book"
+        );
+        let registry = game.pack.assets.registry;
+        let mut with_book = registry.state(state).unwrap().clone();
+        with_book
+            .properties
+            .insert("has_book".into(), "true".into());
+        let with_book = registry.id(&with_book).unwrap();
+        game.world.set(key.0, key.1, key.2, with_book).unwrap();
+        assert_eq!(
+            signal(&mut game, with_book),
+            15,
+            "with HAS_BOOK the book's page signal"
+        );
+    }
+
     /// `LecternBlockEntity.getRedstoneSignal`: `floor(page / (pageCount - 1)
     /// * 14) + (hasBook() ? 1 : 0)`. `hasBook()` asks for a book-content
     /// component, so a stack that only sits in the lectern (a non-book item a

crates/horizon-client/src/game/command_harness.rs

@@ -380,7 +380,7 @@ impl Game<'_> {
                         let (mut meta, mut defaults) = (Vec::new(), Vec::new());
                         super::lan::mob_meta(
                             m,
-                            hub,
+                            hub.sync_data(),
                             None,
                             g.world.game_time,
                             &mut meta,

crates/horizon-client/src/game/creatures.rs

@@ -3,6 +3,8 @@
 //! breeding, conversions, drops), projectiles, player interactions
 //! (feeding, shearing, milking, spawn eggs, throwing) and rendering.
 mod mob_geometry;
+mod mob_group;
+mod prep;
 mod projectile_world;
 mod pufferfish_player;
 mod spawn_eggs;
@@ -1048,18 +1050,11 @@ impl Game<'_> {
             teams: std::sync::Arc::new(self.mob_teams()),
             items: Some(self.items()),
             lid_openers: &self.lid_openers,
-            mod_effects: if self.remote.is_none() {
-                Some(self)
-            } else {
-                None
-            },
+            mod_effects: self.remote.is_none().then_some(self as _),
             // A warden's away suspects are UUIDs: on the server side, where the
             // entity UUIDs live, they resolve back to the mobs that are here.
-            who_lookup: if self.remote.is_none() {
-                Some(self)
-            } else {
-                None
-            },
+            who_lookup: self.remote.is_none().then_some(self as _),
+            foreign: None,
             border: Some(self.border.geometry()),
             // The first player in the level: a server has no player of its own.
             difficulty: self.local_difficulty(
@@ -1187,50 +1182,32 @@ impl Game<'_> {
         self.init_sulfur_attribute_maps();
         let uuid = self.level.player_uuid;
         let level_id = self.dim.id();
-        // LivingEntity.isPushable reads onClimbable live; a mob's cached one
-        // is only refreshed by its own push step, so one that has not ticked
-        // yet (the first tick after a load, a fresh spawn) or has moved since
-        // would be pushed out of a ladder or scaffolding. Refreshed here, from
-        // where each mob stands now, before the snapshot others push it by.
-        let climbing: Vec<bool> = {
-            let w = self.mob_world_base(&[]);
-            self.mobs.iter().map(|m| m.on_climbable(&w)).collect()
-        };
-        for (m, climbing) in self.mobs.iter_mut().zip(climbing) {
-            m.on_climbable = climbing;
-        }
+        // Each mob's climbing, refreshed, and its snapshot entry (creatures/prep.rs).
         let mut snapshot = std::mem::take(&mut self.creatures.seen_buf);
-        snapshot.extend(self.mobs.iter().map(|m| seen(m, uuid)));
+        self.prepare_mobs(uuid, &mut snapshot);
         let mut events = std::mem::take(&mut self.creatures.event_buf);
         events.clear();
         // SulfurCube.primeWhenOnPoweredPosition runs before super.tick adopts
         // the body: the cubes whose adopted body explodes, as this tick starts.
         let primable = self.sulfur_cubes_adopted_explosive();
-        let mut mobs = std::mem::take(&mut self.mobs);
+        let mobs = std::mem::take(&mut self.mobs);
         // Each mob's target as the mods last heard it, by its place in `mobs`
         // (`native_living_change_target`).
         let mut targets = std::mem::take(&mut self.creatures.target_buf);
         targets.clear();
         targets.extend(mobs.iter().map(|m| m.ai.target));
-        let mut dead = Vec::new();
-        let mut corpses = Vec::new();
-        let mut died = std::mem::take(&mut self.died);
+        let died = std::mem::take(&mut self.died);
         let mut ticked = std::mem::take(&mut self.creatures.ticked);
         ticked.clear();
         let ticking = self.entity_ticking_area();
-        // With guests on, each mob sees its nearest player (game/lan_mobs.rs).
-        let mut guests = self.lan_guest_views();
-        // What stands in a guest's slot while a mob holds that guest's view for
-        // its tick (see `lent` below).
-        let mut spare = guests.first().map(|g| g.1.clone());
         let players = self.level_players();
         let nearest = super::natural_spawning::NearestPlayers::new(&players);
-        let mut bindings = match self.lan.as_mut() {
+        // With guests on, each mob sees its nearest player (game/lan_mobs.rs).
+        let guests = self.lan_guest_views();
+        let bindings = match self.lan.as_mut() {
             Some(host) if !guests.is_empty() => std::mem::take(&mut host.mob_players),
             _ => Default::default(),
         };
-        // The guests' entity ids by connection, for the shulker bullets fired
-        // at the guest a mob sees.
         let guest_ids: Vec<(u32, u32)> = match self.lan.as_ref() {
             Some(host) if !guests.is_empty() => host
                 .hub
@@ -1240,23 +1217,9 @@ impl Game<'_> {
                 .collect(),
             _ => Vec::new(),
         };
-        let mut guest_events = Vec::new();
         // The block writes since the last tick: mobs re-path around them.
         let path_writes = std::mem::take(self.world.path_writes.get_or_insert_with(Vec::new));
-        // The mobs whose bodies go this tick: (id, feet, width, height, split).
-        // 🤖 Agent: the last field is the grown slime, magma cube or sulfur
-        // cube the body leaves behind (`AbstractCubeMob.remove`), captured as
-        // the body goes because it has left the list by the time it splits.
-        let mut gone = Vec::new();
-        let level_random;
-        {
-            // The level random, lent to the mobs (brain behaviours draw from
-            // it) and taken back after their ticks.
-            let mut scratch = self.world.random.clone();
-            let mut pushes = horizon_client::mobs::PendingPushes::default();
-            let mut keep = Vec::with_capacity(mobs.len());
-            let mut grid =
-                horizon_client::mobs::seen_grid::SeenGrid::new(snapshot.iter().map(|s| s.bounds));
+        let group = {
             // All but the snapshot is the same for every mob of the tick, so
             // build it once: the player view and the teams scan every UUID.
             self.leash_boats = self
@@ -1266,87 +1229,52 @@ impl Game<'_> {
                 .collect();
             let mut base = self.mob_world_base(&[]);
             let path_changes = horizon_client::mobs::collision_changes(&base, &path_writes);
-            // Each mob keeps its snapshot slot all tick; the first slot of an
-            // id answers, as a scan would.
-            let mut index = horizon_client::fast_hash::FxHashMap::with_capacity_and_hasher(
-                snapshot.len(),
-                Default::default(),
-            );
-            let mut villager_slots = Vec::new();
-            let mut solid_slots = Vec::new();
-            for (i, s) in snapshot.iter().enumerate() {
-                index.entry(s.id).or_insert(i as u32);
-                if s.species == horizon_client::mobs::Species::Villager {
-                    villager_slots.push(i as u32);
-                }
-                if s.may_block_mobs() {
-                    solid_slots.push(i as u32);
-                }
-            }
             laps.lap("creature_prepare");
             let single_player = horizon_client::mobs::single_player_pos(
                 players.iter().map(|p| (p.position, p.spectator)),
             );
+            let nearest = prep::MobsNearest::new(&nearest, &mobs, single_player.is_none());
             let mut despawn_w = base.despawn_view();
             let mut cached_difficulty: Option<((i32, i32), spawning::LocalDifficulty)> = None;
-            for i in 0..mobs.len() {
-                let m = &mut mobs[i];
-                let (bx, bz) = (m.position.x.floor() as i32, m.position.z.floor() as i32);
-                let seen_guest = if guests.is_empty() {
-                    None
-                } else {
-                    let bound = bindings.get(&m.id).copied();
-                    let conn = super::lan_mobs::player_for(m, base.player.as_ref(), &guests, bound);
-                    match conn {
-                        Some(conn) => {
-                            bindings.insert(m.id, conn);
-                        }
-                        None => {
-                            bindings.remove(&m.id);
-                        }
-                    }
-                    conn
-                };
+            let shared = mob_group::Shared {
+                uuid,
+                level_id,
+                path_changes: &path_changes,
+                guest_ids: &guest_ids,
+            };
+            let start = mob_group::Start {
+                events,
+                // The level random, lent to the mobs and taken back after their ticks.
+                random: self.world.random.clone(),
+                guests,
+                player: base.player.take(),
+                bindings,
+                died,
+                ticked,
+                targets,
+            };
+            let mut group = mob_group::Group::new(mobs.into(), snapshot, start);
+            for i in 0..group.mobs.len() {
+                let position = group.mobs[i].position;
+                let (bx, bz) = (position.x.floor() as i32, position.z.floor() as i32);
+                let seen_guest = group.bind_guest(i);
                 if !base.world.loaded(bx, bz) {
-                    keep.push(!m.hostile() || m.persistent);
+                    group.unloaded(i);
                     continue;
                 }
-                let nearest_player = single_player.unwrap_or_else(|| {
-                    nearest
-                        .nearest(m.position.x, m.position.y, m.position.z)
-                        .map(|p| p.position)
-                });
+                let nearest_player = single_player.unwrap_or_else(|| nearest.at(i, position));
                 // ServerLevel.tick: every loaded entity checks despawning, but
                 // only those in entity-ticking chunks (within the simulation
-                // distance) tick. The despawn check reads the nearest player
-                // and the difficulty, not the player the mob sees.
+                // distance) tick.
                 if !ticking(bx, bz) {
                     despawn_w.nearest_player = nearest_player;
-                    let kept = horizon_client::mobs::ai::despawn_check(m, &despawn_w);
-                    if !kept {
-                        died.remove(&m.id);
-                        snapshot[i].alive = false;
-                    }
-                    keep.push(kept);
+                    group.rest(i, &despawn_w);
                     continue;
                 }
-                let controlling_player_movement = horizon_client::mobs::equine::controlled(m)
-                    .then(|| {
-                        base.player
-                            .as_ref()
-                            .filter(|p| p.riding_vehicle == Some(m.id))
-                            .map(|p| p.velocity)
-                            .or_else(|| {
-                                guests
-                                    .iter()
-                                    .find(|(_, p)| p.riding_vehicle == Some(m.id))
-                                    .map(|(_, p)| p.velocity)
-                            })
-                    })
-                    .flatten();
+                let controlling = group.controlling(i);
                 // A mod canceled its tick: it stays as it is.
-                if round.canceled(m.id) {
-                    keep.push(true);
+                if round.canceled(group.mobs[i].id) {
+                    group.keep.push(true);
                     continue;
                 }
                 // getCurrentDifficultyAt(blockPosition()): the mob's own chunk,
@@ -1356,175 +1284,34 @@ impl Game<'_> {
                     Some((key, difficulty)) if key == chunk_key => difficulty,
                     _ => {
                         let difficulty =
-                            self.local_difficulty([m.position.x, m.position.y, m.position.z]);
+                            self.local_difficulty([position.x, position.y, position.z]);
                         cached_difficulty = Some((chunk_key, difficulty));
                         difficulty
                     }
                 };
-                horizon_client::mobs::apply_mob_pushes(m, &mut pushes);
-                horizon_client::mobs::blocks_changed(m, &path_changes);
-                let from = events.len();
-                // The guest's view is lent to the mob for its tick, not copied:
-                // the tick only reads it (`&MobWorld`), and it goes back to the
-                // guest's slot straight after.
-                let lent = seen_guest.and_then(|conn| guests.iter().position(|g| g.0 == conn));
-                let active_player = match (seen_guest, lent) {
-                    (None, _) => base.player.take(),
-                    (Some(_), Some(k)) => spare
-                        .take()
-                        .map(|stand_in| std::mem::replace(&mut guests[k].1, stand_in)),
-                    (Some(_), None) => None,
-                };
-                let mut w = MobWorld {
-                    others: &snapshot,
-                    others_grid: Some(&grid),
-                    others_index: Some(&index),
-                    others_villagers: Some(&villager_slots),
-                    others_solid: Some(&solid_slots),
-                    difficulty,
-                    nearest_player,
-                    player: active_player,
-                    teams: std::sync::Arc::clone(&base.teams),
-                    ..base
-                };
-                w.controlling_player_movement = controlling_player_movement;
-                ticked.insert(mobs[i].id);
-                // `horizon-report`'s cost by mob type (tick_profile.rs).
-                let sampled = super::tick_profile::start().map(|t| (t, mobs[i].kind.clone()));
-                let kept =
-                    horizon_client::mobs::riders::tick(&mut mobs, i, &w, &mut scratch, &mut events);
-                if let Some((started, kind)) = sampled {
-                    super::tick_profile::add(level_id, &kind, started);
-                }
-                let m = &mut mobs[i];
-                if seen_guest.is_none() {
-                    base.player = w.player;
-                } else if let Some(k) = lent {
-                    spare = w.player.map(|v| std::mem::replace(&mut guests[k].1, v));
-                }
-                // Its tick changed its target: the mods may keep the one it had
-                // (`native_living_change_target`), and then what the tick did to
-                // the new one is not dealt.
-                if let Some(&before) = targets.get(i)
-                    && m.ai.target != before
-                {
-                    let named = target_named(
-                        m.ai.target,
-                        seen_guest,
-                        &guests,
-                        &guest_ids,
-                        base.player.as_ref(),
-                    );
-                    if let Some((player, mob)) = named {
-                        let kind = super::native_hooks::mob_kind(m);
-                        let mut cancel = false;
-                        Self::native_living_change_target(
-                            level_id,
-                            m.id,
-                            &kind,
-                            player,
-                            mob,
-                            &mut cancel,
-                        );
-                        if cancel {
-                            if let Some(new) = m.ai.target {
-                                let mut k = from;
-                                while k < events.len() {
-                                    if hits_target(&events[k], new) {
-                                        events.remove(k);
-                                    } else {
-                                        k += 1;
-                                    }
-                                }
-                            }
-                            m.ai.target = before;
-                        }
-                    }
-                    targets[i] = m.ai.target;
-                }
-                // A killed mob's body is removed: its effects that act on
-                // death run (`LivingEntity.remove` with KILLED).
-                if !kept && !m.alive() && !m.removed && !m.extra.effects.is_empty() {
-                    corpses.push(m.clone());
-                }
-                horizon_client::mobs::take_mob_pushes(&mut events, from, &mut pushes);
-                // LivingEntity.tickDeath: after twenty ticks the body is removed.
-                if !kept && !m.removed && !m.alive() && m.death_time >= 20 {
-                    let (width, height) = (f64::from(m.width()), f64::from(m.height()));
-                    // 🤖 Agent: `AbstractCubeMob.remove` splits a grown cube into
-                    // its smaller halves when it is removed, twenty ticks after
-                    // the death. The parent leaves the list before the split
-                    // runs, so its size, position, kind and persistence are
-                    // captured here, as they are at the removal.
-                    let split = if matches!(
-                        m.species(),
-                        Species::Slime | Species::MagmaCube | Species::SulfurCube
-                    ) && m.extra.size > 1
-                    {
-                        Some(m.clone())
-                    } else {
-                        None
-                    };
-                    gone.push((m.id, m.position, width, height, split));
-                }
-                if let Some(conn) = seen_guest {
-                    // ShulkerBullet.finalTarget is the guest's entity: the
-                    // bullet homes on it while the projectiles tick, where
-                    // `Who::Player` is the host's own player.
-                    if let Some(&(_, id)) = guest_ids.iter().find(|g| g.0 == conn) {
-                        for e in &mut events[from..] {
-                            if let MobEvent::Projectile(p) = e
-                                && let ProjectileKind::ShulkerBullet { target, .. } = &mut p.kind
-                                && *target == Some(Who::Player)
-                            {
-                                *target = Some(Who::Mob(id));
-                            }
-                        }
-                    }
-                    let mut k = from;
-                    while k < events.len() {
-                        if super::lan_mobs::acts_on_player(&events[k]) {
-                            guest_events.push((conn, m.id, events.remove(k)));
-                        } else {
-                            k += 1;
-                        }
-                    }
-                }
-                // Killed during its own tick (drowning, burning, falling...).
-                if !m.alive() && !m.removed && died.insert(m.id) {
-                    dead.push(m.clone());
-                }
-                if !kept {
-                    died.remove(&m.id);
-                }
-                keep.push(kept);
-                // ServerLevel ticks entities one after another: the ones
-                // after this mob see it as it is now.
-                snapshot[i] = seen(m, uuid);
-                snapshot[i].alive &= kept;
-                grid.update(i, snapshot[i].bounds);
-                horizon_client::mobs::apply_brain_events(&mut mobs, &mut events, from);
+                let player = (seen_guest, nearest_player, controlling);
+                group.tick(i, player, difficulty, &base, &shared);
             }
             // Targets changed after the mob's own tick (another mob's alert):
             // heard now, and a cancel puts the old one back before its next tick.
-            for (i, m) in mobs.iter_mut().enumerate() {
-                let Some(&before) = targets.get(i) else {
+            for (i, m) in group.mobs.iter_mut().enumerate() {
+                let Some(&before) = group.targets.get(i) else {
                     break;
                 };
                 if m.ai.target == before {
                     continue;
                 }
-                let seen_guest = if guests.is_empty() {
+                let seen_guest = if group.guests.is_empty() {
                     None
                 } else {
-                    bindings.get(&m.id).copied()
+                    group.bindings.get(&m.id).copied()
                 };
                 let named = target_named(
                     m.ai.target,
                     seen_guest,
-                    &guests,
+                    &group.guests,
                     &guest_ids,
-                    base.player.as_ref(),
+                    group.player.as_ref(),
                 );
                 if let Some((player, mob)) = named {
                     let kind = super::native_hooks::mob_kind(m);
@@ -1543,13 +1330,29 @@ impl Game<'_> {
                 }
             }
             laps.lap("creature_loop");
-            let mut flags = keep.into_iter();
-            mobs.retain(|_| flags.next().unwrap_or(true));
-            for m in &mut mobs {
-                horizon_client::mobs::apply_mob_pushes(m, &mut pushes);
+            let mut flags = std::mem::take(&mut group.keep).into_iter();
+            group.mobs.retain(|_| flags.next().unwrap_or(true));
+            for m in &mut group.mobs {
+                horizon_client::mobs::apply_mob_pushes(m, &mut group.pushes);
             }
-            level_random = scratch;
-        }
+            group
+        };
+        let mob_group::Group {
+            mut mobs,
+            mut snapshot,
+            mut events,
+            random: level_random,
+            guests,
+            mut bindings,
+            died,
+            ticked,
+            mut targets,
+            dead,
+            corpses,
+            gone,
+            guest_events,
+            ..
+        } = group;
         self.world.random = level_random;
         if let Some(host) = self.lan.as_mut()
             && !guests.is_empty()
@@ -1565,7 +1368,7 @@ impl Game<'_> {
         // first, as `append` put them; with none, the ticked list goes back
         // as it is instead of being copied into a new one, every mob a tick.
         if self.mobs.is_empty() {
-            self.mobs = mobs;
+            self.mobs = mobs.into();
         } else {
             self.mobs.append(&mut mobs);
         }
@@ -8376,6 +8179,7 @@ mod pearl_tests {
             dragon: None,
             leash_boats: &[],
             who_lookup: None,
+            foreign: None,
             others_grid: None,
             others_index: None,
             others_villagers: None,

crates/horizon-client/src/game/creatures/mob_group.rs

@@ -0,0 +1,338 @@
+//! The mob loop of `tick_creature_entities` over a group of mobs, in their list order, with
+//! everything the loop keeps for them: their snapshot and its indexes, the pushes and events
+//! their ticks give, the level random they draw from, the players they see and what dies. The
+//! serial loop is one group holding every mob; a group of the parallel mob tick (#14457) holds
+//! the mobs no mob outside it can reach this tick.
+use super::*;
+use horizon_client::{
+    fast_hash::{FxHashMap, FxHashSet},
+    mobs::{MobSlot, PendingPushes, seen_grid::SeenGrid, spawning::LocalDifficulty},
+};
+use horizon_core::random::LegacyRandom;
+
+/// What every mob of the loop reads and none writes.
+pub(super) struct Shared<'a> {
+    pub uuid: Option<[i32; 4]>,
+    pub level_id: &'static str,
+    /// The blocks whose collision changed since the last tick: mobs re-path around them.
+    pub path_changes: &'a [[i32; 3]],
+    /// The guests' entity ids by connection, for the shulker bullets fired at the guest a mob
+    /// sees.
+    pub guest_ids: &'a [(u32, u32)],
+}
+
+/// A mob whose body goes this tick: (id, feet, width, height, the grown cube it splits into).
+/// The last field is the grown slime, magma cube or sulfur cube the body leaves behind
+/// (`AbstractCubeMob.remove`), captured as the body goes because it has left the list by the
+/// time it splits.
+pub(super) type Gone = (u32, Vec3, f64, f64, Option<Mob>);
+
+/// A group's mobs and everything the loop keeps for them, in their list order.
+pub(super) struct Group<M: MobSlot> {
+    pub mobs: Vec<M>,
+    /// Each mob's snapshot entry: as it was before the tick until its turn, as its turn left it
+    /// after.
+    pub snapshot: Vec<Seen>,
+    grid: SeenGrid,
+    /// The first slot of each id, as a scan would find it.
+    index: FxHashMap<u32, u32>,
+    villager_slots: Vec<u32>,
+    solid_slots: Vec<u32>,
+    pub pushes: PendingPushes,
+    pub events: Vec<MobEvent>,
+    /// The level random, lent to the mobs (brain behaviours draw from it).
+    pub random: LegacyRandom,
+    pub guests: super::super::lan_mobs::GuestViews,
+    /// What stands in a guest's slot while a mob holds that guest's view for its tick.
+    spare: Option<PlayerView>,
+    /// The host's own player, lent to each mob that sees it for its tick.
+    pub player: Option<PlayerView>,
+    /// The guest each mob sees, by mob id.
+    pub bindings: FxHashMap<u32, u32>,
+    pub died: FxHashSet<u32>,
+    pub ticked: FxHashSet<u32>,
+    /// Each mob's target as the mods last heard it (`native_living_change_target`).
+    pub targets: Vec<Option<Who>>,
+    /// Whether each mob stays in the list, pushed at its turn.
+    pub keep: Vec<bool>,
+    pub dead: Vec<Mob>,
+    pub corpses: Vec<Mob>,
+    pub gone: Vec<Gone>,
+    pub guest_events: Vec<(u32, u32, MobEvent)>,
+}
+
+/// What a group starts from besides its mobs and their snapshot.
+pub(super) struct Start {
+    pub events: Vec<MobEvent>,
+    pub random: LegacyRandom,
+    pub guests: super::super::lan_mobs::GuestViews,
+    pub player: Option<PlayerView>,
+    pub bindings: FxHashMap<u32, u32>,
+    pub died: FxHashSet<u32>,
+    pub ticked: FxHashSet<u32>,
+    pub targets: Vec<Option<Who>>,
+}
+
+impl<M: MobSlot> Group<M> {
+    /// A group of `mobs`, whose snapshot entries are `snapshot`.
+    pub(super) fn new(mobs: Vec<M>, snapshot: Vec<Seen>, start: Start) -> Self {
+        let grid = SeenGrid::new(snapshot.iter().map(|s| s.bounds));
+        let mut index = FxHashMap::with_capacity_and_hasher(snapshot.len(), Default::default());
+        let (mut villager_slots, mut solid_slots) = (Vec::new(), Vec::new());
+        for (i, s) in snapshot.iter().enumerate() {
+            index.entry(s.id).or_insert(i as u32);
+            if s.species == Species::Villager {
+                villager_slots.push(i as u32);
+            }
+            if s.may_block_mobs() {
+                solid_slots.push(i as u32);
+            }
+        }
+        let n = mobs.len();
+        Self {
+            spare: start.guests.first().map(|g| g.1.clone()),
+            mobs,
+            snapshot,
+            grid,
+            index,
+            villager_slots,
+            solid_slots,
+            pushes: PendingPushes::default(),
+            events: start.events,
+            random: start.random,
+            guests: start.guests,
+            player: start.player,
+            bindings: start.bindings,
+            died: start.died,
+            ticked: start.ticked,
+            targets: start.targets,
+            keep: Vec::with_capacity(n),
+            dead: Vec::new(),
+            corpses: Vec::new(),
+            gone: Vec::new(),
+            guest_events: Vec::new(),
+        }
+    }
+
+    /// The guest mob `i` sees now, kept in the bindings: none while no guest is on.
+    pub(super) fn bind_guest(&mut self, i: usize) -> Option<u32> {
+        if self.guests.is_empty() {
+            return None;
+        }
+        let m = self.mobs[i].mob();
+        let bound = self.bindings.get(&m.id).copied();
+        let conn = super::super::lan_mobs::player_for(m, self.player.as_ref(), &self.guests, bound);
+        match conn {
+            Some(conn) => self.bindings.insert(m.id, conn),
+            None => self.bindings.remove(&m.id),
+        };
+        conn
+    }
+
+    /// Mob `i`'s turn when its chunk isn't loaded: it is kept as it is, unless it would despawn.
+    pub(super) fn unloaded(&mut self, i: usize) {
+        let m = self.mobs[i].mob();
+        self.keep.push(!m.hostile() || m.persistent);
+    }
+
+    /// Mob `i`'s turn outside the entity-ticking area: only its despawn check, which reads the
+    /// nearest player and the difficulty, not the player the mob sees.
+    pub(super) fn rest(&mut self, i: usize, despawn_w: &MobWorld) {
+        let m = self.mobs[i].mob_mut();
+        let kept = horizon_client::mobs::ai::despawn_check(m, despawn_w);
+        if !kept {
+            self.died.remove(&m.id);
+            self.snapshot[i].alive = false;
+        }
+        self.keep.push(kept);
+    }
+
+    /// Who controls mob `i`'s moves this tick: the velocity of the player riding it, when it is
+    /// a controlled mount.
+    pub(super) fn controlling(&self, i: usize) -> Option<Vec3> {
+        let m = self.mobs[i].mob();
+        horizon_client::mobs::equine::controlled(m)
+            .then(|| {
+                self.player
+                    .as_ref()
+                    .filter(|p| p.riding_vehicle == Some(m.id))
+                    .map(|p| p.velocity)
+                    .or_else(|| {
+                        self.guests
+                            .iter()
+                            .find(|(_, p)| p.riding_vehicle == Some(m.id))
+                            .map(|(_, p)| p.velocity)
+                    })
+            })
+            .flatten()
+    }
+
+    /// Mob `i`'s own tick, then what the loop does with what it did. `players`: the guest it
+    /// sees (none: the host's player), its nearest player, and the velocity of the player
+    /// controlling it as a mount ([`Self::controlling`]).
+    pub(super) fn tick(
+        &mut self,
+        i: usize,
+        (seen_guest, nearest_player, controlling_player_movement): (
+            Option<u32>,
+            Option<Vec3>,
+            Option<Vec3>,
+        ),
+        difficulty: LocalDifficulty,
+        base: &MobWorld,
+        shared: &Shared,
+    ) {
+        let m = self.mobs[i].mob_mut();
+        horizon_client::mobs::apply_mob_pushes(m, &mut self.pushes);
+        horizon_client::mobs::blocks_changed(m, shared.path_changes);
+        let from = self.events.len();
+        // The guest's view is lent to the mob for its tick, not copied:
+        // the tick only reads it (`&MobWorld`), and it goes back to the
+        // guest's slot straight after.
+        let lent = seen_guest.and_then(|conn| self.guests.iter().position(|g| g.0 == conn));
+        let active_player = match (seen_guest, lent) {
+            (None, _) => self.player.take(),
+            (Some(_), Some(k)) => self
+                .spare
+                .take()
+                .map(|stand_in| std::mem::replace(&mut self.guests[k].1, stand_in)),
+            (Some(_), None) => None,
+        };
+        let mut w = MobWorld {
+            others: &self.snapshot,
+            others_grid: Some(&self.grid),
+            others_index: Some(&self.index),
+            others_villagers: Some(&self.villager_slots),
+            others_solid: Some(&self.solid_slots),
+            difficulty,
+            nearest_player,
+            player: active_player,
+            teams: std::sync::Arc::clone(&base.teams),
+            ..*base
+        };
+        w.controlling_player_movement = controlling_player_movement;
+        self.ticked.insert(self.mobs[i].mob().id);
+        // `horizon-report`'s cost by mob type (tick_profile.rs).
+        let sampled =
+            super::super::tick_profile::start().map(|t| (t, self.mobs[i].mob().kind.clone()));
+        let kept = horizon_client::mobs::riders::tick(
+            &mut self.mobs,
+            i,
+            &w,
+            &mut self.random,
+            &mut self.events,
+        );
+        if let Some((started, kind)) = sampled {
+            super::super::tick_profile::add(shared.level_id, &kind, started);
+        }
+        let m = self.mobs[i].mob_mut();
+        if seen_guest.is_none() {
+            self.player = w.player;
+        } else if let Some(k) = lent {
+            self.spare = w
+                .player
+                .map(|v| std::mem::replace(&mut self.guests[k].1, v));
+        }
+        // Its tick changed its target: the mods may keep the one it had
+        // (`native_living_change_target`), and then what the tick did to
+        // the new one is not dealt.
+        if let Some(&before) = self.targets.get(i)
+            && m.ai.target != before
+        {
+            let named = target_named(
+                m.ai.target,
+                seen_guest,
+                &self.guests,
+                shared.guest_ids,
+                self.player.as_ref(),
+            );
+            if let Some((player, mob)) = named {
+                let kind = super::super::native_hooks::mob_kind(m);
+                let mut cancel = false;
+                Game::native_living_change_target(
+                    shared.level_id,
+                    m.id,
+                    &kind,
+                    player,
+                    mob,
+                    &mut cancel,
+                );
+                if cancel {
+                    if let Some(new) = m.ai.target {
+                        let mut k = from;
+                        while k < self.events.len() {
+                            if hits_target(&self.events[k], new) {
+                                self.events.remove(k);
+                            } else {
+                                k += 1;
+                            }
+                        }
+                    }
+                    m.ai.target = before;
+                }
+            }
+            self.targets[i] = m.ai.target;
+        }
+        // A killed mob's body is removed: its effects that act on
+        // death run (`LivingEntity.remove` with KILLED).
+        if !kept && !m.alive() && !m.removed && !m.extra.effects.is_empty() {
+            self.corpses.push(m.clone());
+        }
+        horizon_client::mobs::take_mob_pushes(&mut self.events, from, &mut self.pushes);
+        // LivingEntity.tickDeath: after twenty ticks the body is removed.
+        if !kept && !m.removed && !m.alive() && m.death_time >= 20 {
+            let (width, height) = (f64::from(m.width()), f64::from(m.height()));
+            // `AbstractCubeMob.remove` splits a grown cube into its smaller halves when it is
+            // removed, twenty ticks after the death. The parent leaves the list before the split
+            // runs, so its size, position, kind and persistence are captured here, as they are
+            // at the removal.
+            let split = if matches!(
+                m.species(),
+                Species::Slime | Species::MagmaCube | Species::SulfurCube
+            ) && m.extra.size > 1
+            {
+                Some(m.clone())
+            } else {
+                None
+            };
+            self.gone.push((m.id, m.position, width, height, split));
+        }
+        if let Some(conn) = seen_guest {
+            // ShulkerBullet.finalTarget is the guest's entity: the
+            // bullet homes on it while the projectiles tick, where
+            // `Who::Player` is the host's own player.
+            if let Some(&(_, id)) = shared.guest_ids.iter().find(|g| g.0 == conn) {
+                for e in &mut self.events[from..] {
+                    if let MobEvent::Projectile(p) = e
+                        && let ProjectileKind::ShulkerBullet { target, .. } = &mut p.kind
+                        && *target == Some(Who::Player)
+                    {
+                        *target = Some(Who::Mob(id));
+                    }
+                }
+            }
+            let mut k = from;
+            while k < self.events.len() {
+                if super::super::lan_mobs::acts_on_player(&self.events[k]) {
+                    self.guest_events.push((conn, m.id, self.events.remove(k)));
+                } else {
+                    k += 1;
+                }
+            }
+        }
+        // Killed during its own tick (drowning, burning, falling...).
+        if !m.alive() && !m.removed && self.died.insert(m.id) {
+            self.dead.push(m.clone());
+        }
+        if !kept {
+            self.died.remove(&m.id);
+        }
+        self.keep.push(kept);
+        // ServerLevel ticks entities one after another: the ones
+        // after this mob see it as it is now.
+        self.snapshot[i] = seen(m, shared.uuid);
+        self.snapshot[i].alive &= kept;
+        self.grid.update(i, self.snapshot[i].bounds);
+        horizon_client::mobs::apply_brain_events(&mut self.mobs, &mut self.events, from);
+    }
+}

crates/horizon-client/src/game/creatures/prep.rs

@@ -0,0 +1,247 @@
+//! What the mob loop (`tick_creature_entities`) prepares for every mob before it runs, on the tick's
+//! helper threads while the parallel tick is on (`horizon_client::parallel`): each answer reads only
+//! that mob, the blocks and the players, so the helpers give what one thread gives.
+use super::super::{Game, natural_spawning::NearestPlayers};
+use horizon_client::mobs::{Mob, MobWorld, Seen};
+use horizon_client::parallel::par_map;
+use horizon_core::geometry::Vec3;
+
+/// From this many mobs a preparation goes to the helper threads.
+const MIN_MOBS: usize = 256;
+
+impl Game<'_> {
+    /// `LivingEntity.isPushable` reads onClimbable live; a mob's cached one is only refreshed by its
+    /// own push step, so one that has not ticked yet (the first tick after a load, a fresh spawn) or
+    /// has moved since would be pushed out of a ladder or scaffolding. Refreshed here, from where
+    /// each mob stands now; then each mob's entry of the snapshot others push it by goes in
+    /// `snapshot`.
+    pub(super) fn prepare_mobs(&mut self, player: Option<[i32; 4]>, snapshot: &mut Vec<Seen>) {
+        let climbing: Vec<bool> = {
+            let w = self.mob_world_base(&[]);
+            let rebuild = hookless(&w);
+            par_map(&self.mobs, MIN_MOBS, |part, out| {
+                let w = rebuild();
+                out.extend(part.iter().map(|m| m.on_climbable(&w)));
+            })
+        };
+        for (m, climbing) in self.mobs.iter_mut().zip(climbing) {
+            m.on_climbable = climbing;
+        }
+        snapshot.extend(par_map(&self.mobs, MIN_MOBS, |part, out| {
+            out.extend(part.iter().map(|m| super::seen(m, player)));
+        }));
+    }
+}
+
+/// The nearest player each mob's despawn check reads (`Mob.checkDespawn`), found for every mob before
+/// the loop. A mob that moved before its turn is asked again where it stands.
+pub(super) struct MobsNearest<'p> {
+    players: &'p NearestPlayers<'p>,
+    /// Each mob's position as the loop started and its nearest player there; empty while the
+    /// parallel tick is off (every mob asks at its turn).
+    found: Vec<(Vec3, Option<Vec3>)>,
+}
+
+impl<'p> MobsNearest<'p> {
+    /// `wanted`: whether the loop asks (not with one player, whose position answers).
+    pub(super) fn new(players: &'p NearestPlayers<'p>, mobs: &[Mob], wanted: bool) -> Self {
+        let found = if wanted && horizon_client::parallel::lanes() > 1 {
+            par_map(mobs, MIN_MOBS, |part, out| {
+                out.extend(
+                    part.iter()
+                        .map(|m| (m.position, Self::ask(players, m.position))),
+                );
+            })
+        } else {
+            Vec::new()
+        };
+        Self { players, found }
+    }
+    fn ask(players: &NearestPlayers, at: Vec3) -> Option<Vec3> {
+        players.nearest(at.x, at.y, at.z).map(|p| p.position)
+    }
+    /// The nearest player to mob `i` of the loop, standing at `at`.
+    pub(super) fn at(&self, i: usize, at: Vec3) -> Option<Vec3> {
+        match self.found.get(i) {
+            Some(&(was, nearest)) if was == at => nearest,
+            _ => Self::ask(self.players, at),
+        }
+    }
+}
+
+/// `base` without its hooks into the game, which can't leave the server thread, as a recipe a
+/// helper thread builds its own world from: native mod effects (the phases on helpers don't run
+/// them) and the UUID lookup.
+fn hookless<'w>(base: &MobWorld<'w>) -> impl Fn() -> MobWorld<'w> + Sync + use<'w> {
+    // Every field but the two hooks, named, so that a field added to `MobWorld` is decided here.
+    let MobWorld {
+        world,
+        registry,
+        facts,
+        fluid_tables,
+        movement,
+        collision,
+        suffocation,
+        data,
+        player,
+        controlling_player_movement,
+        nearest_player,
+        others,
+        dragon,
+        leash_boats,
+        others_grid,
+        others_index,
+        others_villagers,
+        others_solid,
+        daylight,
+        monsters_burn,
+        game_time,
+        day_time,
+        raining,
+        thundering,
+        level,
+        block_tags,
+        landing,
+        biomes,
+        mob_griefing,
+        universal_anger,
+        mob_drops,
+        max_entity_cramming,
+        difficulty,
+        dimension_type,
+        world_seed,
+        sea_level,
+        teams,
+        items,
+        lid_openers,
+        mod_effects: _,
+        who_lookup: _,
+        foreign: _,
+        border,
+    } = base.clone();
+    move || MobWorld {
+        world,
+        registry,
+        facts,
+        fluid_tables,
+        movement,
+        collision,
+        suffocation,
+        data,
+        player: player.clone(),
+        controlling_player_movement,
+        nearest_player,
+        others,
+        dragon,
+        leash_boats,
+        others_grid,
+        others_index,
+        others_villagers,
+        others_solid,
+        daylight,
+        monsters_burn,
+        game_time,
+        day_time,
+        raining,
+        thundering,
+        level,
+        block_tags,
+        landing,
+        biomes,
+        mob_griefing,
+        universal_anger,
+        mob_drops,
+        max_entity_cramming,
+        difficulty,
+        dimension_type,
+        world_seed,
+        sea_level,
+        teams: std::sync::Arc::clone(&teams),
+        items,
+        lid_openers,
+        mod_effects: None,
+        who_lookup: None,
+        foreign: None,
+        border,
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::super::super::natural_spawning::LevelPlayer;
+    use super::*;
+    use horizon_core::geometry::Aabb;
+
+    #[test]
+    #[ignore = "needs prepared resources and scratch homes/cwd"]
+    fn preparation_refreshes_climbing_and_nearest_after_a_move() {
+        let mut game = super::super::super::native_machines::tests::server_game();
+        game.build_registries();
+        game.set_allow_commands(true);
+        for command in [
+            "setblock 8 80 9 stone",
+            "setblock 8 80 8 ladder[facing=north]",
+            "summon pig 8.5 80 8.5 {NoAI:1b}",
+            "summon pig 9.5 80 8.5 {NoAI:1b}",
+        ] {
+            assert!(game.run_command(command), "{command}");
+        }
+        assert_eq!(game.mobs.len(), 2);
+        let examples = game.mobs.clone();
+        game.mobs = (0..MIN_MOBS + 37)
+            .map(|i| {
+                let mut mob = examples[i % 2].clone();
+                mob.id = 10_000 + i as u32;
+                mob.on_climbable = i % 2 != 0; // deliberately stale in both cases
+                mob
+            })
+            .collect::<Vec<_>>()
+            .into();
+        let mut snapshot = Vec::new();
+        game.prepare_mobs(None, &mut snapshot);
+        assert_eq!(snapshot.len(), game.mobs.len());
+        for (i, (m, seen)) in game.mobs.iter().zip(&snapshot).enumerate() {
+            assert_eq!(m.on_climbable, i % 2 == 0, "mob {i}");
+            assert_eq!(seen.id, m.id);
+            assert_eq!(seen.position, m.position);
+            assert_eq!(seen.pushable, i % 2 != 0, "snapshot {i}");
+        }
+
+        let player = |x, spectator| {
+            let position = Vec3::new(x, 80.0, 8.5);
+            LevelPlayer {
+                uuid: 0,
+                position,
+                bounds: Aabb::new(position, position),
+                alive: false, // dead players still count for checkDespawn
+                creative: false,
+                spectator,
+            }
+        };
+        let players = [player(8.5, true), player(7.5, false), player(9.5, false)];
+        let index = NearestPlayers::new(&players);
+        let nearest = MobsNearest::new(&index, &game.mobs, true);
+        for (i, mob) in game.mobs.iter().enumerate() {
+            let expected = if i % 2 == 0 {
+                players[1].position
+            } else {
+                players[2].position
+            };
+            assert_eq!(nearest.at(i, mob.position), Some(expected));
+            // A previous mob's push moved this mob across the nearest-player tie.
+            assert_eq!(
+                nearest.at(i, Vec3::new(10.0, 80.0, 8.5)),
+                Some(players[2].position)
+            );
+        }
+        let no_cache = MobsNearest::new(&index, &game.mobs, false);
+        assert_eq!(
+            no_cache.at(0, game.mobs[0].position),
+            Some(players[1].position)
+        );
+        let spectators = [player(8.5, true)];
+        let index = NearestPlayers::new(&spectators);
+        let nearest = MobsNearest::new(&index, &game.mobs, true);
+        assert_eq!(nearest.at(0, game.mobs[0].position), None);
+    }
+}

crates/horizon-client/src/game/entity.rs

@@ -594,24 +594,9 @@ impl Game<'_> {
         splash: &mut impl FnMut(horizon_client::entities::ItemSplash),
     ) -> (bool, bool) {
         let no_physics = escape_speed.is_some();
-        let heights = item.push_by_fluid(&self.world, &self.res.motion().fluids);
-        // Entity.baseTick splashes before moving, except on the first tick.
-        item.water_entry(heights.0 > 0.0, item.ticks == 0, splash);
-        item.lava_fall(heights.1 > 0.0);
-        let step = item.tick_move_before_block_effects(
-            &self.world,
-            self.res.facts(),
-            self.collision,
-            no_physics,
-            escape_speed,
-            heights,
-            entity_boxes,
-            Some(self.border.geometry()),
-            Some(&self.res.motion().blocks),
-        );
-        item.check_fall(&step, |fall, from, moved| {
-            self.vehicle_fall_reset(fall, from, moved)
-        });
+        let step = self
+            .item_tables()
+            .move_item(item, escape_speed, entity_boxes, splash);
         if let Some(pos) = step.vertical_contact {
             let state = self.world.block(pos.x, pos.y, pos.z);
             let position = item.position;
@@ -630,23 +615,9 @@ impl Game<'_> {
                 return (false, true);
             }
         }
-        item.finish_move_block_effects(
-            &self.world,
-            self.res.facts(),
-            no_physics,
-            Some(&self.res.motion().blocks),
-            step,
-        );
-        item.push_by_bubble_column(
-            &self.world,
-            self.pack.assets.registry,
-            self.res.facts(),
-            self.collision,
-        );
-        let alive = item.tick_settle(&self.world, self.res.facts(), no_physics, step);
-        // ItemEntity.tick refreshes immersion and splashes on water entry.
-        let (water, _) = item.push_by_fluid(&self.world, &self.res.motion().fluids);
-        item.water_entry(water > 0.0, false, splash);
+        let alive = self
+            .item_tables()
+            .settle_item(item, no_physics, step, splash);
         (alive, false)
     }
 

crates/horizon-client/src/game/item_entities.rs

@@ -2,6 +2,8 @@
 //! `tick_experience`), the stacks a destroyed container lets fall, and spawning and merging orbs.
 use super::*;
 
+mod ahead;
+
 impl Game<'_> {
     pub(super) fn tick_items(&mut self) {
         // EntityTickEvent.Pre for the items about to tick, and the Post after them.
@@ -43,6 +45,13 @@ impl Game<'_> {
             horizon_client::entities::ObstacleGrid::new(self.item_entity_collisions());
         let mut nearby = Vec::new();
         let mut bubble_tops = Vec::new();
+        let mut moves = ahead::Moves::work_out(self, &items, &obstacles, round, |item| {
+            !self.joined_during_entity_pass(item.id)
+                && ticking(
+                    item.position.x.floor() as i32,
+                    item.position.z.floor() as i32,
+                )
+        });
         // `ItemUtils.onContainerDestroyed`: what a destroyed item held, and
         // where it lay, for adding to the level once its own tick is done.
         let mut destroyed: Vec<(ItemStack, Vec3)> = Vec::new();
@@ -65,33 +74,35 @@ impl Game<'_> {
             {
                 continue;
             }
-            // `getEntityCollisions(entity, bounds.expandTowards(movement))`: the boxes around
-            // where the item can reach this tick (a block of slack for the next velocity).
-            let reach = Vec3::new(
-                item.velocity.x.abs() + 1.0,
-                item.velocity.y.abs() + 1.0,
-                item.velocity.z.abs() + 1.0,
-            );
-            obstacles.near(item.bounds().inflate(reach), &mut nearby);
-            // ItemEntity.tick's noPhysics (servers only): out of a block, a solid entity
-            // or the border, placed or closed on it, towards the closest open side (#1252).
-            let no_physics = merging
-                && item.inside_block(
-                    &self.world,
-                    self.collision,
+            let mut at_turn = |item: &mut ItemEntity| {
+                // `getEntityCollisions(entity, bounds.expandTowards(movement))`: the boxes around
+                // where the item can reach this tick.
+                obstacles.near(ahead::reach(item), &mut nearby);
+                // ItemEntity.tick's noPhysics (servers only): out of a block, a solid entity
+                // or the border, placed or closed on it, towards the closest open side (#1252).
+                let no_physics = merging
+                    && item.inside_block(
+                        &self.world,
+                        self.collision,
+                        &nearby,
+                        Some(self.border.geometry()),
+                    );
+                // `moveTowardsClosestSpace` draws the speed from the item's own random.
+                let escape_speed = no_physics.then(|| item.random.next_float() * 0.2 + 0.1);
+                let mut entered = Vec::new();
+                let (alive, discarded) = self.tick_item_with_block_contacts(
+                    item,
+                    escape_speed,
                     &nearby,
-                    Some(self.border.geometry()),
+                    false,
+                    &mut |splash| entered.push(splash),
                 );
-            // `moveTowardsClosestSpace` draws the speed from the item's own random.
-            let escape_speed = no_physics.then(|| item.random.next_float() * 0.2 + 0.1);
-            let mut entered = Vec::new();
-            let (alive, discarded) = self.tick_item_with_block_contacts(
-                item,
-                escape_speed,
-                &nearby,
-                false,
-                &mut |splash| entered.push(splash),
-            );
+                (no_physics, alive, discarded, entered)
+            };
+            let (no_physics, alive, discarded, entered) = match moves.take(i, item, &mut at_turn) {
+                Some(moved) => moved,
+                None => at_turn(item),
+            };
             // Entity.doWaterSplashEffect: the server plays the sound (broadcast
             // to clients). Its addParticle is empty.
             for splash in entered {

crates/horizon-client/src/game/item_entities/ahead.rs

@@ -0,0 +1,336 @@
+//! The dropped items' own moves, worked out ahead of the item loop on the tick's helper threads
+//! while the parallel tick is on (`horizon_client::parallel`).
+//!
+//! An item's move (`ItemEntity.tick` up to its merge) reads only itself, the level's blocks, the
+//! obstacle boxes and the border, and the item loop changes none of them: so an item's move worked
+//! out on a copy before the loop is the move its turn would make. Its escape from inside a block
+//! draws the item's own random, which the copy carries. A merge before its turn can change its
+//! count, age and pickup delay: such an item moves at its turn instead. With a native mod loaded
+//! (its hooks see every item) or a mod listening to the round, every item moves at its turn.
+use super::super::{boats::fall_reset, entity::Entity as _, native_ticks};
+use super::*;
+use horizon_client::{
+    blocks::BlockFacts,
+    entities::{ItemEntity, ItemSplash, ItemStep, ObstacleGrid},
+    motion::MotionResources,
+    world::World,
+};
+use horizon_core::geometry::{Aabb, Vec3, border::BorderGeometry};
+use horizon_world::{block_collision::BlockCollisionRegistry, registry::BlockRegistry};
+
+/// From this many items their moves go to the helper threads.
+const MIN_ITEMS: usize = 256;
+
+/// What an item's own move reads.
+pub(in crate::game) struct ItemTables<'a> {
+    world: &'a World,
+    facts: &'a BlockFacts,
+    collision: &'a BlockCollisionRegistry<'a>,
+    motion: &'a MotionResources,
+    registry: &'a BlockRegistry,
+    border: &'a BorderGeometry,
+}
+
+impl<'w> Game<'w> {
+    pub(in crate::game) fn item_tables(&self) -> ItemTables<'_> {
+        ItemTables {
+            world: &self.world,
+            facts: self.res.facts(),
+            collision: self.collision,
+            motion: self.res.motion(),
+            registry: self.pack.assets.registry,
+            border: self.border.geometry(),
+        }
+    }
+}
+
+impl ItemTables<'_> {
+    /// `ItemEntity.tick` up to its move's vertical contact: the fluids, the water splash, the move
+    /// (pushed out at `escape_speed` when inside a block) and its fall.
+    pub(in crate::game) fn move_item(
+        &self,
+        item: &mut ItemEntity,
+        escape_speed: Option<f32>,
+        entity_boxes: &[Aabb],
+        splash: &mut impl FnMut(ItemSplash),
+    ) -> ItemStep {
+        let heights = item.push_by_fluid(self.world, &self.motion.fluids);
+        // Entity.baseTick splashes before moving, except on the first tick.
+        item.water_entry(heights.0 > 0.0, item.ticks == 0, splash);
+        item.lava_fall(heights.1 > 0.0);
+        let step = item.tick_move_before_block_effects(
+            self.world,
+            self.facts,
+            self.collision,
+            escape_speed.is_some(),
+            escape_speed,
+            heights,
+            entity_boxes,
+            Some(self.border),
+            Some(&self.motion.blocks),
+        );
+        item.check_fall(&step, |fall, from, moved| {
+            fall_reset(self.world, self.motion, self.facts, fall, from, moved)
+        });
+        step
+    }
+    /// The rest of the move after its vertical contact: block effects, bubble columns, settling
+    /// and the water entry. Whether the item lives on.
+    pub(in crate::game) fn settle_item(
+        &self,
+        item: &mut ItemEntity,
+        no_physics: bool,
+        step: ItemStep,
+        splash: &mut impl FnMut(ItemSplash),
+    ) -> bool {
+        item.finish_move_block_effects(
+            self.world,
+            self.facts,
+            no_physics,
+            Some(&self.motion.blocks),
+            step,
+        );
+        item.push_by_bubble_column(self.world, self.registry, self.facts, self.collision);
+        let alive = item.tick_settle(self.world, self.facts, no_physics, step);
+        // ItemEntity.tick refreshes immersion and splashes on water entry.
+        let (water, _) = item.push_by_fluid(self.world, &self.motion.fluids);
+        item.water_entry(water > 0.0, false, splash);
+        alive
+    }
+}
+
+/// `getEntityCollisions(entity, bounds.expandTowards(movement))`: the box around an item that its
+/// move reads others' boxes in (a block of slack for the next velocity).
+pub(super) fn reach(item: &ItemEntity) -> Aabb {
+    let v = item.velocity;
+    item.bounds()
+        .inflate(Vec3::new(v.x.abs() + 1.0, v.y.abs() + 1.0, v.z.abs() + 1.0))
+}
+
+/// An item's move as its turn makes it: without physics (inside a block), alive, discarded, and
+/// its splashes.
+pub(super) type Moved = (bool, bool, bool, Vec<ItemSplash>);
+
+/// One item's move, worked out on a copy.
+pub(super) struct Ahead {
+    moved: ItemEntity,
+    result: Moved,
+    /// What a merge before its turn can change: its count, age and pickup delay.
+    started: (u32, i32, i64),
+}
+
+fn started(item: &ItemEntity) -> (u32, i32, i64) {
+    (item.stack.count, item.age, item.pickup_delay)
+}
+
+/// `item`'s whole move on a server with no native mod, from the tables alone.
+fn move_alone(
+    tables: &ItemTables,
+    obstacles: &ObstacleGrid,
+    item: &mut ItemEntity,
+    found: &mut Vec<u32>,
+    near: &mut Vec<Aabb>,
+) -> Moved {
+    obstacles.near_shared(reach(item), found, near);
+    let no_physics = item.inside_block(tables.world, tables.collision, near, Some(tables.border));
+    // `moveTowardsClosestSpace` draws the speed from the item's own random.
+    let escape_speed = no_physics.then(|| item.random.next_float() * 0.2 + 0.1);
+    let mut splashes = Vec::new();
+    let step = tables.move_item(item, escape_speed, near, &mut |s| splashes.push(s));
+    if step.vertical_contact.is_some() && item.stack.count == 0 {
+        return (no_physics, false, true, splashes);
+    }
+    let alive = tables.settle_item(item, no_physics, step, &mut |s| splashes.push(s));
+    (no_physics, alive, false, splashes)
+}
+
+#[cfg(test)]
+thread_local! {
+    /// A test's choice of the path, over the parallel tick's switch.
+    static AHEAD: std::cell::Cell<Option<bool>> = const { std::cell::Cell::new(None) };
+    /// The moves a test's tick put in place, and those a merge made it refuse.
+    static TAKEN: std::cell::Cell<(usize, usize)> = const { std::cell::Cell::new((0, 0)) };
+}
+
+/// Whether the moves are worked out ahead: while the parallel tick is on.
+fn ahead() -> bool {
+    #[cfg(test)]
+    if let Some(ahead) = AHEAD.with(std::cell::Cell::get) {
+        return ahead;
+    }
+    horizon_client::parallel::lanes() > 1
+}
+
+/// The moves worked out ahead, by the item's place in the loop.
+pub(super) struct Moves(Vec<Option<Ahead>>);
+
+impl Moves {
+    /// The moves of the items that will tick (`ticks`), on the helper threads; none while the
+    /// parallel tick is off, for few items, or with a mod that could see an item's move.
+    pub(super) fn work_out(
+        game: &Game,
+        items: &[ItemEntity],
+        obstacles: &ObstacleGrid,
+        round: &native_ticks::TickRound,
+        ticks: impl Fn(&ItemEntity) -> bool,
+    ) -> Self {
+        if !ahead()
+            || items.len() < MIN_ITEMS
+            || game.remote.is_some()
+            || round.listening()
+            || !horizon_client::mod_api::mods().is_empty()
+        {
+            return Self(Vec::new());
+        }
+        let todo: Vec<usize> = (0..items.len()).filter(|&i| ticks(&items[i])).collect();
+        let tables = game.item_tables();
+        let done = horizon_client::parallel::par_map(&todo, MIN_ITEMS / 4, |part, out| {
+            let (mut found, mut near) = (Vec::new(), Vec::new());
+            for &i in part {
+                let mut moved = items[i].clone();
+                let result = move_alone(&tables, obstacles, &mut moved, &mut found, &mut near);
+                let started = started(&items[i]);
+                out.push((
+                    i,
+                    Ahead {
+                        moved,
+                        result,
+                        started,
+                    },
+                ));
+            }
+        });
+        let mut moves: Vec<Option<Ahead>> = (0..items.len()).map(|_| None).collect();
+        for (i, a) in done {
+            moves[i] = Some(a);
+        }
+        Self(moves)
+    }
+    /// Item `i`'s move, put in place, when it was worked out and no merge changed the item since.
+    /// Under `HORIZON_PARALLEL_TICK=verify` the move is made again at its turn, on a copy, and a
+    /// difference is logged.
+    pub(super) fn take(
+        &mut self,
+        i: usize,
+        item: &mut ItemEntity,
+        again: impl FnOnce(&mut ItemEntity) -> Moved,
+    ) -> Option<Moved> {
+        let a = self.0.get_mut(i)?.take()?;
+        let fits = a.started == started(item);
+        #[cfg(test)]
+        TAKEN.with(|t| {
+            let (taken, refused) = t.get();
+            t.set((taken + usize::from(fits), refused + usize::from(!fits)));
+        });
+        if !fits {
+            return None;
+        }
+        if horizon_client::parallel::parallel_tick() == 2 {
+            let mut serial = item.clone();
+            let expected = again(&mut serial);
+            check(item.id, &serial, &expected, &a.moved, &a.result);
+        }
+        *item = a.moved;
+        Some(a.result)
+    }
+}
+
+/// `HORIZON_PARALLEL_TICK=verify`: an item's move worked out ahead against the one at its turn.
+fn check(id: u32, serial: &ItemEntity, expected: &Moved, ahead: &ItemEntity, got: &Moved) {
+    use std::sync::atomic::{AtomicU64, Ordering};
+    static CHECKED: AtomicU64 = AtomicU64::new(0);
+    static DIFFERED: AtomicU64 = AtomicU64::new(0);
+    if format!("{serial:?} {expected:?}") != format!("{ahead:?} {got:?}") {
+        DIFFERED.fetch_add(1, Ordering::Relaxed);
+        log::error!("parallel tick: item {id}'s move worked out ahead differs from its turn's");
+    }
+    let n = CHECKED.fetch_add(1, Ordering::Relaxed) + 1;
+    if n == 1 || n.is_multiple_of(10_000) {
+        log::info!(
+            "parallel tick: {n} item moves worked out ahead checked, {} differed",
+            DIFFERED.load(Ordering::Relaxed)
+        );
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::super::super::native_machines::tests::server_game;
+    use super::{AHEAD, TAKEN};
+    use horizon_client::{chunk::Chunk, items::ItemStack};
+    use std::sync::Arc;
+
+    /// One item tick with the moves worked out ahead leaves every item, and the level's random,
+    /// as the tick that moves each item at its turn: items escaping a block (their own random),
+    /// floating in water, falling, and merging into a later item before its turn (whose move
+    /// is then made at its turn).
+    #[test]
+    #[ignore = "needs prepared resources in HORIZON_TEST_RESOURCES"]
+    fn moves_worked_out_ahead_match_those_made_at_each_turn() {
+        let mut game = server_game();
+        let feet = horizon_core::geometry::Vec3::new(8.5, 70.0, 8.5);
+        game.player = Some(horizon_client::motion::PlayerMotion::new(game.motion, feet).unwrap());
+        // `ChunkMap.prepareEntityTickingChunk`: the chunks within 2 are loaded too.
+        for x in -2..=2 {
+            for z in -2..=2 {
+                game.world
+                    .chunks
+                    .insert((x, z), Arc::new(Chunk::empty(x, z)));
+            }
+        }
+        game.update_chunk_tickets();
+        let registry = game.pack.assets.registry;
+        let stone = registry.default_state("minecraft:stone").unwrap();
+        let water = registry.default_state("minecraft:water").unwrap();
+        for x in 0..16 {
+            for z in 0..16 {
+                let (fx, fz) = (f64::from(x) + 0.5, f64::from(z) + 0.5);
+                let stack = |count| ItemStack::new("minecraft:stone", count);
+                match (x + z) % 5 {
+                    0 => {
+                        game.world.set(x, 65, z, stone).unwrap();
+                        game.spawn_item(stack(1), [fx, 65.3, fz], [0.0; 3], 10);
+                    }
+                    1 => {
+                        game.world.set(x, 65, z, water).unwrap();
+                        game.spawn_item(stack(1), [fx, 65.5, fz], [0.0; 3], 10);
+                    }
+                    2 => {
+                        game.spawn_item(stack(1), [fx, 66.0, fz], [0.0; 3], 0);
+                        game.spawn_item(stack(5), [fx, 66.0, fz], [0.0; 3], 0);
+                    }
+                    _ => game.spawn_item(stack(1), [fx, 67.0, fz], [0.1, 0.2, -0.1], 10),
+                }
+            }
+        }
+        for item in game.dropped.iter_mut() {
+            item.ticks = 39;
+        }
+        game.ticked_entities_through = u32::MAX;
+        let (items, random) = (game.dropped.clone(), game.world.random.clone());
+        let mut runs = Vec::new();
+        for ahead in [false, true] {
+            game.dropped = items.clone();
+            game.world.random = random.clone();
+            AHEAD.with(|a| a.set(Some(ahead)));
+            TAKEN.with(|t| t.set((0, 0)));
+            game.tick_items();
+            let left: Vec<String> = game.dropped.iter().map(|i| format!("{i:?}")).collect();
+            runs.push((
+                left,
+                format!("{:?}", game.world.random),
+                TAKEN.with(|t| t.get()),
+            ));
+        }
+        AHEAD.with(|a| a.set(None));
+        assert!(runs[0].0.len() < items.len(), "some items merged");
+        assert_eq!(runs[0].0, runs[1].0);
+        assert_eq!(runs[0].1, runs[1].1);
+        // The path under test ran: most moves were put in place, and a merge refused some.
+        let (taken, refused) = runs[1].2;
+        assert!(
+            taken >= 200 && refused > 0,
+            "{taken} put in place, {refused} refused"
+        );
+    }
+}

crates/horizon-client/src/game/lan.rs

@@ -4,6 +4,7 @@
 //! costs one `Option` check per tick.
 mod event_timings;
 mod level_event_targets;
+mod mob_states;
 
 use super::{Game, shield::OFFHAND_SLOT};
 use event_timings::EventTimings;
@@ -363,12 +364,12 @@ pub(super) fn baby_index(species: Species) -> Option<u8> {
 /// own: `mob_meta`'s arms hold these mobs' other entries.
 fn mob_variant_meta(
     m: &Mob,
-    hub: &Hub,
+    sync: &SyncData,
     meta: &mut Vec<(u8, i32, Value)>,
     defaults: &mut Vec<(u8, i32, Value)>,
 ) {
     let (kind, e) = (m.kind.id.as_str(), &m.extra);
-    let id = |registry: &str, name: &str| hub.registry_id(registry, &format!("minecraft:{name}"));
+    let id = |registry: &str, name: &str| sync.id(registry, &format!("minecraft:{name}"));
     match kind {
         // Axolotl.DATA_VARIANT (after AgeableMob's two).
         "minecraft:axolotl" => {
@@ -501,7 +502,7 @@ fn mob_variant_meta(
 /// `defaults` gets the defaults of the entries that can go back to theirs.
 pub(super) fn mob_meta(
     m: &Mob,
-    hub: &Hub,
+    sync: &SyncData,
     player_entity: Option<i32>,
     game_time: i64,
     meta: &mut Vec<(u8, i32, Value)>,
@@ -558,7 +559,7 @@ pub(super) fn mob_meta(
     // LivingEntity.DATA_EFFECT_PARTICLES and DATA_EFFECT_AMBIENCE_ID: the
     // swirls of its effects (the ones a mob has are visible, not ambient).
     meta.extend(
-        hub.effect_meta(
+        sync.effect_meta(
             (m.extra.effects.iter())
                 .filter(|e| !m.extra.hidden_effects.contains(&e.0))
                 .map(|e| (e.0, m.extra.ambient_effects.contains(&e.0))),
@@ -590,7 +591,7 @@ pub(super) fn mob_meta(
         }
     }
     let e = &m.extra;
-    let id = |registry: &str, name: &str| hub.registry_id(registry, &format!("minecraft:{name}"));
+    let id = |registry: &str, name: &str| sync.id(registry, &format!("minecraft:{name}"));
     // Pig.DATA_BOOST_TIME and Strider.DATA_BOOST_TIME (18, the first entry
     // after AgeableMob's two). The total `boost()` stored in `counter2`.
     // The default 0 is not registered: vanilla never writes it back, and the
@@ -598,7 +599,7 @@ pub(super) fn mob_meta(
     if matches!(kind, "minecraft:pig" | "minecraft:strider") && e.counter2 > 0 {
         meta.push((18, serializer::INT, Value::Int(e.counter2)));
     }
-    mob_variant_meta(m, hub, meta, defaults);
+    mob_variant_meta(m, sync, meta, defaults);
     mob_raider_meta(m, meta, defaults);
     match kind {
         "minecraft:guardian" | "minecraft:elder_guardian" => {
@@ -1948,14 +1949,15 @@ impl Game<'_> {
         mut kept: Option<&mut super::lan_idle::IdlePass<'_>>,
     ) {
         out.clear();
-        let mut speeds = super::lan_attributes::SpeedByType::default();
+        let mut states = mob_states::Later::new(self.world.game_time);
         // ChunkMap.tick sends the changes of an entity outside the
         // entity-ticking range only when it changed section or needs sync.
         let levels = self.tickets.levels();
         let idle_at = |p: horizon_core::geometry::Vec3| {
             !levels.in_entity_ticking_range((p.x.floor() as i32) >> 4, (p.z.floor() as i32) >> 4)
         };
-        for m in self.mobs.iter().filter(|m| !m.removed && m.id > after) {
+        for (index, m) in (self.mobs.iter().enumerate()).filter(|(_, m)| !m.removed && m.id > after)
+        {
             // The lead's holder: the player the mob sees, another mob or a
             // fence knot (game/lan_entities.rs).
             let player_entity = || {
@@ -1986,6 +1988,7 @@ impl Game<'_> {
             let bare = idle
                 && (kept.as_deref_mut()).is_some_and(|k| k.bare(hub, m.id as i32, kind, &place))
                 && !m.needs_sync;
+            let slot = out.as_slice().len();
             let e = out.add(m.id as i32, entity_uuid(m.id), kind);
             e.position = place.position;
             (e.yaw, e.pitch, e.head_yaw) = place.rotation;
@@ -1999,21 +2002,11 @@ impl Game<'_> {
             if bare {
                 continue;
             }
-            mob_meta(
-                m,
-                hub,
-                player_entity(),
-                self.world.game_time,
-                &mut e.meta,
-                &mut e.defaults,
-            );
-            // HappyGhast.aiStep: requiresPrecisePosition while on a still timeout.
-            e.precise = m.species() == horizon_client::mobs::Species::HappyGhast
-                && horizon_client::mobs::happy_ghast::on_still_timeout(m);
-            mob_freeze_meta(m, &mut e.meta, &mut e.defaults);
+            let (sync, sources) = (hub.sync_data(), self.attr_sources());
+            states.state((slot, index), m, player_entity(), sync, sources, e);
             mob_equipment(m, hub, &mut e.equipment);
-            self.mob_attributes(m, &mut speeds, &mut e.attributes);
         }
+        states.fill(&self.mobs, hub.sync_data(), self.attr_sources(), out);
         for d in self.dropped.iter().filter(|d| d.id > after) {
             let idle = idle_at(d.position);
             let place = super::lan_idle::Place {
@@ -4116,7 +4109,7 @@ mod tests {
         let hub = meta_hub("RaiderMetaTest");
         let meta_of = |m: &Mob| {
             let (mut meta, mut defaults) = (Vec::new(), Vec::new());
-            super::mob_meta(m, &hub, None, 0, &mut meta, &mut defaults);
+            super::mob_meta(m, hub.sync_data(), None, 0, &mut meta, &mut defaults);
             (meta, defaults)
         };
 
@@ -4145,7 +4138,7 @@ mod tests {
             // The same health: the ghoul's own maximum is 40, the zombie's 20.
             m.health = 10.0;
             let (mut meta, mut defaults) = (Vec::new(), Vec::new());
-            super::mob_meta(&m, &hub, None, 0, &mut meta, &mut defaults);
+            super::mob_meta(&m, hub.sync_data(), None, 0, &mut meta, &mut defaults);
             (meta, defaults)
         };
         for (template, id) in [

crates/horizon-client/src/game/lan/mob_states.rs

@@ -0,0 +1,286 @@
+//! The part of each mob's state for its clients (`Game::lan_entities`) that reads only the mob and
+//! the synced tables: its entity data and attribute instances. While the parallel tick is on
+//! (`horizon_client::parallel`) it is built after the pass, on the tick's helper threads, and put in
+//! place; otherwise in place during the pass, as one thread always did.
+use super::super::lan_attributes::{AttrSources, SpeedByType};
+use super::{mob_freeze_meta, mob_meta};
+use horizon_client::lan::{
+    data::SyncData,
+    entity::{EntityState, States},
+};
+use horizon_client::mobs::Mob;
+use horizon_protocol::{entity_data::Value, packets::play::AttributeSnapshot};
+
+/// From this many mobs the states go to the helper threads.
+const MIN_MOBS: usize = 1024;
+
+/// One mob's entity data (`meta`, and the `defaults` its entries can go back to) and attribute
+/// instances; whether its position is sent precisely.
+#[allow(clippy::too_many_arguments)]
+pub(super) fn mob_state(
+    m: &Mob,
+    sync: &SyncData,
+    sources: AttrSources,
+    player_entity: Option<i32>,
+    game_time: i64,
+    speeds: &mut SpeedByType,
+    meta: &mut Vec<(u8, i32, Value)>,
+    defaults: &mut Vec<(u8, i32, Value)>,
+    attributes: &mut Vec<AttributeSnapshot>,
+) -> bool {
+    mob_meta(m, sync, player_entity, game_time, meta, defaults);
+    mob_freeze_meta(m, meta, defaults);
+    sources.mob_attributes(m, speeds, attributes);
+    // HappyGhast.aiStep: requiresPrecisePosition while on a still timeout.
+    m.species() == horizon_client::mobs::Species::HappyGhast
+        && horizon_client::mobs::happy_ghast::on_still_timeout(m)
+}
+
+#[cfg(test)]
+thread_local! {
+    /// A test's choice of the path, over the parallel tick's switch.
+    static DEFER: std::cell::Cell<Option<bool>> = const { std::cell::Cell::new(None) };
+}
+
+/// Whether the states wait for the helper threads: while the parallel tick is on.
+fn deferring() -> bool {
+    #[cfg(test)]
+    if let Some(defer) = DEFER.with(std::cell::Cell::get) {
+        return defer;
+    }
+    horizon_client::parallel::lanes() > 1
+}
+
+/// The mobs' states of one pass: built in place, or (while the parallel tick is on) the mobs whose
+/// states wait for the helper threads: (their slot in the states, their index in the mobs, the
+/// player entity their lead may name).
+pub(super) struct Later {
+    waiting: Option<Vec<(usize, usize, Option<i32>)>>,
+    game_time: i64,
+    speeds: SpeedByType,
+}
+
+impl Later {
+    pub(super) fn new(game_time: i64) -> Self {
+        Self {
+            waiting: deferring().then(Vec::new),
+            game_time,
+            speeds: SpeedByType::default(),
+        }
+    }
+    /// Mob `index`'s state `e`, in `slot` of the states: built now, or left for [`Self::fill`].
+    pub(super) fn state(
+        &mut self,
+        (slot, index): (usize, usize),
+        m: &Mob,
+        player_entity: Option<i32>,
+        sync: &SyncData,
+        sources: AttrSources,
+        e: &mut EntityState,
+    ) {
+        if let Some(waiting) = &mut self.waiting {
+            waiting.push((slot, index, player_entity));
+            return;
+        }
+        let (meta, defaults, attributes) = (&mut e.meta, &mut e.defaults, &mut e.attributes);
+        e.precise = mob_state(
+            m,
+            sync,
+            sources,
+            player_entity,
+            self.game_time,
+            &mut self.speeds,
+            meta,
+            defaults,
+            attributes,
+        );
+    }
+    /// The waiting states, built on the helper threads and put in place.
+    pub(super) fn fill(
+        self,
+        mobs: &[Mob],
+        sync: &SyncData,
+        sources: AttrSources,
+        out: &mut States,
+    ) {
+        let Some(waiting) = self.waiting else {
+            return;
+        };
+        let game_time = self.game_time;
+        let built = horizon_client::parallel::par_map(&waiting, MIN_MOBS, |part, done| {
+            let mut speeds = SpeedByType::default();
+            for &(slot, index, player) in part {
+                let (mut meta, mut defaults, mut attributes) = (Vec::new(), Vec::new(), Vec::new());
+                let precise = mob_state(
+                    &mobs[index],
+                    sync,
+                    sources,
+                    player,
+                    game_time,
+                    &mut speeds,
+                    &mut meta,
+                    &mut defaults,
+                    &mut attributes,
+                );
+                done.push((slot, meta, defaults, attributes, precise));
+            }
+        });
+        let states = out.as_mut_slice();
+        for (slot, meta, defaults, attributes, precise) in built {
+            let e = &mut states[slot];
+            (e.meta, e.defaults, e.attributes, e.precise) = (meta, defaults, attributes, precise);
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use horizon_client::lan::{NetConfig, hub::HubConfig};
+
+    /// The states put in place after the pass are those built in place during it, for mobs whose
+    /// entity data and attributes vary: effects, armor, a horse's rolled jump, variants, a baby,
+    /// a happy ghast, and what `/attribute` changed.
+    #[test]
+    #[ignore = "needs prepared resources and scratch homes/cwd"]
+    fn deferred_states_match_those_built_in_place() {
+        crate::mod_api::in_own_process(
+            "game::lan::mob_states::tests::deferred_states_match_those_built_in_place",
+            deferred_states_match_those_built_in_place_in_own_process,
+        );
+    }
+
+    fn deferred_states_match_those_built_in_place_in_own_process() {
+        let mut game = super::super::super::native_machines::tests::server_game();
+        game.build_registries();
+        game.set_allow_commands(true);
+        for command in [
+            "summon piglin 8.5 80 8.5 {NoAI:1b,PersistenceRequired:1b}",
+            "effect give @e[type=piglin] speed 30 1",
+            "item replace entity @e[type=piglin,limit=1] armor.head with iron_helmet",
+            "summon horse 10.5 80 8.5 {NoAI:1b}",
+            "summon sheep 12.5 80 8.5 {NoAI:1b,Color:3b,Age:-24000}",
+            "summon axolotl 14.5 80 8.5 {NoAI:1b,Variant:2}",
+            "summon happy_ghast 16.5 84 8.5 {NoAI:1b}",
+            "summon pig 18.5 80 8.5 {NoAI:1b}",
+            "summon frog 20.5 80 8.5 {NoAI:1b}",
+            "attribute @e[type=pig,limit=1] minecraft:scale base set 1.5",
+        ] {
+            assert!(game.run_command(command), "{command}");
+        }
+        let examples = game.mobs.clone();
+        game.mobs = (0..MIN_MOBS * 2 + 37)
+            .map(|i| {
+                let mut mob = examples[i % examples.len()].clone();
+                mob.id = 10_000 + i as u32;
+                mob.removed = i % 11 == 0;
+                mob
+            })
+            .collect::<Vec<_>>()
+            .into();
+        game.lan_host(
+            super::super::LanSettings {
+                port: 0,
+                guest_commands: true,
+                force_game_mode: false,
+            },
+            NetConfig::loopback(0, String::new(), 2),
+            HubConfig {
+                default_game_mode: 0,
+                guest_commands: true,
+                force_game_mode: false,
+                hardcore: false,
+                view_distance: 2,
+                simulation_distance: 2,
+                dimension: "minecraft:overworld".into(),
+                allow_flight: false,
+                seed: 0,
+                max_players: 20,
+                integrated: false,
+                levels: Default::default(),
+            },
+        )
+        .unwrap();
+        let mut lan = game.lan.take().unwrap();
+        // The shared tables must retain pack-selected effect colors, not just
+        // vanilla's registry color. This process owns the global pack palette.
+        horizon_client::brewing::set_pack_colors(vec![("speed".into(), 0xAA12_3456)]);
+        let players = Default::default();
+        let mut runs = Vec::new();
+        struct Reset;
+        impl Drop for Reset {
+            fn drop(&mut self) {
+                DEFER.with(|d| d.set(None));
+            }
+        }
+        let _reset = Reset;
+        for after in [0, 10_012] {
+            runs.clear();
+            for defer in [false, true, false] {
+                DEFER.with(|d| d.set(Some(defer)));
+                let mut states = States::default();
+                game.lan_entities(&players, &mut lan.hub, &mut states, false, after, None);
+                runs.push(states.as_slice().to_vec());
+            }
+            let expected_ids: Vec<_> = game
+                .mobs
+                .iter()
+                .filter(|m| !m.removed && m.id > after)
+                .map(|m| m.id as i32)
+                .collect();
+            assert_eq!(
+                runs[0].iter().map(|e| e.id).collect::<Vec<_>>(),
+                expected_ids
+            );
+            assert!(runs[0].iter().filter(|e| !e.meta.is_empty()).count() >= MIN_MOBS);
+            assert!(runs[0].iter().any(|e| !e.attributes.is_empty()));
+            let colored: Vec<_> = runs[0]
+                .iter()
+                .filter(|e| e.kind == "minecraft:piglin")
+                .collect();
+            assert!(!colored.is_empty());
+            for e in colored {
+                let particles = e.meta.iter().find(|e| e.0 == 10).unwrap();
+                let Value::Particles(particles) = &particles.2 else {
+                    panic!("effect particle metadata");
+                };
+                assert_eq!(particles.len(), 1);
+                assert_eq!(
+                    particles[0].data,
+                    Some(horizon_protocol::entity_data::ParticleData::Color(
+                        0xFF12_3456_u32 as i32
+                    ))
+                );
+            }
+            assert_eq!(runs[0], runs[1], "deferred, after {after}");
+            assert_eq!(runs[0], runs[2], "switch off, after {after}");
+
+            // With deferral off the caller can read the metadata immediately,
+            // before fill, and fill must not replace the vectors it already owns.
+            let m = game
+                .mobs
+                .iter()
+                .find(|m| !m.removed && m.id > after)
+                .unwrap();
+            let mut states = States::default();
+            let e = states.add(m.id as i32, 0, &m.kind.id);
+            let mut later = Later::new(game.world.game_time);
+            later.state((0, 0), m, None, lan.hub.sync_data(), game.attr_sources(), e);
+            assert_eq!(e.meta, runs[0][0].meta);
+            assert_eq!(e.defaults, runs[0][0].defaults);
+            assert_eq!(e.attributes, runs[0][0].attributes);
+            let pointers = (e.meta.as_ptr(), e.defaults.as_ptr(), e.attributes.as_ptr());
+            later.fill(
+                &game.mobs,
+                lan.hub.sync_data(),
+                game.attr_sources(),
+                &mut states,
+            );
+            let e = &states.as_slice()[0];
+            assert_eq!(
+                pointers,
+                (e.meta.as_ptr(), e.defaults.as_ptr(), e.attributes.as_ptr())
+            );
+        }
+    }
+}

crates/horizon-client/src/game/lan/sulfur_cube_meta_tests.rs

@@ -22,7 +22,7 @@ fn a_sulfur_cube_sends_its_fuse_at_19_and_from_bucket_at_20() {
     let hub = meta_hub("SulfurCubeMetaTest");
     let meta_of = |m: &Mob| {
         let (mut meta, mut defaults) = (Vec::new(), Vec::new());
-        super::mob_meta(m, &hub, None, 0, &mut meta, &mut defaults);
+        super::mob_meta(m, hub.sync_data(), None, 0, &mut meta, &mut defaults);
         (meta, defaults)
     };
 

crates/horizon-client/src/game/lan_attributes.rs

@@ -103,10 +103,32 @@ impl Game<'_> {
         }
     }
 
+    /// What a mob's synced attribute instances are read from (`mob_attributes`).
+    pub(super) fn attr_sources(&self) -> AttrSources<'_> {
+        AttrSources {
+            catalog: &self.motion.attributes,
+            items: self.items(),
+            commands: &self.cmds.attributes,
+        }
+    }
+}
+
+/// What the synced attribute instances are read from: the attribute catalog, the item table and
+/// what `/attribute` changed. Plain tables, so the tick's helper threads can read them
+/// (`Game::lan_entities`).
+#[derive(Clone, Copy)]
+pub(super) struct AttrSources<'a> {
+    catalog: &'a horizon_world::attributes::entity::Catalog,
+    items: &'a horizon_client::items::Items,
+    commands:
+        &'a std::collections::HashMap<Target, horizon_world::attributes::entity::EntityAttributes>,
+}
+
+impl AttrSources<'_> {
     /// The registry id of a client-syncable attribute the entity type has
     /// (`AttributeMap.getInstance` is null for the others), by name.
     fn synced_attribute(&self, entity: &str, name: &str) -> Option<i32> {
-        let catalog = &self.motion.attributes;
+        let catalog = self.catalog;
         let id = catalog.id(name)?;
         (catalog.attribute(id).syncable && catalog.default_base(entity, id).is_some())
             .then(|| id.registry_id())
@@ -114,7 +136,7 @@ impl Game<'_> {
 
     /// The type's base value of an attribute, as the instance starts.
     fn attribute_default(&self, entity: &str, name: &str) -> f64 {
-        let catalog = &self.motion.attributes;
+        let catalog = self.catalog;
         catalog
             .id(name)
             .and_then(|id| catalog.default_base(entity, id))
@@ -132,7 +154,7 @@ impl Game<'_> {
         out: &mut Vec<AttributeSnapshot>,
     ) {
         for (slot, item) in slots {
-            let Some(info) = item.and_then(|i| self.items().get(i)) else {
+            let Some(info) = item.and_then(|i| self.items.get(i)) else {
                 continue;
             };
             for (attribute, id, amount, op, group) in &info.modifiers {
@@ -306,6 +328,16 @@ impl Game<'_> {
         }
     }
 
+    /// The instances `/attribute` created in an entity's generic map
+    /// replace the ones derived from its state.
+    fn generic_attributes_of(&self, t: Target, out: &mut Vec<AttributeSnapshot>) {
+        if let Some(map) = self.commands.get(&t) {
+            generic_attributes_into(map, out);
+        }
+    }
+}
+
+impl Game<'_> {
     /// An armor stand's synced instances: movement speed (its first
     /// `aiStep`), its equipment's modifiers and what `/attribute` changed.
     pub(super) fn stand_attributes(
@@ -315,18 +347,19 @@ impl Game<'_> {
     ) -> Vec<AttributeSnapshot> {
         const KIND: &str = "minecraft:armor_stand";
         let mut out = Vec::new();
-        if let Some(a) = self.synced_attribute(KIND, "minecraft:movement_speed") {
+        let sources = self.attr_sources();
+        if let Some(a) = sources.synced_attribute(KIND, "minecraft:movement_speed") {
             instance(
                 &mut out,
                 a,
-                self.attribute_default(KIND, "minecraft:movement_speed"),
+                sources.attribute_default(KIND, "minecraft:movement_speed"),
             );
         }
         let slots = horizon_client::entities::ArmorStand::SLOTS
             .into_iter()
             .zip(equipment.iter().map(|s| s.as_ref().map(|s| &s.item)));
-        self.equipment_attributes(KIND, |_| None, slots, &mut out);
-        self.generic_attributes_of(Target::Stand(id), &mut out);
+        sources.equipment_attributes(KIND, |_| None, slots, &mut out);
+        sources.generic_attributes_of(Target::Stand(id), &mut out);
         out
     }
 
@@ -372,14 +405,6 @@ impl Game<'_> {
             }
         }
     }
-
-    /// The instances `/attribute` created in an entity's generic map
-    /// replace the ones derived from its state.
-    fn generic_attributes_of(&self, t: Target, out: &mut Vec<AttributeSnapshot>) {
-        if let Some(map) = self.cmds.attributes.get(&t) {
-            generic_attributes_into(map, out);
-        }
-    }
 }
 
 #[cfg(test)]

crates/horizon-client/src/game/piston_push.rs

@@ -738,14 +738,11 @@ impl Game<'_> {
         });
         self.set_pushed_on_ground(target, limited, resolved);
 
-        // Entity.move updates ground before its tiny-collision return. This
-        // native slice covers only unoccupied server NoAI horizontal moves.
-        if let Target::Entity(EntityHandle::Mob(i)) = target {
+        // shortcut: unoccupied server mobs only, expand when controller/passenger authority is bound.
+        let mob = if let Target::Entity(EntityHandle::Mob(i)) = target {
             let m = &self.mobs[i];
-            let eligible = !self.options.execution_role.is_client()
+            (!self.options.execution_role.is_client()
                 && self.remote.is_none()
-                && limited.y == 0.0
-                && m.no_ai
                 && !m.extra.passenger
                 && m.extra.mount.is_none()
                 && m.rider.is_none()
@@ -757,16 +754,22 @@ impl Game<'_> {
                         .guests
                         .iter()
                         .any(|g| host.hub.here(g) && g.vehicle == Some(m.id))
-                });
-            if eligible {
-                self.mobs[i].on_ground = limited.y < 0.0 && limited.y != resolved.y;
-            }
+                }))
+            .then_some(i)
+        } else {
+            None
+        };
+        // Entity.move updates ground before its tiny-collision return.
+        if let Some(i) = mob
+            && self.mobs[i].no_ai
+        {
+            self.mobs[i].on_ground = limited.y < 0.0 && limited.y != resolved.y;
+        }
+        let tnt_after_move =
+            tnt.is_some() && !self.options.execution_role.is_client() && self.remote.is_none();
+        if mob.is_none() && !tnt_after_move {
+            self.restitute_piston(target, limited, resolved);
         }
-        // `Entity.move(MoverType.PISTON, …)` restitution, which runs whether or
-        // not the step is applied: so a primed TNT in a chunk that does not tick
-        // entities still loses its velocity into walls and ceilings and
-        // bounces on slime (#12022).
-        self.restitute_piston(target, limited, resolved);
         let moved = resolved.length_squared() > 1.0e-7
             || limited.length_squared() - resolved.length_squared() < 1.0e-7;
         if moved {
@@ -775,6 +778,12 @@ impl Game<'_> {
             // A move the collisions cut to nothing still ends in the check.
             self.release_after_block_move(target);
         }
+        // Restitution reads the new landing block even when the push was blocked.
+        if let Some(i) = mob {
+            self.mobs[i].restitute_movement(&self.world, &self.motion.blocks, limited, resolved);
+        } else if tnt_after_move {
+            self.restitute_piston(target, limited, resolved);
+        }
         if let Some((i, from, ..)) = tnt {
             self.finish_tnt_piston_move(i, from, resolved);
         }
@@ -1053,6 +1062,7 @@ mod tests {
 mod order_tests {
     use crate::game::{Audio, Game, Options, Progress, Settings, load_resources};
     use horizon_client::chunk::Chunk;
+    use horizon_core::geometry::Vec3;
     use horizon_core::{geometry::Direction, position::BlockPos};
     use horizon_world::level::pistons::MovingPiston;
     use std::sync::Arc;
@@ -1151,8 +1161,9 @@ mod order_tests {
         }
         // The near stone's box reaches x 5.5 into a TNT from x 5.01: 0.49 plus
         // PUSH_OFFSET; the far stone's old box stops it at x 6.
+        // Vanilla's TNT width is 0.98F, not the exact decimal 0.98.
         assert!(
-            (moved[0] - 0.5).abs() < 1e-9,
+            (moved[0] - 0.500000009536743).abs() < 1e-9,
             "far stone gone first: {}",
             moved[0]
         );
@@ -1251,7 +1262,7 @@ mod order_tests {
     }
 
     /// A primed TNT pushed down by a piston onto a slime block bounces up,
-    /// matching vanilla's slime restitution (~+0.64) even in a lazy chunk (#12022).
+    /// matching vanilla's restitution even in a lazy chunk (#12022).
     #[test]
     #[ignore = "needs prepared resources in HORIZON_TEST_RESOURCES"]
     fn a_pushed_tnt_bounces_on_slime() {
@@ -1312,32 +1323,36 @@ mod order_tests {
             death_ticks: 0,
             id: 1,
         };
-        game.block_ticks.moving_pistons = vec![(BlockPos::new(5, 200, 5), piston)];
-        game.sync_moving_pistons();
-        game.tnts.clear();
-        assert!(game.run_command("summon tnt 5.5 200.5 5.5 {fuse:80,Motion:[0.0d,-0.4d,0.0d]}"));
-        assert!(
-            (game.tnts[0].velocity.y - (-0.4)).abs() < 1.0e-9,
-            "TNT starts with vy -0.4, got {:?}",
-            game.tnts[0].velocity
-        );
-        // Piston advances from progress 0 to 0.5 and pushes the TNT down.
-        let pushes: Vec<_> = game
-            .block_ticks
-            .moving_pistons
-            .iter()
-            .map(|&(pos, old)| (pos, old, old.progress + 0.5))
-            .collect();
-        for (_, piston) in &mut game.block_ticks.moving_pistons {
-            piston.progress += 0.5;
+        // Pinned 26.3 PistonMovingBlockEntity.moveCollidedEntities, without entity ticks.
+        for (floor, y, vy, final_y, final_vy) in [
+            (slime, 200.5, -0.4, 200.0, 0.3510000138133766_f64),
+            (stone, 200.5, -0.4, 200.0, 0.0),
+            (slime, 200.0, -0.4, 200.0, 0.3200000047683716),
+            (slime, 200.00001, -0.4, 200.00001, 0.32000064476813395),
+            (slime, 200.5, 0.3, 200.0, 0.0),
+        ] {
+            game.world.set(5, 199, 5, floor).unwrap();
+            game.block_ticks.moving_pistons = vec![(BlockPos::new(5, 200, 5), piston)];
+            game.sync_moving_pistons();
+            game.tnts.clear();
+            assert!(game.run_command(&format!(
+                "summon tnt 5.5 {y} 5.5 {{fuse:80,Motion:[0.0d,{vy}d,0.0d]}}"
+            )));
+            assert_eq!(game.tnts[0].velocity.y, vy);
+            let pushes: Vec<_> = game
+                .block_ticks
+                .moving_pistons
+                .iter()
+                .map(|&(pos, old)| (pos, old, old.progress + 0.5))
+                .collect();
+            for (_, piston) in &mut game.block_ticks.moving_pistons {
+                piston.progress += 0.5;
+            }
+            game.push_piston_entities(pushes);
+            assert_eq!(game.tnts[0].position, Vec3::new(5.5, final_y, 5.5));
+            assert_eq!(game.tnts[0].velocity.y.to_bits(), final_vy.to_bits());
+            assert!(game.tnts[0].on_ground);
         }
-        game.push_piston_entities(pushes);
-        // The slime block at y=199 bounces the TNT up; vertical velocity must be positive.
-        assert!(
-            game.tnts[0].velocity.y > 0.0,
-            "TNT should bounce up on slime, got vy {:?}",
-            game.tnts[0].velocity
-        );
     }
 }
 
@@ -1602,6 +1617,54 @@ mod h255_mob_piston_ground_tests {
     }
     #[test]
     #[ignore = "needs prepared resources in HORIZON_TEST_RESOURCES"]
+    fn h255_added_tnt_authority_and_exhausted_budget() {
+        use crate::game::execution_role::ExecutionRole;
+        for role in [
+            ExecutionRole::Client,
+            ExecutionRole::IntegratedServer,
+            ExecutionRole::DedicatedServer,
+        ] {
+            with_role_game(role, |game| {
+                assert!(game.run_command("setblock 5 199 5 slime_block"));
+                assert!(
+                    game.run_command("summon tnt 5.5 200.5 5.5 {fuse:80,Motion:[0d,-0.4d,0d]}")
+                );
+                let mut budget = std::collections::HashMap::new();
+                let push = |game: &mut Game<'_>,
+                            budget: &mut std::collections::HashMap<_, _>,
+                            movement| {
+                    let tnt = &game.tnts[0];
+                    game.move_pushed(
+                        Target::Entity(EntityHandle::Tnt(0)),
+                        tnt.bounds(),
+                        [tnt.position.x, tnt.position.z],
+                        0.0,
+                        tnt.on_ground,
+                        movement,
+                        Direction::Down,
+                        budget,
+                    );
+                };
+                push(game, &mut budget, Vec3::ZERO);
+                assert_eq!(game.tnts[0].position.y, 200.5);
+                assert_eq!(game.tnts[0].velocity.y, -0.4);
+                push(game, &mut budget, Vec3::new(0.0, -0.51, 0.0));
+                let expected = if role.is_client() {
+                    0.0_f64
+                } else {
+                    0.3510000138133766
+                };
+                assert_eq!(game.tnts[0].velocity.y.to_bits(), expected.to_bits());
+                game.tnts[0].position.y = 200.5;
+                game.tnts[0].velocity.y = -0.4;
+                push(game, &mut budget, Vec3::new(0.0, -0.51, 0.0));
+                assert_eq!(game.tnts[0].position.y, 200.5);
+                assert_eq!(game.tnts[0].velocity.y, -0.4);
+            });
+        }
+    }
+    #[test]
+    #[ignore = "needs prepared resources in HORIZON_TEST_RESOURCES"]
     fn h255_added_no_request_and_exhausted_budget_keep_state() {
         with_game(|game| {
             let i = added_mob(game, 20, "horse", true, 8.5);
@@ -1613,6 +1676,16 @@ mod h255_mob_piston_ground_tests {
             added_push(game, i, Vec3::new(0.51, 0.0, 0.0));
             assert!(game.mobs[i].on_ground, "occupied path remains excluded");
             assert_eq!(game.mobs[i].position, Vec3::new(9.01, 202.0, 4.5));
+            assert!(game.run_command("setblock 9 201 4 slime_block"));
+            game.mobs[i].position.y = 202.25;
+            game.mobs[i].velocity = Vec3::new(0.0, -1.0, 0.0);
+            game.mobs[i].sync_position = false;
+            added_push(game, i, Vec3::new(0.0, -0.5, 0.0));
+            assert_eq!(
+                game.mobs[i].velocity.y, 0.0,
+                "occupied restitution stays outside this slice"
+            );
+            assert!(!game.mobs[i].sync_position);
             assert!(game.run_command("ride @s dismount"));
             assert!(game.riding.is_none());
             assert!(game.mobs[i].rider.is_none());
@@ -1655,23 +1728,46 @@ mod h255_mob_piston_ground_tests {
     }
     #[test]
     #[ignore = "needs prepared resources in HORIZON_TEST_RESOURCES"]
-    fn h255_added_client_role_ai_and_vertical_paths_keep_native_state() {
+    fn h255_added_client_role_ai_and_vertical_ground_state() {
         use crate::game::execution_role::ExecutionRole as Role;
-        for (role, no_ai, movement) in [
-            (Role::Client, true, Vec3::new(0.25, 0.0, 0.0)),
-            (Role::IntegratedServer, false, Vec3::new(0.25, 0.0, 0.0)),
-            (Role::IntegratedServer, true, Vec3::new(0.0, 0.25, 0.0)),
+        for (role, no_ai, movement, on_ground) in [
+            (Role::Client, true, Vec3::new(0.25, 0.0, 0.0), true),
+            (
+                Role::IntegratedServer,
+                false,
+                Vec3::new(0.25, 0.0, 0.0),
+                true,
+            ),
+            (
+                Role::IntegratedServer,
+                true,
+                Vec3::new(0.0, 0.25, 0.0),
+                false,
+            ),
         ] {
             with_role_game(role, |game| {
                 let i = added_mob(game, 22, "husk", no_ai, 8.5);
                 let before = game.mobs[i].position;
                 added_push(game, i, movement);
                 assert_eq!(game.mobs[i].position, before.add_vector(movement));
-                assert!(
-                    game.mobs[i].on_ground,
-                    "excluded native path remains unchanged; not Java parity"
-                );
+                assert_eq!(game.mobs[i].on_ground, on_ground);
             });
         }
     }
+
+    #[test]
+    #[ignore = "needs prepared resources in HORIZON_TEST_RESOURCES"]
+    fn h255_added_partial_push_reads_slime_after_the_position_changes() {
+        with_game(|game| {
+            assert!(game.run_command("setblock 8 201 4 slime_block"));
+            let i = added_mob(game, 23, "iron_golem", true, 8.5);
+            game.mobs[i].position.y = 202.25;
+            game.mobs[i].velocity = Vec3::new(0.0, -1.0, 0.0);
+            game.mobs[i].sync_position = false;
+            added_push(game, i, Vec3::new(0.0, -0.5, 0.0));
+            assert_eq!(game.mobs[i].position, Vec3::new(8.5, 202.0, 4.5));
+            assert!(game.mobs[i].velocity.y > 0.9);
+            assert!(game.mobs[i].sync_position);
+        });
+    }
 }

crates/horizon-client/src/lan/data.rs

@@ -4,6 +4,7 @@
 //! (`data/minecraft/...`) and the registries report.
 use crate::chunk::LevelShapes;
 use horizon_protocol::{
+    entity_data,
     packets::configuration::{RegistryData, RegistryEntry, RegistryTags, TagEntry, UpdateTags},
     registry::{Builtin, Dynamic},
 };
@@ -506,6 +507,58 @@ impl SyncData {
             .any(|(_, e)| e.from_pack.contains(&true))
     }
 
+    /// `LivingEntity.updateSynchronizedMobEffectParticles`: DATA_EFFECT_PARTICLES
+    /// (10) and DATA_EFFECT_AMBIENCE_ID (11) for the visible `effects`, as
+    /// (effect, ambient). The entries are left out when there is nothing to
+    /// show, which is their default.
+    pub fn effect_meta<'a>(
+        &self,
+        effects: impl IntoIterator<Item = (&'a str, bool)>,
+    ) -> Vec<(u8, i32, entity_data::Value)> {
+        let mut particles = Vec::new();
+        let mut all_ambient = true;
+        for (effect, ambient) in effects {
+            let effect = effect.trim_start_matches("minecraft:");
+            // MobEffect.particleFactory: the omens have their own particles,
+            // the others an entity effect in their colour (faint if ambient).
+            let (name, data) = match effect {
+                "trial_omen" | "raid_omen" => (effect, None),
+                _ => {
+                    let Some(def) = crate::brewing::effect(effect) else {
+                        continue;
+                    };
+                    let alpha = if ambient { 38 } else { 255 };
+                    let argb = (alpha << 24) | def.shown_color() as i32;
+                    (
+                        "entity_effect",
+                        Some(entity_data::ParticleData::Color(argb)),
+                    )
+                }
+            };
+            let Some(id) = self.id("minecraft:particle_type", &format!("minecraft:{name}")) else {
+                continue;
+            };
+            particles.push(entity_data::Particle { id, data });
+            all_ambient &= ambient;
+        }
+        if particles.is_empty() {
+            return Vec::new();
+        }
+        let mut meta = vec![(
+            10,
+            entity_data::serializer::PARTICLES,
+            entity_data::Value::Particles(particles),
+        )];
+        if all_ambient {
+            meta.push((
+                11,
+                entity_data::serializer::BOOLEAN,
+                entity_data::Value::Bool(true),
+            ));
+        }
+        meta
+    }
+
     /// The id of a name in a built-in or synchronized registry.
     pub fn id(&self, registry: &str, name: &str) -> Option<i32> {
         if registry == "minecraft:data_component_type"

crates/horizon-client/src/lan/hub.rs

@@ -24,7 +24,7 @@ use horizon_core::{
 };
 use horizon_protocol::ids::play::clientbound as ids;
 use horizon_protocol::{
-    entity_data::{Particle, ParticleData, Value, serializer},
+    entity_data::{Value, serializer},
     packets::play::{self, Clientbound as C, Serverbound as S},
     text::{literal, translatable},
 };
@@ -3250,39 +3250,12 @@ impl Hub {
         &self,
         effects: impl IntoIterator<Item = (&'a str, bool)>,
     ) -> Vec<(u8, i32, Value)> {
-        let mut particles = Vec::new();
-        let mut all_ambient = true;
-        for (effect, ambient) in effects {
-            let effect = effect.trim_start_matches("minecraft:");
-            // MobEffect.particleFactory: the omens have their own particles,
-            // the others an entity effect in their colour (faint if ambient).
-            let (name, data) = match effect {
-                "trial_omen" | "raid_omen" => (effect, None),
-                _ => {
-                    let Some(def) = crate::brewing::effect(effect) else {
-                        continue;
-                    };
-                    let alpha = if ambient { 38 } else { 255 };
-                    let argb = (alpha << 24) | def.shown_color() as i32;
-                    ("entity_effect", Some(ParticleData::Color(argb)))
-                }
-            };
-            let Some(id) =
-                self.registry_id("minecraft:particle_type", &format!("minecraft:{name}"))
-            else {
-                continue;
-            };
-            particles.push(Particle { id, data });
-            all_ambient &= ambient;
-        }
-        if particles.is_empty() {
-            return Vec::new();
-        }
-        let mut meta = vec![(10, serializer::PARTICLES, Value::Particles(particles))];
-        if all_ambient {
-            meta.push((11, serializer::BOOLEAN, Value::Bool(true)));
-        }
-        meta
+        self.sync.effect_meta(effects)
+    }
+
+    /// The registries and tables the clients are synced with (read by helper threads).
+    pub fn sync_data(&self) -> &SyncData {
+        &self.sync
     }
 
     fn player_state(

crates/horizon-client/src/mobs.rs

@@ -5,6 +5,10 @@
 //! fly and amphibious node evaluators (`path`, `nav`), move/look/jump/body
 //! controls (`control`), `LivingEntity` physics (`living`) and per-species
 //! behaviour, models, animations and sounds.
+mod mob_slot;
+mod pending_pushes;
+pub use mob_slot::MobSlot;
+pub use pending_pushes::{PendingPushes, apply_mob_pushes};
 pub mod ai;
 pub mod allay;
 pub mod ambient;
@@ -2923,28 +2927,6 @@ pub fn take_mob_pushes(events: &mut Vec<MobEvent>, from: usize, pending: &mut Pe
     }
     events.truncate(kept);
 }
-/// Pushes waiting for their mob's tick, in push order. A list while short;
-/// past [`PendingPushes::LIST_MAX`] (a crowd pushing each other) one list
-/// per mob, as scanning the whole list for each mob grows with its square.
-#[derive(Default)]
-pub struct PendingPushes {
-    list: Vec<(u32, Vec3)>,
-    by_mob: crate::fast_hash::FxHashMap<u32, Vec<Vec3>>,
-}
-impl PendingPushes {
-    const LIST_MAX: usize = 64;
-    pub fn push(&mut self, id: u32, v: Vec3) {
-        if self.by_mob.is_empty() && self.list.len() < Self::LIST_MAX {
-            self.list.push((id, v));
-            return;
-        }
-        // Older pushes first, so each mob's stay in order.
-        for (id, v) in self.list.drain(..) {
-            self.by_mob.entry(id).or_default().push(v);
-        }
-        self.by_mob.entry(id).or_default().push(v);
-    }
-}
 /// `ServerLevel.sendBlockUpdated`: for each write that changed a block's collision
 /// shape, a mob whose path runs near it re-paths (`PathNavigation.shouldRecomputePath`).
 /// The writes whose shape changed are found once per tick ([`collision_changes`]).
@@ -2972,50 +2954,28 @@ pub fn collision_changes(w: &MobWorld, writes: &[((i32, i32, i32), u32, u32)]) -
         })
         .collect()
 }
-/// Adds the pending pushes aimed at `m` to its velocity, in push order.
-pub fn apply_mob_pushes(m: &mut Mob, pending: &mut PendingPushes) {
-    // Entity.push(x, y, z) sets needsSync.
-    let mut add = |v: Vec3| {
-        m.velocity = Vec3::new(m.velocity.x + v.x, m.velocity.y + v.y, m.velocity.z + v.z);
-        m.needs_sync = true;
-    };
-    if !pending.list.is_empty() {
-        pending.list.retain(|(id, v)| {
-            if *id != m.id {
-                return true;
-            }
-            add(*v);
-            false
-        });
-    }
-    if !pending.by_mob.is_empty()
-        && let Some(pushes) = pending.by_mob.remove(&m.id)
-    {
-        pushes.into_iter().for_each(add);
-    }
-}
 /// Applies the brain memories a mob's tick set on other mobs (from event
 /// index `from` on), at once: vanilla sets them on the other entity during
 /// the tick, before the mobs after it tick.
-pub fn apply_brain_events(mobs: &mut [Mob], events: &mut Vec<MobEvent>, from: usize) {
+pub fn apply_brain_events<M: MobSlot>(mobs: &mut [M], events: &mut Vec<MobEvent>, from: usize) {
     let mut i = from;
     while i < events.len() {
         match events[i] {
             MobEvent::BreedTarget { mob, partner } => {
-                if let Some(m) = mobs.iter_mut().find(|m| m.id == mob) {
+                if let Some(m) = mobs.iter_mut().map(MobSlot::mob_mut).find(|m| m.id == mob) {
                     animal_brain::set_breed_partner(m, partner);
                 }
                 events.remove(i);
             }
             MobEvent::LockGaze { mob, partner } => {
-                if let Some(m) = mobs.iter_mut().find(|m| m.id == mob) {
+                if let Some(m) = mobs.iter_mut().map(MobSlot::mob_mut).find(|m| m.id == mob) {
                     animal_brain::lock_gaze(m, partner);
                     villager_brain::lock_gaze_half(m, Who::Mob(partner));
                 }
                 events.remove(i);
             }
             MobEvent::Retreat { mob, from, at } => {
-                if let Some(m) = mobs.iter_mut().find(|m| m.id == mob) {
+                if let Some(m) = mobs.iter_mut().map(MobSlot::mob_mut).find(|m| m.id == mob) {
                     piglin::receive_retreat(m, from, at);
                 }
                 events.remove(i);
@@ -3029,7 +2989,11 @@ pub fn apply_brain_events(mobs: &mut [Mob], events: &mut Vec<MobEvent>, from: us
                 events.remove(i);
             }
             MobEvent::SchoolSize { leader, by } => {
-                if let Some(m) = mobs.iter_mut().find(|m| m.id == leader) {
+                if let Some(m) = mobs
+                    .iter_mut()
+                    .map(MobSlot::mob_mut)
+                    .find(|m| m.id == leader)
+                {
                     m.extra.followers += by;
                 }
                 events.remove(i);
@@ -3039,7 +3003,11 @@ pub fn apply_brain_events(mobs: &mut [Mob], events: &mut Vec<MobEvent>, from: us
             MobEvent::Breed {
                 parent, partner, ..
             } => {
-                if let Some(m) = mobs.iter_mut().find(|m| m.id == partner) {
+                if let Some(m) = mobs
+                    .iter_mut()
+                    .map(MobSlot::mob_mut)
+                    .find(|m| m.id == partner)
+                {
                     m.extra.age = 6000;
                     m.extra.in_love = 0;
                     animal_brain::forget_breed_target(m, parent);
@@ -3061,6 +3029,7 @@ pub fn apply_brain_events(mobs: &mut [Mob], events: &mut Vec<MobEvent>, from: us
                 let mut raised = Vec::new();
                 let (aim, fish_pos, fish_silent) = mobs
                     .iter()
+                    .map(MobSlot::mob)
                     .find(|m| m.id == mob)
                     .map(|m| {
                         (
@@ -3075,7 +3044,10 @@ pub fn apply_brain_events(mobs: &mut [Mob], events: &mut Vec<MobEvent>, from: us
                     })
                     .map(|(aim, pos, silent)| (Some(aim), Some(pos), silent))
                     .unwrap_or((None, None, true));
-                if let Some(t) = mobs.iter_mut().find(|m| m.id == target)
+                if let Some(t) = mobs
+                    .iter_mut()
+                    .map(MobSlot::mob_mut)
+                    .find(|m| m.id == target)
                     && t.damage_aimed(amount, source, aim)
                 {
                     t.add_effect("minecraft:poison", poison_ticks, 0);
@@ -3290,6 +3262,9 @@ pub struct MobWorld<'a> {
     /// not in the level (a warden's anger, `AngerManagement.angerByUuid`).
     /// None: this world keeps no entity UUIDs, so nothing is resolved.
     pub who_lookup: Option<&'a dyn WhoLookup>,
+    /// Another thread's mobs, for a lookup by id that misses `others`: a group of the server's
+    /// parallel mob loop (`game/mob_islands.rs`) answers as the serial loop would.
+    pub foreign: Option<&'a dyn ForeignSeen>,
     /// The level's world border, whose walls stop a mob's moves
     /// (`Entity.collectColliders`); None: nothing stops them there.
     pub border: Option<&'a horizon_core::geometry::border::BorderGeometry>,
@@ -3304,6 +3279,10 @@ pub trait ModEffectTicks {
 /// names while it is in the level, never some other entity as a stand-in. Only
 /// the game layer keeps entity UUIDs, so it answers this (`Game::who_by_uuid`)
 /// and the mob layer reaches it through `MobWorld`.
+/// A mob outside this thread's part of the mob loop, as the serial loop would see it now.
+pub trait ForeignSeen {
+    fn seen(&self, id: u32) -> Option<&Seen>;
+}
 pub trait WhoLookup {
     fn who_by_uuid(&self, uuid: [i32; 4]) -> Option<Who>;
 }

crates/horizon-client/src/mobs/aquatic.rs

@@ -749,18 +749,35 @@ fn can_be_followed(s: Species, followers: i32) -> bool {
 }
 /// `startFollowing` on a recruit, applied after the recruiter's tick. A fish
 /// another leader took this tick (its snapshot was older) stays with it.
-pub fn start_following(mobs: &mut [Mob], fish: u32, leader: u32) {
-    let alive = |mobs: &[Mob], id: u32| mobs.iter().any(|m| m.id == id && m.alive() && !m.removed);
-    let Some(current) = mobs.iter().find(|m| m.id == fish).map(|m| m.extra.leader) else {
+pub fn start_following<M: super::MobSlot>(mobs: &mut [M], fish: u32, leader: u32) {
+    let alive = |mobs: &[M], id: u32| {
+        mobs.iter()
+            .map(super::MobSlot::mob)
+            .any(|m| m.id == id && m.alive() && !m.removed)
+    };
+    let Some(current) = mobs
+        .iter()
+        .map(super::MobSlot::mob)
+        .find(|m| m.id == fish)
+        .map(|m| m.extra.leader)
+    else {
         return;
     };
     if current.is_some_and(|l| alive(mobs, l)) {
         return;
     }
-    if let Some(m) = mobs.iter_mut().find(|m| m.id == fish) {
+    if let Some(m) = mobs
+        .iter_mut()
+        .map(super::MobSlot::mob_mut)
+        .find(|m| m.id == fish)
+    {
         m.extra.leader = Some(leader);
     }
-    if let Some(m) = mobs.iter_mut().find(|m| m.id == leader) {
+    if let Some(m) = mobs
+        .iter_mut()
+        .map(super::MobSlot::mob_mut)
+        .find(|m| m.id == leader)
+    {
         m.extra.followers += 1;
     }
 }

crates/horizon-client/src/mobs/goals.rs

@@ -21,7 +21,7 @@ pub struct Cx<'a, 'w> {
     pub lr: &'a mut LegacyRandom,
     pub events: &'a mut Vec<MobEvent>,
 }
-pub trait Goal: std::fmt::Debug {
+pub trait Goal: std::fmt::Debug + Send + Sync {
     fn flags(&self) -> u8;
     /// A goal that draws its first start delay in its constructor (a schooling fish's
     /// `FollowFlockLeaderGoal`): the delay again, when the mob's random is seeded afterwards.

crates/horizon-client/src/mobs/living.rs

@@ -849,7 +849,6 @@ impl Mob {
         self.horizontal_collision = x_hit || z_hit;
         // A server's mobs are authoritative (`isLocalInstanceAuthoritative`):
         // the vertical flags update even without vertical movement.
-        let moved_vertically = wanted.y.abs() > 0.0;
         let vertical = wanted.y != moved.y;
         self.vertical_collision_below = vertical && wanted.y < 0.0;
         self.on_ground = self.vertical_collision_below;
@@ -857,42 +856,51 @@ impl Mob {
         // checkFallDamage reads the landing block (getOnPosLegacy).
         self.supporting = self.supporting_block(w, moved);
         self.species_check_fall_damage(w, moved, events);
-        // Entity.restituteMovementAfterCollisions: the colliding components
-        // bounce back by the entity's bounciness or the block's (a slime
-        // block, a bed), whichever is larger. A mob never sneaks, so nothing
-        // suppresses its bounce there.
-        if moved_vertically && vertical || self.horizontal_collision {
-            // Entity.getOnPos(0.2F): the block whose bounce this move sees.
+        self.restitute_movement(w.world, w.movement, wanted, moved);
+        self.movement_emission(w, moved, events);
+        // Entity.move's last step: soul sand, honey and the like slow the
+        // horizontal speed of whatever the mob stands in or on.
+        let factor = self.block_speed_factor(w);
+        self.velocity = horizon_world::entity_move::scale_by_block_speed(self.velocity, factor);
+    }
+    /// `Entity.restituteMovementAfterCollisions`, shared by travel and piston pushes.
+    pub fn restitute_movement(
+        &mut self,
+        world: &crate::world::World,
+        movement: &horizon_world::movement_properties::BoundMovementProperties,
+        wanted: Vec3,
+        moved: Vec3,
+    ) {
+        let differs = |a: f64, b: f64| (b - a).abs() >= f64::from(1.0e-5_f32);
+        let x_hit = differs(wanted.x, moved.x);
+        let z_hit = differs(wanted.z, moved.z);
+        let vertical = wanted.y != moved.y;
+        if wanted.y.abs() > 0.0 && vertical || x_hit || z_hit {
             let on = (
                 self.position.x.floor() as i32,
                 (self.position.y - f64::from(0.2_f32)).floor() as i32,
                 self.position.z.floor() as i32,
             );
             let input = RestitutionInput {
-                block: w.world.block(on.0, on.1, on.2),
+                block: world.block(on.0, on.1, on.2),
                 collided_x: x_hit,
                 collided_z: z_hit,
                 vertical_collision: vertical,
-                collision_below: self.vertical_collision_below,
+                collision_below: vertical && wanted.y < 0.0,
                 living: true,
                 resolved_movement: moved,
                 fall_distance: self.fall_distance,
             };
-            // The port's phases cannot fail for a mob: a block outside the
-            // registry reads as one with no bounce, never as an error.
+            // Unknown block states have no bounce; the provider cannot fail.
             let _ = restitute_movement(
                 &mut MobRestitution {
                     mob: self,
-                    world: w,
+                    world,
+                    movement,
                 },
                 input,
             );
         }
-        self.movement_emission(w, moved, events);
-        // Entity.move's last step: soul sand, honey and the like slow the
-        // horizontal speed of whatever the mob stands in or on.
-        let factor = self.block_speed_factor(w);
-        self.velocity = horizon_world::entity_move::scale_by_block_speed(self.velocity, factor);
     }
     /// `Entity.getBlockSpeedFactor` at the mob's feet.
     fn block_speed_factor(&self, w: &MobWorld) -> f32 {
@@ -2103,20 +2111,21 @@ pub fn client_water_splash(mob: &mut Mob, out: &mut Vec<MobEvent>) {
 /// mob that never sneaks. The phases themselves are
 /// `movement_restitution::restitute_movement`, the same port the player and
 /// dropped items use.
-struct MobRestitution<'a, 'w> {
+struct MobRestitution<'a> {
     mob: &'a mut Mob,
-    world: &'a MobWorld<'w>,
+    world: &'a crate::world::World,
+    movement: &'a horizon_world::movement_properties::BoundMovementProperties,
 }
-impl<'a, 'w> MobRestitution<'a, 'w> {
+impl<'a> MobRestitution<'a> {
     /// The block's movement data: its coefficients and its bounce tag.
     fn movement(
         &self,
         state: BlockStateId,
     ) -> Result<horizon_world::movement_properties::MovementBlock, ()> {
-        self.world.movement.get(state).map_err(|_| ())
+        self.movement.get(state).map_err(|_| ())
     }
 }
-impl RestitutionWorld for MobRestitution<'_, '_> {
+impl RestitutionWorld for MobRestitution<'_> {
     // The block data cannot fail here: a state the registry does not know
     // reads as a block with no bounce, never as an error.
     type Error = ();
@@ -2162,7 +2171,7 @@ impl RestitutionWorld for MobRestitution<'_, '_> {
                 position.z.floor() as i32,
             ));
         };
-        let state = self.world.world.block(support[0], support[1], support[2]);
+        let state = self.world.block(support[0], support[1], support[2]);
         let keeps_column = self
             .movement(state)
             .is_ok_and(|b| b.wall || b.coefficients.fence_gate);

crates/horizon-client/src/mobs/living_tests.rs

@@ -174,6 +174,7 @@ fn a_mob_bounces_off_a_bouncy_block_and_not_off_stone() {
             dragon: None,
             leash_boats: &[],
             who_lookup: None,
+            foreign: None,
             others_grid: None,
             others_index: None,
             others_villagers: None,
@@ -291,6 +292,7 @@ fn a_ridden_horse_floats_in_deep_water() {
         dragon: None,
         leash_boats: &[],
         who_lookup: None,
+        foreign: None,
         others_grid: None,
         others_index: None,
         others_villagers: None,
@@ -411,6 +413,7 @@ fn a_falling_mob_lands_on_a_shulker_or_a_held_happy_ghast() {
             nearest_player: None,
             others: &others,
             who_lookup: None,
+            foreign: None,
             dragon: None,
             leash_boats: &[],
             others_grid: None,

crates/horizon-client/src/mobs/mob_slot.rs

@@ -0,0 +1,29 @@
+//! [`MobSlot`]: a mob a list hands out.
+use super::Mob;
+
+/// A mob a list hands out: the list's own, or a borrow of one in a larger list (a group of the
+/// server's parallel mob loop, `game/mob_islands.rs`, works on its mobs in place).
+pub trait MobSlot {
+    fn mob(&self) -> &Mob;
+    fn mob_mut(&mut self) -> &mut Mob;
+}
+impl MobSlot for Mob {
+    #[inline]
+    fn mob(&self) -> &Mob {
+        self
+    }
+    #[inline]
+    fn mob_mut(&mut self) -> &mut Mob {
+        self
+    }
+}
+impl MobSlot for &mut Mob {
+    #[inline]
+    fn mob(&self) -> &Mob {
+        self
+    }
+    #[inline]
+    fn mob_mut(&mut self) -> &mut Mob {
+        self
+    }
+}

crates/horizon-client/src/mobs/pending_pushes.rs

@@ -0,0 +1,103 @@
+//! The pushes a mob's tick gives others, waiting for each one's own tick.
+use super::Mob;
+use horizon_core::geometry::Vec3;
+
+/// Pushes waiting for their mob's tick, in push order. A list while short;
+/// past [`PendingPushes::LIST_MAX`] (a crowd pushing each other) one list
+/// per mob, as scanning the whole list for each mob grows with its square.
+#[derive(Default)]
+pub struct PendingPushes {
+    list: Vec<(u32, Vec3)>,
+    by_mob: crate::fast_hash::FxHashMap<u32, Vec<Vec3>>,
+}
+impl PendingPushes {
+    const LIST_MAX: usize = 64;
+    /// `other`'s pushes after these, each mob's in their order.
+    pub fn append(&mut self, other: PendingPushes) {
+        for (id, v) in other.list {
+            self.push(id, v);
+        }
+        for (id, list) in other.by_mob {
+            for v in list {
+                self.push(id, v);
+            }
+        }
+    }
+    pub fn push(&mut self, id: u32, v: Vec3) {
+        if self.by_mob.is_empty() && self.list.len() < Self::LIST_MAX {
+            self.list.push((id, v));
+            return;
+        }
+        // Older pushes first, so each mob's stay in order.
+        for (id, v) in self.list.drain(..) {
+            self.by_mob.entry(id).or_default().push(v);
+        }
+        self.by_mob.entry(id).or_default().push(v);
+    }
+}
+/// Adds the pending pushes aimed at `m` to its velocity, in push order.
+pub fn apply_mob_pushes(m: &mut Mob, pending: &mut PendingPushes) {
+    // Entity.push(x, y, z) sets needsSync.
+    let mut add = |v: Vec3| {
+        m.velocity = Vec3::new(m.velocity.x + v.x, m.velocity.y + v.y, m.velocity.z + v.z);
+        m.needs_sync = true;
+    };
+    if !pending.list.is_empty() {
+        pending.list.retain(|(id, v)| {
+            if *id != m.id {
+                return true;
+            }
+            add(*v);
+            false
+        });
+    }
+    if !pending.by_mob.is_empty()
+        && let Some(pushes) = pending.by_mob.remove(&m.id)
+    {
+        pushes.into_iter().for_each(add);
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::PendingPushes;
+    use horizon_core::geometry::Vec3;
+
+    /// Each mob's pushes, in the order they apply.
+    fn per_mob(p: &PendingPushes) -> std::collections::BTreeMap<u32, Vec<[f64; 3]>> {
+        let mut out = std::collections::BTreeMap::<u32, Vec<[f64; 3]>>::new();
+        for (id, v) in &p.list {
+            out.entry(*id).or_default().push([v.x, v.y, v.z]);
+        }
+        for (id, list) in &p.by_mob {
+            out.entry(*id)
+                .or_default()
+                .extend(list.iter().map(|v| [v.x, v.y, v.z]));
+        }
+        out
+    }
+
+    /// Two lists of pushes, the second appended to the first, apply as one list of them all in
+    /// that order, on either side of the switch from one list to a list per mob.
+    #[test]
+    fn appended_pushes_apply_in_order() {
+        let pushes: Vec<(u32, Vec3)> = (0..200u32)
+            .map(|k| (k * 7 % 11, Vec3::new(f64::from(k), 0.0, -f64::from(k))))
+            .collect();
+        let mut all = PendingPushes::default();
+        for &(id, v) in &pushes {
+            all.push(id, v);
+        }
+        for split in [0, 30, 64, 70, 150, 200] {
+            let (mut first, mut second) = (PendingPushes::default(), PendingPushes::default());
+            for &(id, v) in &pushes[..split] {
+                first.push(id, v);
+            }
+            for &(id, v) in &pushes[split..] {
+                second.push(id, v);
+            }
+            first.append(second);
+            assert_eq!(per_mob(&first), per_mob(&all), "split at {split}");
+        }
+    }
+}

crates/horizon-client/src/mobs/riders.rs

@@ -186,16 +186,23 @@ pub fn seat(attachment: [f64; 3], rider_offset: f64, at: Vec3, yaw: f32) -> Vec3
 
 /// `Entity.startRiding`: the link it makes, `rider` on `vehicle` as its first
 /// passenger (`addPassenger`).
-pub fn link(mobs: &mut [Mob], rider: u32, vehicle: u32) {
+pub fn link<M: super::MobSlot>(mobs: &mut [M], rider: u32, vehicle: u32) {
     if rider == vehicle {
         return;
     }
-    if let Some(m) = mobs.iter_mut().find(|m| m.id == rider) {
+    if let Some(m) = mobs
+        .iter_mut()
+        .map(super::MobSlot::mob_mut)
+        .find(|m| m.id == rider)
+    {
         super::shulker::on_start_riding(m);
         m.extra.mount = Some(vehicle);
         m.extra.passenger = true;
     }
-    if let Some(m) = mobs.iter_mut().find(|m| m.id == vehicle)
+    if let Some(m) = mobs
+        .iter_mut()
+        .map(super::MobSlot::mob_mut)
+        .find(|m| m.id == vehicle)
         && m.extra.rider.is_none()
     {
         m.extra.rider = Some(rider);
@@ -213,9 +220,10 @@ pub fn can_add_passenger(v: &Mob, aboard: usize) -> bool {
 }
 /// `Piglin.getTopPassenger`: down the chain of first passengers from
 /// `vehicle`, stopping at the `counter`th seat.
-fn top_passenger(mobs: &[Mob], vehicle: u32, counter: i32) -> u32 {
+fn top_passenger<M: super::MobSlot>(mobs: &[M], vehicle: u32, counter: i32) -> u32 {
     let first = mobs
         .iter()
+        .map(super::MobSlot::mob)
         .find(|m| m.id == vehicle)
         .and_then(|m| m.extra.rider);
     match first {
@@ -228,17 +236,22 @@ fn top_passenger(mobs: &[Mob], vehicle: u32, counter: i32) -> u32 {
 /// Not the vehicle it is on already, nor one it carries itself, and only
 /// after its boarding cooldown, onto a vehicle with no passenger
 /// (`canRide`, `canAddPassenger`). Returns whether it boarded.
-pub fn start_riding(mobs: &mut [Mob], rider: u32, vehicle: u32) -> bool {
+pub fn start_riding<M: super::MobSlot>(mobs: &mut [M], rider: u32, vehicle: u32) -> bool {
     // `Shulker.startRiding` puts its attach face back before the ride is
     // decided, so a shulker that is refused still ends up lid up.
-    if let Some(m) = mobs.iter_mut().find(|m| m.id == rider) {
+    if let Some(m) = mobs
+        .iter_mut()
+        .map(super::MobSlot::mob_mut)
+        .find(|m| m.id == rider)
+    {
         super::shulker::on_start_riding(m);
     }
-    let Some(r) = mobs.iter().find(|m| m.id == rider) else {
+    let Some(r) = mobs.iter().map(super::MobSlot::mob).find(|m| m.id == rider) else {
         return false;
     };
     let on_hoglin = mobs
         .iter()
+        .map(super::MobSlot::mob)
         .any(|m| m.id == vehicle && m.species() == Species::Hoglin);
     let vehicle = if r.species() == Species::Piglin && r.is_baby() && on_hoglin {
         top_passenger(mobs, vehicle, MAX_PASSENGERS_ON_ONE_HOGLIN)
@@ -252,17 +265,26 @@ pub fn start_riding(mobs: &mut [Mob], rider: u32, vehicle: u32) -> bool {
     // its vehicle.
     let mut up = Some(vehicle);
     while let Some(v) = up {
-        up = mobs.iter().find(|m| m.id == v).and_then(|m| m.extra.mount);
+        up = mobs
+            .iter()
+            .map(super::MobSlot::mob)
+            .find(|m| m.id == v)
+            .and_then(|m| m.extra.mount);
         if up == Some(rider) {
             return false;
         }
     }
-    let Some(v) = mobs.iter().find(|m| m.id == vehicle) else {
+    let Some(v) = mobs
+        .iter()
+        .map(super::MobSlot::mob)
+        .find(|m| m.id == vehicle)
+    else {
         return false;
     };
     let count = usize::from(v.rider.is_some())
         + mobs
             .iter()
+            .map(super::MobSlot::mob)
             .filter(|m| m.extra.mount == Some(vehicle))
             .count()
             .max(usize::from(v.extra.rider.is_some()));
@@ -470,25 +492,26 @@ pub fn solve_body_rot(head: f32, vehicle_body: f32) -> f32 {
 /// Tick a passenger with Mob.getNavigation's recursive vehicle redirection.
 /// Navigation is lent, not copied: goal start/stop/path progress all address
 /// the same object the vehicle ticks. The pathfinder sees the vehicle body.
-pub fn tick(
-    mobs: &mut [Mob],
+pub fn tick<M: super::MobSlot>(
+    mobs: &mut [M],
     index: usize,
     w: &MobWorld,
     random: &mut horizon_core::random::LegacyRandom,
     events: &mut Vec<super::MobEvent>,
 ) -> bool {
     let mut owner = index;
-    if !mobs[index].no_ai {
+    if !mobs[index].mob().no_ai {
         for _ in 0..mobs.len() {
             let Some(vehicle) = mobs[owner]
+                .mob()
                 .extra
                 .mount
-                .and_then(|id| mobs.iter().position(|m| m.id == id))
+                .and_then(|id| mobs.iter().position(|m| m.mob().id == id))
             else {
                 break;
             };
             if vehicle == index
-                || controller(&mobs[vehicle], w).is_none_or(|r| r.id != mobs[owner].id)
+                || controller(mobs[vehicle].mob(), w).is_none_or(|r| r.id != mobs[owner].mob().id)
             {
                 break;
             }
@@ -496,14 +519,14 @@ pub fn tick(
         }
     }
     if owner == index {
-        return mobs[index].tick(w, random, events);
+        return mobs[index].mob_mut().tick(w, random, events);
     }
     let (rider, vehicle) = if index < owner {
         let (left, right) = mobs.split_at_mut(owner);
-        (&mut left[index], &mut right[0])
+        (left[index].mob_mut(), right[0].mob_mut())
     } else {
         let (left, right) = mobs.split_at_mut(index);
-        (&mut right[0], &mut left[owner])
+        (right[0].mob_mut(), left[owner].mob_mut())
     };
     let body = vehicle.path_mob();
     std::mem::swap(&mut rider.ai.nav, &mut vehicle.ai.nav);

crates/horizon-client/src/mobs/seen_grid.rs

@@ -15,6 +15,9 @@ pub struct SeenGrid {
     /// The widest box (along x or z) the grid has held: no box reaches
     /// further than this past its minimum corner.
     reach: f64,
+    /// The entries a query counts when it decides to scan them all instead (at least the grid's
+    /// own): a grid of part of a list answers as the whole list's would.
+    entries: usize,
 }
 
 fn column(x: f64, z: f64) -> (i32, i32) {
@@ -39,6 +42,22 @@ impl SeenGrid {
         }
         grid
     }
+    /// The widest box and the entries it has, for a grid of part of its entries that must meet
+    /// the same entries in the same order (the server's parallel mob tick).
+    pub fn bounds(&self) -> (f64, usize) {
+        (self.reach, self.cell_of.len().max(self.entries))
+    }
+    /// [`Self::bounds`] of the grid of `boxes`, without building it.
+    pub fn bounds_of(boxes: impl Iterator<Item = Aabb>) -> (f64, usize) {
+        boxes.fold((0.0_f64, 0), |(reach, n), b| (reach.max(width(b)), n + 1))
+    }
+    /// Answer queries as a grid of `bounds` would: as wide a margin, and a scan past as many
+    /// entries.
+    pub fn answering_as(mut self, (reach, entries): (f64, usize)) -> Self {
+        self.reach = self.reach.max(reach);
+        self.entries = entries;
+        self
+    }
     /// Entry `i`'s box is now `b` (the mob moved or resized).
     pub fn update(&mut self, i: usize, b: Aabb) {
         self.reach = self.reach.max(width(b));
@@ -78,7 +97,7 @@ impl SeenGrid {
         let hi = column(max_x, max_z);
         let span = |a: i32, b: i32| i64::from(b) - i64::from(a) + 1;
         let (nx, nz) = (span(lo.0, hi.0), span(lo.1, hi.1));
-        let max_cells = (self.cell_of.len() as i64).max(64);
+        let max_cells = (self.cell_of.len().max(self.entries) as i64).max(64);
         if nx < 1 || nz < 1 || nx.saturating_mul(nz) > max_cells {
             return false;
         }
@@ -337,4 +356,46 @@ mod tests {
             rounds * n
         );
     }
+
+    /// A grid of part of a list's boxes that answers as the whole list's grid decides when to scan
+    /// every entry and when to look in columns: the part meets its boxes as the whole would.
+    #[test]
+    fn a_part_answers_as_the_whole_list() {
+        let mut seed = 0x9e37_79b9_7f4a_7c15u64;
+        let mut next = |range: f64| {
+            seed = seed.wrapping_mul(6_364_136_223_846_793_005).wrapping_add(1);
+            (seed >> 11) as f64 / (1u64 << 53) as f64 * range
+        };
+        let boxes: Vec<Aabb> = (0..400)
+            .map(|_| boxed(next(400.0) - 200.0, next(8.0), next(400.0) - 200.0, 0.6))
+            .collect();
+        let whole = SeenGrid::new(boxes.iter().copied());
+        assert_eq!(SeenGrid::bounds_of(boxes.iter().copied()), whole.bounds());
+        let part: Vec<usize> = (0..boxes.len()).step_by(4).collect();
+        let grid = SeenGrid::new(part.iter().map(|&i| boxes[i])).answering_as(whole.bounds());
+        let (mut a, mut b) = (Vec::new(), Vec::new());
+        let mut scanned = 0;
+        for _ in 0..500 {
+            let side = next(100.0);
+            let area = boxed(next(400.0) - 200.0, next(8.0), next(400.0) - 200.0, side);
+            let indexed = whole.candidates(area, &mut a);
+            assert_eq!(grid.candidates(area, &mut b), indexed, "{side} blocks");
+            scanned += usize::from(!indexed);
+            if indexed {
+                let met = |out: &[u32]| -> Vec<usize> {
+                    out.iter()
+                        .map(|&j| part[j as usize])
+                        .filter(|&i| boxes[i].intersects(area))
+                        .collect()
+                };
+                let expected: Vec<usize> = a
+                    .iter()
+                    .map(|&j| j as usize)
+                    .filter(|i| i % 4 == 0 && boxes[*i].intersects(area))
+                    .collect();
+                assert_eq!(met(&b), expected);
+            }
+        }
+        assert!(scanned > 0 && scanned < 500, "{scanned} of 500 scanned");
+    }
 }

crates/horizon-client/src/mobs/villager_brain.rs

@@ -928,12 +928,12 @@ fn chunk_pois(
         return;
     };
     let kind = |state: u32| poi_of_state(w, horizon_world::registry::BlockStateId(state));
-    let min = w.world.bottom_y();
-    for s in (y0 - min) >> 4..=(y1 - min) >> 4 {
-        let section = &chunk.sections[s as usize];
-        for &(i, poi) in w.world.pois.get((qx, s, qz), section, kind).iter() {
+    let (min, mut lists) = (w.world.bottom_y(), Vec::new());
+    w.world.pois.column(chunk, min, (y0, y1), kind, &mut lists);
+    for (base, pois) in &lists {
+        for &(i, poi) in pois.iter() {
             let i = i32::from(i);
-            let y = min + s * 16 + (i >> 8);
+            let y = base + (i >> 8);
             if (y0..=y1).contains(&y) {
                 visit(poi, [qx * 16 + (i & 15), y, qz * 16 + (i >> 4 & 15)]);
             }

crates/horizon-client/src/mobs/world.rs

@@ -604,13 +604,14 @@ impl MobWorld<'_> {
         }
     }
     pub fn seen(&self, id: u32) -> Option<&Seen> {
-        match self.others_index {
+        let own = match self.others_index {
             Some(index) => index
                 .get(&id)
                 .and_then(|&i| self.others.get(i as usize))
                 .filter(|s| s.id == id),
             None => self.others.iter().find(|s| s.id == id),
-        }
+        };
+        own.or_else(|| self.foreign?.seen(id))
     }
     /// Living mobs of the given species within `range` of `from`, nearest first.
     pub fn nearby<'s>(
@@ -773,6 +774,7 @@ pub(crate) fn with_resource_test_world(
         dragon: None,
         leash_boats: &[],
         who_lookup: None,
+        foreign: None,
         others_grid: None,
         others_index: None,
         others_villagers: None,

crates/horizon-client/src/mobs/zombie_villager.rs

@@ -261,6 +261,7 @@ mod tests {
             dragon: None,
             leash_boats: &[],
             who_lookup: None,
+            foreign: None,
             others_grid: None,
             others_index: None,
             others_villagers: None,

crates/horizon-client/src/parallel.rs

@@ -0,0 +1,146 @@
+//! The helper threads of the server tick's parallel phases (`HORIZON_PARALLEL_TICK`): one pool, made
+//! once, instead of threads started for every phase of every tick. A phase hands it work that gives
+//! the same result in whatever order the threads take it, and each part's results stay in order.
+use std::sync::atomic::{AtomicUsize, Ordering};
+use std::sync::{Mutex, OnceLock, PoisonError};
+
+/// `HORIZON_PARALLEL_TICK`: off when unset (0). `1` turns the server tick's parallel phases on (1),
+/// and `verify` does too (2), also checking the phases that can be checked against the serial code.
+/// Each phase gives the serial result either way.
+pub fn parallel_tick() -> u8 {
+    static ON: OnceLock<u8> = OnceLock::new();
+    *ON.get_or_init(|| match std::env::var("HORIZON_PARALLEL_TICK").as_deref() {
+        Ok("1" | "on" | "true") => 1,
+        Ok("verify") => 2,
+        _ => 0,
+    })
+}
+
+/// How many threads a phase may share its work among: `horizon-threads`' count (at most 8) while
+/// the parallel tick is on, else 1.
+pub fn lanes() -> usize {
+    if parallel_tick() > 0 {
+        horizon_threads::worker_count().clamp(1, 8)
+    } else {
+        1
+    }
+}
+
+/// The pool: one thread fewer than a phase's lanes (the server thread takes parts too).
+fn pool() -> &'static rayon::ThreadPool {
+    static POOL: OnceLock<rayon::ThreadPool> = OnceLock::new();
+    POOL.get_or_init(|| {
+        rayon::ThreadPoolBuilder::new()
+            .num_threads(
+                horizon_threads::worker_count()
+                    .clamp(1, 8)
+                    .saturating_sub(1)
+                    .max(1),
+            )
+            .thread_name(|i| format!("tick-helper-{i}"))
+            .build()
+            .expect("the tick's helper threads start")
+    })
+}
+
+/// `f` over `items` in parts on the pool, or all on this thread while the parallel tick is off or
+/// for fewer than `min` items: every part's results, in the order of `items`, as one thread would
+/// give them.
+pub fn par_map<T: Sync, R: Send>(
+    items: &[T],
+    min: usize,
+    f: impl Fn(&[T], &mut Vec<R>) + Sync,
+) -> Vec<R> {
+    par_map_on(items, min, lanes(), f)
+}
+
+/// [`par_map`] on `lanes` threads.
+///
+/// The parts are small (four a thread) and go to whichever thread is free, this one too: a helper
+/// the machine is slow to run (the chunk workers compete for the same cores) holds the phase up by
+/// one part at most instead of an eighth of the work, and the server thread works instead of only
+/// waiting.
+fn par_map_on<T: Sync, R: Send>(
+    items: &[T],
+    min: usize,
+    lanes: usize,
+    f: impl Fn(&[T], &mut Vec<R>) + Sync,
+) -> Vec<R> {
+    if lanes <= 1 || items.len() < min.max(2) {
+        let mut out = Vec::with_capacity(items.len());
+        f(items, &mut out);
+        return out;
+    }
+    let size = items.len().div_ceil(lanes * 4);
+    let count = items.len().div_ceil(size);
+    let next = AtomicUsize::new(0);
+    let slots: Vec<Mutex<Vec<R>>> = (0..count).map(|_| Mutex::new(Vec::new())).collect();
+    let work = || {
+        loop {
+            let k = next.fetch_add(1, Ordering::Relaxed);
+            if k >= count {
+                break;
+            }
+            let part = &items[k * size..((k + 1) * size).min(items.len())];
+            let mut done = Vec::with_capacity(part.len());
+            f(part, &mut done);
+            *slots[k].lock().unwrap_or_else(PoisonError::into_inner) = done;
+        }
+    };
+    pool().in_place_scope(|s| {
+        for _ in 1..lanes {
+            s.spawn(|_| work());
+        }
+        work();
+    });
+    let mut out = Vec::with_capacity(items.len());
+    for slot in slots {
+        out.extend(slot.into_inner().unwrap_or_else(PoisonError::into_inner));
+    }
+    out
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    /// On several threads, every item's result comes back once, in the items' order, for item
+    /// counts that don't divide into the parts evenly; a part may give several results or none.
+    #[test]
+    fn parts_come_back_in_order() {
+        for n in [0usize, 1, 7, 33, 1000, 4097] {
+            let items: Vec<u64> = (0..n as u64).collect();
+            let f = |part: &[u64], out: &mut Vec<u64>| {
+                for &x in part {
+                    // Spin a little so the threads overlap.
+                    let mut h = x;
+                    for _ in 0..200 {
+                        h = h.wrapping_mul(6364136223846793005).wrapping_add(1);
+                    }
+                    if x % 3 != 0 {
+                        out.push(x * 10);
+                    }
+                    if x % 5 == 0 {
+                        out.push(h);
+                    }
+                }
+            };
+            let mut serial = Vec::new();
+            f(&items, &mut serial);
+            assert_eq!(par_map_on(&items, 1, 4, f), serial, "{n} items");
+        }
+    }
+
+    /// A panic in a part reaches the caller instead of being lost on a helper.
+    #[test]
+    fn a_panic_reaches_the_caller() {
+        let items: Vec<u32> = (0..256).collect();
+        let caught = std::panic::catch_unwind(|| {
+            par_map_on(&items, 1, 4, |part, out: &mut Vec<u32>| {
+                assert!(!part.contains(&200), "item 200");
+                out.extend(part);
+            })
+        });
+        assert!(caught.is_err());
+    }
+}

crates/horizon-client/src/random_tick_sections.rs

@@ -117,6 +117,44 @@ impl RandomTickSections {
         });
         Some((y, mask.clone()))
     }
+    /// Builds, on the tick's helper threads (`crate::parallel`), the masks [`Self::next_masked`]
+    /// will build at the first draw in each ticking section of the chunks `keys` names, from their
+    /// blocks now (`chunk_of`), so the random tick loop finds them built. A mask only tells the loop
+    /// which drawn blocks to read: the draws and the ticks are the same either way, and an edit
+    /// before the loop reaches a section drops its mask as usual.
+    pub fn prepare_masks(
+        &mut self,
+        keys: &[(i32, i32)],
+        chunk_of: impl Fn((i32, i32)) -> Option<Arc<Chunk>>,
+        facts: &BlockFacts,
+        candidate: &(impl Fn(u32) -> bool + Sync),
+    ) {
+        let mut todo: Vec<((i32, i32), usize, Arc<Chunk>)> = Vec::new();
+        for &key in keys {
+            let Some(chunk) = chunk_of(key) else {
+                continue;
+            };
+            let cached = Self::refresh(&mut self.chunks, key, &chunk, &mut |id| {
+                facts.get(BlockStateId(id)).random_ticks
+            });
+            for i in cached.sections.mask.iter() {
+                if cached.masks.get(i).is_some_and(Option::is_none) {
+                    todo.push((key, i, Arc::clone(&chunk)));
+                }
+            }
+        }
+        let masks = crate::parallel::par_map(&todo, 128, |part, out| {
+            out.extend(
+                (part.iter())
+                    .map(|(_, i, chunk)| Self::candidate_mask(&chunk.sections[*i], candidate)),
+            );
+        });
+        for ((key, i, _), mask) in todo.into_iter().zip(masks) {
+            if let Some(cached) = self.chunks.get_mut(&key) {
+                cached.masks[i] = Some(Rc::new(mask));
+            }
+        }
+    }
     fn candidate_mask(section: &Section, mut candidate: impl FnMut(u32) -> bool) -> RandomTickMask {
         let blocks = &section.blocks;
         let mut ids = [0u16; 4096];
@@ -481,25 +519,41 @@ mod tick_parity {
             time: 0,
             speed: 3,
         };
+        // More than 128 ticking sections: prepare_masks must exercise its
+        // helper path when HORIZON_PARALLEL_TICK is enabled, not just defer work.
+        for x in 2..=65 {
+            original.chunks.insert((x, 0), chunk(x));
+        }
         Arc::make_mut(original.chunks.get_mut(&(1, 0)).unwrap()).lit = true; // precipitation-only chunk
         let mut cached = original.clone();
         // The masked path reads only the drawn blocks that can tick.
         let mut masked = original.clone();
+        // The masked path again, its masks built ahead of each tick (`prepare_masks`).
+        let mut prepared = original.clone();
+        let mut prep_cache = RandomTickSections::default();
+        let mut d = Ticks::new(12345);
         let mut cache = RandomTickSections::default();
         let mut mask_cache = RandomTickSections::default();
         let (mut a, mut b, mut c) = (Ticks::new(12345), Ticks::new(12345), Ticks::new(12345));
-        for x in -1..=1 {
+        for x in -1..=65 {
             a.add_empty_chunk(ChunkPos::new(x, 0)).unwrap();
             b.add_empty_chunk(ChunkPos::new(x, 0)).unwrap();
             c.add_empty_chunk(ChunkPos::new(x, 0)).unwrap();
+            d.add_empty_chunk(ChunkPos::new(x, 0)).unwrap();
         }
         for step in 0..128 {
-            for level in [&mut original, &mut cached, &mut masked] {
+            for level in [&mut original, &mut cached, &mut masked, &mut prepared] {
                 level.time = step;
                 level.speed = [0, 1, 3, 9][step as usize % 4];
                 level.reads.borrow_mut().clear();
                 level.events.clear();
                 match step {
+                    4 => {
+                        // The large fixture has exercised every tick speed and
+                        // helper-mask invalidation; keep the remaining edit,
+                        // unload and reload cases small as before.
+                        level.chunks.retain(|&(x, _), _| x <= 1);
+                    }
                     12 => {
                         level.chunks.remove(&(-1, 0));
                     }
@@ -528,15 +582,17 @@ mod tick_parity {
             if step == 12 {
                 cache.remove(&(-1, 0));
                 mask_cache.remove(&(-1, 0));
+                prep_cache.remove(&(-1, 0));
             }
             if step == 50 || step == 60 {
                 facts.states[grass.0 as usize].random_ticks = step == 60;
                 cache.clear();
                 mask_cache.clear();
+                prep_cache.clear();
             }
             // A write between ticks.
             if step == 20 {
-                for level in [&mut original, &mut cached, &mut masked] {
+                for level in [&mut original, &mut cached, &mut masked, &mut prepared] {
                     level.set_raw(BlockPos::new(0, MIN_Y + 16, 0), grass);
                 }
             }
@@ -589,6 +645,55 @@ mod tick_parity {
                 };
                 behaviours.tick_chunk_random_masked(&mut masked, &mut c, ChunkPos::new(x, z), next);
             }
+            let keys = lit(&prepared);
+            let candidate = |id| behaviours.random_tick_candidate(BlockStateId(id));
+            prep_cache.prepare_masks(
+                &keys,
+                |key| prepared.chunks.get(&key).cloned(),
+                &facts,
+                &candidate,
+            );
+            for &(x, z) in &keys {
+                // Every ticking section's mask is there before the tick asks.
+                let now = prepared.chunks[&(x, z)].clone();
+                let mut from = MIN_Y;
+                while let Some((y, _)) = prep_cache.next_masked(
+                    (x, z),
+                    &now,
+                    &facts,
+                    |_| unreachable!("built ahead"),
+                    from,
+                ) {
+                    from = y + 16;
+                }
+            }
+            for (x, z) in keys {
+                let next = |level: &mut Level, from| {
+                    let now = level.chunks[&(x, z)].clone();
+                    prep_cache
+                        .next_masked((x, z), &now, &facts, candidate, from)
+                        .map(|(y, mask)| (y, Some(mask)))
+                };
+                behaviours.tick_chunk_random_masked(
+                    &mut prepared,
+                    &mut d,
+                    ChunkPos::new(x, z),
+                    next,
+                );
+            }
+            assert_eq!(
+                original.events, prepared.events,
+                "prepared events at tick {step}"
+            );
+            assert_eq!(
+                a.random_tick_value, d.random_tick_value,
+                "prepared position RNG at tick {step}"
+            );
+            assert_eq!(
+                original.random.clone().next_long(),
+                prepared.random.next_long(),
+                "prepared level RNG at tick {step}"
+            );
             assert_eq!(original.reads, cached.reads, "ordered reads at tick {step}");
             assert_eq!(
                 original.events, masked.events,
@@ -629,6 +734,9 @@ mod tick_parity {
             for (s, t) in c.sections.iter().zip(&masked.chunks[key].sections) {
                 assert_eq!(s.blocks, t.blocks);
             }
+            for (s, t) in c.sections.iter().zip(&prepared.chunks[key].sections) {
+                assert_eq!(s.blocks, t.blocks);
+            }
         }
     }
 }

crates/horizon-core/src/random.rs

@@ -1,8 +1,10 @@
 //! Vanilla's two seeded random streams. No platform/default Rust RNG is used.
 mod hash;
+mod ordered;
 mod positional;
 mod worldgen;
 pub use hash::{seed_from_hash, seed_from_hash_utf16};
+pub use ordered::{LaneDone, OrderedDraws};
 pub use positional::{
     LegacyPositionalRandom, PositionalRandom, XoroshiroPositionalRandom, java_string_hash,
     position_seed,
@@ -104,10 +106,23 @@ pub trait RandomSource: Sized {
     }
 }
 
-#[derive(Clone, Debug)]
+#[derive(Debug)]
 pub struct LegacyRandom {
     state: u64,
     gaussian: Option<f64>,
+    /// A lane of an [`OrderedDraws`] stream: each item's first draw waits for its turn.
+    gate: Option<Box<ordered::Gate>>,
+}
+
+/// A copy draws on its own, from the state it was copied at: it is never gated.
+impl Clone for LegacyRandom {
+    fn clone(&self) -> Self {
+        Self {
+            state: self.state,
+            gaussian: self.gaussian,
+            gate: None,
+        }
+    }
 }
 
 impl LegacyRandom {
@@ -118,6 +133,7 @@ impl LegacyRandom {
         Self {
             state: (seed as u64 ^ Self::MULTIPLIER) & Self::MASK,
             gaussian: None,
+            gate: None,
         }
     }
 
@@ -126,6 +142,7 @@ impl LegacyRandom {
         Self {
             state,
             gaussian: None,
+            gate: None,
         }
     }
 
@@ -134,7 +151,18 @@ impl LegacyRandom {
         self.state
     }
 
+    /// A gated lane's first draw of an item waits for its turn and takes the shared state.
+    #[inline]
+    fn sync(&mut self) {
+        if let Some(gate) = self.gate.as_deref_mut()
+            && !gate.held()
+        {
+            (self.state, self.gaussian) = gate.take();
+        }
+    }
+
     fn bits(&mut self, count: u32) -> i32 {
+        self.sync();
         self.state = self.state.wrapping_mul(Self::MULTIPLIER).wrapping_add(11) & Self::MASK;
         let value = (self.state >> (48 - count)) as i32;
         #[cfg(feature = "rng-trace")]
@@ -149,7 +177,11 @@ impl RandomSource for LegacyRandom {
         LegacyPositionalRandom::new(self.next_long())
     }
     fn set_seed(&mut self, seed: i64) {
-        *self = Self::new(seed);
+        // LegacyRandomSource.setSeed resets the stream and its gaussian cache, not the lane.
+        // Reseeding is an ordered mutation even when this item makes no draws afterwards.
+        self.sync();
+        self.state = (seed as u64 ^ Self::MULTIPLIER) & Self::MASK;
+        self.gaussian = None;
     }
 
     fn next_int(&mut self) -> i32 {
@@ -190,6 +222,7 @@ impl RandomSource for LegacyRandom {
         value as f64 * (1.0 / 9_007_199_254_740_992.0)
     }
     fn next_gaussian(&mut self) -> f64 {
+        self.sync();
         let mut cache = self.gaussian.take();
         let value = marsaglia_polar(&mut cache, || self.next_double());
         self.gaussian = cache;
@@ -352,6 +385,24 @@ mod tests {
         invalid_bounds_do_not_consume_state(XoroshiroRandom::new(0));
     }
 
+    #[test]
+    fn legacy_reseed_resets_the_gaussian_cache_and_stream() {
+        let mut rng = LegacyRandom::new(42);
+        for seed in [i64::MIN, -1, 0, 42, i64::MAX] {
+            // Leave a cached second gaussian before reseeding.
+            rng.next_gaussian();
+            rng.set_seed(seed);
+            let mut fresh = LegacyRandom::new(seed);
+            for _ in 0..3 {
+                assert_eq!(
+                    rng.next_gaussian().to_bits(),
+                    fresh.next_gaussian().to_bits()
+                );
+                assert_eq!(rng.next_long(), fresh.next_long());
+            }
+        }
+    }
+
     #[test]
     fn zero_state_fallback_and_reseed() {
         let mut a = XoroshiroRandom::from_state(0, 0);

crates/horizon-core/src/random/ordered.rs

@@ -0,0 +1,401 @@
+//! One [`LegacyRandom`] stream shared by several threads that each run their part of a list in list
+//! order (the server's parallel mob tick, `game/mob_islands.rs`): every draw comes out exactly as if
+//! the whole list had run on one thread.
+//!
+//! Each thread (a lane) runs a subset of the list's items, in ascending item order, and says which
+//! item it is on ([`LegacyRandom::begin_item`]). The first draw or reseed waits until every other
+//! lane has passed that item, then takes the stream's state; the item's end gives it back
+//! ([`LegacyRandom::end_item`]). So the items before it have made all their draws and the items after
+//! it none. The lane on the smallest item never waits, so the lanes can't deadlock, as long as every
+//! lane has its own thread. An item that never draws or reseeds never waits.
+use super::LegacyRandom;
+use std::sync::{
+    Arc, Mutex, PoisonError,
+    atomic::{AtomicUsize, Ordering},
+};
+
+/// The shared stream and where each lane is.
+#[derive(Debug)]
+pub struct OrderedDraws {
+    /// The stream's state between holders: the generator state and the gaussian cache.
+    state: Mutex<(u64, Option<f64>)>,
+    /// The item each lane is on; `usize::MAX` once it has finished.
+    lanes: Box<[AtomicUsize]>,
+}
+
+/// A lane's hold on the shared stream, inside its [`LegacyRandom`].
+#[derive(Debug)]
+pub(super) struct Gate {
+    draws: Arc<OrderedDraws>,
+    lane: usize,
+    item: usize,
+    held: bool,
+}
+
+impl OrderedDraws {
+    /// `random`'s stream shared by `lanes` lanes, none of them started.
+    pub fn new(random: &LegacyRandom, lanes: usize) -> Arc<Self> {
+        Arc::new(Self {
+            state: Mutex::new((random.state, random.gaussian)),
+            lanes: (0..lanes).map(|_| AtomicUsize::new(0)).collect(),
+        })
+    }
+    /// The random lane `lane` draws from: its draws wait for their turn.
+    pub fn lane(self: &Arc<Self>, lane: usize) -> LegacyRandom {
+        let mut random = LegacyRandom::from_state(0);
+        random.gate = Some(Box::new(Gate {
+            draws: Arc::clone(self),
+            lane,
+            item: 0,
+            held: false,
+        }));
+        random
+    }
+    /// The stream as the lanes left it, once they have all finished.
+    pub fn finish(&self) -> LegacyRandom {
+        let (state, gaussian) = *self.state.lock().unwrap_or_else(PoisonError::into_inner);
+        let mut random = LegacyRandom::from_state(state);
+        random.gaussian = gaussian;
+        random
+    }
+    /// Wait until lane `lane` has finished item `item` (it is on a later one, or done).
+    pub fn wait_past(&self, lane: usize, item: usize) {
+        let mut spins = 0u32;
+        while self.lanes[lane].load(Ordering::Acquire) <= item {
+            spins += 1;
+            if spins < 64 {
+                std::hint::spin_loop();
+            } else {
+                std::thread::yield_now();
+            }
+        }
+    }
+    fn wait_for_turn(&self, lane: usize, item: usize) {
+        let mut spins = 0u32;
+        loop {
+            let ready = self
+                .lanes
+                .iter()
+                .enumerate()
+                .all(|(k, at)| k == lane || at.load(Ordering::Acquire) > item);
+            if ready {
+                return;
+            }
+            spins += 1;
+            if spins < 64 {
+                std::hint::spin_loop();
+            } else {
+                std::thread::yield_now();
+            }
+        }
+    }
+}
+
+impl Gate {
+    /// The shared state for this item's first draw, after the items before it.
+    pub(super) fn take(&mut self) -> (u64, Option<f64>) {
+        self.draws.wait_for_turn(self.lane, self.item);
+        self.held = true;
+        *self
+            .draws
+            .state
+            .lock()
+            .unwrap_or_else(PoisonError::into_inner)
+    }
+    pub(super) fn held(&self) -> bool {
+        self.held
+    }
+}
+
+impl LegacyRandom {
+    /// A gated random's lane starts item `item` (ascending within the lane). Nothing for an
+    /// ungated one.
+    pub fn begin_item(&mut self, item: usize) {
+        if let Some(gate) = self.gate.as_deref_mut() {
+            gate.item = item;
+            gate.draws.lanes[gate.lane].store(item, Ordering::Release);
+        }
+    }
+    /// The item's draws are done: the shared stream goes on from where it left it.
+    pub fn end_item(&mut self) {
+        if let Some(gate) = self.gate.as_deref_mut()
+            && gate.held
+        {
+            *gate
+                .draws
+                .state
+                .lock()
+                .unwrap_or_else(PoisonError::into_inner) = (self.state, self.gaussian);
+            gate.held = false;
+        }
+    }
+    /// The lane has run all its items: no other lane waits for it any more.
+    pub fn finish_lane(&mut self) {
+        self.end_item();
+        if let Some(gate) = self.gate.as_deref_mut() {
+            gate.draws.lanes[gate.lane].store(usize::MAX, Ordering::Release);
+        }
+    }
+    /// A gated random's [`LaneDone`], for its lane's thread to hold while it runs; None for an
+    /// ungated one.
+    pub fn lane_done(&self) -> Option<LaneDone> {
+        self.gate.as_deref().map(|gate| LaneDone {
+            draws: Arc::clone(&gate.draws),
+            lane: gate.lane,
+        })
+    }
+}
+
+/// Marks a lane finished when dropped, also while a panic unwinds its thread: the other lanes
+/// then run to their end instead of waiting for a lane that stopped, and the panic reaches the
+/// caller when it joins the threads.
+#[must_use = "the lane is marked finished when this is dropped"]
+pub struct LaneDone {
+    draws: Arc<OrderedDraws>,
+    lane: usize,
+}
+
+impl Drop for LaneDone {
+    fn drop(&mut self) {
+        self.draws.lanes[self.lane].store(usize::MAX, Ordering::Release);
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::random::RandomSource;
+
+    /// The timeout lives outside the scoped joins: a broken gate must fail, not hang the test suite.
+    fn bounded<T: Send + 'static>(run: impl FnOnce() -> T + Send + 'static) -> T {
+        let (sent, finished) = std::sync::mpsc::channel();
+        std::thread::spawn(move || {
+            let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(run));
+            let _ = sent.send(result);
+        });
+        match finished
+            .recv_timeout(std::time::Duration::from_secs(20))
+            .expect("ordered lanes must finish within the timeout")
+        {
+            Ok(value) => value,
+            Err(panic) => std::panic::resume_unwind(panic),
+        }
+    }
+
+    #[test]
+    fn reseeding_lanes_preserve_the_serial_stream() {
+        let mut initial = LegacyRandom::new(42);
+        initial.next_gaussian(); // A reseed before the first draw must also clear this cache.
+        // Independent serial expectation: replace an ungated generator, without calling set_seed.
+        let mut serial = LegacyRandom::new(11);
+        let mut expected = vec![vec![serial.next_gaussian().to_bits()]];
+        let before_reseed = serial.next_gaussian().to_bits();
+        serial = LegacyRandom::new(-7);
+        expected.push(vec![before_reseed, serial.next_gaussian().to_bits()]);
+        serial = LegacyRandom::new(i64::MIN);
+        expected.push(vec![]); // Reseed-only item must publish the state and clear the cache.
+        expected.push(vec![serial.next_gaussian().to_bits()]);
+        expected.push(vec![
+            serial.next_gaussian().to_bits(),
+            serial.next_int() as u64,
+        ]);
+        let before_reseed = serial.next_int() as u64;
+        serial = LegacyRandom::new(0);
+        expected.push(vec![
+            before_reseed,
+            serial.next_long() as u64,
+            serial.next_gaussian().to_bits(),
+        ]);
+
+        let (got, mut final_random) = bounded(move || {
+            let shared = OrderedDraws::new(&initial, 3);
+            let mut got: Vec<_> = std::thread::scope(|s| {
+                let handles: Vec<_> = (0..3)
+                    .map(|lane| {
+                        let shared = Arc::clone(&shared);
+                        s.spawn(move || {
+                            let mut random = shared.lane(lane);
+                            let mut out = Vec::new();
+                            // Intentionally no LaneDone: finish_lane itself must survive reseeding.
+                            for item in (lane..6).step_by(3) {
+                                random.begin_item(item);
+                                let values = match item {
+                                    0 => {
+                                        random.set_seed(11);
+                                        vec![random.next_gaussian().to_bits()]
+                                    }
+                                    1 => {
+                                        let before = random.next_gaussian().to_bits();
+                                        random.set_seed(-7);
+                                        vec![before, random.next_gaussian().to_bits()]
+                                    }
+                                    2 => {
+                                        random.set_seed(i64::MIN);
+                                        vec![]
+                                    }
+                                    3 => vec![random.next_gaussian().to_bits()],
+                                    4 => vec![
+                                        random.next_gaussian().to_bits(),
+                                        random.next_int() as u64,
+                                    ],
+                                    5 => {
+                                        let before = random.next_int() as u64;
+                                        random.set_seed(0);
+                                        vec![
+                                            before,
+                                            random.next_long() as u64,
+                                            random.next_gaussian().to_bits(),
+                                        ]
+                                    }
+                                    _ => unreachable!(),
+                                };
+                                random.end_item();
+                                out.push((item, values));
+                            }
+                            random.finish_lane();
+                            out
+                        })
+                    })
+                    .collect();
+                handles
+                    .into_iter()
+                    .flat_map(|h| h.join().unwrap())
+                    .collect()
+            });
+            got.sort_by_key(|(item, _)| *item);
+            (
+                got.into_iter().map(|(_, v)| v).collect::<Vec<_>>(),
+                shared.finish(),
+            )
+        });
+        assert_eq!(got, expected);
+        assert_eq!(final_random.state(), serial.state());
+        // finish() must keep the cached gaussian as well as the LCG state.
+        assert_eq!(
+            final_random.next_gaussian().to_bits(),
+            serial.next_gaussian().to_bits()
+        );
+        assert_eq!(final_random.next_long(), serial.next_long());
+    }
+
+    #[test]
+    fn a_reseeded_lane_keeps_its_panic_guard_and_published_stream() {
+        bounded(|| {
+            let shared = OrderedDraws::new(&LegacyRandom::new(42), 2);
+            std::thread::scope(|s| {
+                let panicking = Arc::clone(&shared);
+                let handle = s.spawn(move || {
+                    let mut random = panicking.lane(0);
+                    random.begin_item(0);
+                    random.set_seed(7);
+                    let _done = random.lane_done().expect("reseed must keep the lane guard");
+                    random.next_gaussian();
+                    random.end_item();
+                    panic!("after publishing the reseeded item");
+                });
+                let mut random = shared.lane(1);
+                random.begin_item(1);
+                let mut serial = LegacyRandom::new(7);
+                serial.next_gaussian();
+                assert_eq!(
+                    random.next_gaussian().to_bits(),
+                    serial.next_gaussian().to_bits()
+                );
+                random.finish_lane();
+                assert!(handle.join().is_err(), "the panic must reach the caller");
+                assert_eq!(shared.finish().next_long(), serial.next_long());
+            });
+        });
+    }
+
+    /// Items 0..40 over three lanes, each drawing on some items: the same values as one thread.
+    #[test]
+    fn lanes_draw_in_list_order() {
+        let draws_on = |item: usize| match item % 5 {
+            0 => 0,
+            1 => 1,
+            2 => 3,
+            _ => (item % 3) as i32,
+        };
+        let mut serial = LegacyRandom::new(42);
+        let expected: Vec<Vec<i32>> = (0..40)
+            .map(|i| {
+                (0..draws_on(i))
+                    .map(|_| serial.next_int_bound(1000).unwrap())
+                    .collect()
+            })
+            .collect();
+        let shared = OrderedDraws::new(&LegacyRandom::new(42), 3);
+        let got: Vec<(usize, Vec<i32>)> = std::thread::scope(|s| {
+            let handles: Vec<_> = (0..3)
+                .map(|lane| {
+                    let shared = Arc::clone(&shared);
+                    s.spawn(move || {
+                        let mut random = shared.lane(lane);
+                        let mut out = Vec::new();
+                        for item in (0..40).filter(|i| (i * 7 + 3) % 3 == lane) {
+                            random.begin_item(item);
+                            let values = (0..draws_on(item))
+                                .map(|_| random.next_int_bound(1000).unwrap())
+                                .collect();
+                            random.end_item();
+                            out.push((item, values));
+                        }
+                        random.finish_lane();
+                        out
+                    })
+                })
+                .collect();
+            handles
+                .into_iter()
+                .flat_map(|h| h.join().unwrap())
+                .collect()
+        });
+        let mut got = got;
+        got.sort_by_key(|(i, _)| *i);
+        assert_eq!(
+            got.into_iter().map(|(_, v)| v).collect::<Vec<_>>(),
+            expected
+        );
+        assert_eq!(shared.finish().state(), serial.state());
+    }
+
+    /// A lane that panics part way, holding its [`LaneDone`], doesn't hold up a lane whose later
+    /// items wait for it: that lane finishes, and the panic reaches the caller.
+    #[test]
+    fn a_panicking_lane_lets_the_others_finish() {
+        let (sent, finished) = std::sync::mpsc::channel();
+        std::thread::spawn(move || {
+            let shared = OrderedDraws::new(&LegacyRandom::new(7), 2);
+            let caught = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
+                std::thread::scope(|s| {
+                    let panicking = Arc::clone(&shared);
+                    s.spawn(move || {
+                        let mut random = panicking.lane(0);
+                        let _done = random.lane_done();
+                        for item in [0, 2, 4] {
+                            random.begin_item(item);
+                            random.next_int_bound(10).unwrap();
+                            assert_ne!(item, 2, "item 2");
+                            random.end_item();
+                        }
+                        random.finish_lane();
+                    });
+                    let mut random = shared.lane(1);
+                    let _done = random.lane_done();
+                    for item in [1, 3, 5] {
+                        random.begin_item(item);
+                        random.next_int_bound(10).unwrap();
+                        random.end_item();
+                    }
+                    random.finish_lane();
+                });
+            }));
+            let _ = sent.send(caught.is_err());
+        });
+        let panicked = finished
+            .recv_timeout(std::time::Duration::from_secs(20))
+            .expect("the lanes finished instead of waiting for the one that panicked");
+        assert!(panicked, "the panic reached the caller");
+    }
+}

tools/command_exec_corpus.py

@@ -6291,6 +6291,14 @@ def moving_piston_presentation_diagnostic():
     return moving_piston_presentation(cmd, setup)
 
 
+def piston_mob_bounce():
+    try:
+        from .piston_bounce_corpus import piston_bounce
+    except ImportError:
+        from piston_bounce_corpus import piston_bounce
+    return piston_bounce(cmd, setup)
+
+
 def piston_entity_tick_order():
     """#9337: entities travel before moving pistons push them, not afterward."""
     ops = []
@@ -7481,6 +7489,7 @@ def groups():
         ('wither spawn', wither_spawn()),
         ('riding teleport', riding_teleport()),
         ('piston entity tick order', piston_entity_tick_order()),
+        ('piston mob bounce', piston_mob_bounce()),
         ('primed tnt box', primed_tnt_box()),
         ('primed tnt fall', primed_tnt_fall()),
         ('command block last output', command_block_last_output()),

tools/piston_bounce_corpus.py

@@ -0,0 +1,64 @@
+"""Actual downward piston pushes onto slime, including the blocked move (#12091)."""
+
+
+def piston_bounce(cmd, setup):
+    ops = setup('/tick freeze', '/gamerule spawn_mobs false',
+                '/gamerule advance_time false', '/forceload add 0 0')
+    ops.append({'op': 'settle', 'required_chunks': [[0, 0]]})
+    ops += setup('/tp @s 1.5 200 1.5', '/gamemode creative @s',
+                 '/kill @e[type=!player]', '/fill 0 199 0 15 199 15 stone',
+                 '/fill 0 200 0 15 208 15 air',
+                 '/scoreboard objectives add piston_bounce dummy',
+                 '/execute store result score start piston_bounce run time query gametime')
+    cases = [('slime_block', 'slime_block', False),
+             ('slime_block', 'stone', False),
+             ('stone', 'slime_block', False),
+             ('slime_block', 'slime_block', True)]
+    for index, (floor, top, no_ai) in enumerate(cases):
+        x, z = 4 + index % 2 * 6, 4 + index // 2 * 6
+        selector = f'@e[tag=piston_bounce_{index},limit=1]'
+        for command in (f'/setblock {x} 200 {z} {floor}',
+                        f'/setblock {x} 204 {z} {top}',
+                        f'/setblock {x} 205 {z} piston[facing=down]',
+                        f'/summon wither_skeleton {x}.5 201 {z}.5 '
+                        '{PersistenceRequired:1b,Invulnerable:1b,Motion:[0d,0d,0d],'
+                        f'NoAI:{int(no_ai)}b,Tags:["piston_bounce_{index}"]' + '}'):
+            ops.append(cmd(command, expect_result=1))
+        ops += setup(f'/attribute {selector} movement_speed base set 0')
+        fields = (f'Pos:[{x}.5d,201d,{z}.5d],Motion:[0d,0d,0d],'
+                  'Health:20f' + (',NoAI:1b' if no_ai else ''))
+        ops.append(cmd(f'/execute if entity @e[tag=piston_bounce_{index},limit=1,type=wither_skeleton,nbt={{' + fields + '}]',
+                       expect_result=1))
+        ops.append(cmd(f'/execute unless entity @e[tag=piston_bounce_{index},nbt={{NoGravity:1b}}]',
+                       expect_result=1))
+        if not no_ai:
+            ops.append(cmd(f'/execute unless entity @e[tag=piston_bounce_{index},nbt={{NoAI:1b}}]',
+                           expect_result=1))
+    for tick in range(7):
+        if tick == 1:
+            for x, z in [(4, 4), (10, 4), (4, 10), (10, 10)]:
+                ops.append(cmd(f'/setblock {x + 1} 205 {z} redstone_block', expect_result=1))
+        if tick == 3:
+            # Remove the ceiling so the bounce becomes observable travel.
+            for x, z in [(4, 4), (10, 4), (4, 10), (10, 10)]:
+                ops += setup(f'/fill {x} 203 {z} {x} 205 {z} air')
+                for y in (203, 204, 205):
+                    ops.append(cmd(f'/execute if block {x} {y} {z} air', expect_result=1))
+        if tick:
+            ops += setup('/tick step 1')
+            ops.append({'op': 'tick', 'n': 1})
+        ops += setup('/execute store result score now piston_bounce run time query gametime',
+                     '/scoreboard players operation now piston_bounce -= start piston_bounce')
+        ops.append(cmd('/scoreboard players get now piston_bounce', expect_result=tick))
+        for index in range(len(cases)):
+            selector = f'@e[tag=piston_bounce_{index},limit=1]'
+            for field in ('Pos', 'Motion', 'OnGround'):
+                ops.append(cmd(f'/data get entity {selector} {field}'))
+            if tick in (1, 2):
+                x, z = 4 + index % 2 * 6, 4 + index // 2 * 6
+                ops.append(cmd(f'/execute if block {x} 205 {z} piston[facing=down,extended=true]',
+                               expect_result=1))
+                ops.append(cmd(f'/data get block {x} 204 {z} progress'))
+    return ops + setup('/kill @e[type=!player]', '/fill 0 200 0 15 208 15 air',
+                       '/scoreboard objectives remove piston_bounce', '/forceload remove 0 0',
+                       '/tick unfreeze', '/tp @s 8.5 230 8.5 0 0')