feat: default generated message ports to CDR

.github/workflows/message-codegen.yml

@@ -142,14 +142,14 @@ jobs:
           cmake -S build/sdk-deps/zenoh-cpp -B build/sdk-deps/zenoh-cpp-build -DCMAKE_PREFIX_PATH="$PWD/build/sdk-deps/zenoh-c" -DCMAKE_INSTALL_PREFIX="$PWD/build/sdk-deps/zenoh-cpp-install" -DZENOHCXX_ENABLE_TESTS=OFF -DZENOHCXX_ENABLE_EXAMPLES=OFF
           cmake --install build/sdk-deps/zenoh-cpp-build
           .native-review/bin/python -m pip install pytest pytest-mock pytest-env
-      - name: Verify explicit CDR transports and actual SDK coordinator example
+      - name: Verify default CDR transports and actual SDK coordinator example
         if: runner.os == 'Linux'
         env:
           DIMOS_EXTERNAL_NATIVE_PYTHON: ${{ github.workspace }}/.native-review/bin/python
           DIMOS_MESSAGE_NATIVE_CACHE: ${{ github.workspace }}/build/native-project-cache
           DIMOS_CDR_NATIVE_SDK_PREFIX: ${{ github.workspace }}/build/sdk-deps/zenoh-c;${{ github.workspace }}/build/sdk-deps/zenoh-cpp-install
         run: |
-          .native-review/bin/python -I -m pytest --noconftest --import-mode=importlib -o addopts='' -q "$PWD/dimos/core/test_cdr_transport.py"
+          LCM_DEFAULT_URL=memq:// .native-review/bin/python -I -m pytest --noconftest --import-mode=importlib -o addopts='' -q "$PWD/dimos/core/test_cdr_transport.py" "$PWD/dimos/core/test_transport_factory.py"
           .venv/bin/python -m pytest --noconftest --import-mode=importlib -o addopts='' -o 'markers=native_e2e: compiled native consumer acceptance' -q dimos/message_codegen/test_project_native.py -k actual_coordinator
           cargo test --manifest-path native/rust/dimos-module/Cargo.toml --locked --features cdr --lib --jobs 2
       - name: Verify generated browser schemas, relay encoding and web consumers

dimos/core/cdr_transport.py

@@ -12,7 +12,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-"""Explicit CDR encoding over existing raw transports; no default cutover."""
+"""CDR encoding for generated messages over existing raw transports."""
 
 from __future__ import annotations
 
@@ -36,7 +36,9 @@
 
 
 class CdrTransport(PubSubTransport[T]):
-    """Opt-in plain XCDR1 values over raw LCM or Zenoh.
+    """Plain XCDR1 values over raw LCM or Zenoh.
+
+    The default transport factory selects this class for generated message types.
 
     The explicit /cdr/<package>/msg/<Type> suffix separates these channels
     from existing LCM-encoded streams. Both peers must select the same schema;

dimos/core/test_transport_factory.py

@@ -14,8 +14,14 @@
 
 from __future__ import annotations
 
+from contextlib import ExitStack
+import pickle
+
+from dimos_cdrgen.registry import encode
 import pytest
 
+import dimos.core.cdr_transport as cdr_module
+from dimos.core.cdr_transport import CdrTransport
 from dimos.core.global_config import GlobalConfig
 from dimos.core.transport import (
     LCMTransport,
@@ -31,6 +37,8 @@
     session_config,
     transport_topic,
 )
+from dimos.messages.sensor_msgs.msg import Image as CdrImage
+from dimos.messages.std_msgs.msg import Int32
 from dimos.msgs.geometry_msgs.PoseStamped import PoseStamped
 from dimos.msgs.geometry_msgs.Twist import Twist
 from dimos.msgs.sensor_msgs.Image import Image
