refactor(manipulation): extract execution runtime

[TomCC7]

Summary

• extract coordinator trajectory lifecycle into an internal execution runtime
• migrate manipulation module and UI to runtime-backed lifecycle snapshots
• remove legacy dispatch, partial-robot selection, and freshness-gate paths

Validation

• uv run pytest dimos/manipulation -q (322 passed; 10 expected Drake skips)
• uv run ruff check dimos/manipulation
• uv run ruff format --check dimos/manipulation
• uv run --with mypy mypy dimos/manipulation
• openspec validate extract-manipulation-execution-runtime --strict

OpenSpec

• extract-manipulation-execution-runtime (44/44 tasks complete)

[codecov]

❌ 1 Tests Failed:

Tests completed	Failed	Passed	Skipped
2881	1	2880	41

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

dimos.manipulation.test_execution_runtime::test_stale_reset_deadline_does_not_retire_inflight_action

Stack Traces | 1.06s run time

def test_stale_reset_deadline_does_not_retire_inflight_action() -> None:
        gateway = BlockingGateway({"a": Outcome.ACCEPTED})
        entered = threading.Event()
        release = threading.Event()
        second_entered = threading.Event()
        cancel_calls = 0
    
        def cancel(_task: str) -> Outcome:
            nonlocal cancel_calls
            cancel_calls += 1
            if cancel_calls == 2:
                second_entered.set()
            entered.set()
            assert release.wait(2)
            return Outcome.CANCELLED
    
        gateway.cancel = cancel  # type: ignore[method-assign]
        runtime = ExecutionRuntime(lambda: gateway, action_timeout=0.05, poll_interval=10)
        try:
            runtime.execute_explicit(plan("a"))
            wait_until(lambda: runtime.snapshot().state == LifecycleState.RUNNING)
            operation = runtime.snapshot().operation
            assert operation is not None
            task = operation.tasks[0]
            faulted = replace(operation, tasks=(replace(task, activity=TaskActivity.ACTIVE),))
            runtime._submit(
                lambda: runtime._commit(
                    replace(runtime._context, state=LifecycleState.FAULT, active=faulted)
                )
            )
            first = runtime.reset()
            assert first.accepted and first.value is not None and entered.wait(1)
            old_action = runtime._context.active.tasks[0].action
            assert old_action is not None and old_action.reset_id == first.value.reset_id
            first_result = runtime.wait_for_reset(first.value, timeout=1)
            assert first_result.accepted and first_result.value is not None
            assert not first_result.value.success
    
            second = runtime.reset()
            assert second.accepted and second.value is not None
            runtime._submit(
                lambda: runtime._reduce(
                    _Event(
                        "action_deadline",
                        runtime._context.active.handle.operation_id,
                        runtime._context.active.handle.attempt_id,
                        task.task_id,
                        old_action.action_id,
                        ActionMethod.CANCEL,
                        reset_id=first.value.reset_id,
                    )
                )
            )
>           assert runtime._context.active.tasks[0].action is old_action
E           AssertionError: assert ActionRecord(action_id='3852c0b7-4256-403b-8dd4-47eb52c26bb9', method=<ActionMethod.CANCEL: 'cancel'>, started=255.500806251, deadline=255.550806251, deadline_reported=True, reset_id='333dd182-9ad2-4177-a97e-349f7ef28b9c') is ActionRecord(action_id='3852c0b7-4256-403b-8dd4-47eb52c26bb9', method=<ActionMethod.CANCEL: 'cancel'>, started=255.500806251, deadline=255.550806251, deadline_reported=False, reset_id='333dd182-9ad2-4177-a97e-349f7ef28b9c')
E            +  where ActionRecord(action_id='3852c0b7-4256-403b-8dd4-47eb52c26bb9', method=<ActionMethod.CANCEL: 'cancel'>, started=255.500806251, deadline=255.550806251, deadline_reported=True, reset_id='333dd182-9ad2-4177-a97e-349f7ef28b9c') = TaskRecord(task_id='8b6558b2-73c0-4267-9501-b37a7a1be02a', task_name='a', entry=TaskEntry(planning_group='a', task_name='a', request={'trajectory': 'a'}, robot_name=None), activity=<TaskActivity.UNKNOWN: 'unknown'>, action=ActionRecord(action_id='3852c0b7-4256-403b-8dd4-47eb52c26bb9', method=<ActionMethod.CANCEL: 'cancel'>, started=255.500806251, deadline=255.550806251, deadline_reported=True, reset_id='333dd182-9ad2-4177-a97e-349f7ef28b9c'), cancel_required=True, reset_required=False).action

cancel     = <function test_stale_reset_deadline_does_not_retire_inflight_action.<locals>.cancel at 0x7f2aa6d260c0>
cancel_calls = 1
entered    = <threading.Event at 0x7f2a9ca8a2d0: set>
faulted    = Operation(handle=OperationHandle(plan_id='409aa708-edb6-4c13-8e34-b2a5d9500fff', operation_id='f13bcdcc-c069-4f8f-8215..._index=0, cancel_requested=False, cleanup_required=False, uncertain=False, rejected=False, failed=False, diagnostic='')
first      = CommandResult(accepted=True, value=ResetHandle(reset_id='333dd182-9ad2-4177-a97e-349f7ef28b9c'), diagnostic='', snapsh... diagnostic=''), fault=None, diagnostic=None, shutdown=<ShutdownState.OPEN: 'open'>, shutdown_result=None, revision=8))
first_result = CommandResult(accepted=True, value=ResetResult(handle=ResetHandle(reset_id='333dd182-9ad2-4177-a97e-349f7ef28b9c'), su...stic='fault reconciliation action deadline', shutdown=<ShutdownState.OPEN: 'open'>, shutdown_result=None, revision=10))
gateway    = <dimos.manipulation.test_execution_runtime.BlockingGateway object at 0x7f2a9ca8aae0>
old_action = ActionRecord(action_id='3852c0b7-4256-403b-8dd4-47eb52c26bb9', method=<ActionMethod.CANCEL: 'cancel'>, started=255.500806251, deadline=255.550806251, deadline_reported=False, reset_id='333dd182-9ad2-4177-a97e-349f7ef28b9c')
operation  = Operation(handle=OperationHandle(plan_id='409aa708-edb6-4c13-8e34-b2a5d9500fff', operation_id='f13bcdcc-c069-4f8f-8215..._index=0, cancel_requested=False, cleanup_required=False, uncertain=False, rejected=False, failed=False, diagnostic='')
release    = <threading.Event at 0x7f2a9ca8bb30: set>
runtime    = <dimos.manipulation.execution_runtime.ExecutionRuntime object at 0x7f2a9ca8b020>
second     = CommandResult(accepted=True, value=ResetHandle(reset_id='4d867d08-c6c2-4304-a597-2c5b64a6055d'), diagnostic='', snapsh...stic='fault reconciliation action deadline', shutdown=<ShutdownState.OPEN: 'open'>, shutdown_result=None, revision=11))
second_entered = <threading.Event at 0x7f2a9ca8b5c0: unset>
task       = TaskRecord(task_id='8b6558b2-73c0-4267-9501-b37a7a1be02a', task_name='a', entry=TaskEntry(planning_group='a', task_nam..., robot_name=None), activity=<TaskActivity.ACTIVE: 'active'>, action=None, cancel_required=False, reset_required=False)

dimos/manipulation/test_execution_runtime.py:1214: AssertionError

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