feat(transport): rust zenoh support and api refactor

[aclauer]

Problem

I swear it's not actually 18k lines, most of that is Cargo.lock

Note: currently don't have QoS, that needs to be added later

Closes DIM-XXX

Solution

• rewrite the Rust transport trait to be a real pub sub contract
• add Zenoh transport. key expressions are passed to Rust from Python

How to Test

local ping pong

python3 rust_ping_pong.py --transport=zenoh
python3 rust_ping_pong.py --transport=lcm

remote ping pong (run on different machines, connected to same network)

DIMOS_TRANSPORT=zenoh ./target/release/pong <<< '{"topics":{"data":"dimos_pingpong_data","confirm":"dimos_pingpong_confirm"},"config":{"sample_config":0}}'

DIMOS_TRANSPORT=zenoh ./target/release/ping <<< '{"topics":{"data":"dimos_pingpong_data","confirm":"dimos_pingpong_confirm"},"config":null}'

benchmark between lcm and zenoh locally

cd dimos/native/rust
cargo run --release --example transport_benchmark

Contributor License Agreement

• I have read and approved the CLA.

[codecov]

❌ 1 Tests Failed:

Tests completed	Failed	Passed	Skipped
2688	1	2687	71

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

dimos.perception.test_image_embedding.TestImageEmbedding::test_clip_embedding_initialization

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

Stack Traces | 25.7s 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     = {127126300702400, 127126309095104, 127126325880512, 127126334273216, 127127495358144, 127127503750848, ...}
deadline   = 5920673.903167983
expected_persistent_thread_prefixes = ['Dask-Offload', 'Thread-auto_conversion']
leaked     = [<TMonitor(Thread-249, started daemon 127126283916992)>]
live_new_threads = <function monitor_threads.<locals>.live_new_threads at 0x739f487df420>
request    = <SubRequest 'monitor_threads' for <Function test_clip_embedding_initialization>>
t          = <TMonitor(Thread-249, started daemon 127126283916992)>
test_name  = 'dimos/perception/test_image_embedding.py::TestImageEmbedding::test_clip_embedding_initialization'
thread_names = ['Thread-249']
truly_new  = [<TMonitor(Thread-249, started daemon 127126283916992)>]

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.}

[greptile-apps]

Greptile Summary

This PR adds Zenoh support and refactors native-module transport handling. The main changes are:

• A publish/subscribe transport trait for Rust native modules.
• Runtime selection between LCM and Zenoh via DIMOS_TRANSPORT.
• Zenoh channel/key-expression mapping from Python to Rust.
• Updated Rust ping-pong examples and a transport benchmark.

Confidence Score: 4/5

This is close, but the LCM receive path should be fixed before merging.

• The LCM transport still routes all subscribed channels through one synchronous callback loop.
• A slow decode on one channel can delay or drop messages for unrelated channels.
• The Zenoh subscription lifetime and Python-to-Rust channel mapping look consistent in the reviewed changes.

native/rust/dimos-module/src/lcm.rs

Important Files Changed

Filename	Overview
native/rust/dimos-module/src/lcm.rs	LCM now implements the subscribe callback contract, but its shared receive loop can still block unrelated channels.
native/rust/dimos-module/src/zenoh.rs	Adds the Zenoh transport using session-scoped background subscriptions.
dimos/core/native_module.py	Passes the selected transport to native binaries and sends transport-specific channel strings.
dimos/core/transport.py	Adds channel properties for LCM channel names and Zenoh key expressions.
dimos/protocol/pubsub/impl/zenohpubsub.py	Moves Zenoh key-expression generation onto the topic model.

_{Reviews (4): Last reviewed commit: "Precommit please" | Re-trigger Greptile}