fix: record canonical TF and replay default CDR stream names

.github/workflows/message-codegen.yml

@@ -151,6 +151,7 @@ jobs:
         run: |
           LCM_DEFAULT_URL=memq:// .native-review/bin/python -I -m pytest --noconftest --import-mode=importlib -o addopts='' -q "$PWD/dimos/core/test_cdr_transport.py" "$PWD/dimos/core/test_transport_factory.py"
           .venv/bin/python -m pytest --noconftest --import-mode=importlib -o addopts='' -o 'markers=native_e2e: compiled native consumer acceptance' -q dimos/message_codegen/test_project_native.py -k actual_coordinator
+          .native-review/bin/python -I -c 'from dimos.memory import module; from dimos.memory.store import mcap; import pytest; import sys; assert all("site-packages" in str(value.__file__) for value in (module,mcap)); raise SystemExit(pytest.main(sys.argv[1:]))' --noconftest --import-mode=importlib -o addopts='' -q "$PWD/dimos/memory/test_module.py" "$PWD/dimos/memory/store/test_cdr_recording.py" "$PWD/dimos/memory/store/test_mcap.py"
           cargo test --manifest-path native/rust/dimos-module/Cargo.toml --locked --features cdr --lib --jobs 2
       - name: Verify generated browser schemas, relay encoding and web consumers
         if: runner.os == 'Linux'

dimos/memory/module.py

@@ -35,7 +35,9 @@
 from dimos.memory.store.sqlite import SqliteStore
 from dimos.memory.stream import Stream
 from dimos.memory.type.recording import OnExisting
-from dimos.msgs.tf2_msgs.TFMessage import TFMessage
+from dimos.messages.tf2_msgs.msg import TFMessage
+from dimos.msgs.geometry_msgs.Pose import Pose
+from dimos.msgs.time import to_seconds
 from dimos.types.timestamped import message_timestamp
 from dimos.utils.data import backup_file
 from dimos.utils.logging_config import setup_logger
@@ -48,7 +50,6 @@
         SemanticSearch as SemanticSearch,
         SemanticSearchConfig as SemanticSearchConfig,
     )
-    from dimos.msgs.geometry_msgs.Pose import Pose
 
 logger = setup_logger()
 
@@ -401,7 +402,14 @@ async def _resolve_pose(self, name: str, msg: Any, ts: float) -> Pose | None:
         transform = self._tf.get(
             self.config.root_frame, frame_id, time_point=ts, time_tolerance=self.config.tf_tolerance
         )
-        return transform.to_pose() if transform is not None else None
+        if transform is None:
+            return None
+        # Keep the recorder's internal Pose/pose-setter API; adapt only the TF boundary.
+        position = transform.transform.translation
+        rotation = transform.transform.rotation
+        return Pose(
+            position.x, position.y, position.z, rotation.x, rotation.y, rotation.z, rotation.w
+        )
 
     def _collect_pose_setters(self) -> dict[str, PoseSetter]:
         """Map stream name -> bound ``@pose_setter_for`` method."""
@@ -422,7 +430,11 @@ def _record_tf(self) -> None:
         def on_tf(msg: TFMessage) -> None:
             try:
                 for transform in msg.transforms:
-                    tf_stream.append(TFMessage(transform), ts=transform.ts, pose=None)
+                    tf_stream.append(
+                        TFMessage(transforms=[transform]),
+                        ts=to_seconds(transform.header.stamp),
+                        pose=None,
+                    )
             except sqlite3.ProgrammingError:
                 # A late LCM callback raced teardown and hit the closed store.
                 pass

dimos/memory/store/mcap.py

@@ -19,10 +19,10 @@
 topic -> codec), while ``streams`` may map friendly stream names to topics. See
 ``dimos.robot.unitree.go2.dds.store.Go2McapStore`` for the Go2 DDS wiring.
 
-Channels published by dimos itself need neither: a dimos topic carries its
-message type in its own name, so it is decoded and named from the wire with no
-registry at all (``_dimos_wire``). An injected codec still wins where both
-apply.
+Canonical CDR channels decode through the installed generated-type registry
+after validating their embedded ros2msg schema. Their wire topic supplies the
+logical stream name; legacy topics keep their separate legacy decoder. An
+injected codec still wins where both apply.
 
 Read-only: no append, blobs, vectors, or embeddings. Payloads decode lazily on
 ``obs.data``; ts and counts are cheap (counts come from the mcap index).
