JUST TRYING THING OUT refactor(rpc): replace pickle with CBOR in RPC and the worker pipe

AGENTS.md

@@ -321,6 +321,8 @@ class MySkillContainer(Module):
 
 If multiple modules match the spec, use `.remappings()` to resolve. Source: `dimos/spec/utils.py`, `dimos/core/coordination/blueprints.py`.
 
+RPC arguments, results and worker-pipe messages are CBOR (`dimos/protocol/encode/cbor.py`), not pickle. JSON-like values, tuples, sets, numpy arrays, enums, dataclasses (their fields only: state that `__post_init__` derives stays behind), pydantic models (private attributes stay behind), `Path`, message types with `lcm_encode`, classes and module-level functions cross as themselves. Anything else raises `TypeError` at the call site; give such a class `cbor_encode()` and a `cbor_decode()` classmethod.
+
 ### Adding a New Skill
 
 1. Pick the right container (robot-specific or `dimos/agents/skills/`).

dimos/agents/mcp/mcp_client.py

@@ -137,9 +137,6 @@ def __init__(self, **kwargs: Any) -> None:
         self._mcp_url: str | None = None
         Thread(target=_preload_agent_deps, name="McpClient-preload", daemon=True).start()
 
-    def __reduce__(self) -> Any:
-        return (self.__class__, (), {})
-
     def _server_url(self) -> str:
         """The MCP endpoint: configured, else the blueprint's McpServer, else the default."""
         if self._mcp_url is None:

dimos/agents/skills/demo_calculator_skill.py

@@ -13,13 +13,16 @@
 # limitations under the License.
 
 from dimos.agents.annotation import skill
+from dimos.core.core import rpc
 from dimos.core.module import Module
 
 
 class DemoCalculatorSkill(Module):
+    @rpc
     def start(self) -> None:
         super().start()
 
+    @rpc
     def stop(self) -> None:
         super().stop()
 

dimos/agents/skills/demo_robot.py

@@ -15,6 +15,7 @@
 
 from reactivex import interval
 
+from dimos.core.core import rpc
 from dimos.core.module import Module
 from dimos.core.stream import Out
 from dimos.mapping.models import LatLon
@@ -23,10 +24,12 @@
 class DemoRobot(Module):
     gps_location: Out[LatLon]
 
+    @rpc
     def start(self) -> None:
         super().start()
         self.register_disposable(interval(1.0).subscribe(lambda _: self._publish_gps_location()))
 
+    @rpc
     def stop(self) -> None:
         super().stop()
 

dimos/agents/skills/osm.py

@@ -18,6 +18,7 @@
 from reactivex.disposable import Disposable
 
 from dimos.agents.annotation import skill
+from dimos.core.core import rpc
 from dimos.core.module import Module
 from dimos.core.stream import In
 from dimos.mapping.models import LatLon
@@ -40,6 +41,7 @@ def __init__(self, **kwargs: Any) -> None:
         self._latest_location = None
         self._current_location_map = CurrentLocationMap(QwenVlModel())
 
+    @rpc
     def start(self) -> None:
         super().start()
         if hasattr(self.gps_location, "subscribe"):
@@ -49,6 +51,7 @@ def start(self) -> None:
                 "OsmSkill: gps_location stream does not support direct subscribe (RemoteIn)"
             )
 
+    @rpc
     def stop(self) -> None:
         super().stop()
 

dimos/agents/voice_input.py

@@ -92,9 +92,6 @@ def __init__(self, **kwargs: Any) -> None:
         # and start() waits only for whatever is left.
         self._pipeline = run_in_thread(whisper_pipeline, "whisper-pipeline")
 
-    def __reduce__(self) -> Any:
-        return (self.__class__, (), {})
-
     @rpc
     def start(self) -> None:
         requirement_error = ffmpeg_requirement()

dimos/codebase_checks/test_rpc_overrides.py

