feat(transport): integrate shared CDR codecs and native SDK

.github/workflows/cdr-native.yml

@@ -0,0 +1,37 @@
+name: CDR native interoperability
+
+on:
+  pull_request:
+  workflow_dispatch:
+  push:
+    branches: [main]
+
+permissions:
+  contents: read
+
+jobs:
+  native:
+    runs-on: ubuntu-latest
+    timeout-minutes: 30
+    env:
+      DIMOS_ALLOW_MISSING_COCKPIT: "1"
+      CARGO_TARGET_DIR: ${{ github.workspace }}/target
+    steps:
+      - uses: actions/checkout@v7
+        with:
+          lfs: false
+          persist-credentials: false
+      - uses: astral-sh/setup-uv@v10.2.0
+        with:
+          python-version: '3.12'
+          enable-cache: true
+      - name: Install build and media dependencies
+        run: |
+          sudo apt-get update
+          sudo apt-get install -y cmake pkg-config clang libclang-dev portaudio19-dev libturbojpeg ffmpeg
+      - name: Install Python project and test dependencies
+        run: uv sync --frozen --group tests
+      - name: Test native SDK
+        run: |
+          cargo test --locked --manifest-path native/rust/dimos-module/Cargo.toml
+          cargo test --locked --manifest-path native/rust/dimos-module-macros/Cargo.toml

dimos/core/coordination/test_blueprints.py

@@ -38,7 +38,7 @@
 from dimos.core.core import rpc
 from dimos.core.module import Module
 from dimos.core.stream import In, Out
-from dimos.core.transport import LCMTransport
+from dimos.core.transport import pLCMTransport
 from dimos.spec.utils import Spec
 
 
@@ -139,7 +139,7 @@ def test_config_rejects_unknown_fields_and_sections() -> None:
 
 
 def test_transports() -> None:
-    custom_transport = LCMTransport("/custom_topic", Data1)
+    custom_transport = pLCMTransport("/custom_topic")
     blueprint_set = autoconnect(ModuleA.blueprint(), ModuleB.blueprint()).transports(
         {("data1", Data1): custom_transport}
     )
@@ -414,12 +414,12 @@ def test_explicit_instance_name_sets_blueprint_identity() -> None:
 def test_namespace_prefixes_pinned_transports() -> None:
     blueprint = (
         autoconnect(ModuleA.blueprint(), ModuleB.blueprint())
-        .transports({("data1", Data1): LCMTransport("/custom_topic", Data1)})
+        .transports({("data1", Data1): pLCMTransport("/custom_topic")})
         .namespace("robot0")
     )
 
     transport = blueprint.transport_map[("robot0/data1", Data1)]
-    assert transport.topic.pattern == "/robot0/custom_topic"
+    assert transport.topic == "/robot0/custom_topic"
     assert ("data1", Data1) not in blueprint.transport_map
 
 

dimos/core/test_dds_transport.py

@@ -0,0 +1,37 @@
+# Copyright 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.
+
+from dataclasses import dataclass
+
+import pytest
+
+from dimos.core.transport import DDSTransport
+
+idl = pytest.importorskip("cyclonedds.idl")
+
+
+@dataclass
+class NativeSample(idl.IdlStruct):
+    value: int
+
+
+def test_dds_transport_accepts_native_idl_type():
+    transport = DDSTransport("native_sample", NativeSample)
+    try:
+        assert transport.topic.name == "native_sample"
+        assert transport.topic.data_type is NativeSample
+        sample = NativeSample(value=42)
+        assert transport.topic.data_type.deserialize(sample.serialize()) == sample
+    finally:
+        transport.stop()

dimos/core/test_transport_factory.py