@@ -85,12 +85,22 @@ def decode(self, data: bytes) -> Any:
 def _dimos_wire(topic: str) -> tuple[str, type] | None:
     """Split a dimos wire topic into its port name and message type.
 
-    A dimos topic embeds the type in its last segment,
+    Canonical topics use ``[dimos/]<port>/cdr/<package>/msg/<Type>``.
+    The installed registry supplies their exact generated class; MCAP schema
+    validation still happens independently before decoding.
+
+    A legacy dimos topic embeds the type in its last segment,
     ``dimos/<port>/<msg_name>`` (built by ``zenohpubsub.Topic.key_expr``), so
     every dimos channel describes itself and no per-topic registry is needed.
     None for anything that is not one, which leaves DDS and app channels to
     the injected codecs.
     """
+    normalized = topic.lstrip("/")
+    base, marker, schema_name = normalized.rpartition("/cdr/")
+    if marker:
+        msg_type = message_types().get(schema_name)
+        port = base.removeprefix("dimos/")
+        return (port, msg_type) if port and msg_type is not None else None
     prefix, _, rest = topic.partition("/")
     port, _, msg_name = rest.rpartition("/")
     if prefix != "dimos" or not port:
@@ -259,6 +269,13 @@ def __init__(
                             f"stream {name!r} would name both {taken!r} and {ch.topic!r};"
                             " pass streams= to name them apart"
                         )
+                if "/cdr/" in ch.topic:
+                    wire = _dimos_wire(ch.topic)
+                    if wire is not None and ch.message_encoding != "cdr":
+                        raise ValueError(f"CDR topic requires cdr encoding: {ch.topic}")
+                    if wire is not None and declared is not None:
+                        if declared.name != getattr(wire[1], "__msgtype__", None):
+                            raise ValueError(f"CDR topic/schema mismatch: {ch.topic}")
                 if ch.message_encoding == "cdr" and ch.topic not in self._codecs:
                     # Unknown CDR stays raw, even when a legacy topic names a type.
                     self._codecs[ch.topic] = _BYTES_CODEC

dimos/memory/store/test_cdr_recording.py

@@ -14,23 +14,32 @@
 
 """CDR passes through real mem2 storage and self-describing MCAP replay."""
 
+from contextlib import ExitStack
+from functools import partial
 from pathlib import Path
 import subprocess
 import sys
+import threading
 
 from dimos_cdrgen.registry import encode, schema
 from mcap.reader import make_reader
+from mcap.writer import Writer
 import pytest
 
+import dimos.core.cdr_transport as cdr_module
+from dimos.core.global_config import global_config
+from dimos.core.transport_factory import make_transport
 from dimos.memory.codecs.base import codec_for, codec_from_id, codec_id
 from dimos.memory.codecs.cdr import CdrCodec
 from dimos.memory.store.mcap import McapStore
 from dimos.memory.store.sqlite import SqliteStore
+from dimos.memory.tap import TransportRecorder
 from dimos.messages.builtin_interfaces.msg import Time
 from dimos.messages.dimos_msgs.msg import LineSegment3D, LineSegments3D
 from dimos.messages.geometry_msgs.msg import Point
 from dimos.messages.std_msgs.msg import Header
 from dimos.protocol.cdr_mcap import CdrMcapWriter
+from dimos.protocol.service.lcmservice import LCMConfig
 from dimos.types.timestamped import message_timestamp
 
 
@@ -165,3 +174,88 @@ def test_documented_recording_example_runs_and_refuses_to_overwrite(tmp_path):
     assert repeated.returncode != 0
     assert "FileExistsError" in repeated.stderr
     assert {name: (destination / name).read_bytes() for name in original} == original
