Feat/integrate zenoh

[leshy]

Problem

Closes DIM-XXX

Solution

Breaking Changes

How to Test

Contributor License Agreement

• I have read and approved the CLA.

[leshy]

running
dimos --transport=zenoh --dtop --replay --replay-dir=unitree_go2_bigoffice run unitree-go2

it crashes on import error (not below RuntimeError, I found this in code)

            raise RuntimeError(
                "transport='zenoh' but eclipse-zenoh is not installed. "
                "Install with: uv sync --extra zenoh"
            )

but executing uv sync --extra zenoh uninstalls everything else from my uv env

[vrinek]

@leshy I am not very familiar with uv and Python's package management yet. To unblock this, and simplify, I can make Zenoh a required dependency. What do you think?

[vrinek]

Running the benchmarks for longer (50k msgs, 10sec) prints more reliable results. QoS at default of DROP + RELIABLE + DATA.

[leshy]

@leshy I am not very familiar with uv and Python's package management yet. To unblock this, and simplify, I can make Zenoh a required dependency. What do you think?

let's learn how to use UV, it's our main package manager, this is not the last feature of this type etc @paul-nechifor will know

[greptile-apps]

Greptile Summary

This PR integrates eclipse-zenoh as an optional pub/sub transport alongside the existing LCM stack, adding ZenohService (singleton session management), ZenohPubSubBase/Zenoh/PickleZenoh pub/sub implementations, ZenohTransport/pZenohTransport wrappers, and wires them into _get_transport_for and the Rerun bridge.

• _default_pubsubs in bridge.py silently returns [LCM()] when transport="zenoh" but eclipse-zenoh is not installed — no warning is logged. _get_transport_for raises a RuntimeError for the same condition; the bridge should at minimum emit a warning so operators can diagnose why Zenoh messages are missing from the viewer.

Confidence Score: 4/5

Safe to merge with one P1 fix: the silent LCM fallback in _default_pubsubs should log a warning when Zenoh is configured but unavailable.

One P1 finding: _default_pubsubs silently falls back to LCM when Zenoh is configured but not installed, causing invisible message loss in the Rerun bridge without any diagnostic message. The rest of the implementation is well-structured, follows existing patterns, has good lock discipline, and is covered by comprehensive tests.

dimos/visualization/rerun/bridge.py — silent Zenoh unavailability fallback needs a warning log

Important Files Changed

Filename	Overview
dimos/visualization/rerun/bridge.py	Updated _default_pubsubs to support Zenoh transport; silently falls back to LCM when Zenoh is configured but unavailable, with no warning logged
dimos/protocol/service/zenohservice.py	New file: Zenoh singleton session manager with per-config caching; minor issues with module-level side effect and unlocked session property read
dimos/protocol/pubsub/impl/zenohpubsub.py	New file: raw-bytes and encoder-composed Zenoh pub/sub; well-structured with proper lock discipline, unsubscribe idempotency, and good topic/key-expression conversion
dimos/core/transport.py	Adds ZenohTransport and pZenohTransport under ZENOH_AVAILABLE guard, following established DDSTransport pattern with RLock for thread-safe lazy start
dimos/core/coordination/module_coordinator.py	_get_transport_for updated to route to Zenoh transports when configured; correctly raises RuntimeError if Zenoh is unavailable; _run_configurators skips LCM checks for Zenoh transport

Sequence Diagram

sequenceDiagram
    participant App
    participant ModuleCoordinator
    participant _get_transport_for
    participant ZenohTransport
    participant Zenoh (LCMEncoderMixin)
    participant ZenohPubSubBase
    participant ZenohService
    participant zenoh.Session

    App->>ModuleCoordinator: deploy_blueprint()
    ModuleCoordinator->>_get_transport_for: (name, stream_type)
    _get_transport_for-->>ZenohTransport: ZenohTransport(topic, type)
    ZenohTransport->>ZenohService: start()
    ZenohService->>zenoh.Session: open(config) [once per session_key]
    ZenohTransport->>Zenoh (LCMEncoderMixin): publish(LCMTopic, msg)
    Zenoh (LCMEncoderMixin)->>ZenohPubSubBase: publish(topic, bytes)
    ZenohPubSubBase->>zenoh.Session: publisher.put(bytes)
    zenoh.Session-->>ZenohPubSubBase: callback(sample)
    ZenohPubSubBase-->>Zenoh (LCMEncoderMixin): decode bytes → msg
    Zenoh (LCMEncoderMixin)-->>App: subscriber callback(msg, topic)

