diff --git a/dimos/core/global_config.py b/dimos/core/global_config.py
index 7b657260a16..355d0016dfe 100644
--- a/dimos/core/global_config.py
+++ b/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.
diff --git a/dimos/experimental/memory/cdr_schema.py b/dimos/experimental/memory/cdr_schema.py
new file mode 100644
index 00000000000..624576014c4
--- /dev/null
+++ b/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)}
diff --git a/dimos/experimental/memory/rust/src/cdr.rs b/dimos/experimental/memory/rust/src/cdr.rs
new file mode 100644
index 00000000000..b06527c0500
--- /dev/null
+++ b/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(())
+    }
+}
diff --git a/dimos/experimental/memory/rust/src/decoding.rs b/dimos/experimental/memory/rust/src/decoding.rs
index 79b6b98a5c1..adb51abef6b 100644
--- a/dimos/experimental/memory/rust/src/decoding.rs
+++ b/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(),
diff --git a/dimos/experimental/memory/rust/src/encoding.rs b/dimos/experimental/memory/rust/src/encoding.rs
index 00eacc3823b..8e016741acf 100644
--- a/dimos/experimental/memory/rust/src/encoding.rs
+++ b/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"))
+        }
     })
 }
 
diff --git a/dimos/experimental/memory/rust/src/lib.rs b/dimos/experimental/memory/rust/src/lib.rs
index 74df3f3f9f9..c05bdaabc40 100644
--- a/dimos/experimental/memory/rust/src/lib.rs
+++ b/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());
+    }
+}
diff --git a/dimos/experimental/memory/rust/src/store/mcap.rs b/dimos/experimental/memory/rust/src/store/mcap.rs
index 23213affe24..cb4a43d81b6 100644
--- a/dimos/experimental/memory/rust/src/store/mcap.rs
+++ b/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(),
diff --git a/dimos/experimental/memory/rust_cli_recorder.py b/dimos/experimental/memory/rust_cli_recorder.py
index 2214635c4e2..060424400aa 100644
--- a/dimos/experimental/memory/rust_cli_recorder.py
+++ b/dimos/experimental/memory/rust_cli_recorder.py
@@ -30,6 +30,7 @@
 from dimos.core.global_config import global_config
 from dimos.core.transport import LCMTransport, ZenohTransport
 from dimos.core.transport_factory import session_config
+from dimos.experimental.memory.cdr_schema import cdr_name, recording_schema
 from dimos.experimental.memory.rust_recorder import RustRecorderConfig, RustStreamSpec
 from dimos.memory.store.sqlite import SqliteStore
 from dimos.memory.tap import matching, recording_dir
@@ -57,6 +58,14 @@ class RustRecordingPlan:
 
 def prepare_rust_recorder() -> Path:
     """Build the native recorder when it is unavailable or explicitly requested."""
+    if global_config.record_executable is not None:
+        executable = Path(global_config.record_executable).expanduser().resolve()
+        if not executable.is_file():
+            raise FileNotFoundError(f"Rust recorder executable not found: {executable}")
+        if not os.access(executable, os.X_OK):
+            raise PermissionError(f"Rust recorder is not executable: {executable}")
+        return executable
+
     config = RustRecorderConfig()
     rust_dir, executable_path = config.resolve_paths()
     executable = Path(executable_path)
@@ -103,9 +112,12 @@ def make_plan(transports: dict[tuple[str, type], Any]) -> RustRecordingPlan:
     for index, ((name, payload_type), transport) in enumerate(transports.items()):
         if name not in selected:
             continue
-        if not hasattr(payload_type, "lcm_encode") or not hasattr(payload_type, "lcm_decode"):
+        generated = cdr_name(payload_type) is not None
+        if not generated and (
+            not hasattr(payload_type, "lcm_encode") or not hasattr(payload_type, "lcm_decode")
+        ):
             logger.info(
-                "--record: stream is not a DimOS LCM message type; skipped",
+                "--record: stream is not a supported message type; skipped",
                 stream=name,
                 payload_type=str(payload_type),
             )
@@ -131,7 +143,10 @@ def make_plan(transports: dict[tuple[str, type], Any]) -> RustRecordingPlan:
                 port=port,
                 name=name,
                 payload_type=f"{payload_type.__module__}.{payload_type.__qualname__}",
-                codec="jpeg" if issubclass(payload_type, Image) else "lcm",
+                codec="cdr"
+                if generated
+                else ("jpeg" if issubclass(payload_type, Image) else "lcm"),
+                cdr_schema=recording_schema(payload_type) if generated else None,
             )
         )
         topics[port] = transport.channel
@@ -299,4 +314,6 @@ def stop(self) -> None:
                 thread.join(timeout=DEFAULT_THREAD_JOIN_TIMEOUT)
         if process.returncode not in (None, 0):
             logger.warning("Rust recorder stopped with an error", status=process.returncode)
+        if process.returncode == 0:
+            logger.info("Recording saved", artifact_path=str(self.plan.path))
         self._process = None
diff --git a/dimos/experimental/memory/rust_recorder.py b/dimos/experimental/memory/rust_recorder.py
index 8324c0469f8..f77e625ba57 100644
--- a/dimos/experimental/memory/rust_recorder.py
+++ b/dimos/experimental/memory/rust_recorder.py
@@ -12,7 +12,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-"""Experimental native recorder for high-throughput LCM-backed memory streams."""
+"""Experimental native recorder for legacy LCM and generated CDR streams."""
 
 from __future__ import annotations
 
@@ -27,6 +27,7 @@
 from dimos.core.module import Module
 from dimos.core.native_module import NativeModule, NativeModuleConfig
 from dimos.core.stream import In
+from dimos.experimental.memory.cdr_schema import cdr_name, recording_schema
 from dimos.memory.store.sqlite import SqliteStore
 from dimos.memory.type.recording import OnExisting
 from dimos.msgs.sensor_msgs.Image import Image
@@ -37,7 +38,7 @@
 
 logger = setup_logger()
 
-_SUPPORTED_NATIVE_CODECS = {"lcm", "jpeg", "lz4+lcm", "json"}
+_SUPPORTED_NATIVE_CODECS = {"lcm", "jpeg", "lz4+lcm", "json", "cdr"}
 
 
 class RustStreamSpec(BaseModel):
@@ -49,6 +50,7 @@ class RustStreamSpec(BaseModel):
     codec: str
     timestamp_field: str | None = None
     json_schema: dict[str, Any] | None = None
+    cdr_schema: dict[str, Any] | None = None
 
     @model_validator(mode="after")
     def _json_options(self) -> RustStreamSpec:
@@ -56,6 +58,10 @@ def _json_options(self) -> RustStreamSpec:
             self.timestamp_field is not None or self.json_schema is not None
         ):
             raise ValueError("JSON options require the json codec")
+        if (self.codec == "cdr") != (self.cdr_schema is not None):
+            raise ValueError(
+                "The cdr codec requires its resolved schema; other codecs must not carry one"
+            )
         return self
 
 
@@ -220,7 +226,9 @@ def _stream_specs(self) -> list[RustStreamSpec]:
         for port_name, port in self.inputs.items():
             if getattr(port, "_transport", None) is None:
                 continue
-            if port is self.tf:
+            if port is self.tf and not self.config.record_tf:
+                continue
+            if port is self.tf and cdr_name(port.type) is None:
                 if self.config.record_tf:
                     specs.append(
                         RustStreamSpec(
@@ -243,6 +251,7 @@ def _stream_specs(self) -> list[RustStreamSpec]:
                     codec=codec,
                     timestamp_field=self.config.stream_timestamp_fields.get(stream_name),
                     json_schema=self.config.stream_json_schemas.get(stream_name),
+                    cdr_schema=recording_schema(port.type) if codec == "cdr" else None,
                 )
             )
 
@@ -261,12 +270,14 @@ def _collect_topics(self) -> dict[str, str]:
 
     @staticmethod
     def _default_codec(payload_type: type[Any]) -> str:
+        if cdr_name(payload_type) is not None:
+            return "cdr"
         if issubclass(payload_type, Image):
             return "jpeg"
         if hasattr(payload_type, "lcm_encode") and hasattr(payload_type, "lcm_decode"):
             return "lcm"
         raise TypeError(
-            f"RustRecorder only supports LCM-backed messages, got {payload_type.__qualname__}"
+            f"RustRecorder supports legacy LCM or registered CDR messages, got {payload_type.__qualname__}"
         )
 
     @staticmethod
@@ -276,6 +287,10 @@ def _validate_codec(stream_name: str, payload_type: type[Any], codec: str) -> No
                 f"Unsupported native codec {codec!r} for stream {stream_name!r}; "
                 f"choose one of {sorted(_SUPPORTED_NATIVE_CODECS)}"
             )
+        if codec == "cdr":
+            if cdr_name(payload_type) is None:
+                raise TypeError("The cdr codec requires a generated, registry-owned message type")
+            return
         if codec == "json" and payload_type is not String:
             raise TypeError("JSON codec requires std_msgs.String")
         if codec == "jpeg" and not issubclass(payload_type, Image):
diff --git a/dimos/experimental/memory/test_cdr_schema.py b/dimos/experimental/memory/test_cdr_schema.py
new file mode 100644
index 00000000000..4c81d3ef2e3
--- /dev/null
+++ b/dimos/experimental/memory/test_cdr_schema.py
@@ -0,0 +1,31 @@
+# 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.
+
+"""Recording reuses registered schemas without inspecting payload fields."""
+
+from dimos_cdrgen.registry import schema
+import pytest
+
+from dimos.experimental.memory.cdr_schema import cdr_name, recording_schema
+from dimos.messages.geometry_msgs.msg import PoseStamped
+
+
+def test_registry_owner_and_complete_schema_are_reused():
+    assert cdr_name(PoseStamped) == "geometry_msgs/msg/PoseStamped"
+    assert recording_schema(PoseStamped) == {
+        "name": "geometry_msgs/msg/PoseStamped",
+        "data": schema("geometry_msgs/msg/PoseStamped"),
+    }
+    with pytest.raises(TypeError, match="registry-owned"):
+        recording_schema(bytes)
diff --git a/dimos/experimental/memory/test_rust_cli_recorder.py b/dimos/experimental/memory/test_rust_cli_recorder.py
index 1b23bb3145e..447c3df7abc 100644
--- a/dimos/experimental/memory/test_rust_cli_recorder.py
+++ b/dimos/experimental/memory/test_rust_cli_recorder.py
@@ -27,9 +27,11 @@
 from dimos.core.global_config import global_config
 from dimos.core.transport import LCMTransport, ZenohTransport
 from dimos.experimental.memory import rust_cli_recorder
+from dimos.experimental.memory.cdr_schema import recording_schema
 from dimos.experimental.memory.rust_cli_recorder import RustRecordingPlan, RustRecordingSession
 from dimos.experimental.memory.rust_recorder import RustStreamSpec
 from dimos.memory.store.sqlite import SqliteStore
+from dimos.messages.geometry_msgs.msg import PoseStamped as GeneratedPoseStamped
 from dimos.msgs.geometry_msgs.PoseStamped import PoseStamped
 from dimos.msgs.sensor_msgs.Image import Image
 from dimos.protocol.pubsub.impl.zenohpubsub import Topic as ZenohTopic
@@ -359,3 +361,40 @@ def build(*args, **kwargs):
     assert rust_cli_recorder.prepare_rust_recorder() == executable
     assert popen.call_args.kwargs["cwd"] == str(tmp_path / "dimos/experimental/memory/rust")
     assert popen.call_args.args[0] == "nix build -L path:."
+
+
+@pytest.mark.parametrize("backend", ["lcm", "zenoh"])
+def test_plan_records_generated_messages_with_schema_on_plain_topic(backend):
+    transport = (
+        LCMTransport("pose", GeneratedPoseStamped, url="memq://")
+        if backend == "lcm"
+        else ZenohTransport("pose", GeneratedPoseStamped)
+    )
+    try:
+        plan = rust_cli_recorder.make_plan({("pose", GeneratedPoseStamped): transport})
+        assert plan.backend == backend
+        assert plan.topics == {"stream_0": "pose"}
+        assert plan.streams[0].codec == "cdr"
+        assert plan.streams[0].cdr_schema == recording_schema(GeneratedPoseStamped)
+    finally:
+        transport.stop()
+
+
+def test_prepare_uses_explicit_executable_without_build(tmp_path, monkeypatch, mocker):
+    executable = tmp_path / "recorder"
+    executable.write_text("#!/bin/sh\nexit 0\n")
+    executable.chmod(0o755)
+    monkeypatch.setattr(global_config, "record_executable", str(executable))
+    build = mocker.patch.object(rust_cli_recorder.subprocess, "Popen")
+    assert rust_cli_recorder.prepare_rust_recorder() == executable
+    build.assert_not_called()
+
+
+@pytest.mark.parametrize("exists", [False, True])
+def test_prepare_rejects_unusable_explicit_executable(tmp_path, monkeypatch, exists):
+    executable = tmp_path / "recorder"
+    if exists:
+        executable.touch()
+    monkeypatch.setattr(global_config, "record_executable", str(executable))
+    with pytest.raises(PermissionError if exists else FileNotFoundError):
+        rust_cli_recorder.prepare_rust_recorder()
diff --git a/dimos/experimental/memory/test_rust_recorder.py b/dimos/experimental/memory/test_rust_recorder.py
index f61140f2ee8..ae044f109ea 100644
--- a/dimos/experimental/memory/test_rust_recorder.py
+++ b/dimos/experimental/memory/test_rust_recorder.py
@@ -16,12 +16,15 @@
 
 from pathlib import Path
 from typing import Any, TypeVar