+
+
+@pytest.mark.parametrize("backend", ["lcm", "zenoh"])
+def test_factory_capture_sqlite_mcap_auto_named_replay(backend, monkeypatch, tmp_path):
+    # Isolate only session configuration; codecs, pub/sub and storage stay real.
+    monkeypatch.setattr(cdr_module, "LCMConfig", partial(LCMConfig, url="memq://"))
+    for key, value in {
+        "transport": backend,
+        "robot_ip": None,
+        "robot_ips": "",
+        "zenoh_connect": "",
+        "zenoh_mode": "peer",
+        "zenoh_multicast": False,
+        "zenoh_gossip": False,
+        "zenoh_scouting": False,
+    }.items():
+        monkeypatch.setattr(global_config, key, value)
+    expected = message()
+    expected.header.stamp.nanosec = 123456789
+    database = tmp_path / "live.db"
+    received = threading.Event()
+    with SqliteStore(path=str(database)) as store, ExitStack() as cleanup:
+        transport = make_transport("sensor/segments", LineSegments3D)
+        cleanup.callback(transport.stop)
+        recorder = TransportRecorder(store)
+        cleanup.callback(recorder.close)
+        unsubscribe = recorder.tap("segments", LineSegments3D, transport)
+        assert unsubscribe is not None
+        cleanup.callback(unsubscribe)
+        live = store.stream("segments", LineSegments3D).live()
+        cleanup.callback(live.stop)
+        cleanup.enter_context(live.subscribe(lambda _: received.set()))
+        transport.publish(expected)
+        assert received.wait(5), "default-factory payload was not recorded"
+        assert recorder.dropped == 0
+    path = tmp_path / "live.mcap"
+    with SqliteStore(path=str(database), must_exist=True) as store, CdrMcapWriter(path) as writer:
+        captured = next(iter(store.streams.segments)).data
+        assert type(captured) is LineSegments3D
+        assert encode(captured) == encode(expected)
+        writer.write(
+            transport.channel,
+            encode(captured),
+            schema_name=LineSegments3D.__msgtype__,
+            schema=schema(LineSegments3D.__msgtype__),
+            log_time_ns=20_000_000_000,
+            publish_time_ns=17_123456789,
+        )
+    with McapStore(path=str(path)) as store:
+        assert store.list_streams() == ["sensor_segments"]
+        values = list(store.replay().streams.sensor_segments.iterate_ts())
+        assert values == [(20.0, expected)]
+        assert values[0][1].header.stamp.nanosec == 123456789
+    with path.open("rb") as raw:
+        definition, channel, recorded = next(make_reader(raw).iter_messages())
+        assert recorded.publish_time == 17_123456789
+        assert recorded.log_time == 20_000_000_000
+        assert "MSG: std_msgs/Header" in definition.data.decode()
+        assert channel.topic == transport.channel
+
+
+def test_canonical_wire_topic_cannot_claim_a_different_schema(tmp_path):
+    path = tmp_path / "wrong-topic.mcap"
+    with CdrMcapWriter(path) as writer:
+        writer.write(
+            "dimos/pose/cdr/std_msgs/msg/Header",
+            encode(message()),
+            schema_name=LineSegments3D.__msgtype__,
+            schema=schema(LineSegments3D.__msgtype__),
+            log_time_ns=1,
+        )
+    with pytest.raises(ValueError, match="topic/schema mismatch"):
+        McapStore(path=str(path))
+
+
+def test_canonical_channel_never_falls_back_to_legacy_encoding(tmp_path):
+    path = tmp_path / "wrong-encoding.mcap"
+    with path.open("wb") as raw:
+        writer = Writer(raw)
+        writer.start()
+        channel = writer.register_channel("dimos/pose/cdr/std_msgs/msg/Header", "lcm", 0)
+        writer.add_message(channel, log_time=1, publish_time=1, data=b"legacy")
+        writer.finish()
+    with pytest.raises(ValueError, match="requires cdr encoding"):
+        McapStore(path=str(path))

dimos/memory/test_module.py

@@ -17,19 +17,30 @@
 from __future__ import annotations
 
 from collections.abc import Iterator
+from contextlib import ExitStack
+from functools import partial
+import threading
 
 import pytest
 
+import dimos.core.cdr_transport as cdr_module
+from dimos.core.global_config import GlobalConfig
 from dimos.core.module import ModuleConfig
 from dimos.core.stream import In, Out
+from dimos.core.transport_factory import make_transport
 from dimos.memory.module import Recorder, StreamModule
+from dimos.memory.store.sqlite import SqliteStore
 from dimos.memory.stream import Stream
 from dimos.memory.transform import Transformer
 from dimos.memory.type.observation import Observation
 from dimos.messages.builtin_interfaces.msg import Time as GeneratedTime
 from dimos.messages.dimos_msgs.msg import LineSegments3D as GeneratedSegments