Loading

_{Reviews (1): Last reviewed commit: "Merge branch 'dev' into feat/integrate-z..." | Re-trigger Greptile}

[greptile-apps]

Silent Zenoh unavailability fallback

When transport == "zenoh" but eclipse-zenoh is not installed, _default_pubsubs silently returns [LCM()] with no log warning. The bridge will subscribe to LCM only, so Zenoh-published messages are invisible in the viewer — and there is no indication of why.

_get_transport_for in module_coordinator.py raises RuntimeError("transport='zenoh' but eclipse-zenoh is not installed") for the same condition. _default_pubsubs should at minimum emit a warning so operators can diagnose the mismatch.

Suggested change

	if transport == "zenoh":
	from dimos.core.transport import ZENOH_AVAILABLE
	if ZENOH_AVAILABLE:
	from dimos.protocol.pubsub.impl.zenohpubsub import Zenoh
	return [Zenoh(), LCM()]
	return [LCM()]
	if ZENOH_AVAILABLE:
	from dimos.protocol.pubsub.impl.zenohpubsub import Zenoh
	return [Zenoh(), LCM()]
	logger.warning(
	"transport='zenoh' but eclipse-zenoh is not installed; "
	"RerunBridge falling back to LCM only. "
	"Install with: uv sync --extra zenoh"
	)

[greptile-apps]

Side-effecting call placed between import statements

zenoh.init_log_from_env_or("warn") is interleaved between two import statements. While the intent (initialize Rust tracing before any zenoh operation) is sound, PEP 8 requires all imports to come first. Consider moving this call to the end of the import block or wrapping it in a function called at session creation time.

[greptile-apps]

session property reads shared dict without holding _sessions_lock

start() writes to _sessions under _sessions_lock, but session reads from it without the lock. The check-then-use pattern is logically racy and inconsistent with the lock discipline established in start(). Consider acquiring _sessions_lock in the session property before the if key not in _sessions check.

[Copilot]

Pull request overview

This PR introduces an optional Zenoh-based transport path alongside the existing LCM transport, wiring it through core transport selection, adding a Zenoh pubsub implementation + session service, and updating the Rerun viewer bridge and tests to exercise Zenoh integration.

Changes:

• Add a zenoh optional dependency extra (eclipse-zenoh) and lockfile entries.
• Implement Zenoh session management (ZenohService) and Zenoh pubsub (ZenohPubSubBase, Zenoh, PickleZenoh), plus core transport wrappers and coordinator branching.
• Extend existing test/benchmark harnesses and the Rerun bridge to operate with Zenoh (when available).

Reviewed changes

Copilot reviewed 12 out of 13 changed files in this pull request and generated 7 comments.

Show a summary per file

File	Description
uv.lock	Adds eclipse-zenoh package and a new zenoh extra in the resolved lock metadata.
pyproject.toml	Defines a new zenoh extra (eclipse-zenoh>=1.0.0,<2.0).
dimos/core/global_config.py	Adds transport setting (default lcm) to control transport selection.
dimos/core/transport.py	Adds ZENOH_AVAILABLE detection and Zenoh transport wrappers when available.
dimos/core/coordination/module_coordinator.py	Switches stream transport selection based on global_config.transport; gates LCM configurators.
dimos/visualization/rerun/bridge.py	Resolves pubsub backend lazily and normalizes entity paths for Zenoh topics.
dimos/protocol/service/zenohservice.py	Introduces singleton Zenoh session management service.
dimos/protocol/service/test_zenohservice.py	Adds unit tests for Zenoh session singleton behavior.
dimos/protocol/pubsub/impl/zenohpubsub.py	Implements Zenoh pubsub backend + key-expression/topic conversion helpers.
dimos/protocol/pubsub/impl/test_zenohpubsub.py	Adds integration tests for Zenoh pubsub behavior and conversions.
dimos/protocol/pubsub/test_spec.py	Extends the generic pubsub spec tests to include Zenoh variants when available.
dimos/protocol/pubsub/benchmark/testdata.py	Adds Zenoh to the pubsub benchmark testcase list when available.
dimos/core/test_zenoh_transport.py	Adds tests for coordinator branching and Zenoh transport wrapper lifecycle.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[Copilot]

