refactor(manipulation): compose canonical pick and place

[ruthwikdasyam]

Summary

• Replace the legacy pick-and-place subclass with a capability-composed module that coordinates OSR, grasp generation, and manipulation execution.
• Add stable-ID scan, explicit grasp selection, verified pick, and explicit place workflow skills.
• Compose the canonical workflow into real and simulated xArm perception blueprints.

Stack

• Depends on feat(perception): add configurable OSR scan backends #3713 (ruthwik/feat/osr-owlvit-edgetam).
• This is the third PR in the heuristic grasp -> OSR -> canonical pick-and-place stack.

Validation

• TMPDIR=/dev/shm uv run --no-sync pytest dimos/manipulation/test_pick_and_place_unit.py dimos/perception/experimental/test_object_scene_registration_temporal.py dimos/robot/test_all_blueprints_generation.py -q
• Ruff format/check and mypy on all changed files.
• Imported real and simulated xArm perception blueprints.

Deliberately Deferred

• Public method-description polish, candidate filtering/reranking, visualization, obstacle suppression, and table/box features.

[codecov]

❌ 2 Tests Failed:

Tests completed	Failed	Passed	Skipped
4897	2	4895	82

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

dimos.e2e_tests.test_manipulation_planning_groups::test_dual_arm_plans_and_dispatches_both_arms_through_control_coordinator

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

Stack Traces | 120s run time

lcm_spy = <dimos.e2e_tests.lcm_spy.LcmSpy object at 0x7e4cb93b7260>
start_blueprint = <function start_blueprint.<locals>.set_name_and_start at 0x7e4c9a690860>

    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 through one trajectory task."""
>       _start_openarm_mock_planner(start_blueprint, lcm_spy)

lcm_spy    = <dimos.e2e_tests.lcm_spy.LcmSpy object at 0x7e4cb93b7260>
start_blueprint = <function start_blueprint.<locals>.set_name_and_start at 0x7e4c9a690860>

dimos/e2e_tests/test_manipulation_planning_groups.py:190: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
dimos/e2e_tests/test_manipulation_planning_groups.py:151: 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 0x7e4cb93b7260>
        start_blueprint = <function start_blueprint.<locals>.set_name_and_start at 0x7e4c9a690860>
dimos/e2e_tests/lcm_spy.py:97: in wait_for_saved_topic
    wait_until(
        condition  = <function LcmSpy.wait_for_saved_topic.<locals>.condition at 0x7e4c9a690e00>
        self       = <dimos.e2e_tests.lcm_spy.LcmSpy object at 0x7e4cb93b7260>
        timeout    = 120.0
        topic      = '/coordinator_joint_state#sensor_msgs.JointState'
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

predicate = <function LcmSpy.wait_for_saved_topic.<locals>.condition at 0x7e4c9a690e00>

    def wait_until(
        predicate: Callable[[], bool],
        *,
        timeout: float,
        interval: float = 0.1,
        message: str | None = None,
    ) -> None:
        """Poll ``predicate`` until it returns truthy or ``timeout`` elapses."""
        deadline = time.monotonic() + timeout
        while time.monotonic() < deadline:
            if predicate():
                return
            time.sleep(interval)
>       raise TimeoutError(message or f"Timed out after {timeout}s waiting for condition")
E       TimeoutError: Timeout waiting for topic /coordinator_joint_state#sensor_msgs.JointState

deadline   = 4424188.152184639
interval   = 0.1
message    = 'Timeout waiting for topic /coordinator_joint_state#sensor_msgs.JointState'
predicate  = <function LcmSpy.wait_for_saved_topic.<locals>.condition at 0x7e4c9a690e00>
timeout    = 120.0

.../utils/testing/waiting.py:35: TimeoutError

dimos.e2e_tests.test_manipulation_planning_groups::test_single_arm_plans_and_executes_through_control_coordinator

Flake rate in main: 25.00% (Passed 12 times, Failed 4 times)

Stack Traces | 120s run time

lcm_spy = <dimos.e2e_tests.lcm_spy.LcmSpy object at 0x7e4dc0060d40>
start_blueprint = <function start_blueprint.<locals>.set_name_and_start at 0x7e4c9a690400>

    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 0x7e4dc0060d40>
start_blueprint = <function start_blueprint.<locals>.set_name_and_start at 0x7e4c9a690400>

dimos/e2e_tests/test_manipulation_planning_groups.py:159: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
dimos/e2e_tests/test_manipulation_planning_groups.py:151: 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 0x7e4dc0060d40>
        start_blueprint = <function start_blueprint.<locals>.set_name_and_start at 0x7e4c9a690400>
dimos/e2e_tests/lcm_spy.py:97: in wait_for_saved_topic
    wait_until(
        condition  = <function LcmSpy.wait_for_saved_topic.<locals>.condition at 0x7e4c9a60bb00>
        self       = <dimos.e2e_tests.lcm_spy.LcmSpy object at 0x7e4dc0060d40>
        timeout    = 120.0
        topic      = '/coordinator_joint_state#sensor_msgs.JointState'
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

predicate = <function LcmSpy.wait_for_saved_topic.<locals>.condition at 0x7e4c9a60bb00>

    def wait_until(
        predicate: Callable[[], bool],
        *,
        timeout: float,
        interval: float = 0.1,
        message: str | None = None,
    ) -> None:
        """Poll ``predicate`` until it returns truthy or ``timeout`` elapses."""
        deadline = time.monotonic() + timeout
        while time.monotonic() < deadline:
            if predicate():
                return
            time.sleep(interval)
>       raise TimeoutError(message or f"Timed out after {timeout}s waiting for condition")
E       TimeoutError: Timeout waiting for topic /coordinator_joint_state#sensor_msgs.JointState

deadline   = 4424067.980965734
interval   = 0.1
message    = 'Timeout waiting for topic /coordinator_joint_state#sensor_msgs.JointState'
predicate  = <function LcmSpy.wait_for_saved_topic.<locals>.condition at 0x7e4c9a60bb00>
timeout    = 120.0

.../utils/testing/waiting.py:35: 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.}

[greptile-apps]

Greptile Summary

This change replaces the inherited pick-and-place implementation with composed scene registration, grasp generation, and manipulation services, and wires the new skills into real and simulated xArm blueprints. The held-object scan-and-place flow, blueprint imports, movement capability declarations, and targeted pick-and-place tests completed successfully. However, a failed plan for the first grasp proposal stops the operation instead of trying later ranked proposals that may be reachable.

Confidence Score: 4/5

Not ready to merge because a reachable object can be left unpicked when the highest-ranked grasp is not plannable.

The failure was reproduced with a realistic-fake workflow harness that exercised an unplannable first grasp and a plan-valid second grasp. The targeted unit suite passed, but it does not cover retrying ranked grasp proposals after a planning failure.

Files Needing Attention: dimos/manipulation/pick_and_place_module.py needs a bounded ranked-grasp retry path, with corresponding coverage in dimos/manipulation/test_pick_and_place_unit.py.

T-Rex Logs

What T-Rex did

• Ran the focused PR 3715 capability workflow harness to validate the posted P1 finding and collected its outputs across the base retry, composed harness, targeted unit tests, and xArm blueprint imports.
• Validated the contract by running the capability workflow harness and the unit-test suite, observing a planning failure with planned_candidate_x_values [0.1] and no selected rank, running pytest with 17 tests passing, and completing the real/sim xArm blueprint import successfully.
• Posted a second P1 finding-comment-proof to document the additional review finding.

_{Ran code and verified through T-Rex}

Comments Outside Diff (1)

1. General comment

   pick_object abandons ranked grasp fallback after the first planning failure

   • Bug
     • pick_object selects only candidates.candidates[0] at dimos/manipulation/pick_and_place_module.py:144 and returns immediately from the pregrasp motion failure at lines 155-158. The executed harness provided two ranked proposals: rank 0 was deliberately unplannable while rank 1 was plan-valid. The observed result was PLANNING_FAILED with planned candidate positions [0.1]; candidate rank 1 (x=0.2) was never planned. This regresses the base workflow’s bounded candidate loop and contradicts the new API’s “Generate ranked grasps” contract.
   • Cause
     • The PR replaced the previous bounded loop over candidate grasps with a single hard-coded candidates.candidates[0] selection, while retaining ranked-candidate terminology and GRASP_ATTEMPTS_EXHAUSTED in the manipulation error contract.
   • Fix
     • Iterate over the ranked candidates (with an explicit bounded attempt limit), attempt each candidate’s pregrasp/grasp plan until one completes, return the winning rank and score, and return GRASP_ATTEMPTS_EXHAUSTED only after all eligible candidates fail. Preserve safe gripper/state handling between attempts.

   _{Ran code and verified through T-Rex}

_{Reviews (1): Last reviewed commit: "chore(manipulation): use default grasp R..." | Re-trigger Greptile}

addresssed comments