+from dimos.messages.geometry_msgs.msg import Quaternion, Transform, TransformStamped, Vector3
 from dimos.messages.std_msgs.msg import Header as GeneratedHeader
+from dimos.messages.tf2_msgs.msg import TFMessage
+from dimos.msgs.geometry_msgs.Pose import Pose
 from dimos.protocol.rpc.pubsubrpc import LCMRPC
+from dimos.protocol.service.lcmservice import LCMConfig
 
 
 class Double(Transformer[int, int]):
@@ -102,11 +113,20 @@ async def test_recorder_reads_generated_header_without_network(mocker, tmp_path)
     recorder = Recorder(rpc_transport=LCMRPC, db_path=tmp_path / "unused.db", record_tf=False)
     transform_tree = mocker.Mock()
     mocker.patch.object(recorder, "_tf", transform_tree)
-    transform_tree.get.return_value.to_pose.return_value = mocker.sentinel.pose
+    transform_tree.get.return_value = TransformStamped(
+        GeneratedHeader(GeneratedTime(0, 0), "world"),
+        "sensor_frame",
+        Transform(Vector3(1.0, 2.0, 3.0), Quaternion(0.0, 0.0, 0.0, 1.0)),
+    )
     message = GeneratedSegments(GeneratedHeader(GeneratedTime(0, 0), "sensor_frame"), [])
     try:
         assert recorder._resolve_ts("segments", message) == 0.0
-        assert await recorder._resolve_pose("segments", message, 0.0) is mocker.sentinel.pose
+        pose = await recorder._resolve_pose("segments", message, 0.0)
+        assert type(pose) is Pose
+        assert pose.position.x == 1.0
+        assert pose.position.y == 2.0
+        assert pose.position.z == 3.0
+        assert pose.orientation.w == 1.0
         transform_tree.get.assert_called_once_with(
             "world",
             "sensor_frame",
@@ -116,3 +136,52 @@ async def test_recorder_reads_generated_header_without_network(mocker, tmp_path)
     finally:
         recorder.stop()
     assert not (tmp_path / "unused.db").exists()
+
+
+class MemqRPC(LCMRPC):
+    def __init__(self, **kwargs):
+        super().__init__(url="memq://", **kwargs)
+
+
+def test_recorder_tf_cdr_boundary_records_each_transform_without_changing_header(
+    mocker, monkeypatch, tmp_path
+):
+    monkeypatch.setattr(cdr_module, "LCMConfig", partial(LCMConfig, url="memq://"))
+    mocker.patch("dimos.core.module.get_loop", return_value=(None, None))
+    mocker.patch.object(LCMRPC, "start")
+    mocker.patch.object(LCMRPC, "serve_module_rpc")
+    path = tmp_path / "tf.db"
+    transforms = [
+        TransformStamped(
+            GeneratedHeader(GeneratedTime(0, nanos), "world"),
+            "base_link",
+            Transform(Vector3(1.0, 2.0, 3.0), Quaternion(0.0, 0.0, 0.0, 1.0)),
+        )
+        for nanos in (0, 1)
+    ]
+    ready = threading.Event()
+    observed = []
+
+    def receive(observation):
+        observed.append(observation)
+        if len(observed) == 2:
+            ready.set()
+
+    with ExitStack() as cleanup:
+        recorder = Recorder(rpc_transport=MemqRPC, db_path=path)
+        cleanup.callback(recorder.stop)
+        transport = make_transport("tf", TFMessage, g=GlobalConfig(transport="lcm"))
+        cleanup.callback(transport.stop)
+        recorder.set_transport("tf", transport)
+        recorder._record_tf()
+        live = recorder.store.streams.tf.live()
+        cleanup.callback(live.stop)
+        cleanup.enter_context(live.subscribe(receive))
+        transport.publish(TFMessage(transforms=transforms))
+        assert ready.wait(5), "TF recorder did not append both transforms"
+        assert [observation.ts for observation in observed] == [0.0, 1e-9]
+    with SqliteStore(path=str(path), must_exist=True) as store:
+        values = list(store.streams.tf)
+        assert [item.data for item in values] == [TFMessage([value]) for value in transforms]
+        assert [item.data.transforms[0].header.stamp.nanosec for item in values] == [0, 1]
+        assert transforms[1].header.frame_id == "world"