feat: dimos spy - universal transport spy over LCM + Zenoh

[leshy]

replaces lcmspy

[mintlify]

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

Project	Status	Preview	Updated (UTC)
dimensional	🟢 Ready	View Preview	Jul 6, 2026, 4:36 PM

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

[greptile-apps]

Greptile Summary

This PR adds a universal transport spy to replace the old LCM-only spy. The main changes are:

• New dimos spy command for LCM and Zenoh traffic.
• lcmspy kept as an LCM-only alias.
• Strict parsing for spy transport filters and stray arguments.
• Best-effort default startup when one backend is unavailable.
• New tests and docs for the spy CLI behavior.

Confidence Score: 5/5

This looks safe to merge.

• No blocking issues found in the changed code.

Important Files Changed

Filename	Overview
dimos/robot/cli/dimos.py	Adds the spy command and routes lcmspy through the shared spy entry point.
dimos/utils/cli/spy/run_spy.py	Adds argument parsing, startup wiring, the Textual UI, and the LCM-only console shim.
dimos/utils/cli/spy/core.py	Adds transport sources, per-topic stats, default source discovery, and best-effort source startup.
pyproject.toml	Points the lcmspy console entry to the new spy shim.

_{Reviews (18): Last reviewed commit: "Merge branch 'main' into feat/ivan/spy" | Re-trigger Greptile}

[codecov]

❌ 1 Tests Failed:

Tests completed	Failed	Passed	Skipped
2722	1	2721	71

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

dimos.protocol.pubsub.test_spec::test_high_volume_messages[ros_context-topic1-values1]

Stack Traces | 4.59s run time

pubsub_context = <function ros_context at 0x7c356c32ad40>
topic = RawROSTopic(topic='/test_ros_topic', ros_type=<class 'geometry_msgs.msg._vector3.Vector3'>, qos=<rclpy.qos.QoSProfile object at 0x7c356c1f5cb0>)
values = [geometry_msgs.msg.Vector3(x=1.0, y=2.0, z=3.0), geometry_msgs.msg.Vector3(x=4.0, y=5.0, z=6.0), geometry_msgs.msg.Vector3(x=7.0, y=8.0, z=9.0)]

    @pytest.mark.self_hosted
    @pytest.mark.skipif_macos_bug
    @pytest.mark.parametrize("pubsub_context, topic, values", testdata)
    def test_high_volume_messages(
        pubsub_context: Callable[[], Any], topic: Any, values: list[Any]
    ) -> None:
        """Test that all 5k messages are received correctly.
        Limited to 5k because ros transport cannot handle more.
        Might want to have separate expectations per transport later
        """
        with pubsub_context() as x:
            # Create a list to capture received messages
            received_messages: list[Any] = []
            last_message_time = [time.time()]  # Use list to allow modification in callback
    
            # Define callback function
            def callback(message: Any, topic: Any) -> None:
                received_messages.append(message)
                last_message_time[0] = time.time()
    
            # Subscribe to the topic
            x.subscribe(topic, callback)
    
            # Publish 5000 messages
            num_messages = 5000
            for _ in range(num_messages):
                x.publish(topic, values[0])
    
            # Wait until no messages received for 0.5 seconds
            timeout = 2.0  # Maximum time to wait
            stable_duration = 0.1  # Time without new messages to consider done
            start_time = time.time()
    
            while time.time() - start_time < timeout:
                if time.time() - last_message_time[0] >= stable_duration:
                    break
                time.sleep(0.1)
    
            # Capture count and clear list to avoid printing huge list on failure
            received_len = len(received_messages)
            received_messages.clear()
>           assert received_len == num_messages, f"Expected {num_messages} messages, got {received_len}"
E           AssertionError: Expected 5000 messages, got 2900
E           assert 2900 == 5000

_          = 4999
callback   = <function test_high_volume_messages.<locals>.callback at 0x7c356b64e840>
last_message_time = [1783453930.6014397]
num_messages = 5000
pubsub_context = <function ros_context at 0x7c356c32ad40>
received_len = 2900
received_messages = [geometry_msgs.msg.Vector3(x=1.0, y=2.0, z=3.0), geometry_msgs.msg.Vector3(x=1.0, y=2.0, z=3.0), geometry_msgs.msg.Vec....0, y=2.0, z=3.0), geometry_msgs.msg.Vector3(x=1.0, y=2.0, z=3.0), geometry_msgs.msg.Vector3(x=1.0, y=2.0, z=3.0), ...]
stable_duration = 0.1
start_time = 1783453928.4367003
timeout    = 2.0
topic      = RawROSTopic(topic='/test_ros_topic', ros_type=<class 'geometry_msgs.msg._vector3.Vector3'>, qos=<rclpy.qos.QoSProfile object at 0x7c356c1f5cb0>)
values     = [geometry_msgs.msg.Vector3(x=1.0, y=2.0, z=3.0), geometry_msgs.msg.Vector3(x=4.0, y=5.0, z=6.0), geometry_msgs.msg.Vector3(x=7.0, y=8.0, z=9.0)]
x          = <dimos.protocol.pubsub.impl.rospubsub.RawROS object at 0x7c35202fdcd0>

