feat: integrate GraspGenX grasp proposals

[TomCC7]

Summary

• add ranked grasp proposal message contracts
• add the GraspGenX provider and runtime boundary
• add a standalone deterministic GraspGenX demo and focused tests

Stack

Stack 1 of 5. Base: main. Next: #3363.

Verification

• 38 focused tests passed
• Ruff passed on the changed Python files
• git diff --check passed

[codecov]

❌ 6 Tests Failed:

Tests completed	Failed	Passed	Skipped
3699	6	3693	175

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

dimos.manipulation.test_roboplan_world::test_native_planner_preserves_other_robot_and_auxiliary_joint_state

Stack Traces | 0.015s run time

fake_roboplan = None
robot_config = RobotModelConfig(rpc_transport=<class 'dimos.protocol.rpc.pubsubrpc.LCMRPC'>, default_rpc_timeout=120.0, rpc_timeouts=...ration=2.0, joint_name_mapping={}, gripper_hardware_id=None, tf_extra_links=[], home_joints=None, pre_grasp_offset=0.1)
mocker = <pytest_mock.plugin.MockerFixture object at 0xff7fa49798b0>

    def test_native_planner_preserves_other_robot_and_auxiliary_joint_state(
        fake_roboplan: None,
        robot_config: RobotModelConfig,
        mocker: MockerFixture,
    ) -> None:
        world, first_id, second_id, second_config = _make_two_robot_world(fake_roboplan, robot_config)
        world.sync_from_joint_state(
            second_id,
            JointState(name=["joint1", "joint2"], position=[0.3, 0.1]),
        )
        selection = _selection((robot_config, second_config), "arm/manipulator")
        observed_positions: dict[str, float] = {}
        native_plan = FakeRRT.plan
    
        def capture_scene_state(
            planner: FakeRRT,
            q_start: FakeJointConfiguration,
            q_goal: FakeJointConfiguration,
        ) -> FakeJointPath:
            observed_positions.update(
                zip(
                    planner.scene.native_joint_names,
                    planner.scene.current_positions,
                    strict=True,
                )
            )
            return native_plan(planner, q_start, q_goal)
    
        mocker.patch.object(FakeRRT, "plan", autospec=True, side_effect=capture_scene_state)
    
        result = world.plan_selected_joint_path(
            world,
            selection,
            JointState(name=list(selection.joint_names), position=[0.15, -0.1]),
            JointState(name=list(selection.joint_names), position=[0.2, 0.1]),
        )
    
>       assert result.status == PlanningStatus.SUCCESS
E       AssertionError: assert <PlanningStatus.INVALID_START: 4> == <PlanningStatus.SUCCESS: 1>
E        +  where <PlanningStatus.INVALID_START: 4> = PlanningResult(status=<PlanningStatus.INVALID_START: 4>, path=[], planning_time=0.0, path_length=0.0, iterations=0, message='Requested start state does not match current scene state', timestamps=None).status
E        +  and   <PlanningStatus.SUCCESS: 1> = PlanningStatus.SUCCESS