@@ -18,7 +18,9 @@
 
 from dimos.core.global_config import GlobalConfig
 from dimos.core.transport import (
+    DDSTransport,
     LCMTransport,
+    ROSTransport,
     ZenohTransport,
     pLCMTransport,
     pZenohTransport,
@@ -31,6 +33,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
@@ -82,8 +86,9 @@ def test_make_transport_zenoh_pickled() -> None:
     assert t.topic == "dimos/human_input"
 
 
-def test_default_zenoh_qos_high_rate_sensor_types_drop() -> None:
-    assert default_zenoh_qos("/camera/color", Image) == QOS_LATEST_WINS
+@pytest.mark.parametrize("message_type", [Image, CdrImage])
+def test_default_zenoh_qos_high_rate_sensor_types_drop(message_type) -> None:
+    assert default_zenoh_qos("/camera/color", message_type) == QOS_LATEST_WINS
 
 
 def test_default_zenoh_qos_agent_channels_never_drop() -> None:
@@ -155,3 +160,33 @@ 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_factory_binds_generated_messages(config):
+    transport = make_transport("generated", Int32, g=config)
+    assert transport.topic.message.encoding == "cdr"
+    assert transport.topic.message.message_type is Int32
+
+
+@pytest.mark.parametrize("config", [LCM, ZENOH])
+@pytest.mark.parametrize("message_type", [dict, int, str, bool])
+def test_factory_preserves_pickle_for_python_types(config, message_type):
+    transport = make_transport("python", message_type, g=config)
+    expected = pLCMTransport if config.transport == "lcm" else pZenohTransport
+    assert isinstance(transport, expected)
+
+
+@pytest.mark.parametrize("transport_type", [ROSTransport, DDSTransport])
+def test_generated_messages_require_ros_dds_type_adapters(transport_type):
+    with pytest.raises(TypeError, match="type adapter"):
+        transport_type("generated", Int32)
+
+
+@pytest.mark.parametrize("config", [LCM, ZENOH])
+def test_factory_does_not_pickle_unregistered_cdr_types(config):
+    class Unregistered:
+        __msgtype__ = "test_msgs/msg/Unregistered"
+
+    with pytest.raises(TypeError):
+        make_transport("invalid", Unregistered, g=config)

dimos/core/test_zenoh_transport.py

@@ -48,27 +48,19 @@
 from dimos.protocol.service.zenohservice import ZenohSessionPool
 
 
-class TypedMsg:
-    """A fake typed message with lcm_encode for testing."""
-
-    @staticmethod
-    def lcm_encode() -> bytes:
-        return b""
-
-
 class UntypedMsg:
     """A message without lcm_encode. Triggers pickle transport."""
 
     pass
 
 
 class ProducerModule(Module):
-    typed_data: Out[TypedMsg]
+    typed_data: Out[Image]
     untyped_data: Out[UntypedMsg]
 
 
 class ConsumerModule(Module):
-    typed_data: In[TypedMsg]
+    typed_data: In[Image]
     untyped_data: In[UntypedMsg]
 
 
@@ -147,7 +139,7 @@ def test_invalid_transport_is_rejected_on_update() -> None:
 
 
 def test_lcm_transport_returned_when_transport_is_lcm(use_lcm, blueprint) -> None:
-    transport = _get_transport_for(blueprint, "typed_data", TypedMsg)
+    transport = _get_transport_for(blueprint, "typed_data", Image)
     assert isinstance(transport, LCMTransport)
 
 
@@ -157,7 +149,7 @@ def test_lcm_pickle_transport_returned_for_untyped_when_lcm(use_lcm, blueprint)
 
 
 def test_zenoh_transport_returned_when_transport_is_zenoh(use_zenoh, blueprint) -> None:
-    transport = _get_transport_for(blueprint, "typed_data", TypedMsg)
+    transport = _get_transport_for(blueprint, "typed_data", Image)
     assert isinstance(transport, ZenohTransport)
 
 

dimos/core/transport.py

@@ -29,6 +29,7 @@
 
 from dimos.core.stream import Stream, Transport
 from dimos.msgs.protocol import DimosMsg
+from dimos.protocol.pubsub.codec import describe_message
 from dimos.protocol.pubsub.impl.lcmpubsub import LCM, PickleLCM, Topic as LCMTopic
 from dimos.protocol.pubsub.impl.webrtc.providers.broker import BrokerConfig
 from dimos.protocol.pubsub.impl.webrtc.providers.spec import AudioProvider, ProviderConfig
@@ -198,6 +199,11 @@ class ROSTransport(PubSubTransport[DimosMsg]):
     _ros: DimosROS | None = None
 
     def __init__(self, topic: str, msg_type: type[DimosMsg], **kwargs: Any) -> None:
+        # TODO: Revisit generated-message adapters; ROS also uses CDR encoding.
+        if hasattr(msg_type, "__msgtype__") and describe_message(msg_type).encoding == "cdr":
+            raise TypeError(
+                "Generated messages require a ROS type adapter; raw CDR is not a ROS message"
+            )
         # rclpy (about 0.1 s with rospubsub): only loaded when a ROSTransport exists.
         from dimos.protocol.pubsub.impl.rospubsub import ROSTopic
 
@@ -241,6 +247,11 @@ def stop(self) -> None:
 
 class DDSTransport(PubSubTransport[T]):
     def __init__(self, topic: str, type: type, **kwargs) -> None:  # type: ignore[no-untyped-def]
+        # TODO: Revisit generated-message adapters; DDS also uses CDR encoding.
+        if hasattr(type, "__msgtype__") and describe_message(type).encoding == "cdr":
+            raise TypeError(
+                "Generated messages require a DDS type adapter; raw CDR is not a DDS type"
+            )
         # cyclonedds: see DDS_AVAILABLE.
         from dimos.protocol.pubsub.impl.ddspubsub import DDS, Topic as DDSTopic
 
@@ -280,7 +291,7 @@ def subscribe(
         return self.dds.subscribe(self.topic, lambda msg, topic: callback(msg))
 
 
-M = TypeVar("M", bound=DimosMsg)
+M = TypeVar("M")
 
 
 def _rebuild_webrtc_transport(
@@ -296,10 +307,9 @@ class WebRTCTransport(PubSubTransport[M]):
     also be used directly with an explicit ``config``. Two modes:
 
     * **Raw bytes** (``msg_type=None``): messages pass through as ``bytes``.
-    * **Typed LCM** (``msg_type=SomeMsg``): LCM-encoded on ``broadcast()``,
-      LCM-decoded on ``subscribe()`` (foreign types on the shared channel are skipped) — so multiple
-      transports sharing one multiplexed DataChannel each receive only
-      their own message type.
+    * **Typed messages**: bind a shared codec from ``msg_type``. Generated
+      messages use raw CDR; legacy messages keep their LCM wire format.
+      Peers must agree on the schema; remote schema validation is deferred.
 
     The transport itself holds no connection: the picklable ``config``
     resolves to a per-process singleton provider on first use, so transports
@@ -322,6 +332,7 @@ def __init__(
     ) -> None:
         super().__init__(topic)
         self._msg_type = msg_type
+        self._message = describe_message(msg_type) if msg_type is not None else None
         self._config = config or self._config_cls(**config_kwargs)
         self._pubsub: WebRTCPubSub | None = None
         # Guards first-use init: concurrent subscribe()/broadcast() must not
@@ -337,7 +348,7 @@ def broadcast(self, _: Stream[M] | None, msg: M) -> None:
         if not self._started:
             self.start()
         assert self._pubsub is not None
-        data = msg.lcm_encode() if self._msg_type is not None else msg
+        data = self._message.codec.encode(msg) if self._message is not None else cast("bytes", msg)
         self._pubsub.publish(self.topic, data)  # type: ignore[arg-type]
 
     def subscribe(
@@ -347,18 +358,15 @@ def subscribe(
             self.start()
         assert self._pubsub is not None
 
-        if self._msg_type is not None:
-            msg_type = self._msg_type
+        if self._message is not None:
+            descriptor = self._message
 
             def _typed_cb(data: bytes, _topic: str) -> None:
-                # The channel is multiplexed (e.g. the browser sends Twists and
-                # Poses on cmd_unreliable); lcm_decode verifies the wire
-                # fingerprint and raises on other types — skip those.
                 try:
-                    msg = msg_type.lcm_decode(data)
+                    msg = descriptor.codec.decode(data)
                 except ValueError:
                     return
-                callback(msg)  # type: ignore[arg-type]
+                callback(msg)
 
             return self._pubsub.subscribe(self.topic, _typed_cb)
         return self._pubsub.subscribe(self.topic, lambda msg, _topic: callback(msg))  # type: ignore[arg-type]
@@ -503,7 +511,7 @@ class CloudflareAudioTransport(WebRTCAudioTransport):
 
 
 class ZenohTransport(PubSubTransport[T]):
-    """Zenoh transport with LCM encoding for typed DimosMsg.
+    """Zenoh transport with a codec bound to the declared message type.
 
     Accepts either a plain topic string plus message type, or a full
     `ZenohTopic` carrying per-topic settings: `ZenohTransport(ZenohTopic("bla",
@@ -546,7 +554,7 @@ def stop(self) -> None:
     def broadcast(self, _: Stream[T] | None, msg: T) -> None:
         if not self._started:
             self.start()
-        self.zenoh.publish(self.topic, cast("DimosMsg", msg))
+        self.zenoh.publish(self.topic, msg)
 
     def subscribe(
         self, callback: Callable[[T], None], selfstream: Stream[T] | None = None

dimos/core/transport_factory.py

@@ -27,6 +27,7 @@
     pLCMTransport,
     pZenohTransport,
 )
+from dimos.protocol.pubsub.codec import describe_message
 from dimos.protocol.pubsub.impl.zenohpubsub import (
     QOS_LATEST_WINS,
     QOS_NEVER_DROP,
@@ -91,16 +92,19 @@ 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", ""))
+    type_name = getattr(msg_type, "msg_name", "")
+    if msg_type is not None and hasattr(msg_type, "__msgtype__"):
+        type_name = describe_message(msg_type).name.replace("/msg/", ".")
+    return default_zenoh_qos_for(name, type_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.
+    Generated messages use CDR and legacy messages keep LCM encoding.
+    Other Python types and undeclared types preserve the existing pickle transport.
 
     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,7 +114,11 @@ 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
+    use_pickled = msg_type is None or (
+        not hasattr(msg_type, "__msgtype__") and getattr(msg_type, "lcm_encode", None) is None
+    )
+    if msg_type is not None and not use_pickled:
+        describe_message(msg_type)
     topic = transport_topic(name, g)
     if g.transport == "zenoh":
         ztopic = ZenohTopic(