.../protocol/pubsub/test_spec.py:365: AssertionError

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

dimos.e2e_tests.test_dimsim_spatial_memory::test_go_to_the_bed

Flake rate in main: 20.35% (Passed 90 times, Failed 23 times)

Stack Traces | 559s run time

lcm_spy = <dimos.e2e_tests.lcm_spy.LcmSpy object at 0x7c5c260ecad0>
start_blueprint = <function start_blueprint.<locals>.set_name_and_start at 0x7c5c268ee520>
human_input = <function human_input.<locals>.send_human_input at 0x7c5c268ee5c0>
dim_sim = <dimos.e2e_tests.dim_sim_client.DimSimClient object at 0x7c5c2605ac60>
explore_house = <function explore_house.<locals>.explore at 0x7c5c268eeac0>

    @pytest.mark.self_hosted_large
    def test_go_to_the_bed(lcm_spy, start_blueprint, human_input, dim_sim, explore_house) -> None:
        start_blueprint(
            "run",
            "unitree-go2-agentic",
            simulator="dimsim",
        )
        lcm_spy.save_topic(".../McpClient/on_system_modules/res")
        lcm_spy.wait_for_saved_topic(".../McpClient/on_system_modules/res", timeout=1200.0)
    
        explore_house()
    
        human_input("go to the bed")
    
>       lcm_spy.wait_until_odom_position(-3.567, -1.332, threshold=2, timeout=180)

dim_sim    = <dimos.e2e_tests.dim_sim_client.DimSimClient object at 0x7c5c2605ac60>
explore_house = <function explore_house.<locals>.explore at 0x7c5c268eeac0>
human_input = <function human_input.<locals>.send_human_input at 0x7c5c268ee5c0>
lcm_spy    = <dimos.e2e_tests.lcm_spy.LcmSpy object at 0x7c5c260ecad0>
start_blueprint = <function start_blueprint.<locals>.set_name_and_start at 0x7c5c268ee520>

dimos/e2e_tests/test_dimsim_spatial_memory.py:32: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
dimos/e2e_tests/lcm_spy.py:167: in wait_until_odom_position
    self.wait_for_message_result(
        predicate  = <function LcmSpy.wait_until_odom_position.<locals>.predicate at 0x7c5c268eeca0>
        self       = <dimos.e2e_tests.lcm_spy.LcmSpy object at 0x7c5c260ecad0>
        threshold  = 2
        timeout    = 180
        x          = -3.567
        y          = -1.332
dimos/e2e_tests/lcm_spy.py:153: in wait_for_message_result
    wait_until(
        event      = <threading.Event at 0x7c5c2605b770: unset>
        fail_message = 'Failed to get to position x=-3.567, y=-1.332'
        listener   = <function LcmSpy.wait_for_message_result.<locals>.listener at 0x7c5c268eeb60>
        predicate  = <function LcmSpy.wait_until_odom_position.<locals>.predicate at 0x7c5c268eeca0>
        self       = <dimos.e2e_tests.lcm_spy.LcmSpy object at 0x7c5c260ecad0>
        timeout    = 180
        topic      = '/odom#geometry_msgs.PoseStamped'
        type       = <class 'dimos.msgs.geometry_msgs.PoseStamped.PoseStamped'>
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

predicate = <bound method Event.is_set of <threading.Event at 0x7c5c2605b770: unset>>

    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: Failed to get to position x=-3.567, y=-1.332

deadline   = 3519794.279166228
interval   = 0.1
message    = 'Failed to get to position x=-3.567, y=-1.332'
predicate  = <bound method Event.is_set of <threading.Event at 0x7c5c2605b770: unset>>
timeout    = 180

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

dimos.perception.test_image_embedding.TestImageEmbedding::test_clip_embedding_initialization

Flake rate in main: 30.00% (Passed 7 times, Failed 3 times)

Stack Traces | 27.9s run time

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

    @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-249']. Please look at the first test that fails and fix that.

before     = {135308191344320, 135308199737024, 135308216522432, 135308224915136, 135308241700544, 135308250093248, ...}
deadline   = 5934259.892136311
expected_persistent_thread_prefixes = ['Dask-Offload', 'Thread-auto_conversion']
leaked     = [<TMonitor(Thread-249, started daemon 135308174558912)>]
live_new_threads = <function monitor_threads.<locals>.live_new_threads at 0x7b1046b776a0>
request    = <SubRequest 'monitor_threads' for <Function test_clip_embedding_initialization>>
t          = <TMonitor(Thread-249, started daemon 135308174558912)>
test_name  = 'dimos/perception/test_image_embedding.py::TestImageEmbedding::test_clip_embedding_initialization'
thread_names = ['Thread-249']
truly_new  = [<TMonitor(Thread-249, started daemon 135308174558912)>]

dimos/conftest.py:273: 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.}

[leshy]

Merged Paul's autofix PR #2748 into this branch at bf819c1 — all 15 commits taken (authorship preserved via merge). They resolve several of the earlier review threads (type ignores, inline imports, thread-safety of the stats read, degrade-warning logging, test teardown). Gate green: 54 spy/CLI tests + codebase checks pass, mypy + ruff clean.