diff --git a/.github/workflows/message-codegen.yml b/.github/workflows/message-codegen.yml index 53952452bd..7c1dd07e4b 100644 --- a/.github/workflows/message-codegen.yml +++ b/.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' diff --git a/dimos/memory/module.py b/dimos/memory/module.py index 738d1905fc..16c75e1b53 100644 --- a/dimos/memory/module.py +++ b/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 diff --git a/dimos/memory/store/mcap.py b/dimos/memory/store/mcap.py index 4f4602c1e3..f2ab6bd9f3 100644 --- a/dimos/memory/store/mcap.py +++ b/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/]/cdr//msg/``. + 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//`` (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 diff --git a/dimos/memory/store/test_cdr_recording.py b/dimos/memory/store/test_cdr_recording.py index c772a48a38..41a1616731 100644 --- a/dimos/memory/store/test_cdr_recording.py +++ b/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)) diff --git a/dimos/memory/test_module.py b/dimos/memory/test_module.py index c21ac9c913..3f23556ddf 100644 --- a/dimos/memory/test_module.py +++ b/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"