diff --git a/crates/cranelift/src/compiler/component.rs b/crates/cranelift/src/compiler/component.rs
index 054f864b4be5..29ec6441e813 100644
--- a/crates/cranelift/src/compiler/component.rs
+++ b/crates/cranelift/src/compiler/component.rs
@@ -742,6 +742,14 @@ impl<'a> TrampolineCompiler<'a> {
                     |_, _| {},
                 );
             }
+            Trampoline::CheckBlocking => {
+                self.translate_libcall(
+                    host::check_blocking,
+                    TrapSentinel::Falsy,
+                    WasmArgs::InRegisters,
+                    |_, _| {},
+                );
+            }
             Trampoline::ContextGet { instance, slot } => {
                 self.translate_libcall(
                     host::context_get,
diff --git a/crates/environ/src/component.rs b/crates/environ/src/component.rs
index e1c7962d383b..0eed0d14ad54 100644
--- a/crates/environ/src/component.rs
+++ b/crates/environ/src/component.rs
@@ -132,6 +132,7 @@ macro_rules! foreach_builtin_component_function {
                 caller_instance: u32,
                 callee_instance: u32,
                 task_return_type: u32,
+                callee_async: u32,
                 string_encoding: u32,
                 result_count_or_max_if_async: u32,
                 storage: ptr_u8,
@@ -186,6 +187,8 @@ macro_rules! foreach_builtin_component_function {
             #[cfg(feature = "component-model-async")]
             error_context_transfer(vmctx: vmctx, src_idx: u32, src_table: u32, dst_table: u32) -> u64;
             #[cfg(feature = "component-model-async")]
+            check_blocking(vmctx: vmctx) -> bool;
+            #[cfg(feature = "component-model-async")]
             context_get(vmctx: vmctx, caller_instance: u32, slot: u32) -> u64;
             #[cfg(feature = "component-model-async")]
             context_set(vmctx: vmctx, caller_instance: u32, slot: u32, val: u32) -> bool;
diff --git a/crates/environ/src/component/dfg.rs b/crates/environ/src/component/dfg.rs
index 9df5116e2b04..30f87358aff4 100644
--- a/crates/environ/src/component/dfg.rs
+++ b/crates/environ/src/component/dfg.rs
@@ -478,6 +478,7 @@ pub enum Trampoline {
     FutureTransfer,
     StreamTransfer,
     ErrorContextTransfer,
+    CheckBlocking,
     ContextGet {
         instance: RuntimeComponentInstanceIndex,
         slot: u32,
@@ -1160,6 +1161,7 @@ impl LinearizeDfg<'_> {
             Trampoline::FutureTransfer => info::Trampoline::FutureTransfer,
             Trampoline::StreamTransfer => info::Trampoline::StreamTransfer,
             Trampoline::ErrorContextTransfer => info::Trampoline::ErrorContextTransfer,
+            Trampoline::CheckBlocking => info::Trampoline::CheckBlocking,
             Trampoline::ContextGet { instance, slot } => info::Trampoline::ContextGet {
                 instance: *instance,
                 slot: *slot,
diff --git a/crates/environ/src/component/info.rs b/crates/environ/src/component/info.rs
index efbacb648af4..b2254ae48df2 100644
--- a/crates/environ/src/component/info.rs
+++ b/crates/environ/src/component/info.rs
@@ -1112,6 +1112,10 @@ pub enum Trampoline {
     /// component does not invalidate the handle in the original component.
     ErrorContextTransfer,
 
+    /// An intrinsic used by FACT-generated modules to check whether an
+    /// async-typed function may be called via a sync lower.
+    CheckBlocking,
+
     /// Intrinsic used to implement the `context.get` component model builtin.
     ///
     /// The payload here represents that this is accessing the Nth slot of local
@@ -1242,6 +1246,7 @@ impl Trampoline {
             FutureTransfer => format!("future-transfer"),
             StreamTransfer => format!("stream-transfer"),
             ErrorContextTransfer => format!("error-context-transfer"),
+            CheckBlocking => format!("check-blocking"),
             ContextGet { .. } => format!("context-get"),
             ContextSet { .. } => format!("context-set"),
             ThreadIndex => format!("thread-index"),
diff --git a/crates/environ/src/component/translate/adapt.rs b/crates/environ/src/component/translate/adapt.rs
index ad3a00f525d7..76a75813de9e 100644
--- a/crates/environ/src/component/translate/adapt.rs
+++ b/crates/environ/src/component/translate/adapt.rs
@@ -345,6 +345,7 @@ fn fact_import_to_core_def(
         fact::Import::ErrorContextTransfer => {
             simple_intrinsic(dfg::Trampoline::ErrorContextTransfer)
         }
+        fact::Import::CheckBlocking => simple_intrinsic(dfg::Trampoline::CheckBlocking),
     }
 }
 
diff --git a/crates/environ/src/fact.rs b/crates/environ/src/fact.rs
index d2a8689f2ab4..3fa31386df20 100644
--- a/crates/environ/src/fact.rs
+++ b/crates/environ/src/fact.rs
@@ -47,6 +47,7 @@ pub static PREPARE_CALL_FIXED_PARAMS: &[ValType] = &[
     ValType::I32,     // caller_instance
     ValType::I32,     // callee_instance
     ValType::I32,     // task_return_type
+    ValType::I32,     // callee_async
     ValType::I32,     // string_encoding
     ValType::I32,     // result_count_or_max_if_async
 ];
@@ -89,6 +90,8 @@ pub struct Module<'a> {
     imported_stream_transfer: Option<FuncIndex>,
     imported_error_context_transfer: Option<FuncIndex>,
 
+    imported_check_blocking: Option<FuncIndex>,
+
     // Current status of index spaces from the imports generated so far.
     imported_funcs: PrimaryMap<FuncIndex, Option<CoreDef>>,
     imported_memories: PrimaryMap<MemoryIndex, CoreDef>,
@@ -259,6 +262,7 @@ impl<'a> Module<'a> {
             imported_future_transfer: None,
             imported_stream_transfer: None,
             imported_error_context_transfer: None,
+            imported_check_blocking: None,
             exports: Vec::new(),
         }
     }
@@ -712,6 +716,17 @@ impl<'a> Module<'a> {
         )
     }
 
+    fn import_check_blocking(&mut self) -> FuncIndex {
+        self.import_simple(
+            "async",
+            "check-blocking",
+            &[],
+            &[],
+            Import::CheckBlocking,
+            |me| &mut me.imported_check_blocking,
+        )
+    }
+
     fn translate_helper(&mut self, helper: Helper) -> FunctionId {
         *self.helper_funcs.entry(helper).or_insert_with(|| {
             // Generate a fresh `Function` with a unique id for what we're about to
@@ -870,6 +885,9 @@ pub enum Import {
     /// An intrinisic used by FACT-generated modules to (partially or entirely) transfer
     /// ownership of an `error-context`.
     ErrorContextTransfer,
+    /// An intrinsic used by FACT-generated modules to check whether an
+    /// async-typed function may be called via a sync lower.
+    CheckBlocking,
 }
 
 impl Options {
diff --git a/crates/environ/src/fact/trampoline.rs b/crates/environ/src/fact/trampoline.rs
index 799e26036168..e334876ef1f4 100644
--- a/crates/environ/src/fact/trampoline.rs
+++ b/crates/environ/src/fact/trampoline.rs
@@ -544,6 +544,11 @@ impl<'a, 'b> Compiler<'a, 'b> {
         self.instruction(I32Const(
             i32::try_from(self.types[adapter.lift.ty].results.as_u32()).unwrap(),
         ));
+        self.instruction(I32Const(if self.types[adapter.lift.ty].async_ {
+            1
+        } else {
+            0
+        }));
         self.instruction(I32Const(i32::from(
             adapter.lift.options.string_encoding as u8,
         )));
@@ -748,6 +753,11 @@ impl<'a, 'b> Compiler<'a, 'b> {
             );
         }
 
+        if self.types[adapter.lift.ty].async_ {
+            let check_blocking = self.module.import_check_blocking();
+            self.instruction(Call(check_blocking.as_u32()));
+        }
+
         if self.emit_resource_call {
             let enter = self.module.import_resource_enter_call();
             self.instruction(Call(enter.as_u32()));
diff --git a/crates/environ/src/trap_encoding.rs b/crates/environ/src/trap_encoding.rs
index 15efa99b40fa..e81682e08f2c 100644
--- a/crates/environ/src/trap_encoding.rs
+++ b/crates/environ/src/trap_encoding.rs
@@ -112,6 +112,9 @@ pub enum Trap {
     /// scenario where a component instance tried to call an import or intrinsic
     /// when it wasn't allowed to, e.g. from a post-return function.
     CannotLeaveComponent,
+
+    /// A synchronous task attempted to make a potentially blocking call.
+    CannotBlockSyncTask,
     // if adding a variant here be sure to update the `check!` macro below
 }
 
@@ -154,6 +157,7 @@ impl Trap {
             DisabledOpcode
             AsyncDeadlock
             CannotLeaveComponent
+            CannotBlockSyncTask
         }
 
         None
@@ -190,6 +194,7 @@ impl fmt::Display for Trap {
             DisabledOpcode => "pulley opcode disabled at compile time was executed",
             AsyncDeadlock => "deadlock detected: event loop cannot make further progress",
             CannotLeaveComponent => "cannot leave component instance",
+            CannotBlockSyncTask => "cannot block a synchronous task before returning",
         };
         write!(f, "wasm trap: {desc}")
     }
diff --git a/crates/misc/component-async-tests/wit/test.wit b/crates/misc/component-async-tests/wit/test.wit
index 0a0bad8e684e..af19a6dd57c9 100644
--- a/crates/misc/component-async-tests/wit/test.wit
+++ b/crates/misc/component-async-tests/wit/test.wit
@@ -122,7 +122,7 @@ interface resource-stream {
     foo: func();
   }
 
-  foo: func(count: u32) -> stream<x>;
+  foo: async func(count: u32) -> stream<x>;
 }
 
 interface closed {
@@ -157,7 +157,7 @@ interface cancel {
     leak-task-after-cancel,
   }
 
-  run: func(mode: mode, cancel-delay-millis: u64);
+  run: async func(mode: mode, cancel-delay-millis: u64);
 }
 
 interface intertask {
diff --git a/crates/test-programs/src/bin/async_read_resource_stream.rs b/crates/test-programs/src/bin/async_read_resource_stream.rs
index 089f634c47ad..75ca85cbd822 100644
--- a/crates/test-programs/src/bin/async_read_resource_stream.rs
+++ b/crates/test-programs/src/bin/async_read_resource_stream.rs
@@ -15,7 +15,7 @@ struct Component;
 impl Guest for Component {
     async fn run() {
         let mut count = 7;
-        let mut stream = resource_stream::foo(count);
+        let mut stream = resource_stream::foo(count).await;
 
         while let Some(x) = stream.next().await {
             if count > 0 {
diff --git a/crates/test-util/src/component_fuzz.rs b/crates/test-util/src/component_fuzz.rs
index df2457ae6982..162b7075fbfe 100644
--- a/crates/test-util/src/component_fuzz.rs
+++ b/crates/test-util/src/component_fuzz.rs
@@ -1671,10 +1671,23 @@ impl<'a> TestCase<'a> {
             None
         };
 
+        let mut options = u.arbitrary::<TestCaseOptions>()?;
+
+        // Sync tasks cannot call async functions via a sync lower, nor can they
+        // block in other ways (e.g. by calling `waitable-set.wait`, returning
+        // `CALLBACK_CODE_WAIT`, etc.) prior to returning.  Therefore,
+        // async-ness cascades to the callers:
+        if options.host_async {
+            options.guest_callee_async = true;
+        }
+        if options.guest_callee_async {
+            options.guest_caller_async = true;
+        }
+
         Ok(Self {
             params,
             result,
-            options: u.arbitrary()?,
+            options,
         })
     }
 
diff --git a/crates/test-util/src/wast.rs b/crates/test-util/src/wast.rs
index ff8667644247..d42c3ffc4a1e 100644
--- a/crates/test-util/src/wast.rs
+++ b/crates/test-util/src/wast.rs
@@ -52,6 +52,26 @@ pub fn find_tests(root: &Path) -> Result<Vec<WastTest>> {
         &FindConfig::Infer(component_test_config),
     )
     .with_context(|| format!("failed to add tests from `{}`", cm_tests.display()))?;
+
+    // Temporarily work around upstream tests that loop forever.
+    //
+    // Now that `thread.yield` and `CALLBACK_CODE_YIELD` are both no-ops in
+    // non-blocking contexts, these tests need to be updated; meanwhile, we skip
+    // them.
+    //
+    // TODO: remove this once
+    // https://github.com/WebAssembly/component-model/pull/578 has been merged:
+    {
+        let skip_list = &["drop-subtask.wast", "async-calls-sync.wast"];
+        tests.retain(|test| {
+            test.path
+                .file_name()
+                .and_then(|name| name.to_str())
+                .map(|name| !skip_list.contains(&name))
+                .unwrap_or(true)
+        });
+    }
+
     Ok(tests)
 }
 
@@ -679,6 +699,26 @@ impl WastTest {
             "component-model/test/values/trap-in-post-return.wast",
             "component-model/test/wasmtime/resources.wast",
             "component-model/test/wasm-tools/naming.wast",
+            // TODO: remove these once
+            // https://github.com/WebAssembly/component-model/pull/578 has been
+            // merged:
+            "component-model/test/async/async-calls-sync.wast",
+            "component-model/test/async/backpressure-deadlock.wast",
+            "component-model/test/async/cancel-stream.wast",
+            "component-model/test/async/cancel-subtask.wast",
+            "component-model/test/async/deadlock.wast",
+            "component-model/test/async/drop-subtask.wast",
+            "component-model/test/async/drop-waitable-set.wast",
+            "component-model/test/async/empty-wait.wast",
+            "component-model/test/async/fused.wast",
+            "component-model/test/async/future-read.wast",
+            "component-model/test/async/partial-stream-copies.wast",
+            "component-model/test/async/passing-resources.wast",
+            "component-model/test/async/stackful.wast",
+            "component-model/test/async/trap-if-block-and-sync.wast",
+            "component-model/test/async/trap-if-done.wast",
+            "component-model/test/async/wait-during-callback.wast",
+            "component-model/test/async/zero-length.wast",
         ];
         if failing_component_model_tests
             .iter()
diff --git a/crates/wasmtime/src/runtime/component/concurrent.rs b/crates/wasmtime/src/runtime/component/concurrent.rs
index 2d746ed17181..265ce6900114 100644
--- a/crates/wasmtime/src/runtime/component/concurrent.rs
+++ b/crates/wasmtime/src/runtime/component/concurrent.rs
@@ -574,6 +574,7 @@ enum SuspendReason {
     Waiting {
         set: TableId<WaitableSet>,
         thread: QualifiedThreadId,
+        skip_may_block_check: bool,
     },
     /// The fiber has finished handling its most recent work item and is waiting
     /// for another (or to be dropped if it is no longer needed).
@@ -582,7 +583,10 @@ enum SuspendReason {
     /// chance to run.
     Yielding { thread: QualifiedThreadId },
     /// The fiber was explicitly suspended with a call to `thread.suspend` or `thread.switch-to`.
-    ExplicitlySuspending { thread: QualifiedThreadId },
+    ExplicitlySuspending {
+        thread: QualifiedThreadId,
+        skip_may_block_check: bool,
+    },
 }
 
 /// Represents a pending call into guest code for a given guest task.
@@ -600,7 +604,10 @@ enum GuestCallKind {
     },
     /// Indicates that a new guest task call is pending and may be executed
     /// using the specified closure.
-    StartImplicit(Box<dyn FnOnce(&mut dyn VMStore) -> Result<()> + Send + Sync>),
+    ///
+    /// If the closure returns `Ok(Some(call))`, the `call` should be run
+    /// immediately using `handle_guest_call`.
+    StartImplicit(Box<dyn FnOnce(&mut dyn VMStore) -> Result<Option<GuestCall>> + Send + Sync>),
     StartExplicit(Box<dyn FnOnce(&mut dyn VMStore) -> Result<()> + Send + Sync>),
 }
 
@@ -705,6 +712,12 @@ pub(crate) enum WaitResult {
     Completed,
 }
 
+/// Raise a trap if the calling task is synchronous and trying to block prior to
+/// returning a value.
+pub(crate) fn check_blocking(store: &mut dyn VMStore) -> Result<()> {
+    store.concurrent_state_mut().check_blocking()
+}
+
 /// Poll the specified future until it completes on behalf of a guest->host call
 /// using a sync-lowered import.
 ///
@@ -796,6 +809,7 @@ pub(crate) fn poll_and_block<R: Send + Sync + 'static>(
             store.suspend(SuspendReason::Waiting {
                 set,
                 thread: caller,
+                skip_may_block_check: false,
             })?;
         }
     }
@@ -812,53 +826,58 @@ pub(crate) fn poll_and_block<R: Send + Sync + 'static>(
 
 /// Execute the specified guest call.
 fn handle_guest_call(store: &mut dyn VMStore, call: GuestCall) -> Result<()> {
-    match call.kind {
-        GuestCallKind::DeliverEvent { instance, set } => {
-            let (event, waitable) = instance
-                .get_event(store, call.thread.task, set, true)?
-                .unwrap();
-            let state = store.concurrent_state_mut();
-            let task = state.get_mut(call.thread.task)?;
-            let runtime_instance = task.instance;
-            let handle = waitable.map(|(_, v)| v).unwrap_or(0);
+    let mut next = Some(call);
+    while let Some(call) = next.take() {
+        match call.kind {
+            GuestCallKind::DeliverEvent { instance, set } => {
+                let (event, waitable) = instance
+                    .get_event(store, call.thread.task, set, true)?
+                    .unwrap();
+                let state = store.concurrent_state_mut();
+                let task = state.get_mut(call.thread.task)?;
+                let runtime_instance = task.instance;
+                let handle = waitable.map(|(_, v)| v).unwrap_or(0);
 
-            log::trace!(
-                "use callback to deliver event {event:?} to {:?} for {waitable:?}",
-                call.thread,
-            );
+                log::trace!(
+                    "use callback to deliver event {event:?} to {:?} for {waitable:?}",
+                    call.thread,
+                );
 
-            let old_thread = state.guest_thread.replace(call.thread);
-            log::trace!(
-                "GuestCallKind::DeliverEvent: replaced {old_thread:?} with {:?} as current thread",
-                call.thread
-            );
+                let old_thread = state.guest_thread.replace(call.thread);
+                log::trace!(
+                    "GuestCallKind::DeliverEvent: replaced {old_thread:?} with {:?} as current thread",
+                    call.thread
+                );
 
-            store.maybe_push_call_context(call.thread.task)?;
+                store.maybe_push_call_context(call.thread.task)?;
 
-            let state = store.concurrent_state_mut();
-            state.enter_instance(runtime_instance);
+                let state = store.concurrent_state_mut();
+                state.enter_instance(runtime_instance);
 
-            let callback = state.get_mut(call.thread.task)?.callback.take().unwrap();
+                let callback = state.get_mut(call.thread.task)?.callback.take().unwrap();
 
-            let code = callback(store, runtime_instance, event, handle)?;
+                let code = callback(store, runtime_instance, event, handle)?;
 
-            let state = store.concurrent_state_mut();
+                let state = store.concurrent_state_mut();
 
-            state.get_mut(call.thread.task)?.callback = Some(callback);
-            state.exit_instance(runtime_instance)?;
+                state.get_mut(call.thread.task)?.callback = Some(callback);
+                state.exit_instance(runtime_instance)?;
 
-            store.maybe_pop_call_context(call.thread.task)?;
+                store.maybe_pop_call_context(call.thread.task)?;
 
-            instance.handle_callback_code(store, call.thread, runtime_instance, code)?;
+                next = instance.handle_callback_code(store, call.thread, runtime_instance, code)?;
 
-            store.concurrent_state_mut().guest_thread = old_thread;
-            log::trace!("GuestCallKind::DeliverEvent: restored {old_thread:?} as current thread");
-        }
-        GuestCallKind::StartImplicit(fun) => {
-            fun(store)?;
-        }
-        GuestCallKind::StartExplicit(fun) => {
-            fun(store)?;
+                store.concurrent_state_mut().guest_thread = old_thread;
+                log::trace!(
+                    "GuestCallKind::DeliverEvent: restored {old_thread:?} as current thread"
+                );
+            }
+            GuestCallKind::StartImplicit(fun) => {
+                next = fun(store)?;
+            }
+            GuestCallKind::StartExplicit(fun) => {
+                fun(store)?;
+            }
         }
     }
 
@@ -1431,7 +1450,7 @@ impl StoreOpaque {
                 SuspendReason::ExplicitlySuspending { thread, .. } => {
                     state.get_mut(thread.thread)?.state = GuestThreadState::Suspended(fiber);
                 }
-                SuspendReason::Waiting { set, thread } => {
+                SuspendReason::Waiting { set, thread, .. } => {
                     let old = state
                         .get_mut(set)?
                         .waiting
@@ -1471,6 +1490,26 @@ impl StoreOpaque {
             None
         };
 
+        // We should not have reached here unless either there's no current
+        // task, or the current task is permitted to block.  In addition, we
+        // special-case `thread.switch-to` and waiting for a subtask to go from
+        // `starting` to `started`, both of which we consider non-blocking
+        // operations despite requiring a suspend.
+        assert!(
+            matches!(
+                reason,
+                SuspendReason::ExplicitlySuspending {
+                    skip_may_block_check: true,
+                    ..
+                } | SuspendReason::Waiting {
+                    skip_may_block_check: true,
+                    ..
+                }
+            ) || old_guest_thread
+                .map(|thread| self.concurrent_state_mut().may_block(thread.task))
+                .unwrap_or(true)
+        );
+
         let suspend_reason = &mut self.concurrent_state_mut().suspend_reason;
         assert!(suspend_reason.is_none());
         *suspend_reason = Some(reason);
@@ -1526,6 +1565,7 @@ impl StoreOpaque {
         self.suspend(SuspendReason::Waiting {
             set,
             thread: caller,
+            skip_may_block_check: false,
         })?;
         let state = self.concurrent_state_mut();
         waitable.join(state, old_set)
@@ -1583,13 +1623,16 @@ impl Instance {
 
     /// Handle the `CallbackCode` returned from an async-lifted export or its
     /// callback.
+    ///
+    /// If this returns `Ok(Some(call))`, then `call` should be run immediately
+    /// using `handle_guest_call`.
     fn handle_callback_code(
         self,
         store: &mut StoreOpaque,
         guest_thread: QualifiedThreadId,
         runtime_instance: RuntimeComponentInstanceIndex,
         code: u32,
-    ) -> Result<()> {
+    ) -> Result<Option<GuestCall>> {
         let (code, set) = unpack_callback_code(code);
 
         log::trace!("received callback code from {guest_thread:?}: {code} (set: {set})");
@@ -1607,7 +1650,7 @@ impl Instance {
             Ok(TableId::<WaitableSet>::new(set))
         };
 
-        match code {
+        Ok(match code {
             callback_code::EXIT => {
                 log::trace!("implicit thread {guest_thread:?} completed");
                 self.cleanup_thread(store, guest_thread, runtime_instance)?;
@@ -1627,22 +1670,36 @@ impl Instance {
                         task.callback = None;
                     }
                 }
+                None
             }
             callback_code::YIELD => {
-                // Push this thread onto the "low priority" queue so it runs after
-                // any other threads have had a chance to run.
                 let task = state.get_mut(guest_thread.task)?;
                 assert!(task.event.is_none());
                 task.event = Some(Event::None);
-                state.push_low_priority(WorkItem::GuestCall(GuestCall {
+                let call = GuestCall {
                     thread: guest_thread,
                     kind: GuestCallKind::DeliverEvent {
                         instance: self,
                         set: None,
                     },
-                }));
+                };
+                if state.may_block(guest_thread.task) {
+                    // Push this thread onto the "low priority" queue so it runs
+                    // after any other threads have had a chance to run.
+                    state.push_low_priority(WorkItem::GuestCall(call));
+                    None
+                } else {
+                    // Yielding in a non-blocking context is defined as a no-op
+                    // according to the spec, so we must run this thread
+                    // immediately without allowing any others to run.
+                    Some(call)
+                }
             }
             callback_code::WAIT | callback_code::POLL => {
+                // The task may only return `WAIT` or `POLL` if it was created
+                // for a call to an async export).  Otherwise, we'll trap.
+                state.check_blocking_for(guest_thread.task)?;
+
                 let set = get_set(store, set)?;
                 let state = store.concurrent_state_mut();
 
@@ -1690,11 +1747,10 @@ impl Instance {
                         _ => unreachable!(),
                     }
                 }
+                None
             }
             _ => bail!("unsupported callback code: {code}"),
-        }
-
-        Ok(())
+        })
     }
 
     fn cleanup_thread(
@@ -1864,10 +1920,9 @@ impl Instance {
                 // function returns a `i32` result.
                 let code = unsafe { storage[0].assume_init() }.get_i32() as u32;
 
-                self.handle_callback_code(store, guest_thread, callee_instance, code)?;
-
-                Ok(())
-            }) as Box<dyn FnOnce(&mut dyn VMStore) -> Result<()> + Send + Sync>
+                self.handle_callback_code(store, guest_thread, callee_instance, code)
+            })
+                as Box<dyn FnOnce(&mut dyn VMStore) -> Result<Option<GuestCall>> + Send + Sync>
         } else {
             let token = StoreToken::new(store.as_context_mut());
             Box::new(move |store: &mut dyn VMStore| {
@@ -2003,7 +2058,7 @@ impl Instance {
                     }
                 }
 
-                Ok(())
+                Ok(None)
             })
         };
 
@@ -2038,12 +2093,20 @@ impl Instance {
         caller_instance: RuntimeComponentInstanceIndex,
         callee_instance: RuntimeComponentInstanceIndex,
         task_return_type: TypeTupleIndex,
+        callee_async: bool,
         memory: *mut VMMemoryDefinition,
         string_encoding: u8,
         caller_info: CallerInfo,
     ) -> Result<()> {
         self.id().get(store.0).check_may_leave(caller_instance)?;
 
+        if let (CallerInfo::Sync { .. }, true) = (&caller_info, callee_async) {
+            // A task may only call an async-typed function via a sync lower if
+            // it was created by a call to an async export.  Otherwise, we'll
+            // trap.
+            store.0.concurrent_state_mut().check_blocking()?;
+        }
+
         enum ResultInfo {
             Heap { results: u32 },
             Stack { result_count: u32 },
@@ -2182,6 +2245,7 @@ impl Instance {
             },
             None,
             callee_instance,
+            callee_async,
         )?;
 
         let guest_task = state.push(new_task)?;
@@ -2275,6 +2339,7 @@ impl Instance {
         let async_caller = storage.is_none();
         let state = store.0.concurrent_state_mut();
         let guest_thread = state.guest_thread.unwrap();
+        let callee_async = state.get_mut(guest_thread.task)?.async_function;
         let may_enter_after_call = state
             .get_mut(guest_thread.task)?
             .call_post_return_automatically();
@@ -2368,6 +2433,14 @@ impl Instance {
             store.0.suspend(SuspendReason::Waiting {
                 set,
                 thread: caller,
+                // Normally, `StoreOpaque::suspend` would assert it's being
+                // called from a context where blocking is allowed.  However, if
+                // `async_caller` is `true`, we'll only "block" long enough for
+                // the callee to start, i.e. we won't repeat this loop, so we
+                // tell `suspend` it's okay even if we're not allowed to block.
+                // Alternatively, if the callee is not an async function, then
+                // we know it won't block anyway.
+                skip_may_block_check: async_caller || !callee_async,
             })?;
 
             let state = store.0.concurrent_state_mut();
@@ -2850,6 +2923,14 @@ impl Instance {
         payload: u32,
     ) -> Result<u32> {
         self.id().get(store).check_may_leave(caller)?;
+
+        if !self.options(store, options).async_ {
+            // The caller may only call `waitable-set.wait` from an async task
+            // (i.e. a task created via a call to an async export).
+            // Otherwise, we'll trap.
+            store.concurrent_state_mut().check_blocking()?;
+        }
+
         let &CanonicalOptions {
             cancellable,
             instance: caller_instance,
@@ -2879,6 +2960,14 @@ impl Instance {
         payload: u32,
     ) -> Result<u32> {
         self.id().get(store).check_may_leave(caller)?;
+
+        if !self.options(store, options).async_ {
+            // The caller may only call `waitable-set.poll` from an async task
+            // (i.e. a task created via a call to an async export).
+            // Otherwise, we'll trap.
+            store.concurrent_state_mut().check_blocking()?;
+        }
+
         let &CanonicalOptions {
             cancellable,
             instance: caller_instance,
@@ -3057,13 +3146,31 @@ impl Instance {
         yielding: bool,
         to_thread: Option<u32>,
     ) -> Result<WaitResult> {
+        self.id().get(store).check_may_leave(caller)?;
+
+        if to_thread.is_none() {
+            let state = store.concurrent_state_mut();
+            if yielding {
+                // This is a `thread.yield` call
+                if !state.may_block(state.guest_thread.unwrap().task) {
+                    // The spec defines `thread.yield` to be a no-op in a
+                    // non-blocking context, so we return immediately without giving
+                    // any other thread a chance to run.
+                    return Ok(WaitResult::Completed);
+                }
+            } else {
+                // The caller may only call `thread.suspend` from an async task
+                // (i.e. a task created via a call to an async export).
+                // Otherwise, we'll trap.
+                state.check_blocking()?;
+            }
+        }
+
         // There could be a pending cancellation from a previous uncancellable wait
         if cancellable && store.concurrent_state_mut().take_pending_cancellation() {
             return Ok(WaitResult::Cancelled);
         }
 
-        self.id().get(store).check_may_leave(caller)?;
-
         if let Some(thread) = to_thread {
             self.resume_suspended_thread(store, caller, thread, true)?;
         }
@@ -3077,6 +3184,10 @@ impl Instance {
         } else {
             SuspendReason::ExplicitlySuspending {
                 thread: guest_thread,
+                // Tell `StoreOpaque::suspend` it's okay to suspend here since
+                // we're handling a `thread.switch-to` call; otherwise it would
+                // panic if we called it in a non-blocking context.
+                skip_may_block_check: to_thread.is_some(),
             }
         };
 
@@ -3134,6 +3245,7 @@ impl Instance {
                 store.suspend(SuspendReason::Waiting {
                     set,
                     thread: guest_thread,
+                    skip_may_block_check: false,
                 })?;
             }
         }
@@ -3187,6 +3299,14 @@ impl Instance {
         task_id: u32,
     ) -> Result<u32> {
         self.id().get(store).check_may_leave(caller_instance)?;
+
+        if !async_ {
+            // The caller may only sync call `subtask.cancel` from an async task
+            // (i.e. a task created via a call to an async export).  Otherwise,
+            // we'll trap.
+            store.concurrent_state_mut().check_blocking()?;
+        }
+
         let (rep, is_host) =
             self.id().get_mut(store).guest_tables().0[caller_instance].subtask_rep(task_id)?;
         let (waitable, expected_caller_instance) = if is_host {
@@ -3346,6 +3466,7 @@ pub trait VMComponentAsyncStore {
         caller_instance: RuntimeComponentInstanceIndex,
         callee_instance: RuntimeComponentInstanceIndex,
         task_return_type: TypeTupleIndex,
+        callee_async: bool,
         string_encoding: u8,
         result_count: u32,
         storage: *mut ValRaw,
@@ -3505,6 +3626,7 @@ impl<T: 'static> VMComponentAsyncStore for StoreInner<T> {
         caller_instance: RuntimeComponentInstanceIndex,
         callee_instance: RuntimeComponentInstanceIndex,
         task_return_type: TypeTupleIndex,
+        callee_async: bool,
         string_encoding: u8,
         result_count_or_max_if_async: u32,
         storage: *mut ValRaw,
@@ -3523,6 +3645,7 @@ impl<T: 'static> VMComponentAsyncStore for StoreInner<T> {
                 caller_instance,
                 callee_instance,
                 task_return_type,
+                callee_async,
                 memory,
                 string_encoding,
                 match result_count_or_max_if_async {
@@ -4063,6 +4186,9 @@ pub(crate) struct GuestTask {
     /// The state of the host future that represents an async task, which must
     /// be dropped before we can delete the task.
     host_future_state: HostFutureState,
+    /// Indicates whether this task was created for a call to an async-lifted
+    /// export.
+    async_function: bool,
 }
 
 impl GuestTask {
@@ -4103,6 +4229,7 @@ impl GuestTask {
         caller: Caller,
         callback: Option<CallbackFn>,
         component_instance: RuntimeComponentInstanceIndex,
+        async_function: bool,
     ) -> Result<Self> {
         let sync_call_set = state.push(WaitableSet::default())?;
         let host_future_state = match &caller {
@@ -4137,6 +4264,7 @@ impl GuestTask {
             exited: false,
             threads: HashSet::new(),
             host_future_state,
+            async_function,
         })
     }
 
@@ -4750,6 +4878,24 @@ impl ConcurrentState {
             false
         }
     }
+
+    fn check_blocking(&mut self) -> Result<()> {
+        let task = self.guest_thread.unwrap().task;
+        self.check_blocking_for(task)
+    }
+
+    fn check_blocking_for(&mut self, task: TableId<GuestTask>) -> Result<()> {
+        if self.may_block(task) {
+            Ok(())
+        } else {
+            Err(Trap::CannotBlockSyncTask.into())
+        }
+    }
+
+    fn may_block(&mut self, task: TableId<GuestTask>) -> bool {
+        let task = self.get_mut(task).unwrap();
+        task.async_function || task.returned_or_cancelled()
+    }
 }
 
 /// Provide a type hint to compiler about the shape of a parameter lower
@@ -4908,7 +5054,9 @@ pub(crate) fn prepare_call<T, R>(
 
     let instance = handle.instance().id().get(store.0);
     let options = &instance.component().env_component().options[options];
-    let task_return_type = instance.component().types()[ty].results;
+    let ty = &instance.component().types()[ty];
+    let async_function = ty.async_;
+    let task_return_type = ty.results;
     let component_instance = raw_options.instance;
     let callback = options.callback.map(|i| instance.runtime_callback(i));
     let memory = options
@@ -4965,6 +5113,7 @@ pub(crate) fn prepare_call<T, R>(
             ) as CallbackFn
         }),
         component_instance,
+        async_function,
     )?;
     task.function_index = Some(handle.index());
 
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 9d6098696791..a04040f8e649 100644
--- a/crates/wasmtime/src/runtime/component/concurrent/futures_and_streams.rs
+++ b/crates/wasmtime/src/runtime/component/concurrent/futures_and_streams.rs
@@ -3093,6 +3093,13 @@ impl Instance {
         address: u32,
         count: u32,
     ) -> Result<ReturnCode> {
+        if !self.options(store.0, options).async_ {
+            // The caller may only sync call `{stream,future}.write` from an
+            // async task (i.e. a task created via a call to an async export).
+            // Otherwise, we'll trap.
+            store.0.concurrent_state_mut().check_blocking()?;
+        }
+
         let address = usize::try_from(address).unwrap();
         let count = usize::try_from(count).unwrap();
         self.check_bounds(store.0, options, ty, address, count)?;
@@ -3315,6 +3322,13 @@ impl Instance {
         address: u32,
         count: u32,
     ) -> Result<ReturnCode> {
+        if !self.options(store.0, options).async_ {
+            // The caller may only sync call `{stream,future}.read` from an
+            // async task (i.e. a task created via a call to an async export).
+            // Otherwise, we'll trap.
+            store.0.concurrent_state_mut().check_blocking()?;
+        }
+
         let address = usize::try_from(address).unwrap();
         let count = usize::try_from(count).unwrap();
         self.check_bounds(store.0, options, ty, address, count)?;
@@ -3689,6 +3703,13 @@ impl Instance {
         async_: bool,
         writer: u32,
     ) -> Result<ReturnCode> {
+        if !async_ {
+            // The caller may only sync call `{stream,future}.cancel-write` from
+            // an async task (i.e. a task created via a call to an async
+            // export).  Otherwise, we'll trap.
+            store.concurrent_state_mut().check_blocking()?;
+        }
+
         let (rep, state) =
             get_mut_by_index_from(self.id().get_mut(store).table_for_transmit(ty), ty, writer)?;
         let id = TableId::<TransmitHandle>::new(rep);
@@ -3723,6 +3744,13 @@ impl Instance {
         async_: bool,
         reader: u32,
     ) -> Result<ReturnCode> {
+        if !async_ {
+            // The caller may only sync call `{stream,future}.cancel-read` from
+            // an async task (i.e. a task created via a call to an async
+            // export).  Otherwise, we'll trap.
+            store.concurrent_state_mut().check_blocking()?;
+        }
+
         let (rep, state) =
             get_mut_by_index_from(self.id().get_mut(store).table_for_transmit(ty), ty, reader)?;
         let id = TableId::<TransmitHandle>::new(rep);
diff --git a/crates/wasmtime/src/runtime/component/concurrent_disabled.rs b/crates/wasmtime/src/runtime/component/concurrent_disabled.rs
index 5fb313801fe3..ae40f56c0fbc 100644
--- a/crates/wasmtime/src/runtime/component/concurrent_disabled.rs
+++ b/crates/wasmtime/src/runtime/component/concurrent_disabled.rs
@@ -21,6 +21,10 @@ fn should_have_failed_validation<T>(what: &str) -> Result<T> {
     ))
 }
 
+pub(crate) fn check_blocking(_: &mut dyn VMStore) -> Result<()> {
+    Ok(())
+}
+
 pub(crate) fn poll_and_block<R: Send + Sync + 'static>(
     _store: &mut dyn VMStore,
     future: impl Future<Output = Result<R>> + Send + 'static,
diff --git a/crates/wasmtime/src/runtime/component/func/host.rs b/crates/wasmtime/src/runtime/component/func/host.rs
index a573342303fd..8694cc62f2fb 100644
--- a/crates/wasmtime/src/runtime/component/func/host.rs
+++ b/crates/wasmtime/src/runtime/component/func/host.rs
@@ -41,18 +41,37 @@ enum HostResult<T> {
     Future(Pin<Box<dyn Future<Output = Result<T>> + Send>>),
 }
 
+trait FunctionStyle {
+    const ASYNC: bool;
+}
+
+struct SyncStyle;
+
+impl FunctionStyle for SyncStyle {
+    const ASYNC: bool = false;
+}
+
+#[cfg(feature = "component-model-async")]
+struct AsyncStyle;
+
+#[cfg(feature = "component-model-async")]
+impl FunctionStyle for AsyncStyle {
+    const ASYNC: bool = true;
+}
+
 impl HostFunc {
-    fn from_canonical<T, F, P, R>(func: F) -> Arc<HostFunc>
+    fn from_canonical<T, F, P, R, S>(func: F) -> Arc<HostFunc>
     where
         F: Fn(StoreContextMut<'_, T>, P) -> HostResult<R> + Send + Sync + 'static,
         P: ComponentNamedList + Lift + 'static,
         R: ComponentNamedList + Lower + 'static,
         T: 'static,
+        S: FunctionStyle + 'static,
     {
-        let entrypoint = Self::entrypoint::<T, F, P, R>;
+        let entrypoint = Self::entrypoint::<T, F, P, R, S>;
         Arc::new(HostFunc {
             entrypoint,
-            typecheck: Box::new(typecheck::<P, R>),
+            typecheck: Box::new(typecheck::<P, R, S>),
             func: Box::new(func),
         })
     }
@@ -64,7 +83,7 @@ impl HostFunc {
         P: ComponentNamedList + Lift + 'static,
         R: ComponentNamedList + Lower + 'static,
     {
-        Self::from_canonical::<T, _, _, _>(move |store, params| {
+        Self::from_canonical::<T, _, _, _, SyncStyle>(move |store, params| {
             HostResult::Done(func(store, params))
         })
     }
@@ -81,7 +100,7 @@ impl HostFunc {
         R: ComponentNamedList + Lower + 'static,
     {
         let func = Arc::new(func);
-        Self::from_canonical::<T, _, _, _>(move |store, params| {
+        Self::from_canonical::<T, _, _, _, AsyncStyle>(move |store, params| {
             let func = func.clone();
             HostResult::Future(Box::pin(
                 store.wrap_call(move |accessor| func(accessor, params)),
@@ -89,7 +108,7 @@ impl HostFunc {
         })
     }
 
-    extern "C" fn entrypoint<T, F, P, R>(
+    extern "C" fn entrypoint<T, F, P, R, S>(
         cx: NonNull<VMOpaqueContext>,
         data: NonNull<u8>,
         ty: u32,
@@ -102,11 +121,12 @@ impl HostFunc {
         P: ComponentNamedList + Lift,
         R: ComponentNamedList + Lower + 'static,
         T: 'static,
+        S: FunctionStyle,
     {
         let data = SendSyncPtr::new(NonNull::new(data.as_ptr() as *mut F).unwrap());
         unsafe {
             call_host_and_handle_result::<T>(cx, |store, instance| {
-                call_host(
+                call_host::<_, _, _, _, S>(
                     store,
                     instance,
                     TypeFuncIndex::from_u32(ty),
@@ -118,7 +138,7 @@ impl HostFunc {
         }
     }
 
-    fn new_dynamic_canonical<T, F>(func: F) -> Arc<HostFunc>
+    fn new_dynamic_canonical<T, F, S>(func: F) -> Arc<HostFunc>
     where
         F: Fn(
                 StoreContextMut<'_, T>,
@@ -130,13 +150,21 @@ impl HostFunc {
             + Sync
             + 'static,
         T: 'static,
+        S: FunctionStyle,
     {
         Arc::new(HostFunc {
-            entrypoint: dynamic_entrypoint::<T, F>,
-            // This function performs dynamic type checks and subsequently does
-            // not need to perform up-front type checks. Instead everything is
-            // dynamically managed at runtime.
-            typecheck: Box::new(move |_expected_index, _expected_types| Ok(())),
+            entrypoint: dynamic_entrypoint::<T, F, S>,
+            // This function performs dynamic type checks on its parameters and
+            // results and subsequently does not need to perform up-front type
+            // checks. However, we _do_ verify async-ness here.
+            typecheck: Box::new(move |ty, types| {
+                let ty = &types.types[ty];
+                if S::ASYNC != ty.async_ {
+                    bail!("type mismatch with async");
+                }
+
+                Ok(())
+            }),
             func: Box::new(func),
         })
     }
@@ -148,7 +176,7 @@ impl HostFunc {
             + Sync
             + 'static,
     {
-        Self::new_dynamic_canonical::<T, _>(
+        Self::new_dynamic_canonical::<T, _, SyncStyle>(
             move |store, ty, mut params_and_results, result_start| {
                 let (params, results) = params_and_results.split_at_mut(result_start);
                 let result = func(store, ty, params, results).map(move |()| params_and_results);
@@ -172,7 +200,7 @@ impl HostFunc {
             + 'static,
     {
         let func = Arc::new(func);
-        Self::new_dynamic_canonical::<T, _>(
+        Self::new_dynamic_canonical::<T, _, AsyncStyle>(
             move |store, ty, mut params_and_results, result_start| {
                 let func = func.clone();
                 Box::pin(store.wrap_call(move |accessor| {
@@ -199,12 +227,16 @@ impl HostFunc {
     }
 }
 
-fn typecheck<P, R>(ty: TypeFuncIndex, types: &InstanceType<'_>) -> Result<()>
+fn typecheck<P, R, S>(ty: TypeFuncIndex, types: &InstanceType<'_>) -> Result<()>
 where
     P: ComponentNamedList + Lift,
     R: ComponentNamedList + Lower,
+    S: FunctionStyle,
 {
     let ty = &types.types[ty];
+    if S::ASYNC != ty.async_ {
+        bail!("type mismatch with async");
+    }
     P::typecheck(&InterfaceType::Tuple(ty.params), types)
         .context("type mismatch with parameters")?;
     R::typecheck(&InterfaceType::Tuple(ty.results), types).context("type mismatch with results")?;
@@ -225,6 +257,7 @@ where
 /// * `Return` - the result of the host function
 /// * `F` - the `closure` to actually receive the `Params` and return the
 ///   `Return`
+/// * `S` - the expected `FunctionStyle`
 ///
 /// It's expected that `F` will "un-tuple" the arguments to pass to a host
 /// closure.
@@ -232,7 +265,7 @@ where
 /// This function is in general `unsafe` as the validity of all the parameters
 /// must be upheld. Generally that's done by ensuring this is only called from
 /// the select few places it's intended to be called from.
-unsafe fn call_host<T, Params, Return, F>(
+unsafe fn call_host<T, Params, Return, F, S>(
     store: StoreContextMut<'_, T>,
     instance: Instance,
     ty: TypeFuncIndex,
@@ -244,12 +277,13 @@ where
     F: Fn(StoreContextMut<'_, T>, Params) -> HostResult<Return> + Send + Sync + 'static,
     Params: Lift,
     Return: Lower + 'static,
+    S: FunctionStyle,
 {
     let (component, store) = instance.component_and_store_mut(store.0);
     let mut store = StoreContextMut(store);
     let vminstance = instance.id().get(store.0);
     let opts = &vminstance.component().env_component().options[options];
-    let async_ = opts.async_;
+    let async_lower = opts.async_;
     let caller_instance = opts.instance;
     let mut flags = vminstance.instance_flags(caller_instance);
 
@@ -265,7 +299,7 @@ where
     let param_tys = InterfaceType::Tuple(ty.params);
     let result_tys = InterfaceType::Tuple(ty.results);
 
-    if async_ {
+    if async_lower {
         #[cfg(feature = "component-model-async")]
         {
             let mut storage = unsafe { Storage::<'_, Params, u32>::new_async::<Return>(storage) };
@@ -340,6 +374,13 @@ where
             );
         }
     } else {
+        if S::ASYNC {
+            // The caller has synchronously lowered an async function, meaning
+            // the caller can only call it from an async task (i.e. a task
+            // created via a call to an async export).  Otherwise, we'll trap.
+            concurrent::check_blocking(store.0)?;
+        }
+
         let mut storage = unsafe { Storage::<'_, Params, Return>::new_sync(storage) };
         let mut lift = LiftContext::new(store.0.store_opaque_mut(), options, instance);
         lift.enter_call();
@@ -698,7 +739,7 @@ where
     }
 }
 
-unsafe fn call_host_dynamic<T, F>(
+unsafe fn call_host_dynamic<T, F, S>(
     store: StoreContextMut<'_, T>,
     instance: Instance,
     ty: TypeFuncIndex,
@@ -717,12 +758,13 @@ where
         + Sync
         + 'static,
     T: 'static,
+    S: FunctionStyle,
 {
     let (component, store) = instance.component_and_store_mut(store.0);
     let mut store = StoreContextMut(store);
     let vminstance = instance.id().get(store.0);
     let opts = &component.env_component().options[options];
-    let async_ = opts.async_;
+    let async_lower = opts.async_;
     let caller_instance = opts.instance;
     let mut flags = vminstance.instance_flags(caller_instance);
 
@@ -741,7 +783,7 @@ where
     let mut params_and_results = Vec::new();
     let mut lift = &mut LiftContext::new(store.0.store_opaque_mut(), options, instance);
     lift.enter_call();
-    let max_flat = if async_ {
+    let max_flat = if async_lower {
         MAX_FLAT_ASYNC_PARAMS
     } else {
         MAX_FLAT_PARAMS
@@ -763,7 +805,7 @@ where
         params_and_results.push(Val::Bool(false));
     }
 
-    if async_ {
+    if async_lower {
         #[cfg(feature = "component-model-async")]
         {
             let retptr = if result_tys.types.len() == 0 {
@@ -819,6 +861,13 @@ where
             );
         }
     } else {
+        if S::ASYNC {
+            // The caller has synchronously lowered an async function, meaning
+            // the caller can only call it from an async task (i.e. a task
+            // created via a call to an async export).  Otherwise, we'll trap.
+            concurrent::check_blocking(store.0)?;
+        }
+
         let future = closure(store.as_context_mut(), ty, params_and_results, result_start);
         let result_vals = concurrent::poll_and_block(store.0, future, caller_instance)?;
         let result_vals = &result_vals[result_start..];
@@ -913,7 +962,7 @@ pub(crate) fn validate_inbounds_dynamic(
     Ok(ptr)
 }
 
-extern "C" fn dynamic_entrypoint<T, F>(
+extern "C" fn dynamic_entrypoint<T, F, S>(
     cx: NonNull<VMOpaqueContext>,
     data: NonNull<u8>,
     ty: u32,
@@ -932,11 +981,12 @@ where
         + Sync
         + 'static,
     T: 'static,
+    S: FunctionStyle,
 {
     let data = SendSyncPtr::new(NonNull::new(data.as_ptr() as *mut F).unwrap());
     unsafe {
         call_host_and_handle_result(cx, |store, instance| {
-            call_host_dynamic::<T, _>(
+            call_host_dynamic::<T, _, S>(
                 store,
                 instance,
                 TypeFuncIndex::from_u32(ty),
diff --git a/crates/wasmtime/src/runtime/vm/component/libcalls.rs b/crates/wasmtime/src/runtime/vm/component/libcalls.rs
index 5df911dfb6f8..41976d6a6710 100644
--- a/crates/wasmtime/src/runtime/vm/component/libcalls.rs
+++ b/crates/wasmtime/src/runtime/vm/component/libcalls.rs
@@ -864,6 +864,7 @@ unsafe fn prepare_call(
     caller_instance: u32,
     callee_instance: u32,
     task_return_type: u32,
+    callee_async: u32,
     string_encoding: u32,
     result_count_or_max_if_async: u32,
     storage: *mut u8,
@@ -878,6 +879,7 @@ unsafe fn prepare_call(
             RuntimeComponentInstanceIndex::from_u32(caller_instance),
             RuntimeComponentInstanceIndex::from_u32(callee_instance),
             TypeTupleIndex::from_u32(task_return_type),
+            callee_async != 0,
             u8::try_from(string_encoding).unwrap(),
             result_count_or_max_if_async,
             storage.cast::<crate::ValRaw>(),
@@ -977,6 +979,11 @@ fn error_context_transfer(
     instance.error_context_transfer(store, src_idx, src_table, dst_table)
 }
 
+#[cfg(feature = "component-model-async")]
+fn check_blocking(store: &mut dyn VMStore, _instance: Instance) -> Result<()> {
+    crate::component::concurrent::check_blocking(store)
+}
+
 #[cfg(feature = "component-model-async")]
 unsafe impl HostResultHasUnwindSentinel for ResourcePair {
     type Abi = u64;
diff --git a/tests/all/component_model/async.rs b/tests/all/component_model/async.rs
index b1e3bf8d55ab..0bac8c0dca80 100644
--- a/tests/all/component_model/async.rs
+++ b/tests/all/component_model/async.rs
@@ -473,33 +473,33 @@ async fn task_deletion() -> Result<()> {
                     (export "waitable-set.new" (func $waitable-set.new))))
                 (with "libc" (instance $libc))))
 
-        (func (export "explicit-thread-calls-return-stackful") (result u32)
+        (func (export "explicit-thread-calls-return-stackful") async (result u32)
             (canon lift (core func $cm "explicit-thread-calls-return-stackful") async))
-        (func (export "explicit-thread-calls-return-stackless") (result u32)
+        (func (export "explicit-thread-calls-return-stackless") async (result u32)
             (canon lift (core func $cm "explicit-thread-calls-return-stackless") async (callback (func $cm "cb"))))
-        (func (export "explicit-thread-suspends-sync") (result u32)
+        (func (export "explicit-thread-suspends-sync") async (result u32)
             (canon lift (core func $cm "explicit-thread-suspends-sync")))
-        (func (export "explicit-thread-suspends-stackful") (result u32)
+        (func (export "explicit-thread-suspends-stackful") async (result u32)
             (canon lift (core func $cm "explicit-thread-suspends-stackful") async))
-        (func (export "explicit-thread-suspends-stackless") (result u32)
+        (func (export "explicit-thread-suspends-stackless") async (result u32)
             (canon lift (core func $cm "explicit-thread-suspends-stackless") async (callback (func $cm "cb"))))
-        (func (export "explicit-thread-yield-loops-sync") (result u32)
+        (func (export "explicit-thread-yield-loops-sync") async (result u32)
             (canon lift (core func $cm "explicit-thread-yield-loops-sync")))
-        (func (export "explicit-thread-yield-loops-stackful") (result u32)
+        (func (export "explicit-thread-yield-loops-stackful") async (result u32)
             (canon lift (core func $cm "explicit-thread-yield-loops-stackful") async))
-        (func (export "explicit-thread-yield-loops-stackless") (result u32)
+        (func (export "explicit-thread-yield-loops-stackless") async (result u32)
             (canon lift (core func $cm "explicit-thread-yield-loops-stackless") async (callback (func $cm "cb"))))
     )
 
     (component $D
-        (import "explicit-thread-calls-return-stackful" (func $explicit-thread-calls-return-stackful (result u32)))
-        (import "explicit-thread-calls-return-stackless" (func $explicit-thread-calls-return-stackless (result u32)))
-        (import "explicit-thread-suspends-sync" (func $explicit-thread-suspends-sync (result u32)))
-        (import "explicit-thread-suspends-stackful" (func $explicit-thread-suspends-stackful (result u32)))
-        (import "explicit-thread-suspends-stackless" (func $explicit-thread-suspends-stackless (result u32)))
-        (import "explicit-thread-yield-loops-sync" (func $explicit-thread-yield-loops-sync (result u32)))
-        (import "explicit-thread-yield-loops-stackful" (func $explicit-thread-yield-loops-stackful (result u32)))
-        (import "explicit-thread-yield-loops-stackless" (func $explicit-thread-yield-loops-stackless (result u32)))
+        (import "explicit-thread-calls-return-stackful" (func $explicit-thread-calls-return-stackful async (result u32)))
+        (import "explicit-thread-calls-return-stackless" (func $explicit-thread-calls-return-stackless async (result u32)))
+        (import "explicit-thread-suspends-sync" (func $explicit-thread-suspends-sync async (result u32)))
+        (import "explicit-thread-suspends-stackful" (func $explicit-thread-suspends-stackful async (result u32)))
+        (import "explicit-thread-suspends-stackless" (func $explicit-thread-suspends-stackless async (result u32)))
+        (import "explicit-thread-yield-loops-sync" (func $explicit-thread-yield-loops-sync async (result u32)))
+        (import "explicit-thread-yield-loops-stackful" (func $explicit-thread-yield-loops-stackful async (result u32)))
+        (import "explicit-thread-yield-loops-stackless" (func $explicit-thread-yield-loops-stackless async (result u32)))
 
         (core module $Memory (memory (export "mem") 1))
         (core instance $memory (instantiate $Memory))
@@ -620,7 +620,7 @@ async fn task_deletion() -> Result<()> {
             (export "subtask.cancel" (func $subtask.cancel))
             (export "thread.yield" (func $thread.yield))
         ))))
-        (func (export "run") (result u32) (canon lift (core func $dm "run")))
+        (func (export "run") async (result u32) (canon lift (core func $dm "run")))
     )
 
     (instance $c (instantiate $C))
@@ -728,7 +728,7 @@ async fn cancel_host_future() -> Result<()> {
     ))
   ))
 
-  (func (export "run") (param "f" $f)
+  (func (export "run") async (param "f" $f)
     (canon lift
       (core func $i "run")
       (memory $libc "memory")
diff --git a/tests/all/component_model/bindgen.rs b/tests/all/component_model/bindgen.rs
index 3e8975f7cc38..b49a4f95910b 100644
--- a/tests/all/component_model/bindgen.rs
+++ b/tests/all/component_model/bindgen.rs
@@ -214,7 +214,7 @@ mod one_import_concurrent {
 
                 export bar: async func();
             }
-        ",
+        "
     });
 
     #[tokio::test]
@@ -229,7 +229,7 @@ mod one_import_concurrent {
             r#"
                 (component
                     (import "foo" (instance $foo-instance
-                        (export "foo" (func))
+                        (export "foo" (func async))
                     ))
                     (core module $libc
                         (memory (export "memory") 1)
@@ -255,7 +255,7 @@ mod one_import_concurrent {
                         ))
                     ))
 
-                    (func $f (export "bar")
+                    (func $f (export "bar") async
                         (canon lift (core func $i "bar") async (callback (func $i "callback")))
                     )
 
diff --git a/tests/all/component_model/import.rs b/tests/all/component_model/import.rs
index d5cb4502fe98..6a65dc3a6312 100644
--- a/tests/all/component_model/import.rs
+++ b/tests/all/component_model/import.rs
@@ -492,7 +492,7 @@ async fn stack_and_heap_args_and_rets_concurrent() -> Result<()> {
 }
 
 async fn test_stack_and_heap_args_and_rets(concurrent: bool) -> Result<()> {
-    let (body, async_lower_opts, async_lift_opts) = if concurrent {
+    let (body, async_lower_opts, async_lift_opts, async_type) = if concurrent {
         (
             r#"
     (import "host" "f1" (func $f1 (param i32 i32) (result i32)))
@@ -549,6 +549,7 @@ async fn test_stack_and_heap_args_and_rets(concurrent: bool) -> Result<()> {
             "#,
             "async",
             r#"async (callback (func $m "callback"))"#,
+            "async",
         )
     } else {
         (
@@ -594,6 +595,7 @@ async fn test_stack_and_heap_args_and_rets(concurrent: bool) -> Result<()> {
             "#,
             "",
             "",
+            "",
         )
     };
 
@@ -604,10 +606,10 @@ async fn test_stack_and_heap_args_and_rets(concurrent: bool) -> Result<()> {
                       string string string string
                       string string string string
                       string))
-  (import "f1" (func $f1 (param "a" u32) (result u32)))
-  (import "f2" (func $f2 (param "a" $many_params) (result u32)))
-  (import "f3" (func $f3 (param "a" u32) (result string)))
-  (import "f4" (func $f4 (param "a" $many_params) (result string)))
+  (import "f1" (func $f1 {async_type} (param "a" u32) (result u32)))
+  (import "f2" (func $f2 {async_type} (param "a" $many_params) (result u32)))
+  (import "f3" (func $f3 {async_type} (param "a" u32) (result string)))
+  (import "f4" (func $f4 {async_type} (param "a" $many_params) (result string)))
 
   (core module $libc
     {REALLOC_AND_FREE}
@@ -710,7 +712,7 @@ async fn test_stack_and_heap_args_and_rets(concurrent: bool) -> Result<()> {
     ))
   ))
 
-  (func (export "run")
+  (func (export "run") {async_type}
     (canon lift (core func $m "run") {async_lift_opts})
   )
 )
diff --git a/tests/all/component_model/resources.rs b/tests/all/component_model/resources.rs
index 1df11aba2b09..68f339576100 100644
--- a/tests/all/component_model/resources.rs
+++ b/tests/all/component_model/resources.rs
@@ -1323,7 +1323,7 @@ async fn drop_on_owned_resource() -> Result<()> {
             (component
                 (import "t" (type $t (sub resource)))
                 (import "[constructor]t" (func $ctor (result (own $t))))
-                (import "[method]t.foo" (func $foo (param "self" (borrow $t))))
+                (import "[method]t.foo" (func $foo async (param "self" (borrow $t))))
 
                 (core func $ctor (canon lower (func $ctor)))
                 (core func $drop (canon resource.drop $t))
@@ -1352,7 +1352,7 @@ async fn drop_on_owned_resource() -> Result<()> {
                         (export "drop" (func $drop))
                     ))
                 ))
-                (func (export "f") (canon lift (core func $i "f")))
+                (func (export "f") async (canon lift (core func $i "f")))
             )
         "#,
     )?;
@@ -1364,11 +1364,9 @@ async fn drop_on_owned_resource() -> Result<()> {
     linker
         .root()
         .resource("t", ResourceType::host::<MyType>(), |_, _| Ok(()))?;
-    linker
-        .root()
-        .func_wrap_concurrent("[constructor]t", |_cx, ()| {
-            Box::pin(async { Ok((Resource::<MyType>::new_own(300),)) })
-        })?;
+    linker.root().func_wrap("[constructor]t", |_, ()| {
+        Ok((Resource::<MyType>::new_own(300),))
+    })?;
     linker
         .root()
         .func_wrap_concurrent("[method]t.foo", |_cx, (r,): (Resource<MyType>,)| {
diff --git a/tests/all/pulley_provenance_test_async_component.wat b/tests/all/pulley_provenance_test_async_component.wat
index 75f2a1c6ac2e..b5f04c7ec7bb 100644
--- a/tests/all/pulley_provenance_test_async_component.wat
+++ b/tests/all/pulley_provenance_test_async_component.wat
@@ -1,9 +1,9 @@
 (component
-  (import "sleep" (func $sleep))
+  (import "sleep" (func $sleep async))
 
   (component $A
-    (import "run-stackless" (func $run_stackless))
-    (import "run-stackful" (func $run_stackful))
+    (import "run-stackless" (func $run_stackless async))
+    (import "run-stackful" (func $run_stackful async))
     (core module $libc (memory (export "memory") 1))
     (core instance $libc (instantiate $libc))
 
@@ -91,8 +91,8 @@
         (export "run-stackless" (func $run_stackless))
         (export "run-stackful" (func $run_stackful))))))
 
-    (func (export "run-stackless") (canon lift (core func $i "run-stackless") async (callback (func $i "cb"))))
-    (func (export "run-stackful") (canon lift (core func $i "run-stackful") async)))
+    (func (export "run-stackless") async (canon lift (core func $i "run-stackless") async (callback (func $i "cb"))))
+    (func (export "run-stackful") async (canon lift (core func $i "run-stackful") async)))
 
   (instance $a (instantiate $A
     (with "run-stackless" (func $sleep))
diff --git a/tests/misc_testsuite/component-model-threading/many-threads-indexed.wast b/tests/misc_testsuite/component-model-threading/many-threads-indexed.wast
index be37c06e4bba..62cb8a9f3f94 100644
--- a/tests/misc_testsuite/component-model-threading/many-threads-indexed.wast
+++ b/tests/misc_testsuite/component-model-threading/many-threads-indexed.wast
@@ -121,6 +121,6 @@
             (with "libc" (instance $libc))))
 
     ;; Export the main entry point
-    (func (export "run") (result u32) (canon lift (core func $i "run"))))
+    (func (export "run") async (result u32) (canon lift (core func $i "run"))))
 
 (assert_return (invoke "run") (u32.const 42))
diff --git a/tests/misc_testsuite/component-model-threading/stackful-cancellation.wast b/tests/misc_testsuite/component-model-threading/stackful-cancellation.wast
index e1ca4ab55b6b..5e87a86a351f 100644
--- a/tests/misc_testsuite/component-model-threading/stackful-cancellation.wast
+++ b/tests/misc_testsuite/component-model-threading/stackful-cancellation.wast
@@ -222,18 +222,18 @@
                     (export "waitable-set.new" (func $waitable-set.new))))
                 (with "libc" (instance $libc))))
 
-        (func (export "run-yield") (result u32) (canon lift (core func $cm "run-yield") async))
-        (func (export "run-yield-to") (param "fut" $FT) (result u32) (canon lift (core func $cm "run-yield-to") async))
-        (func (export "run-suspend") (param "fut" $FT) (result u32) (canon lift (core func $cm "run-suspend") async))
-        (func (export "run-switch-to") (param "fut" $FT) (result u32) (canon lift (core func $cm "run-switch-to") async))
+        (func (export "run-yield") async (result u32) (canon lift (core func $cm "run-yield") async))
+        (func (export "run-yield-to") async (param "fut" $FT) (result u32) (canon lift (core func $cm "run-yield-to") async))
+        (func (export "run-suspend") async (param "fut" $FT) (result u32) (canon lift (core func $cm "run-suspend") async))
+        (func (export "run-switch-to") async (param "fut" $FT) (result u32) (canon lift (core func $cm "run-switch-to") async))
     )
 
     (component $D
         (type $FT (future))
-        (import "run-yield" (func $run-yield (result u32)))
-        (import "run-yield-to" (func $run-yield-to (param "fut" $FT) (result u32)))
-        (import "run-suspend" (func $run-suspend (param "fut" $FT) (result u32)))
-        (import "run-switch-to" (func $run-switch-to (param "fut" $FT) (result u32)))
+        (import "run-yield" (func $run-yield async (result u32)))
+        (import "run-yield-to" (func $run-yield-to async (param "fut" $FT) (result u32)))
+        (import "run-suspend" (func $run-suspend async (param "fut" $FT) (result u32)))
+        (import "run-switch-to" (func $run-switch-to async (param "fut" $FT) (result u32)))
 
         (core module $Memory (memory (export "mem") 1))
         (core instance $memory (instantiate $Memory))
@@ -369,7 +369,7 @@
             (export "future.write" (func $future.write))
             (export "thread.yield" (func $thread.yield))
         ))))
-        (func (export "run") (result u32) (canon lift (core func $dm "run")))
+        (func (export "run") async (result u32) (canon lift (core func $dm "run")))
     )
 
     (instance $c (instantiate $C))
diff --git a/tests/misc_testsuite/component-model-threading/threading-builtins.wast b/tests/misc_testsuite/component-model-threading/threading-builtins.wast
index 2ef8104eeec0..5a09d41693bf 100644
--- a/tests/misc_testsuite/component-model-threading/threading-builtins.wast
+++ b/tests/misc_testsuite/component-model-threading/threading-builtins.wast
@@ -97,6 +97,6 @@
             (with "libc" (instance $libc))))
 
     ;; Export the main entry point
-    (func (export "run") (result u32) (canon lift (core func $i "run"))))
+    (func (export "run") async (result u32) (canon lift (core func $i "run"))))
 
 (assert_return (invoke "run") (u32.const 42))
diff --git a/tests/misc_testsuite/component-model/async/backpressure-deadlock.wast b/tests/misc_testsuite/component-model/async/backpressure-deadlock.wast
index 8b497dc3d4d9..5a45f89302a6 100644
--- a/tests/misc_testsuite/component-model/async/backpressure-deadlock.wast
+++ b/tests/misc_testsuite/component-model/async/backpressure-deadlock.wast
@@ -91,7 +91,7 @@
     ))
   ))
 
-  (func (export "f") (canon lift (core func $i "f")))
+  (func (export "f") async (canon lift (core func $i "f")))
 )
 
 (assert_trap (invoke "f") "deadlock detected")
diff --git a/tests/misc_testsuite/component-model/async/future-cancel-read-dropped.wast b/tests/misc_testsuite/component-model/async/future-cancel-read-dropped.wast
index 31c1fe73816c..c74742c175e6 100644
--- a/tests/misc_testsuite/component-model/async/future-cancel-read-dropped.wast
+++ b/tests/misc_testsuite/component-model/async/future-cancel-read-dropped.wast
@@ -58,7 +58,7 @@
     ))
   ))
 
-  (func (export "f") (result u32) (canon lift (core func $i "f")))
+  (func (export "f") async (result u32) (canon lift (core func $i "f")))
 )
 
 ;; Note that there's no way for `future.read` to return DROPPED since the write
diff --git a/tests/misc_testsuite/component-model/async/future-cancel-write-completed.wast b/tests/misc_testsuite/component-model/async/future-cancel-write-completed.wast
index ca19f1c694b5..82a689baaf70 100644
--- a/tests/misc_testsuite/component-model/async/future-cancel-write-completed.wast
+++ b/tests/misc_testsuite/component-model/async/future-cancel-write-completed.wast
@@ -90,7 +90,7 @@
     ))
   ))
 
-  (func (export "f") (result u32) (canon lift (core func $i "f")))
+  (func (export "f") async (result u32) (canon lift (core func $i "f")))
 )
 
 (assert_return (invoke "f") (u32.const 0))
diff --git a/tests/misc_testsuite/component-model/async/future-cancel-write-dropped.wast b/tests/misc_testsuite/component-model/async/future-cancel-write-dropped.wast
index c331181a2502..f0a7a312bfdf 100644
--- a/tests/misc_testsuite/component-model/async/future-cancel-write-dropped.wast
+++ b/tests/misc_testsuite/component-model/async/future-cancel-write-dropped.wast
@@ -51,7 +51,7 @@
     ))
   ))
 
-  (func (export "f") (result u32) (canon lift (core func $i "f")))
+  (func (export "f") async (result u32) (canon lift (core func $i "f")))
 )
 
 (assert_return (invoke "f") (u32.const 1)) ;; expect DROPPED status (not CANCELLED)
diff --git a/tests/misc_testsuite/component-model/async/future-read.wast b/tests/misc_testsuite/component-model/async/future-read.wast
index db0529dcc138..934f909590e4 100644
--- a/tests/misc_testsuite/component-model/async/future-read.wast
+++ b/tests/misc_testsuite/component-model/async/future-read.wast
@@ -49,7 +49,7 @@
     ))
   ))
 
-  (func (export "run")
+  (func (export "run") async
     (canon lift (core func $i "run")))
 )
 
@@ -134,7 +134,7 @@
         (export "read" (func $read))
       ))
     ))
-    (func (export "run") (param "x" $future)
+    (func (export "run") async (param "x" $future)
       (canon lift (core func $i "run") async (callback (func $i "cb"))))
   )
   (instance $child (instantiate $child))
@@ -158,7 +158,7 @@
     ))
   ))
 
-  (func (export "run")
+  (func (export "run") async
     (canon lift (core func $i "run")))
 )
 
@@ -219,7 +219,7 @@
     ))
   ))
 
-  (func (export "run")
+  (func (export "run") async
     (canon lift (core func $i "run")))
 )
 
diff --git a/tests/misc_testsuite/component-model/async/partial-stream-copies.wast b/tests/misc_testsuite/component-model/async/partial-stream-copies.wast
index 314a614e0a62..e8a2caadd778 100644
--- a/tests/misc_testsuite/component-model/async/partial-stream-copies.wast
+++ b/tests/misc_testsuite/component-model/async/partial-stream-copies.wast
@@ -234,7 +234,7 @@
       (export "stream.drop-writable" (func $stream.drop-writable))
       (export "transform" (func $transform'))
     ))))
-    (func (export "run") (result u32) (canon lift (core func $dm "run")))
+    (func (export "run") async (result u32) (canon lift (core func $dm "run")))
   )
 
   (instance $c (instantiate $C))
diff --git a/tests/misc_testsuite/component-model/async/subtask-wait.wast b/tests/misc_testsuite/component-model/async/subtask-wait.wast
index d2c64da15d0a..b16c474ea604 100644
--- a/tests/misc_testsuite/component-model/async/subtask-wait.wast
+++ b/tests/misc_testsuite/component-model/async/subtask-wait.wast
@@ -17,11 +17,11 @@
     )
 
     (core instance $a (instantiate $a))
-    (func (export "run")
+    (func (export "run") async
       (canon lift (core func $a "run") async (callback (func $a "run-cb"))))
   )
   (component $B
-    (import "a" (instance $a (export "run" (func))))
+    (import "a" (instance $a (export "run" (func async))))
 
     (core module $libc (memory (export "memory") 1))
     (core instance $libc (instantiate $libc))
@@ -71,7 +71,7 @@
         (export "wait" (func $wait))
       ))
     ))
-    (func (export "run")
+    (func (export "run") async
       (canon lift (core func $b "run")))
   )
 
diff --git a/tests/misc_testsuite/component-model/async/trap-if-block-and-sync.wast b/tests/misc_testsuite/component-model/async/trap-if-block-and-sync.wast
new file mode 100644
index 000000000000..9abf5b26ea2d
--- /dev/null
+++ b/tests/misc_testsuite/component-model/async/trap-if-block-and-sync.wast
@@ -0,0 +1,377 @@
+;;! component_model_async = true
+;;! component_model_threading = true
+;;! reference_types = true
+;;! gc_types = true
+;;! multi_memory = true
+
+;; TODO: remove this in favor of the upstream version once
+;; https://github.com/WebAssembly/component-model/pull/578 has been merged.
+
+;; The $Tester component has two nested components $C and $D, where $D imports
+;; and calls $C. $C contains utilities used by $D to perform all the tests.
+;; Most of the tests trap, $Tester exports 1 function per test and a fresh
+;; $Tester is created to run each test.
+(component definition $Tester
+  (component $C
+    (core module $Memory (memory (export "mem") 1))
+    (core instance $memory (instantiate $Memory))
+    (core module $CM
+      (import "" "mem" (memory 1))
+      (import "" "waitable.join" (func $waitable.join (param i32 i32)))
+      (import "" "waitable-set.new" (func $waitable-set.new (result i32)))
+      (import "" "waitable-set.wait" (func $waitable-set.wait (param i32 i32) (result i32)))
+      (import "" "future.new" (func $future.new (result i64)))
+      (import "" "future.read" (func $future.read (param i32 i32) (result i32)))
+      (import "" "future.write" (func $future.write (param i32 i32) (result i32)))
+
+      ;; $ws is waited on by 'blocker' and added to by 'unblocker'
+      (global $ws (mut i32) (i32.const 0))
+      (func $start (global.set $ws (call $waitable-set.new)))
+      (start $start)
+
+      (func (export "blocker") (result i32)
+        ;; wait on $ws, which is initially empty, but will be populated with
+        ;; a completed future when "unblocker" synchronously barges in.
+        (local $ret i32)
+        (local.set $ret (call $waitable-set.wait (global.get $ws) (i32.const 0)))
+        (if (i32.ne (i32.const 4 (; FUTURE_READ ;)) (local.get $ret))
+          (then unreachable))
+        (if (i32.ne (i32.const 0 (; COMPLETED ;)) (i32.load (i32.const 4)))
+          (then unreachable))
+
+        (i32.const 42)
+      )
+
+      (func (export "unblocker") (result i32)
+        (local $ret i32) (local $ret64 i64)
+        (local $futr i32) (local $futw i32)
+
+        ;; create read/write futures that will be used to unblock 'blocker'
+        (local.set $ret64 (call $future.new))
+        (local.set $futr (i32.wrap_i64 (local.get $ret64)))
+        (local.set $futw (i32.wrap_i64 (i64.shr_u (local.get $ret64) (i64.const 32))))
+
+        ;; perform a future.read which will block, and add this future to the waitable-set
+        ;; being waited on by 'blocker'
+        (local.set $ret (call $future.read (local.get $futr) (i32.const 0xdeadbeef)))
+        (if (i32.ne (i32.const -1 (; BLOCKED ;)) (local.get $ret))
+          (then unreachable))
+        (call $waitable.join (local.get $futr) (global.get $ws))
+
+        ;; perform a future.write which will rendezvous with the write and complete
+        (local.set $ret (call $future.write (local.get $futw) (i32.const 0xdeadbeef)))
+        (if (i32.ne (i32.const 0 (; COMPLETED ;)) (local.get $ret))
+          (then unreachable))
+
+        (i32.const 43)
+      )
+
+      (func (export "sync-async-func")
+        unreachable
+      )
+      (func (export "async-async-func") (result i32)
+        unreachable
+      )
+      (func (export "async-async-func-cb") (param i32 i32 i32) (result i32)
+        unreachable
+      )
+    )
+    (type $FT (future))
+    (canon waitable.join (core func $waitable.join))
+    (canon waitable-set.new (core func $waitable-set.new))
+    (canon waitable-set.wait (memory $memory "mem") (core func $waitable-set.wait))
+    (canon future.new $FT (core func $future.new))
+    (canon future.read $FT async (core func $future.read))
+    (canon future.write $FT async (core func $future.write))
+    (core instance $cm (instantiate $CM (with "" (instance
+      (export "mem" (memory $memory "mem"))
+      (export "waitable.join" (func $waitable.join))
+      (export "waitable-set.new" (func $waitable-set.new))
+      (export "waitable-set.wait" (func $waitable-set.wait))
+      (export "future.new" (func $future.new))
+      (export "future.read" (func $future.read))
+      (export "future.write" (func $future.write))
+    ))))
+    (func (export "blocker") async (result u32) (canon lift (core func $cm "blocker")))
+    (func (export "unblocker") (result u32) (canon lift (core func $cm "unblocker")))
+    (func (export "sync-async-func") async (canon lift (core func $cm "sync-async-func")))
+    (func (export "async-async-func") async (canon lift (core func $cm "async-async-func") async (callback (func $cm "async-async-func-cb"))))
+  )
+  (component $D
+    (import "c" (instance $c
+      (export "blocker" (func async (result u32)))
+      (export "unblocker" (func (result u32)))
+      (export "sync-async-func" (func async))
+      (export "async-async-func" (func async))
+    ))
+
+    (core module $Memory (memory (export "mem") 1))
+    (core instance $memory (instantiate $Memory))
+    (core module $Core
+      (import "" "mem" (memory 1))
+      (import "" "task.return" (func $task.return (param i32)))
+      (import "" "subtask.cancel" (func $subtask.cancel (param i32) (result i32)))
+      (import "" "thread.yield" (func $thread.yield (result i32)))
+      (import "" "thread.suspend" (func $thread.suspend (result i32)))
+      (import "" "waitable.join" (func $waitable.join (param i32 i32)))
+      (import "" "waitable-set.new" (func $waitable-set.new (result i32)))
+      (import "" "waitable-set.wait" (func $waitable-set.wait (param i32 i32) (result i32)))
+      (import "" "waitable-set.poll" (func $waitable-set.poll (param i32 i32) (result i32)))
+      (import "" "stream.read" (func $stream.read (param i32 i32 i32) (result i32)))
+      (import "" "stream.write" (func $stream.write (param i32 i32 i32) (result i32)))
+      (import "" "future.read" (func $future.read (param i32 i32) (result i32)))
+      (import "" "future.write" (func $future.write (param i32 i32) (result i32)))
+      (import "" "stream.cancel-read" (func $stream.cancel-read (param i32) (result i32)))
+      (import "" "stream.cancel-write" (func $stream.cancel-write (param i32) (result i32)))
+      (import "" "future.cancel-read" (func $future.cancel-read (param i32) (result i32)))
+      (import "" "future.cancel-write" (func $future.cancel-write (param i32) (result i32)))
+      (import "" "blocker" (func $blocker (param i32) (result i32)))
+      (import "" "unblocker" (func $unblocker (result i32)))
+      (import "" "await-sync-async-func" (func $await-sync-async-func))
+      (import "" "await-async-async-func" (func $await-async-async-func))
+
+      (func (export "sync-barges-in") (result i32)
+        (local $ret i32) (local $retp1 i32) (local $retp2 i32)
+        (local $subtask i32) (local $ws i32) (local $event_code i32)
+
+        (local.set $retp1 (i32.const 8))
+        (local.set $retp2 (i32.const 12))
+
+        ;; call $blocker which will block during a synchronous function.
+        (local.set $ret (call $blocker (local.get $retp1)))
+        (if (i32.ne (i32.const 1 (; STARTED ;)) (i32.and (local.get $ret) (i32.const 0xf)))
+          (then unreachable))
+        (local.set $subtask (i32.shr_u (local.get $ret) (i32.const 4)))
+
+        ;; normally calling another function would hit backpressure until
+        ;; $blocker was done, but calling the sync-typed function $unblocker
+        ;; barges in synchronously.
+        (local.set $ret (call $unblocker))
+        (if (i32.ne (local.get $ret) (i32.const 43))
+          (then unreachable))
+
+        ;; now wait to confirm that $subtask was actually unblocked
+        (local.set $ws (call $waitable-set.new))
+        (call $waitable.join (local.get $subtask) (local.get $ws))
+        (local.set $event_code (call $waitable-set.wait (local.get $ws) (local.get $retp2)))
+        (if (i32.ne (i32.const 1 (; SUBTASK ;)) (local.get $event_code))
+          (then unreachable))
+        (if (i32.ne (local.get $subtask) (i32.load (local.get $retp2)))
+          (then unreachable))
+        (if (i32.ne (i32.const 2 (; RETURNED=2 ;)) (i32.load offset=4 (local.get $retp2)))
+          (then unreachable))
+        (if (i32.ne (i32.const 42) (i32.load (local.get $retp1)))
+          (then unreachable))
+
+        (i32.const 44)
+      )
+
+      (func (export "unreachable-cb") (param i32 i32 i32) (result i32)
+        unreachable
+      )
+      (func (export "return-42-cb") (param i32 i32 i32) (result i32)
+        (call $task.return (i32.const 42))
+        (i32.const 0 (; EXIT ;))
+      )
+
+      (func (export "trap-if-sync-call-async1")
+        (call $await-sync-async-func)
+      )
+      (func (export "trap-if-sync-call-async2")
+        (call $await-async-async-func)
+      )
+      (func (export "trap-if-suspend")
+        (call $thread.suspend)
+        unreachable
+      )
+      (func (export "trap-if-wait")
+        (call $waitable-set.wait (call $waitable-set.new) (i32.const 0xdeadbeef))
+        unreachable
+      )
+      (func (export "trap-if-wait-cb") (result i32)
+        (i32.or
+          (i32.const 2 (; WAIT ;))
+          (i32.shl (call $waitable-set.new) (i32.const 4)))
+      )
+      (func (export "trap-if-poll")
+        (call $waitable-set.poll (call $waitable-set.new) (i32.const 0xdeadbeef))
+        unreachable
+      )
+      (func (export "trap-if-poll-cb") (result i32)
+        (i32.or
+          (i32.const 3 (; POLL ;))
+          (i32.shl (call $waitable-set.new) (i32.const 4)))
+      )
+      (func (export "yield-is-fine") (result i32)
+        (drop (call $thread.yield))
+        (i32.const 42)
+      )
+      (func (export "yield-is-fine-cb") (result i32)
+        (i32.or
+          (i32.const 1 (; YIELD ;))
+          (i32.shl (i32.const 0xdead) (i32.const 4)))
+      )
+      (func (export "trap-if-sync-cancel")
+        (call $subtask.cancel (i32.const 0xdeadbeef))
+        unreachable
+      )
+      (func (export "trap-if-sync-stream-read")
+        (call $stream.read (i32.const 0xdead) (i32.const 0xbeef) (i32.const 0xdeadbeef))
+        unreachable
+      )
+      (func (export "trap-if-sync-stream-write")
+        (call $stream.write (i32.const 0xdead) (i32.const 0xbeef) (i32.const 0xdeadbeef))
+        unreachable
+      )
+      (func (export "trap-if-sync-future-read")
+        (call $future.read (i32.const 0xdead) (i32.const 0xdeadbeef))
+        unreachable
+      )
+      (func (export "trap-if-sync-future-write")
+        (call $future.write (i32.const 0xdead) (i32.const 0xdeadbeef))
+        unreachable
+      )
+      (func (export "trap-if-sync-stream-cancel-read")
+        (call $stream.cancel-read (i32.const 0xdead))
+        unreachable
+      )
+      (func (export "trap-if-sync-stream-cancel-write")
+        (call $stream.cancel-write (i32.const 0xdead))
+        unreachable
+      )
+      (func (export "trap-if-sync-future-cancel-read")
+        (call $future.cancel-read (i32.const 0xdead) (i32.const 0xdeadbeef))
+        unreachable
+      )
+      (func (export "trap-if-sync-future-cancel-write")
+        (call $future.cancel-write (i32.const 0xdead) (i32.const 0xdeadbeef))
+        unreachable
+      )
+    )
+    (type $FT (future u8))
+    (type $ST (stream u8))
+    (canon task.return (result u32) (core func $task.return))
+    (canon subtask.cancel (core func $subtask.cancel))
+    (canon thread.yield (core func $thread.yield))
+    (canon thread.suspend (core func $thread.suspend))
+    (canon waitable.join (core func $waitable.join))
+    (canon waitable-set.new (core func $waitable-set.new))
+    (canon waitable-set.wait (memory $memory "mem") (core func $waitable-set.wait))
+    (canon waitable-set.poll (memory $memory "mem") (core func $waitable-set.poll))
+    (canon stream.read $ST (memory $memory "mem") (core func $stream.read))
+    (canon stream.write $ST (memory $memory "mem") (core func $stream.write))
+    (canon future.read $FT (memory $memory "mem") (core func $future.read))
+    (canon future.write $FT (memory $memory "mem") (core func $future.write))
+    (canon stream.cancel-read $ST (core func $stream.cancel-read))
+    (canon stream.cancel-write $ST (core func $stream.cancel-write))
+    (canon future.cancel-read $FT (core func $future.cancel-read))
+    (canon future.cancel-write $FT (core func $future.cancel-write))
+    (canon lower (func $c "blocker") (memory $memory "mem") async (core func $blocker'))
+    (canon lower (func $c "unblocker") (core func $unblocker'))
+    (canon lower (func $c "sync-async-func") (core func $await-sync-async-func'))
+    (canon lower (func $c "async-async-func") (core func $await-async-async-func'))
+    (core instance $core (instantiate $Core (with "" (instance
+      (export "mem" (memory $memory "mem"))
+      (export "task.return" (func $task.return))
+      (export "subtask.cancel" (func $subtask.cancel))
+      (export "thread.yield" (func $thread.yield))
+      (export "thread.suspend" (func $thread.suspend))
+      (export "waitable.join" (func $waitable.join))
+      (export "waitable-set.new" (func $waitable-set.new))
+      (export "waitable-set.wait" (func $waitable-set.wait))
+      (export "waitable-set.poll" (func $waitable-set.poll))
+      (export "stream.read" (func $stream.read))
+      (export "stream.write" (func $stream.write))
+      (export "future.read" (func $future.read))
+      (export "future.write" (func $future.write))
+      (export "stream.cancel-read" (func $stream.cancel-read))
+      (export "stream.cancel-write" (func $stream.cancel-write))
+      (export "future.cancel-read" (func $future.cancel-read))
+      (export "future.cancel-write" (func $future.cancel-write))
+      (export "blocker" (func $blocker'))
+      (export "unblocker" (func $unblocker'))
+      (export "await-sync-async-func" (func $await-sync-async-func'))
+      (export "await-async-async-func" (func $await-async-async-func'))
+    ))))
+    (func (export "sync-barges-in") async (result u32) (canon lift (core func $core "sync-barges-in")))
+    (func (export "trap-if-suspend") (canon lift (core func $core "trap-if-suspend")))
+    (func (export "trap-if-wait") (canon lift (core func $core "trap-if-wait")))
+    (func (export "trap-if-wait-cb") (canon lift (core func $core "trap-if-wait-cb") async (callback (func $core "unreachable-cb"))))
+    (func (export "trap-if-poll") (canon lift (core func $core "trap-if-poll")))
+    (func (export "trap-if-poll-cb") (canon lift (core func $core "trap-if-poll-cb") async (callback (func $core "unreachable-cb"))))
+    (func (export "yield-is-fine") (result u32) (canon lift (core func $core "yield-is-fine")))
+    (func (export "yield-is-fine-cb") (result u32) (canon lift (core func $core "yield-is-fine-cb") async (callback (func $core "return-42-cb"))))
+    (func (export "trap-if-sync-call-async1") (canon lift (core func $core "trap-if-sync-call-async1")))
+    (func (export "trap-if-sync-call-async2") (canon lift (core func $core "trap-if-sync-call-async2")))
+    (func (export "trap-if-sync-cancel") (canon lift (core func $core "trap-if-sync-cancel")))
+    (func (export "trap-if-sync-stream-read") (canon lift (core func $core "trap-if-sync-stream-read")))
+    (func (export "trap-if-sync-stream-write") (canon lift (core func $core "trap-if-sync-stream-write")))
+    (func (export "trap-if-sync-future-read") (canon lift (core func $core "trap-if-sync-future-read")))
+    (func (export "trap-if-sync-future-write") (canon lift (core func $core "trap-if-sync-future-write")))
+    (func (export "trap-if-sync-stream-cancel-read") (canon lift (core func $core "trap-if-sync-stream-cancel-read")))
+    (func (export "trap-if-sync-stream-cancel-write") (canon lift (core func $core "trap-if-sync-stream-cancel-write")))
+    (func (export "trap-if-sync-future-cancel-read") (canon lift (core func $core "trap-if-sync-future-cancel-read")))
+    (func (export "trap-if-sync-future-cancel-write") (canon lift (core func $core "trap-if-sync-future-cancel-write")))
+  )
+  (instance $c (instantiate $C))
+  (instance $d (instantiate $D (with "c" (instance $c))))
+  (func (export "sync-barges-in") (alias export $d "sync-barges-in"))
+  (func (export "trap-if-sync-call-async1") (alias export $d "trap-if-sync-call-async1"))
+  (func (export "trap-if-sync-call-async2") (alias export $d "trap-if-sync-call-async2"))
+  (func (export "trap-if-suspend") (alias export $d "trap-if-suspend"))
+  (func (export "trap-if-wait") (alias export $d "trap-if-wait"))
+  (func (export "trap-if-wait-cb") (alias export $d "trap-if-wait-cb"))
+  (func (export "trap-if-poll") (alias export $d "trap-if-poll"))
+  (func (export "trap-if-poll-cb") (alias export $d "trap-if-poll-cb"))
+  (func (export "yield-is-fine") (alias export $d "yield-is-fine"))
+  (func (export "yield-is-fine-cb") (alias export $d "yield-is-fine-cb"))
+  (func (export "trap-if-sync-cancel") (alias export $d "trap-if-sync-cancel"))
+  (func (export "trap-if-sync-stream-read") (alias export $d "trap-if-sync-stream-read"))
+  (func (export "trap-if-sync-stream-write") (alias export $d "trap-if-sync-stream-write"))
+  (func (export "trap-if-sync-future-read") (alias export $d "trap-if-sync-future-read"))
+  (func (export "trap-if-sync-future-write") (alias export $d "trap-if-sync-future-write"))
+  (func (export "trap-if-sync-stream-cancel-read") (alias export $d "trap-if-sync-stream-cancel-read"))
+  (func (export "trap-if-sync-stream-cancel-write") (alias export $d "trap-if-sync-stream-cancel-write"))
+  (func (export "trap-if-sync-future-cancel-read") (alias export $d "trap-if-sync-future-cancel-read"))
+  (func (export "trap-if-sync-future-cancel-write") (alias export $d "trap-if-sync-future-cancel-write"))
+)
+
+(component instance $i $Tester)
+(assert_return (invoke "sync-barges-in") (u32.const 44))
+
+(component instance $i $Tester)
+(assert_trap (invoke "trap-if-sync-call-async1") "cannot block a synchronous task before returning")
+(component instance $i $Tester)
+(assert_trap (invoke "trap-if-sync-call-async2") "cannot block a synchronous task before returning")
+(component instance $i $Tester)
+(assert_trap (invoke "trap-if-suspend") "cannot block a synchronous task before returning")
+(component instance $i $Tester)
+(assert_trap (invoke "trap-if-wait") "cannot block a synchronous task before returning")
+(component instance $i $Tester)
+(assert_trap (invoke "trap-if-wait-cb") "cannot block a synchronous task before returning")
+(component instance $i $Tester)
+(assert_trap (invoke "trap-if-poll") "cannot block a synchronous task before returning")
+(component instance $i $Tester)
+(assert_trap (invoke "trap-if-poll-cb") "cannot block a synchronous task before returning")
+(component instance $i $Tester)
+(assert_return (invoke "yield-is-fine") (u32.const 42))
+(component instance $i $Tester)
+(assert_return (invoke "yield-is-fine-cb") (u32.const 42))
+(component instance $i $Tester)
+(assert_trap (invoke "trap-if-sync-cancel") "cannot block a synchronous task before returning")
+(component instance $i $Tester)
+(assert_trap (invoke "trap-if-sync-stream-read") "cannot block a synchronous task before returning")
+(component instance $i $Tester)
+(assert_trap (invoke "trap-if-sync-stream-write") "cannot block a synchronous task before returning")
+(component instance $i $Tester)
+(assert_trap (invoke "trap-if-sync-future-read") "cannot block a synchronous task before returning")
+(component instance $i $Tester)
+(assert_trap (invoke "trap-if-sync-future-write") "cannot block a synchronous task before returning")
+(component instance $i $Tester)
+(assert_trap (invoke "trap-if-sync-stream-cancel-read") "cannot block a synchronous task before returning")
+(component instance $i $Tester)
+(assert_trap (invoke "trap-if-sync-stream-cancel-write") "cannot block a synchronous task before returning")
+(component instance $i $Tester)
+(assert_trap (invoke "trap-if-sync-future-cancel-read") "cannot block a synchronous task before returning")
+(component instance $i $Tester)
+(assert_trap (invoke "trap-if-sync-future-cancel-write") "cannot block a synchronous task before returning")
diff --git a/tests/misc_testsuite/component-model/async/trap-if-done.wast b/tests/misc_testsuite/component-model/async/trap-if-done.wast
index b8c832b052c8..d4220cbc6300 100644
--- a/tests/misc_testsuite/component-model/async/trap-if-done.wast
+++ b/tests/misc_testsuite/component-model/async/trap-if-done.wast
@@ -423,14 +423,14 @@
       (export "stream-drop-writable" (func $stream-drop-writable'))
     ))))
     (func (export "trap-after-future-eager-write") (canon lift (core func $core "trap-after-future-eager-write")))
-    (func (export "trap-after-future-async-write") (canon lift (core func $core "trap-after-future-async-write")))
+    (func (export "trap-after-future-async-write") async (canon lift (core func $core "trap-after-future-async-write")))
     (func (export "trap-after-future-reader-dropped") (canon lift (core func $core "trap-after-future-reader-dropped")))
     (func (export "trap-after-future-eager-read") (param "bool" bool) (result $FT) (canon lift (core func $core "trap-after-future-eager-read")))
-    (func (export "trap-after-future-async-read") (param "bool" bool) (result $FT) (canon lift (core func $core "trap-after-future-async-read")))
+    (func (export "trap-after-future-async-read") async (param "bool" bool) (result $FT) (canon lift (core func $core "trap-after-future-async-read")))
     (func (export "trap-after-stream-reader-eager-dropped") (canon lift (core func $core "trap-after-stream-reader-eager-dropped")))
-    (func (export "trap-after-stream-reader-async-dropped") (canon lift (core func $core "trap-after-stream-reader-async-dropped")))
+    (func (export "trap-after-stream-reader-async-dropped") async (canon lift (core func $core "trap-after-stream-reader-async-dropped")))
     (func (export "trap-after-stream-writer-eager-dropped") (param "bool" bool) (result $ST) (canon lift (core func $core "trap-after-stream-writer-eager-dropped")))
-    (func (export "trap-after-stream-writer-async-dropped") (param "bool" bool) (result $ST) (canon lift (core func $core "trap-after-stream-writer-async-dropped")))
+    (func (export "trap-after-stream-writer-async-dropped") async (param "bool" bool) (result $ST) (canon lift (core func $core "trap-after-stream-writer-async-dropped")))
   )
   (instance $c (instantiate $C))
   (instance $d (instantiate $D (with "c" (instance $c))))
diff --git a/tests/misc_testsuite/component-model/async/wait-forever.wast b/tests/misc_testsuite/component-model/async/wait-forever.wast
index ce3b0e8fcb7e..7f73f2ee995f 100644
--- a/tests/misc_testsuite/component-model/async/wait-forever.wast
+++ b/tests/misc_testsuite/component-model/async/wait-forever.wast
@@ -30,7 +30,7 @@
       ))
     ))
 
-    (func (export "run")
+    (func (export "run") async
       (canon lift (core func $i "run") async (callback (func $i "cb"))))
   )
   (instance $child (instantiate $child))
@@ -49,7 +49,7 @@
     ))
   ))
 
-  (func (export "run")
+  (func (export "run") async
     (canon lift (core func $i "run")))
 )
 
diff --git a/tests/misc_testsuite/component-model/async/wait-forever2.wast b/tests/misc_testsuite/component-model/async/wait-forever2.wast
index e6df132f056d..dc0a3a616b26 100644
--- a/tests/misc_testsuite/component-model/async/wait-forever2.wast
+++ b/tests/misc_testsuite/component-model/async/wait-forever2.wast
@@ -26,14 +26,14 @@
       (export "mem" (memory $memory "mem"))
       (export "waitable-set.new" (func $waitable-set.new))
     ))))
-    (func (export "f") (result u32) (canon lift
+    (func (export "f") async (result u32) (canon lift
       (core func $cm "f")
       async (memory $memory "mem") (callback (func $cm "cb"))
     ))
   )
 
   (component $D
-    (import "f" (func $f (result u32)))
+    (import "f" (func $f async (result u32)))
 
     (core module $Memory (memory (export "mem") 1))
     (core instance $memory (instantiate $Memory))
@@ -65,7 +65,7 @@
       (export "waitable-set.wait" (func $waitable-set.wait))
       (export "f" (func $f'))
     ))))
-    (func (export "f") (result u32) (canon lift (core func $dm "g")))
+    (func (export "f") async (result u32) (canon lift (core func $dm "g")))
   )
 
   (instance $c (instantiate $C))