+from unittest.mock import Mock
 
 import pytest
 
 from dimos.core.global_config import GlobalConfig
 from dimos.core.module import Module
 from dimos.core.stream import In
+from dimos.core.transport import PubSubTransport
+from dimos.experimental.memory.cdr_schema import recording_schema
 from dimos.experimental.memory.rust_recorder import (
     RustMcapStoreConfig,
     RustRecorder,
@@ -31,8 +34,10 @@
 )
 from dimos.memory.store.sqlite import SqliteStore
 from dimos.memory.type.recording import OnExisting
+from dimos.messages.geometry_msgs.msg import PoseStamped as NativePoseStamped
 from dimos.msgs.geometry_msgs.PoseStamped import PoseStamped
 from dimos.msgs.sensor_msgs.Image import Image
+from dimos.protocol.rpc.zenohrpc import ZenohRPC
 
 
 class SampleRustRecorder(RustRecorder):
@@ -48,19 +53,14 @@ class UnsupportedRustRecorder(RustRecorder):
     values: In[Unsupported]
 
 
-class FakeTransport:
-    def __init__(self, channel: str) -> None:
-        self.channel = channel
-
-    def stop(self) -> None:
-        pass
-
-
 TRecorder = TypeVar("TRecorder", bound=RustRecorder)
 
 
 @pytest.fixture
-def make_recorder() -> Any:
+def make_recorder(mocker) -> Any:
+    mocker.patch.object(ZenohRPC, "start")
+    mocker.patch.object(ZenohRPC, "serve_module_rpc")
+    mocker.patch("dimos.core.module.get_loop", return_value=(None, None))
     recorders: list[RustRecorder] = []
 
     def make(recorder_type: type[TRecorder], **kwargs: Any) -> TRecorder:
@@ -75,7 +75,7 @@ def make(recorder_type: type[TRecorder], **kwargs: Any) -> TRecorder:
 
 def connect(recorder: RustRecorder, **channels: str) -> None:
     for name, channel in channels.items():
