diff --git a/crates/test-util/src/wast.rs b/crates/test-util/src/wast.rs
index 6c29339431bc..227d61c42374 100644
--- a/crates/test-util/src/wast.rs
+++ b/crates/test-util/src/wast.rs
@@ -581,10 +581,12 @@ impl WastTest {
             return true;
         }
 
-        // Wasmtime doesn't implement `assert_{invalid,malformed}_custom`
         let unsupported = [
+            // Wasmtime doesn't implement `assert_{invalid,malformed}_custom`
             "spec_testsuite/proposals/extended-name-section/custom/name_annot.wast",
             "spec_testsuite/proposals/extended-name-section/custom/name.wast",
+            // Waiting on WebAssembly/component-model#755 to land
+            "component-model/test/async/zero-length.wast",
         ];
         if unsupported.iter().any(|part| self.path.ends_with(part)) {
             return true;
diff --git a/crates/wasmtime/src/runtime/component/concurrent/futures_and_streams.rs b/crates/wasmtime/src/runtime/component/concurrent/futures_and_streams.rs
index 01fe86d3ab66..4001bf6c6766 100644
--- a/crates/wasmtime/src/runtime/component/concurrent/futures_and_streams.rs
+++ b/crates/wasmtime/src/runtime/component/concurrent/futures_and_streams.rs
@@ -497,13 +497,7 @@ impl<D: 'static> DirectDestination<'_, D> {
 
         let memory = instance
             .options_memory_mut(self.store.0, options)
-            .get_mut(
-                if count == 0 {
-                    0
-                } else {
-                    address + guest_offset.as_usize()
-                }..,
-            )
+            .get_mut(address + guest_offset.as_usize()..)
             .and_then(|b| b.get_mut(..(count.as_usize() - guest_offset.as_usize())));
         match memory {
             Some(memory) => Ok(memory),
@@ -904,12 +898,6 @@ impl<'a, T> Source<'a, T> {
                 bail_bug!("expected WriteState::GuestReady");
             };
 
-            // Zero-length writes are exempt from bounds/alignment checks, so
-            // there's nothing to lift.
-            if guest_offset >= count {
-                return Ok(());
-            }
-
             let cx = &mut LiftContext::new(store.0.store_opaque_mut(), options, instance)?;
             let ty = ty.payload(cx.types);
             let old_remaining = buffer.remaining_capacity();
@@ -1022,13 +1010,7 @@ impl<D: 'static> DirectSource<'_, D> {
 
         let memory = instance
             .options_memory(self.store.0, options)
-            .get(
-                if count == 0 {
-                    0
-                } else {
-                    address + guest_offset.as_usize()
-                }..,
-            )
+            .get(address + guest_offset.as_usize()..)
             .and_then(|b| b.get(..(count.as_usize() - guest_offset.as_usize())));
         match memory {
             Some(memory) => Ok(memory),
@@ -3390,8 +3372,9 @@ impl Instance {
         let read_types = read_component.types();
         let count = count.as_usize();
 
-        // Validate `write_ty` w.r.t. `write_address` to ensure it's properly
-        // aligned and in-bounds.
+        // Double-check that both buffers are properly aligned and in-bounds.
+        // This was already validated in `check_bounds` when the read/write was
+        // issued, so failure here is a bug.
         //
         // FIXME(#4311): this and the rest of this function use the 32-bit
         // layout of payloads, which needs memory64 handling.
@@ -3404,15 +3387,18 @@ impl Instance {
             Some(abi) => usize::try_from(abi.size)? * count,
             None => usize::try_from(write_abi.size32)? * count,
         };
-        if write_length_in_bytes > 0 {
+        if write_payload_ty.is_some() {
             if write_address % usize::try_from(write_abi.align32)? != 0 {
-                bail!("write pointer not aligned");
+                bail_bug!("write pointer not aligned");
             }
-            write_instance
+            if write_instance
                 .options_memory(store, write_options)
                 .get(write_address..)
                 .and_then(|b| b.get(..write_length_in_bytes))
-                .ok_or_else(|| crate::format_err!("write pointer out of bounds"))?;
+                .is_none()
+            {
+                bail_bug!("write pointer out of bounds");
+            }
         }
 
         let read_payload_ty = read_ty.payload(read_types);
@@ -3424,15 +3410,18 @@ impl Instance {
             Some(abi) => usize::try_from(abi.size)? * count,
             None => usize::try_from(read_abi.size32)? * count,
         };
-        if read_length_in_bytes > 0 {
+        if read_payload_ty.is_some() {
             if read_address % usize::try_from(read_abi.align32)? != 0 {
-                bail!("read pointer not aligned");
+                bail_bug!("read pointer not aligned");
             }
-            read_instance
+            if read_instance
                 .options_memory(store, read_options)
                 .get(read_address..)
                 .and_then(|b| b.get(..read_length_in_bytes))
-                .ok_or_else(|| crate::format_err!("read pointer out of bounds"))?;
+                .is_none()
+            {
+                bail_bug!("read pointer out of bounds");
+            }
         }
 
         if write_runtime_instance == read_runtime_instance
@@ -3571,14 +3560,11 @@ impl Instance {
             TransmitIndex::Future(ty) => types[types[ty].ty].payload,
             TransmitIndex::Stream(ty) => types[types[ty].ty].payload,
         };
+        // Empty payloads ignore the pointer entirely, but otherwise the
+        // alignment and bounds checks apply even to zero-length reads/writes.
         let Some(payload) = payload else {
             return Ok(());
         };
-        // As-written the spec currently exempts zero-length reads/writes from
-        // bounds/alignment checks.
-        if count == 0 {
-            return Ok(());
-        }
         let abi = types.canonical_abi(&payload);
         // FIXME(#4311): needs memory64 handling
         let size = usize::try_from(abi.size32)?;
diff --git a/tests/all/component_model/async.rs b/tests/all/component_model/async.rs
index d5c7c75010b3..aa094ab0d135 100644
--- a/tests/all/component_model/async.rs
+++ b/tests/all/component_model/async.rs
@@ -2609,7 +2609,7 @@ async fn zero_length_reads_and_writes() -> Result<()> {
                     (local $stream-reader i32)
                     (local $ret i32)
 
-                    (local.set $ret (call $stream.read (local.get $stream) (i32.const 0xFFFFFF) (i32.const 0)))
+                    (local.set $ret (call $stream.read (local.get $stream) (i32.const 0x10000) (i32.const 0)))
                     (if (i32.ne (local.get $ret) (i32.const 1 (; DROPPED ;))) (then unreachable))
 
                     (local.set $tmp (call $stream.new))
@@ -2618,7 +2618,7 @@ async fn zero_length_reads_and_writes() -> Result<()> {
 
                     (call $task.return (local.get $stream-reader))
 
-                    (local.set $ret (call $stream.write (local.get $stream-writer) (i32.const 0xFFFFFF) (i32.const 0)))
+                    (local.set $ret (call $stream.write (local.get $stream-writer) (i32.const 0x10000) (i32.const 0)))
                     (if (i32.ne (local.get $ret) (i32.const -1 (; BLOCKED ;))) (then unreachable))
 
                     (local.set $set (call $waitable-set.new))
@@ -2677,8 +2677,9 @@ async fn zero_length_reads_and_writes() -> Result<()> {
             _finish: bool,
         ) -> Poll<Result<StreamResult>> {
             // This used to panic on zero-length reads with out-of-bounds
-            // addresses:
-            _ = destination.as_direct(store, 1).remaining();
+            // addresses, and should work with an address at the very end of
+            // memory:
+            _ = destination.as_direct(store, 1).remaining()?;
             Poll::Ready(Ok(StreamResult::Dropped))
         }
     }
@@ -2696,8 +2697,9 @@ async fn zero_length_reads_and_writes() -> Result<()> {
             _finish: bool,
         ) -> Poll<Result<StreamResult>> {
             // This used to panic on zero-length reads with out-of-bounds
-            // addresses:
-            _ = source.as_direct(store).remaining();
+            // addresses, and should work with an address at the very end of
+            // memory:
+            _ = source.as_direct(store).remaining()?;
             Poll::Ready(Ok(StreamResult::Dropped))
         }
     }
diff --git a/tests/all/component_model/stream_contract.rs b/tests/all/component_model/stream_contract.rs
index ed1f9ce9edb7..b8d999fd1086 100644
--- a/tests/all/component_model/stream_contract.rs
+++ b/tests/all/component_model/stream_contract.rs
@@ -227,8 +227,8 @@ async fn consumer_completed_without_consuming() -> Result<()> {
     assert_eq!(guest_write(false, 100, 0).await?, 0);
     assert_eq!(guest_write(true, 100, 4).await?, 4 << 4); // COMPLETED(4)
 
-    // A zero-length write isn't bounds-checked, even when it's consumed.
-    assert_eq!(guest_write(true, u32::MAX, 0).await?, 0);
+    // A zero-length write is still bounds-checked.
+    assert!(guest_write(true, u32::MAX, 0).await.is_err());
     Ok(())
 }