diff --git a/.DS_Store b/.DS_Store
new file mode 100644
index 0000000..b9492ea
Binary files /dev/null and b/.DS_Store differ
diff --git a/Cargo.lock b/Cargo.lock
index 6e00d24..bc68e00 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -3141,14 +3141,15 @@ dependencies = [
 
 [[package]]
 name = "phoxal"
-version = "0.4.0"
+version = "0.5.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "069b276097c1231660a800e2fc27719cf4cc9d42e0f8928589d96da8db877847"
+checksum = "28e98fd325a693585a7ba0db4dfbffa940b913217630bd412c3e82f077f77ce4"
 dependencies = [
  "anyhow",
  "arc-swap",
  "async-trait",
  "clap",
+ "const_format",
  "derive-new",
  "derive_setters",
  "dotenvy",
diff --git a/Cargo.toml b/Cargo.toml
index 7ed2f39..3adc032 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -22,7 +22,7 @@ serde = { version = "1.0.228", features = ["derive"] }
 tokio = { version = "1", features = ["macros", "rt", "rt-multi-thread", "time", "signal", "sync", "fs"] }
 tracing = "0.1.44"
 urdf-rs = "0.9.0"
-phoxal = { version = "0.4.0" }
+phoxal = { version = "0.5.0" }
 
 [workspace.lints.rust]
 unused = "warn"
diff --git a/src/.DS_Store b/src/.DS_Store
new file mode 100644
index 0000000..693a1c6
Binary files /dev/null and b/src/.DS_Store differ
diff --git a/src/assets.rs b/src/assets.rs
index 8370213..6d91406 100644
--- a/src/assets.rs
+++ b/src/assets.rs
@@ -1,7 +1,10 @@
 use std::collections::{BTreeSet, HashMap};
 
 use anyhow::{Result, anyhow};
-use phoxal::api::asset::v1::{GetRequest as AssetRequest, GetResponse as AssetResponse};
+use phoxal::api::v1::{
+    asset::{GetRequest as AssetRequest, GetResponse as AssetResponse},
+    topic,
+};
 use phoxal::bus::Bus;
 use phoxal::bus::query::Retry;
 use phoxal::model::structure::Structure;
@@ -10,7 +13,14 @@ use tracing::warn;
 use urdf_rs::Geometry;
 
 pub(crate) async fn fetch_asset(bus: &Bus, path: &str, retry: &Retry) -> Result<Vec<u8>> {
-    match phoxal::api::asset::v1::get::query(bus, &AssetRequest::new(path), retry).await? {
+    match bus
+        .request(
+            &topic::new().v1().asset().get(),
+            &AssetRequest::new(path),
+            retry,
+        )
+        .await?
+    {
         Some(AssetResponse::Ok { bytes }) => Ok(bytes),
         Some(AssetResponse::NotFound) => Err(anyhow!("asset '{path}' was not found")),
         Some(AssetResponse::InvalidPath(reason)) => {
diff --git a/src/capability/accelerometer.rs b/src/capability/accelerometer.rs
index d8e3cb1..2f37e12 100644
--- a/src/capability/accelerometer.rs
+++ b/src/capability/accelerometer.rs
@@ -1,6 +1,6 @@
 use anyhow::Result;
-use phoxal::api::component::v1::capability::accelerometer::Sample;
-use phoxal::bus::pubsub::Stamped;
+use phoxal::api::v1::component::capability::accelerometer::Sample;
+use phoxal::bus::typed::Received;
 use rerun::{Arrows3D, RecordingStream};
 
 use crate::logging::helpers::set_logical_time;
@@ -8,12 +8,12 @@ use crate::logging::helpers::set_logical_time;
 pub(crate) fn log(
     rec: &RecordingStream,
     capability_entity_path: &str,
-    stamped: &Stamped<Sample>,
+    received: &Received<Sample>,
 ) -> Result<()> {
-    set_logical_time(rec, stamped.timestamp_ns);
+    set_logical_time(rec, received.at_ns);
     rec.log(
         format!("{capability_entity_path}/acceleration").as_str(),
-        &Arrows3D::from_vectors([*stamped.data.linear_acceleration()])
+        &Arrows3D::from_vectors([*received.value.linear_acceleration()])
             .with_origins([[0.0, 0.0, 0.0]]),
     )?;
     Ok(())
diff --git a/src/capability/battery.rs b/src/capability/battery.rs
index ab807ee..8561035 100644
--- a/src/capability/battery.rs
+++ b/src/capability/battery.rs
@@ -1,6 +1,6 @@
 use anyhow::Result;
-use phoxal::api::component::v1::capability::battery::State;
-use phoxal::bus::pubsub::Stamped;
+use phoxal::api::v1::component::capability::battery::State;
+use phoxal::bus::typed::Received;
 use rerun::{RecordingStream, Scalars};
 
 use crate::logging::helpers::set_logical_time;
@@ -8,12 +8,12 @@ use crate::logging::helpers::set_logical_time;
 pub(crate) fn log(
     rec: &RecordingStream,
     capability_entity_path: &str,
-    stamped: &Stamped<State>,
+    received: &Received<State>,
 ) -> Result<()> {
-    set_logical_time(rec, stamped.timestamp_ns);
+    set_logical_time(rec, received.at_ns);
     rec.log(
         format!("{capability_entity_path}/battery_percentage").as_str(),
-        &Scalars::new([f64::from(stamped.data.percentage())]),
+        &Scalars::new([f64::from(received.value.percentage())]),
     )?;
     Ok(())
 }
diff --git a/src/capability/camera.rs b/src/capability/camera.rs
index e737ee7..e7f38df 100644
--- a/src/capability/camera.rs
+++ b/src/capability/camera.rs
@@ -1,6 +1,6 @@
 use anyhow::{Result, anyhow, ensure};
-use phoxal::api::component::v1::capability::camera::{Encoding, Frame};
-use phoxal::bus::pubsub::Stamped;
+use phoxal::api::v1::component::capability::camera::{Encoding, Frame};
+use phoxal::bus::typed::Received;
 use phoxal::model::component::v1::capability::Capability;
 use rerun::{Image, RecordingStream};
 
@@ -12,12 +12,12 @@ pub(crate) fn log(
     rec: &RecordingStream,
     scene: &SceneModel,
     capability_id: &str,
-    stamped: &Stamped<Frame>,
+    received: &Received<Frame>,
 ) -> Result<()> {
-    let frame = &stamped.data;
+    let frame = &received.value;
     let resolution = [frame.width(), frame.height()];
 
-    set_logical_time(rec, stamped.timestamp_ns);
+    set_logical_time(rec, received.at_ns);
     let image_entity_path = scene.image_entity_path(capability_id)?;
     if let Some(pinhole) = camera_pinhole(scene, capability_id, resolution)? {
         rec.log(image_entity_path.as_str(), &pinhole)?;
diff --git a/src/capability/depth.rs b/src/capability/depth.rs
index 1b13041..61e5dde 100644
--- a/src/capability/depth.rs
+++ b/src/capability/depth.rs
@@ -1,7 +1,7 @@
 use anyhow::{Result, anyhow};
 use image::{ExtendedColorType, ImageEncoder, codecs::png::PngEncoder};
-use phoxal::api::component::v1::capability::depth::{Depth, MILLIMETERS_PER_METER};
-use phoxal::bus::pubsub::Stamped;
+use phoxal::api::v1::component::capability::depth::{Depth, MILLIMETERS_PER_METER};
+use phoxal::bus::typed::Received;
 use phoxal::model::component::v1::capability::Capability;
 use rerun::{EncodedDepthImage, RecordingStream, components::MediaType};
 
@@ -13,7 +13,7 @@ pub(crate) fn log(
     rec: &RecordingStream,
     scene: &SceneModel,
     capability_id: &str,
-    stamped: &Stamped<Depth>,
+    received: &Received<Depth>,
 ) -> Result<()> {
     let capability = scene.capability(capability_id)?;
     let Capability::Depth(depth) = capability else {
@@ -21,9 +21,9 @@ pub(crate) fn log(
             "capability '{capability_id}' is not a depth capability"
         ));
     };
-    let resolution = payload_resolution(&stamped.data, [depth.width_px, depth.height_px])?;
+    let resolution = payload_resolution(&received.value, [depth.width_px, depth.height_px])?;
 
-    set_logical_time(rec, stamped.timestamp_ns);
+    set_logical_time(rec, received.at_ns);
     let image_entity_path = scene.image_entity_path(capability_id)?;
     if let Some(pinhole) = pinhole_from_horizontal_fov(resolution, depth.field_of_view_rad) {
         rec.log(image_entity_path.as_str(), &pinhole)?;
@@ -32,7 +32,7 @@ pub(crate) fn log(
     let depth_entity_path = scene.depth_entity_path(capability_id)?;
     rec.log(
         depth_entity_path.as_str(),
-        &EncodedDepthImage::new(samples_to_png(stamped.data.samples_mm(), resolution)?)
+        &EncodedDepthImage::new(samples_to_png(received.value.samples_mm(), resolution)?)
             .with_media_type(MediaType::png())
             .with_meter(MILLIMETERS_PER_METER)
             .with_point_fill_ratio(0.35),
@@ -80,7 +80,7 @@ fn samples_to_native_bytes(samples: &[u16]) -> Vec<u8> {
 #[cfg(test)]
 mod tests {
     use image::ImageReader;
-    use phoxal::api::component::v1::capability::depth::Depth;
+    use phoxal::api::v1::component::capability::depth::Depth;
 
     use super::{payload_resolution, samples_to_png};
 
diff --git a/src/capability/encoder.rs b/src/capability/encoder.rs
index e7fd090..f1232aa 100644
--- a/src/capability/encoder.rs
+++ b/src/capability/encoder.rs
@@ -1,6 +1,6 @@
 use anyhow::Result;
-use phoxal::api::component::v1::capability::encoder::Sample;
-use phoxal::bus::pubsub::Stamped;
+use phoxal::api::v1::component::capability::encoder::Sample;
+use phoxal::bus::typed::Received;
 use rerun::{RecordingStream, Scalars};
 
 use crate::logging::helpers::set_logical_time;
@@ -8,12 +8,12 @@ use crate::logging::helpers::set_logical_time;
 pub(crate) fn log(
     rec: &RecordingStream,
     capability_entity_path: &str,
-    stamped: &Stamped<Sample>,
+    received: &Received<Sample>,
 ) -> Result<()> {
-    set_logical_time(rec, stamped.timestamp_ns);
+    set_logical_time(rec, received.at_ns);
     rec.log(
         format!("{capability_entity_path}/ticks").as_str(),
-        &Scalars::new([stamped.data.ticks() as f64]),
+        &Scalars::new([received.value.ticks() as f64]),
     )?;
     Ok(())
 }
diff --git a/src/capability/gnss.rs b/src/capability/gnss.rs
index 17c63de..fe51c43 100644
--- a/src/capability/gnss.rs
+++ b/src/capability/gnss.rs
@@ -1,6 +1,6 @@
 use anyhow::Result;
-use phoxal::api::component::v1::capability::gnss::Sample;
-use phoxal::bus::pubsub::Stamped;
+use phoxal::api::v1::component::capability::gnss::Sample;
+use phoxal::bus::typed::Received;
 use rerun::{RecordingStream, Scalars};
 
 use crate::logging::helpers::set_logical_time;
@@ -8,12 +8,12 @@ use crate::logging::helpers::set_logical_time;
 pub(crate) fn log(
     rec: &RecordingStream,
     capability_entity_path: &str,
-    stamped: &Stamped<Sample>,
+    received: &Received<Sample>,
 ) -> Result<()> {
-    set_logical_time(rec, stamped.timestamp_ns);
+    set_logical_time(rec, received.at_ns);
     rec.log(
         format!("{capability_entity_path}/gnss_altitude").as_str(),
-        &Scalars::new([stamped.data.altitude()]),
+        &Scalars::new([received.value.altitude()]),
     )?;
     Ok(())
 }
diff --git a/src/capability/gyroscope.rs b/src/capability/gyroscope.rs
index 1d976c5..86af783 100644
--- a/src/capability/gyroscope.rs
+++ b/src/capability/gyroscope.rs
@@ -1,6 +1,6 @@
 use anyhow::Result;
-use phoxal::api::component::v1::capability::gyroscope::Sample;
-use phoxal::bus::pubsub::Stamped;
+use phoxal::api::v1::component::capability::gyroscope::Sample;
+use phoxal::bus::typed::Received;
 use rerun::{Arrows3D, RecordingStream};
 
 use crate::logging::helpers::set_logical_time;
@@ -8,12 +8,13 @@ use crate::logging::helpers::set_logical_time;
 pub(crate) fn log(
     rec: &RecordingStream,
     capability_entity_path: &str,
-    stamped: &Stamped<Sample>,
+    received: &Received<Sample>,
 ) -> Result<()> {
-    set_logical_time(rec, stamped.timestamp_ns);
+    set_logical_time(rec, received.at_ns);
     rec.log(
         format!("{capability_entity_path}/angular_velocity").as_str(),
-        &Arrows3D::from_vectors([*stamped.data.angular_velocity()]).with_origins([[0.0, 0.0, 0.0]]),
+        &Arrows3D::from_vectors([*received.value.angular_velocity()])
+            .with_origins([[0.0, 0.0, 0.0]]),
     )?;
     Ok(())
 }
diff --git a/src/capability/imu.rs b/src/capability/imu.rs
index b99d95e..75f0318 100644
--- a/src/capability/imu.rs
+++ b/src/capability/imu.rs
@@ -1,6 +1,6 @@
 use anyhow::Result;
-use phoxal::api::component::v1::capability::imu::Sample;
-use phoxal::bus::pubsub::Stamped;
+use phoxal::api::v1::component::capability::imu::Sample;
+use phoxal::bus::typed::Received;
 use rerun::{Quaternion, RecordingStream, Transform3D};
 
 use crate::logging::helpers::set_logical_time;
@@ -9,10 +9,10 @@ use crate::visuals::reorder_wxyz_to_xyzw;
 pub(crate) fn log(
     rec: &RecordingStream,
     capability_entity_path: &str,
-    stamped: &Stamped<Sample>,
+    received: &Received<Sample>,
 ) -> Result<()> {
-    set_logical_time(rec, stamped.timestamp_ns);
-    if let Some(orientation) = stamped.data.orientation() {
+    set_logical_time(rec, received.at_ns);
+    if let Some(orientation) = received.value.orientation() {
         rec.log(
             format!("{capability_entity_path}/orientation").as_str(),
             &Transform3D::from_translation_rotation(
diff --git a/src/capability/lidar.rs b/src/capability/lidar.rs
index 09c53c4..d7c283a 100644
--- a/src/capability/lidar.rs
+++ b/src/capability/lidar.rs
@@ -1,6 +1,6 @@
 use anyhow::{Result, anyhow};
-use phoxal::api::component::v1::capability::lidar::Scan;
-use phoxal::bus::pubsub::Stamped;
+use phoxal::api::v1::component::capability::lidar::Scan;
+use phoxal::bus::typed::Received;
 use phoxal::model::component::v1::CapabilityRef;
 use phoxal::model::component::v1::capability::{Capability, LidarOutput};
 use rerun::{Clear, Points3D, RecordingStream};
@@ -13,11 +13,11 @@ pub(crate) fn log(
     scene: &SceneModel,
     capability_entity_path: &str,
     capability_id: &str,
-    stamped: &Stamped<Scan>,
+    received: &Received<Scan>,
 ) -> Result<()> {
-    set_logical_time(rec, stamped.timestamp_ns);
+    set_logical_time(rec, received.at_ns);
 
-    match &stamped.data {
+    match &received.value {
         Scan::Ranges(scan) => {
             let config = lidar_config(scene, capability_id)?;
             if config.output != LidarOutput::Ranges {
diff --git a/src/capability/magnetometer.rs b/src/capability/magnetometer.rs
index 9cf90ae..b5a4804 100644
--- a/src/capability/magnetometer.rs
+++ b/src/capability/magnetometer.rs
@@ -1,6 +1,6 @@
 use anyhow::Result;
-use phoxal::api::component::v1::capability::magnetometer::Sample;
-use phoxal::bus::pubsub::Stamped;
+use phoxal::api::v1::component::capability::magnetometer::Sample;
+use phoxal::bus::typed::Received;
 use rerun::{Arrows3D, RecordingStream};
 
 use crate::logging::helpers::set_logical_time;
@@ -8,12 +8,12 @@ use crate::logging::helpers::set_logical_time;
 pub(crate) fn log(
     rec: &RecordingStream,
     capability_entity_path: &str,
-    stamped: &Stamped<Sample>,
+    received: &Received<Sample>,
 ) -> Result<()> {
-    set_logical_time(rec, stamped.timestamp_ns);
+    set_logical_time(rec, received.at_ns);
     rec.log(
         format!("{capability_entity_path}/magnetic_field").as_str(),
-        &Arrows3D::from_vectors([*stamped.data.magnetic_field()]).with_origins([[0.0, 0.0, 0.0]]),
+        &Arrows3D::from_vectors([*received.value.magnetic_field()]).with_origins([[0.0, 0.0, 0.0]]),
     )?;
     Ok(())
 }
diff --git a/src/capability/microphone.rs b/src/capability/microphone.rs
deleted file mode 100644
index 619c6d5..0000000
--- a/src/capability/microphone.rs
+++ /dev/null
@@ -1,19 +0,0 @@
-use anyhow::Result;
-use phoxal::api::component::v1::capability::microphone::Frame;
-use phoxal::bus::pubsub::Stamped;
-use rerun::{RecordingStream, Scalars};
-
-use crate::logging::helpers::set_logical_time;
-
-pub(crate) fn log(
-    rec: &RecordingStream,
-    capability_entity_path: &str,
-    stamped: &Stamped<Frame>,
-) -> Result<()> {
-    set_logical_time(rec, stamped.timestamp_ns);
-    rec.log(
-        format!("{capability_entity_path}/audio_bytes").as_str(),
-        &Scalars::new([stamped.data.data().len() as f64]),
-    )?;
-    Ok(())
-}
diff --git a/src/capability/mmwave.rs b/src/capability/mmwave.rs
index 477b7b9..862852a 100644
--- a/src/capability/mmwave.rs
+++ b/src/capability/mmwave.rs
@@ -1,6 +1,6 @@
 use anyhow::Result;
-use phoxal::api::component::v1::capability::mmwave::Scan;
-use phoxal::bus::pubsub::Stamped;
+use phoxal::api::v1::component::capability::mmwave::Scan;
+use phoxal::bus::typed::Received;
 use rerun::{Arrows3D, Clear, Points3D, RecordingStream, Scalars};
 
 use crate::logging::helpers::set_logical_time;
@@ -8,23 +8,23 @@ use crate::logging::helpers::set_logical_time;
 pub(crate) fn log(
     rec: &RecordingStream,
     capability_entity_path: &str,
-    stamped: &Stamped<Scan>,
+    received: &Received<Scan>,
 ) -> Result<()> {
-    set_logical_time(rec, stamped.timestamp_ns);
-    let positions = stamped
-        .data
+    set_logical_time(rec, received.at_ns);
+    let positions = received
+        .value
         .detections()
         .iter()
         .map(|detection| detection.position)
         .collect::<Vec<_>>();
-    let velocities = stamped
-        .data
+    let velocities = received
+        .value
         .detections()
         .iter()
         .map(|detection| detection.velocity)
         .collect::<Vec<_>>();
-    let snr = stamped
-        .data
+    let snr = received
+        .value
         .detections()
         .iter()
         .map(|detection| f64::from(detection.snr))
diff --git a/src/capability/mod.rs b/src/capability/mod.rs
index 731878c..adf0c1a 100644
--- a/src/capability/mod.rs
+++ b/src/capability/mod.rs
@@ -8,123 +8,119 @@ mod gyroscope;
 mod imu;
 mod lidar;
 mod magnetometer;
-mod microphone;
 mod mmwave;
 mod range;
-mod speaker;
 
 use crate::scene::SceneModel;
 use anyhow::Result;
-use phoxal::api::component::v1::capability::accelerometer::Sample as AccelerometerSample;
-use phoxal::api::component::v1::capability::battery::State as BatteryState;
-use phoxal::api::component::v1::capability::camera::Frame as CameraFrame;
-use phoxal::api::component::v1::capability::depth::Depth;
-use phoxal::api::component::v1::capability::encoder::Sample as EncoderSample;
-use phoxal::api::component::v1::capability::gnss::Sample as GnssSample;
-use phoxal::api::component::v1::capability::gyroscope::Sample as GyroscopeSample;
-use phoxal::api::component::v1::capability::imu::Sample as ImuSample;
-use phoxal::api::component::v1::capability::lidar::Scan as LidarScan;
-use phoxal::api::component::v1::capability::magnetometer::Sample as MagnetometerSample;
-use phoxal::api::component::v1::capability::microphone::Frame as MicrophoneFrame;
-use phoxal::api::component::v1::capability::mmwave::Scan as MmwaveScan;
-use phoxal::api::component::v1::capability::range::Sample as RangeSample;
-use phoxal::api::component::v1::capability::speaker::audio::Audio;
-use phoxal::bus::pubsub::Stamped;
+use phoxal::api::v1::component::capability::accelerometer::Sample as AccelerometerSample;
+use phoxal::api::v1::component::capability::battery::State as BatteryState;
+use phoxal::api::v1::component::capability::camera::Frame as CameraFrame;
+use phoxal::api::v1::component::capability::depth::Depth;
+use phoxal::api::v1::component::capability::encoder::Sample as EncoderSample;
+use phoxal::api::v1::component::capability::gnss::Sample as GnssSample;
+use phoxal::api::v1::component::capability::gyroscope::Sample as GyroscopeSample;
+use phoxal::api::v1::component::capability::imu::Sample as ImuSample;
+use phoxal::api::v1::component::capability::lidar::Scan as LidarScan;
+use phoxal::api::v1::component::capability::magnetometer::Sample as MagnetometerSample;
+use phoxal::api::v1::component::capability::mmwave::Scan as MmwaveScan;
+use phoxal::api::v1::component::capability::range::Sample as RangeSample;
+use phoxal::bus::typed::Received;
 use rerun::RecordingStream;
 
 pub(crate) fn log_imu(
     rec: &RecordingStream,
     scene: &SceneModel,
     capability_id: &str,
-    stamped: &Stamped<ImuSample>,
+    received: &Received<ImuSample>,
 ) -> Result<()> {
-    imu::log(rec, scene.capability_entity_path(capability_id)?, stamped)
+    imu::log(rec, scene.capability_entity_path(capability_id)?, received)
 }
 
 pub(crate) fn log_accelerometer(
     rec: &RecordingStream,
     scene: &SceneModel,
     capability_id: &str,
-    stamped: &Stamped<AccelerometerSample>,
+    received: &Received<AccelerometerSample>,
 ) -> Result<()> {
-    accelerometer::log(rec, scene.capability_entity_path(capability_id)?, stamped)
+    accelerometer::log(rec, scene.capability_entity_path(capability_id)?, received)
 }
 
 pub(crate) fn log_gyroscope(
     rec: &RecordingStream,
     scene: &SceneModel,
     capability_id: &str,
-    stamped: &Stamped<GyroscopeSample>,
+    received: &Received<GyroscopeSample>,
 ) -> Result<()> {
-    gyroscope::log(rec, scene.capability_entity_path(capability_id)?, stamped)
+    gyroscope::log(rec, scene.capability_entity_path(capability_id)?, received)
 }
 
 pub(crate) fn log_gnss(
     rec: &RecordingStream,
     scene: &SceneModel,
     capability_id: &str,
-    stamped: &Stamped<GnssSample>,
+    received: &Received<GnssSample>,
 ) -> Result<()> {
-    gnss::log(rec, scene.capability_entity_path(capability_id)?, stamped)
+    gnss::log(rec, scene.capability_entity_path(capability_id)?, received)
 }
 
 pub(crate) fn log_battery(
     rec: &RecordingStream,
     scene: &SceneModel,
     capability_id: &str,
-    stamped: &Stamped<BatteryState>,
+    received: &Received<BatteryState>,
 ) -> Result<()> {
-    battery::log(rec, scene.capability_entity_path(capability_id)?, stamped)
+    battery::log(rec, scene.capability_entity_path(capability_id)?, received)
 }
 
 pub(crate) fn log_encoder(
     rec: &RecordingStream,
     scene: &SceneModel,
     capability_id: &str,
-    stamped: &Stamped<EncoderSample>,
+    received: &Received<EncoderSample>,
 ) -> Result<()> {
-    encoder::log(rec, scene.capability_entity_path(capability_id)?, stamped)
+    encoder::log(rec, scene.capability_entity_path(capability_id)?, received)
 }
 
 pub(crate) fn log_camera(
     rec: &RecordingStream,
     scene: &SceneModel,
     capability_id: &str,
-    stamped: &Stamped<CameraFrame>,
+    received: &Received<CameraFrame>,
 ) -> Result<()> {
-    camera::log(rec, scene, capability_id, stamped)
+    camera::log(rec, scene, capability_id, received)
 }
 
 pub(crate) fn log_depth(
     rec: &RecordingStream,
     scene: &SceneModel,
     capability_id: &str,
-    stamped: &Stamped<Depth>,
+    received: &Received<Depth>,
 ) -> Result<()> {
-    depth::log(rec, scene, capability_id, stamped)
+    depth::log(rec, scene, capability_id, received)
 }
 
 pub(crate) fn log_range(
     rec: &RecordingStream,
     scene: &SceneModel,
     capability_id: &str,
-    stamped: &Stamped<RangeSample>,
+    received: &Received<RangeSample>,
 ) -> Result<()> {
-    range::log(rec, scene.capability_entity_path(capability_id)?, stamped)
+    range::log(rec, scene.capability_entity_path(capability_id)?, received)
 }
 
 pub(crate) fn log_lidar(
     rec: &RecordingStream,
     scene: &SceneModel,
     capability_id: &str,
-    stamped: &Stamped<LidarScan>,
+    received: &Received<LidarScan>,
 ) -> Result<()> {
     lidar::log(
         rec,
         scene,
         scene.capability_entity_path(capability_id)?,
         capability_id,
-        stamped,
+        received,
     )
 }
 
@@ -132,34 +128,16 @@ pub(crate) fn log_magnetometer(
     rec: &RecordingStream,
     scene: &SceneModel,
     capability_id: &str,
-    stamped: &Stamped<MagnetometerSample>,
+    received: &Received<MagnetometerSample>,
 ) -> Result<()> {
-    magnetometer::log(rec, scene.capability_entity_path(capability_id)?, stamped)
-}
-
-pub(crate) fn log_microphone(
-    rec: &RecordingStream,
-    scene: &SceneModel,
-    capability_id: &str,
-    stamped: &Stamped<MicrophoneFrame>,
-) -> Result<()> {
-    microphone::log(rec, scene.capability_entity_path(capability_id)?, stamped)
+    magnetometer::log(rec, scene.capability_entity_path(capability_id)?, received)
 }
 
 pub(crate) fn log_mmwave(
     rec: &RecordingStream,
     scene: &SceneModel,
     capability_id: &str,
-    stamped: &Stamped<MmwaveScan>,
-) -> Result<()> {
-    mmwave::log(rec, scene.capability_entity_path(capability_id)?, stamped)
-}
-
-pub(crate) fn log_speaker_data(
-    rec: &RecordingStream,
-    scene: &SceneModel,
-    capability_id: &str,
-    stamped: &Stamped<Audio>,
+    received: &Received<MmwaveScan>,
 ) -> Result<()> {
-    speaker::log_data(rec, scene.capability_entity_path(capability_id)?, stamped)
+    mmwave::log(rec, scene.capability_entity_path(capability_id)?, received)
 }
diff --git a/src/capability/range.rs b/src/capability/range.rs
index c7c6c2a..016ecd5 100644
--- a/src/capability/range.rs
+++ b/src/capability/range.rs
@@ -1,6 +1,6 @@
 use anyhow::Result;
-use phoxal::api::component::v1::capability::range::Sample;
-use phoxal::bus::pubsub::Stamped;
+use phoxal::api::v1::component::capability::range::Sample;
+use phoxal::bus::typed::Received;
 use rerun::{Color, LineStrips3D, Points3D, RecordingStream, TextLog, TextLogLevel};
 
 use crate::logging::helpers::set_logical_time;
@@ -8,10 +8,10 @@ use crate::logging::helpers::set_logical_time;
 pub(crate) fn log(
     rec: &RecordingStream,
     capability_entity_path: &str,
-    stamped: &Stamped<Sample>,
+    received: &Received<Sample>,
 ) -> Result<()> {
-    set_logical_time(rec, stamped.timestamp_ns);
-    let distance_m = stamped.data.distance_m();
+    set_logical_time(rec, received.at_ns);
+    let distance_m = received.value.distance_m();
     let rounded_distance_m = (distance_m * 100.0).round() / 100.0;
 
     rec.log(
diff --git a/src/capability/speaker.rs b/src/capability/speaker.rs
deleted file mode 100644
index b5466d5..0000000
--- a/src/capability/speaker.rs
+++ /dev/null
@@ -1,19 +0,0 @@
-use anyhow::Result;
-use phoxal::api::component::v1::capability::speaker::audio::Audio;
-use phoxal::bus::pubsub::Stamped;
-use rerun::{RecordingStream, Scalars};
-
-use crate::logging::helpers::set_logical_time;
-
-pub(crate) fn log_data(
-    rec: &RecordingStream,
-    capability_entity_path: &str,
-    stamped: &Stamped<Audio>,
-) -> Result<()> {
-    set_logical_time(rec, stamped.timestamp_ns);
-    rec.log(
-        format!("{capability_entity_path}/audio_output_bytes").as_str(),
-        &Scalars::new([stamped.data.data().len() as f64]),
-    )?;
-    Ok(())
-}
diff --git a/src/event.rs b/src/event.rs
index 1927701..7465e6b 100644
--- a/src/event.rs
+++ b/src/event.rs
@@ -1,23 +1,21 @@
-use phoxal::api::component::v1::capability::accelerometer::Sample as AccelerometerSample;
-use phoxal::api::component::v1::capability::battery::State as BatteryState;
-use phoxal::api::component::v1::capability::camera::Frame as CameraFrame;
-use phoxal::api::component::v1::capability::depth::Depth;
-use phoxal::api::component::v1::capability::encoder::Sample as EncoderSample;
-use phoxal::api::component::v1::capability::gnss::Sample as GnssSample;
-use phoxal::api::component::v1::capability::gyroscope::Sample as GyroscopeSample;
-use phoxal::api::component::v1::capability::imu::Sample as ImuSample;
-use phoxal::api::component::v1::capability::lidar::Scan as LidarScan;
-use phoxal::api::component::v1::capability::magnetometer::Sample as MagnetometerSample;
-use phoxal::api::component::v1::capability::microphone::Frame as MicrophoneFrame;
-use phoxal::api::component::v1::capability::mmwave::Scan as MmwaveScan;
-use phoxal::api::component::v1::capability::range::Sample as RangeSample;
-use phoxal::api::component::v1::capability::speaker::audio::Audio;
-use phoxal::api::drive::v1::State as DriveState;
-use phoxal::api::frame::v1::FrameTransform;
-use phoxal::api::joint::v1::JointState;
-use phoxal::api::odometry::v1::OdometryEstimate;
-use phoxal::api::presence::Heartbeat;
-use phoxal::bus::pubsub::Stamped;
+use phoxal::api::v1::component::capability::accelerometer::Sample as AccelerometerSample;
+use phoxal::api::v1::component::capability::battery::State as BatteryState;
+use phoxal::api::v1::component::capability::camera::Frame as CameraFrame;
+use phoxal::api::v1::component::capability::depth::Depth;
+use phoxal::api::v1::component::capability::encoder::Sample as EncoderSample;
+use phoxal::api::v1::component::capability::gnss::Sample as GnssSample;
+use phoxal::api::v1::component::capability::gyroscope::Sample as GyroscopeSample;
+use phoxal::api::v1::component::capability::imu::Sample as ImuSample;
+use phoxal::api::v1::component::capability::lidar::Scan as LidarScan;
+use phoxal::api::v1::component::capability::magnetometer::Sample as MagnetometerSample;
+use phoxal::api::v1::component::capability::mmwave::Scan as MmwaveScan;
+use phoxal::api::v1::component::capability::range::Sample as RangeSample;
+use phoxal::api::v1::drive::State as DriveState;
+use phoxal::api::v1::frame::FrameTransform;
+use phoxal::api::v1::joint::JointState;
+use phoxal::api::v1::odometry::OdometryEstimate;
+use phoxal::api::v1::presence::Heartbeat;
+use phoxal::bus::typed::Received;
 
 macro_rules! define_event_enum {
     (
@@ -34,15 +32,15 @@ macro_rules! define_event_enum {
             )*
             Joint {
                 joint_id: String,
-                message: Stamped<JointState>,
+                message: Received<JointState>,
             },
             Camera {
                 capability_id: String,
-                message: Stamped<CameraFrame>,
+                message: Received<CameraFrame>,
             },
             Depth {
                 capability_id: String,
-                message: Stamped<Depth>,
+                message: Received<Depth>,
             },
             $(
                 $cap_variant {
@@ -56,22 +54,20 @@ macro_rules! define_event_enum {
 
 define_event_enum! {
     pub(crate) enum Event {
-        Frame(Stamped<FrameTransform>),
-        Odometry(Stamped<OdometryEstimate>),
-        DriveState(Stamped<DriveState>),
-        PresenceHeartbeat(Stamped<Heartbeat>),
+        Frame(Received<FrameTransform>),
+        Odometry(Received<OdometryEstimate>),
+        DriveState(Received<DriveState>),
+        PresenceHeartbeat(Received<Heartbeat>),
         ---
-        Imu(Stamped<ImuSample>),
-        Accelerometer(Stamped<AccelerometerSample>),
-        Gyroscope(Stamped<GyroscopeSample>),
-        Gnss(Stamped<GnssSample>),
-        Battery(Stamped<BatteryState>),
-        Encoder(Stamped<EncoderSample>),
-        Range(Stamped<RangeSample>),
-        Lidar(Stamped<LidarScan>),
-        Magnetometer(Stamped<MagnetometerSample>),
-        Microphone(Stamped<MicrophoneFrame>),
-        Mmwave(Stamped<MmwaveScan>),
-        SpeakerData(Stamped<Audio>),
+        Imu(Received<ImuSample>),
+        Accelerometer(Received<AccelerometerSample>),
+        Gyroscope(Received<GyroscopeSample>),
+        Gnss(Received<GnssSample>),
+        Battery(Received<BatteryState>),
+        Encoder(Received<EncoderSample>),
+        Range(Received<RangeSample>),
+        Lidar(Received<LidarScan>),
+        Magnetometer(Received<MagnetometerSample>),
+        Mmwave(Received<MmwaveScan>),
     }
 }
diff --git a/src/frame_graph.rs b/src/frame_graph.rs
index 21b004b..8cf8d46 100644
--- a/src/frame_graph.rs
+++ b/src/frame_graph.rs
@@ -2,8 +2,8 @@ use std::collections::{HashMap, HashSet};
 
 use anyhow::{Result, anyhow};
 use nalgebra::{Isometry3, Quaternion as NaQuaternion, Translation3, UnitQuaternion};
-use phoxal::api::frame::v1::FrameTransform;
-use phoxal::api::odometry::v1::OdometryEstimate;
+use phoxal::api::v1::frame::FrameTransform;
+use phoxal::api::v1::odometry::OdometryEstimate;
 use rerun::{Quaternion, Transform3D};
 
 #[derive(Debug, Clone)]
@@ -124,8 +124,8 @@ mod tests {
     use std::collections::HashMap;
 
     use anyhow::Result;
-    use phoxal::api::frame::v1::{FrameId, FrameTransform, Source};
-    use phoxal::api::odometry::v1::{
+    use phoxal::api::v1::frame::{FrameId, FrameTransform, Source};
+    use phoxal::api::v1::odometry::{
         Covariance, OdometryEstimate, PoseEstimate, Status, StatusMode, VelocityEstimate,
     };
 
diff --git a/src/logging/drive.rs b/src/logging/drive.rs
index 429112f..521dd12 100644
--- a/src/logging/drive.rs
+++ b/src/logging/drive.rs
@@ -1,6 +1,6 @@
 use anyhow::Result;
-use phoxal::api::drive::v1::State;
-use phoxal::bus::pubsub::Stamped;
+use phoxal::api::v1::drive::State;
+use phoxal::bus::typed::Received;
 use rerun::{RecordingStream, Scalars};
 
 use crate::logging::helpers::set_logical_time;
@@ -8,19 +8,19 @@ use crate::paths::RUNTIME_ROOT;
 
 pub(crate) fn log_drive_state_update(
     rec: &RecordingStream,
-    stamped: &Stamped<State>,
+    received: &Received<State>,
 ) -> Result<()> {
-    set_logical_time(rec, stamped.timestamp_ns);
+    set_logical_time(rec, received.at_ns);
     let path_root = format!("{RUNTIME_ROOT}/drive/state");
-    log_target(rec, &format!("{path_root}/target"), stamped.data.target)?;
+    log_target(rec, &format!("{path_root}/target"), received.value.target)?;
     log_target(
         rec,
         &format!("{path_root}/limited_target"),
-        stamped.data.limited_target,
+        received.value.limited_target,
     )?;
     rec.log(
         format!("{path_root}/stopped"),
-        &Scalars::new([f64::from(u8::from(stamped.data.stop_reason.is_some()))]),
+        &Scalars::new([f64::from(u8::from(received.value.stop_reason.is_some()))]),
     )?;
     Ok(())
 }
@@ -28,7 +28,7 @@ pub(crate) fn log_drive_state_update(
 fn log_target(
     rec: &RecordingStream,
     path_root: &str,
-    target: phoxal::api::drive::v1::Target,
+    target: phoxal::api::v1::drive::Target,
 ) -> Result<()> {
     rec.log(
         format!("{path_root}/linear_x_mps"),
diff --git a/src/logging/helpers.rs b/src/logging/helpers.rs
index 87b0a48..99c48a2 100644
--- a/src/logging/helpers.rs
+++ b/src/logging/helpers.rs
@@ -1,5 +1,7 @@
 use rerun::{RecordingStream, TimeCell};
 
-pub(crate) fn set_logical_time(rec: &RecordingStream, timestamp_ns: u64) {
-    rec.set_time("time", TimeCell::from_duration_nanos(timestamp_ns as i64));
+pub(crate) fn set_logical_time(rec: &RecordingStream, timestamp_ns: Option<u64>) {
+    if let Some(timestamp_ns) = timestamp_ns {
+        rec.set_time("time", TimeCell::from_duration_nanos(timestamp_ns as i64));
+    }
 }
diff --git a/src/logging/joint.rs b/src/logging/joint.rs
index 4cfe0e8..53c6102 100644
--- a/src/logging/joint.rs
+++ b/src/logging/joint.rs
@@ -1,6 +1,6 @@
 use anyhow::Result;
-use phoxal::api::joint::v1::JointState as JointData;
-use phoxal::bus::pubsub::Stamped;
+use phoxal::api::v1::joint::JointState as JointData;
+use phoxal::bus::typed::Received;
 use rerun::{RecordingStream, Scalars};
 
 use crate::logging::helpers::set_logical_time;
@@ -8,9 +8,9 @@ use crate::logging::helpers::set_logical_time;
 pub(crate) fn log_joint_update(
     rec: &RecordingStream,
     joint_entity_path: &str,
-    stamped: &Stamped<JointData>,
+    received: &Received<JointData>,
 ) -> Result<()> {
-    set_logical_time(rec, stamped.timestamp_ns);
-    rec.log(joint_entity_path, &Scalars::new([stamped.data.value]))?;
+    set_logical_time(rec, received.at_ns);
+    rec.log(joint_entity_path, &Scalars::new([received.value.value]))?;
     Ok(())
 }
diff --git a/src/logging/odometry.rs b/src/logging/odometry.rs
index f08870c..a9edac6 100644
--- a/src/logging/odometry.rs
+++ b/src/logging/odometry.rs
@@ -1,6 +1,6 @@
 use anyhow::Result;
-use phoxal::api::odometry::v1::OdometryEstimate as OdometryData;
-use phoxal::bus::pubsub::Stamped;
+use phoxal::api::v1::odometry::OdometryEstimate as OdometryData;
+use phoxal::bus::typed::Received;
 use rerun::{RecordingStream, Scalars};
 
 use crate::logging::helpers::set_logical_time;
@@ -8,16 +8,16 @@ use crate::paths::RUNTIME_ODOMETRY_ENTITY_PATH;
 
 pub(crate) fn log_odometry_update(
     rec: &RecordingStream,
-    stamped: &Stamped<OdometryData>,
+    received: &Received<OdometryData>,
 ) -> Result<()> {
-    set_logical_time(rec, stamped.timestamp_ns);
+    set_logical_time(rec, received.at_ns);
     rec.log(
         format!("{RUNTIME_ODOMETRY_ENTITY_PATH}/pose/position"),
-        &Scalars::new(stamped.data.pose.translation_m),
+        &Scalars::new(received.value.pose.translation_m),
     )?;
     rec.log(
         format!("{RUNTIME_ODOMETRY_ENTITY_PATH}/pose/orientation_xyzw"),
-        &Scalars::new(stamped.data.pose.rotation_xyzw),
+        &Scalars::new(received.value.pose.rotation_xyzw),
     )?;
     Ok(())
 }
diff --git a/src/logging/static_scene.rs b/src/logging/static_scene.rs
index 76a343f..b94e13c 100644
--- a/src/logging/static_scene.rs
+++ b/src/logging/static_scene.rs
@@ -32,7 +32,7 @@ pub(crate) fn log_static_scene(
         }
     }
 
-    set_logical_time(rec, 0);
+    set_logical_time(rec, Some(0));
     scene.log_all_render_anchors(rec, graph, component_selection)?;
     Ok(())
 }
diff --git a/src/main.rs b/src/main.rs
index 9cf3697..86b93f7 100644
--- a/src/main.rs
+++ b/src/main.rs
@@ -91,7 +91,7 @@ fn handle_event(
 ) -> Result<()> {
     match event {
         Event::Frame(message) => {
-            frame_graph.update_frame(message.data.clone());
+            frame_graph.update_frame(message.value.clone());
             scene.log_frame_anchors(rec, frame_graph, component_selection)
         }
         Event::Joint { joint_id, message } => log_joint_update(
@@ -100,7 +100,7 @@ fn handle_event(
             &message,
         ),
         Event::Odometry(message) => {
-            frame_graph.update_odometry(&message.data);
+            frame_graph.update_odometry(&message.value);
             log_odometry_update(rec, &message)?;
             scene.log_odometry_anchors(rec, frame_graph, component_selection)
         }
@@ -150,17 +150,9 @@ fn handle_event(
             capability_id,
             message,
         } => capability::log_magnetometer(rec, scene, &capability_id, &message),
-        Event::Microphone {
-            capability_id,
-            message,
-        } => capability::log_microphone(rec, scene, &capability_id, &message),
         Event::Mmwave {
             capability_id,
             message,
         } => capability::log_mmwave(rec, scene, &capability_id, &message),
-        Event::SpeakerData {
-            capability_id,
-            message,
-        } => capability::log_speaker_data(rec, scene, &capability_id, &message),
     }
 }
diff --git a/src/scene.rs b/src/scene.rs
index 8e68942..9e11ea9 100644
--- a/src/scene.rs
+++ b/src/scene.rs
@@ -1,8 +1,8 @@
 use std::collections::HashMap;
 
 use anyhow::{Context, Result, anyhow};
-use phoxal::api::frame::v1::{FrameId, FrameTransform as FrameData, Source};
-use phoxal::api::joint::v1::JointId;
+use phoxal::api::v1::frame::{FrameId, FrameTransform as FrameData, Source};
+use phoxal::api::v1::joint::JointId;
 use phoxal::bus::Bus;
 use phoxal::bus::query::Retry;
 use phoxal::model::component::v1::capability::{CameraMode, Capability, StructuralTarget};
diff --git a/src/subscriptions.rs b/src/subscriptions.rs
index dfdcc6d..64f1555 100644
--- a/src/subscriptions.rs
+++ b/src/subscriptions.rs
@@ -1,42 +1,18 @@
-use std::future::Future;
 use std::time::Duration;
 
 use crate::event::Event;
 use crate::scene::SceneModel;
 use crate::selection::ComponentSelection;
-use phoxal::api::component::v1::capability::camera::Frame as CameraFrame;
-use phoxal::api::component::v1::capability::depth::Depth as DepthFrame;
-use phoxal::api::frame::v1::FrameId;
-use phoxal::api::joint::v1::JointId;
+use phoxal::api::v1::topic;
 use phoxal::bus::Bus;
-use phoxal::bus::pubsub::Stamped;
-use phoxal::bus::zenoh::{TypedSchema, TypedSubscriber};
-use phoxal::model::component::v1::CapabilityRef;
+use phoxal::bus::topic::{ANY, PubSub, Topic};
+use phoxal::bus::typed::Received;
 use phoxal::model::component::v1::capability::Capability;
-use phoxal::model::robot::v1::{ResolvedLocalizeBackend, resolve_localize_backend};
-use phoxal::model::v1::Robot;
-use phoxal::runtime::debug_input_topic;
 use serde::de::DeserializeOwned;
 use tokio::sync::mpsc;
 use tracing::warn;
 
 const SUBSCRIPTION_RETRY_DELAY: Duration = Duration::from_secs(1);
-const LOCALIZE_RUNTIME_ID: &str = "localize";
-
-#[derive(Debug, Clone, Copy, PartialEq, Eq)]
-enum LocalizeDebugInputPayload {
-    Camera,
-    Depth,
-    Imu,
-}
-
-#[derive(Debug, Clone, PartialEq, Eq)]
-struct LocalizeDebugInputSubscription {
-    key: &'static str,
-    topic_name: String,
-    capability_id: String,
-    payload: LocalizeDebugInputPayload,
-}
 
 pub(crate) fn spawn_all(
     bus: Bus,
@@ -44,84 +20,45 @@ pub(crate) fn spawn_all(
     scene: &SceneModel,
     component_selection: &ComponentSelection,
 ) {
-    for frame_id in scene.default_frames.keys() {
-        let frame_id = frame_id.clone();
-        let typed_frame_id = FrameId::new(frame_id.clone());
-        spawn_subscription_loop(
-            bus.clone(),
-            phoxal::api::frame::v1::data::path(&typed_frame_id),
-            move |bus| {
-                let typed_frame_id = typed_frame_id.clone();
-                async move {
-                    phoxal::api::frame::v1::data::subscriber_builder(&bus, &typed_frame_id)
-                        .await
-                        .map_err(phoxal::bus::Error::from)
-                }
-            },
-            tx.clone(),
-            Event::Frame,
-        );
-    }
+    spawn_subscription_loop(
+        bus.clone(),
+        topic::new().v1().frame().data(),
+        tx.clone(),
+        Event::Frame,
+    );
 
     for joint in &scene.structure.joints {
-        let joint_name = joint.name.clone();
-        let joint_id = JointId::new(joint_name.clone());
+        let joint_id = joint.name.clone();
         spawn_subscription_loop(
             bus.clone(),
-            phoxal::api::joint::v1::data::path(&joint_id),
-            move |bus| {
-                let joint_id = joint_id.clone();
-                async move {
-                    phoxal::api::joint::v1::data::subscriber_builder(&bus, &joint_id)
-                        .await
-                        .map_err(phoxal::bus::Error::from)
-                }
-            },
+            topic::new().v1().joint(joint_id.clone()).data(),
             tx.clone(),
-            {
-                let joint_id = joint_name;
-                move |message| Event::Joint {
-                    joint_id: joint_id.clone(),
-                    message,
-                }
+            move |message| Event::Joint {
+                joint_id: joint_id.clone(),
+                message,
             },
         );
     }
+
     spawn_subscription_loop(
         bus.clone(),
-        phoxal::api::odometry::v1::data::path(),
-        move |bus| async move {
-            phoxal::api::odometry::v1::data::subscriber_builder(&bus)
-                .await
-                .map_err(phoxal::bus::Error::from)
-        },
+        topic::new().v1().odometry().estimate(),
         tx.clone(),
         Event::Odometry,
     );
     spawn_subscription_loop(
         bus.clone(),
-        phoxal::api::drive::v1::state::path(),
-        move |bus| async move {
-            phoxal::api::drive::v1::state::subscriber_builder(&bus)
-                .await
-                .map_err(phoxal::bus::Error::from)
-        },
+        topic::new().v1().drive().state(),
         tx.clone(),
         Event::DriveState,
     );
     spawn_subscription_loop(
         bus.clone(),
-        phoxal::api::presence::heartbeat::path(),
-        move |bus| async move {
-            phoxal::api::presence::heartbeat::subscriber_builder(&bus)
-                .await
-                .map_err(phoxal::bus::Error::from)
-        },
+        topic::new().v1().presence().heartbeat(),
         tx.clone(),
         Event::PresenceHeartbeat,
     );
 
-    spawn_localize_debug_input_subscriptions(bus.clone(), tx.clone(), scene);
     spawn_capability_subscriptions(bus, tx, scene, component_selection);
 }
 
@@ -132,24 +69,15 @@ fn spawn_capability_subscriptions(
     component_selection: &ComponentSelection,
 ) {
     macro_rules! spawn_capability {
-        ($component_id:expr, $local_capability_id:expr, $kind:expr, $build:expr, $event_variant:ident) => {{
+        ($component_id:expr, $local_capability_id:expr, $topic:expr, $event_variant:ident) => {{
             let capability_id = format!("{}.{}", $component_id, $local_capability_id);
-            let component_id = $component_id.to_string();
-            let local_capability_id = $local_capability_id.to_string();
-            spawn_subscription_loop(
-                bus.clone(),
-                capability_subscription_path(&component_id, $kind, &local_capability_id),
-                move |bus| {
-                    let component_id = component_id.clone();
-                    let local_capability_id = local_capability_id.clone();
-                    async move { $build(bus, component_id, local_capability_id).await }
-                },
-                tx.clone(),
-                move |message| Event::$event_variant {
+            let topic = $topic($component_id.to_string(), $local_capability_id.to_string());
+            spawn_subscription_loop(bus.clone(), topic, tx.clone(), move |message| {
+                Event::$event_variant {
                     capability_id: capability_id.clone(),
                     message,
-                },
-            );
+                }
+            });
         }};
     }
 
@@ -162,365 +90,158 @@ fn spawn_capability_subscriptions(
         };
         for (local_capability_id, capability) in &component.capabilities {
             match capability {
-                Capability::Imu(_) => {}
-                Capability::Accelerometer(_) => {
-                    spawn_capability!(
-                        component_id,
-                        local_capability_id,
-                        phoxal::api::component::v1::capability::accelerometer::KIND,
-                        |bus: Bus, component_id: String, local_capability_id: String| async move {
-                            phoxal::api::component::v1::capability::accelerometer::subscriber_builder(
-                                &bus,
-                                &component_id,
-                                &local_capability_id,
-                            )
-                            .await
-                            .map_err(phoxal::bus::Error::from)
-                        },
-                        Accelerometer
-                    );
-                }
-                Capability::Gyroscope(_) => {
-                    spawn_capability!(
-                        component_id,
-                        local_capability_id,
-                        phoxal::api::component::v1::capability::gyroscope::KIND,
-                        |bus: Bus, component_id: String, local_capability_id: String| async move {
-                            phoxal::api::component::v1::capability::gyroscope::subscriber_builder(
-                                &bus,
-                                &component_id,
-                                &local_capability_id,
-                            )
-                            .await
-                            .map_err(phoxal::bus::Error::from)
-                        },
-                        Gyroscope
-                    );
-                }
-                Capability::Gnss(_) => {
-                    spawn_capability!(
-                        component_id,
-                        local_capability_id,
-                        phoxal::api::component::v1::capability::gnss::KIND,
-                        |bus: Bus, component_id: String, local_capability_id: String| async move {
-                            phoxal::api::component::v1::capability::gnss::subscriber_builder(
-                                &bus,
-                                &component_id,
-                                &local_capability_id,
-                            )
-                            .await
-                            .map_err(phoxal::bus::Error::from)
-                        },
-                        Gnss
-                    );
-                }
-                Capability::Battery(_) => {
-                    spawn_capability!(
-                        component_id,
-                        local_capability_id,
-                        phoxal::api::component::v1::capability::battery::KIND,
-                        |bus: Bus, component_id: String, local_capability_id: String| async move {
-                            phoxal::api::component::v1::capability::battery::subscriber_builder(
-                                &bus,
-                                &component_id,
-                                &local_capability_id,
-                            )
-                            .await
-                            .map_err(phoxal::bus::Error::from)
-                        },
-                        Battery
-                    );
-                }
+                Capability::Imu(_) => spawn_capability!(
+                    component_id,
+                    local_capability_id,
+                    |component_id, capability_id| topic::new()
+                        .v1()
+                        .component(component_id)
+                        .imu(capability_id)
+                        .data(),
+                    Imu
+                ),
+                Capability::Accelerometer(_) => spawn_capability!(
+                    component_id,
+                    local_capability_id,
+                    |component_id, capability_id| topic::new()
+                        .v1()
+                        .component(component_id)
+                        .accelerometer(capability_id)
+                        .data(),
+                    Accelerometer
+                ),
+                Capability::Gyroscope(_) => spawn_capability!(
+                    component_id,
+                    local_capability_id,
+                    |component_id, capability_id| topic::new()
+                        .v1()
+                        .component(component_id)
+                        .gyroscope(capability_id)
+                        .data(),
+                    Gyroscope
+                ),
+                Capability::Gnss(_) => spawn_capability!(
+                    component_id,
+                    local_capability_id,
+                    |component_id, capability_id| topic::new()
+                        .v1()
+                        .component(component_id)
+                        .gnss(capability_id)
+                        .data(),
+                    Gnss
+                ),
+                Capability::Battery(_) => spawn_capability!(
+                    component_id,
+                    local_capability_id,
+                    |component_id, capability_id| topic::new()
+                        .v1()
+                        .component(component_id)
+                        .battery(capability_id)
+                        .data(),
+                    Battery
+                ),
                 Capability::EmergencyStop(_) => {}
                 Capability::Motor(_) => {}
-                Capability::Encoder(_) => {
-                    spawn_capability!(
-                        component_id,
-                        local_capability_id,
-                        phoxal::api::component::v1::capability::encoder::KIND,
-                        |bus: Bus, component_id: String, local_capability_id: String| async move {
-                            phoxal::api::component::v1::capability::encoder::subscriber_builder(
-                                &bus,
-                                &component_id,
-                                &local_capability_id,
-                            )
-                            .await
-                            .map_err(phoxal::bus::Error::from)
-                        },
-                        Encoder
-                    );
-                }
-                Capability::Camera(_) | Capability::Depth(_) => {}
-                Capability::Range(_) => {
-                    spawn_capability!(
-                        component_id,
-                        local_capability_id,
-                        phoxal::api::component::v1::capability::range::KIND,
-                        |bus: Bus, component_id: String, local_capability_id: String| async move {
-                            phoxal::api::component::v1::capability::range::subscriber_builder(
-                                &bus,
-                                &component_id,
-                                &local_capability_id,
-                            )
-                            .await
-                            .map_err(phoxal::bus::Error::from)
-                        },
-                        Range
-                    );
-                }
-                Capability::Lidar(_) => {
-                    spawn_capability!(
-                        component_id,
-                        local_capability_id,
-                        phoxal::api::component::v1::capability::lidar::KIND,
-                        |bus: Bus, component_id: String, local_capability_id: String| async move {
-                            phoxal::api::component::v1::capability::lidar::subscriber_builder(
-                                &bus,
-                                &component_id,
-                                &local_capability_id,
-                            )
-                            .await
-                            .map_err(phoxal::bus::Error::from)
-                        },
-                        Lidar
-                    );
-                }
-                Capability::Magnetometer(_) => {
-                    spawn_capability!(
-                        component_id,
-                        local_capability_id,
-                        phoxal::api::component::v1::capability::magnetometer::KIND,
-                        |bus: Bus, component_id: String, local_capability_id: String| async move {
-                            phoxal::api::component::v1::capability::magnetometer::subscriber_builder(
-                                &bus,
-                                &component_id,
-                                &local_capability_id,
-                            )
-                            .await
-                            .map_err(phoxal::bus::Error::from)
-                        },
-                        Magnetometer
-                    );
-                }
-                Capability::Microphone(_) => {
-                    spawn_capability!(
-                        component_id,
-                        local_capability_id,
-                        phoxal::api::component::v1::capability::microphone::KIND,
-                        |bus: Bus, component_id: String, local_capability_id: String| async move {
-                            phoxal::api::component::v1::capability::microphone::subscriber_builder(
-                                &bus,
-                                &component_id,
-                                &local_capability_id,
-                            )
-                            .await
-                            .map_err(phoxal::bus::Error::from)
-                        },
-                        Microphone
-                    );
-                }
-                Capability::Mmwave(_) => {
-                    spawn_capability!(
-                        component_id,
-                        local_capability_id,
-                        phoxal::api::component::v1::capability::mmwave::KIND,
-                        |bus: Bus, component_id: String, local_capability_id: String| async move {
-                            phoxal::api::component::v1::capability::mmwave::subscriber_builder(
-                                &bus,
-                                &component_id,
-                                &local_capability_id,
-                            )
-                            .await
-                            .map_err(phoxal::bus::Error::from)
-                        },
-                        Mmwave
-                    );
-                }
-                Capability::Speaker(_) => {
-                    let capability_id = format!("{component_id}.{local_capability_id}");
-                    let audio_component_id = component_id.to_string();
-                    let audio_local_capability_id = local_capability_id.to_string();
-
-                    spawn_subscription_loop(
-                        bus.clone(),
-                        phoxal::api::component::v1::capability::speaker::audio::path(
-                            &audio_component_id,
-                            &audio_local_capability_id,
-                        ),
-                        move |bus| {
-                            let component_id = audio_component_id.clone();
-                            let local_capability_id = audio_local_capability_id.clone();
-                            async move {
-                                phoxal::api::component::v1::capability::speaker::audio::subscriber_builder(
-                                    &bus,
-                                    &component_id,
-                                    &local_capability_id,
-                                )
-                                .await
-                                .map_err(phoxal::bus::Error::from)
-                            }
-                        },
-                        tx.clone(),
-                        {
-                            let capability_id = capability_id.clone();
-                            move |message| Event::SpeakerData {
-                                capability_id: capability_id.clone(),
-                                message,
-                            }
-                        },
-                    );
-                }
+                Capability::Encoder(_) => spawn_capability!(
+                    component_id,
+                    local_capability_id,
+                    |component_id, capability_id| topic::new()
+                        .v1()
+                        .component(component_id)
+                        .encoder(capability_id)
+                        .data(),
+                    Encoder
+                ),
+                Capability::Camera(_) => spawn_capability!(
+                    component_id,
+                    local_capability_id,
+                    |component_id, capability_id| topic::new()
+                        .v1()
+                        .component(component_id)
+                        .camera(capability_id)
+                        .profile(ANY)
+                        .data(),
+                    Camera
+                ),
+                Capability::Depth(_) => spawn_capability!(
+                    component_id,
+                    local_capability_id,
+                    |component_id, capability_id| topic::new()
+                        .v1()
+                        .component(component_id)
+                        .depth(capability_id)
+                        .profile(ANY)
+                        .data(),
+                    Depth
+                ),
+                Capability::Range(_) => spawn_capability!(
+                    component_id,
+                    local_capability_id,
+                    |component_id, capability_id| topic::new()
+                        .v1()
+                        .component(component_id)
+                        .range(capability_id)
+                        .data(),
+                    Range
+                ),
+                Capability::Lidar(_) => spawn_capability!(
+                    component_id,
+                    local_capability_id,
+                    |component_id, capability_id| topic::new()
+                        .v1()
+                        .component(component_id)
+                        .lidar(capability_id)
+                        .data(),
+                    Lidar
+                ),
+                Capability::Magnetometer(_) => spawn_capability!(
+                    component_id,
+                    local_capability_id,
+                    |component_id, capability_id| topic::new()
+                        .v1()
+                        .component(component_id)
+                        .magnetometer(capability_id)
+                        .data(),
+                    Magnetometer
+                ),
+                Capability::Microphone(_) => {}
+                Capability::Mmwave(_) => spawn_capability!(
+                    component_id,
+                    local_capability_id,
+                    |component_id, capability_id| topic::new()
+                        .v1()
+                        .component(component_id)
+                        .mmwave(capability_id)
+                        .data(),
+                    Mmwave
+                ),
+                Capability::Speaker(_) => {}
                 Capability::Led(_) => {}
             }
         }
     }
 }
 
-fn capability_subscription_path(
-    component_id: &str,
-    _kind: &'static str,
-    capability_id: &str,
-) -> String {
-    phoxal::api::component::v1::capability::default_profile_path(component_id, capability_id)
-}
-
-fn spawn_localize_debug_input_subscriptions(bus: Bus, tx: mpsc::Sender<Event>, scene: &SceneModel) {
-    for subscription in localize_debug_input_subscriptions(&scene.configuration) {
-        match subscription.payload {
-            LocalizeDebugInputPayload::Camera => {
-                let topic_name = subscription.topic_name.clone();
-                let capability_id = subscription.capability_id.clone();
-                spawn_subscription_loop(
-                    bus.clone(),
-                    topic_name.clone(),
-                    move |bus| {
-                        let topic_name = topic_name.clone();
-                        async move {
-                            phoxal::bus::pubsub::subscriber_builder::<Stamped<CameraFrame>>(
-                                &bus,
-                                &topic_name,
-                            )
-                            .await
-                            .map_err(phoxal::bus::Error::from)
-                        }
-                    },
-                    tx.clone(),
-                    move |message| Event::Camera {
-                        capability_id: capability_id.clone(),
-                        message,
-                    },
-                );
-            }
-            LocalizeDebugInputPayload::Depth => {
-                let topic_name = subscription.topic_name.clone();
-                let capability_id = subscription.capability_id.clone();
-                spawn_subscription_loop(
-                    bus.clone(),
-                    topic_name.clone(),
-                    move |bus| {
-                        let topic_name = topic_name.clone();
-                        async move {
-                            phoxal::bus::pubsub::subscriber_builder::<Stamped<DepthFrame>>(
-                                &bus,
-                                &topic_name,
-                            )
-                            .await
-                            .map_err(phoxal::bus::Error::from)
-                        }
-                    },
-                    tx.clone(),
-                    move |message| Event::Depth {
-                        capability_id: capability_id.clone(),
-                        message,
-                    },
-                );
-            }
-            LocalizeDebugInputPayload::Imu => {
-                let topic_name = subscription.topic_name.clone();
-                let capability_id = subscription.capability_id.clone();
-                spawn_subscription_loop(
-                    bus.clone(),
-                    topic_name.clone(),
-                    move |bus| {
-                        let topic_name = topic_name.clone();
-                        async move {
-                            phoxal::bus::pubsub::subscriber_builder::<
-                                Stamped<phoxal::api::component::v1::capability::imu::Sample>,
-                            >(&bus, &topic_name)
-                            .await
-                            .map_err(phoxal::bus::Error::from)
-                        }
-                    },
-                    tx.clone(),
-                    move |message| Event::Imu {
-                        capability_id: capability_id.clone(),
-                        message,
-                    },
-                );
-            }
-        }
-    }
-}
-
-fn localize_debug_input_subscriptions(
-    configuration: &Robot,
-) -> Vec<LocalizeDebugInputSubscription> {
-    match resolve_localize_backend(&configuration.manifest, &configuration.components) {
-        ResolvedLocalizeBackend::OrbSlam3RgbdInertial { camera, depth, imu } => vec![
-            localize_debug_input_subscription("camera", &camera, LocalizeDebugInputPayload::Camera),
-            localize_debug_input_subscription("depth", &depth, LocalizeDebugInputPayload::Depth),
-            localize_debug_input_subscription("imu", &imu, LocalizeDebugInputPayload::Imu),
-        ],
-        ResolvedLocalizeBackend::OrbSlam3Rgbd { camera, depth } => vec![
-            localize_debug_input_subscription("camera", &camera, LocalizeDebugInputPayload::Camera),
-            localize_debug_input_subscription("depth", &depth, LocalizeDebugInputPayload::Depth),
-        ],
-        ResolvedLocalizeBackend::DeadReckoning | ResolvedLocalizeBackend::GnssAnchored { .. } => {
-            Vec::new()
-        }
-    }
-}
-
-fn localize_debug_input_subscription(
-    key: &'static str,
-    capability: &CapabilityRef,
-    payload: LocalizeDebugInputPayload,
-) -> LocalizeDebugInputSubscription {
-    LocalizeDebugInputSubscription {
-        key,
-        topic_name: debug_input_topic(LOCALIZE_RUNTIME_ID, key),
-        capability_id: capability.to_string(),
-        payload,
-    }
-}
-
-fn spawn_subscription_loop<T, Setup, Fut, ToEvent>(
+fn spawn_subscription_loop<T, ToEvent>(
     bus: Bus,
-    topic_name: String,
-    setup: Setup,
+    topic: Topic<PubSub<T>>,
     tx: mpsc::Sender<Event>,
     to_event: ToEvent,
 ) where
-    T: DeserializeOwned + TypedSchema + Send + Sync + 'static,
-    Setup: Fn(Bus) -> Fut + Send + Sync + 'static,
-    Fut: Future<Output = phoxal::bus::Result<TypedSubscriber<T>>> + Send,
-    ToEvent: Fn(T) -> Event + Send + Sync + 'static,
+    T: DeserializeOwned + Send + Sync + 'static,
+    ToEvent: Fn(Received<T>) -> Event + Send + Sync + 'static,
 {
+    let topic_name = topic.key().into_owned();
     tokio::spawn(async move {
         loop {
-            match setup(bus.clone()).await {
+            match bus.subscriber(&topic).await {
                 Ok(subscriber) => loop {
-                    match subscriber.recv_async().await {
-                        Ok(Ok(message)) => {
+                    match subscriber.recv().await {
+                        Ok(message) => {
                             if tx.send(to_event(message)).await.is_err() {
                                 return;
                             }
                         }
-                        Ok(Err(error)) => {
-                            warn!(%error, topic = %topic_name, "Subscription payload decode failed; retrying");
-                            tokio::time::sleep(SUBSCRIPTION_RETRY_DELAY).await;
-                            break;
-                        }
                         Err(error) => {
                             warn!(%error, topic = %topic_name, "Subscription receive failed; retrying");
                             tokio::time::sleep(SUBSCRIPTION_RETRY_DELAY).await;
@@ -539,255 +260,70 @@ fn spawn_subscription_loop<T, Setup, Fut, ToEvent>(
 
 #[cfg(test)]
 mod tests {
-    use std::collections::BTreeMap;
-
-    use phoxal::model::component::v1::Component as ComponentConfig;
-    use phoxal::model::component::v1::capability::{
-        Camera, CameraMode, Capability, Depth as DepthCapability, Gnss, GnssCoordinateSystem, Imu,
-        StructuralTarget,
-    };
-    use phoxal::model::robot::v1::{
-        Component as ModelComponent, ComponentSource, Components, Identity, KinematicConfig,
-        Motion, Phoxal, PhoxalRuntimes, Robot as RobotManifest, Role, SourcePath,
-    };
-    use phoxal::model::structure::Structure;
-    use phoxal::model::v1::Robot;
+    use phoxal::api::v1::topic;
+    use phoxal::bus::topic::{ANY, PubSub, Topic};
 
-    use super::{
-        LocalizeDebugInputPayload, LocalizeDebugInputSubscription, capability_subscription_path,
-        localize_debug_input_subscriptions,
-    };
+    fn topic_key<T>(topic: Topic<PubSub<T>>) -> String {
+        topic.key().into_owned()
+    }
 
     #[test]
-    fn unchanged_capability_subscriptions_use_default_profile() {
+    fn component_subscriptions_use_two_id_data_topics() {
         assert_eq!(
-            capability_subscription_path(
-                "front_left_tof",
-                phoxal::api::component::v1::capability::range::KIND,
-                "range"
+            topic_key(
+                topic::new()
+                    .v1()
+                    .component("front_left_tof")
+                    .range("range")
+                    .data()
             ),
-            "component/front_left_tof/range/profile/default"
+            "v1/component/front_left_tof/range/range/data"
         );
         assert_eq!(
-            capability_subscription_path(
-                "left_drive",
-                phoxal::api::component::v1::capability::encoder::KIND,
-                "encoder"
+            topic_key(
+                topic::new()
+                    .v1()
+                    .component("left_drive")
+                    .encoder("encoder")
+                    .data()
             ),
-            "component/left_drive/encoder/profile/default"
+            "v1/component/left_drive/encoder/encoder/data"
         );
-    }
-
-    #[test]
-    fn standalone_accelerometer_subscriptions_stay_on_default_profile() {
         assert_eq!(
-            capability_subscription_path(
-                "imu",
-                phoxal::api::component::v1::capability::accelerometer::KIND,
-                "accelerometer"
+            topic_key(
+                topic::new()
+                    .v1()
+                    .component("imu")
+                    .accelerometer("accelerometer")
+                    .data()
             ),
-            "component/imu/accelerometer/profile/default"
+            "v1/component/imu/accelerometer/accelerometer/data"
         );
     }
 
     #[test]
-    fn orb_slam_inertial_uses_localize_debug_inputs_with_source_capability_ids() {
-        let robot = localize_rgbd_inertial_robot();
-
+    fn camera_and_depth_subscriptions_use_profile_wildcards() {
         assert_eq!(
-            localize_debug_input_subscriptions(&robot),
-            vec![
-                LocalizeDebugInputSubscription {
-                    key: "camera",
-                    topic_name: "runtime/localize/debug/input/camera".to_string(),
-                    capability_id: "front_camera.rgb".to_string(),
-                    payload: LocalizeDebugInputPayload::Camera,
-                },
-                LocalizeDebugInputSubscription {
-                    key: "depth",
-                    topic_name: "runtime/localize/debug/input/depth".to_string(),
-                    capability_id: "front_camera.depth".to_string(),
-                    payload: LocalizeDebugInputPayload::Depth,
-                },
-                LocalizeDebugInputSubscription {
-                    key: "imu",
-                    topic_name: "runtime/localize/debug/input/imu".to_string(),
-                    capability_id: "imu.imu".to_string(),
-                    payload: LocalizeDebugInputPayload::Imu,
-                },
-            ]
-        );
-    }
-
-    #[test]
-    fn dead_reckoning_and_gnss_localize_create_no_media_debug_input_subscriptions() {
-        let mut dead_reckoning = localize_rgbd_inertial_robot();
-        for component in dead_reckoning.manifest.components.values_mut() {
-            component.roles.clear();
-        }
-        assert!(localize_debug_input_subscriptions(&dead_reckoning).is_empty());
-
-        let mut gnss = dead_reckoning;
-        gnss.manifest.components.sources.insert(
-            "gps_module".to_string(),
-            ComponentSource::Path(SourcePath {
-                path: "./components/gps_module".into(),
-            }),
-        );
-        gnss.manifest.components.insert(
-            "gps".to_string(),
-            ModelComponent {
-                component: "gps_module".to_string(),
-                mount_link: "base_link".to_string(),
-                driver: None,
-                roles: BTreeMap::from([("gnss".to_string(), vec![Role::Localization])]),
-                parameters: BTreeMap::new(),
-            },
+            topic_key(
+                topic::new()
+                    .v1()
+                    .component("front_camera")
+                    .camera("rgb")
+                    .profile(ANY)
+                    .data()
+            ),
+            "v1/component/front_camera/camera/rgb/profile/*/data"
         );
-        gnss.components.insert(
-            "gps_module".to_string(),
-            ComponentConfig {
-                gtin: None,
-                capabilities: BTreeMap::from([(
-                    "gnss".to_string(),
-                    Capability::Gnss(Gnss {
-                        target: link_target(),
-                        publish_rate_hz: 5.0,
-                        coordinate_system: GnssCoordinateSystem::Local,
-                    }),
-                )]),
-            },
+        assert_eq!(
+            topic_key(
+                topic::new()
+                    .v1()
+                    .component("front_camera")
+                    .depth("depth")
+                    .profile(ANY)
+                    .data()
+            ),
+            "v1/component/front_camera/depth/depth/profile/*/data"
         );
-
-        assert!(localize_debug_input_subscriptions(&gnss).is_empty());
-    }
-
-    fn link_target() -> StructuralTarget {
-        StructuralTarget::Link {
-            id: "camera_link".to_string(),
-        }
-    }
-
-    fn localize_rgbd_inertial_robot() -> Robot {
-        Robot {
-            manifest: RobotManifest {
-                phoxal: Phoxal {
-                    cli_min_version: "^0.6".to_string(),
-                },
-                identity: Identity {
-                    id: "testbot".to_string(),
-                    namespace: "dev".to_string(),
-                },
-                structure: "structure.urdf".into(),
-                phoxal_runtimes: PhoxalRuntimes {
-                    version: "^0.1".to_string(),
-                    overrides: BTreeMap::new(),
-                },
-                user_runtimes: BTreeMap::new(),
-                tools: BTreeMap::new(),
-                network: None,
-                motion: Motion {
-                    kinematic: KinematicConfig::Omnidirectional {
-                        actuators: Vec::new(),
-                        encoders: Vec::new(),
-                    },
-                },
-                components: Components {
-                    sources: BTreeMap::from([
-                        (
-                            "rgbd_camera".to_string(),
-                            ComponentSource::Path(SourcePath {
-                                path: "./components/rgbd_camera".into(),
-                            }),
-                        ),
-                        (
-                            "imu_module".to_string(),
-                            ComponentSource::Path(SourcePath {
-                                path: "./components/imu_module".into(),
-                            }),
-                        ),
-                    ]),
-                    instances: BTreeMap::from([
-                        (
-                            "front_camera".to_string(),
-                            ModelComponent {
-                                component: "rgbd_camera".to_string(),
-                                mount_link: "base_link".to_string(),
-                                driver: None,
-                                roles: BTreeMap::from([
-                                    ("rgb".to_string(), vec![Role::Localization]),
-                                    ("depth".to_string(), vec![Role::Localization]),
-                                ]),
-                                parameters: BTreeMap::new(),
-                            },
-                        ),
-                        (
-                            "imu".to_string(),
-                            ModelComponent {
-                                component: "imu_module".to_string(),
-                                mount_link: "base_link".to_string(),
-                                driver: None,
-                                roles: BTreeMap::from([(
-                                    "imu".to_string(),
-                                    vec![Role::Localization],
-                                )]),
-                                parameters: BTreeMap::new(),
-                            },
-                        ),
-                    ]),
-                },
-            },
-            components: BTreeMap::from([
-                (
-                    "rgbd_camera".to_string(),
-                    ComponentConfig {
-                        gtin: None,
-                        capabilities: BTreeMap::from([
-                            (
-                                "rgb".to_string(),
-                                Capability::Camera(Camera {
-                                    target: link_target(),
-                                    mode: CameraMode::Rgb,
-                                    publish_rate_hz: 30.0,
-                                    width_px: 640,
-                                    height_px: 480,
-                                    field_of_view_rad: Some(1.0),
-                                }),
-                            ),
-                            (
-                                "depth".to_string(),
-                                Capability::Depth(DepthCapability {
-                                    target: link_target(),
-                                    publish_rate_hz: 30.0,
-                                    width_px: 640,
-                                    height_px: 400,
-                                    field_of_view_rad: Some(1.0),
-                                    min_range_m: None,
-                                    max_range_m: None,
-                                }),
-                            ),
-                        ]),
-                    },
-                ),
-                (
-                    "imu_module".to_string(),
-                    ComponentConfig {
-                        gtin: None,
-                        capabilities: BTreeMap::from([(
-                            "imu".to_string(),
-                            Capability::Imu(Imu {
-                                target: link_target(),
-                                publish_rate_hz: 100.0,
-                                axes: None,
-                            }),
-                        )]),
-                    },
-                ),
-            ]),
-            structure: Structure::from_urdf_str(
-                r#"<robot name="testbot"><link name="base_link"/></robot>"#,
-            )
-            .expect("test structure parses"),
-        }
     }
 }
diff --git a/src/system.rs b/src/system.rs
index d190ad2..675ab8e 100644
--- a/src/system.rs
+++ b/src/system.rs
@@ -1,6 +1,6 @@
 use anyhow::Result;
-use phoxal::api::presence::Heartbeat as PresenceHeartbeat;
-use phoxal::bus::pubsub::Stamped;
+use phoxal::api::v1::presence::Heartbeat as PresenceHeartbeat;
+use phoxal::bus::typed::Received;
 use rerun::{RecordingStream, TextLog};
 
 use crate::logging::helpers::set_logical_time;
@@ -8,9 +8,9 @@ use crate::paths::SYSTEM_ROOT;
 
 pub(crate) fn log_presence_heartbeat(
     rec: &RecordingStream,
-    stamped: &Stamped<PresenceHeartbeat>,
+    received: &Received<PresenceHeartbeat>,
 ) -> Result<()> {
-    set_logical_time(rec, stamped.timestamp_ns);
+    set_logical_time(rec, received.at_ns);
     rec.log(
         format!("{SYSTEM_ROOT}/presence/heartbeat"),
         &TextLog::new("ping"),