Feat/webrtc transport with Sam config merged on top

dimos/control/blueprints/mobile.py

@@ -73,33 +73,29 @@ def _flowbase_twist_base(
 
 
 # Mock holonomic twist base (3-DOF: vx, vy, wz)
-coordinator_mock_twist_base = ControlCoordinator.blueprint(
-    hardware=[_mock_twist_base()],
-    tasks=[
-        TaskConfig(
-            name="vel_base",
-            type="velocity",
-            joint_names=_base_joints,
-            priority=10,
-        ),
-    ],
-).remappings([(ControlCoordinator, "twist_command", "cmd_vel")])
+coordinator_mock_twist_base = (
+    ControlCoordinator.blueprint(
+        hardware=[_mock_twist_base()],
+        tasks=[
+            TaskConfig(
+                name="vel_base",
+                type="velocity",
+                joint_names=_base_joints,
+                priority=10,
+            ),
+        ],
+    )
+    .transports(
+        {
+            ("joint_state", JointState): LCMTransport.spec("/coordinator/joint_state", JointState),
+            ("twist_command", Twist): LCMTransport.spec("/cmd_vel", Twist),
+        }
+    )
+    .remappings([(ControlCoordinator, "twist_command", "cmd_vel")])
+)
 
 # FlowBase holonomic twist base (3-DOF: vx, vy, wz) over Portal RPC
-coordinator_flowbase = ControlCoordinator.blueprint(
-    hardware=[_flowbase_twist_base()],
-    tasks=[
-        TaskConfig(
-            name="vel_base",
-            type="velocity",
-            joint_names=_base_joints,
-            priority=10,
-        ),
-    ],
-).remappings([(ControlCoordinator, "twist_command", "cmd_vel")])
-
-# FlowBase + WASD pygame keyboard teleop in a single blueprint
-coordinator_flowbase_keyboard_teleop = autoconnect(
+coordinator_flowbase = (
     ControlCoordinator.blueprint(
         hardware=[_flowbase_twist_base()],
         tasks=[
@@ -110,9 +106,40 @@ def _flowbase_twist_base(
                 priority=10,
             ),
         ],
-    ),
-    KeyboardTeleop.blueprint(),
-).remappings([(ControlCoordinator, "twist_command", "cmd_vel")])
+    )
+    .transports(
+        {
+            ("joint_state", JointState): LCMTransport.spec("/coordinator/joint_state", JointState),
+            ("twist_command", Twist): LCMTransport.spec("/cmd_vel", Twist),
+        }
+    )
+    .remappings([(ControlCoordinator, "twist_command", "cmd_vel")])
+)
+
+# FlowBase + WASD pygame keyboard teleop in a single blueprint
+coordinator_flowbase_keyboard_teleop = (
+    autoconnect(
+        ControlCoordinator.blueprint(
+            hardware=[_flowbase_twist_base()],
+            tasks=[
+                TaskConfig(
+                    name="vel_base",
+                    type="velocity",
+                    joint_names=_base_joints,
+                    priority=10,
+                ),
+            ],
+        ),
+        KeyboardTeleop.blueprint(),
+    )
+    .transports(
+        {
+            ("twist_command", Twist): LCMTransport.spec("/cmd_vel", Twist),
+            ("joint_state", JointState): LCMTransport.spec("/coordinator/joint_state", JointState),
+        }
+    )
+    .remappings([(ControlCoordinator, "twist_command", "cmd_vel")])
+)
 
 # FlowBase + Livox MID-360 + FastLio2 SLAM + nav stack with click-to-drive in Rerun. The velocity
 # sink is ControlCoordinator + FlowBaseAdapter
@@ -170,21 +197,26 @@ def _flowbase_twist_base(
         ),
         RerunWebSocketServer.blueprint(),
     )
+    .transports(
+        {
+            ("twist_command", Twist): LCMTransport.spec("/cmd_vel", Twist),
+            ("joint_state", JointState): LCMTransport.spec("/coordinator/joint_state", JointState),
+        }
+    )
     .remappings(
         [
             (FastLio2, "lidar", "registered_scan"),
             # SimplePlanner / FarPlanner owns way_point — disconnect MovementManager's
             # redundant pass-through copy (matches unitree-g1-nav-onboard).
             (MovementManager, "way_point", "_mgr_way_point_unused"),
             # MovementManager.cmd_vel publishes to LCM /cmd_vel by default; the
-            # coordinator's twist_command listens on the same name.
+            # coordinator's twist_command listens on the same topic.
             (ControlCoordinator, "twist_command", "cmd_vel"),
         ]
     )
     .global_config(n_workers=8)
 )
 