-        getattr(recorder, name).transport = FakeTransport(channel)  # type: ignore[assignment]
+        getattr(recorder, name).transport = Mock(spec=PubSubTransport, channel=channel)
 
 
 def test_specs_use_native_defaults_remapping_and_configured_workers(
@@ -209,7 +209,7 @@ def test_unsupported_python_payload_fails_before_native_process_starts(
     )
     connect(recorder, values="/values")
 
-    with pytest.raises(TypeError, match="only supports LCM-backed messages"):
+    with pytest.raises(TypeError, match="supports legacy LCM or registered CDR messages"):
         recorder._stream_specs()
 
 
@@ -369,3 +369,49 @@ def test_replay_never_prepares_native_sources_or_process(tmp_path, make_recorder
     spawn.assert_not_called()
     start.assert_called_once_with(recorder)
     assert not path.exists()
+
+
+class GeneratedRustRecorder(RustRecorder):
+    pose: In[NativePoseStamped]
+
+
+def test_generated_values_resolve_cdr_and_full_schema_without_legacy_type_copy(
+    tmp_path, make_recorder
+):
+    recorder = make_recorder(
+        GeneratedRustRecorder,
+        store=RustSqliteStoreConfig(path=str(tmp_path / "native.db")),
+        record_tf=False,
+    )
+    connect(recorder, pose="example/pose")
+    specs = recorder._stream_specs()
+    assert len(specs) == 1
+    assert specs[0].codec == "cdr"
+    assert specs[0].cdr_schema == recording_schema(NativePoseStamped)
+    assert (
+        specs[0].payload_type == f"{NativePoseStamped.__module__}.{NativePoseStamped.__qualname__}"
+    )
+    recorder._prepare_store(specs)
+    with SqliteStore(path=str(tmp_path / "native.db")) as store:
+        assert store.stream("pose").data_type is NativePoseStamped
+
+
+def test_generated_values_cannot_silently_select_a_legacy_wire_codec(tmp_path, make_recorder):
+    recorder = make_recorder(
+        GeneratedRustRecorder,
+        store=RustSqliteStoreConfig(path=str(tmp_path / "native.db")),
+        record_tf=False,
+        stream_codecs={"pose": "lcm"},
+    )
+    connect(recorder, pose="example/pose")
+    with pytest.raises(TypeError, match="LCM-backed"):
+        recorder._stream_specs()
+    assert not (tmp_path / "native.db").exists()
+
+
+def test_cdr_stream_schema_is_required_and_cannot_leak_into_other_codecs():
+    options = dict(port="pose", name="pose", payload_type="geometry_msgs/msg/PoseStamped")
+    with pytest.raises(ValueError, match="requires its resolved schema"):
+        RustStreamSpec(**options, codec="cdr")
+    with pytest.raises(ValueError, match="other codecs must not carry one"):
+        RustStreamSpec(**options, codec="lcm", cdr_schema=recording_schema(NativePoseStamped))
diff --git a/dimos/experimental/memory/test_rust_recorder_e2e.py b/dimos/experimental/memory/test_rust_recorder_e2e.py
index d41ebe3921e..50a24483d36 100644
--- a/dimos/experimental/memory/test_rust_recorder_e2e.py
+++ b/dimos/experimental/memory/test_rust_recorder_e2e.py
@@ -14,6 +14,7 @@
 
 from __future__ import annotations
 
+from contextlib import ExitStack
 import importlib.util
 import os
 from pathlib import Path
@@ -25,13 +26,14 @@
 from typing import cast
 import uuid
 
+from dimos_cdrgen.registry import encode as encode_cdr
 import numpy as np
 import pytest
 
 from dimos.constants import DIMOS_PROJECT_ROOT
 from dimos.core.global_config import global_config
 from dimos.core.stream import In
-from dimos.core.transport import LCMTransport, ZenohTransport
+from dimos.core.transport import LCMTransport, PubSubTransport, ZenohTransport
 from dimos.experimental.memory import rust_cli_recorder
 from dimos.experimental.memory.rust_cli_recorder import RustRecordingSession
 from dimos.experimental.memory.rust_recorder import (
@@ -42,15 +44,25 @@
 )
 from dimos.memory.codecs.lcm import LcmCodec
 from dimos.memory.codecs.lz4 import Lz4Codec
+from dimos.memory.store.cdr_mcap import CdrMcapStore
 from dimos.memory.store.mcap import McapStore
 from dimos.memory.store.sqlite import SqliteStore
 from dimos.memory.type.observation import Observation
+from dimos.messages.builtin_interfaces.msg import Time as NativeTime
+from dimos.messages.geometry_msgs.msg import (
+    Point as NativePoint,
+    Pose as NativePose,
+    PoseStamped as NativePoseStamped,
+    Quaternion as NativeQuaternion,
+)
+from dimos.messages.std_msgs.msg import Header as NativeHeader
 from dimos.msgs.geometry_msgs.Transform import Transform
 from dimos.msgs.geometry_msgs.Vector3 import Vector3
 from dimos.msgs.sensor_msgs.Image import Image, ImageFormat
 from dimos.msgs.sensor_msgs.Imu import Imu
 from dimos.msgs.tf2_msgs.TFMessage import TFMessage
-from dimos.protocol.pubsub.impl.zenohpubsub import Topic as ZenohTopic
+from dimos.protocol.pubsub.impl.zenohpubsub import Topic as ZenohTopic, ZenohPubSubBase
+from dimos.protocol.rpc.zenohrpc import ZenohRPC
 from dimos.protocol.service.zenohservice import ZenohConfig, ZenohSessionPool
 
 pytestmark = pytest.mark.self_hosted
@@ -420,3 +432,140 @@ def test_tf_records_over_zenoh_and_replays_through_python(
             "base_link",
             "camera",
         ]
+
+
+class CdrInteropRustRecorder(RustRecorder):
+    pose: In[NativePoseStamped]
+
+
+def _stop_owned_process(process):
+    if process.poll() is None:
+        process.terminate()
+        try:
+            process.wait(timeout=10)
+        except subprocess.TimeoutExpired:
+            process.kill()
+            process.wait(timeout=10)
+
+
+@pytest.mark.parametrize(
+    "store_kind",
+    [
+        "sqlite",
+        pytest.param(
+            "mcap", marks=pytest.mark.skipif(not _MCAP_AVAILABLE, reason="mcap not installed")
+        ),
+    ],
+)
+def test_cdr_native_artifact_replays_raw_payload_with_reception_time(
+    tmp_path,
+    rust_recorder_executable,
+    store_kind,
+    monkeypatch,
+    mocker,
+):
+    artifact = tmp_path / ("native.db" if store_kind == "sqlite" else "native.mcap")
+    store = (
+        RustSqliteStoreConfig(path=str(artifact))
+        if store_kind == "sqlite"
+        else RustMcapStoreConfig(path=str(artifact))
+    )
+    # RPC is unrelated to this storage acceptance; the actual raw pub/sub and
+    # native process below remain real and bound only to loopback.
+    mocker.patch.object(ZenohRPC, "start")
+    mocker.patch.object(ZenohRPC, "serve_module_rpc")
+    monkeypatch.setattr(global_config, "transport", "zenoh")
+    endpoint = f"tcp/127.0.0.1:{_free_port()}"
+    channel = "example/pose/" + uuid.uuid4().hex
+    with ExitStack() as resources:
+        recorder = CdrInteropRustRecorder(
+            source_dir=None,
+            build_command=None,
+            executable=str(rust_recorder_executable),
+            store=store,
+            record_tf=False,
+            encoding_threads=2,
+            session=ZenohConfig(
+                mode="peer",
+                connect=[],
+                listen=[endpoint],
+                multicast=False,
+                gossip=False,
+                connect_timeout=0,
+            ),
+        )
+        resources.callback(recorder.stop)
+        recorder.pose.transport = mocker.Mock(spec=PubSubTransport, channel=channel)
+        specs = recorder._stream_specs()
+        assert specs[0].codec == "cdr"
+        recorder._prepare_store(specs)
+        recorder.config.streams = specs
+        pool = ZenohSessionPool()
+        resources.callback(pool.close_all)
+        publisher = ZenohPubSubBase(
+            session_pool=pool,
+            mode="client",
+            connect=[endpoint],
+            multicast=False,
+            gossip=False,
+            scouting=False,
+            connect_timeout=5,
+        )
+        resources.callback(publisher.stop)
+        process = resources.enter_context(
+            subprocess.Popen(
+                [str(rust_recorder_executable)],
+                cwd=_RUST_PACKAGE,
+                env={
+                    **os.environ,
+                    "DIMOS_TRANSPORT": "zenoh",
+                    "RUST_LOG": "info,dimos_memory_recorder=debug",
+                },
+                stdin=subprocess.PIPE,
+                stdout=subprocess.DEVNULL,
+                stderr=subprocess.PIPE,
+            )
+        )
+        resources.callback(_stop_owned_process, process)
+        assert process.stdin is not None
+        process.stdin.write(recorder._stdin_blob({"pose": channel}))
+        process.stdin.close()
+        _wait_for_log(process, "memory recorder ready")
+        publisher.start()
+        expected = NativePoseStamped(
+            NativeHeader(NativeTime(1_700_000_000, 123_456_789), "map"),
+            NativePose(NativePoint(1.5, -2.0, 3.0), NativeQuaternion(0.0, 0.0, 0.0, 1.0)),
+        )
+        payload = encode_cdr(expected)
+        received_after = time.time()
+        publisher.publish(ZenohTopic(channel), payload)
+        _wait_for_log(process, "memory recorder batch written")
+        process.terminate()
+        assert process.wait(timeout=10) == 0
+    memory = (
+        SqliteStore(path=str(artifact))
+        if store_kind == "sqlite"
+        else CdrMcapStore(path=str(artifact))
+    )
+    with memory:
+        observations = list(memory.stream("pose"))
+        assert len(observations) == 1
+        actual = observations[0]
+        assert received_after <= actual.ts <= time.time()
+        assert type(actual.data) is NativePoseStamped
+        assert type(actual.data.header) is NativeHeader
+        assert actual.data == expected
+        assert encode_cdr(actual.data) == payload
+    if store_kind == "mcap":
+        # MCAP is optional; this parametrization is skipped when it is absent.
+        reader_module = importlib.import_module("mcap.reader")
+        with artifact.open("rb") as source:
+            records = list(reader_module.make_reader(source).iter_messages())
+        assert len(records) == 1
+        actual_schema, actual_channel, message = records[0]
+        assert message.publish_time == message.log_time
+        assert message.data == payload
+        assert actual_channel.message_encoding == "cdr"
+        assert actual_schema.encoding == "ros2msg"
+        assert actual_schema.name == "geometry_msgs/msg/PoseStamped"
+        assert actual_schema.data.decode() == specs[0].cdr_schema["data"]
diff --git a/dimos/memory/codecs/base.py b/dimos/memory/codecs/base.py
index a33667e36f5..5fd6342b9c8 100644
--- a/dimos/memory/codecs/base.py
+++ b/dimos/memory/codecs/base.py
@@ -17,6 +17,8 @@
 import importlib
 from typing import Any, Protocol, TypeVar, runtime_checkable
 
+from dimos.memory.codecs.cdr import CdrCodec
+
 T = TypeVar("T")
 
 
@@ -38,6 +40,8 @@ def codec_for(payload_type: type[Any] | None = None, sample: Any = None) -> Code
     from dimos.memory.codecs.pickle import PickleCodec
 
     if payload_type is not None:
+        if hasattr(payload_type, "__msgtype__"):
+            return CdrCodec(payload_type)
         from dimos.msgs.sensor_msgs.Image import Image, ImageFormat
 
         if issubclass(payload_type, Image):
@@ -115,6 +119,8 @@ def _make_one(name: str, payload_module: str, inner: Codec[Any] | None = None) -
         from dimos.memory.codecs.jpeg import JpegCodec
 
         return JpegCodec()
+    if name == "cdr":
+        return CdrCodec(resolve_payload_type(payload_module))
     if name == "lcm":
         from dimos.memory.codecs.lcm import LcmCodec
 
diff --git a/dimos/memory/codecs/cdr.py b/dimos/memory/codecs/cdr.py
new file mode 100644
index 00000000000..25771667631
--- /dev/null
+++ b/dimos/memory/codecs/cdr.py
@@ -0,0 +1,42 @@
+# 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.
+
+"""CDR storage for installed generated values; independent of legacy codecs."""
+
+from __future__ import annotations
+
+from typing import Generic, TypeVar
+
+from dimos.protocol.pubsub.codec import describe_message
+
+T = TypeVar("T")
+
+
+class CdrCodec(Generic[T]):
+    def __init__(self, msg_type: type[T]) -> None:
+        descriptor = describe_message(msg_type)
+        if descriptor.encoding != "cdr":
+            raise TypeError(f"{msg_type!r} is not an installed generated CDR message type")
+        self._msg_type = msg_type
+        self._codec = descriptor.codec
+
+    @property
+    def payload_type(self) -> type[T]:
+        return self._msg_type
+
+    def encode(self, value: T) -> bytes:
+        return self._codec.encode(value)
+
+    def decode(self, data: bytes) -> T:
+        return self._codec.decode(data)
diff --git a/dimos/memory/store/cdr_mcap.py b/dimos/memory/store/cdr_mcap.py
new file mode 100644
index 00000000000..c1d1225e84d
--- /dev/null
+++ b/dimos/memory/store/cdr_mcap.py
@@ -0,0 +1,255 @@
+# 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.
+
+"""Read-only CDR MCAP storage using embedded schemas and installed generated types."""
+
+from __future__ import annotations
+
+from collections.abc import Iterator
+from dataclasses import replace
+from functools import partial
+from typing import Any, Protocol, runtime_checkable
+
+from dimos_cdrgen.registry import message_types, schema as cdr_schema
+
+from dimos.memory.backend import Backend
+from dimos.memory.codecs.base import codec_for
+from dimos.memory.codecs.cdr import CdrCodec
+from dimos.memory.notifier.subject import SubjectNotifier
+from dimos.memory.observationstore.base import ObservationStore, ObservationStoreConfig
+from dimos.memory.store.base import Store, StoreConfig
+from dimos.memory.type.filter import StreamQuery
+from dimos.memory.type.observation import Observation
+
+
+@runtime_checkable
+class StreamCodec(Protocol):
+    """What the store needs to turn a channel's stored bytes into a payload."""
+
+    @property
+    def payload_type(self) -> type: ...
+
+    def decode(self, data: bytes) -> Any: ...
+
+
+class _BytesCodec:
+    """Identity codec: hands back a codecless channel's stored bytes as ``Stream[bytes]``."""
+
+    payload_type = bytes
+
+    def decode(self, data: bytes) -> bytes:
+        return data
+
+
+_BYTES_CODEC = _BytesCodec()
+
+
+def _slug(topic: str) -> str:
+    return topic.removeprefix("rt/").lstrip("/").replace("/", "_")
+
+
+class CdrMcapObservationStoreConfig(ObservationStoreConfig):
+    name: str = "<mcap>"
+
+
+class CdrMcapObservationStore(ObservationStore[Any]):
+    """Read-only metadata/query over one mcap channel. Payloads load lazily."""
+
+    config: CdrMcapObservationStoreConfig
+
+    def __init__(
+        self,
+        *,
+        name: str,
+        path: str,
+        topic: str,
+        codec: StreamCodec,
+        count: int,
+        observation_uses_publish_time: bool,
+    ) -> None:
+        super().__init__(name=name)
+        self._path = path
+        self._topic = topic
+        self._codec = codec
+        self._count = count
+        # Immutable channel metadata: this store never writes and each iterator
+        # owns its own file reader, so timestamp selection has no async state.
+        self._observation_uses_publish_time = observation_uses_publish_time
+
+    @property
+    def name(self) -> str:
+        return self.config.name
+
+    def _iter(self, reverse: bool = False) -> Iterator[Observation[Any]]:
+        from mcap.reader import make_reader  # optional mcap dependency
+
+        decode, dtype, n = self._codec.decode, self._codec.payload_type, self._count
+        with open(self._path, "rb") as f:
+            msgs = make_reader(f).iter_messages(topics=[self._topic], reverse=reverse)
+            for i, (_schema, _channel, message) in enumerate(msgs):
+                observation_time = (
+                    message.publish_time
+                    if self._observation_uses_publish_time
+                    else message.log_time
+                )
+                yield Observation(
+                    id=(n - 1 - i) if reverse else i,
+                    ts=observation_time / 1e9,
+                    data_type=dtype,
+                    _loader=partial(decode, message.data),
+                )
+
+    def query(self, q: StreamQuery) -> Iterator[Observation[Any]]:
+        # MCAP is natively log-time ordered, so id ordering never needs a sort.
+        # Native DimOS recordings expose publish_time as observation ts; source
+        # time can differ from log/reception order and must use the generic sort.
+        if q.order_field == "id" or (
+            q.order_field == "ts" and not self._observation_uses_publish_time
+        ):
+            it = self._iter(reverse=q.order_desc)
+            q = replace(q, order_field=None, order_desc=False)
+            return q.apply(it)
+        return q.apply(self._iter())
+
+    def count(self, q: StreamQuery) -> int:
+        if not q.filters and q.search_text is None and q.search_vec is None:
+            n = self._count
+            if q.offset_val:
+                n = max(0, n - q.offset_val)
+            if q.limit_val is not None:
+                n = min(n, q.limit_val)
+            return n
+        return sum(1 for _ in self.query(q))
+
+    def fetch_by_ids(self, ids: list[int]) -> list[Observation[Any]]:
+        want = set(ids)
+        return [o for o in self._iter() if o.id in want]
+
+    def insert(self, obs: Observation[Any]) -> int:
+        raise NotImplementedError("CdrMcapStore is read-only")
+
+
+class CdrMcapStoreConfig(StoreConfig):
+    path: str = ""
+
+
+class CdrMcapStore(Store):
+    """A memory store backed by an mcap file (read-only).
+
+    Only CDR channels are accepted. Unknown schemas are exposed as raw bytes. Names default to
+    the slugified topic (see :func:`_slug`); ``streams`` (friendly name -> topic)
+    overrides the name for specific topics.
+    """
+
+    config: CdrMcapStoreConfig
+
+    def __init__(
+        self,
+        *,
+        streams: dict[str, str] | None = None,
+        **kwargs: Any,
+    ) -> None:
+        from mcap.reader import make_reader  # optional mcap dependency
+
+        super().__init__(**kwargs)
+        self._codecs: dict[str, StreamCodec] = {}
+        name_of = {topic: name for name, topic in (streams or {}).items()}  # topic -> override
+        with open(self.config.path, "rb") as f:
+            summary = make_reader(f).get_summary()
+        self._stream_topic: dict[str, str] = {}  # stream name -> topic
+        self._available: dict[str, int] = {}  # stream name -> message count
+        self._observation_uses_publish_time: dict[str, bool] = {}
+        # Channels with no registered codec are still exposed, as Stream[bytes] via
+        # _BYTES_CODEC — reachable but undecoded. _raw maps their stream name to the
+        # source schema so summary() can flag them [raw bytes: <schema>].
+        self._raw: dict[str, str | None] = {}  # raw stream name -> source schema
+        channel_signatures: dict[str, tuple[Any, ...]] = {}
+        if summary is not None and summary.statistics is not None:
+            for cid, ch in summary.channels.items():
+                if ch.message_encoding != "cdr":
+                    raise ValueError(f"CdrMcapStore requires CDR channels: {ch.topic!r}")
+                declared = summary.schemas.get(ch.schema_id)
+                signature = (
+                    ch.message_encoding,
+                    (declared.name, declared.encoding, declared.data) if declared else None,
+                    ch.metadata.get("dimos.observation_time"),
+                )
+                if ch.topic in channel_signatures and channel_signatures[ch.topic] != signature:
+                    raise ValueError(
+                        f"Conflicting schemas or timing policies for topic {ch.topic!r}"
+                    )
+                channel_signatures[ch.topic] = signature
+                count = summary.statistics.channel_message_counts.get(cid, 0)
+                name = name_of.get(ch.topic) or _slug(ch.topic)
+                taken = self._stream_topic.get(name)
+                if taken is not None and taken != ch.topic:
+                    raise ValueError(
+                        f"stream {name!r} would name both {taken!r} and {ch.topic!r};"
+                        " pass streams= to name them apart"
+                    )
+                if ch.topic not in self._codecs:
+                    # Unknown schemas remain available as raw bytes.
+                    self._codecs[ch.topic] = _BYTES_CODEC
+                    self._raw[name] = declared.name if declared else None
+                    if declared is not None and declared.encoding == "ros2msg":
+                        kind = message_types().get(declared.name)
+                        if kind is not None:
+                            if declared.data.decode("utf-8") != cdr_schema(declared.name):
+                                raise ValueError(f"CDR schema mismatch: {declared.name}")
+                            self._codecs[ch.topic] = CdrCodec(kind)
+                            self._raw.pop(name, None)
+                self._stream_topic[name] = ch.topic
+                self._available[name] = self._available.get(name, 0) + count
+                self._observation_uses_publish_time[name] = (
+                    ch.metadata.get("dimos.observation_time") == "publish_time"
+                )
+
+    def list_streams(self) -> list[str]:
+        return sorted(set(self._available) | set(self._streams))
+
+    def summary(self) -> str:
+        """Base summary, tagging codecless streams with ``[raw bytes: <schema>]``."""
+        lines = []
+        for name, stream in self.streams.items():
+            line = stream.summary()  # "Stream(\"name\"): ..."
+            if name in self._raw:
+                head = str(stream)  # "Stream(\"name\")"
+                line = f"{head} [raw bytes: {self._raw[name] or '?'}]{line[len(head) :]}"
+            lines.append(line)
+        return "\n".join(lines)
+
+    def _create_backend(
+        self, name: str, payload_type: type | None = None, **config: Any
+    ) -> Backend[Any]:
+        if name not in self._available:
+            raise KeyError(f"No stream {name!r}. Available: {sorted(self._available)}")
+        topic = self._stream_topic[name]
+        codec = self._codecs[topic]
+        ptype = codec.payload_type
+        obs = CdrMcapObservationStore(
+            name=name,
+            path=self.config.path,
+            topic=topic,
+            codec=codec,
+            count=self._available[name],
+            observation_uses_publish_time=self._observation_uses_publish_time[name],
+        )
+        return Backend(
+            metadata_store=obs,
+            codec=codec_for(ptype),  # storage codec, unused (blob_store=None)
+            data_type=ptype,
+            blob_store=None,
+            vector_store=None,
+            notifier=SubjectNotifier(),
+        )
diff --git a/dimos/memory/store/test_cdr_recording.py b/dimos/memory/store/test_cdr_recording.py
new file mode 100644
index 00000000000..aab309c06b3
--- /dev/null
+++ b/dimos/memory/store/test_cdr_recording.py
@@ -0,0 +1,140 @@
+# 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.
+
+"""Read-only CDR stores decode self-describing MCAP recordings."""
+
+from contextlib import contextmanager
+
+from dimos_cdrgen.registry import encode, schema
+from mcap.writer import Writer
+import pytest
+
+from dimos.memory.codecs.base import codec_for, codec_from_id, codec_id
+from dimos.memory.codecs.cdr import CdrCodec
+from dimos.memory.store.cdr_mcap import CdrMcapStore
+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
+
+
+def message(sec=17):
+    return LineSegments3D(
+        header=Header(stamp=Time(sec=sec, nanosec=250_000_000), frame_id="imu"),
+        segments=[LineSegment3D(Point(1.0, 2.0, 3.0), Point(4.0, 5.0, 6.0), 0.75)],
+    )
+
+
+@contextmanager
+def mcap_fixture(path):
+    with path.open("wb") as output:
+        writer = Writer(output)
+        writer.start(profile="ros2")
+        try:
+            yield writer
+        finally:
+            writer.finish()
+
+
+def test_cdr_codec_restores_generated_values():
+    expected = message()
+    codec = codec_for(LineSegments3D)
+    assert isinstance(codec, CdrCodec)
+    assert codec_id(codec) == "cdr"
+    rebuilt = codec_from_id("cdr", f"{LineSegments3D.__module__}.{LineSegments3D.__qualname__}")
+    assert rebuilt.decode(encode(expected)) == expected
+    with pytest.raises(TypeError, match="declared type"):
+        codec.encode(Header(Time(0, 0), "wrong-type"))
+    with pytest.raises(TypeError, match="Unsupported"):
+        CdrCodec(str)
+
+
+@pytest.mark.parametrize("payload", [b"malformed", b"\x00\x07\x00\x00" + bytes(4)])
+def test_cdr_storage_rejects_invalid_encapsulation(payload):
+    with pytest.raises(ValueError, match="XCDR1 encapsulation"):
+        CdrCodec(LineSegments3D).decode(payload)
+
+
+def test_mismatched_known_schema_is_rejected_before_decoding(tmp_path):
+    path = tmp_path / "bad.mcap"
+    with mcap_fixture(path) as writer:
+        schema_id = writer.register_schema(
+            LineSegments3D.__msgtype__, "ros2msg", b"int32 changed\n"
+        )
+        channel = writer.register_channel("imu", "cdr", schema_id)
+        writer.add_message(channel, log_time=1, publish_time=1, data=encode(message()))
+    with pytest.raises(ValueError, match="CDR schema mismatch"):
+        CdrMcapStore(path=str(path))
+
+
+def test_unknown_cdr_schema_never_uses_a_legacy_topic_decoder(tmp_path):
+    path = tmp_path / "unknown.mcap"
+    with mcap_fixture(path) as writer:
+        schema_id = writer.register_schema("future_msgs/msg/New", "ros2msg", b"uint32 value\n")
+        channel = writer.register_channel("dimos/imu/sensor_msgs.Imu", "cdr", schema_id)
+        writer.add_message(channel, log_time=1, publish_time=1, data=b"unknown bytes")
+    with CdrMcapStore(path=str(path)) as store:
+        observations = list(store.stream("dimos_imu_sensor_msgs.Imu"))
+        assert observations[0].data == b"unknown bytes"
+        assert "raw bytes: future_msgs/msg/New" in store.summary()
+
+
+def test_conflicting_channel_schemas_are_rejected(tmp_path):
+    path = tmp_path / "conflict.mcap"
+    with mcap_fixture(path) as writer:
+        for text in (b"uint32 value\n", b"uint64 value\n"):
+            schema_id = writer.register_schema("future_msgs/msg/New", "ros2msg", text)
+            channel = writer.register_channel("same", "cdr", schema_id)
+            writer.add_message(channel, log_time=1, publish_time=1, data=b"")
+    with pytest.raises(ValueError, match="Conflicting schemas"):
+        CdrMcapStore(path=str(path))
+
+
+def test_cdr_store_rejects_non_cdr_channels(tmp_path):
+    path = tmp_path / "legacy.mcap"
+    with mcap_fixture(path) as writer:
+        channel = writer.register_channel("imu", "lcm", 0)
+        writer.add_message(channel, log_time=1, publish_time=1, data=b"legacy")
+    with pytest.raises(ValueError, match="requires CDR channels"):
+        CdrMcapStore(path=str(path))
+
+
+def test_cdr_store_remaps_topics_and_queries_source_time(tmp_path):
+    path = tmp_path / "source-order.mcap"
+    with mcap_fixture(path) as writer:
+        schema_id = writer.register_schema(
+            LineSegments3D.__msgtype__, "ros2msg", schema(LineSegments3D.__msgtype__).encode()
+        )
+        channel = writer.register_channel(
+            "/sensors/segments",
+            "cdr",
+            schema_id,
+            metadata={"dimos.observation_time": "publish_time"},
+        )
+        for log_ns, sec in ((1, 20), (2, 10)):
+            writer.add_message(
+                channel,
+                log_time=log_ns,
+                publish_time=sec * 1_000_000_000,
+                data=encode(message(sec)),
+            )
+    with CdrMcapStore(path=str(path), streams={"segments": "/sensors/segments"}) as store:
+        stream = store.stream("segments")
+        assert store.list_streams() == ["segments"]
+        assert stream.count() == 2
+        assert stream.order_by("ts").first().data == message(10)
+        assert stream.order_by("ts", desc=True).first().data == message(20)
+        assert stream.order_by("id", desc=True).first().data == message(10)
+        with pytest.raises(NotImplementedError, match="read-only"):
+            stream.append(message(30), ts=30)
diff --git a/dimos/navigation/trajectory_follower/fancy/rust/flake.lock b/dimos/navigation/trajectory_follower/fancy/rust/flake.lock
index 9a4b4df94e3..eeb07ccc345 100644
--- a/dimos/navigation/trajectory_follower/fancy/rust/flake.lock
+++ b/dimos/navigation/trajectory_follower/fancy/rust/flake.lock
@@ -90,11 +90,11 @@
       },
       "locked": {
         "dir": "dimos/navigation/local_planner/rust",
-        "lastModified": 1792533441,
-        "narHash": "sha256-XF2lT9/357inKL+G8SmftVXicRfeMugelCoPDB8XOOc=",
+        "lastModified": 1792745145,
+        "narHash": "sha256-cJLM/uV6rBI3BAo0B8Tn/F225s48HdNIu7cKaLLC7PE=",
         "owner": "dimensionalOS",
         "repo": "dimos",
-        "rev": "b173fa6dcd3d7046429e6405b3826f8b96952cf8",
+        "rev": "e5c07673474aabf2256f53fd660511514610b91e",
         "type": "github"
       },
       "original": {
diff --git a/examples/cdr-recording/README.md b/examples/cdr-recording/README.md
new file mode 100644
index 00000000000..ede61c00def
--- /dev/null
+++ b/examples/cdr-recording/README.md
@@ -0,0 +1,87 @@
+# Record a standard ROS 2 message
+
+**WIP:** this example tests the CDR recording path. Business blueprints have not
+been migrated. No robot, ROS installation, or API key is required.
+
+The example publishes `geometry_msgs/msg/PoseStamped` at 10 Hz over Zenoh. Its
+pose moves around a one-metre circle in the `map` frame. The normal DimOS CLI
+starts the Rust recorder, which stores the original CDR bytes in MCAP. Python
+does not write the recording.
+
+## Install from this PR
+
+Use Linux, Python 3.12, [uv](https://docs.astral.sh/uv/getting-started/installation/),
+and a current stable [Rust toolchain](https://rustup.rs/). The native build needs
+a C/C++ compiler, CMake, pkg-config, and NASM. On Ubuntu:
+
+```bash
+sudo apt-get install build-essential cmake pkg-config nasm libturbojpeg0 libgl1 libglib2.0-0 git-lfs
+git lfs install
+git clone --branch cc/feat/cdr-unit-recording https://github.com/dimensionalOS/dimos.git dimos-cdr-review
+cd dimos-cdr-review
+```
+
+If you already checked out this PR, start at its repository root instead. All
+commands below run from that root; no earlier PR checkout is needed.
+
+```bash
+uv venv .venv-cdr-recording --python 3.12
+DIMOS_ALLOW_MISSING_COCKPIT=1 uv pip install --python .venv-cdr-recording/bin/python -e . -e examples/cdr-recording
+source .venv-cdr-recording/bin/activate
+cargo build --locked --manifest-path dimos/experimental/memory/rust/Cargo.toml --target-dir build/cdr-recorder -j2
+```
+
+The install generates the built-in messages. `DIMOS_ALLOW_MISSING_COCKPIT=1`
+allows this headless example to install without a built web frontend. The small
+example package registers a blueprint with the existing `dimos.blueprints` entry
+point mechanism.
+
+## Record
+
+```bash
+dimos --transport zenoh --viewer none --n-workers 1 \
+  --record mcap --record-engine rust --record-topics pose \
+  --record-executable "$PWD/build/cdr-recorder/debug/dimos-memory-recorder" \
+  run dimos-cdr-recording-example.pose
+```
+
+If DimOS offers to raise the shared-memory limit, answer **N**. This small
+example works without that system change.
+
+Wait for `Experimental Rust recorder ready`. Let the example run for about
+15 seconds, then press **Ctrl+C once**. Wait for shutdown to finish. The
+`Recording saved` log prints the full `artifact_path` to `memory.mcap`.
+
+The recorder binary comes from this checkout. `--record-executable` selects it
+without invoking the default Nix build. Zenoh uses the normal local discovery
+settings; this example does not connect to robot hardware.
+
+## Inspect in Foxglove
+
+1. Open the printed `memory.mcap` file in Foxglove.
+2. Add a **Raw Messages** panel and select the `pose` topic.
+3. Confirm the schema is `geometry_msgs/msg/PoseStamped`, `header.frame_id` is
+   `map`, and `pose.position.x` and `pose.position.y` change during playback.
+4. Add a **3D** panel. Set the display frame to `map` and enable `pose`.
+5. Play the recording. The pose arrow moves around a circle. No TF topic is
+   required because the pose and display frame are the same.
+
+The MCAP contains a `ros2msg` schema with the PoseStamped definition and all its
+nested definitions. Its channel uses `cdr` encoding. Foxglove reads that schema
+and decodes the recorded bytes; it does not need the DimOS Python package.
+
+Recording times use reception time. The original `header.stamp` remains in the
+payload. See [Foxglove's supported pose types](https://docs.foxglove.dev/docs/visualization/panels/3d#pose).
+
+## Read with Python
+
+Replace the path below with the printed `artifact_path`:
+
+```python
+from dimos.memory.store.cdr_mcap import CdrMcapStore
+
+with CdrMcapStore(path="/path/from/log/memory.mcap") as store:
+    print(store.list_streams())  # ['pose']
+    observation = next(iter(store.stream("pose")))
+    print(observation.ts, observation.data)
+```
diff --git a/examples/cdr-recording/cdr_recording_example.py b/examples/cdr-recording/cdr_recording_example.py
new file mode 100644
index 00000000000..2f8eac9303b
--- /dev/null
+++ b/examples/cdr-recording/cdr_recording_example.py
@@ -0,0 +1,53 @@
+# 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.
+
+"""Publish a standard ROS 2 pose for the CLI Rust recorder."""
+
+import math
+import time
+
+from reactivex import interval
+
+from dimos.core.core import rpc
+from dimos.core.module import Module
+from dimos.core.stream import Out
+from dimos.messages.builtin_interfaces.msg import Time
+from dimos.messages.geometry_msgs.msg import Point, Pose, PoseStamped, Quaternion
+from dimos.messages.std_msgs.msg import Header
+
+
+class MovingPose(Module):
+    """Publish a pose moving around a one-metre circle at 10 Hz."""
+
+    pose: Out[PoseStamped]
+
+    @rpc
+    def start(self) -> None:
+        super().start()
+        self.register_disposable(interval(0.1).subscribe(self._publish))
+
+    def _publish(self, tick: int) -> None:
+        angle = tick * 0.05
+        sec, nanosec = divmod(time.time_ns(), 1_000_000_000)
+        self.pose.publish(
+            PoseStamped(
+                header=Header(stamp=Time(sec=sec, nanosec=nanosec), frame_id="map"),
+                pose=Pose(
+                    position=Point(x=math.cos(angle), y=math.sin(angle), z=0.0),
+                    orientation=Quaternion(
+                        x=0.0, y=0.0, z=math.sin(angle / 2), w=math.cos(angle / 2)
+                    ),
+                ),
+            )
+        )
diff --git a/examples/cdr-recording/pyproject.toml b/examples/cdr-recording/pyproject.toml
new file mode 100644
index 00000000000..9c5367471b7
--- /dev/null
+++ b/examples/cdr-recording/pyproject.toml
@@ -0,0 +1,15 @@
+[build-system]
+requires = ["setuptools>=70"]
+build-backend = "setuptools.build_meta"
+
+[project]
+name = "dimos-cdr-recording-example"
+version = "0.1.0"
+requires-python = ">=3.12"
+dependencies = ["dimos", "mcap>=1.2"]
+
+[project.entry-points."dimos.blueprints"]
+pose = "cdr_recording_example:MovingPose"
+
+[tool.setuptools]
+py-modules = ["cdr_recording_example"]