diff --git a/src/io/formats/bag/parallel.rs b/src/io/formats/bag/parallel.rs
index ac13978..77b9f30 100644
--- a/src/io/formats/bag/parallel.rs
+++ b/src/io/formats/bag/parallel.rs
@@ -65,30 +65,26 @@ impl BagFormat {
 
     /// Open a BAG reader from a transport source.
     #[cfg(feature = "remote")]
-    pub fn open_from_transport(
-        mut transport: Box<dyn crate::io::transport::Transport>,
+    pub async fn open_from_transport(
+        transport: Box<dyn crate::io::transport::Transport>,
         path: String,
     ) -> Result<ParallelBagReader> {
-        use std::pin::Pin;
-        use std::task::{Context, Poll, Waker};
+        use std::future::poll_fn;
 
         let mut data = Vec::new();
         let mut buffer = vec![0u8; 64 * 1024];
-        let waker = Waker::noop();
-        let mut cx = Context::from_waker(waker);
-        let mut pinned_transport = unsafe { Pin::new_unchecked(transport.as_mut()) };
+        let mut pinned_transport = Box::into_pin(transport);
 
         loop {
-            match pinned_transport.as_mut().poll_read(&mut cx, &mut buffer) {
-                Poll::Ready(Ok(0)) => break,
-                Poll::Ready(Ok(n)) => data.extend_from_slice(&buffer[..n]),
-                Poll::Ready(Err(e)) => {
+            match poll_fn(|cx| pinned_transport.as_mut().poll_read(cx, &mut buffer)).await {
+                Ok(0) => break,
+                Ok(n) => data.extend_from_slice(&buffer[..n]),
+                Err(e) => {
                     return Err(CodecError::encode(
                         "Transport",
                         format!("Failed to read from {path}: {e}"),
                     ));
                 }
-                Poll::Pending => std::thread::yield_now(),
             }
         }
 
@@ -346,7 +342,7 @@ impl ParallelBagReader {
 
 impl FormatReader for ParallelBagReader {
     #[cfg(feature = "remote")]
-    fn open_from_transport(
+    async fn open_from_transport(
         _transport: Box<dyn crate::io::transport::Transport>,
         _path: String,
     ) -> Result<Self>
diff --git a/src/io/formats/bag/sequential.rs b/src/io/formats/bag/sequential.rs
index 0736680..1d81bc1 100644
--- a/src/io/formats/bag/sequential.rs
+++ b/src/io/formats/bag/sequential.rs
@@ -165,7 +165,7 @@ impl SequentialBagReader {
 
 impl FormatReader for SequentialBagReader {
     #[cfg(feature = "remote")]
-    fn open_from_transport(
+    async fn open_from_transport(
         _transport: Box<dyn crate::io::transport::Transport>,
         _path: String,
     ) -> Result<Self>
diff --git a/src/io/formats/mcap/parallel.rs b/src/io/formats/mcap/parallel.rs
index 9d8ddd8..583ac66 100644
--- a/src/io/formats/mcap/parallel.rs
+++ b/src/io/formats/mcap/parallel.rs
@@ -685,7 +685,7 @@ impl ParallelMcapReader {
 
 impl FormatReader for ParallelMcapReader {
     #[cfg(feature = "remote")]
-    fn open_from_transport(
+    async fn open_from_transport(
         _transport: Box<dyn crate::io::transport::Transport>,
         _path: String,
     ) -> Result<Self>
diff --git a/src/io/formats/mcap/reader.rs b/src/io/formats/mcap/reader.rs
index 5505705..349e62a 100644
--- a/src/io/formats/mcap/reader.rs
+++ b/src/io/formats/mcap/reader.rs
@@ -59,30 +59,26 @@ impl McapFormat {
 
     /// Open an MCAP reader from a transport source.
     #[cfg(feature = "remote")]
-    pub fn open_from_transport(
-        mut transport: Box<dyn crate::io::transport::Transport>,
+    pub async fn open_from_transport(
+        transport: Box<dyn crate::io::transport::Transport>,
         path: String,
     ) -> Result<McapReader> {
-        use std::pin::Pin;
-        use std::task::{Context, Poll, Waker};
+        use std::future::poll_fn;
 
         let mut data = Vec::new();
         let mut buffer = vec![0u8; 64 * 1024];
-        let waker = Waker::noop();
-        let mut cx = Context::from_waker(waker);
-        let mut pinned_transport = unsafe { Pin::new_unchecked(transport.as_mut()) };
+        let mut pinned_transport = Box::into_pin(transport);
 
         loop {
-            match pinned_transport.as_mut().poll_read(&mut cx, &mut buffer) {
-                Poll::Ready(Ok(0)) => break,
-                Poll::Ready(Ok(n)) => data.extend_from_slice(&buffer[..n]),
-                Poll::Ready(Err(e)) => {
+            match poll_fn(|cx| pinned_transport.as_mut().poll_read(cx, &mut buffer)).await {
+                Ok(0) => break,
+                Ok(n) => data.extend_from_slice(&buffer[..n]),
+                Err(e) => {
                     return Err(CodecError::encode(
                         "Transport",
                         format!("Failed to read from {path}: {e}"),
                     ));
                 }
-                Poll::Pending => std::thread::yield_now(),
             }
         }
 
@@ -271,7 +267,7 @@ impl McapReader {
 
 impl FormatReader for McapReader {
     #[cfg(feature = "remote")]
-    fn open_from_transport(
+    async fn open_from_transport(
         _transport: Box<dyn crate::io::transport::Transport>,
         _path: String,
     ) -> Result<Self>
diff --git a/src/io/formats/mcap/sequential.rs b/src/io/formats/mcap/sequential.rs
index 6233704..611fb6b 100644
--- a/src/io/formats/mcap/sequential.rs
+++ b/src/io/formats/mcap/sequential.rs
@@ -233,7 +233,7 @@ impl SequentialMcapReader {
 
 impl FormatReader for SequentialMcapReader {
     #[cfg(feature = "remote")]
-    fn open_from_transport(
+    async fn open_from_transport(
         _transport: Box<dyn crate::io::transport::Transport>,
         _path: String,
     ) -> Result<Self>
diff --git a/src/io/formats/mcap/two_pass.rs b/src/io/formats/mcap/two_pass.rs
index 1832281..ee4e205 100644
--- a/src/io/formats/mcap/two_pass.rs
+++ b/src/io/formats/mcap/two_pass.rs
@@ -583,7 +583,7 @@ impl TwoPassMcapReader {
 
 impl FormatReader for TwoPassMcapReader {
     #[cfg(feature = "remote")]
-    fn open_from_transport(
+    async fn open_from_transport(
         _transport: Box<dyn crate::io::transport::Transport>,
         _path: String,
     ) -> Result<Self>
diff --git a/src/io/formats/rrd/parallel.rs b/src/io/formats/rrd/parallel.rs
index 3c37d2f..075ea93 100644
--- a/src/io/formats/rrd/parallel.rs
+++ b/src/io/formats/rrd/parallel.rs
@@ -470,7 +470,7 @@ impl Iterator for RrdDecodedMessageWithTimestampStream<'_> {
 
 impl FormatReader for ParallelRrdReader {
     #[cfg(feature = "remote")]
-    fn open_from_transport(
+    async fn open_from_transport(
         _transport: Box<dyn crate::io::transport::Transport>,
         _path: String,
     ) -> Result<Self>
diff --git a/src/io/formats/rrd/reader.rs b/src/io/formats/rrd/reader.rs
index dc63bf9..6bf235a 100644
--- a/src/io/formats/rrd/reader.rs
+++ b/src/io/formats/rrd/reader.rs
@@ -56,30 +56,26 @@ impl RrdFormat {
 
     /// Open an RRD reader from a transport source.
     #[cfg(feature = "remote")]
-    pub fn open_from_transport(
-        mut transport: Box<dyn crate::io::transport::Transport>,
+    pub async fn open_from_transport(
+        transport: Box<dyn crate::io::transport::Transport>,
         path: String,
     ) -> Result<ParallelRrdReader> {
-        use std::pin::Pin;
-        use std::task::{Context, Poll, Waker};
+        use std::future::poll_fn;
 
         let mut data = Vec::new();
         let mut buffer = vec![0u8; 64 * 1024];
-        let waker = Waker::noop();
-        let mut cx = Context::from_waker(waker);
-        let mut pinned_transport = unsafe { Pin::new_unchecked(transport.as_mut()) };
+        let mut pinned_transport = Box::into_pin(transport);
 
         loop {
-            match pinned_transport.as_mut().poll_read(&mut cx, &mut buffer) {
-                Poll::Ready(Ok(0)) => break,
-                Poll::Ready(Ok(n)) => data.extend_from_slice(&buffer[..n]),
-                Poll::Ready(Err(e)) => {
+            match poll_fn(|cx| pinned_transport.as_mut().poll_read(cx, &mut buffer)).await {
+                Ok(0) => break,
+                Ok(n) => data.extend_from_slice(&buffer[..n]),
+                Err(e) => {
                     return Err(CodecError::encode(
                         "Transport",
                         format!("Failed to read from {path}: {e}"),
                     ));
                 }
-                Poll::Pending => std::thread::yield_now(),
             }
         }
 
@@ -394,7 +390,7 @@ impl RrdReader {
 
 impl FormatReader for RrdReader {
     #[cfg(feature = "remote")]
-    fn open_from_transport(
+    async fn open_from_transport(
         _transport: Box<dyn crate::io::transport::Transport>,
         _path: String,
     ) -> Result<Self>
diff --git a/src/io/reader/mod.rs b/src/io/reader/mod.rs
index 0686b4d..6cda2e0 100644
--- a/src/io/reader/mod.rs
+++ b/src/io/reader/mod.rs
@@ -248,30 +248,68 @@ impl RoboReader {
                 let path_for_detection = path.split('?').next().unwrap_or(path);
                 let path_obj = std::path::Path::new(path_for_detection);
                 let format = detect_format(path_obj)?;
+                let path_owned = path.to_string();
 
                 match format {
                     FileFormat::Mcap => {
+                        let open_path = path_owned.clone();
+                        let error_path = path_owned.clone();
+                        let reader = std::thread::spawn(move || {
+                            shared_runtime().block_on(McapFormat::open_from_transport(
+                                transport, open_path,
+                            ))
+                        })
+                        .join()
+                        .map_err(|_| {
+                            CodecError::encode(
+                                "RoboReader",
+                                format!(
+                                    "Failed to join MCAP transport initialization for '{error_path}'"
+                                ),
+                            )
+                        })??;
                         return Ok(Self {
-                            inner: Box::new(McapFormat::open_from_transport(
-                                transport,
-                                path.to_string(),
-                            )?),
+                            inner: Box::new(reader),
                         });
                     }
                     FileFormat::Bag => {
+                        let open_path = path_owned.clone();
+                        let error_path = path_owned.clone();
+                        let reader = std::thread::spawn(move || {
+                            shared_runtime()
+                                .block_on(BagFormat::open_from_transport(transport, open_path))
+                        })
+                        .join()
+                        .map_err(|_| {
+                            CodecError::encode(
+                                "RoboReader",
+                                format!(
+                                    "Failed to join BAG transport initialization for '{error_path}'"
+                                ),
+                            )
+                        })??;
                         return Ok(Self {
-                            inner: Box::new(BagFormat::open_from_transport(
-                                transport,
-                                path.to_string(),
-                            )?),
+                            inner: Box::new(reader),
                         });
                     }
                     FileFormat::Rrd => {
+                        let open_path = path_owned.clone();
+                        let error_path = path_owned.clone();
+                        let reader = std::thread::spawn(move || {
+                            shared_runtime()
+                                .block_on(RrdFormat::open_from_transport(transport, open_path))
+                        })
+                        .join()
+                        .map_err(|_| {
+                            CodecError::encode(
+                                "RoboReader",
+                                format!(
+                                    "Failed to join RRD transport initialization for '{error_path}'"
+                                ),
+                            )
+                        })??;
                         return Ok(Self {
-                            inner: Box::new(RrdFormat::open_from_transport(
-                                transport,
-                                path.to_string(),
-                            )?),
+                            inner: Box::new(reader),
                         });
                     }
                     FileFormat::Unknown => {
@@ -467,7 +505,7 @@ impl RoboReader {
 
 impl FormatReader for RoboReader {
     #[cfg(feature = "remote")]
-    fn open_from_transport(
+    async fn open_from_transport(
         transport: Box<dyn crate::io::transport::Transport>,
         path: String,
     ) -> Result<Self>
@@ -479,9 +517,9 @@ impl FormatReader for RoboReader {
         let format = detect_format(path_obj)?;
 
         let inner: Box<dyn FormatReader> = match format {
-            FileFormat::Mcap => Box::new(McapFormat::open_from_transport(transport, path)?),
-            FileFormat::Bag => Box::new(BagFormat::open_from_transport(transport, path)?),
-            FileFormat::Rrd => Box::new(RrdFormat::open_from_transport(transport, path)?),
+            FileFormat::Mcap => Box::new(McapFormat::open_from_transport(transport, path).await?),
+            FileFormat::Bag => Box::new(BagFormat::open_from_transport(transport, path).await?),
+            FileFormat::Rrd => Box::new(RrdFormat::open_from_transport(transport, path).await?),
             FileFormat::Unknown => {
                 return Err(CodecError::parse(
                     "RoboReader",
@@ -568,7 +606,7 @@ mod tests {
 
     impl FormatReader for MockReader {
         #[cfg(feature = "remote")]
-        fn open_from_transport(
+        async fn open_from_transport(
             _transport: Box<dyn crate::io::transport::Transport>,
             path: String,
         ) -> Result<Self>
@@ -925,6 +963,124 @@ mod tests {
         assert!(result.unwrap().is_none());
     }
 
+    /// Regression test for transport readers:
+    /// Pending must not cause busy-spin in open_from_transport.
+    #[test]
+    #[cfg(feature = "remote")]
+    fn test_transport_open_does_not_busy_spin_on_pending() {
+        use crate::io::traits::FormatReader;
+        use crate::io::transport::Transport;
+        use std::io;
+        use std::pin::Pin;
+        use std::sync::Arc;
+        use std::sync::atomic::{AtomicUsize, Ordering};
+        use std::task::{Context, Poll};
+        use std::time::{Duration, Instant};
+
+        struct PendingThenReadyTransport {
+            data: Vec<u8>,
+            pos: usize,
+            delay: Duration,
+            start: Instant,
+            wake_scheduled: bool,
+            poll_count: Arc<AtomicUsize>,
+        }
+
+        impl PendingThenReadyTransport {
+            fn new(data: Vec<u8>, delay: Duration, poll_count: Arc<AtomicUsize>) -> Self {
+                Self {
+                    data,
+                    pos: 0,
+                    delay,
+                    start: Instant::now(),
+                    wake_scheduled: false,
+                    poll_count,
+                }
+            }
+        }
+
+        impl Transport for PendingThenReadyTransport {
+            fn poll_read(
+                mut self: Pin<&mut Self>,
+                cx: &mut Context<'_>,
+                buf: &mut [u8],
+            ) -> Poll<io::Result<usize>> {
+                self.poll_count.fetch_add(1, Ordering::Relaxed);
+
+                if self.start.elapsed() < self.delay {
+                    if !self.wake_scheduled {
+                        self.wake_scheduled = true;
+                        let waker = cx.waker().clone();
+                        let remaining = self.delay.saturating_sub(self.start.elapsed());
+                        std::thread::spawn(move || {
+                            std::thread::sleep(remaining);
+                            waker.wake();
+                        });
+                    }
+                    return Poll::Pending;
+                }
+
+                let this = self.as_mut().get_mut();
+                if this.pos >= this.data.len() {
+                    return Poll::Ready(Ok(0));
+                }
+
+                let n = buf.len().min(this.data.len() - this.pos);
+                buf[..n].copy_from_slice(&this.data[this.pos..this.pos + n]);
+                this.pos += n;
+                Poll::Ready(Ok(n))
+            }
+
+            fn poll_seek(
+                mut self: Pin<&mut Self>,
+                _cx: &mut Context<'_>,
+                pos: u64,
+            ) -> Poll<io::Result<u64>> {
+                let this = self.as_mut().get_mut();
+                this.pos = pos as usize;
+                Poll::Ready(Ok(pos))
+            }
+
+            fn position(&self) -> u64 {
+                self.pos as u64
+            }
+
+            fn len(&self) -> Option<u64> {
+                Some(self.data.len() as u64)
+            }
+
+            fn is_seekable(&self) -> bool {
+                true
+            }
+        }
+
+        let fixture_path = std::path::PathBuf::from(env!("CARGO_MANIFEST_DIR"))
+            .join("tests/fixtures/robocodec_test_0.mcap");
+        if !fixture_path.exists() {
+            eprintln!("Skipping test: fixture not found");
+            return;
+        }
+
+        let data = std::fs::read(fixture_path).expect("failed to read mcap fixture");
+        let poll_count = Arc::new(AtomicUsize::new(0));
+        let transport = PendingThenReadyTransport::new(
+            data,
+            Duration::from_millis(25),
+            Arc::clone(&poll_count),
+        );
+
+        let result = shared_runtime().block_on(RoboReader::open_from_transport(
+            Box::new(transport),
+            "pending.mcap".to_string(),
+        ));
+
+        assert!(result.is_ok(), "open_from_transport should succeed");
+        assert!(
+            poll_count.load(Ordering::Relaxed) < 200,
+            "poll_read should not busy-spin while Pending"
+        );
+    }
+
     /// Test that BAG opens via FormatReader::open_from_transport
     /// Regression test: Previously BAG returned "unsupported" error
     #[test]
@@ -947,7 +1103,10 @@ mod tests {
             Box::new(MemoryTransport::new(data)) as Box<dyn crate::io::transport::Transport>;
 
         // This should NOT return "unsupported" error anymore
-        let result = RoboReader::open_from_transport(transport, "test.bag".to_string());
+        let result = shared_runtime().block_on(RoboReader::open_from_transport(
+            transport,
+            "test.bag".to_string(),
+        ));
 
         match result {
             Ok(reader) => {
@@ -984,7 +1143,10 @@ mod tests {
             Box::new(MemoryTransport::new(data)) as Box<dyn crate::io::transport::Transport>;
 
         // This should NOT return "unsupported" error anymore
-        let result = RoboReader::open_from_transport(transport, "test.rrd".to_string());
+        let result = shared_runtime().block_on(RoboReader::open_from_transport(
+            transport,
+            "test.rrd".to_string(),
+        ));
 
         match result {
             Ok(reader) => {
diff --git a/src/io/s3/reader.rs b/src/io/s3/reader.rs
index 910a36c..5fc4f85 100644
--- a/src/io/s3/reader.rs
+++ b/src/io/s3/reader.rs
@@ -804,7 +804,7 @@ impl S3Reader {
 
 impl FormatReader for S3Reader {
     #[cfg(feature = "remote")]
-    fn open_from_transport(
+    async fn open_from_transport(
         _transport: Box<dyn crate::io::transport::Transport>,
         _path: String,
     ) -> crate::Result<Self>
diff --git a/src/io/streaming/reader.rs b/src/io/streaming/reader.rs
index 7d4de98..b57e92a 100644
--- a/src/io/streaming/reader.rs
+++ b/src/io/streaming/reader.rs
@@ -104,15 +104,14 @@ impl StreamingRoboReader {
                 let format = detect_format(path_obj)?;
 
                 let inner: Box<dyn FormatReader> = match format {
-                    FileFormat::Mcap => Box::new(McapFormat::open_from_transport(
-                        transport,
-                        path.to_string(),
-                    )?),
+                    FileFormat::Mcap => Box::new(
+                        McapFormat::open_from_transport(transport, path.to_string()).await?,
+                    ),
                     FileFormat::Bag => {
-                        Box::new(BagFormat::open_from_transport(transport, path.to_string())?)
+                        Box::new(BagFormat::open_from_transport(transport, path.to_string()).await?)
                     }
                     FileFormat::Rrd => {
-                        Box::new(RrdFormat::open_from_transport(transport, path.to_string())?)
+                        Box::new(RrdFormat::open_from_transport(transport, path.to_string()).await?)
                     }
                     FileFormat::Unknown => {
                         return Err(CodecError::parse(
diff --git a/src/io/streaming/stream.rs b/src/io/streaming/stream.rs
index 8a61139..b3d4244 100644
--- a/src/io/streaming/stream.rs
+++ b/src/io/streaming/stream.rs
@@ -210,9 +210,7 @@ impl FrameStream {
 
     /// Process a message and return any completed frames.
     pub fn process_message(&mut self, msg: TimestampedMessage) -> Vec<AlignedFrame> {
-        self.message_buffer.push(msg.clone());
-        self.progress
-            .set_messages_buffered(self.message_buffer.len());
+        let log_time = msg.log_time;
 
         // Extract state data if this is a state topic
         if self.config.state_topics.contains(&msg.topic)
@@ -222,8 +220,17 @@ impl FrameStream {
             entries.push((msg.log_time, state));
         }
 
+        // Only buffer image-topic messages (which are searched by find_image_at_time).
+        // Non-image/non-state messages are never used, so skip them to avoid
+        // cloning megabytes of image data for irrelevant topics.
+        if self.config.image_topics.contains(&msg.topic) {
+            self.message_buffer.push(msg);
+            self.progress
+                .set_messages_buffered(self.message_buffer.len());
+        }
+
         // Check if we should emit frames
-        self.try_emit_frames(msg.log_time)
+        self.try_emit_frames(log_time)
     }
 
     /// Drain any remaining frames from the buffer.
@@ -298,6 +305,22 @@ impl FrameStream {
             self.next_frame_time = Some(frame_time + frame_interval_ns);
         }
 
+        // Evict old messages that can no longer match any future frame.
+        // Without this, the buffer grows unboundedly (hundreds of thousands of
+        // messages, including MB-sized images), causing O(n) scans to stall.
+        if let Some(next_frame) = self.next_frame_time {
+            let image_tolerance = 16_666_667u64; // ~16ms, same as find_image_at_time
+            let msg_cutoff = next_frame.saturating_sub(image_tolerance);
+            self.message_buffer.retain(|msg| msg.log_time >= msg_cutoff);
+            self.progress
+                .set_messages_buffered(self.message_buffer.len());
+
+            let state_cutoff = next_frame.saturating_sub(self.config.max_state_latency_ns);
+            for entries in self.state_buffer.values_mut() {
+                entries.retain(|(time, _)| *time >= state_cutoff);
+            }
+        }
+
         frames
     }
 
diff --git a/src/io/traits.rs b/src/io/traits.rs
index 46f7cae..523d207 100644
--- a/src/io/traits.rs
+++ b/src/io/traits.rs
@@ -62,6 +62,7 @@ impl<T> DecodedMessageIterator for T where
 ///     println!("Messages: {}", reader.message_count());
 /// }
 /// ```
+#[allow(async_fn_in_trait)]
 pub trait FormatReader: Send + Sync {
     /// Open a reader from any transport source.
     ///
@@ -86,7 +87,7 @@ pub trait FormatReader: Send + Sync {
     /// - The data is not a valid file for this format
     /// - Required metadata cannot be extracted
     #[cfg(feature = "remote")]
-    fn open_from_transport(
+    async fn open_from_transport(
         transport: Box<dyn crate::io::transport::Transport>,
         path: String,
     ) -> Result<Self>
@@ -775,7 +776,7 @@ mod tests {
 
         impl FormatReader for TestReader {
             #[cfg(feature = "remote")]
-            fn open_from_transport(
+            async fn open_from_transport(
                 _transport: Box<dyn crate::io::transport::Transport>,
                 _path: String,
             ) -> Result<Self>
@@ -952,7 +953,7 @@ mod tests {
 
         impl FormatReader for TestReader {
             #[cfg(feature = "remote")]
-            fn open_from_transport(
+            async fn open_from_transport(
                 _transport: Box<dyn crate::io::transport::Transport>,
                 _path: String,
             ) -> Result<Self>
@@ -1022,7 +1023,7 @@ mod tests {
 
         impl FormatReader for TestReader {
             #[cfg(feature = "remote")]
-            fn open_from_transport(
+            async fn open_from_transport(
                 _transport: Box<dyn crate::io::transport::Transport>,
                 _path: String,
             ) -> Result<Self>
@@ -1087,7 +1088,7 @@ mod tests {
 
         impl FormatReader for TestReader {
             #[cfg(feature = "remote")]
-            fn open_from_transport(
+            async fn open_from_transport(
                 _transport: Box<dyn crate::io::transport::Transport>,
                 _path: String,
             ) -> Result<Self>
diff --git a/tests/bag_transport_tests.rs b/tests/bag_transport_tests.rs
index 1f43109..c19c119 100644
--- a/tests/bag_transport_tests.rs
+++ b/tests/bag_transport_tests.rs
@@ -28,8 +28,11 @@ fn bag_transport_from_fixture(filename: &str) -> Box<dyn robocodec::io::transpor
 #[cfg(feature = "remote")]
 fn test_bag_format_open_from_transport() {
     let transport = bag_transport_from_fixture("robocodec_test_15.bag");
-    let reader = BagFormat::open_from_transport(transport, "memory://test.bag".to_string())
-        .expect("Failed to open BAG via transport");
+    let reader = tokio_test::block_on(BagFormat::open_from_transport(
+        transport,
+        "memory://test.bag".to_string(),
+    ))
+    .expect("Failed to open BAG via transport");
 
     // Should have at least one channel
     assert!(
@@ -57,10 +60,10 @@ fn test_bag_format_transport_channels_match_local() {
     let bag_path = fixture_path("robocodec_test_15.bag");
 
     // Open via transport-based reader
-    let transport_reader = BagFormat::open_from_transport(
+    let transport_reader = tokio_test::block_on(BagFormat::open_from_transport(
         bag_transport_from_fixture("robocodec_test_15.bag"),
         "memory://test.bag".to_string(),
-    )
+    ))
     .expect("Failed to open with transport");
     let transport_channels: HashMap<_, _> = transport_reader
         .channels()
@@ -111,10 +114,10 @@ fn test_bag_format_transport_channels_match_local() {
 #[test]
 #[cfg(feature = "remote")]
 fn test_robo_reader_open_from_transport_bag() {
-    let reader = RoboReader::open_from_transport(
+    let reader = tokio_test::block_on(RoboReader::open_from_transport(
         bag_transport_from_fixture("robocodec_test_15.bag"),
         "memory://test.bag".to_string(),
-    )
+    ))
     .expect("Failed to open RoboReader from transport");
 
     assert!(matches!(
diff --git a/tests/rrd_transport_tests.rs b/tests/rrd_transport_tests.rs
index 09b5163..17b6e9f 100644
--- a/tests/rrd_transport_tests.rs
+++ b/tests/rrd_transport_tests.rs
@@ -27,10 +27,10 @@ fn rrd_transport_from_fixture(filename: &str) -> Box<dyn robocodec::io::transpor
 #[test]
 #[cfg(feature = "remote")]
 fn test_rrd_format_open_from_transport() {
-    let reader = RrdFormat::open_from_transport(
+    let reader = tokio_test::block_on(RrdFormat::open_from_transport(
         rrd_transport_from_fixture("file1.rrd"),
         "memory://test.rrd".to_string(),
-    )
+    ))
     .expect("Failed to open RRD via transport");
 
     // Should have at least one channel
@@ -59,10 +59,10 @@ fn test_rrd_format_transport_channels_match_local() {
     let rrd_path = fixture_path("file1.rrd");
 
     // Open via transport-based reader
-    let transport_reader = RrdFormat::open_from_transport(
+    let transport_reader = tokio_test::block_on(RrdFormat::open_from_transport(
         rrd_transport_from_fixture("file1.rrd"),
         "memory://test.rrd".to_string(),
-    )
+    ))
     .expect("Failed to open with transport");
     let transport_channels: HashMap<_, _> = transport_reader
         .channels()
@@ -103,10 +103,10 @@ fn test_rrd_format_transport_channels_match_local() {
 #[test]
 #[cfg(feature = "remote")]
 fn test_robo_reader_open_from_transport_rrd() {
-    let reader = RoboReader::open_from_transport(
+    let reader = tokio_test::block_on(RoboReader::open_from_transport(
         rrd_transport_from_fixture("file1.rrd"),
         "memory://test.rrd".to_string(),
-    )
+    ))
     .expect("Failed to open RoboReader from transport");
 
     assert!(matches!(