feat(transport): WebRTC DataChannel transport (cloudflare SFU)

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 | Transport[Any]] = field(
         default_factory=lambda: MappingProxyType({})
     )
     global_config_overrides: Mapping[str, Any] = field(default_factory=lambda: MappingProxyType({}))
@@ -185,9 +207,26 @@ 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():
+            # Raw transport instances (plain `LCMTransport(...)` pins) have no
+            # config override surface — only deferred specs participate.
+            cls = spec.config_cls if isinstance(spec, TransportSpec) else None
+            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 | Transport[Any]]
+    ) -> "Blueprint":
         return replace(self, transport_map=MappingProxyType({**self.transport_map, **transports}))
 
     def global_config(self, **kwargs: Any) -> "Blueprint":
@@ -219,6 +258,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 TransportSpec, 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,
@@ -72,9 +74,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
@@ -256,7 +258,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:
@@ -270,7 +274,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) or _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]
@@ -297,6 +303,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)
@@ -307,7 +315,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()
@@ -344,6 +352,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))
@@ -368,7 +378,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]
@@ -579,27 +589,26 @@ def _is_name_unique(blueprint: Blueprint, name: str) -> bool:
 
 
 def _get_transport_for(blueprint: Blueprint, name: str, stream_type: type) -> PubSubTransport[Any]:
-    mapped = blueprint.transport_map.get((name, stream_type), None)
-    if mapped is not None:
-        return _coerce_transport_to_backend(mapped)
-
     topic = f"/{name}" if _is_name_unique(blueprint, name) else f"/{short_id()}"
     return make_transport(topic, stream_type)
 
 
-def _coerce_transport_to_backend(transport: PubSubTransport[Any]) -> PubSubTransport[Any]:
+def _coerce_transport_to_backend(transport: Transport[Any]) -> Transport[Any]:
     """Rebuild an explicitly-mapped LCM/Zenoh transport for the active backend.
 
-    Blueprints pin specific channels in their `transport_map` with e.g. `LCMTransport("/cmd_vel",
-    Twist)`. So the global transport switch reaches those too, rebuild the plain LCM<->Zenoh pair
-    via the factory when it doesn't match `global_config.transport`. Deliberate non-default choices
-    (`JpegLcmTransport`, SHM, ROS, DDS, ...) are exact-type-checked out and left untouched.
+    Blueprints pin specific channels in their `transport_map` with e.g. `LCMTransport.spec(
+    "/cmd_vel", Twist)`. So the global transport switch reaches those too, rebuild the plain
+    LCM<->Zenoh pair via the factory when it doesn't match `global_config.transport`. Deliberate
+    non-default choices (`JpegLcmTransport`, SHM, ROS, DDS, WebRTC, ...) are exact-type-checked
+    out and left untouched.
     """
     want = global_config.transport
     is_pickled = type(transport) in (pLCMTransport, pZenohTransport)
     is_lcm = type(transport) in (LCMTransport, pLCMTransport)
     is_zenoh = type(transport) in (ZenohTransport, pZenohTransport)
-    if not ((want == "zenoh" and is_lcm) or (want == "lcm" and is_zenoh)):
+    if not isinstance(transport, PubSubTransport) or not (
+        (want == "zenoh" and is_lcm) or (want == "lcm" and is_zenoh)
+    ):
         return transport
 
     if is_pickled:
@@ -612,6 +621,32 @@ def _coerce_transport_to_backend(transport: PubSubTransport[Any]) -> PubSubTrans
     return make_transport(logical, msg_type)
 
 
+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, then gets coerced to the active pubsub backend. Returns
+    ready-to-use instances pickled into module workers.
+    """
+    materialized: dict[tuple[str, type], Transport[Any]] = {}
+    for key, spec in blueprint.transport_map.items():
+        if not isinstance(spec, TransportSpec):
+            # Plain transport instance pinned directly in the blueprint — use
+            # as-is (modulo the global lcm/zenoh backend switch).
+            materialized[key] = _coerce_transport_to_backend(spec)
+            continue
+        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] = _coerce_transport_to_backend(spec.build(config=config))
+    return materialized
+
+
 def _verify_no_name_conflicts(blueprint: Blueprint) -> None:
     name_to_types: dict[Any, set[type]] = defaultdict(set)
     name_to_modules: dict[Any, list[tuple[type, type]]] = defaultdict(list)
@@ -650,7 +685,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:

dimos/core/stream.py

@@ -22,6 +22,7 @@
     TypeVar,
 )
 
+from pydantic import BaseModel
 import reactivex as rx
 from reactivex import operators as ops
 from reactivex.disposable import Disposable
@@ -80,6 +81,11 @@ class State(enum.Enum):
 
 
 class Transport(Resource, ObservableMixin[T]):
+    # Transports that expose a pydantic config override surface set this to the
+    # config class; the blueprint config flow picks them up automatically. None
+    # means "no overridable config" (LCM/SHM transports).
+    _config_cls: type[BaseModel] | None = None
+
     # used by local Output; selfstream is None when publishing without a source stream
     def broadcast(self, selfstream: Out[T] | None, value: T) -> None:
         raise NotImplementedError