feat(expert-program): validate parallel joint segments

embodichain/lab/gym/envs/expert_program/__init__.py

@@ -155,6 +155,10 @@
     create_simulation_expert_program_adapter,
 )
 from .simulation_handover import ConfiguredHandOverPoseProvider
+from .simulation_parallel_safety import (
+    CuroboParallelCommandSafetyValidator,
+    CuroboParallelSafetyValidatorFactory,
+)
 from .simulation_policies import (
     SimulationSegmentPolicyPort,
     default_simulation_settle_presets,
@@ -189,6 +193,8 @@
     "ControlPartResourceBinding",
     "ConfiguredHandOverPoseProvider",
     "CyclicPoseTargetCfg",
+    "CuroboParallelCommandSafetyValidator",
+    "CuroboParallelSafetyValidatorFactory",
     "DeclarativeCfgValue",
     "DemoBridgeError",
     "EXPERT_PROGRAM_SCHEMA_VERSION",

embodichain/lab/gym/envs/expert_program/catalog.py

@@ -1457,14 +1457,27 @@ def create_parallel_safety_validator(
         *,
         simulation: object,
         robot: object,
+        scene_registry: SceneRegistry,
+        engine: AtomicActionEngine,
     ) -> ParallelCommandSafetyValidator | None:
         """Create and strictly validate the registration-owned live safety gate."""
         self.assert_unchanged()
         factory = self.parallel_safety_factory
         if factory is None:
             return None
+        if type(scene_registry) is not SceneRegistry:
+            raise TypeError("scene_registry must be exactly SceneRegistry.")
+        if not isinstance(engine, AtomicActionEngine):
+            raise TypeError("engine must be an AtomicActionEngine.")
+        if engine.robot is not robot:
+            raise ValueError("engine and factory must reference the exact same robot.")
         with self._parallel_safety_validator_lock:
-            validator = factory.create(simulation=simulation, robot=robot)
+            validator = factory.create(
+                simulation=simulation,
+                robot=robot,
+                scene_registry=scene_registry,
+                engine=engine,
+            )
             if not isinstance(validator, ParallelCommandSafetyValidator):
                 raise TypeError(
                     "parallel_safety_factory.create() must return a "

embodichain/lab/gym/envs/expert_program/extensions.py

@@ -28,7 +28,7 @@
 from dataclasses import dataclass, fields, is_dataclass
 from enum import Enum
 from types import MappingProxyType
-from typing import ClassVar, Protocol, runtime_checkable
+from typing import ClassVar, Protocol, runtime_checkable, TYPE_CHECKING
 
 import torch
 
@@ -57,6 +57,10 @@
     RuntimeTransportActionEncoder,
 )
 
+if TYPE_CHECKING:
+    from embodichain.lab.sim.atomic_actions import AtomicActionEngine
+    from embodichain.lab.sim.skills import SceneRegistry
+
 VersionedKey = tuple[str, str]
 """Exact ``(provider_or_projector_id, revision)`` registry key."""
 
@@ -450,8 +454,10 @@ def create(
         *,
         simulation: object,
         robot: object,
+        scene_registry: SceneRegistry,
+        engine: AtomicActionEngine,
     ) -> ParallelCommandSafetyValidator:
-        """Create one live validator bound to the exact simulation and robot."""
+        """Create one live gate bound to the exact assembled runtime."""
 
 
 @dataclass(frozen=True, slots=True)

embodichain/lab/gym/envs/expert_program/simulation_environment.py

@@ -1060,6 +1060,8 @@ def create_parallel_command_safety_validator(
         validator = self._registration.create_parallel_safety_validator(
             simulation=self._simulation,
             robot=self._robot,
+            scene_registry=scene_registry,
+            engine=engine,
         )
         if not isinstance(validator, ParallelCommandSafetyValidator):
             raise TypeError(