-
 # Mock arm (7-DOF) + mock holonomic base (3-DOF)
 _mock_arm_hw = HardwareComponent(
     hardware_id="arm",
@@ -193,20 +225,29 @@ def _flowbase_twist_base(
     adapter_type="mock",
 )
 
-coordinator_mobile_manip_mock = ControlCoordinator.blueprint(
-    hardware=[_mock_arm_hw, _mock_twist_base()],
-    tasks=[
-        TaskConfig(
-            name="traj_arm",
-            type="trajectory",
-            joint_names=_mock_arm_hw.joints,
-            priority=10,
-        ),
-        TaskConfig(
-            name="vel_base",
-            type="velocity",
-            joint_names=_base_joints,
-            priority=10,
-        ),
-    ],
-).remappings([(ControlCoordinator, "twist_command", "cmd_vel")])
+coordinator_mobile_manip_mock = (
+    ControlCoordinator.blueprint(
+        hardware=[_mock_arm_hw, _mock_twist_base()],
+        tasks=[
+            TaskConfig(
+                name="traj_arm",
+                type="trajectory",
+                joint_names=_mock_arm_hw.joints,
+                priority=10,
+            ),
+            TaskConfig(
+                name="vel_base",
+                type="velocity",
+                joint_names=_base_joints,
+                priority=10,
+            ),
+        ],
+    )
+    .transports(
+        {
+            ("joint_state", JointState): LCMTransport.spec("/coordinator/joint_state", JointState),
+            ("twist_command", Twist): LCMTransport.spec("/cmd_vel", Twist),
+        }
+    )
+    .remappings([(ControlCoordinator, "twist_command", "cmd_vel")])
+)

dimos/core/coordination/blueprints.py

@@ -21,15 +21,14 @@
 from types import MappingProxyType
 from typing import TYPE_CHECKING, Any, Literal, Union, get_args, get_origin, get_type_hints
 
-from pydantic import create_model
+from pydantic import BaseModel, create_model
 
 if TYPE_CHECKING:
     from dimos.protocol.service.system_configurator.base import SystemConfigurator
 
 from dimos.core.global_config import GlobalConfig
 from dimos.core.module import ModuleBase, is_module_type
-from dimos.core.stream import In, Out
-from dimos.core.transport import PubSubTransport
+from dimos.core.stream import In, Out, Transport
 from dimos.spec.utils import Spec, is_spec
 from dimos.utils.logging_config import setup_logger
 
@@ -141,6 +140,29 @@ def create(cls, module: type[ModuleBase], kwargs: dict[str, Any]) -> Self:
         )
 
 
+@dataclass(frozen=True)
+class TransportSpec:
+    """Deferred transport construction: a transport class plus its ctor args.
+
+    Blueprint authors declare transports via ``Cls.spec(...)`` so nothing is
+    constructed at blueprint-definition time. The coordinator materializes
+    specs at build time, once CLI/env/config overrides have resolved.
+    """
+
+    cls: type[Transport[Any]]
+    args: tuple[Any, ...] = ()
+    kwargs: Mapping[str, Any] = field(default_factory=dict)
+
+    @property
+    def config_cls(self) -> type[BaseModel] | None:
+        # Set by transports that expose a pydantic config override surface
+        return self.cls._config_cls
+
+    def build(self, config: BaseModel | None = None) -> Transport[Any]:
+        extra = {"config": config} if config is not None else {}
+        return self.cls(*self.args, **self.kwargs, **extra)
+
+
 # These fields cannot be pickled.
 _PROXY_FIELDS = ("transport_map", "global_config_overrides", "remapping_map")
 
@@ -149,7 +171,7 @@ def create(cls, module: type[ModuleBase], kwargs: dict[str, Any]) -> Self:
 class Blueprint:
     blueprints: tuple[BlueprintAtom, ...]
     disabled_modules_tuple: tuple[type[ModuleBase], ...] = field(default_factory=tuple)