@@ -155,3 +163,35 @@ def test_apply_transport_arg_bad_value_exits(argv: list[str]) -> None:
         apply_transport_arg(argv, g=g)
     assert exc.value.code == 2
     assert g.transport == "lcm"  # config left untouched
+
+
+@pytest.mark.parametrize("config", [LCM, ZENOH])
+def test_generated_factory_publishes_cdr_instead_of_pickle(config, mocker):
+    boundary = "LCMPubSubBase" if config.transport == "lcm" else "ZenohPubSubBase"
+    factory = mocker.patch.object(cdr_module, boundary, autospec=True)
+    transport = make_transport("reading", Int32, g=config)
+    with ExitStack() as resources:
+        resources.callback(transport.stop)
+        transport.publish(Int32(42))
+        factory.return_value.publish.assert_called_once_with(transport.topic, encode(Int32(42)))
+        prefix = "dimos/" if config.transport == "zenoh" else ""
+        assert transport.channel == prefix + "reading/cdr/std_msgs/msg/Int32"
+        restored = pickle.loads(pickle.dumps(transport))
+        assert type(restored) is CdrTransport
+        assert restored.channel == transport.channel
+
+
+def test_generated_sensor_qos_survives_factory_and_pickle():
+    sensor = make_transport("camera", CdrImage, g=ZENOH)
+    seed = make_transport("loaded_map", CdrImage, g=ZENOH)
+    assert sensor.publish_qos == QOS_LATEST_WINS.to_wire()
+    assert pickle.loads(pickle.dumps(seed)).publish_qos == QOS_NEVER_DROP.to_wire()
+    assert make_transport("camera", CdrImage, g=LCM).publish_qos is None
+
+
+def test_unregistered_generated_claim_is_rejected_instead_of_pickled():
+    class Unregistered:
+        __msgtype__ = "std_msgs/msg/Int32"
+
+    with pytest.raises(TypeError, match="registry-owned"):
+        make_transport("reading", Unregistered, g=ZENOH)

dimos/core/transport_factory.py

@@ -20,6 +20,7 @@
 import sys
 from typing import TYPE_CHECKING, Any, NoReturn, get_args
 