GlobalConfig.transport is a free-form str, but the code branches on exact string values (e.g. "lcm" vs "zenoh"). To avoid silent fallback on typos and to get early validation from Pydantic settings, consider typing this as a Literal[...] (or an enum) consistent with viewer: ViewerBackend.

[Copilot]

When transport == "zenoh" but ZENOH_AVAILABLE is false, this silently falls back to [LCM()]. That can hide a misconfiguration (the rest of the system raises for transport='zenoh' without the extra). Consider raising a clear error or at least logging a warning so the viewer doesn't quietly listen on the wrong transport.

Suggested change

	return [Zenoh(), LCM()]
	return [Zenoh(), LCM()]
	raise RuntimeError(
	"transport='zenoh' was requested for the rerun bridge, but Zenoh support "
	"is not available. Install the Zenoh extra/dependencies or choose a "
	"different transport."
	)

[Copilot]

ZenohService caches sessions globally in _sessions but never closes/evicts them (and stop() explicitly does not close the session). If code creates services with multiple distinct configs over the process lifetime, sessions (and Zenoh background threads/resources) will accumulate without a cleanup path. Consider adding reference counting, an explicit close()/shutdown() API, or closing/removing the session when no live service instances are using a given session_key.

[Copilot]

These tests import dimos.protocol.service.zenohservice at module import time, which imports zenoh unconditionally. If the zenoh extra isn't installed (the default in most CI/dev envs), pytest collection will crash before any skipif can apply. Guard the module with pytest.importorskip("zenoh") / pytest.skip(..., allow_module_level=True) or move the import behind a ZENOH_AVAILABLE check so the file is safely skipped when Zenoh isn't installed.

Suggested change

	pytest.importorskip("zenoh")

[Copilot]

This test module imports ZenohPubSubBase/zenohservice at import time; because zenohservice.py imports zenoh unconditionally, pytest collection will fail when the zenoh extra isn't installed. Add a module-level pytest.importorskip("zenoh") (or skip based on ZENOH_AVAILABLE) before importing Zenoh-dependent modules.

Suggested change

	pytest.importorskip("zenoh")

[Copilot]

_retry_until is imported from dimos.protocol.pubsub.impl.test_zenohpubsub, which itself imports Zenoh modules unconditionally. This makes the entire test file fail to import when Zenoh isn't installed, even though most tests here are not Zenoh-dependent. Consider duplicating _retry_until locally (or moving it to a non-Zenoh test utility module) and guarding Zenoh-only imports/tests with pytest.importorskip("zenoh") / skipif(not ZENOH_AVAILABLE) at module level.

[Copilot]

_run_configurators() now skips all LCM system configurators when global_config.transport != "lcm". However ModuleConfig still uses LCM for RPC (LCMRPC) and TF (LCMTF) regardless of the stream transport, so skipping multicast/buffer configuration can break core functionality even in transport="zenoh" mode. Consider always running LCM configurators (or at least the subset required for RPC/TF), and gating only the parts that are strictly tied to LCM pubsub streams.

Suggested change

	lcm_checks = lcm_configurators() if global_config.transport == "lcm" else []
	# LCM is still used by core facilities such as RPC/TF even when the
	# selected stream transport is not "lcm", so LCM configurators must
	# always run here. Only configurators that are strictly specific to
	# optional LCM pubsub streams should be gated elsewhere.
	lcm_checks = lcm_configurators()