refactor(manipulation): cut over to single robot model

dimos/control/tasks/cartesian_ik_task/cartesian_ik_task.py

@@ -170,11 +170,9 @@ def __init__(self, name: str, config: CartesianIKTaskConfig) -> None:
         self._joint_names = frozenset(config.joint_names)
         self._joint_names_list = list(config.joint_names)
         self._num_joints = len(config.joint_names)
-        expected_joints = config.control_ik.robot_model.get_coordinator_joint_names()
-        if config.joint_names != expected_joints:
-            raise ValueError(
-                f"CartesianIKTask {name}: task joints must match RobotModelConfig coordinator joints"
-            )
+        model_joint_names = list(config.control_ik.robot_model.joint_names)
+        if list(config.joint_names) != model_joint_names:
+            raise ValueError(f"CartesianIKTask {name}: task joints must match model joints exactly")
 
         # Create IK solver from model
         self._ik: PinkControlIK = create_pink_control_ik(config.control_ik)

dimos/control/tasks/cartesian_ik_task/pink_control_ik.py

@@ -207,7 +207,7 @@ def _end_effector_link(robot: RobotModelConfig) -> str:
         if len(tip_links) != 1:
             raise ValueError(
                 f"Pink control requires exactly one pose-targetable planning group; "
-                f"robot '{robot.name}' has {len(tip_links)}"
+                f"model has {len(tip_links)}"
             )
         return tip_links[0]
 
@@ -216,23 +216,23 @@ def _build_mapping(
         model: pinocchio.Model,
         robot: RobotModelConfig,
     ) -> _CoordinateMapping:
-        joint_names = tuple(robot.get_coordinator_joint_names())
+        joint_names = tuple(robot.joint_names)
         if not joint_names or len(set(joint_names)) != len(joint_names):
             raise ValueError("control task joints must be unique and non-empty")
 
         q_indices: list[int] = []
         v_indices: list[int] = []
         q_widths: list[int] = []
         joint_ids: set[int] = set()
-        for urdf_name in (robot.get_urdf_joint_name(name) for name in joint_names):
-            if not model.existJointName(urdf_name):
-                raise ValueError(f"control joint mapping references unknown joint: {urdf_name}")
-            joint_id = int(model.getJointId(urdf_name))
+        for joint_name in joint_names:
+            if not model.existJointName(joint_name):
+                raise ValueError(f"control model references unknown joint: {joint_name}")
+            joint_id = int(model.getJointId(joint_name))
             if joint_id <= 0 or joint_id >= len(model.joints):
-                raise ValueError(f"invalid control joint index for {urdf_name}")
+                raise ValueError(f"invalid control joint index for {joint_name}")
             joint = model.joints[joint_id]
             if int(joint.nv) != 1 or int(joint.nq) not in (1, 2):
-                raise ValueError(f"control joint must be one-DoF: {urdf_name}")
+                raise ValueError(f"control joint must be one-DoF: {joint_name}")
             q_indices.append(int(joint.idx_q))
             v_indices.append(int(joint.idx_v))
             q_widths.append(int(joint.nq))

dimos/control/tasks/cartesian_ik_task/test_cartesian_ik_task.py