@@ -0,0 +1,73 @@
+# 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.
+
+"""An override of an @rpc method must carry the decorator itself.
+
+``Module.rpcs`` reads the ``__rpc__`` marker off each attribute, so a plain override
+hides the base method from the module's proxy. The host then fetches it with a
+GetAttrRequest, and the worker cannot send a bound method back: ``start()`` on such a
+module fails at deploy time.
+"""
+
+import pytest
+
+from dimos.core.global_config import global_config
+from dimos.core.module import ModuleBase
+from dimos.robot.all_blueprints import all_blueprints
+from dimos.robot.get_all_blueprints import OptionalDependencyError, load_blueprint
+from dimos.robot.test_all_blueprints import SELF_HOSTED_BLUEPRINTS
+
+
+def _plain_overrides(module: type[ModuleBase]) -> list[str]:
+    """Public methods without @rpc that some base class declares with it."""
+    return sorted(
+        name
+        for name in dir(module)
+        if not name.startswith("_")
+        and callable(getattr(module, name, None))
+        and not hasattr(getattr(module, name), "__rpc__")
+        and any(hasattr(vars(base).get(name), "__rpc__") for base in module.__mro__)
+    )
+
+
+@pytest.mark.parametrize(
+    "blueprint_name",
+    [
+        pytest.param(name, marks=pytest.mark.self_hosted)
+        if name in SELF_HOSTED_BLUEPRINTS
+        else name
+        for name in sorted(all_blueprints)
+    ],
+)
+def test_rpc_overrides_keep_the_decorator(
+    blueprint_name: str, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    # The multi-robot blueprints read ROBOT_IPS at import time.
+    monkeypatch.setattr(global_config, "robot_ips", "192.0.2.10,192.0.2.11")
+    try:
+        blueprint = load_blueprint(blueprint_name)
+    except OptionalDependencyError as e:
+        pytest.skip(f"Skipping due to missing optional dependency: {e}")
+
+    violations = sorted(
+        {
+            f"{atom.module.__module__}.{atom.module.__name__}.{name}"
+            for atom in blueprint.blueprints
+            for name in _plain_overrides(atom.module)
+        }
+    )
+    assert not violations, (
+        f"Blueprint {blueprint_name!r} has @rpc methods overridden without the decorator; "
+        f"add @rpc to: {violations}"
+    )

dimos/control/connection/test_connection_module.py

@@ -22,7 +22,6 @@
 
 from collections.abc import Callable, Iterator
 import math
-import pickle
 import threading
 import time
 from typing import Any
@@ -33,6 +32,7 @@
 from dimos.control.connection.connection_module import ConnectionStatus
 from dimos.control.contract.limits import Limits
 from dimos.msgs.sensor_msgs.JointState import JointState
+from dimos.protocol.encode.cbor import decode, encode
 
 JOINT1 = "mock/joint1/position"
 DEADMAN_S = 0.05
@@ -300,7 +300,7 @@ def test_status_is_an_rpc_that_crosses_processes(rig: Callable[..., Rig]) -> Non
     r = rig()
     status = r.module.status()
     assert isinstance(status, ConnectionStatus)
-    assert pickle.loads(pickle.dumps(status)) == status
+    assert decode(encode(status)) == status
     assert len(status.session_id) == 32
     assert status.connected
     assert status.last_command_time is None

dimos/control/contract/keys.py

@@ -133,6 +133,13 @@ def __new__(cls, value: str) -> Key:
         key._source, key._resource, key._interface = parts
         return key
 
+    def cbor_encode(self) -> str:
+        return str(self)
+
+    @classmethod
+    def cbor_decode(cls, value: str) -> Key:
+        return cls(value)
+
     @classmethod
     def of(cls, source: str, resource: str, interface: str) -> Key:
         """Build a name from its three parts.

dimos/control/contract/test_keys.py

@@ -17,11 +17,10 @@
 
 from __future__ import annotations
 
-import pickle
-
 import pytest
 
 from dimos.control.contract.keys import Interface, Key, is_valid_segment
+from dimos.protocol.encode.cbor import decode, encode
 
 
 @pytest.mark.parametrize("segment", ["arm", "joint1", "left_hip_pitch", "A", "_", "9", "a_1_B"])
@@ -107,9 +106,9 @@ def test_error_quotes_the_key() -> None:
         Key("arm/joint1")
 
 
-def test_keys_pickle_with_their_parts() -> None:
+def test_keys_cross_processes_with_their_parts() -> None:
     """Descriptions cross a process boundary, so the parts must survive."""
-    restored = pickle.loads(pickle.dumps(Key.of("arm", "joint1", Interface.POSITION)))
+    restored = decode(encode(Key.of("arm", "joint1", Interface.POSITION)))
 
     assert restored == "arm/joint1/position"
     assert restored.source == "arm"

dimos/control/contract/test_limits.py

@@ -24,7 +24,6 @@
 from collections.abc import Sequence
 import math
 from pathlib import Path
-import pickle
 from typing import Any
 
 import pytest
@@ -40,6 +39,7 @@
 )
 from dimos.msgs.geometry_msgs.Twist import Twist
 from dimos.msgs.geometry_msgs.Vector3 import Vector3
+from dimos.protocol.encode.cbor import decode, encode
 from dimos.robot.assets.model import JointDescription, RobotModel
 
 JOINTS = ("j1", "j2", "j3")
@@ -264,7 +264,7 @@ def test_a_config_that_commands_nothing_is_refused(model: RobotModel) -> None:
 
 def test_the_table_crosses_processes(model: RobotModel) -> None:
     table = limits(config(model=model, pd=True, base_range=RANGE))
-    assert pickle.loads(pickle.dumps(table)) == table
+    assert decode(encode(table)) == table
 
 
 def test_limits_refuse_unless_told_to_clamp() -> None:

dimos/control/contract/test_sequence.py

@@ -16,11 +16,10 @@
 
 from __future__ import annotations
 
-import pickle
-
 import pytest
 
 from dimos.control.contract.sequence import Freshness, is_newer
+from dimos.protocol.encode.cbor import decode, encode
 
 
 def test_anything_is_newer_than_nothing() -> None:
@@ -78,9 +77,9 @@ def test_marking_again_refreshes() -> None:
     assert not fresh.is_stale(now=20.0, timeout_s=1.0)
 
 
-def test_freshness_pickles() -> None:
+def test_freshness_crosses_processes() -> None:
     """Freshness travels with a source snapshot over RPC."""
     fresh = Freshness()
     fresh.mark(42.0)
 
-    assert pickle.loads(pickle.dumps(fresh)).last_receipt == 42.0
+    assert decode(encode(fresh)).last_receipt == 42.0

dimos/core/baked_host.py

@@ -128,7 +128,7 @@ def baked_host(
 ) -> type[BakedHost]:
     """Build the NativeModule subclass that drives a baked host binary.
 
-    Assign the result to `name` at module level: deploying to a worker pickles
+    Assign the result to `name` at module level: deploying to a worker sends
     the class by that path.
     """
     if not members:
@@ -160,6 +160,6 @@ def baked_host(
         "_remaps": remaps,
     }
     host_cls = type(name, (BakedHost,), namespace)
-    # Nothing names the config class, so put it where pickle will look for it.
+    # Nothing names the config class, so put it where the codec will look for it.
     setattr(sys.modules[caller], config_cls.__name__, config_cls)
     return host_cls

dimos/core/coordination/blueprint_config/parser.py

@@ -80,11 +80,11 @@
 )
 from dimos.core.coordination.blueprints import (
     Blueprint,
-    TransportSpec,
     config_key,
     transport_config_name,
 )
 from dimos.core.global_config import GlobalConfig
+from dimos.core.stream import TransportSpec
 
 
 def split_run_arguments(tokens: Sequence[str]) -> tuple[tuple[str, ...], tuple[str, ...]]:

dimos/core/coordination/blueprint_config/schema.py

@@ -28,7 +28,8 @@
     all_annotation_types,
     scalar_annotation_types,
 )
-from dimos.core.coordination.blueprints import BlueprintAtom, TransportSpec
+from dimos.core.coordination.blueprints import BlueprintAtom
+from dimos.core.stream import TransportSpec
 
 
 @dataclass(frozen=True, slots=True)

dimos/core/coordination/blueprint_config/test_parsed.py

@@ -21,9 +21,8 @@
 
 from dimos.core.coordination.blueprint_config.errors import BlueprintConfigError
 from dimos.core.coordination.blueprint_config.parser import BlueprintConfigParser
-from dimos.core.coordination.blueprints import TransportSpec
 from dimos.core.module import Module, ModuleConfig
-from dimos.core.stream import Stream, Transport
+from dimos.core.stream import Stream, Transport, TransportSpec
 
 
 class NestedConfig(BaseModel):

dimos/core/coordination/blueprint_config/test_parser.py

@@ -25,9 +25,9 @@
     BlueprintConfigParser,
     split_run_arguments,
 )
-from dimos.core.coordination.blueprints import TransportSpec, autoconnect
+from dimos.core.coordination.blueprints import autoconnect
 from dimos.core.module import Module, ModuleConfig
-from dimos.core.stream import Stream, Transport
+from dimos.core.stream import Stream, Transport, TransportSpec
 
 
 class NestedConfig(BaseModel):