Default to Zenoh transport on macOS and document replay workflow

dimos/core/coordination/module_coordinator.py

@@ -29,7 +29,7 @@
 from dimos.core.global_config import GlobalConfig, global_config
 from dimos.core.module import ModuleBase, ModuleSpec
 from dimos.core.resource import Resource
-from dimos.core.transport import LCMTransport, PubSubTransport, pLCMTransport
+from dimos.core.transport import ZENOH_AVAILABLE, LCMTransport, PubSubTransport, pLCMTransport
 from dimos.spec.utils import spec_annotation_compliance, spec_structural_compliance
 from dimos.utils.generic import short_id
 from dimos.utils.logging_config import setup_logger
@@ -542,7 +542,23 @@ def _get_transport_for(blueprint: Blueprint, name: str, stream_type: type) -> Pu
 
     use_pickled = getattr(stream_type, "lcm_encode", None) is None
     topic = f"/{name}" if _is_name_unique(blueprint, name) else f"/{short_id()}"
-    transport = pLCMTransport(topic) if use_pickled else LCMTransport(topic, stream_type)
+
+    if global_config.transport == "zenoh":
+        if not ZENOH_AVAILABLE:
+            raise RuntimeError(
+                "transport='zenoh' but eclipse-zenoh is not installed. "
+                "Install with: uv sync --extra zenoh"
+            )
+        from dimos.core.transport import ZenohTransport, pZenohTransport
+
+        zenoh_topic = f"dimos{topic}"
+        transport = (
+            pZenohTransport(zenoh_topic)
+            if use_pickled
+            else ZenohTransport(zenoh_topic, stream_type)
+        )
+    else:
+        transport = pLCMTransport(topic) if use_pickled else LCMTransport(topic, stream_type)
 
     return transport
 
@@ -612,7 +628,8 @@ def _run_configurators(blueprint: Blueprint) -> None:
     from dimos.protocol.service.system_configurator.base import configure_system
     from dimos.protocol.service.system_configurator.lcm_config import lcm_configurators
 
-    configurators = [*lcm_configurators(), *blueprint.configurator_checks]
+    lcm_checks = lcm_configurators() if global_config.transport == "lcm" else []
+    configurators = [*lcm_checks, *blueprint.configurator_checks]
 
     try:
         configure_system(configurators)

dimos/core/global_config.py

@@ -12,20 +12,30 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import platform
 import re
 from typing import Literal, TypeAlias
 
+from pydantic import Field
 from pydantic_settings import BaseSettings, SettingsConfigDict
 
+from dimos.core.transport import ZENOH_AVAILABLE
 from dimos.models.vl.types import VlModelName
 
 ViewerBackend: TypeAlias = Literal["rerun", "rerun-web", "rerun-connect", "foxglove", "none"]
+TransportBackend: TypeAlias = Literal["lcm", "zenoh"]
 
 
 def _get_all_numbers(s: str) -> list[float]:
     return [float(x) for x in re.findall(r"-?\d+\.?\d*", s)]
 
 
+def _default_transport() -> TransportBackend:
+    if platform.system() == "Darwin" and ZENOH_AVAILABLE:
+        return "zenoh"
+    return "lcm"
+
+
 class GlobalConfig(BaseSettings):
     robot_ip: str | None = None
     robot_ips: str | None = None
@@ -52,6 +62,7 @@ class GlobalConfig(BaseSettings):
     nerf_speed: float = 1.0
     planner_robot_speed: float | None = None
     mcp_port: int = 9990
+    transport: TransportBackend = Field(default_factory=_default_transport)
     dtop: bool = False
     obstacle_avoidance: bool = True
     detection_model: VlModelName = "moondream"
@@ -61,6 +72,7 @@ class GlobalConfig(BaseSettings):
         env_file=".env",
         env_file_encoding="utf-8",
         extra="ignore",
+        validate_assignment=True,
     )
 
     def update(self, **kwargs: object) -> None:

dimos/core/test_utils.py

@@ -0,0 +1,43 @@
+# Copyright 2025-2026 Dimensional Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Shared test helpers that must remain import-safe in minimal environments."""
+
+from __future__ import annotations
+
+from collections.abc import Callable
+import threading
+
+
+def retry_until(
+    event: threading.Event,
+    action: Callable[[], None],
+    timeout: float = 2.0,
+    interval: float = 0.01,
+) -> None:
+    """Retry an action until an Event fires.
+
+    Useful for async test paths where the producer may need to retry briefly
+    before the subscriber is fully ready.
+    """
+    deadline = threading.Event()
+    timer = threading.Timer(timeout, deadline.set)
+    timer.start()
+    try:
+        while not event.is_set() and not deadline.is_set():
+            action()
+            event.wait(interval)
+    finally:
+        timer.cancel()
+    assert event.is_set(), f"Timed out after {timeout}s waiting for event"