feat(recording): add CDR MCAP store and native recording

dimos/core/global_config.py

@@ -47,6 +47,7 @@ class GlobalConfig(BaseSettings):
     replay_db: str = "go2_short"
     record: Literal["", "sqlite", "mcap"] = ""
     record_engine: Literal["python", "rust"] = Field(default="python", validate_default=True)
+    record_executable: str | None = None  # Optional prebuilt Rust recorder path.
     record_topics: str = "*"  # comma-separated globs on the topic slug (/a/b -> a_b)
     record_encoding_threads: int | None = Field(default=None, ge=1)
     # How every zenoh session this process opens joins the network.

dimos/experimental/memory/cdr_schema.py

@@ -0,0 +1,32 @@
+# 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.
+
+"""Resolve schema metadata once when registering a raw CDR recording stream."""
+
+from typing import Any
+
+from dimos_cdrgen.registry import message_types, schema
+
+
+def cdr_name(message_type: type[Any]) -> str | None:
+    if not hasattr(message_type, "__msgtype__"):
+        return None
+    return next((name for name, cls in message_types().items() if cls is message_type), None)
+
+
+def recording_schema(message_type: type[Any]) -> dict[str, Any]:
+    name = cdr_name(message_type)
+    if name is None:
+        raise TypeError("CDR recording requires an installed, registry-owned message type")
+    return {"name": name, "data": schema(name)}

dimos/experimental/memory/rust/src/cdr.rs

@@ -0,0 +1,39 @@
+// 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.
+
+//! Schema metadata for opaque CDR recording.
+
+use anyhow::{ensure, Result};
+use serde::{Deserialize, Serialize};
+
+#[derive(Clone, Debug, Deserialize, Serialize)]
+#[serde(deny_unknown_fields)]
+pub struct CdrSchema {
+    pub name: String,
+    pub data: String,
+}
+
+impl CdrSchema {
+    pub fn validate(&self) -> Result<()> {
+        ensure!(
+            self.name.split('/').count() == 3 && self.name.contains("/msg/"),
+            "CDR schema requires package/msg/Type identity"
+        );
+        ensure!(
+            !self.data.trim().is_empty(),
+            "CDR recording requires complete ros2msg schema data"
+        );
+        Ok(())
+    }
+}

dimos/experimental/memory/rust/src/decoding.rs