-    transport_map: Mapping[tuple[str, type], PubSubTransport[Any]] = field(
+    transport_map: Mapping[tuple[str, type], TransportSpec] = field(
         default_factory=lambda: MappingProxyType({})
     )
     global_config_overrides: Mapping[str, Any] = field(default_factory=lambda: MappingProxyType({}))
@@ -185,9 +207,22 @@ def config(self) -> type:
             for b in self.blueprints
         }
         configs["g"] = (GlobalConfig | None, None)
+        transport_fields: dict[str, Any] = {}
+        seen: set[type] = set()
+        for spec in self.transport_map.values():
+            cls = spec.config_cls
+            if cls is None or cls in seen:
+                continue
+            seen.add(cls)
+            transport_fields[transport_config_name(cls)] = (cls | None, None)
+        if transport_fields:
+            transports_model = create_model(
+                "TransportsConfig", __config__={"extra": "forbid"}, **transport_fields
+            )
+            configs["transports"] = (transports_model | None, None)
         return create_model("BlueprintConfig", __config__={"extra": "forbid"}, **configs)  # type: ignore[call-overload,no-any-return]
 
-    def transports(self, transports: dict[tuple[str, type], Any]) -> "Blueprint":
+    def transports(self, transports: dict[tuple[str, type], TransportSpec]) -> "Blueprint":
         return replace(self, transport_map=MappingProxyType({**self.transport_map, **transports}))
 
     def global_config(self, **kwargs: Any) -> "Blueprint":
@@ -219,6 +254,10 @@ def active_blueprints(self) -> tuple[BlueprintAtom, ...]:
         return tuple(bp for bp in self.blueprints if bp.module not in disabled)
 
 
