WIP: feat: add manipulation planning groups

[TomCC7]

Summary

• add first-class manipulation planning group models, SRDF/fallback discovery, resolved joint naming, and world group resolution
• add group-scoped kinematics/planner APIs plus GeneratedPlan preview/execution projection through coordinator trajectory tasks
• add OpenArm mock planner/coordinator large E2E coverage for single-arm and dual-arm planning flows
• add OpenSpec artifacts and user/contributor docs for planning groups

Verification

• uv run pytest dimos/manipulation -q
• uv run pytest dimos/e2e_tests/test_manipulation_planning_groups.py -m self_hosted_large -q
• uv run pytest --collect-only dimos/e2e_tests/test_manipulation_planning_groups.py -q
• openspec validate add-planning-groups
• pre-commit hooks via git commit

Notes

• PR remains draft.
• OpenSpec manual QA tasks 8.12 and 8.13 remain unchecked because no SRDF-backed chain fixture or auxiliary pose-planning fixture was available in this environment.

[codecov]

❌ 3 Tests Failed:

Tests completed	Failed	Passed	Skipped
182	3	179	49

View the top 3 failed test(s) by shortest run time

dimos.manipulation.test_manipulation_module.TestManipulationModuleIntegration::test_ee_pose

Stack Traces | 0.464s run time