@@ -40,7 +40,6 @@
 
 def _robot(path: Path) -> RobotModelConfig:
     return RobotModelConfig(
-        name="tiny",
         model_path=path,
         base_pose=PoseStamped(position=[0, 0, 0], orientation=[0, 0, 0, 1]),
         joint_names=["joint1"],
@@ -135,6 +134,26 @@ def test_factory_rejects_invalid_default_pink_configuration() -> None:
         control_task_registry.create("cartesian_ik", config, hardware={})
 
 
+def test_cartesian_task_rejects_task_joints_that_do_not_match_model_order(
+    tmp_path: Path, mocker
+) -> None:
+    backend = _FakeControlIK()
+    mocker.patch(
+        "dimos.control.tasks.cartesian_ik_task.cartesian_ik_task.create_pink_control_ik",
+        return_value=backend,
+    )
+    model = _robot(tmp_path / "unused.urdf")
+
+    with pytest.raises(ValueError, match="task joints must match model joints exactly"):
+        CartesianIKTask(
+            "cartesian",
+            CartesianIKTaskConfig(
+                joint_names=["different_joint"],
+                control_ik=PinkControlIKConfig(robot_model=model),
+            ),
+        )
+
+
 @pytest.mark.parametrize(
     "module_name",
     [

dimos/control/tasks/cartesian_ik_task/test_pink_control_ik.py

@@ -84,7 +84,6 @@ def _robot(
     joint_names = joints or ["joint1", "joint2"]
     joint_count = len(joint_names)
     return RobotModelConfig(
-        name="tiny",
         model_path=path,
         base_pose=PoseStamped(position=[0, 0, 0], orientation=[0, 0, 0, 1]),
         joint_names=joint_names,
@@ -184,17 +183,15 @@ def prepare(
     }
 
 
-def test_pink_validates_named_frame_and_exact_joint_mapping(tmp_path: Path) -> None:
+def test_pink_validates_named_frame_and_exact_model_joints(tmp_path: Path) -> None:
     model_path = _write_urdf(tmp_path)
 
     with pytest.raises(ValueError, match="end-effector frame"):
         create_pink_control_ik(
             PinkControlIKConfig(robot_model=_robot(model_path, frame="missing")),
         )
 
-    mismatched = _robot(model_path).model_copy(
-        update={"joint_name_mapping": {"joint1": "missing", "joint2": "joint2"}}
-    )
+    mismatched = _robot(model_path).model_copy(update={"joint_names": ["missing", "joint2"]})
     with pytest.raises(ValueError, match="unknown joint"):
         create_pink_control_ik(
             PinkControlIKConfig(robot_model=mismatched),

dimos/control/tasks/eef_twist_task/test_eef_twist_task.py

@@ -81,7 +81,6 @@ def fake_ik(mocker) -> FakeIK:
 def _fake_robot_model() -> RobotModelConfig:
     local_joints = ["joint1", "joint2", "joint3"]
     return RobotModelConfig(
-        name="fake",
         model_path="fake.urdf",
         base_pose=PoseStamped(position=[0, 0, 0], orientation=[0, 0, 0, 1]),
         joint_names=local_joints,
@@ -93,10 +92,6 @@ def _fake_robot_model() -> RobotModelConfig:
                 tip_link="tool",
             )
         ],
-        joint_name_mapping={
-            f"arm/joint{index}": joint_name
-            for index, joint_name in enumerate(local_joints, start=1)
-        },
         home_joints=[0.0, 0.0, 0.0],
     )
 
@@ -106,7 +101,7 @@ def task(fake_ik: FakeIK) -> EEFTwistTask:
     return EEFTwistTask(
         "eef",
         EEFTwistTaskConfig(
-            joint_names=["arm/joint1", "arm/joint2", "arm/joint3"],
+            joint_names=["joint1", "joint2", "joint3"],
             control_ik=PinkControlIKConfig(robot_model=_fake_robot_model()),
             timeout=0.3,
             max_joint_delta_deg=15.0,
@@ -122,7 +117,7 @@ def _state(
     values = [0.0, 0.0, 0.0] if positions is None else positions
     return CoordinatorState(
         joints=JointStateSnapshot(
-            joint_positions={f"arm/joint{i + 1}": value for i, value in enumerate(values)},
+            joint_positions={f"joint{i + 1}": value for i, value in enumerate(values)},
         ),
         t_now=t_now,
         dt=dt,
@@ -143,7 +138,7 @@ def test_first_nonzero_command_activates_seeds_from_fk_and_outputs_servo_positio
 
     assert output is not None
     assert output.mode == ControlMode.SERVO_POSITION
-    assert output.joint_names == ["arm/joint1", "arm/joint2", "arm/joint3"]
+    assert output.joint_names == ["joint1", "joint2", "joint3"]
     assert output.positions == [0.01, 0.02, 0.03]
     assert fake_ik.solve_calls[0].translation[0] > 0.0
 
@@ -264,7 +259,7 @@ def test_preemption_discards_last_commanded_solve_seed(task: EEFTwistTask, fake_
     assert task.on_ee_twist_command(_twist(), t_now=1.0)
     assert task.compute(_state(1.01)) is not None
 
-    task.on_preempted("higher_priority", frozenset(["arm/joint1"]))
+    task.on_preempted("higher_priority", frozenset(["joint1"]))
     output = task.compute(_state(1.02, positions=[0.5, 0.0, 0.0]))
 
     assert output is not None
@@ -277,7 +272,7 @@ def gripper_task(fake_ik: FakeIK) -> EEFTwistTask:
     return EEFTwistTask(
         "eef",
         EEFTwistTaskConfig(
-            joint_names=["arm/joint1", "arm/joint2", "arm/joint3"],
+            joint_names=["joint1", "joint2", "joint3"],
             control_ik=PinkControlIKConfig(robot_model=_fake_robot_model()),
             timeout=0.0,
             max_joint_delta_deg=15.0,

dimos/control/tasks/teleop_task/test_teleop_task.py

@@ -72,7 +72,6 @@ def solve(
 
 def _robot(path: Path) -> RobotModelConfig:
     return RobotModelConfig(
-        name="arm",
         model_path=path,
         base_pose=PoseStamped(position=[0, 0, 0], orientation=[0, 0, 0, 1]),
         joint_names=["joint1", "joint2"],
@@ -84,7 +83,6 @@ def _robot(path: Path) -> RobotModelConfig:
                 tip_link="tool",
             )
         ],
-        joint_name_mapping={"arm/joint1": "joint1", "arm/joint2": "joint2"},
         home_joints=[0.0, 0.0],
     )
 
@@ -102,7 +100,7 @@ def _state(
     return CoordinatorState(
         joints=JointStateSnapshot(
             joint_positions={
-                f"arm/joint{index + 1}": position for index, position in enumerate(positions)
+                f"joint{index + 1}": position for index, position in enumerate(positions)
             }
         ),
         t_now=t_now,
@@ -146,7 +144,7 @@ def task(tmp_path: Path, fake_ik: _FakePinkIK) -> TeleopIKTask:
     return TeleopIKTask(
         "teleop_arm",
         TeleopIKTaskConfig(
-            joint_names=["arm/joint1", "arm/joint2"],
+            joint_names=["joint1", "joint2"],
             control_ik=_pink_config(tmp_path / "unused.urdf"),
             hand="right",
             min_dt=0.02,
@@ -161,7 +159,7 @@ def gripper_task(tmp_path: Path, fake_ik: _FakePinkIK) -> TeleopIKTask:
     return TeleopIKTask(
         "teleop_arm",
         TeleopIKTaskConfig(
-            joint_names=["arm/joint1", "arm/joint2"],
+            joint_names=["joint1", "joint2"],
             control_ik=_pink_config(tmp_path / "unused.urdf"),
             hand="right",
             gripper_joint="arm/gripper",
@@ -263,9 +261,9 @@ def test_gripper_claim_interpolation_and_hold_output(
 
     output = gripper_task.compute(_state(1.01, (0.4, 0.5)))
 
-    assert gripper_task.claim().joints == frozenset({"arm/joint1", "arm/joint2", "arm/gripper"})
+    assert gripper_task.claim().joints == frozenset({"joint1", "joint2", "arm/gripper"})
     assert output is not None
-    assert output.joint_names == ["arm/joint1", "arm/joint2", "arm/gripper"]
+    assert output.joint_names == ["joint1", "joint2", "arm/gripper"]
     assert output.positions == pytest.approx([0.4, 0.5, 0.6])
 
 
@@ -279,11 +277,11 @@ def test_pose_is_rejected_before_engage_and_after_release(task: TeleopIKTask) ->
     assert not task.on_cartesian_command(_delta(), 2.3)
 
 
-def test_factory_requires_pink_configuration_and_matching_model(tmp_path: Path) -> None:
+def test_factory_requires_pink_configuration_and_matching_model_dof(tmp_path: Path) -> None:
     legacy = TaskConfig(
         name="teleop",
         type="teleop_ik",
-        joint_names=["arm/joint1", "arm/joint2"],
+        joint_names=["joint1", "joint2"],
         params={"model_path": "legacy.xml", "ee_joint_id": 2, "hand": "right"},
     )
     with pytest.raises(ValueError, match="control_ik"):
@@ -292,11 +290,23 @@ def test_factory_requires_pink_configuration_and_matching_model(tmp_path: Path)
     mismatched = TaskConfig(
         name="teleop",
         type="teleop_ik",
-        joint_names=["wrong/joint1", "wrong/joint2"],
+        joint_names=["joint1"],
         params={
             "control_ik": {"robot_model": _robot(tmp_path / "unused.urdf")},
             "hand": "right",
         },
     )
-    with pytest.raises(ValueError, match="task joints must match"):
+    with pytest.raises(ValueError, match="task joints must match model joints exactly"):
         create_task(mismatched, {})
+
+    wrong_names = TaskConfig(
+        name="teleop",
+        type="teleop_ik",
+        joint_names=["other1", "other2"],
+        params={
+            "control_ik": {"robot_model": _robot(tmp_path / "unused.urdf")},
+            "hand": "right",
+        },
+    )
+    with pytest.raises(ValueError, match="task joints must match model joints exactly"):
+        create_task(wrong_names, {})