feat: record generated messages as self-contained MCAP (3/6)

.github/scripts/wheel_smoke.py

@@ -49,7 +49,7 @@ def main() -> None:
     image_type = types["sensor_msgs/msg/Image"]
     value = image_type(height=1, width=2, step=2, encoding="mono8", data=[3, 7])
     assert list(image_type.decode(value.encode()).data) == [3, 7]
-    assert "MSG: std_msgs/msg/Header" in image_type.schema
+    assert "MSG: std_msgs/Header" in image_type.schema
     dist = Path(locate.__file__).resolve().parent / "_relay_dist"
     for rel in REQUIRED:
         if not (dist / rel).is_file():

.github/workflows/message-codegen.yml

@@ -8,6 +8,9 @@ on:
       - native/rust/dimos-lcm-transport/**
       - scripts/*message*codegen*
       - scripts/test_message_packages.sh
+      - scripts/test_message_mcap.sh
+      - dimos/protocol/cdr_mcap.py
+      - dimos/protocol/test_cdr_mcap.py
       - scripts/vendor_message_definitions.py
       - .github/workflows/message-codegen.yml
       - .github/docker/message-codegen-reference.Dockerfile
@@ -25,18 +28,31 @@ jobs:
       - uses: actions/checkout@v7
       - uses: astral-sh/setup-uv@v10.0.1
       - uses: dtolnay/rust-toolchain@stable
+      - uses: actions/setup-node@v6
+        with:
+          node-version: '22'
       - name: Set up a ROS-free Python environment
         run: |
           test ! -d /opt/ros
           uv venv .venv --python 3.12
-          uv pip install --python .venv/bin/python pybind11==3.0.1 rosbags==0.11.0 numpy pytest pytest-asyncio pytest-env lcm-dimos-fork setuptools wheel
+          uv pip install --python .venv/bin/python pybind11==3.0.1 rosbags==0.11.0 numpy pytest pytest-asyncio pytest-env lcm-dimos-fork setuptools wheel mcap==1.4.0 rerun-sdk==0.32.0 pillow
           .venv/bin/python -c "import importlib.util; assert importlib.util.find_spec('rclpy') is None"
       - name: Build the pinned standalone CDR library
         run: bash scripts/setup_message_codegen.sh
       - name: Generate, build, test, and demonstrate all three languages
         run: bash scripts/test_message_codegen.sh
       - name: Build and consume wheel, sdist, CMake, and Cargo packages
         run: bash scripts/test_message_packages.sh
+      - name: Record generated CDR and verify both viewer decoders
+        run: bash scripts/test_message_mcap.sh
+      - name: Retain self-contained recording and viewer evidence
+        uses: actions/upload-artifact@v7
+        with:
+          name: message-mcap-evidence
+          path: |
+            build/message-codegen/viewers/demo.mcap
+            build/message-codegen/viewers/demo.rrd
+            build/message-codegen/viewers/evidence/
       - name: Retain external application and packages
         uses: actions/upload-artifact@v7
         with:

dimos/message_codegen/definitions.py

@@ -196,5 +196,8 @@ def schema(self, name: str) -> str:
         sections = [root.text.rstrip() + "\n"]
         for message in sorted(closure, key=lambda item: item.name):
             if message.name != name:
-                sections.append(f"{'=' * 80}\nMSG: {message.name}\n{message.text.rstrip()}\n")
+                # Concatenated .msg sections use the package/resource spelling
+                # used by field references. The outer schema name stays pkg/msg/T.
+                resource_name = message.name.replace("/msg/", "/")
+                sections.append(f"{'=' * 80}\nMSG: {resource_name}\n{message.text.rstrip()}\n")
         return "".join(sections)

dimos/protocol/cdr_mcap.py

@@ -0,0 +1,97 @@
+# 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.
+
+"""Self-describing CDR recording, independent of transport and generated classes."""
+
+from __future__ import annotations
+
+from pathlib import Path
+from types import TracebackType
+
+from mcap.writer import CompressionType, Writer
+from typing_extensions import Self
+
+
+class CdrMcapWriter:
+    """Write raw generated CDR with embedded ROS2 definitions and chunk compression.
+
+    Callers supply source and reception timestamps explicitly. No message package
+    is imported, so a recorder can accept schemas from external stream providers.
+    """
+
+    def __init__(self, path: str | Path) -> None:
+        self._stream = Path(path).open("wb")
+        try:
+            self._writer = Writer(self._stream, compression=CompressionType.ZSTD)
+            self._writer.start(profile="ros2", library="dimos")
+        except BaseException:
+            self._stream.close()
+            raise
+        self._schemas: dict[tuple[str, str], int] = {}
+        self._channels: dict[tuple[str, int], int] = {}
+        self._closed = False
+
+    def write(
+        self,
+        topic: str,
+        payload: bytes,
+        *,
+        schema_name: str,
+        schema: str,
+        log_time_ns: int,
+        publish_time_ns: int | None = None,
+        sequence: int = 0,
+    ) -> None:
+        if self._closed:
+            raise ValueError("MCAP writer is closed")
+        schema_key = (schema_name, schema)
+        if schema_key not in self._schemas:
+            self._schemas[schema_key] = self._writer.register_schema(
+                name=schema_name, encoding="ros2msg", data=schema.encode("utf-8")
+            )
+        schema_id = self._schemas[schema_key]
+        channel_key = (topic, schema_id)
+        if channel_key not in self._channels:
+            self._channels[channel_key] = self._writer.register_channel(
+                topic=topic,
+                message_encoding="cdr",
+                schema_id=schema_id,
+                metadata={"offered_qos_profiles": "[]"},
+            )
+        self._writer.add_message(
+            channel_id=self._channels[channel_key],
+            data=payload,
+            log_time=log_time_ns,
+            publish_time=log_time_ns if publish_time_ns is None else publish_time_ns,
+            sequence=sequence,
+        )
+
+    def close(self) -> None:
+        if not self._closed:
+            self._closed = True
+            try:
+                self._writer.finish()
+            finally:
+                self._stream.close()
+
+    def __enter__(self) -> Self:
+        return self
+
+    def __exit__(
+        self,
+        exc_type: type[BaseException] | None,
+        exc: BaseException | None,
+        traceback: TracebackType | None,
+    ) -> None:
+        self.close()

dimos/protocol/test_cdr_mcap.py

@@ -0,0 +1,110 @@
+# 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.
+
+from mcap.reader import make_reader
+from mcap.records import Chunk
+from mcap.stream_reader import StreamReader
+import pytest
+from rosbags.typesys import Stores, get_types_from_msg, get_typestore
+
+from dimos.message_codegen.definitions import Definitions
+from dimos.protocol.cdr_mcap import CdrMcapWriter
+
+
+@pytest.fixture
+def point_message():
+    name = "geometry_msgs/msg/Point"
+    schema = Definitions([]).schema(name)
+    reference = get_typestore(Stores.ROS2_JAZZY)
+    value = reference.types[name](x=1.25, y=-2.5, z=3.0)
+    return name, schema, bytes(reference.serialize_cdr(value, name))
+
+
+def test_embeds_schemas_and_preserves_cdr_timestamps_and_sequence(tmp_path, point_message):
+    name, schema, payload = point_message
+    path = tmp_path / "capture.mcap"
+    with CdrMcapWriter(path) as writer:
+        for topic, timestamp in (("/first", 100), ("/second", 200)):
+            writer.write(
+                topic,
+                payload,
+                schema_name=name,
+                schema=schema,
+                log_time_ns=timestamp,
+                publish_time_ns=0,
+                sequence=7,
+            )
+    with path.open("rb") as stream:
+        reader = make_reader(stream, validate_crcs=True)
+        assert reader.get_header().profile == "ros2"
+        summary = reader.get_summary()
+        assert len(summary.schemas) == 1
+        assert len(summary.channels) == 2
+        messages = list(reader.iter_messages())
+    assert [message.log_time for _, _, message in messages] == [100, 200]
+    reference = get_typestore(Stores.EMPTY)
+    for stored_schema, channel, message in messages:
+        assert stored_schema.encoding == "ros2msg"
+        assert stored_schema.name == name
+        assert stored_schema.data.decode() == schema
+        assert channel.message_encoding == "cdr"
+        assert channel.metadata["offered_qos_profiles"] == "[]"
+        assert message.data == payload
+        assert message.publish_time == 0
+        assert message.sequence == 7
+        reference.register(get_types_from_msg(stored_schema.data.decode(), stored_schema.name))
+        decoded = reference.deserialize_cdr(message.data, name)
+        assert (decoded.x, decoded.y, decoded.z) == (1.25, -2.5, 3.0)
+
+
+def test_unstamped_messages_use_reception_time_and_chunks_are_compressed(tmp_path, point_message):
+    name, schema, payload = point_message
+    path = tmp_path / "capture.mcap"
+    with CdrMcapWriter(path) as writer:
+        writer.write("/point", payload, schema_name=name, schema=schema, log_time_ns=123)
+    with path.open("rb") as stream:
+        messages = list(make_reader(stream).iter_messages())
+    assert messages[0][2].publish_time == 123
+    with path.open("rb") as stream:
+        chunks = [
+            record
+            for record in StreamReader(stream, emit_chunks=True).records
+            if isinstance(record, Chunk)
+        ]
+    assert chunks
+    assert all(chunk.compression == "zstd" for chunk in chunks)
+
+
+def test_context_finalizes_recording_when_producer_raises(tmp_path, point_message):
+    name, schema, payload = point_message
+    path = tmp_path / "capture.mcap"
+    with pytest.raises(RuntimeError, match="producer"), CdrMcapWriter(path) as writer:
+        writer.write("/point", payload, schema_name=name, schema=schema, log_time_ns=123)
+        raise RuntimeError("producer failed")
+    with path.open("rb") as stream:
+        assert make_reader(stream).get_summary().statistics.message_count == 1
+    with pytest.raises(ValueError, match="closed"):
+        writer.write("/point", payload, schema_name=name, schema=schema, log_time_ns=124)
+
+
+def test_nested_schema_uses_resolvable_resource_names():
+    schema = Definitions([]).schema("sensor_msgs/msg/Image")
+
+    assert "MSG: std_msgs/Header\n" in schema
+    assert "MSG: builtin_interfaces/Time\n" in schema
+    assert set(get_types_from_msg(schema, "sensor_msgs/msg/Image")) == {
+        "sensor_msgs/msg/Image",
+        "std_msgs/msg/Header",
+        "builtin_interfaces/msg/Time",
+    }

examples/message-codegen/EVIDENCE.md

@@ -96,3 +96,36 @@ content check (1,928 entries; 10.7 MB). A fresh environment with both wheels
 discovered 141 types and accepted a built-in Point inside external Telemetry.
 Two separately installed CMake packages can include the same standard Point
 without duplicate definitions. Focused tests now pass 34 cases.
+
+## Stage 3 automated recording checks
+
+`bash scripts/test_message_mcap.sh` passed on 2026-09-21 with Rerun 0.32.0,
+MCAP 1.4.0, and the locked Foxglove JavaScript packages. The artifact contains
+150 messages and is 3,123,201 bytes, including all schema dependencies.
+
+```text
+4 passed
+Foxglove libraries decoded 30 /camera/image messages from embedded definitions.
+Foxglove libraries decoded 30 /camera/compressed messages from embedded definitions.
+Foxglove libraries decoded 30 /robot/pose messages from embedded definitions.
+Foxglove libraries decoded 30 /telemetry messages from embedded definitions.
+Foxglove libraries decoded 30 /tf messages from embedded definitions.
+Rerun imported 30 /telemetry rows as demo_msgs.msg.Telemetry:message.
+Rerun imported 30 /camera/image rows as Image:buffer.
+Rerun imported 30 /camera/compressed rows as EncodedImage:blob.
+Rerun imported 30 /robot/pose rows as InstancePoses3D:translations.
+Rerun imported 30 /tf rows as Transform3D:translation.
+Custom telemetry is a structured Arrow value with nested fields.
+```
+
+The stage-1 conformance/buffer/raw-transport demos and stage-2 installed-package
+demo were rerun after correcting dependency section names to `package/Type`.
+The Rerun browser viewer displayed both images and exposed custom Telemetry's
+`sequence: 29`, `reading.temperature: 22.9`, and `hops: [1, 2, 3]` from the MCAP.
+Local screenshots are in `build/message-codegen/viewers/`.
+
+Full viewer acceptance is still pending. Foxglove's application redirected to
+sign-in in the isolated browser; no credentials were entered. Its decoding
+libraries pass independently, but that does not establish application UI
+acceptance. Rerun's direct-file screen capture also needs final confirmation
+of pose visualization and complete semantic playback before task 3.5 is closed.

examples/message-codegen/README.md

@@ -172,3 +172,54 @@ type discovery validates conflicting qualified definitions before loading types.
 Generation, installation from built artifacts, and runtime require no schema
 service or runtime download. Dependency setup can use the network; cached Cargo
 builds and local wheel installation are exercised offline by the demo.
+
+## Stage 3: one recording for Foxglove and Rerun
+
+After building the stage-1 demo, install the viewer-check dependencies and run:
+
+```bash
+uv pip install --python .venv/bin/python mcap==1.4.0 rerun-sdk==0.32.0 pillow
+bash scripts/test_message_mcap.sh
+```
+
+Node.js 22 and npm are required for the independent Foxglove decoder check. The
+script installs its locked packages with `npm ci`. It writes
+`build/message-codegen/viewers/demo.mcap`: 30 frames each of raw RGB Image, PNG
+CompressedImage, PoseStamped, TFMessage, and nested custom Telemetry. Source
+timestamps start at 2023-11-14 22:13:20 UTC and advance by 100 ms. Reception
+timestamps are 1 ms later. Each channel contains ordinary encapsulated CDR;
+the MCAP chunks use Zstandard compression.
+
+The checks read the embedded definitions using Foxglove's official libraries
+and the native Rerun 0.32.0 importer. Rerun must produce image, encoded-image,
+pose, and transform components plus a structured custom telemetry column. It
+must import all 30 rows per topic. This catches unresolved schema dependencies
+that can otherwise silently omit custom data. CI uploads the MCAP, converted
+RRD, and decoder evidence as `message-mcap-evidence`.
+
+To inspect the file in Rerun directly:
+
+```bash
+RERUN_ANALYTICS_ENABLED=false .venv/bin/rerun build/message-codegen/viewers/demo.mcap
+```
+
+For a browser-hosted viewer, add `--serve-web --bind 127.0.0.1 --port auto` and
+open the URL printed by Rerun. No ROS or `demo_msgs` installation is needed by
+the viewer. Raw and compressed images should show the same changing gradient.
+Select `/robot/pose` in Streams to inspect the map-frame position, and select
+`/telemetry` to inspect the nested message fields. The CLI conversion explicitly
+enables `ros2msg` and `ros2_reflection`; the saved `demo.rrd` is also available
+for examining these decoder results.
+
+In Foxglove, sign in, open the **same** local `demo.mcap`, add an Image panel for
+`/camera/image`, a 3D panel for `/robot/pose` with fixed frame `map`, and a Raw
+Messages panel for `/telemetry`. Scrub from first to last frame: sequence should
+change from 0 to 29 and `reading.temperature` from 20.0 to 22.9. The compressed
+image topic should display the same pixels. Foxglove's web app currently
+requires authentication; library decoding is automated, but application UI
+acceptance remains pending until an authenticated viewer is available.
+
+Stop the Rerun process with Ctrl-C and close its browser tab. The script starts
+no persistent services. Remove `build/message-codegen/viewers/` to clean up
+recordings and `examples/message-codegen/viewer-checker/node_modules/` to clean
+up the Node dependencies.