self = <dimos.manipulation.test_manipulation_module.TestManipulationModuleIntegration object at 0x7cb9f4631580>
module = <dimos.manipulation.manipulation_module.ManipulationModule object at 0x7cb9df3ca4e0>
joint_state_zeros = JointState(ts=1781763858.6393812, frame_id='', name=['arm/joint1', 'arm/joint2', 'arm/joint3', 'arm/joint4', 'arm/join....0, 0.0, 0.0, 0.0, 0.0, 0.0], velocity=[0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0], effort=[0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0])

    def test_ee_pose(self, module, joint_state_zeros):
        """Test getting end-effector pose."""
        module._on_joint_state(joint_state_zeros)
    
>       pose = module.get_ee_pose()

joint_state_zeros = JointState(ts=1781763858.6393812, frame_id='', name=['arm/joint1', 'arm/joint2', 'arm/joint3', 'arm/joint4', 'arm/join....0, 0.0, 0.0, 0.0, 0.0, 0.0], velocity=[0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0], effort=[0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0])
module     = <dimos.manipulation.manipulation_module.ManipulationModule object at 0x7cb9df3ca4e0>
self       = <dimos.manipulation.test_manipulation_module.TestManipulationModuleIntegration object at 0x7cb9f4631580>

dimos/manipulation/test_manipulation_module.py:200: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
dimos/manipulation/manipulation_module.py:434: in get_ee_pose
    return self._world_monitor.get_ee_pose(robot[1], joint_state=None)
        robot      = ('test_arm', 'robot_1', RobotModelConfig(rpc_transport=<class 'dimos.protocol.rpc.pubsubrpc.LCMRPC'>, default_rpc_time...nipulation.planning.trajectory_generator.joint_trajectory_generator.JointTrajectoryGenerator object at 0x7cb9d842f890>)
        robot_name = None
        self       = <dimos.manipulation.manipulation_module.ManipulationModule object at 0x7cb9df3ca4e0>
.../planning/monitor/world_monitor.py:364: in get_ee_pose
    self._unique_pose_group_id_for_robot(robot_id),
        joint_state = None
        robot_id   = 'robot_1'
        self       = <dimos.manipulation.planning.monitor.world_monitor.WorldMonitor object at 0x7cb9df3ca660>
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <dimos.manipulation.planning.monitor.world_monitor.WorldMonitor object at 0x7cb9df3ca660>
robot_id = 'robot_1'

    def _unique_pose_group_id_for_robot(self, robot_id: WorldRobotID) -> PlanningGroupID:
        robot_name = self._world.get_robot_config(robot_id).name
        pose_group_ids = [
            group.id
            for group in self._world.list_planning_groups()
            if group.robot_name == robot_name and group.has_pose_target
        ]
        if len(pose_group_ids) != 1:
>           raise ValueError(
                f"Robot '{robot_name}' has {len(pose_group_ids)} pose-targetable planning groups; "
                "call get_group_pose/get_group_jacobian with an explicit planning group ID"
            )
E           ValueError: Robot 'test_arm' has 0 pose-targetable planning groups; call get_group_pose/get_group_jacobian with an explicit planning group ID

pose_group_ids = []
robot_id   = 'robot_1'
robot_name = 'test_arm'
self       = <dimos.manipulation.planning.monitor.world_monitor.WorldMonitor object at 0x7cb9df3ca660>

.../planning/monitor/world_monitor.py:435: ValueError

dimos.e2e_tests.test_manipulation_planning_groups::test_dual_arm_plans_and_dispatches_both_arms_through_control_coordinator

Stack Traces | 123s run time

lcm_spy = <dimos.e2e_tests.lcm_spy.LcmSpy object at 0x77b4f99fe3c0>
start_blueprint = <function start_blueprint.<locals>.set_name_and_start at 0x77b4f7720720>

    def test_dual_arm_plans_and_dispatches_both_arms_through_control_coordinator(
        lcm_spy: LcmSpy,
        start_blueprint: Callable[..., DimosCliCall],
    ) -> None:
        """Plan one generated plan over both arms and dispatch both JTC tasks."""
>       _start_openarm_mock_planner(start_blueprint, lcm_spy)

lcm_spy    = <dimos.e2e_tests.lcm_spy.LcmSpy object at 0x77b4f99fe3c0>
start_blueprint = <function start_blueprint.<locals>.set_name_and_start at 0x77b4f7720720>

dimos/e2e_tests/test_manipulation_planning_groups.py:182: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
dimos/e2e_tests/test_manipulation_planning_groups.py:145: in _start_openarm_mock_planner
    lcm_spy.wait_for_saved_topic(JOINT_STATE_TOPIC, timeout=120.0)
        lcm_spy    = <dimos.e2e_tests.lcm_spy.LcmSpy object at 0x77b4f99fe3c0>
        start_blueprint = <function start_blueprint.<locals>.set_name_and_start at 0x77b4f7720720>
dimos/e2e_tests/lcm_spy.py:112: in wait_for_saved_topic
    self.wait_until(
        condition  = <function LcmSpy.wait_for_saved_topic.<locals>.condition at 0x77b4f7720b80>
        self       = <dimos.e2e_tests.lcm_spy.LcmSpy object at 0x77b4f99fe3c0>
        timeout    = 120.0
        topic      = '/coordinator/joint_state#sensor_msgs.JointState'
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <dimos.e2e_tests.lcm_spy.LcmSpy object at 0x77b4f99fe3c0>

    def wait_until(
        self,
        *,
        condition: Callable[[], bool],
        timeout: float,
        error_message: str,
        poll_interval: float = 0.1,
    ) -> None:
        start_time = time.time()
        while time.time() - start_time < timeout:
            if condition():
                return
            time.sleep(poll_interval)
>       raise TimeoutError(error_message)
E       TimeoutError: Timeout waiting for topic /coordinator/joint_state#sensor_msgs.JointState

condition  = <function LcmSpy.wait_for_saved_topic.<locals>.condition at 0x77b4f7720b80>
error_message = 'Timeout waiting for topic /coordinator/joint_state#sensor_msgs.JointState'
poll_interval = 0.1
self       = <dimos.e2e_tests.lcm_spy.LcmSpy object at 0x77b4f99fe3c0>
start_time = 1781764445.7750642
timeout    = 120.0

dimos/e2e_tests/lcm_spy.py:105: TimeoutError

dimos.e2e_tests.test_manipulation_planning_groups::test_single_arm_plans_and_executes_through_control_coordinator

Stack Traces | 123s run time

lcm_spy = <dimos.e2e_tests.lcm_spy.LcmSpy object at 0x77b4f99fbc80>
start_blueprint = <function start_blueprint.<locals>.set_name_and_start at 0x77b4f77ea5c0>

    def test_single_arm_plans_and_executes_through_control_coordinator(
        lcm_spy: LcmSpy,
        start_blueprint: Callable[..., DimosCliCall],
    ) -> None:
        """Plan with one arm and execute through its trajectory task."""
>       _start_openarm_mock_planner(start_blueprint, lcm_spy)

lcm_spy    = <dimos.e2e_tests.lcm_spy.LcmSpy object at 0x77b4f99fbc80>
start_blueprint = <function start_blueprint.<locals>.set_name_and_start at 0x77b4f77ea5c0>

dimos/e2e_tests/test_manipulation_planning_groups.py:153: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
dimos/e2e_tests/test_manipulation_planning_groups.py:145: in _start_openarm_mock_planner
    lcm_spy.wait_for_saved_topic(JOINT_STATE_TOPIC, timeout=120.0)
        lcm_spy    = <dimos.e2e_tests.lcm_spy.LcmSpy object at 0x77b4f99fbc80>
        start_blueprint = <function start_blueprint.<locals>.set_name_and_start at 0x77b4f77ea5c0>
dimos/e2e_tests/lcm_spy.py:112: in wait_for_saved_topic
    self.wait_until(
        condition  = <function LcmSpy.wait_for_saved_topic.<locals>.condition at 0x77b4f77eade0>
        self       = <dimos.e2e_tests.lcm_spy.LcmSpy object at 0x77b4f99fbc80>
        timeout    = 120.0
        topic      = '/coordinator/joint_state#sensor_msgs.JointState'
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <dimos.e2e_tests.lcm_spy.LcmSpy object at 0x77b4f99fbc80>

    def wait_until(
        self,
        *,
        condition: Callable[[], bool],
        timeout: float,
        error_message: str,
        poll_interval: float = 0.1,
    ) -> None:
        start_time = time.time()
        while time.time() - start_time < timeout:
            if condition():
                return
            time.sleep(poll_interval)
>       raise TimeoutError(error_message)
E       TimeoutError: Timeout waiting for topic /coordinator/joint_state#sensor_msgs.JointState

condition  = <function LcmSpy.wait_for_saved_topic.<locals>.condition at 0x77b4f77eade0>
error_message = 'Timeout waiting for topic /coordinator/joint_state#sensor_msgs.JointState'
poll_interval = 0.1
self       = <dimos.e2e_tests.lcm_spy.LcmSpy object at 0x77b4f99fbc80>
start_time = 1781764322.7665458
timeout    = 120.0

dimos/e2e_tests/lcm_spy.py:105: TimeoutError

To view more test analytics, go to the Test Analytics Dashboard
_{📋 Got 3 mins? Take this short survey to help us improve Test Analytics.}

[TomCC7]

plan_to_pose and plan_to_joints should be just wrapper of the more generic functions, no duplication