feat(sim): add external simulation provider contract

dimos/core/coordination/blueprint_config/parser.py

@@ -75,6 +75,7 @@
     plain,
     plain_mapping,
     snapshot_mapping,
+    validated_model_values,
 )
 from dimos.core.coordination.blueprints import (
     Blueprint,
@@ -421,7 +422,7 @@ def _validate_modules(
                 raise BlueprintConfigError(
                     format_validation_error(module.atom.name, error)
                 ) from error
-            dumped = model.model_dump(mode="python", exclude_unset=True)
+            dumped = validated_model_values(model)
             dumped.pop("g", None)
             dumped.pop("instance_name", None)
             parsed[module.atom.name] = dumped

dimos/core/coordination/blueprint_config/test_parser.py

@@ -13,7 +13,9 @@
 # limitations under the License.
 
 from collections.abc import Callable
+from dataclasses import dataclass
 from pathlib import Path
+import pickle
 from typing import Annotated, Any, Literal
 
 from pydantic import BaseModel, Field
@@ -452,6 +454,14 @@ def __str__(self) -> str:
         return "Anchor:\n  multi\n  line"
 
 
+@dataclass(frozen=True)
+class CallableAnchor:
+    prefix: str
+
+    def __call__(self, value: Any) -> str:
+        return f"{self.prefix}:{value}"
+
+
 class ArbitraryConfig(ModuleConfig):
     scaling: Anchor = Field(default_factory=Anchor)
     hybrid: Anchor | str = "fallback"
@@ -490,6 +500,18 @@ def test_blueprint_pinned_arbitrary_value_survives_filtering() -> None:
     assert isinstance(parsed.module_kwargs("arbitrarymodule")["scaling"], Anchor)
 
 
+def test_blueprint_pinned_callable_dataclass_survives_worker_serialization() -> None:
+    blueprint = ArbitraryModule.blueprint(handlers={"scene": CallableAnchor("render")})
+
+    parsed = BlueprintConfigParser(blueprint).parse(environ={})
+    worker_kwargs = pickle.loads(pickle.dumps(parsed.module_kwargs("arbitrarymodule")))
+    worker_config = ArbitraryConfig.model_validate(worker_kwargs)
+    handler = worker_config.handlers["scene"]
+
+    assert isinstance(handler, CallableAnchor)
+    assert handler("apartment") == "render:apartment"
+
+
 def test_format_help_uses_nested_parent_default_instance() -> None:
     class NestedRequiredConfig(BaseModel):
         value: int

dimos/core/coordination/blueprint_config/values.py

@@ -55,6 +55,31 @@ def plain(value: Any) -> Any:
     return _copy_opaque(value)
 
 
+def validated_model_values(model: BaseModel) -> dict[str, Any]:
+    """Copy explicitly set validated fields without serializing runtime objects."""
+    return {
+        name: _validated_value(getattr(model, name))
+        for name in type(model).model_fields
+        if name in model.model_fields_set
+    }
+
+
+def _validated_value(value: Any) -> Any:
+    if isinstance(value, BaseModel):
+        return validated_model_values(value)
+    if isinstance(value, Mapping):
+        return {_copy_opaque(key): _validated_value(item) for key, item in value.items()}
+    if isinstance(value, list):
+        return [_validated_value(item) for item in value]
+    if isinstance(value, tuple):
+        return tuple(_validated_value(item) for item in value)
+    if isinstance(value, set):
+        return {_validated_value(item) for item in value}
+    if isinstance(value, frozenset):
+        return frozenset(_validated_value(item) for item in value)
+    return _copy_opaque(value)
+
+
 def deep_merge(destination: dict[str, Any], incoming: Mapping[str, Any]) -> None:
     for key, value in incoming.items():
         if key in destination and isinstance(destination[key], dict) and isinstance(value, Mapping):

dimos/core/coordination/coordinator_rpc.py

@@ -19,6 +19,8 @@
 
 from dimos.core.global_config import global_config
 from dimos.core.transport_factory import rpc_backend
+from dimos.protocol.rpc.zenohrpc import ZenohRPC
+from dimos.protocol.service.zenohservice import ZENOH_LOCAL_ROUTER_ENDPOINT
 from dimos.utils.logging_config import setup_logger
 
 if TYPE_CHECKING:
@@ -51,7 +53,12 @@ def serve(cls, coordinator: RPCInspectable) -> CoordinatorRPC:
     @classmethod
     def connect(cls, *, timeout: float) -> CoordinatorRPC:
         """Attach to a running Coordinator, raising `TimeoutError` if none answers."""
-        rpc = rpc_backend()()
+        backend = rpc_backend()
+        rpc = (
+            ZenohRPC(mode="client", connect=[ZENOH_LOCAL_ROUTER_ENDPOINT])
+            if backend is ZenohRPC
+            else backend()
+        )
         rpc.start()
         client = cls(rpc)
         deadline = time.monotonic() + timeout

dimos/core/coordination/module_coordinator.py

@@ -19,6 +19,7 @@
 import dataclasses
 import importlib
 import inspect
+import os
 import shutil
 import sys
 import threading
@@ -41,6 +42,11 @@
     pZenohTransport,
 )
 from dimos.core.transport_factory import make_transport
+from dimos.protocol.service.zenohservice import (
+    ZENOH_LOCAL_ROUTER_ENDPOINT,
+    ZENOH_ROUTER_ENDPOINT_ENV,
+    ZenohRouter,
+)
 from dimos.spec.utils import is_spec, spec_annotation_compliance, spec_structural_compliance
 from dimos.utils.generic import short_id
 from dimos.utils.logging_config import setup_logger
@@ -89,13 +95,24 @@ def __init__(
         self._modules_lock = threading.RLock()
         self._rpc_lock = threading.RLock()
         self._coordinator_rpc: CoordinatorRPC | None = None
+        self._zenoh_router: ZenohRouter | None = None
+        self._previous_zenoh_router_endpoint: str | None = None
 
     def start(self) -> None:
         from dimos.core.o3dpickle import register_picklers
 
         register_picklers()
-        for m in self._managers.values():
-            m.start()
+        if self._global_config.transport == "zenoh":
+            self._zenoh_router = ZenohRouter()
+            self._zenoh_router.start()
+            self._previous_zenoh_router_endpoint = os.environ.get(ZENOH_ROUTER_ENDPOINT_ENV)
+            os.environ[ZENOH_ROUTER_ENDPOINT_ENV] = ZENOH_LOCAL_ROUTER_ENDPOINT
+        try:
+            for m in self._managers.values():
+                m.start()
+        except BaseException:
+            self._stop_zenoh_router()
+            raise
         self._started = True
 
     def stop(self) -> None:
@@ -119,9 +136,21 @@ def _stop_manager(m: WorkerManager) -> None:
                 logger.error("Error stopping manager", manager=type(m).__name__, exc_info=True)
 
         safe_thread_map(tuple(self._managers.values()), _stop_manager)
+        self._stop_zenoh_router()
+
+    def _stop_zenoh_router(self) -> None:
+        if self._zenoh_router is not None:
+            self._zenoh_router.stop()
+            self._zenoh_router = None
+        if self._global_config.transport != "zenoh":
+            return
+        if self._previous_zenoh_router_endpoint is None:
+            os.environ.pop(ZENOH_ROUTER_ENDPOINT_ENV, None)
+        else:
+            os.environ[ZENOH_ROUTER_ENDPOINT_ENV] = self._previous_zenoh_router_endpoint
 
     def start_rpc_service(self) -> None:
-        """Expose the coordinator's API as @rpc methods over LCM."""
+        """Expose the coordinator's API over the configured RPC transport."""
         with self._rpc_lock:
             if self._coordinator_rpc is not None:
                 return

dimos/core/global_config.py

@@ -53,6 +53,7 @@ class GlobalConfig(BaseSettings):
     can_port: str | None = None
     device_path: str | None = None  # device path for real robot (e.g. /dev/ttyUSB0)
     simulation: str = ""
+    simulation_provider: str = ""
     replay: bool = False
     replay_db: str = "go2_short"
     new_memory: bool = False

dimos/core/native_module.py

@@ -60,6 +60,10 @@ class MyCppModule(NativeModule):
 from dimos.core.core import rpc
 from dimos.core.global_config import global_config
 from dimos.core.module import Module, ModuleConfig
+from dimos.protocol.service.zenohservice import (
+    ZENOH_LOCAL_ROUTER_ENDPOINT,
+    ZENOH_ROUTER_ENDPOINT_ENV,
+)
 from dimos.utils.logging_config import setup_logger
 
 if sys.platform.startswith("linux"):
@@ -254,6 +258,8 @@ def start(self) -> None:
 
         # set transport so native modules know which one to spawn
         env["DIMOS_TRANSPORT"] = global_config.transport
+        if global_config.transport == "zenoh":
+            env[ZENOH_ROUTER_ENDPOINT_ENV] = ZENOH_LOCAL_ROUTER_ENDPOINT
 
         # set Rust logging to match Python level
         env["RUST_LOG"] = _PYTHON_TO_RUST_LEVELS.get(

dimos/hardware/manipulators/sim/adapter.py

@@ -27,7 +27,7 @@
     JointLimits,
     ManipulatorInfo,
 )
-from dimos.simulation.engines.mujoco_shm import (
+from dimos.hardware.simulation.shared_memory import (
     ManipShmReader,
     shm_key_from_path,
 )

dimos/hardware/manipulators/sim/test_shm_adapter.py

@@ -23,7 +23,7 @@
 import dimos.hardware.manipulators.sim.adapter as adapter_mod
 from dimos.hardware.manipulators.sim.adapter import ShmMujocoAdapter
 from dimos.hardware.manipulators.spec import ControlMode, ManipulatorAdapter
-from dimos.simulation.engines.mujoco_shm import ManipShmWriter
+from dimos.hardware.simulation.shared_memory import ManipShmWriter
 
 ARM_DOF = 7