+from dimos.core.cdr_transport import CdrTransport
 from dimos.core.global_config import GlobalConfig, TransportBackend, global_config
 from dimos.core.transport import (
     LCMTransport,
@@ -58,7 +59,12 @@ def transport_topic(name: str, g: GlobalConfig = global_config) -> str:
 
 # High-rate sensor streams: drop stale frames under congestion, never stall the
 # publisher. Matched by message type since that is what makes them high-rate.
-_LATEST_WINS_TYPES = ("sensor_msgs.Image", "sensor_msgs.PointCloud2")
+_LATEST_WINS_TYPES = (
+    "sensor_msgs.Image",
+    "sensor_msgs.PointCloud2",
+    "sensor_msgs/msg/Image",
+    "sensor_msgs/msg/PointCloud2",
+)
 # Low-rate channels where a drop loses something that never comes back: a whole
 # turn of agent/human conversation, a one-shot robot action verb, a push-to-talk
 # chunk where one gap discards the whole utterance, or a region of a seeded map
@@ -91,16 +97,20 @@ def default_zenoh_qos_for(name: str, msg_name: str) -> ZenohQoS | None:
 
 def default_zenoh_qos(name: str, msg_type: type | None = None) -> ZenohQoS | None:
     """Default publisher QoS for a logical channel; None = zenoh defaults."""
-    return default_zenoh_qos_for(name, getattr(msg_type, "msg_name", ""))
+    return default_zenoh_qos_for(
+        name, getattr(msg_type, "__msgtype__", getattr(msg_type, "msg_name", ""))
+    )
 
 
 def make_transport(
     name: str, msg_type: type | None = None, *, g: GlobalConfig = global_config
 ) -> PubSubTransport[Any]:
     """Construct the active-backend pub/sub transport for a logical channel.
 
-    A pickled (self-describing) transport is used when no `msg_type` is given or
-    the type has no `lcm_encode`. Otherwise a typed transport is used.
+    Registry-owned generated messages use CDR on the selected raw backend.
+    Legacy types with `lcm_encode` retain their explicit legacy encoding; other
+    Python values use pickle. CDR channels have a separate `/cdr/` suffix;
+    there is no encoding negotiation or fallback decoding.
 
     A channel name alone doesn't fully define a topic: backends have per-topic
     settings (Zenoh publisher QoS, for one) that live on their Topic objects.
@@ -110,8 +120,15 @@ def make_transport(
     transport map. LCM (UDP multicast) has no per-topic settings.
     """
 
-    use_pickled = msg_type is None or getattr(msg_type, "lcm_encode", None) is None
     topic = transport_topic(name, g)
+    if msg_type is not None and hasattr(msg_type, "__msgtype__"):
+        return CdrTransport(
+            topic,
+            msg_type,
+            backend=g.transport,
+            qos=default_zenoh_qos(name, msg_type) if g.transport == "zenoh" else None,
+        )
+    use_pickled = msg_type is None or getattr(msg_type, "lcm_encode", None) is None
     if g.transport == "zenoh":
         ztopic = ZenohTopic(
             topic, None if use_pickled else msg_type, qos=default_zenoh_qos(name, msg_type)

dimos/models/embedding/clip.py

@@ -17,12 +17,12 @@
 from functools import cached_property
 from typing import overload
 
-from dimos.messages.sensor_msgs.msg import Image
 from PIL import Image as PILImage
 import torch
 import torch.nn.functional as functional
 from transformers import CLIPModel as HFCLIPModel, CLIPProcessor
 
+from dimos.messages.sensor_msgs.msg import Image
 from dimos.models.base import HuggingFaceModel
 from dimos.models.embedding.base import Embedding, EmbeddingModel, HuggingFaceEmbeddingModelConfig
 from dimos.msgs.image import image_to_rgb

dimos/models/embedding/dino.py

@@ -17,12 +17,12 @@
 from functools import cached_property
 from typing import Any, overload
 
-from dimos.messages.sensor_msgs.msg import Image
 from PIL import Image as PILImage
 import torch
 import torch.nn.functional as functional
 from transformers import AutoImageProcessor, AutoModel
 
+from dimos.messages.sensor_msgs.msg import Image
 from dimos.models.base import HuggingFaceModel
 from dimos.models.embedding.base import Embedding, EmbeddingModel, HuggingFaceEmbeddingModelConfig
 from dimos.msgs.image import image_to_rgb

dimos/models/embedding/siglip.py

@@ -17,12 +17,12 @@
 from functools import cached_property
 from typing import overload
 
-from dimos.messages.sensor_msgs.msg import Image
 from PIL import Image as PILImage
 import torch
 import torch.nn.functional as functional
 from transformers import SiglipModel as HFSiglipModel, SiglipProcessor
 
+from dimos.messages.sensor_msgs.msg import Image
 from dimos.models.base import HuggingFaceModel
 from dimos.models.embedding.base import Embedding, EmbeddingModel, HuggingFaceEmbeddingModelConfig
 from dimos.msgs.image import image_to_rgb

dimos/models/embedding/siglip2.py

@@ -17,12 +17,12 @@
 from functools import cached_property
 from typing import overload
 
-from dimos.messages.sensor_msgs.msg import Image
 from PIL import Image as PILImage
 import torch
 import torch.nn.functional as functional
 from transformers import SiglipModel as HFSiglipModel, SiglipProcessor
 
+from dimos.messages.sensor_msgs.msg import Image
 from dimos.models.base import HuggingFaceModel
 from dimos.models.embedding.base import Embedding, EmbeddingModel, HuggingFaceEmbeddingModelConfig
 from dimos.msgs.image import image_to_rgb

dimos/models/embedding/test_generated_images.py

@@ -29,20 +29,25 @@
 from dimos.msgs.image import image_from_array, image_view
 
 
-@pytest.mark.parametrize("model_type", [CLIPModel,DINOModel,SigLIPModel,SigLIP2Model])
-def test_generated_embeddings_keep_zero_source_time_and_convert_rgb(mocker,model_type):
-    processor,network=mocker.Mock(),mocker.Mock()
-    features=torch.tensor([[3.0,4.0]])
-    processor.return_value=BatchFeature(data={"pixel_values":torch.zeros(1,3,4,8)})
-    network.get_image_features.return_value=features
-    network.return_value=BaseModelOutputWithPooling(last_hidden_state=features[:,None,:],pooler_output=features)
-    mocker.patch.object(model_type,"_processor",new_callable=PropertyMock,return_value=processor)
-    mocker.patch.object(model_type,"_model",new_callable=PropertyMock,return_value=network)
-    image=image_from_array(np.full((4,8,3),[0,0,255],dtype=np.uint8),encoding="bgr8",
-        header=Header(frame_id="camera",stamp=Time(sec=0,nanosec=0)))
-    model=model_type(device="cpu")
-    result=model.embed(image)
+@pytest.mark.parametrize("model_type", [CLIPModel, DINOModel, SigLIPModel, SigLIP2Model])
+def test_generated_embeddings_keep_zero_source_time_and_convert_rgb(mocker, model_type):
+    processor, network = mocker.Mock(), mocker.Mock()
+    features = torch.tensor([[3.0, 4.0]])
+    processor.return_value = BatchFeature(data={"pixel_values": torch.zeros(1, 3, 4, 8)})
+    network.get_image_features.return_value = features
+    network.return_value = BaseModelOutputWithPooling(
+        last_hidden_state=features[:, None, :], pooler_output=features
+    )
+    mocker.patch.object(model_type, "_processor", new_callable=PropertyMock, return_value=processor)
+    mocker.patch.object(model_type, "_model", new_callable=PropertyMock, return_value=network)
+    image = image_from_array(
+        np.full((4, 8, 3), [0, 0, 255], dtype=np.uint8),
+        encoding="bgr8",
+        header=Header(frame_id="camera", stamp=Time(sec=0, nanosec=0)),
+    )
+    model = model_type(device="cpu")
+    result = model.embed(image)
     assert result.timestamp == 0.0
-    torch.testing.assert_close(result.vector,torch.tensor([0.6,0.8]))
-    assert processor.call_args.kwargs["images"][0].getpixel((0,0)) == (255,0,0)
-    np.testing.assert_array_equal(image_view(image)[0,0],[0,0,255])
+    torch.testing.assert_close(result.vector, torch.tensor([0.6, 0.8]))
+    assert processor.call_args.kwargs["images"][0].getpixel((0, 0)) == (255, 0, 0)
+    np.testing.assert_array_equal(image_view(image)[0, 0], [0, 0, 255])

dimos/models/embedding/treid.py

@@ -19,11 +19,11 @@
 
 from functools import cached_property
 
-from dimos.messages.sensor_msgs.msg import Image
 import torch
 import torch.nn.functional as functional
 from torchreid import utils as torchreid_utils
 
+from dimos.messages.sensor_msgs.msg import Image
 from dimos.models.base import LocalModel
 from dimos.models.embedding.base import Embedding, EmbeddingModel, EmbeddingModelConfig
 from dimos.msgs.image import image_to_bgr

dimos/models/segmentation/edge_tam.py

@@ -20,14 +20,14 @@
 import tempfile
 from typing import TYPE_CHECKING, Any, Protocol, TypedDict
 
-from dimos.messages.sensor_msgs.msg import Image
 from hydra.utils import instantiate
 import numpy as np
 from numpy.typing import NDArray
 from omegaconf import OmegaConf
 from PIL import Image as PILImage
 import torch
 
+from dimos.messages.sensor_msgs.msg import Image
 from dimos.models.base import default_torch_device
 from dimos.msgs.image import image_to_bgr, image_to_jpeg
 from dimos.perception.detection.detectors.base import Detector

dimos/models/segmentation/test_yoloe.py

@@ -16,11 +16,11 @@
 from typing import Any
 from unittest.mock import MagicMock
 
-from dimos.messages.sensor_msgs.msg import Image
-from dimos.messages.std_msgs.msg import Header
 import numpy as np
 from ultralytics.models.yolo.yoloe import YOLOEVPSegPredictor
 
+from dimos.messages.sensor_msgs.msg import Image
+from dimos.messages.std_msgs.msg import Header
 from dimos.models.segmentation.yoloe import YoloeBoxSegmenter
 from dimos.msgs.image import image_from_array, image_to_bgr
 from dimos.msgs.time import time_from_seconds, to_seconds

dimos/models/segmentation/yoloe.py

@@ -19,11 +19,11 @@
 from dataclasses import replace
 from typing import Protocol
 
-from dimos.messages.sensor_msgs.msg import Image
 import numpy as np
 from numpy.typing import NDArray
 from ultralytics.models.yolo.yoloe import YOLOEVPSegPredictor
 
+from dimos.messages.sensor_msgs.msg import Image
 from dimos.msgs.image import image_to_bgr
 from dimos.perception.detection.detectors.yoloe import Yoloe2DDetector, YoloePromptMode
 from dimos.perception.detection.type.detection2d.bbox import Bbox, Detection2DBBox

examples/external-messages/demo_modules.py

@@ -30,7 +30,6 @@
 from reactivex.disposable import Disposable
 import zenoh
 
-from dimos.core.cdr_transport import CdrTransport
 from dimos.core.coordination.blueprints import autoconnect
 from dimos.core.coordination.module_coordinator import ModuleCoordinator
 from dimos.core.core import rpc
@@ -112,40 +111,22 @@ def run(cpp: Path, rust: Path) -> None:
             gossip=False,
             connect_timeout=10,
         )
-        blueprint = (
-            autoconnect(
-                Probe.blueprint(),
-                CppRelay.blueprint(
-                    executable=str(cpp.resolve()), stdin_config=True, session=session
-                ),
-                RustRelay.blueprint(
-                    executable=str(rust.resolve()), stdin_config=True, session=session
-                ),
-            )
-            .transports(
-                {
-                    (name, DeviceReading): CdrTransport.spec(
-                        "example/" + name,
-                        DeviceReading,
-                        backend="zenoh",
-                        **session.model_dump(),
-                    )
-                    for name in ("raw", "cpp", "done")
-                }
-            )
-            .global_config(
-                viewer="none",
-                n_workers=3,
-                transport="zenoh",
-                robot_ip=None,
-                robot_ips="",
-                zenoh_connect=endpoint,
-                zenoh_mode="client",
-                zenoh_multicast=False,
-                zenoh_gossip=False,
-                zenoh_scouting=False,
-                zenoh_connect_timeout=10,
-            )
+        blueprint = autoconnect(
+            Probe.blueprint(),
+            CppRelay.blueprint(executable=str(cpp.resolve()), stdin_config=True, session=session),
+            RustRelay.blueprint(executable=str(rust.resolve()), stdin_config=True, session=session),
+        ).global_config(
+            viewer="none",
+            n_workers=3,
+            transport="zenoh",
+            robot_ip=None,
+            robot_ips="",
+            zenoh_connect=endpoint,
+            zenoh_mode="client",
+            zenoh_multicast=False,
+            zenoh_gossip=False,
+            zenoh_scouting=False,
+            zenoh_connect_timeout=10,
         )
         coordinator = ModuleCoordinator.build(blueprint)
         resources.callback(coordinator.stop)