@@ -214,6 +214,7 @@ def _refresh_collision_tool_geometry(
                     meshes = v.meshes
                     break
         mesh_root = Path(_mesh_dir) / "meshes"
+        role_counts: dict[str, int] = {}
         try:
             for spec in meshes:
                 path = mesh_root / spec.file
@@ -222,10 +223,14 @@ def _refresh_collision_tool_geometry(
                 # rotation branch here when a non-identity rpy appears.
                 T = np.eye(4, dtype=np.float64)
                 T[:3, 3] = spec.origin
-                # tool: namespace so colliding-pair reports speak the defined
-                # vocabulary (link names / shape: / install: / tool:) instead
-                # of raw mesh filenames.
-                geom_name = f"tool:{tool_key}:{spec.file}"
+                # tool: namespace with the mesh's semantic role so colliding-
+                # pair reports speak the defined vocabulary (link names /
+                # shape: / install: / tool:), never raw mesh filenames.
+                # Repeated roles (two jaws) get a positional suffix for the
+                # attach/detach bookkeeping's name uniqueness.
+                role = spec.role.name.lower()
+                n = role_counts[role] = role_counts.get(role, 0) + 1
+                geom_name = f"tool:{tool_key}:{role}" + (f"_{n}" if n > 1 else "")
                 collision.attach_mesh_to_frame(
                     geom_name,
                     str(path),
 
@@ -29,12 +29,18 @@ def collision_blocked(checker, current_q, target_q) -> bool:
     """Whether streaming toward ``target_q`` must stop.
 
     Approaching: blocked when ``target_q`` collides. Already colliding (a
-    keep-out placed over the arm, or a tool-attach overlap): blocked only when
-    the motion goes *deeper* — escaping is allowed, mirroring
-    :func:`guard_joint_path`'s start-in-collision rule.
+    keep-out placed over the arm, or a tool-attach overlap): blocked when the
+    motion contacts anything *new* or goes *deeper* — escaping is allowed,
+    mirroring :func:`guard_joint_path`'s start-in-collision rule (a global
+    min-distance trend alone can't tell an improving start-collision from a
+    new shallower one, so pairs are checked too). The pair sets allocate, but
+    only in this already-colliding branch — never in the steady-state path.
     """
     if checker.in_collision(current_q):
-        return bool(
+        new_pairs = set(checker.colliding_pairs(target_q)) - set(
+            checker.colliding_pairs(current_q)
+        )
+        return bool(new_pairs) or bool(
             checker.min_distance(target_q)
             < checker.min_distance(current_q) - _ESCAPE_TOL
         )
 
@@ -23,6 +23,12 @@ class SetShapesCommand(SystemCommand[SetShapesCmd]):
     dropped with) motion. The controller mirrors the applied shapes to the
     planner subprocess via ``sync_shapes`` after this executes.
 
+    A world change also invalidates committed motion that now violates it
+    (MoveIt-style): the segment player re-guards the streaming trajectory's
+    remaining waypoints on the version bump, and every trajectory again at
+    activation, halting with the collision error instead of driving into the
+    new keep-out.
+
     This is the codec boundary: the wire form is rebuilt into waldoctl
     ``Shape`` objects here (running their construction-time validation), and
     everything downstream — state, checkers, subprocess syncs, readback —
 
@@ -294,18 +294,20 @@ def gate_joint_enable_collision(checker, q_rad, joint_en, q_step, qlim=None) ->
     collides — self, tool, or keep-out shape. Proximity-gated (skip the
     per-direction checks when the arm is farther than ``_ENABLE_NEAR_M`` from
     collision) so it stays cheap. When the arm is ALREADY colliding (a keep-out
-    placed over it), escaping directions stay enabled — grey only those that go
-    deeper — mirroring the jog/planner escape semantics; otherwise every button
-    would grey with no way out. ``qlim`` is optional ``(low, high)`` rows: the
-    probe is clamped so a pose past the mechanical stop (which the jog itself
-    can never reach) can't grey the button. Not njit — it calls the C++ checker.
+    placed over it), escaping directions stay enabled — grey those that go
+    deeper or contact anything new — mirroring the jog/planner escape
+    semantics; otherwise every button would grey with no way out. ``qlim`` is
+    optional ``(low, high)`` rows: the probe is clamped so a pose past the
+    mechanical stop (which the jog itself can never reach) can't grey the
+    button. Not njit — it calls the C++ checker.
     """
     if checker is None:
         return
     md_now = checker.min_distance(q_rad)
     if md_now >= _ENABLE_NEAR_M:
         return
     inside = checker.in_collision(q_rad)
+    pairs_now = set(checker.colliding_pairs(q_rad)) if inside else None
     for j in range(6):
         for slot, sign in ((2 * j, 1.0), (2 * j + 1, -1.0)):
             if joint_en[slot]:
@@ -316,8 +318,10 @@ def gate_joint_enable_collision(checker, q_rad, joint_en, q_step, qlim=None) ->
                         q_step[j] = qlim[0][j]
                     elif q_step[j] > qlim[1][j]:
                         q_step[j] = qlim[1][j]
-                if inside:
-                    if checker.min_distance(q_step) < md_now - _ESCAPE_TOL:
+                if pairs_now is not None:
+                    if set(checker.colliding_pairs(q_step)) - pairs_now or (
+                        checker.min_distance(q_step) < md_now - _ESCAPE_TOL
+                    ):
                         joint_en[slot] = 0
                 elif checker.in_collision(q_step):
                     joint_en[slot] = 0
@@ -420,15 +424,18 @@ def _compute_cart_enable(
     near = md_now < _ENABLE_NEAR_M
     # Already colliding: keep escaping directions enabled (see the joint gate).
     inside = near and checker is not None and checker.in_collision(q_rad)
+    pairs_now = set(checker.colliding_pairs(q_rad)) if inside else None
 
     # Check IK (and, when near collision, the solved config) for each target
     for i in range(12):
         try:
             ik = solve_ik(robot, targets[i], q_rad, quiet_logging=True)
             ok = bool(ik.success)
             if ok and near and checker is not None:
-                if inside:
-                    ok = checker.min_distance(ik.q) >= md_now - _ESCAPE_TOL
+                if pairs_now is not None:
+                    ok = checker.min_distance(ik.q) >= md_now - _ESCAPE_TOL and not (
+                        set(checker.colliding_pairs(ik.q)) - pairs_now
+                    )
                 elif checker.in_collision(ik.q):
                     ok = False
             out[i] = 1 if ok else 0
 
@@ -16,10 +16,12 @@
 from collections import deque
 from typing import TYPE_CHECKING
 
+import numpy as np
 from pinokin import arrays_equal_n
 
+from parol6.commands._collision_guard import guard_joint_path
 from parol6.commands.base import CommandBase, ExecutionStatusCode
-from parol6.config import SETTLE_MAX_TICKS
+from parol6.config import COLLISION_PATH_SAMPLES, SETTLE_MAX_TICKS, steps_to_rad
 from parol6.protocol.wire import CommandCode
 from parol6.server.command_executor import _format_cmd_params
 from parol6.server.command_registry import create_command_from_struct
@@ -32,6 +34,7 @@
 )
 from parol6.utils.error_catalog import RobotError, make_error
 from parol6.utils.error_codes import ErrorCode
+from parol6.utils.errors import TrajectoryPlanningError
 from waldoctl import ActionState
 
 if TYPE_CHECKING:
@@ -57,6 +60,7 @@ class SegmentPlayer:
         "_inline_activated",
         "_settling",
         "_settle_ticks",
+        "_last_shapes_version",
     )
 
     def __init__(self, planner: MotionPlanner) -> None:
@@ -68,6 +72,7 @@ def __init__(self, planner: MotionPlanner) -> None:
         self._inline_activated: bool = False
         self._settling: bool = False
         self._settle_ticks: int = 0
+        self._last_shapes_version: int = 0
 
     @property
     def active(self) -> bool:
@@ -89,6 +94,18 @@ def tick(self, state: ControllerState) -> bool:
                 state.queued_duration += seg.duration
             seg = self._planner.poll_segment()
 
+        # MoveIt-style invalidation: a world change (SET_SHAPES bumps
+        # shapes_version) re-guards the streaming trajectory's remaining
+        # waypoints. The unchanged path pays only this int compare.
+        if state.shapes_version != self._last_shapes_version:
+            self._last_shapes_version = state.shapes_version
+            if isinstance(self._active, TrajectorySegment):
+                start = self._step - 1 if self._step > 0 else 0
+                if not self._world_guard(
+                    self._active, self._active.trajectory_steps[start:], state
+                ):
+                    return False
+
         # Process active segment or activate next
         max_immediate = 8  # prevent infinite recursion on back-to-back instant commands
         for _ in range(max_immediate):
@@ -97,6 +114,8 @@ def tick(self, state: ControllerState) -> bool:
                 if not self._buffer:
                     return False
                 self._activate_next(state)
+                if self._active is None:
+                    continue  # activation-time world guard rejected the segment
 
             active = self._active
 
@@ -162,8 +181,19 @@ def tick(self, state: ControllerState) -> bool:
         return self._active is not None
 
     def _activate_next(self, state: ControllerState) -> None:
-        """Promote next buffered segment to active."""
-        self._active = self._buffer.popleft()
+        """Promote next buffered segment to active.
+
+        Trajectory segments are re-guarded against the *current* world first:
+        their plan-time guard may predate a world change — the planner FIFO
+        orders plans and shape syncs by submission, so a segment planned
+        against the old world can arrive here after SET_SHAPES applied.
+        """
+        seg = self._buffer.popleft()
+        if isinstance(seg, TrajectorySegment) and not self._world_guard(
+            seg, seg.trajectory_steps, state
+        ):
+            return
+        self._active = seg
         self._step = 0
         self._inline_cmd = None
         self._inline_activated = False
@@ -245,6 +275,51 @@ def _on_failure(
         self._planner.cancel()
         self._drain_planner_queue(state)
 
+    def _world_guard(
+        self, seg: Segment, steps: np.ndarray, state: ControllerState
+    ) -> bool:
+        """Validate trajectory waypoints (motor steps) against the current
+        collision world; on violation, halt playback like an ErrorSegment.
+
+        Runs only at segment activation and on a world change, never per-tick;
+        rows are subsampled *before* the steps→rad conversion so the cost stays
+        ~COLLISION_PATH_SAMPLES checker calls regardless of trajectory length.
+        ``guard_joint_path`` keeps the escape semantics — a keep-out dropped
+        onto the arm still permits the escaping remainder. Inline segments
+        (Home/Gripper) are not trajectory playback and are not re-guarded.
+        """
+        n = len(steps)
+        if n == 0:
+            return True
+        target = max(2, COLLISION_PATH_SAMPLES + 2)
+        if n > target:
+            idx = np.unique(np.linspace(0, n - 1, target).round().astype(int))
+            steps = steps[idx]
+        q = np.empty((len(steps), 6), dtype=np.float64)
+        for i in range(len(steps)):
+            steps_to_rad(steps[i], q[i])
+        try:
+            guard_joint_path(q)
+        except TrajectoryPlanningError as exc:
+            logger.error(
+                "Command %d invalidated by world change: %s",
+                seg.command_index,
+                exc.robot_error,
+            )
+            state.error = exc.robot_error
+            pairs = exc.colliding_pairs
+            state.collision_active = bool(pairs)
+            state.collision_pairs = tuple(pairs) if pairs else ()
+            state.action_state = ActionState.ERROR
+            state.action_current = ""
+            state.action_params = ""
+            self._active = None
+            self._buffer.clear()
+            self._planner.cancel()
+            self._drain_planner_queue(state)
+            return False
+        return True
+
     def cancel(self, state: ControllerState) -> None:
         """Clear buffer, drain stale segments, and stop playback."""
         self._active = None