+def transport_config_name(cls: type) -> str:
+    return cls.__name__.removesuffix("Config").lower()
+
+
 def autoconnect(*blueprints: Blueprint) -> Blueprint:
     all_blueprints = tuple(_eliminate_duplicates([bp for bs in blueprints for bp in bs.blueprints]))
     all_transports = dict(  # type: ignore[var-annotated]

dimos/core/coordination/module_coordinator.py

@@ -23,12 +23,14 @@
 import threading
 from typing import TYPE_CHECKING, Any, NamedTuple, cast
 
+from dimos.core.coordination.blueprints import transport_config_name
 from dimos.core.coordination.coordinator_rpc import CoordinatorRPC
 from dimos.core.coordination.worker_manager import WorkerManager
 from dimos.core.coordination.worker_manager_python import WorkerManagerPython
 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.stream import Transport
 from dimos.core.transport import LCMTransport, PubSubTransport, pLCMTransport
 from dimos.spec.utils import is_spec, spec_annotation_compliance, spec_structural_compliance
 from dimos.utils.generic import short_id
@@ -65,9 +67,9 @@ def __init__(
         self._deployed_modules = {}
         self._deployed_atoms: dict[type[ModuleBase], BlueprintAtom] = {}
         self._resolved_module_refs: dict[tuple[type[ModuleBase], str], type[ModuleBase]] = {}
-        self._transport_registry: dict[tuple[str, type], PubSubTransport[Any]] = {}
+        self._transport_registry: dict[tuple[str, type], Transport[Any]] = {}
         self._class_aliases: dict[type[ModuleBase], type[ModuleBase]] = {}
-        self._module_transports: dict[type[ModuleBase], dict[str, PubSubTransport[Any]]] = {}
+        self._module_transports: dict[type[ModuleBase], dict[str, Transport[Any]]] = {}
         self._started = False
         self._modules_lock = threading.RLock()
         self._coordinator_rpc: CoordinatorRPC | None = None
@@ -249,7 +251,9 @@ def _send_on_system_modules(self) -> None:
             if hasattr(module, "on_system_modules"):
                 module.on_system_modules(modules)
 
-    def _connect_streams(self, blueprint: Blueprint) -> None:
+    def _connect_streams(
+        self, blueprint: Blueprint, transports: Mapping[tuple[str, type], Transport[Any]]
+    ) -> None:
         streams: dict[tuple[str, type], list[tuple[type, str]]] = defaultdict(list)
 
         for bp in blueprint.active_blueprints:
@@ -263,7 +267,9 @@ def _connect_streams(self, blueprint: Blueprint) -> None:
             if key in self._transport_registry:
                 transport = self._transport_registry[key]
             else:
-                transport = _get_transport_for(blueprint, remapped_name, stream_type)
+                transport = transports.get(key)
+                if transport is None:
+                    transport = _get_transport_for(blueprint, remapped_name, stream_type)
             self._transport_registry[key] = transport
             for module, original_name in streams[key]:
                 instance = self.get_instance(module)  # type: ignore[assignment]
@@ -290,6 +296,8 @@ def build(
         blueprint_args = blueprint_args or {}
         if "g" in blueprint_args:
             global_config.update(**blueprint_args.pop("g"))
+        transport_overrides = blueprint_args.pop("transports", None) or {}
+        transports = _materialize_transports(blueprint, transport_overrides)
 
         _run_configurators(blueprint)
         _check_requirements(blueprint)
@@ -300,7 +308,7 @@ def build(
         coordinator.start()
 
         _deploy_all_modules(blueprint, coordinator, global_config, blueprint_args)
-        coordinator._connect_streams(blueprint)
+        coordinator._connect_streams(blueprint, transports)
         _connect_module_refs(blueprint, coordinator)
 
         coordinator.build_all_modules()
@@ -337,6 +345,8 @@ def _load_blueprint(
         blueprint_args = blueprint_args or {}
         if "g" in blueprint_args:
             self._global_config.update(**blueprint_args.pop("g"))
+        transport_overrides = blueprint_args.pop("transports", None) or {}
+        transports = _materialize_transports(blueprint, transport_overrides)
 
         # Scale worker pool.
         n_extra = int(blueprint.global_config_overrides.get("n_workers", 0))
@@ -361,7 +371,7 @@ def _load_blueprint(
         before = set(self._deployed_modules)
 
         _deploy_all_modules(blueprint, self, self._global_config, blueprint_args)
-        self._connect_streams(blueprint)
+        self._connect_streams(blueprint, transports)
         _connect_module_refs(blueprint, self, existing_modules=before)
 
         new_modules = [proxy for cls, proxy in self._deployed_modules.items() if cls not in before]
@@ -572,15 +582,29 @@ def _is_name_unique(blueprint: Blueprint, name: str) -> bool:
 
 
 def _get_transport_for(blueprint: Blueprint, name: str, stream_type: type) -> PubSubTransport[Any]:
-    transport = blueprint.transport_map.get((name, stream_type), None)
-    if transport:
-        return transport
-
     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)
+    return pLCMTransport(topic) if use_pickled else LCMTransport(topic, stream_type)
+
 
-    return transport
+def _materialize_transports(
+    blueprint: Blueprint, overrides: Mapping[str, Mapping[str, Any]]
+) -> dict[tuple[str, type], Transport[Any]]:
+    """Build the blueprint's declared transports, merging CLI/env config overrides.
+
+    WebRTC transports get a freshly constructed provider config from the
+    resolved ``transports.<name>.*`` overrides; everything else builds from the
+    spec as-is. Returns ready-to-use instances pickled into module workers.
+    """
+    materialized: dict[tuple[str, type], Transport[Any]] = {}
+    for key, spec in blueprint.transport_map.items():
+        config = None
+        config_cls = spec.config_cls
+        if config_cls is not None:
+            sub = overrides.get(transport_config_name(config_cls), {})
+            config = config_cls(**sub)
+        materialized[key] = spec.build(config=config)
+    return materialized
 
 
 def _verify_no_name_conflicts(blueprint: Blueprint) -> None:
@@ -621,7 +645,7 @@ def _verify_no_name_conflicts(blueprint: Blueprint) -> None:
 
 def _verify_no_conflicts_with_existing(
     blueprint: Blueprint,
-    existing_registry: dict[tuple[str, type], PubSubTransport[Any]],
+    existing_registry: dict[tuple[str, type], Transport[Any]],
 ) -> None:
     """Check that a new blueprint's streams don't conflict with already-registered transports."""
     if not existing_registry: