Extract manipulation plan execution manager by TomCC7 · Pull Request #3183 · dimensionalOS/dimos

TomCC7 · 2026-07-25T07:51:41Z

Summary

extract whole-plan validation, trajectory preparation, dispatch, replacement, rollback, and cancellation into PlanExecutionManager
preserve the existing manipulation execute/cancel RPCs and state projection without changing ControlCoordinator
reject partial execution of multi-robot plans and document acceptance, freshness, replacement, uncertainty, and coordinator ownership semantics
add focused race/failure tests and simulated single/dual-arm coordinator coverage

Verification

openspec validate extract-plan-execution-manager
focused manipulation suite: 66 passed
opted-in planning-group coordinator e2e suite: 2 passed
strict mypy on production/new execution files and e2e harness
Ruff check and format checks on touched Python files
bin/doclinks and bin/run-doc-codeblocks --ci on the updated manipulation docs
confirmed dimos/robot/all_blueprints.py is unchanged

OpenSpec

All 27 implementation and verification tasks are complete.

mintlify · 2026-07-25T07:52:44Z

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

Project	Status	Preview	Updated (UTC)
dimensional	🟢 Ready	View Preview	Jul 25, 2026, 7:52 AM

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

codecov · 2026-07-25T08:00:15Z

❌ 6 Tests Failed:

Tests completed	Failed	Passed	Skipped
3499	6	3493	71

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

dimos.manipulation.test_manipulation_module.TestCoordinatorIntegration::test_execute_rejected_by_coordinator

Stack Traces | 0.373s run time

self = <dimos.manipulation.test_manipulation_module.TestCoordinatorIntegration object at 0x749cdf659df0>
module = <dimos.manipulation.manipulation_module.ManipulationModule object at 0x749ccb179610>
joint_state_zeros = JointState(ts=1784991780.4664311, 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_execute_rejected_by_coordinator(self, module, joint_state_zeros):
        """Test handling of coordinator rejection."""
        module._on_joint_state(joint_state_zeros)
    
        module.plan_to_joints(JointState(position=[0.05] * 7))
    
        # Mock coordinator to reject
        mock_client = MagicMock()
        mock_client.task_invoke.return_value = False
        module._coordinator_client = mock_client
    
        result = module.execute()
    
        assert result is False
>       assert module._state == ManipulationState.FAULT
E       assert <ManipulationState.COMPLETED: 3> == <ManipulationState.FAULT: 4>
E        +  where <ManipulationState.COMPLETED: 3> = <dimos.manipulation.manipulation_module.ManipulationModule object at 0x749ccb179610>._state
E        +  and   <ManipulationState.FAULT: 4> = ManipulationState.FAULT

joint_state_zeros = JointState(ts=1784991780.4664311, 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])
mock_client = <MagicMock id='128216779546192'>
module     = <dimos.manipulation.manipulation_module.ManipulationModule object at 0x749ccb179610>
result     = False
self       = <dimos.manipulation.test_manipulation_module.TestCoordinatorIntegration object at 0x749cdf659df0>

dimos/manipulation/test_manipulation_module.py:302: AssertionError

dimos.manipulation.test_manipulation_module.TestManipulationModuleIntegration::test_trajectory_name_translation

Stack Traces | 0.449s run time

self = <dimos.manipulation.test_manipulation_module.TestManipulationModuleIntegration object at 0x749cdf659940>
module = <dimos.manipulation.manipulation_module.ManipulationModule object at 0x749ccb1758e0>
joint_state_zeros = JointState(ts=1784991774.5330343, 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_trajectory_name_translation(self, module, joint_state_zeros):
        """Test that trajectory joint names are translated for coordinator."""
        module._on_joint_state(joint_state_zeros)
    
        success = module.plan_to_joints(JointState(position=[0.05] * 7))
        assert success is True
    
        assert module._last_plan is not None
>       traj = module._split_plan_trajectory_by_robot(module._last_plan)["test_arm"]
E       AttributeError: 'ManipulationModule' object has no attribute '_split_plan_trajectory_by_robot'

joint_state_zeros = JointState(ts=1784991774.5330343, 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 0x749ccb1758e0>
self       = <dimos.manipulation.test_manipulation_module.TestManipulationModuleIntegration object at 0x749cdf659940>
success    = True

dimos/manipulation/test_manipulation_module.py:245: AttributeError

dimos.manipulation.test_manipulation_module.TestManipulationModuleIntegration::test_plan_to_joints

Stack Traces | 0.46s run time

self = <dimos.manipulation.test_manipulation_module.TestManipulationModuleIntegration object at 0x749cdf6589e0>
module = <dimos.manipulation.manipulation_module.ManipulationModule object at 0x749ccb11e810>
joint_state_zeros = JointState(ts=1784991771.8420339, 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_plan_to_joints(self, module, joint_state_zeros):
        """Test planning to a joint configuration."""
        module._on_joint_state(joint_state_zeros)
    
        target = JointState(position=[0.1, 0.1, 0.1, 0.1, 0.1, 0.1, 0.1])
        success = module.plan_to_joints(target)
    
        assert success is True
        assert module._state == ManipulationState.COMPLETED
        assert module.has_planned_path() is True
    
        assert module._last_plan is not None
>       traj = module._split_plan_trajectory_by_robot(module._last_plan)["test_arm"]
E       AttributeError: 'ManipulationModule' object has no attribute '_split_plan_trajectory_by_robot'

joint_state_zeros = JointState(ts=1784991771.8420339, 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 0x749ccb11e810>
self       = <dimos.manipulation.test_manipulation_module.TestManipulationModuleIntegration object at 0x749cdf6589e0>
success    = True
target     = JointState(ts=1784991771.844232, frame_id='', name=[], position=[0.1, 0.1, 0.1, 0.1, 0.1, 0.1, 0.1], velocity=[], effort=[])

dimos/manipulation/test_manipulation_module.py:173: AttributeError

dimos.manipulation.test_manipulation_module.TestManipulationModuleIntegration::test_plan_to_explicit_joint_target

Stack Traces | 0.463s run time

self = <dimos.manipulation.test_manipulation_module.TestManipulationModuleIntegration object at 0x749cdf658d40>
module = <dimos.manipulation.manipulation_module.ManipulationModule object at 0x749ccb175d60>
joint_state_zeros = JointState(ts=1784991772.7087033, 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_plan_to_explicit_joint_target(self, module, joint_state_zeros):
        """Test planning to an explicit planning-group joint target."""
        module._on_joint_state(joint_state_zeros)
    
        success = module.plan_to_joint_targets(
            {"test_arm/manipulator": JointState(position=[0.05] * 7)}
        )
    
        assert success is True
        assert module._state == ManipulationState.COMPLETED
        assert module._last_plan is not None
        assert module._last_plan.group_ids == ("test_arm/manipulator",)
        assert module.has_planned_path() is True
>       assert module._split_plan_trajectory_by_robot(module._last_plan) is not None
E       AttributeError: 'ManipulationModule' object has no attribute '_split_plan_trajectory_by_robot'

joint_state_zeros = JointState(ts=1784991772.7087033, 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 0x749ccb175d60>
self       = <dimos.manipulation.test_manipulation_module.TestManipulationModuleIntegration object at 0x749cdf658d40>
success    = True

dimos/manipulation/test_manipulation_module.py:191: AttributeError

dimos.manipulation.test_manipulation_module.TestCoordinatorIntegration::test_execute_with_mock_coordinator

Stack Traces | 5.4s run time

request = <SubRequest 'monitor_threads' for <Function test_execute_with_mock_coordinator>>

    @pytest.fixture(autouse=True)
    def monitor_threads(request):
        # Capture threads before test runs
        test_name = request.node.nodeid
        with _seen_threads_lock:
            _before_test_threads[test_name] = {
                t.ident for t in threading.enumerate() if t.ident is not None
            }
    
        yield
    
        with _seen_threads_lock:
            before = _before_test_threads.get(test_name, set())
    
        # Threads intentionally left running for the whole process and cleaned up on
        # exit, so they don't count as per-test leaks.
        expected_persistent_thread_prefixes = [
            "Dask-Offload",
            # HuggingFace safetensors conversion thread - no user cleanup API
            # https://github..../transformers/issues/29513
            "Thread-auto_conversion",
        ]
    
        def live_new_threads():
            # Threads created during this test that are still running. A thread that
            # has already stopped is not a leak -- it's done, just not yet reaped
            # from threading's registry -- so we key on is_alive(), not presence.
            result = []
            for t in threading.enumerate():
                if t.ident is None or t.ident in before or t.name == "MainThread":
                    continue
                if any(t.name.startswith(prefix) for prefix in expected_persistent_thread_prefixes):
                    continue
                if t.is_alive():
                    result.append(t)
            return result
    
        # Some C extensions tear their callback threads down asynchronously, so a
        # thread can stay alive briefly after the test cleaned up its owner (notably
        # zenoh's pyo3-closure threads, freed shortly after the session is closed).
        # A single snapshot races that teardown and flags a thread that is about to
        # exit. Give new threads a grace period to drain; only ones that stay alive
        # are real leaks (a genuinely leaked thread never exits).
        deadline = time.monotonic() + 5.0
        leaked = live_new_threads()
        while leaked and time.monotonic() < deadline:
            time.sleep(0.02)
            leaked = live_new_threads()
    
        if not leaked:
            return
    
        with _seen_threads_lock:
            # Report each leaked thread only once across the session.
            truly_new = [t for t in leaked if t.ident not in _seen_threads]
            for t in leaked:
                _seen_threads.add(t.ident)
    
        if not truly_new:
            return
    
        thread_names = [t.name for t in truly_new]
    
>       pytest.fail(
            f"Non-closed threads created during this test. Thread names: {thread_names}. "
            "Please look at the first test that fails and fix that."
        )
E       Failed: Non-closed threads created during this test. Thread names: ['Thread-107 (_lcm_loop)']. Please look at the first test that fails and fix that.

before     = {128216144537280, 128226488743616, 128226830968640}
deadline   = 1036036.315181099
expected_persistent_thread_prefixes = ['Dask-Offload', 'Thread-auto_conversion']
leaked     = [<Thread(Thread-107 (_lcm_loop), started daemon 128216756909760)>]
live_new_threads = <function monitor_threads.<locals>.live_new_threads at 0x749d15b3a3e0>
request    = <SubRequest 'monitor_threads' for <Function test_execute_with_mock_coordinator>>
t          = <Thread(Thread-107 (_lcm_loop), started daemon 128216756909760)>
test_name  = 'dimos/manipulation/test_manipulation_module.py::TestCoordinatorIntegration::test_execute_with_mock_coordinator'
thread_names = ['Thread-107 (_lcm_loop)']
truly_new  = [<Thread(Thread-107 (_lcm_loop), started daemon 128216756909760)>]

dimos/conftest.py:288: Failed

dimos.manipulation.test_manipulation_module.TestCoordinatorIntegration::test_state_transitions_during_execution

Stack Traces | 5.41s run time

request = <SubRequest 'monitor_threads' for <Function test_state_transitions_during_execution>>

    @pytest.fixture(autouse=True)
    def monitor_threads(request):
        # Capture threads before test runs
        test_name = request.node.nodeid
        with _seen_threads_lock:
            _before_test_threads[test_name] = {
                t.ident for t in threading.enumerate() if t.ident is not None
            }
    
        yield
    
        with _seen_threads_lock:
            before = _before_test_threads.get(test_name, set())
    
        # Threads intentionally left running for the whole process and cleaned up on
        # exit, so they don't count as per-test leaks.
        expected_persistent_thread_prefixes = [
            "Dask-Offload",
            # HuggingFace safetensors conversion thread - no user cleanup API
            # https://github..../transformers/issues/29513
            "Thread-auto_conversion",
        ]
    
        def live_new_threads():
            # Threads created during this test that are still running. A thread that
            # has already stopped is not a leak -- it's done, just not yet reaped
            # from threading's registry -- so we key on is_alive(), not presence.
            result = []
            for t in threading.enumerate():
                if t.ident is None or t.ident in before or t.name == "MainThread":
                    continue
                if any(t.name.startswith(prefix) for prefix in expected_persistent_thread_prefixes):
                    continue
                if t.is_alive():
                    result.append(t)
            return result
    
        # Some C extensions tear their callback threads down asynchronously, so a
        # thread can stay alive briefly after the test cleaned up its owner (notably
        # zenoh's pyo3-closure threads, freed shortly after the session is closed).
        # A single snapshot races that teardown and flags a thread that is about to
        # exit. Give new threads a grace period to drain; only ones that stay alive
        # are real leaks (a genuinely leaked thread never exits).
        deadline = time.monotonic() + 5.0
        leaked = live_new_threads()
        while leaked and time.monotonic() < deadline:
            time.sleep(0.02)
            leaked = live_new_threads()
    
        if not leaked:
            return
    
        with _seen_threads_lock:
            # Report each leaked thread only once across the session.
            truly_new = [t for t in leaked if t.ident not in _seen_threads]
            for t in leaked:
                _seen_threads.add(t.ident)
    
        if not truly_new:
            return
    
        thread_names = [t.name for t in truly_new]
    
>       pytest.fail(
            f"Non-closed threads created during this test. Thread names: {thread_names}. "
            "Please look at the first test that fails and fix that."
        )
E       Failed: Non-closed threads created during this test. Thread names: ['Thread-115 (_lcm_loop)']. Please look at the first test that fails and fix that.

before     = {128216136144576, 128216517834432, 128216756909760, 128226488743616, 128226830968640}
deadline   = 1036047.16050382
expected_persistent_thread_prefixes = ['Dask-Offload', 'Thread-auto_conversion']
leaked     = [<Thread(Thread-115 (_lcm_loop), started daemon 128215913854656)>]
live_new_threads = <function monitor_threads.<locals>.live_new_threads at 0x749ccb1b7560>
request    = <SubRequest 'monitor_threads' for <Function test_state_transitions_during_execution>>
t          = <Thread(Thread-115 (_lcm_loop), started daemon 128215913854656)>
test_name  = 'dimos/manipulation/test_manipulation_module.py::TestCoordinatorIntegration::test_state_transitions_during_execution'
thread_names = ['Thread-115 (_lcm_loop)']
truly_new  = [<Thread(Thread-115 (_lcm_loop), started daemon 128215913854656)>]

dimos/conftest.py:288: Failed

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

mintlify · 2026-07-25T08:55:06Z

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

Project	Status	Preview	Updated (UTC)
dimensional	🟡 Building	–	Jul 25, 2026, 7:51 AM

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

TomCC7 · 2026-07-25T22:33:57Z

+    UNKNOWN = auto()
+
+
+class ExecutionDispatchOutcome(Enum):


how is this different from coordinator dispatch outcome can we merge them

TomCC7 · 2026-07-25T22:34:44Z

+
+    plan_start_tolerance: float = 1e-6
+
+    def __post_init__(self) -> None:


use pydantic here

TomCC7 · 2026-07-25T22:46:50Z

+
+
+@dataclass(frozen=True)
+class ExecutionTarget:


discuss with me about whether using pydantic is good here. it's good for value validation for sure

TomCC7 · 2026-07-25T22:47:47Z

+        ...
+
+
+class CoordinatorExecutionAdapter:


maybe should put this into control coordinator's position? Since this looks like a generic client

TomCC7 · 2026-07-25T22:58:11Z

        # Coordinator integration (lazy initialized)
        self._coordinator_client: RPCClient | None = None
+        self._execution_manager: PlanExecutionManager | None = None
+        self._execution_manager_client: RPCClient | None = None


I guess we should not give execution manager and client a none type at all - none type is only because we need to initilaize in a later lifecycle. we should just declare here and just assign in later init, and there won't be any reference to this variable by the logic of whole system:

self._execution_manager: PlanExecutionManager

TomCC7 · 2026-07-25T22:59:14Z

        return True

    def _get_coordinator_client(self) -> RPCClient | None:
        """Get or create coordinator RPC client (lazy init)."""


let's just don't lazy init here. init normally as other classes in this module as we almost always need this

TomCC7 · 2026-07-25T23:13:29Z

-        # Translate joint names
-        coordinator_names = [
-            robot_config.get_coordinator_joint_name(j) for j in trajectory.joint_names
+    def _get_execution_manager(self) -> PlanExecutionManager | None:


again no lazy creation

TomCC7 · 2026-07-25T23:14:09Z

+        self._state_lock = threading.Lock()
+        self._dispatch_lock = threading.Lock()
+        self._cancel_lock = threading.Lock()


validate that all three locks are required here and can't be merged

TomCC7 · 2026-07-25T23:14:33Z

+                        self._possibly_active_tasks.add(dispatch.task_name)
+
+                    outcome = self._coordinator.execute(dispatch.task_name, dispatch.trajectory)
+                    if outcome is CoordinatorDispatchOutcome.ACCEPTED:


should use match

TomCC7 · 2026-07-25T23:14:52Z

+                self._cancellation_in_progress = True
+                self._generation += 1
+            try:
+                with self._dispatch_lock:


so dirty three locks are used together...

feat: extract manipulation plan execution manager

945f2c0

mintlify Bot deployed to staging - docs July 25, 2026 07:52 View deployment

TomCC7 force-pushed the cc/feat/manip-execution branch from d5c492a to 945f2c0 Compare July 25, 2026 07:53

mintlify Bot deployed to staging - docs July 25, 2026 07:54 View deployment

Merge remote-tracking branch 'origin/main' into cc/feat/manip-execution

fe81dc1

mintlify Bot deployed to staging - docs July 25, 2026 14:55 View deployment

TomCC7 commented Jul 25, 2026

View reviewed changes


		plan_start_tolerance: float = 1e-6

		def __post_init__(self) -> None:



		@dataclass(frozen=True)
		class ExecutionTarget:

Uh oh!

Conversation

TomCC7 commented Jul 25, 2026

Summary

Verification

OpenSpec

Uh oh!

mintlify Bot commented Jul 25, 2026

Uh oh!

codecov Bot commented Jul 25, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

❌ 6 Tests Failed:

Uh oh!

mintlify Bot commented Jul 25, 2026

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

1 participant

codecov Bot commented Jul 25, 2026 •

edited

Loading