feat: replay blueprint - dimos --replay-db <memory.db> run replay

[spomichter]

Summary

dimos --replay-db <memory.db> run replay plays a --record recording back onto the bus as a blueprint, rosbag-play style: every recorded stream is an Out port of a ReplayModule, so lcmspy/dimos spy, Rerun, and any consumer see it exactly like a live robot.

dimos --replay-db recordings/<run-id>/memory.db run replay                        # all streams, recorded timing, viewer up
dimos --replay-db <file> run replay --replay.speed 2 --replay.loop true
dimos --replay-db <file> --viewer none run replay                                 # bus only
dimos --replay-db <file> run replay unitree-go2 --disable go2-connection          # feed a consumer blueprint

Design

• ReplayModule (dimos/memory/replay_module.py): start() opens the store, pins one replay anchor at the earliest recorded ts, and subscribes store.replay().stream(name).observable() to each Out. speed, loop, seek, duration are module config (--replay.speed etc.).
• replay_module(dataset) builds the subclass with one Out[type] per recorded stream (types from the store registry), so autoconnect can wire them; the blueprint reads global_config.replay_db at import (--replay-db must be a file path). ReplayModule.__init__ also creates any missing Out from its dataset config, so a worker that imported the port-less class (workers re-import blueprint modules before receiving the global config) ends up with the same ports. No env vars; lifecycle.py is untouched. Without a recording the class has no ports, so the blueprint registry never touches a database.
• replay blueprint (dimos/memory/blueprints.py) = vis_module + Replay. The Rerun layout is generated from the recorded types: one 2D view per Image stream, one 3D world view.
• Replay.pin_anchor(): without it the anchor pins on the first subscribed stream and streams that start earlier lose their first frames.
• --replay / ReplayConnection are untouched; that path fakes a robot from the shipped datasets, this one replays any recording generically.

Docs: to be added to docs/usage/recording.md once #3737 lands.

Test plan

• test_replay_module.py: ports and stream_types come from the recording, topic filter, bad pattern raises listing names, empty dataset -> no ports; a module instance publishes the recorded messages in order on its Outs
• pytest dimos/memory dimos/cli dimos/core/coordination + blueprint registry: passing; mypy clean
• dimos --replay-db recordings/<run-id>/memory.db run replay (Zenoh, Xvfb): bridge deploys, all 7 recorded streams back on the bus at recorded rates (lidar/odom/tf 1.7 Hz, color_image 1.9, maps 0.3)
• Rerun on Xvfb, screenshot mid-replay: generated layout shows color_image + 3D world (lidar, global map, costmap), 11 entities, 27.6 s timeline

[codecov]

❌ 2 Tests Failed:

Tests completed	Failed	Passed	Skipped
4866	2	4864	87

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

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

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

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

    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   = 4363168.594427279
interval   = 0.1
message    = 'Timeout waiting for topic /coordinator_joint_state#sensor_msgs.JointState'
predicate  = <function LcmSpy.wait_for_saved_topic.<locals>.condition at 0x7337bc196840>
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 0x7337bd6385c0>
start_blueprint = <function start_blueprint.<locals>.set_name_and_start at 0x7337bc1947c0>

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

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

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

    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   = 4363048.428148823
interval   = 0.1
message    = 'Timeout waiting for topic /coordinator_joint_state#sensor_msgs.JointState'
predicate  = <function LcmSpy.wait_for_saved_topic.<locals>.condition at 0x7337bc195f80>
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.}