@@ -30,6 +30,8 @@ use crate::{Codec, StreamConfig, IMAGE_PAYLOAD_TYPE};
 /// Transport-decoded payload consumed by storage codecs.
 #[derive(Debug)]
 pub(crate) enum DecodedPayload {
+    /// Plain CDR is preserved byte-for-byte, including its original Header.
+    Cdr(Vec<u8>),
     /// An LCM payload whose storage codecs operate on its canonical wire form.
     Lcm(Vec<u8>),
     /// UTF-8 JSON after removing the existing String transport envelope.
@@ -51,6 +53,12 @@ pub(crate) fn decode(
     data: &[u8],
     reception_ts: f64,
 ) -> Result<Vec<DecodedObservation>> {
+    if stream.codec == Codec::Cdr {
+        return Ok(vec![DecodedObservation {
+            ts: reception_ts,
+            payload: DecodedPayload::Cdr(data.to_vec()),
+        }]);
+    }
     if stream.codec == Codec::Json {
         anyhow::ensure!(
             stream.payload_type == "dimos.msgs.std_msgs.String.String",
@@ -194,6 +202,7 @@ mod tests {
 
     fn json_stream(field: Option<&str>) -> StreamConfig {
         StreamConfig {
+            cdr_schema: None,
             port: "events".into(),
             name: "events".into(),
             payload_type: "dimos.msgs.std_msgs.String.String".into(),

dimos/experimental/memory/rust/src/encoding.rs

@@ -35,6 +35,10 @@ pub(crate) fn encode(
     observation: DecodedObservation,
 ) -> Result<StoredObservation> {
     let data = match stream.codec {
+        Codec::Cdr => match observation.payload {
+            DecodedPayload::Cdr(data) => data,
+            _ => return Err(anyhow!("CDR storage requires a CDR transport payload")),
+        },
         Codec::Lcm => lcm_encode(observation.payload)?,
         Codec::Json => match observation.payload {
             DecodedPayload::Json(data) => data,
@@ -53,7 +57,9 @@ fn lcm_encode(payload: DecodedPayload) -> Result<Vec<u8>> {
     Ok(match payload {
         DecodedPayload::Lcm(data) => data,
         DecodedPayload::Image(image) => image.encode(),
-        DecodedPayload::Json(_) => return Err(anyhow!("LCM codec cannot store JSON payload")),
+        DecodedPayload::Json(_) | DecodedPayload::Cdr(_) => {
+            return Err(anyhow!("LCM codec cannot store non-LCM payload"))
+        }
     })
 }
 

dimos/experimental/memory/rust/src/lib.rs

@@ -31,6 +31,7 @@ use tracing::{debug, info};
 use crate::encoding::StoredObservation;
 use crate::store::{Observation, RecordingStore};
 
+pub mod cdr;
 mod decoding;
 mod encoding;
 pub mod store;
@@ -44,6 +45,7 @@ const FLUSH_INTERVAL: Duration = Duration::from_millis(100);
 #[derive(Clone, Copy, Debug, Deserialize, Serialize, PartialEq, Eq)]
 #[serde(rename_all = "lowercase")]
 pub enum Codec {
+    Cdr,
     Lcm,
     Json,
     Jpeg,
@@ -54,6 +56,7 @@ pub enum Codec {
 impl Codec {
     pub(crate) fn id(self) -> &'static str {
         match self {
+            Self::Cdr => "cdr",
             Self::Lcm => "lcm",
             Self::Json => "json",
             Self::Jpeg => "jpeg",
@@ -73,11 +76,13 @@ pub struct StreamConfig {
     pub timestamp_field: Option<String>,
     #[serde(default)]
     pub json_schema: Option<serde_json::Value>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub cdr_schema: Option<cdr::CdrSchema>,
 }
 
 impl StreamConfig {
     pub fn is_tf(&self) -> bool {
-        self.payload_type == TF_PAYLOAD_TYPE
+        self.codec != Codec::Cdr && self.payload_type == TF_PAYLOAD_TYPE
     }
 }
 
@@ -152,6 +157,18 @@ pub struct RecorderEngine {
 impl RecorderEngine {
     pub fn start(config: RecorderConfig) -> Result<Self> {
         for stream in &config.streams {
+            if stream.codec == Codec::Cdr {
+                stream
+                    .cdr_schema
+                    .as_ref()
+                    .context("CDR codec requires its resolved schema")?
+                    .validate()?;
+            } else {
+                anyhow::ensure!(
+                    stream.cdr_schema.is_none(),
+                    "Only the cdr codec accepts a CDR schema"
+                );
+            }
             anyhow::ensure!(
                 stream.codec == Codec::Json
                     || (stream.timestamp_field.is_none() && stream.json_schema.is_none()),
@@ -467,6 +484,7 @@ mod tests {
 
     fn stream(name: &str, codec: Codec, is_tf: bool) -> Arc<StreamConfig> {
         Arc::new(StreamConfig {
+            cdr_schema: None,
             port: name.to_string(),
             name: name.to_string(),
             payload_type: if is_tf {
@@ -482,6 +500,7 @@ mod tests {
 
     fn typed_stream(name: &str, payload_type: &str, codec: Codec) -> Arc<StreamConfig> {
         Arc::new(StreamConfig {
+            cdr_schema: None,
             port: name.to_string(),
             name: name.to_string(),
             payload_type: payload_type.to_string(),
@@ -562,6 +581,7 @@ mod tests {
         let mut message = Imu::default();
         message.header.stamp = Time { sec: 42, nsec: 25 };
         let stream = Arc::new(StreamConfig {
+            cdr_schema: None,
             port: "imu".to_string(),
             name: "imu".to_string(),
             payload_type: "dimos.msgs.sensor_msgs.Imu.Imu".to_string(),
@@ -817,6 +837,7 @@ mod tests {
         let message = lcm_msgs::std_msgs::String { data: text.into() };
         let schema = serde_json::json!({"type":"object","properties":{"sent":{"type":"number"}}});
         let stream = Arc::new(StreamConfig {
+            cdr_schema: None,
             port: "events".into(),
             name: "events".into(),
             payload_type: "dimos.msgs.std_msgs.String.String".into(),
@@ -870,3 +891,95 @@ mod tests {
         }
     }
 }
+
+#[cfg(test)]
+mod cdr_recording_tests {
+    use super::*;
+    use crate::cdr::CdrSchema;
+    use tempfile::tempdir;
+
+    #[test]
+    fn cdr_raw_storage_keeps_schema_and_opaque_payload_with_reception_time() {
+        let directory = tempdir().unwrap();
+        let path = directory.path().join("cdr.mcap");
+        let schema_text = "int32 sec\nuint32 nanosec\nstring label\n";
+        let stream = Arc::new(StreamConfig {
+            port: "events".into(),
+            name: "events".into(),
+            payload_type: "test_msgs.StampedEvent".into(),
+            codec: Codec::Cdr,
+            timestamp_field: None,
+            json_schema: None,
+            cdr_schema: Some(CdrSchema {
+                name: "test_msgs/msg/StampedEvent".into(),
+                data: schema_text.into(),
+            }),
+        });
+        let payload = vec![0, 1, 0, 0, 255, 123, 45];
+        let encoded = process(&stream, &payload, 99.0).unwrap().pop().unwrap();
+        assert_eq!(encoded.data, payload);
+        assert_eq!(encoded.ts, 99.0);
+        let config = store::RecordingStoreConfig::Mcap {
+            path: path.to_str().unwrap().into(),
+        };
+        let mut output = store::open(&config, &[(*stream).clone()], 1).unwrap();
+        output
+            .write_batch(&[Observation {
+                stream,
+                source_ts: encoded.ts,
+                reception_ts: 99.0,
+                data: encoded.data,
+            }])
+            .unwrap();
+        output.finish().unwrap();
+        let bytes = std::fs::read(path).unwrap();
+        let messages = mcap::MessageStream::new(&bytes)
+            .unwrap()
+            .collect::<Result<Vec<_>, _>>()
+            .unwrap();
+        assert_eq!(messages.len(), 1);
+        let message = &messages[0];
+        assert_eq!(message.publish_time, 99_000_000_000);
+        assert_eq!(message.log_time, 99_000_000_000);
+        assert_eq!(message.channel.message_encoding, "cdr");
+        assert_eq!(message.data.as_ref(), payload);
+        let actual_schema = message.channel.schema.as_ref().unwrap();
+        assert_eq!(actual_schema.name, "test_msgs/msg/StampedEvent");
+        assert_eq!(actual_schema.encoding, "ros2msg");
+        assert_eq!(actual_schema.data.as_ref(), schema_text.as_bytes());
+    }
+
+    #[test]
+    fn cdr_stream_preflight_rejects_missing_or_cross_codec_schema() {
+        let directory = tempdir().unwrap();
+        let mut stream = StreamConfig {
+            port: "sample".into(),
+            name: "sample".into(),
+            payload_type: "test.Sample".into(),
+            codec: Codec::Cdr,
+            cdr_schema: None,
+            timestamp_field: None,
+            json_schema: None,
+        };
+        let config = |stream| RecorderConfig {
+            store: store::RecordingStoreConfig::Mcap {
+                path: directory
+                    .path()
+                    .join("should-not-exist.mcap")
+                    .to_str()
+                    .unwrap()
+                    .into(),
+            },
+            encoding_threads: 1,
+            streams: vec![stream],
+        };
+        assert!(RecorderEngine::start(config(stream.clone())).is_err());
+        stream.cdr_schema = Some(CdrSchema {
+            name: "test_msgs/msg/Clock".into(),
+            data: "float64 ts\n".into(),
+        });
+        stream.codec = Codec::Lcm;
+        assert!(RecorderEngine::start(config(stream)).is_err());
+        assert!(!directory.path().join("should-not-exist.mcap").exists());
+    }
+}

dimos/experimental/memory/rust/src/store/mcap.rs

@@ -32,9 +32,26 @@ pub struct McapRecordingStore {
 
 impl McapRecordingStore {
     pub fn open(path: &str, streams: &[StreamConfig], compression_threads: usize) -> Result<Self> {
+        for stream in streams {
+            if stream.codec == crate::Codec::Cdr {
+                stream
+                    .cdr_schema
+                    .as_ref()
+                    .context("CDR MCAP requires its complete schema")?
+                    .validate()?;
+            }
+        }
         let file = File::create(path).with_context(|| format!("failed to create {path}"))?;
+        let profile = if streams
+            .iter()
+            .all(|stream| stream.codec == crate::Codec::Cdr)
+        {
+            "ros2"
+        } else {
+            "dimos"
+        };
         let options = WriteOptions::new()
-            .profile("dimos")
+            .profile(profile)
             .library("dimos-memory-recorder")
             .compression(Some(Compression::Zstd))
             .compression_threads(compression_threads.try_into().unwrap_or(u32::MAX));
@@ -50,10 +67,17 @@ impl McapRecordingStore {
                 ("dimos.port".to_string(), stream.port.clone()),
                 (
                     "dimos.observation_time".to_string(),
-                    "publish_time".to_string(),
+                    if stream.cdr_schema.is_some() {
+                        "log_time"
+                    } else {
+                        "publish_time"
+                    }
+                    .to_string(),
                 ),
             ]);
-            let schema_id = if let Some(schema) = &stream.json_schema {
+            let schema_id = if let Some(schema) = &stream.cdr_schema {
+                writer.add_schema(&schema.name, "ros2msg", schema.data.as_bytes())?
+            } else if let Some(schema) = &stream.json_schema {
                 writer.add_schema(&stream.name, "jsonschema", &serde_json::to_vec(schema)?)?
             } else {
                 0
@@ -145,6 +169,7 @@ mod tests {
     fn mcap_rejects_negative_source_time_instead_of_writing_zero() {
         let file = NamedTempFile::new().unwrap();
         let stream = Arc::new(StreamConfig {
+            cdr_schema: None,
             port: "events".into(),
             name: "events".into(),
             payload_type: "dimos.msgs.std_msgs.String.String".into(),
@@ -190,6 +215,7 @@ mod tests {
     fn mcap_preserves_zero_source_time_and_rejects_invalid_reception_time() {
         let file = NamedTempFile::new().unwrap();
         let stream = Arc::new(StreamConfig {
+            cdr_schema: None,
             port: "events".into(),
             name: "events".into(),
             payload_type: "dimos.msgs.std_msgs.String.String".into(),