capture_scene_state = <function test_native_planner_preserves_other_robot_and_auxiliary_joint_state.<locals>.capture_scene_state at 0xff7fa4a56d40>
fake_roboplan = None
first_id   = 'robot_1'
mocker     = <pytest_mock.plugin.MockerFixture object at 0xff7fa49798b0>
native_plan = <function FakeRRT.plan at 0xff7fc5eb3100>
observed_positions = {}
result     = PlanningResult(status=<PlanningStatus.INVALID_START: 4>, path=[], planning_time=0.0, path_length=0.0, iterations=0, message='Requested start state does not match current scene state', timestamps=None)
robot_config = RobotModelConfig(rpc_transport=<class 'dimos.protocol.rpc.pubsubrpc.LCMRPC'>, default_rpc_timeout=120.0, rpc_timeouts=...ration=2.0, joint_name_mapping={}, gripper_hardware_id=None, tf_extra_links=[], home_joints=None, pre_grasp_offset=0.1)
second_config = RobotModelConfig(rpc_transport=<class 'dimos.protocol.rpc.pubsubrpc.LCMRPC'>, default_rpc_timeout=120.0, rpc_timeouts=...ration=2.0, joint_name_mapping={}, gripper_hardware_id=None, tf_extra_links=[], home_joints=None, pre_grasp_offset=0.1)
second_id  = 'robot_2'
selection  = PlanningGroupSelection(groups=(PlanningGroup(id='arm/manipulator', robot_name='arm', group_name='manipulator', joint_n...tcp', source='srdf'),), group_ids=('arm/manipulator',), joint_names=('arm/joint1', 'arm/joint2'), robot_names=('arm',))
world      = <dimos.manipulation.planning.world.roboplan_world.RoboPlanWorld object at 0xff7fa4979430>

dimos/manipulation/test_roboplan_world.py:1858: AssertionError

dimos.codebase_checks.test_no_dunder_new::test_no_dunder_new

Stack Traces | 1.92s run time

def test_no_dunder_new() -> None:
        """Fail if any test file calls `__new__` to bypass `__init__`."""
        dimos_dir = DIMOS_PROJECT_ROOT / "dimos"
        hits = find_dunder_new_calls()
        if hits:
            listing = "\n".join(
                f"  - {p.relative_to(dimos_dir)}:{lineno}: {line.strip()}" for p, lineno, line in hits
            )
>           raise AssertionError(
                f"Found __new__ call(s) in test files:\n{listing}\n\n"
                "Tests must construct objects with the real constructor: __init__ is "
                "code under test too, and an object assembled by hand silently rots "
                "when the constructor changes. If __init__ does heavy work, mock the "
                "collaborators it needs instead of skipping it. Only if that is truly "
                "impossible, add the call to the WHITELIST in "
                "dimos/codebase_checks/test_no_dunder_new.py."
            )
E           AssertionError: Found __new__ call(s) in test files:
E             - manipulation/test_table_collision.py:23: module = object.__new__(ManipulationModule)
E           
E           Tests must construct objects with the real constructor: __init__ is code under test too, and an object assembled by hand silently rots when the constructor changes. If __init__ does heavy work, mock the collaborators it needs instead of skipping it. Only if that is truly impossible, add the call to the WHITELIST in dimos/codebase_checks/test_no_dunder_new.py.

dimos_dir  = PosixPath('.../dimos/dimos/dimos')
hits       = [(PosixPath('.../dimos/dimos/dimos/manipulation/test_table_collision.py'), 23, '    module = object.__new__(ManipulationModule)')]
listing    = '  - manipulation/test_table_collision.py:23: module = object.__new__(ManipulationModule)'

dimos/codebase_checks/test_no_dunder_new.py:67: AssertionError

dimos.codebase_checks.test_inline_heavy_imports::test_heavy_imports_are_inline

Stack Traces | 2.48s run time

def test_heavy_imports_are_inline() -> None:
        """Fail if any file imports cv2/open3d/rerun at module level."""
        hits = find_eager_heavy_imports()
        if hits:
            listing = "\n".join(
                f"  - dimos/{f}:{line}: `{module}`"
                for f, lines in sorted(hits.items())
                for line, module in lines
            )
>           raise AssertionError(
                f"Found module-level import(s) of {'/'.join(HEAVY_MODULES)}:\n{listing}\n\n"
                "These libraries load large native extensions into every process that "
                "transitively imports the module. Import them inside the function or "
                "method that uses them; imports needed only for type annotations go "
                "under `if TYPE_CHECKING:`."
            )
E           AssertionError: Found module-level import(s) of cv2/open3d/rerun:
E             - .../manipulation/visualization/pose_overlay.py:17: `cv2`
E             - .../manipulation/visualization/rerun.py:20: `rerun.blueprint`
E           
E           These libraries load large native extensions into every process that transitively imports the module. Import them inside the function or method that uses them; imports needed only for type annotations go under `if TYPE_CHECKING:`.

hits       = {'manipulation/visualization/pose_overlay.py': [(17, 'cv2')], 'manipulation/visualization/rerun.py': [(20, 'rerun.blueprint')]}
listing    = '  - .../manipulation/visualization/pose_overlay.py:17: `cv2`\n  - .../manipulation/visualization/rerun.py:20: `rerun.blueprint`'

dimos/codebase_checks/test_inline_heavy_imports.py:92: AssertionError

dimos.codebase_checks.test_import_from_source::test_import_from_source

Stack Traces | 8.33s run time

def test_import_from_source() -> None:
        """Fail if any name is imported from a module that only re-imported it."""
        violations = find_reexport_imports()
        if violations:
            listing = "\n".join(
                f"  - {p.relative_to(DIMOS_PROJECT_ROOT)}:{line}: `{name}` imported from "
                f"{src}, but defined in {origin}"
                for p, line, name, src, origin in sorted(violations)
            )
>           raise AssertionError(
                f"Found import(s) that pull a name from a re-exporter:\n{listing}\n\n"
                "Import each name straight from the module that defines it (shown above). "
                "If a module re-exports a name on purpose, mark its import with "
                "`# noqa: F401` or the `from x import Y as Y` form, and that re-export "
                "will be allowed."
            )
E           AssertionError: Found import(s) that pull a name from a re-exporter:
E             - dimos/manipulation/test_picknplace.py:30: `IKStatus` imported from dimos.manipulation.planning.spec.models, but defined in dimos.manipulation.planning.spec.enums
E           
E           Import each name straight from the module that defines it (shown above). If a module re-exports a name on purpose, mark its import with `# noqa: F401` or the `from x import Y as Y` form, and that re-export will be allowed.

listing    = '  - dimos/manipulation/test_picknplace.py:30: `IKStatus` imported from dimos.manipulation.planning.spec.models, but defined in dimos.manipulation.planning.spec.enums'
violations = [(PosixPath('.../dimos/manipulation/test_picknplace.py'), 30, 'IKStatus', 'dimos.manipulation.planning.spec.models', 'dimos.manipulation.planning.spec.enums')]

dimos/codebase_checks/test_import_from_source.py:148: AssertionError

dimos.manipulation.test_picknplace::test_picknplace_blueprint_accepts_short_backend_and_grasp_options

Stack Traces | 15.7s run time

def test_picknplace_blueprint_accepts_short_backend_and_grasp_options() -> None:
        config = BlueprintConfigParser(picknplace)
    
>       options = config.parse(
            overrides={"osr": {"det": "moondream", "seg": "edgetam"}, "pnp": {"grasp": "graspgenx"}}
        )

config     = <dimos.core.coordination.blueprint_config.parser.BlueprintConfigParser object at 0xff80904284a0>

dimos/manipulation/test_picknplace.py:221: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
.../coordination/blueprint_config/parser.py:168: in parse
    key: plain(value)
        cli_tokens = ()
        config_path = None
        environ    = None
        global_overrides = None
        overrides  = {'osr': {'det': 'moondream', 'seg': 'edgetam'}, 'pnp': {'grasp': 'graspgenx'}}
        schema     = <[RuntimeError("Failed to pull LFS file .../dimos/data/.lfs/xarm_description.tar.gz after 3 attemp.../xarm_description.tar.gz']' returned non-zero exit status 1.") raised in repr()] ParserSchema object at 0xff8090453fc0>
        self       = <dimos.core.coordination.blueprint_config.parser.BlueprintConfigParser object at 0xff80904284a0>
.../coordination/blueprint_config/values.py:50: in plain
    return [plain(item) for item in value]
        value      = [<[RuntimeError("Failed to pull LFS file .../dimos/data/.lfs/xarm_description.tar.gz after 3 attem..._description.tar.gz']' returned non-zero exit status 1.") raised in repr()] RobotModelConfig object at 0xff7fc6945bd0>]
.../coordination/blueprint_config/values.py:46: in plain
    return {key: plain(item) for key, item in value.model_dump(mode="python").items()}
        value      = <[RuntimeError("Failed to pull LFS file .../dimos/data/.lfs/xarm_description.tar.gz after 3 attemp...m_description.tar.gz']' returned non-zero exit status 1.") raised in repr()] RobotModelConfig object at 0xff7fc6945bd0>
.../coordination/blueprint_config/values.py:45: in plain
    if isinstance(value, BaseModel):
        value      = <[RuntimeError("Failed to pull LFS file .../dimos/data/.lfs/xarm_description.tar.gz after 3 attemp.../.lfs/xarm_description.tar.gz']' returned non-zero exit status 1.") raised in repr()] LfsPath object at 0xff8016a235d0>
dimos/utils/data.py:364: in __getattribute__
    resolved = object.__getattribute__(self, "_ensure_downloaded")()
        name       = '__class__'
        self       = <[RuntimeError("Failed to pull LFS file .../dimos/data/.lfs/xarm_description.tar.gz after 3 attemp.../.lfs/xarm_description.tar.gz']' returned non-zero exit status 1.") raised in repr()] LfsPath object at 0xff8016a235d0>
dimos/utils/data.py:347: in _ensure_downloaded
    cache = get_data(filename)
        cache      = None
        filename   = 'xarm_description/urdf/xarm_device.urdf.xacro'
        self       = <[RuntimeError("Failed to pull LFS file .../dimos/data/.lfs/xarm_description.tar.gz after 3 attemp.../.lfs/xarm_description.tar.gz']' returned non-zero exit status 1.") raised in repr()] LfsPath object at 0xff8016a235d0>
dimos/utils/data.py:304: in get_data
    archive_path = _decompress_archive(_pull_lfs_archive(archive_name))
        archive_name = 'xarm_description'
        data_dir   = PosixPath('.../dimos/dimos/data')
        file_path  = PosixPath('.../dimos/dimos/data/xarm_description/urdf/xarm_device.urdf.xacro')
        name       = 'xarm_description/urdf/xarm_device.urdf.xacro'
        nested_path = PosixPath('urdf/xarm_device.urdf.xacro')
        path_parts = ('xarm_description', 'urdf', 'xarm_device.urdf.xacro')
dimos/utils/data.py:248: in _pull_lfs_archive
    _lfs_pull(file_path, repo_root)
        file_path  = PosixPath('.../dimos/data/.lfs/xarm_description.tar.gz')
        filename   = 'xarm_description'
        repo_root  = PosixPath('.../work/dimos/dimos')
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

file_path = PosixPath('.../dimos/data/.lfs/xarm_description.tar.gz')
repo_root = PosixPath('.../work/dimos/dimos')

    def _lfs_pull(file_path: Path, repo_root: Path, *, retries: int = 2) -> None:
        relative_path = file_path.relative_to(repo_root)
    
        env = os.environ.copy()
        env["GIT_LFS_FORCE_PROGRESS"] = "1"
    
        last_err: subprocess.CalledProcessError | None = None
        for attempt in range(1, retries + 2):  # retries + 1 total attempts
            try:
                subprocess.run(
                    ["git", "lfs", "pull", "--include", str(relative_path)],
                    cwd=repo_root,
                    check=True,
                    env=env,
                )
                return
            except subprocess.CalledProcessError as e:
                last_err = e
                if attempt <= retries:
                    time.sleep(attempt)  # 1s, 2s backoff
    
>       raise RuntimeError(
            f"Failed to pull LFS file {file_path} after {retries + 1} attempts: {last_err}"
        )
E       RuntimeError: Failed to pull LFS file .../dimos/data/.lfs/xarm_description.tar.gz after 3 attempts: Command '['git', 'lfs', 'pull', '--include', 'data/.lfs/xarm_description.tar.gz']' returned non-zero exit status 1.

attempt    = 3
env        = {'ACCEPT_EULA': 'Y', 'ACTIONS_ID_TOKEN_REQUEST_TOKEN': 'eyJhbGciOiJSUzI1NiIsImtpZCI6IjM4ODI2YjE3LTZhMzAtNWY5Yi1iMTY5LT...-version=2.0', 'ACTIONS_ORCHESTRATION_ID': '55dbf82f-b238-4e09-b461-0f80bf14bf88.tests.ubuntu-24_04-arm_3_14_fal', ...}
file_path  = PosixPath('.../dimos/data/.lfs/xarm_description.tar.gz')
last_err   = CalledProcessError(1, ['git', 'lfs', 'pull', '--include', 'data/.lfs/xarm_description.tar.gz'])
relative_path = PosixPath('data/.lfs/xarm_description.tar.gz')
repo_root  = PosixPath('.../work/dimos/dimos')
retries    = 2

dimos/utils/data.py:216: RuntimeError

View the full list of 1 ❄️ flaky test(s)

::dimos.manipulation.demo_grasp_visualization.test_demo

Flake rate in main: 100.00% (Passed 0 times, Failed 1 times)

Stack Traces | 0s run time

ImportError while importing test module '.../manipulation/demo_grasp_visualization/test_demo.py'.
Hint: make sure your test modules/packages have valid Python names.
Traceback:
.../visualization/viser/runtime.py:25: in <module>
    from viser import ViserServer as ViserServer
E   ModuleNotFoundError: No module named 'viser'

The above exception was the direct cause of the following exception:
.../manipulation/demo_grasp_visualization/test_demo.py:24: in <module>
    from dimos.manipulation.demo_grasp_visualization import __main__
.../manipulation/demo_grasp_visualization/__main__.py:20: in <module>
    from .demo import DEFAULT_MAX_CANDIDATES, run_contributor_demo
.../manipulation/demo_grasp_visualization/demo.py:43: in <module>
    from dimos.manipulation.visualization.viser.visualizer import ViserManipulationVisualizer
.../visualization/viser/visualizer.py:28: in <module>
    from dimos.manipulation.visualization.viser.gui import ViserPanelGui
.../visualization/viser/gui.py:32: in <module>
    from dimos.manipulation.visualization.viser.runtime import VISER_INSTALL_HINT
.../visualization/viser/runtime.py:29: in <module>
    raise ModuleNotFoundError(VISER_INSTALL_HINT) from e
E   ModuleNotFoundError: Viser manipulation visualization requires Viser with URDF support. Install it with: uv sync --extra manipulation

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

[greptile-apps]

Greptile Summary

Adds a GraspGenX grasp-proposal integration and demonstration workflow.

• Introduces ranked TCP-frame grasp candidate messages, configuration, runtime adapter, and model checkpoint loading.
• Adds a deterministic banana-scene demo with static grasp visualization and focused tests.
• Adds an optional GraspGenX dependency extra, pinned source revision, compatibility overrides, and lockfile updates.
• Changes the shared grasp-generation protocol from PoseArray-based generate_grasps to GraspCandidateArray-based propose_grasps.

Confidence Score: 4/5

The interface mismatch between GraspingModule and the new GraspGenSpec/GraspGenXModule must be fixed before merging because the existing grasp-generation skill cannot use the new provider.

The shared protocol now exposes a one-argument propose_grasps method returning GraspCandidateArray, but the orchestrator still invokes the removed two-argument generate_grasps method and consumes a PoseArray-shaped result.

Files Needing Attention: dimos/manipulation/grasping/grasp_gen_spec.py, dimos/manipulation/grasping/grasping.py

Important Files Changed

Filename	Overview
dimos/manipulation/grasping/grasp_gen_spec.py	Replaces the shared API without coordinating the existing GraspingModule call and result contract.
dimos/manipulation/grasping/grasp_gen_x.py	Adds the validated, ranked GraspGenX adapter, but its new API is not consumable by the existing orchestrator.
dimos/manipulation/grasping/grasp_gen_x_runtime.py	Adds lazy optional-runtime loading, pinned checkpoint retrieval, sweep-volume configuration, and tensor conversion.
dimos/msgs/manipulation_msgs/GraspCandidate.py	Adds a finite-score TCP grasp candidate message.
dimos/msgs/manipulation_msgs/GraspCandidateArray.py	Adds an ordered candidate collection carrying the source point-cloud header.
pyproject.toml	Adds the GraspGenX extra, pinned Git source, and global compatibility overrides.
dimos/manipulation/demo_graspgenx/demo.py	Adds one-shot inference, output validation, rendering orchestration, and reliable adapter cleanup.

_{Reviews (1): Last reviewed commit: "feat(manip): add standalone GraspGenX de..." | Re-trigger Greptile}

[greptile-apps]

Grasp provider contract is broken

When GraspingModule.generate_grasps uses the new provider, it calls the removed two-argument generate_grasps method even though the revised spec and GraspGenXModule expose only one-argument propose_grasps. This produces a missing-method/RPC failure; changing only the method name still leaves the extra argument and the caller's incompatible PoseArray.poses handling.

[mintlify]

Preview deployment for your docs. Learn more about Mintlify Previews.

Project	Status	Preview	Updated (UTC)
dimensional	🟢 Ready	View Preview	Aug 5, 2026, 5:32 AM

💡 Tip: Enable Workflows to automatically generate PRs for you.

[greptile-apps]

Too many files changed for review (104 files, 100 file limit).

Bypass the limit by tagging @greptile-apps to review.