diff --git a/crates/environ/src/component/dfg.rs b/crates/environ/src/component/dfg.rs
index fad632be02ef..69e456038884 100644
--- a/crates/environ/src/component/dfg.rs
+++ b/crates/environ/src/component/dfg.rs
@@ -267,7 +267,6 @@ pub enum CoreDef {
     InstanceFlags(RuntimeComponentInstanceIndex),
     Trampoline(TrampolineIndex),
     UnsafeIntrinsic(ModuleInternedTypeIndex, UnsafeIntrinsic),
-    TaskMayBlock,
 
     /// This is a special variant not present in `info::CoreDef` which
     /// represents that this definition refers to a fused adapter function. This
@@ -913,7 +912,6 @@ impl LinearizeDfg<'_> {
                 }
                 info::CoreDef::UnsafeIntrinsic(*i)
             }
-            CoreDef::TaskMayBlock => info::CoreDef::TaskMayBlock,
         }
     }
 
diff --git a/crates/environ/src/component/info.rs b/crates/environ/src/component/info.rs
index 679564a33a6c..0e96aa047038 100644
--- a/crates/environ/src/component/info.rs
+++ b/crates/environ/src/component/info.rs
@@ -392,10 +392,6 @@ pub enum CoreDef {
     Trampoline(TrampolineIndex),
     /// An intrinsic for compile-time builtins.
     UnsafeIntrinsic(UnsafeIntrinsic),
-    /// Reference to a wasm global which represents a runtime-managed boolean
-    /// indicating whether the currently-running task may perform a blocking
-    /// operation.
-    TaskMayBlock,
 }
 
 impl<T> From<CoreExport<T>> for CoreDef
diff --git a/crates/environ/src/component/translate.rs b/crates/environ/src/component/translate.rs
index 4a80c5f927b2..b345274a68a7 100644
--- a/crates/environ/src/component/translate.rs
+++ b/crates/environ/src/component/translate.rs
@@ -613,7 +613,6 @@ impl<'a, 'data> Translator<'a, 'data> {
 
                 let known_func = match arg {
                     CoreDef::InstanceFlags(_) => unreachable!("instance flags are not a function"),
-                    CoreDef::TaskMayBlock => unreachable!("task_may_block is not a function"),
 
                     // We could in theory inline these trampolines, so it could
                     // potentially make sense to record that we know this
diff --git a/crates/environ/src/component/translate/adapt.rs b/crates/environ/src/component/translate/adapt.rs
index 61ad82bbe7bf..aee064705348 100644
--- a/crates/environ/src/component/translate/adapt.rs
+++ b/crates/environ/src/component/translate/adapt.rs
@@ -455,8 +455,7 @@ impl PartitionAdapterModules {
             // These items can't transitively depend on an adapter
             dfg::CoreDef::Trampoline(_)
             | dfg::CoreDef::InstanceFlags(_)
-            | dfg::CoreDef::UnsafeIntrinsic(..)
-            | dfg::CoreDef::TaskMayBlock => {}
+            | dfg::CoreDef::UnsafeIntrinsic(..) => {}
         }
     }
 
diff --git a/crates/environ/src/fact.rs b/crates/environ/src/fact.rs
index 5282d3f0f442..6070299bd5f5 100644
--- a/crates/environ/src/fact.rs
+++ b/crates/environ/src/fact.rs
@@ -113,8 +113,6 @@ pub struct Module<'a> {
     helper_worklist: Vec<(FunctionId, Helper)>,
 
     exports: Vec<(u32, String)>,
-
-    task_may_block: Option<GlobalIndex>,
 }
 
 struct AdapterData {
@@ -298,7 +296,6 @@ impl<'a> Module<'a> {
             imported_unsafe_intrinsics: HashMap::new(),
             imported_traps: HashMap::new(),
             exports: Vec::new(),
-            task_may_block: None,
         }
     }
 
@@ -491,25 +488,6 @@ impl<'a> Module<'a> {
         idx
     }
 
-    fn import_task_may_block(&mut self) -> GlobalIndex {
-        if let Some(task_may_block) = self.task_may_block {
-            task_may_block
-        } else {
-            let task_may_block = self.import_global(
-                "instance",
-                "task_may_block",
-                GlobalType {
-                    val_type: ValType::I32,
-                    mutable: true,
-                    shared: false,
-                },
-                CoreDef::TaskMayBlock,
-            );
-            self.task_may_block = Some(task_may_block);
-            task_may_block
-        }
-    }
-
     fn import_transcoder(&mut self, transcoder: transcode::Transcoder) -> FuncIndex {
         *self
             .imported_transcoders
diff --git a/crates/environ/src/fact/trampoline.rs b/crates/environ/src/fact/trampoline.rs
index 132f373adb59..d79ed0fb7b5b 100644
--- a/crates/environ/src/fact/trampoline.rs
+++ b/crates/environ/src/fact/trampoline.rs
@@ -769,25 +769,7 @@ impl<'a, 'b> Compiler<'a, 'b> {
         let saved_lower_may_leave =
             self.trap_if_not_may_leave(adapter.lower.flags, Trap::CannotLeaveComponent);
 
-        let old_task_may_block = if self.module.tunables.concurrency_support {
-            // Save, clear, and later restore the `may_block` field.
-            let task_may_block = self.module.import_task_may_block();
-            let old_task_may_block = if self.types[adapter.lift.ty].async_ {
-                self.instruction(GlobalGet(task_may_block.as_u32()));
-                self.instruction(I32Eqz);
-                self.instruction(If(BlockType::Empty));
-                self.trap(Trap::CannotBlockSyncTask);
-                self.instruction(End);
-                None
-            } else {
-                let task_may_block = self.module.import_task_may_block();
-                self.instruction(GlobalGet(task_may_block.as_u32()));
-                let old_task_may_block = self.local_set_new_tmp(ValType::I32);
-                self.instruction(I32Const(0));
-                self.instruction(GlobalSet(task_may_block.as_u32()));
-                Some(old_task_may_block)
-            };
-
+        if self.module.tunables.concurrency_support {
             // Push a task onto the current task stack.
             //
             // Note that for sync-to-sync calls, we replace this call with
@@ -809,8 +791,6 @@ impl<'a, 'b> Compiler<'a, 'b> {
             ));
             let enter_sync_call = self.module.import_enter_sync_call();
             self.instruction(Call(enter_sync_call.as_u32()));
-
-            old_task_may_block
         } else if self.emit_resource_call {
             assert!(!self.types[adapter.lift.ty].async_);
             self.instruction(I32Const(
@@ -822,10 +802,7 @@ impl<'a, 'b> Compiler<'a, 'b> {
             ));
             let enter_sync_call = self.module.import_enter_sync_call();
             self.instruction(Call(enter_sync_call.as_u32()));
-            None
-        } else {
-            None
-        };
+        }
 
         // Perform the translation of arguments. Note that the `may_leave` flag
         // is cleared around this invocation for the callee as per the
@@ -874,7 +851,9 @@ impl<'a, 'b> Compiler<'a, 'b> {
         // With all the arguments on the stack the actual target function is
         // now invoked. The core wasm results of the function are then placed
         // into locals for result translation afterwards.
+
         self.instruction(Call(adapter.callee.as_u32()));
+
         let mut result_locals = Vec::with_capacity(lift_sig.results.len());
         let mut temps = Vec::new();
         for ty in lift_sig.results.iter().rev() {
@@ -944,16 +923,6 @@ impl<'a, 'b> Compiler<'a, 'b> {
             self.free_temp_local(tmp);
         }
 
-        if self.module.tunables.concurrency_support {
-            // Restore old `may_block_field`
-            if let Some(old_task_may_block) = old_task_may_block {
-                let task_may_block = self.module.import_task_may_block();
-                self.instruction(LocalGet(old_task_may_block.idx));
-                self.instruction(GlobalSet(task_may_block.as_u32()));
-                self.free_temp_local(old_task_may_block);
-            }
-        }
-
         self.exit_exception_barrier();
 
         self.finish()
diff --git a/crates/wasmtime/src/runtime/component/concurrent.rs b/crates/wasmtime/src/runtime/component/concurrent.rs
index 079ce49e6e20..f7c15af53d66 100644
--- a/crates/wasmtime/src/runtime/component/concurrent.rs
+++ b/crates/wasmtime/src/runtime/component/concurrent.rs
@@ -665,7 +665,6 @@ 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).
@@ -675,13 +674,9 @@ enum SuspendReason {
     Yielding {
         thread: QualifiedThreadId,
         cancellable: bool,
-        skip_may_block_check: bool,
     },
     /// The fiber was explicitly suspended with a call to `thread.suspend` or `thread.switch-to`.
-    ExplicitlySuspending {
-        thread: QualifiedThreadId,
-        skip_may_block_check: bool,
-    },
+    ExplicitlySuspending { thread: QualifiedThreadId },
 }
 
 /// Represents a pending call into guest code for a given guest task.
@@ -702,7 +697,7 @@ enum GuestCallKind {
     ///
     /// 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>),
+    StartImplicit(Box<dyn FnOnce(&mut dyn VMStore) -> Result<()> + Send + Sync>),
     StartExplicit(Box<dyn FnOnce(&mut dyn VMStore) -> Result<()> + Send + Sync>),
 }
 
@@ -723,18 +718,17 @@ impl fmt::Debug for GuestCallKind {
 /// The target of a suspension intrinsic.
 #[derive(Copy, Clone, Debug)]
 pub enum SuspensionTarget {
-    SomeSuspended(u32),
-    Some(u32),
+    Resume(u32),
+    Promote(u32),
     None,
 }
 
-impl SuspensionTarget {
-    fn is_none(&self) -> bool {
-        matches!(self, SuspensionTarget::None)
-    }
-    fn is_some(&self) -> bool {
-        !self.is_none()
-    }
+/// Behavior for `resume_thread`.
+#[derive(Copy, Clone, Debug)]
+pub enum ResumeThread {
+    Promote,
+    Resume,
+    ResumeLater,
 }
 
 /// Represents a pending call into guest code for a given guest thread.
@@ -755,15 +749,16 @@ impl GuestCall {
     /// - the call is for a not-yet started task and the (sub-)component
     /// instance to be called has backpressure enabled
     fn is_ready(&self, store: &mut StoreOpaque) -> Result<bool> {
-        let instance = store
-            .concurrent_state_mut()?
-            .get_mut(self.thread.task)?
-            .instance;
+        let task = store.concurrent_state_mut()?.get_mut(self.thread.task)?;
+        let async_typed = task.async_typed;
+        let instance = task.instance;
         let state = store.instance_state(instance).concurrent_state();
 
         let ready = match &self.kind {
             GuestCallKind::DeliverEvent { .. } => !state.do_not_enter,
-            GuestCallKind::StartImplicit(_) => !(state.do_not_enter || state.backpressure > 0),
+            GuestCallKind::StartImplicit(_) => {
+                !async_typed || !(state.do_not_enter || state.backpressure > 0)
+            }
             GuestCallKind::StartExplicit(_) => true,
         };
         log::trace!(
@@ -787,11 +782,21 @@ enum WorkItem {
     /// A host task to be pushed to `ConcurrentState::futures`.
     PushFuture(AlwaysMut<HostTaskFuture>),
     /// A fiber to resume.
-    ResumeFiber(StoreFiber<'static>),
+    ResumeFiber {
+        instance: RuntimeInstance,
+        thread: QualifiedThreadId,
+        fiber: StoreFiber<'static>,
+    },
     /// A thread to resume.
-    ResumeThread(RuntimeComponentInstanceIndex, QualifiedThreadId),
+    ResumeThread {
+        instance: RuntimeInstance,
+        thread: QualifiedThreadId,
+    },
     /// A pending call into guest code for a given guest task.
-    GuestCall(RuntimeComponentInstanceIndex, GuestCall),
+    GuestCall {
+        instance: RuntimeInstance,
+        call: GuestCall,
+    },
     /// A job to run on a worker fiber.
     WorkerFunction(AlwaysMut<Box<dyn FnOnce(&mut dyn VMStore) -> Result<()> + Send>>),
 }
@@ -800,16 +805,22 @@ impl fmt::Debug for WorkItem {
     fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
         match self {
             Self::PushFuture(_) => f.debug_tuple("PushFuture").finish(),
-            Self::ResumeFiber(_) => f.debug_tuple("ResumeFiber").finish(),
-            Self::ResumeThread(instance, thread) => f
-                .debug_tuple("ResumeThread")
-                .field(instance)
-                .field(thread)
+            Self::ResumeFiber {
+                instance, thread, ..
+            } => f
+                .debug_struct("ResumeFiber")
+                .field("instance", instance)
+                .field("thread", thread)
                 .finish(),
-            Self::GuestCall(instance, call) => f
-                .debug_tuple("GuestCall")
-                .field(instance)
-                .field(call)
+            Self::ResumeThread { instance, thread } => f
+                .debug_struct("ResumeThread")
+                .field("instance", instance)
+                .field("thread", thread)
+                .finish(),
+            Self::GuestCall { instance, call } => f
+                .debug_struct("GuestCall")
+                .field("instance", instance)
+                .field("call", call)
                 .finish(),
             Self::WorkerFunction(_) => f.debug_tuple("WorkerFunction").finish(),
         }
@@ -882,6 +893,9 @@ pub(crate) fn poll_and_block<R: Send + Sync + 'static>(
         // then use `GuestThread::sync_call_set` to wait for the task to
         // complete, suspending the current fiber until it does so.
         Poll::Pending => {
+            let caller_instance = store.concurrent_state_mut()?.get_mut(caller.task)?.instance;
+            store.trap_if_may_not_suspend(caller_instance)?;
+
             let state = store.concurrent_state_mut()?;
             state.push_future(future);
 
@@ -891,7 +905,6 @@ pub(crate) fn poll_and_block<R: Send + Sync + 'static>(
             store.suspend(SuspendReason::Waiting {
                 set,
                 thread: caller,
-                skip_may_block_check: false,
             })?;
 
             // Remove the `task` from the `sync_call_set` to ensure that when
@@ -914,70 +927,59 @@ 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<()> {
-    let mut next = Some(call);
-    while let Some(call) = next.take() {
-        match call.kind {
-            GuestCallKind::DeliverEvent { instance, set } => {
-                let (event, waitable) =
-                    match instance.get_event(store, call.thread.task, set, true)? {
-                        Some(pair) => pair,
-                        None => bail_bug!("delivering non-present event"),
-                    };
-                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,
-                );
-
-                let old_thread = store.set_thread(call.thread)?;
-                log::trace!(
-                    "GuestCallKind::DeliverEvent: replaced {old_thread:?} with {:?} as current thread",
-                    call.thread
-                );
-
-                store.enter_instance(runtime_instance);
-
-                let Some(callback) = store
-                    .concurrent_state_mut()?
-                    .get_mut(call.thread.task)?
-                    .callback
-                    .take()
-                else {
-                    bail_bug!("guest task callback field not present")
-                };
+    match call.kind {
+        GuestCallKind::DeliverEvent { instance, set } => {
+            let (event, waitable) = match instance.get_event(store, call.thread.task, set, true)? {
+                Some(pair) => pair,
+                None => bail_bug!("delivering non-present event"),
+            };
+            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,
+            );
+
+            let old_thread = store.set_thread(call.thread)?;
+            log::trace!(
+                "GuestCallKind::DeliverEvent: replaced {old_thread:?} with {:?} as current thread",
+                call.thread
+            );
+
+            store.enter_instance(runtime_instance);
+
+            let Some(callback) = store
+                .concurrent_state_mut()?
+                .get_mut(call.thread.task)?
+                .callback
+                .take()
+            else {
+                bail_bug!("guest task callback field not present")
+            };
 
-                let code = callback(store, event, handle)?;
+            let code = callback(store, event, handle)?;
 
-                store
-                    .concurrent_state_mut()?
-                    .get_mut(call.thread.task)?
-                    .callback = Some(callback);
+            store
+                .concurrent_state_mut()?
+                .get_mut(call.thread.task)?
+                .callback = Some(callback);
 
-                store.exit_instance(runtime_instance)?;
+            store.exit_instance(runtime_instance)?;
 
-                store.set_thread(old_thread)?;
+            store.set_thread(old_thread)?;
 
-                next = instance.handle_callback_code(
-                    store,
-                    call.thread,
-                    runtime_instance.index,
-                    code,
-                )?;
+            instance.handle_callback_code(store, call.thread, runtime_instance.index, code)?;
 
-                log::trace!(
-                    "GuestCallKind::DeliverEvent: restored {old_thread:?} as current thread"
-                );
-            }
-            GuestCallKind::StartImplicit(fun) => {
-                next = fun(store)?;
-            }
-            GuestCallKind::StartExplicit(fun) => {
-                fun(store)?;
-            }
+            log::trace!("GuestCallKind::DeliverEvent: restored {old_thread:?} as current thread");
+        }
+        GuestCallKind::StartImplicit(fun) => {
+            fun(store)?;
+        }
+        GuestCallKind::StartExplicit(fun) => {
+            fun(store)?;
         }
     }
 
@@ -1197,10 +1199,7 @@ impl<T> StoreContextMut<'_, T> {
     pub(super) async fn run_concurrent_trap_on_idle<R>(
         self,
         fun: impl AsyncFnOnce(&Accessor<T>) -> R,
-    ) -> Result<R>
-    where
-        T: Send + 'static,
-    {
+    ) -> Result<R> {
         self.do_run_concurrent(fun, true).await
     }
 
@@ -1208,10 +1207,7 @@ impl<T> StoreContextMut<'_, T> {
         mut self,
         fun: impl AsyncFnOnce(&Accessor<T>) -> R,
         trap_on_idle: bool,
-    ) -> Result<R>
-    where
-        T: Send + 'static,
-    {
+    ) -> Result<R> {
         debug_assert!(self.0.concurrency_support());
         check_recursive_run();
         let token = StoreToken::new(self.as_context_mut());
@@ -1223,6 +1219,11 @@ impl<T> StoreContextMut<'_, T> {
 
         impl<'a, T, V> Drop for Dropper<'a, T, V> {
             fn drop(&mut self) {
+                self.store
+                    .0
+                    .concurrent_state_mut_already_forced_current_thread()
+                    .event_loop_running = false;
+
                 tls::set(self.store.0, || {
                     // SAFETY: Here we drop the value without moving it for the
                     // first and only time -- per the contract for `Drop::drop`,
@@ -1234,6 +1235,9 @@ impl<T> StoreContextMut<'_, T> {
         }
 
         let accessor = &Accessor::new(token);
+        self.0
+            .concurrent_state_mut_already_forced_current_thread()
+            .event_loop_running = true;
         let dropper = &mut Dropper {
             store: self,
             value: ManuallyDrop::new(fun(accessor)),
@@ -1257,10 +1261,7 @@ impl<T> StoreContextMut<'_, T> {
         mut self,
         mut future: Pin<&mut impl Future<Output = R>>,
         trap_on_idle: bool,
-    ) -> Result<R>
-    where
-        T: Send + 'static,
-    {
+    ) -> Result<R> {
         struct Reset<'a, T: 'static> {
             store: StoreContextMut<'a, T>,
             futures: Option<FuturesUnordered<HostTaskFuture>>,
@@ -1374,6 +1375,39 @@ impl<T> StoreContextMut<'_, T> {
                             if trap_on_idle {
                                 // `trap_on_idle` is true, so we exit
                                 // immediately.
+
+                                // If there are any tasks belonging to
+                                // an instance which may not suspend, trap with
+                                // `CannotBlockSyncTask`:
+                                let instances = reset
+                                    .store
+                                    .0
+                                    .concurrent_state_mut()?
+                                    .table
+                                    .get_mut()
+                                    .iter_mut()
+                                    .filter_map(|entry| {
+                                        if let Some(task) = entry.downcast_ref::<GuestTask>() {
+                                            Some(task.instance)
+                                        } else {
+                                            None
+                                        }
+                                    })
+                                    .collect::<Vec<_>>();
+
+                                for instance in instances {
+                                    if reset
+                                        .store
+                                        .0
+                                        .instance_state(instance)
+                                        .concurrent_state()
+                                        .do_not_suspend
+                                    {
+                                        return Poll::Ready(Err(Trap::CannotBlockSyncTask.into()));
+                                    }
+                                }
+
+                                // Otherwise, trap with `AsyncDeadlock`:
                                 Poll::Ready(Err(Trap::AsyncDeadlock.into()))
                             } else {
                                 // `trap_on_idle` is false, so we assume
@@ -1416,7 +1450,9 @@ impl<T> StoreContextMut<'_, T> {
                         fn drop(&mut self) {
                             while let Some(item) = self.ready.next() {
                                 match item {
-                                    WorkItem::ResumeFiber(mut fiber) => fiber.dispose(self.store.0),
+                                    WorkItem::ResumeFiber { mut fiber, .. } => {
+                                        fiber.dispose(self.store.0)
+                                    }
                                     WorkItem::PushFuture(future) => {
                                         tls::set(self.store.0, move || drop(future))
                                     }
@@ -1469,10 +1505,7 @@ impl<T> StoreContextMut<'_, T> {
     }
 
     /// Handle the specified work item, possibly resuming a fiber if applicable.
-    async fn handle_work_item(self, item: WorkItem) -> Result<()>
-    where
-        T: Send,
-    {
+    async fn handle_work_item(self, item: WorkItem) -> Result<()> {
         log::trace!("handle work item {item:?}");
         match item {
             WorkItem::PushFuture(future) => {
@@ -1481,10 +1514,10 @@ impl<T> StoreContextMut<'_, T> {
                     .futures_mut()?
                     .push(future.into_inner());
             }
-            WorkItem::ResumeFiber(fiber) => {
+            WorkItem::ResumeFiber { fiber, .. } => {
                 self.0.resume_fiber(fiber).await?;
             }
-            WorkItem::ResumeThread(_, thread) => {
+            WorkItem::ResumeThread { thread, .. } => {
                 if let GuestThreadState::Ready { fiber, .. } = mem::replace(
                     &mut self.0.concurrent_state_mut()?.get_mut(thread.thread)?.state,
                     GuestThreadState::Running,
@@ -1494,7 +1527,7 @@ impl<T> StoreContextMut<'_, T> {
                     bail_bug!("cannot resume non-pending thread {thread:?}");
                 }
             }
-            WorkItem::GuestCall(_, call) => {
+            WorkItem::GuestCall { call, .. } => {
                 if call.is_ready(self.0)? {
                     self.run_on_worker(WorkerItem::GuestCall(call)).await?;
                 } else {
@@ -1529,26 +1562,43 @@ impl<T> StoreContextMut<'_, T> {
     }
 
     /// Execute the specified guest call on a worker fiber.
-    async fn run_on_worker(self, item: WorkerItem) -> Result<()>
-    where
-        T: Send,
-    {
+    async fn run_on_worker(self, item: WorkerItem) -> Result<()> {
         let worker = if let Some(fiber) = self.0.concurrent_state_mut()?.worker.take() {
             fiber
         } else {
-            fiber::make_fiber(self.0, move |store| {
-                loop {
-                    let Some(item) = store.concurrent_state_mut()?.worker_item.take() else {
-                        bail_bug!("worker_item not present when resuming fiber")
-                    };
-                    match item {
-                        WorkerItem::GuestCall(call) => handle_guest_call(store, call)?,
-                        WorkerItem::Function(fun) => fun.into_inner()(store)?,
-                    }
+            // SAFETY: the `make_fiber_unchecked` function is unsafe because the
+            // returned fiber is unconditionally `Send` as opposed to being
+            // conditionally send depending on the argument (in this case
+            // `self.0`). This `async` function, however, is conditionally
+            // `Send` depending on `self`, in this case `StoreContextMut<T>`,
+            // which is already going to be conditionally `Send` depending on
+            // `T`.
+            //
+            // The returned fiber is possibly stored within the `Store<T>` as
+            // well. If `T: Send` then that's fine and everything's dandy. If
+            // `T: !Send`, however, then the store is already not-`Send` meaning
+            // that putting more actually-not-`Send` things inside of it isn't
+            // an issue.
+            //
+            // The main issue here is that the returned fiber effectively can't
+            // get transferred outside the context of the store. That's an
+            // implementation detail we'll have to rely on, but is currently
+            // true.
+            unsafe {
+                fiber::make_fiber_unchecked(self.0, move |store| {
+                    loop {
+                        let Some(item) = store.concurrent_state_mut()?.worker_item.take() else {
+                            bail_bug!("worker_item not present when resuming fiber")
+                        };
+                        match item {
+                            WorkerItem::GuestCall(call) => handle_guest_call(store, call)?,
+                            WorkerItem::Function(fun) => fun.into_inner()(store)?,
+                        }
 
-                    store.suspend(SuspendReason::NeedWork)?;
-                }
-            })?
+                        store.suspend(SuspendReason::NeedWork)?;
+                    }
+                })?
+            }
         };
 
         let worker_item = &mut self.0.concurrent_state_mut()?.worker_item;
@@ -1729,6 +1779,26 @@ impl StoreOpaque {
         Ok(false)
     }
 
+    fn enter_sync_call(&mut self, callee: RuntimeInstance) -> Result<()> {
+        log::trace!("enter sync-typed call {callee:?}");
+        let state = self.instance_state(callee).concurrent_state();
+        if cfg!(debug_assertions) && state.do_not_suspend {
+            bail_bug!("attempted to reenter instance while sync call in progress");
+        }
+        state.do_not_suspend = true;
+        Ok(())
+    }
+
+    fn exit_sync_call(&mut self, callee: RuntimeInstance) -> Result<()> {
+        log::trace!("exit sync-typed call {callee:?}");
+        let state = self.instance_state(callee).concurrent_state();
+        if cfg!(debug_assertions) && !state.do_not_suspend {
+            bail_bug!("do_not_suspend flag switched back to false unexpectedly");
+        }
+        state.do_not_suspend = false;
+        Ok(())
+    }
+
     /// Push a `GuestTask` onto the task stack for either a sync-to-sync,
     /// guest-to-guest call or a sync host-to-guest call.
     ///
@@ -1743,10 +1813,10 @@ impl StoreOpaque {
     pub(crate) fn enter_guest_sync_call(
         &mut self,
         guest_caller: Option<RuntimeInstance>,
-        callee_async: bool,
+        callee_async_typed: bool,
         callee: RuntimeInstance,
     ) -> Result<()> {
-        log::trace!("enter sync call {callee:?}");
+        log::trace!("enter sync-lifted call {callee:?}");
         if !self.concurrency_support() {
             return self.enter_call_not_concurrent();
         }
@@ -1781,7 +1851,8 @@ impl StoreOpaque {
             },
             None,
             callee,
-            callee_async,
+            callee_async_typed,
+            true,
         )?;
 
         Instance::from_wasmtime(self, callee.instance).add_guest_thread_to_instance_table(
@@ -1791,10 +1862,14 @@ impl StoreOpaque {
         )?;
         self.set_thread(guest_thread)?;
 
+        if !callee_async_typed {
+            self.enter_sync_call(callee)?;
+        }
+
         Ok(())
     }
 
-    /// Pop a `GuestTask` previously pushed using `enter_sync_call`.
+    /// Pop a `GuestTask` previously pushed using `enter_guest_sync_call`.
     ///
     /// NB: for sync-to-sync, guest-to-guest calls we delay task construction in
     /// fused adapters and then when the call returns we check to see if the
@@ -1805,21 +1880,28 @@ impl StoreOpaque {
         if !self.concurrency_support() {
             return Ok(self.exit_call_not_concurrent());
         }
+
         let thread = match self.set_thread(CurrentThread::None)?.guest() {
             Some(t) => *t,
             None => bail_bug!("expected task when exiting"),
         };
         let task = self.concurrent_state_mut()?.get_mut(thread.task)?;
         let instance = task.instance;
+
         let caller = match &task.caller {
             &Caller::Guest { thread } => thread.into(),
             &Caller::Host { caller, .. } => caller,
         };
         task.lift_result = None;
         task.exited = true;
+
+        if !task.async_typed {
+            self.exit_sync_call(instance)?;
+        }
+
         self.set_thread(caller)?;
 
-        log::trace!("exit sync call {instance:?}");
+        log::trace!("exit sync-lifted call {instance:?}");
         self.cleanup_thread(thread, instance, CleanupTask::Yes)?;
 
         Ok(())
@@ -1942,24 +2024,6 @@ impl StoreOpaque {
             *self.vm_store_context_mut().component_context_mut() = context;
         }
 
-        // Each time we switch threads, we conservatively set `task_may_block`
-        // to `false` for the component instance we're switching away from (if
-        // any), meaning it will be `false` for any new thread created for that
-        // instance unless explicitly set otherwise.
-        //
-        // Additionally if we're switching to a new thread, set its component
-        // instance's `task_may_block` according to where it left off.
-        let state = self.concurrent_state_mut()?;
-        if let Some(old_task) = old_thread.guest_task() {
-            let instance = state.get_mut(old_task)?.instance.instance;
-            self.component_instance_mut(instance)
-                .set_task_may_block(false)
-        }
-
-        if thread.guest_task().is_some() {
-            self.set_task_may_block()?;
-        }
-
         // Keep the JIT-visible current-thread pointer in sync.
         *self.vm_store_context_mut().current_thread_mut() = if thread.is_none() {
             VMLazyThread::none()
@@ -1970,34 +2034,30 @@ impl StoreOpaque {
         Ok(old_thread)
     }
 
-    /// Set the global variable representing whether the current task may block
-    /// prior to entering Wasm code.
-    fn set_task_may_block(&mut self) -> Result<()> {
-        let guest_thread = self.current_guest_thread()?;
-        let state = self.concurrent_state_mut()?;
-        let instance = state.get_mut(guest_thread.task)?.instance.instance;
-        let may_block = self.concurrent_state_mut()?.may_block(guest_thread.task)?;
-        self.component_instance_mut(instance)
-            .set_task_may_block(may_block);
-        Ok(())
-    }
-
-    pub(crate) fn check_blocking(&mut self) -> Result<()> {
-        if !self.concurrency_support() {
-            return Ok(());
-        }
-        let task = self.current_guest_thread()?.task;
-        let state = self.concurrent_state_mut()?;
-        let instance = state.get_mut(task)?.instance.instance;
-        let task_may_block = self.component_instance(instance).get_task_may_block();
-
-        if task_may_block {
+    fn trap_if_may_not_suspend(&mut self, instance: RuntimeInstance) -> Result<()> {
+        if self.may_suspend(instance)? {
             Ok(())
         } else {
             Err(Trap::CannotBlockSyncTask.into())
         }
     }
 
+    fn may_suspend(&mut self, instance: RuntimeInstance) -> Result<bool> {
+        // Call this for the side effect of forcing any deferred task creation,
+        // which may influence the value of `ConcurrentState::do_not_suspend`
+        // below:
+        self.concurrent_state_mut()?;
+
+        Ok(!self.concurrency_support()
+            || !self
+                .instance_state(instance)
+                .concurrent_state()
+                .do_not_suspend
+            || self
+                .concurrent_state_mut()?
+                .promote_instance_local_thread_work_items(instance)?)
+    }
+
     /// Record that we're about to enter a (sub-)component instance which does
     /// not support more than one concurrent, stackful activation, meaning it
     /// cannot be entered again until the next call returns.
@@ -2033,7 +2093,7 @@ impl StoreOpaque {
             let call = GuestCall { thread, kind };
             if call.is_ready(self)? {
                 self.concurrent_state_mut()?
-                    .push_high_priority(WorkItem::GuestCall(instance.index, call));
+                    .push_high_priority(WorkItem::GuestCall { instance, call });
             } else {
                 self.instance_state(instance)
                     .concurrent_state()
@@ -2112,8 +2172,8 @@ impl StoreOpaque {
                 } => {
                     state.get_mut(thread.thread)?.state =
                         GuestThreadState::Ready { fiber, cancellable };
-                    let instance = state.get_mut(thread.task)?.instance.index;
-                    state.push_low_priority(WorkItem::ResumeThread(instance, thread));
+                    let instance = state.get_mut(thread.task)?.instance;
+                    state.push_low_priority(WorkItem::ResumeThread { instance, thread });
                 }
                 SuspendReason::ExplicitlySuspending { thread, .. } => {
                     state.get_mut(thread.thread)?.state = GuestThreadState::Suspended(fiber);
@@ -2157,35 +2217,14 @@ impl StoreOpaque {
             CurrentThread::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.
-        debug_assert!(
-            matches!(
-                reason,
-                SuspendReason::ExplicitlySuspending {
-                    skip_may_block_check: true,
-                    ..
-                } | SuspendReason::Waiting {
-                    skip_may_block_check: true,
-                    ..
-                } | SuspendReason::Yielding {
-                    skip_may_block_check: true,
-                    ..
-                }
-            ) || old_guest_thread
-                .guest_task()
-                .map(|task| self.concurrent_state_mut()?.may_block(task))
-                .transpose()?
-                .unwrap_or(true)
-        );
-
         let suspend_reason = &mut self.concurrent_state_mut()?.suspend_reason;
         assert!(suspend_reason.is_none());
         *suspend_reason = Some(reason);
 
+        if !self.fiber_async_state_mut().can_block() {
+            return Err(format_err!("future dropped"));
+        }
+
         self.with_blocking(|_, cx| cx.suspend(StoreFiberYield::ReleaseStore))?;
 
         if task.is_some() {
@@ -2195,7 +2234,11 @@ impl StoreOpaque {
         Ok(())
     }
 
-    fn wait_for_event(&mut self, waitable: Waitable) -> Result<()> {
+    fn wait_for_event(
+        &mut self,
+        caller_instance: RuntimeInstance,
+        waitable: Waitable,
+    ) -> Result<()> {
         let caller = self.current_guest_thread()?;
         let state = self.concurrent_state_mut()?;
 
@@ -2203,10 +2246,12 @@ impl StoreOpaque {
 
         let set = state.get_mut(caller.thread)?.sync_call_set;
         waitable.join(state, Some(set))?;
+
+        self.trap_if_may_not_suspend(caller_instance)?;
+
         self.suspend(SuspendReason::Waiting {
             set,
             thread: caller,
-            skip_may_block_check: false,
         })?;
         let state = self.concurrent_state_mut()?;
         waitable.join(state, None)
@@ -2239,6 +2284,18 @@ impl StoreOpaque {
         runtime_instance: RuntimeInstance,
         cleanup_task: CleanupTask,
     ) -> Result<()> {
+        // If a thread exits while the instance to which it belongs is running a
+        // sync-typed task that wants to block, and if there are no other
+        // eligible threads to run, we must trap here.
+        //
+        // However, if this thread never ran at all, we skip that check.
+        match cleanup_task {
+            CleanupTask::Yes => {
+                self.trap_if_may_not_suspend(runtime_instance)?;
+            }
+            CleanupTask::No => {}
+        }
+
         let state = self.concurrent_state_mut()?;
         let thread_data = state.get_mut(guest_thread.thread)?;
         let sync_call_set = thread_data.sync_call_set;
@@ -2360,6 +2417,15 @@ impl StoreOpaque {
         assert_eq!((bits << 1) >> 1, bits);
         Ok(Some((bits << 1) | u32::from(is_host)))
     }
+
+    pub(crate) fn queue_task(
+        &mut self,
+        task: impl FnOnce(&mut dyn VMStore) -> Result<()> + Send + 'static,
+    ) -> Result<()> {
+        self.concurrent_state_mut()?
+            .push_high_priority(WorkItem::WorkerFunction(AlwaysMut::new(Box::new(task))));
+        Ok(())
+    }
 }
 
 enum CleanupTask {
@@ -2428,7 +2494,18 @@ impl Instance {
         guest_thread: QualifiedThreadId,
         runtime_instance: RuntimeComponentInstanceIndex,
         code: u32,
-    ) -> Result<Option<GuestCall>> {
+    ) -> Result<()> {
+        if cfg!(debug_assertions)
+            && store
+                .instance_state(self.runtime_instance(runtime_instance))
+                .concurrent_state()
+                .do_not_suspend
+        {
+            bail_bug!(
+                "should not be possible to run a callback while a sync call is outstanding for the same instance"
+            );
+        }
+
         let (code, set) = unpack_callback_code(code);
 
         log::trace!("received callback code from {guest_thread:?}: {code} (set: {set})");
@@ -2444,7 +2521,7 @@ impl Instance {
             Ok(TableId::<WaitableSet>::new(set))
         };
 
-        Ok(match code {
+        match code {
             callback_code::EXIT => {
                 log::trace!("implicit thread {guest_thread:?} completed");
                 let task = store.concurrent_state_mut()?.get_mut(guest_thread.task)?;
@@ -2455,7 +2532,6 @@ impl Instance {
                     self.runtime_instance(runtime_instance),
                     CleanupTask::Yes,
                 )?;
-                None
             }
             callback_code::YIELD => {
                 let task = state.get_mut(guest_thread.task)?;
@@ -2475,23 +2551,14 @@ impl Instance {
                         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(runtime_instance, 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)
-                }
+                // 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 {
+                    instance: self.runtime_instance(runtime_instance),
+                    call,
+                });
             }
             callback_code::WAIT => {
-                // The task may only return `WAIT` 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()?;
 
@@ -2499,16 +2566,16 @@ impl Instance {
                     || !state.get_mut(set)?.ready.is_empty()
                 {
                     // An event is immediately available; deliver it ASAP.
-                    state.push_high_priority(WorkItem::GuestCall(
-                        runtime_instance,
-                        GuestCall {
+                    state.push_high_priority(WorkItem::GuestCall {
+                        instance: self.runtime_instance(runtime_instance),
+                        call: GuestCall {
                             thread: guest_thread,
                             kind: GuestCallKind::DeliverEvent {
                                 instance: self,
                                 set: Some(set),
                             },
                         },
-                    ));
+                    });
                 } else {
                     // No event is immediately available.
                     //
@@ -2532,10 +2599,11 @@ impl Instance {
                         bail_bug!("set's waiting set already had this thread registered");
                     }
                 }
-                None
             }
             _ => bail!(Trap::UnsupportedCallbackCode),
-        })
+        }
+
+        Ok(())
     }
 
     /// Add the specified guest call to the "high priority" work item queue, to
@@ -2673,8 +2741,7 @@ impl Instance {
                 let code = unsafe { storage[0].assume_init() }.get_i32() as u32;
 
                 self.handle_callback_code(store, guest_thread, callee_instance.index, code)
-            })
-                as Box<dyn FnOnce(&mut dyn VMStore) -> Result<Option<GuestCall>> + Send + Sync>
+            }) as Box<dyn FnOnce(&mut dyn VMStore) -> Result<()> + Send + Sync>
         } else {
             let token = StoreToken::new(store.as_context_mut());
             Box::new(move |store: &mut dyn VMStore| {
@@ -2696,6 +2763,15 @@ impl Instance {
                     store.enter_instance(callee_instance);
                 }
 
+                let callee_async_typed = store
+                    .concurrent_state_mut()?
+                    .get_mut(guest_thread.task)?
+                    .async_typed;
+
+                if !callee_async_typed {
+                    store.enter_sync_call(callee_instance)?;
+                }
+
                 // SAFETY: See the documentation for `make_call` to review the
                 // contract we must uphold for `call` here.
                 //
@@ -2704,6 +2780,10 @@ impl Instance {
                 // over must be valid.
                 let storage = call(store)?;
 
+                if !callee_async_typed {
+                    store.exit_sync_call(callee_instance)?;
+                }
+
                 if !async_ {
                     // This is a sync-lifted export, so now is when we lift the
                     // result, optionally call the post-return function, if any,
@@ -2761,20 +2841,20 @@ impl Instance {
 
                 // This is a callback-less call, so the implicit thread has now completed
                 store.cleanup_thread(guest_thread, callee_instance, CleanupTask::Yes)?;
-                Ok(None)
+                Ok(())
             })
         };
 
         store
             .0
             .concurrent_state_mut()?
-            .push_high_priority(WorkItem::GuestCall(
-                callee_instance.index,
-                GuestCall {
+            .push_high_priority(WorkItem::GuestCall {
+                instance: callee_instance,
+                call: GuestCall {
                     thread: guest_thread,
                     kind: GuestCallKind::StartImplicit(fun),
                 },
-            ));
+            });
 
         Ok(())
     }
@@ -2799,18 +2879,11 @@ impl Instance {
         caller_instance: RuntimeComponentInstanceIndex,
         callee_instance: RuntimeComponentInstanceIndex,
         task_return_type: TypeTupleIndex,
-        callee_async: bool,
+        callee_async_typed: bool,
         memory: *mut VMMemoryDefinition,
         string_encoding: StringEncoding,
         caller_info: CallerInfo,
     ) -> Result<()> {
-        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.check_blocking()?;
-        }
-
         enum ResultInfo {
             Heap { results: u32 },
             Stack { result_count: u32 },
@@ -2972,7 +3045,10 @@ impl Instance {
             Caller::Guest { thread: old_thread },
             None,
             self.runtime_instance(callee_instance),
-            callee_async,
+            callee_async_typed,
+            // We don't know whether the callee export was lifted sync or async
+            // yet, but we'll update this in `start_call`:
+            false,
         )?;
 
         // Make the new thread the current one so that `Self::start_call` knows
@@ -3041,7 +3117,11 @@ impl Instance {
         let async_caller = storage.is_none();
         let guest_thread = store.0.current_guest_thread()?;
         let state = store.0.concurrent_state_mut()?;
-        let callee_async = state.get_mut(guest_thread.task)?.async_function;
+
+        if !state.event_loop_running {
+            bail_bug!("Instance::start_call called without a running event loop");
+        }
+
         let callee = SendSyncPtr::new(callee);
         let param_count = usize::try_from(param_count)?;
         assert!(param_count <= MAX_FLAT_PARAMS);
@@ -3049,6 +3129,9 @@ impl Instance {
         assert!(result_count <= MAX_FLAT_RESULTS);
 
         let task = state.get_mut(guest_thread.task)?;
+
+        task.async_lifted = (flags & START_FLAG_ASYNC_CALLEE) != 0;
+
         if let Some(callback) = NonNull::new(callback) {
             // We're calling an async-lifted export with a callback, so store
             // the callback and related context as part of the task so we can
@@ -3112,14 +3195,6 @@ 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()?;
@@ -3155,6 +3230,7 @@ impl Instance {
                 // The callee hasn't returned yet, and the caller is calling via
                 // a sync-lowered import, so we loop and keep waiting until the
                 // callee returns.
+                store.0.trap_if_may_not_suspend(caller_instance)?;
             }
         };
 
@@ -3581,18 +3657,12 @@ impl Instance {
         set: u32,
         payload: u32,
     ) -> Result<u32> {
-        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.check_blocking()?;
-        }
-
         let &CanonicalOptions {
             cancellable,
             instance: caller_instance,
             ..
         } = &self.id().get(store).component().env_component().options[options];
+        let caller = self.runtime_instance(caller_instance);
         let rep = store
             .instance_state(self.runtime_instance(caller_instance))
             .handle_table()
@@ -3600,6 +3670,7 @@ impl Instance {
 
         self.waitable_check(
             store,
+            caller,
             cancellable,
             WaitableCheck::Wait,
             WaitableCheckParams {
@@ -3623,13 +3694,15 @@ impl Instance {
             instance: caller_instance,
             ..
         } = &self.id().get(store).component().env_component().options[options];
+        let caller = self.runtime_instance(caller_instance);
         let rep = store
-            .instance_state(self.runtime_instance(caller_instance))
+            .instance_state(caller)
             .handle_table()
             .waitable_set_rep(set)?;
 
         self.waitable_check(
             store,
+            caller,
             cancellable,
             WaitableCheck::Poll,
             WaitableCheckParams {
@@ -3724,50 +3797,75 @@ impl Instance {
         store: &mut StoreOpaque,
         runtime_instance: RuntimeComponentInstanceIndex,
         thread_idx: u32,
-        high_priority: bool,
-        allow_ready: bool,
-    ) -> Result<()> {
+        how: ResumeThread,
+    ) -> Result<bool> {
         let thread_id =
             GuestThread::from_instance(self.id().get_mut(store), runtime_instance, thread_idx)?;
         let state = store.concurrent_state_mut()?;
         let guest_thread = QualifiedThreadId::qualify(state, thread_id)?;
         let thread = state.get_mut(guest_thread.thread)?;
+        let high_priority = match how {
+            ResumeThread::Promote | ResumeThread::Resume => true,
+            ResumeThread::ResumeLater => false,
+        };
+
+        match (&how, &thread.state) {
+            // Promotion is a noop unless the thread is in a ready state.
+            (ResumeThread::Promote, GuestThreadState::Ready { .. }) => {}
+            (ResumeThread::Promote, _) => return Ok(false),
+
+            // When resuming a thread it must be in a suspended state otherwise
+            // this operation is a trap.
+            (
+                ResumeThread::Resume | ResumeThread::ResumeLater,
+                GuestThreadState::NotStartedExplicit(_) | GuestThreadState::Suspended(_),
+            ) => {}
+            (ResumeThread::Resume | ResumeThread::ResumeLater, _) => {
+                bail!(Trap::CannotResumeThread)
+            }
+        }
 
         match mem::replace(&mut thread.state, GuestThreadState::Running) {
             GuestThreadState::NotStartedExplicit(start_func) => {
                 log::trace!("starting thread {guest_thread:?}");
-                let guest_call = WorkItem::GuestCall(
-                    runtime_instance,
-                    GuestCall {
+                let guest_call = WorkItem::GuestCall {
+                    instance: self.runtime_instance(runtime_instance),
+                    call: GuestCall {
                         thread: guest_thread,
                         kind: GuestCallKind::StartExplicit(Box::new(move |store| {
                             start_func(store, guest_thread)
                         })),
                     },
-                );
+                };
                 store
                     .concurrent_state_mut()?
                     .push_work_item(guest_call, high_priority);
             }
             GuestThreadState::Suspended(fiber) => {
                 log::trace!("resuming thread {thread_id:?} that was suspended");
-                store
-                    .concurrent_state_mut()?
-                    .push_work_item(WorkItem::ResumeFiber(fiber), high_priority);
+                store.concurrent_state_mut()?.push_work_item(
+                    WorkItem::ResumeFiber {
+                        instance: self.runtime_instance(runtime_instance),
+                        thread: guest_thread,
+                        fiber,
+                    },
+                    high_priority,
+                );
             }
-            GuestThreadState::Ready { fiber, cancellable } if allow_ready => {
+            GuestThreadState::Ready { fiber, cancellable } => {
                 log::trace!("resuming thread {thread_id:?} that was ready");
                 thread.state = GuestThreadState::Ready { fiber, cancellable };
                 store
                     .concurrent_state_mut()?
-                    .promote_thread_work_item(guest_thread);
+                    .promote_thread_work_items(guest_thread)?;
             }
-            other => {
+            other @ (GuestThreadState::NotStartedImplicit
+            | GuestThreadState::Running
+            | GuestThreadState::Completed) => {
                 thread.state = other;
-                bail!(Trap::CannotResumeThread);
             }
         }
-        Ok(())
+        Ok(true)
     }
 
     fn add_guest_thread_to_instance_table(
@@ -3787,8 +3885,9 @@ impl Instance {
         Ok(guest_id)
     }
 
-    /// Helper function for the `thread.yield`, `thread.yield-to-suspended`, `thread.suspend`,
-    /// `thread.suspend-to`, and `thread.suspend-to-suspended` intrinsics.
+    /// Helper function for the `thread.yield`, thread.suspend`,
+    /// `thread.suspend-then-resume`, `thread.suspend-then-promote`,
+    /// `thread.yield-then-resume`, and `thread.yield-then-promote` intrinsics.
     pub(crate) fn suspension_intrinsic(
         self,
         store: &mut StoreOpaque,
@@ -3797,58 +3896,42 @@ impl Instance {
         yielding: bool,
         to_thread: SuspensionTarget,
     ) -> Result<WaitResult> {
-        let guest_thread = store.current_guest_thread()?;
-        if to_thread.is_none() {
-            let state = store.concurrent_state_mut()?;
-            if yielding {
-                // This is a `thread.yield` call
-                if !state.may_block(guest_thread.task)? {
-                    // In a non-blocking context, a `thread.yield` may trigger
-                    // other threads in the same component instance to run.
-                    if !state.promote_instance_local_thread_work_item(caller) {
-                        // No other threads are runnable, so just return
-                        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.
-                store.check_blocking()?;
-            }
-        }
-
         // There could be a pending cancellation from a previous uncancellable wait
         if cancellable && store.take_pending_cancellation()? {
             return Ok(WaitResult::Cancelled);
         }
 
-        match to_thread {
-            SuspensionTarget::SomeSuspended(thread) => {
-                self.resume_thread(store, caller, thread, true, false)?
+        let check_suspend = match to_thread {
+            SuspensionTarget::Promote(thread) => {
+                !self.resume_thread(store, caller, thread, ResumeThread::Promote)?
             }
-            SuspensionTarget::Some(thread) => {
-                self.resume_thread(store, caller, thread, true, true)?
+            SuspensionTarget::Resume(thread) => {
+                if !self.resume_thread(store, caller, thread, ResumeThread::Resume)? {
+                    bail_bug!("resumed thread should have been ready");
+                }
+                false
             }
-            SuspensionTarget::None => { /* nothing to do */ }
+            SuspensionTarget::None => true,
+        };
+
+        if check_suspend && !store.may_suspend(self.runtime_instance(caller))? {
+            return if yielding {
+                Ok(WaitResult::Completed)
+            } else {
+                Err(Trap::CannotBlockSyncTask.into())
+            };
         }
 
+        let guest_thread = store.current_guest_thread()?;
+
         let reason = if yielding {
             SuspendReason::Yielding {
                 thread: guest_thread,
                 cancellable,
-                // Tell `StoreOpaque::suspend` it's okay to suspend here since
-                // we're handling a `thread.yield-to-suspended` call; otherwise it would
-                // panic if we called it in a non-blocking context.
-                skip_may_block_check: to_thread.is_some(),
             }
         } else {
             SuspendReason::ExplicitlySuspending {
                 thread: guest_thread,
-                // Tell `StoreOpaque::suspend` it's okay to suspend here since
-                // we're handling a `thread.suspend-to(-suspended)` call; otherwise it would
-                // panic if we called it in a non-blocking context.
-                skip_may_block_check: to_thread.is_some(),
             }
         };
 
@@ -3865,6 +3948,7 @@ impl Instance {
     fn waitable_check(
         self,
         store: &mut StoreOpaque,
+        caller: RuntimeInstance,
         cancellable: bool,
         check: WaitableCheck,
         params: WaitableCheckParams,
@@ -3886,8 +3970,11 @@ impl Instance {
                     || (matches!(task.event, Some(Event::Cancelled)) && !cancellable))
                     && state.get_mut(set)?.ready.is_empty()
                 {
+                    store.trap_if_may_not_suspend(caller)?;
+
                     if cancellable {
-                        let old = state
+                        let old = store
+                            .concurrent_state_mut()?
                             .get_mut(guest_thread.thread)?
                             .wake_on_cancel
                             .replace(set);
@@ -3899,7 +3986,6 @@ impl Instance {
                     store.suspend(SuspendReason::Waiting {
                         set,
                         thread: guest_thread,
-                        skip_may_block_check: false,
                     })?;
                 }
             }
@@ -3959,13 +4045,6 @@ impl Instance {
         async_: bool,
         task_id: u32,
     ) -> Result<u32> {
-        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.check_blocking()?;
-        }
-
         let (rep, is_host) = store
             .instance_state(self.runtime_instance(caller_instance))
             .handle_table()
@@ -3977,7 +4056,7 @@ impl Instance {
         };
         let concurrent_state = store.concurrent_state_mut()?;
 
-        log::trace!("subtask_cancel {waitable:?} (handle {task_id})");
+        log::trace!("subtask_cancel {waitable:?} (handle {task_id}; async {async_})");
 
         if !async_ {
             waitable.trap_if_in_waitable_set(concurrent_state)?;
@@ -4034,7 +4113,7 @@ impl Instance {
                 // `Event::Cancelled` if it was already cancelled), but that's
                 // okay -- this should supersede the previous state.
                 task.event = Some(Event::Cancelled);
-                let runtime_instance = task.instance.index;
+                let runtime_instance = task.instance;
                 for thread in task.threads.clone() {
                     let thread = QualifiedThreadId {
                         task: guest_task,
@@ -4044,17 +4123,21 @@ impl Instance {
                     if let Some(set) = thread_mut.wake_on_cancel.take() {
                         // The thread is in a cancellable wait, so wake it up:
                         let item = match concurrent_state.get_mut(set)?.waiting.remove(&thread) {
-                            Some(WaitMode::Fiber(fiber)) => WorkItem::ResumeFiber(fiber),
-                            Some(WaitMode::Callback(instance)) => WorkItem::GuestCall(
-                                runtime_instance,
-                                GuestCall {
+                            Some(WaitMode::Fiber(fiber)) => WorkItem::ResumeFiber {
+                                instance: runtime_instance,
+                                thread,
+                                fiber,
+                            },
+                            Some(WaitMode::Callback(instance)) => WorkItem::GuestCall {
+                                instance: runtime_instance,
+                                call: GuestCall {
                                     thread,
                                     kind: GuestCallKind::DeliverEvent {
                                         instance,
                                         set: None,
                                     },
                                 },
-                            ),
+                            },
                             None => bail_bug!("thread not present in wake_on_cancel set"),
                         };
                         concurrent_state.push_high_priority(item);
@@ -4063,13 +4146,6 @@ impl Instance {
                         store.suspend(SuspendReason::Yielding {
                             thread: caller,
                             cancellable: false,
-                            // We've already checked that for a sync version of
-                            // this intrinsic we're allowed to block (start of
-                            // the function here), and otherwise this is similar
-                            // to `suspension_intrinsic` where we're doing a
-                            // brief yield to deliver the event, so there's no
-                            // need to check may-block again.
-                            skip_may_block_check: true,
                         })?;
                         break;
                     } else if let GuestThreadState::Ready {
@@ -4078,13 +4154,11 @@ impl Instance {
                     {
                         // The thread is in a cancellable yield, so yield back
                         // to it.
-                        concurrent_state.promote_thread_work_item(thread);
+                        concurrent_state.promote_thread_work_items(thread)?;
                         let caller = store.current_guest_thread()?;
                         store.suspend(SuspendReason::Yielding {
                             thread: caller,
                             cancellable: false,
-                            // See the comment above for why this is `true`
-                            skip_may_block_check: true,
                         })?;
                         break;
                     }
@@ -4112,7 +4186,7 @@ impl Instance {
             // Wait for this waitable to get signaled with its terminal status
             // from the completion callback enqueued by `first_poll`. Once
             // that's done fall through to the sahred
-            store.wait_for_event(waitable)?;
+            store.wait_for_event(self.runtime_instance(caller_instance), waitable)?;
 
             // .. fall through to determine what event's in store for us.
         }
@@ -4762,6 +4836,8 @@ pub struct GuestThread {
     instance_rep: Option<u32>,
     /// Scratch waitable set used to watch subtasks during synchronous calls.
     sync_call_set: TableId<WaitableSet>,
+    /// Whether this thread was explicitly created.
+    explicit: bool,
 }
 
 impl GuestThread {
@@ -4790,6 +4866,7 @@ impl GuestThread {
             state: GuestThreadState::NotStartedImplicit,
             instance_rep: None,
             sync_call_set,
+            explicit: false,
         })
     }
 
@@ -4811,6 +4888,7 @@ impl GuestThread {
             state: GuestThreadState::NotStartedExplicit(start_func),
             instance_rep: None,
             sync_call_set,
+            explicit: true,
         })
     }
 }
@@ -4893,9 +4971,12 @@ 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-typed
+    /// export.
+    async_typed: bool,
     /// Indicates whether this task was created for a call to an async-lifted
     /// export.
-    async_function: bool,
+    async_lifted: bool,
 
     decremented_interesting_task_count: bool,
 }
@@ -4941,7 +5022,8 @@ impl GuestTask {
         caller: Caller,
         callback: Option<CallbackFn>,
         instance: RuntimeInstance,
-        async_function: bool,
+        async_typed: bool,
+        async_lifted: bool,
     ) -> Result<QualifiedThreadId> {
         let host_future_state = match &caller {
             Caller::Guest { .. } => HostFutureState::NotApplicable,
@@ -4972,7 +5054,8 @@ impl GuestTask {
             exited: false,
             threads: HashSet::new(),
             host_future_state,
-            async_function,
+            async_typed,
+            async_lifted,
             decremented_interesting_task_count: false,
         })?;
         let new_thread = GuestThread::new_implicit(state, task)?;
@@ -5126,17 +5209,21 @@ impl Waitable {
                 assert!(wake_on_cancel.is_none() || wake_on_cancel == Some(set));
 
                 let item = match mode {
-                    WaitMode::Fiber(fiber) => WorkItem::ResumeFiber(fiber),
-                    WaitMode::Callback(instance) => WorkItem::GuestCall(
-                        state.get_mut(thread.task)?.instance.index,
-                        GuestCall {
+                    WaitMode::Fiber(fiber) => WorkItem::ResumeFiber {
+                        instance: state.get_mut(thread.task)?.instance,
+                        thread,
+                        fiber,
+                    },
+                    WaitMode::Callback(instance) => WorkItem::GuestCall {
+                        instance: state.get_mut(thread.task)?.instance,
+                        call: GuestCall {
                             thread,
                             kind: GuestCallKind::DeliverEvent {
                                 instance,
                                 set: Some(set),
                             },
                         },
-                    ),
+                    },
                 };
                 state.push_high_priority(item);
             }
@@ -5225,6 +5312,9 @@ pub struct ConcurrentInstanceState {
     backpressure: u16,
     /// Whether this instance can be entered
     do_not_enter: bool,
+    /// Whether this instance may suspend (i.e. whether this instance is
+    /// currently running a sync-typed function).
+    do_not_suspend: bool,
     /// Pending calls for this instance which require `Self::backpressure` to be
     /// `true` and/or `Self::do_not_enter` to be false before they can proceed.
     pending: BTreeMap<QualifiedThreadId, GuestCallKind>,
@@ -5361,6 +5451,9 @@ pub struct ConcurrentState {
     ///
     /// Used in the implementation of `Accessor::poll_ready_for_concurrent_call`.
     ready_for_concurrent_call_waker: Option<Waker>,
+
+    /// Whether the `StoreContextMut::poll_until` event loop is running.
+    event_loop_running: bool,
 }
 
 impl Default for ConcurrentState {
@@ -5378,6 +5471,7 @@ impl Default for ConcurrentState {
             interesting_tasks: 0,
             interesting_tasks_empty_waker: None,
             ready_for_concurrent_call_waker: None,
+            event_loop_running: false,
         }
     }
 }
@@ -5425,7 +5519,7 @@ impl ConcurrentState {
         }
 
         let mut handle_item = |item| match item {
-            WorkItem::ResumeFiber(fiber) => {
+            WorkItem::ResumeFiber { fiber, .. } => {
                 fibers.push(fiber);
             }
             WorkItem::PushFuture(future) => {
@@ -5435,8 +5529,9 @@ impl ConcurrentState {
                     .unwrap()
                     .push(future.into_inner());
             }
-            WorkItem::ResumeThread(..) | WorkItem::GuestCall(..) | WorkItem::WorkerFunction(..) => {
-            }
+            WorkItem::ResumeThread { .. }
+            | WorkItem::GuestCall { .. }
+            | WorkItem::WorkerFunction(_) => {}
         };
 
         for item in mem::take(&mut self.high_priority) {
@@ -5477,6 +5572,7 @@ impl ConcurrentState {
             interesting_tasks: _,
             interesting_tasks_empty_waker: _,
             ready_for_concurrent_call_waker: _,
+            event_loop_running: _,
         } = self;
 
         for entry in table.get_mut().iter_mut() {
@@ -5500,15 +5596,16 @@ impl ConcurrentState {
         }
 
         let mut handle_item = |item: &mut WorkItem| match item {
-            WorkItem::ResumeFiber(fiber) => {
+            WorkItem::ResumeFiber { fiber, .. } => {
                 fiber.trace_gc_roots(modules, unwind, gc_roots_list);
             }
             WorkItem::PushFuture(_future) => {
                 // TODO(cm-gc): once futures can contain GC roots, we will need
                 // to trace them.
             }
-            WorkItem::ResumeThread(..) | WorkItem::GuestCall(..) | WorkItem::WorkerFunction(..) => {
-            }
+            WorkItem::ResumeThread { .. }
+            | WorkItem::GuestCall { .. }
+            | WorkItem::WorkerFunction(_) => {}
         };
 
         for item in high_priority {
@@ -5582,59 +5679,81 @@ impl ConcurrentState {
         }
     }
 
-    fn promote_instance_local_thread_work_item(
+    fn promote_instance_local_thread_work_items(
         &mut self,
-        current_instance: RuntimeComponentInstanceIndex,
-    ) -> bool {
-        self.promote_work_items_matching(|item: &WorkItem| match item {
-            WorkItem::ResumeThread(instance, _) | WorkItem::GuestCall(instance, _) => {
-                *instance == current_instance
-            }
-            _ => false,
+        current_instance: RuntimeInstance,
+    ) -> Result<bool> {
+        log::trace!("promote thread work items for {current_instance:?}");
+        self.promote_work_items_matching(|state: &mut Self, item: &WorkItem| {
+            Ok(match item {
+                WorkItem::ResumeThread { instance, thread }
+                | WorkItem::ResumeFiber {
+                    instance, thread, ..
+                }
+                | WorkItem::GuestCall {
+                    instance,
+                    call: GuestCall { thread, .. },
+                } => {
+                    // According to the spec, only certain threads may be
+                    // resumed when an instance has a sync-typed task in
+                    // progress:
+                    *instance == current_instance
+                        && (state.get_mut(thread.thread)?.explicit || {
+                            let task = state.get_mut(thread.task)?;
+                            !task.async_typed || (task.async_lifted && task.callback.is_none())
+                        })
+                }
+                _ => false,
+            })
         })
     }
 
-    fn promote_thread_work_item(&mut self, thread: QualifiedThreadId) -> bool {
-        self.promote_work_items_matching(|item: &WorkItem| match item {
-            WorkItem::ResumeThread(_, t) | WorkItem::GuestCall(_, GuestCall { thread: t, .. }) => {
-                *t == thread
-            }
-            _ => false,
+    fn promote_thread_work_items(&mut self, thread: QualifiedThreadId) -> Result<bool> {
+        self.promote_work_items_matching(|_: &mut Self, item: &WorkItem| {
+            Ok(match item {
+                WorkItem::ResumeThread {
+                    thread: item_thread,
+                    ..
+                }
+                | WorkItem::GuestCall {
+                    call:
+                        GuestCall {
+                            thread: item_thread,
+                            ..
+                        },
+                    ..
+                } => *item_thread == thread,
+                _ => false,
+            })
         })
     }
 
-    fn promote_work_items_matching<F>(&mut self, mut predicate: F) -> bool
+    fn promote_work_items_matching<F>(&mut self, mut predicate: F) -> Result<bool>
     where
-        F: FnMut(&WorkItem) -> bool,
+        F: FnMut(&mut Self, &WorkItem) -> Result<bool>,
     {
-        // If there's a high-priority work item to resume the current guest thread,
-        // we don't need to promote anything, but we return true to indicate that
-        // work is pending for the current instance.
-        if self.high_priority.iter().any(&mut predicate) {
-            true
-        }
-        // Otherwise, look for a low-priority work item that matches the current
-        // instance and promote it to high-priority.
-        else if let Some(idx) = self.low_priority.iter().position(&mut predicate) {
-            let item = self.low_priority.remove(idx).unwrap();
-            self.push_high_priority(item);
-            true
-        } else {
-            false
+        for item in mem::take(&mut self.high_priority) {
+            if predicate(self, &item)? {
+                self.push_high_priority(item);
+            } else {
+                self.push_low_priority(item);
+            }
         }
-    }
 
-    fn check_blocking_for(&mut self, task: TableId<GuestTask>) -> Result<()> {
-        if self.may_block(task)? {
-            Ok(())
-        } else {
-            Err(Trap::CannotBlockSyncTask.into())
+        if self.high_priority.is_empty() {
+            // Note the use of `.rev()` here to preserve ordering given that
+            // items are popped from the back of `self.low_priority` by
+            // `poll_until` and pushed to the front by `push_low_priority`.
+            for item in mem::take(&mut self.low_priority).into_iter().rev() {
+                if predicate(self, &item)? {
+                    self.push_high_priority(item);
+                } else {
+                    self.push_low_priority(item);
+                }
+            }
         }
-    }
 
-    fn may_block(&mut self, task: TableId<GuestTask>) -> Result<bool> {
-        let task = self.get_mut(task)?;
-        Ok(task.async_function || task.returned_or_cancelled())
+        Ok(!self.high_priority.is_empty())
     }
 
     /// Used by `ResourceTables` to acquire the current `CallContext` for the
@@ -5842,7 +5961,8 @@ pub(crate) fn prepare_call<T, R>(
     let instance = handle.instance().id().get(store.0);
     let options = &instance.component().env_component().options[options];
     let ty = &instance.component().types()[ty];
-    let async_function = ty.async_;
+    let async_typed = ty.async_;
+    let async_lifted = raw_options.async_;
     let task_return_type = ty.results;
     let component_instance = raw_options.instance;
     let callback = options.callback.map(|i| instance.runtime_callback(i));
@@ -5887,7 +6007,8 @@ pub(crate) fn prepare_call<T, R>(
             }) as CallbackFn
         }),
         instance,
-        async_function,
+        async_typed,
+        async_lifted,
     )?;
 
     if !store.0.may_enter(instance)? {
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 4daa4392766a..5c7744a383c1 100644
--- a/crates/wasmtime/src/runtime/component/concurrent/futures_and_streams.rs
+++ b/crates/wasmtime/src/runtime/component/concurrent/futures_and_streams.rs
@@ -3502,13 +3502,6 @@ impl Instance {
     ) -> Result<ReturnCode> {
         let count = ItemCount::new(count)?;
 
-        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.check_blocking()?;
-        }
-
         let address = usize::try_from(address)?;
         self.check_bounds(store.0, options, ty, address, count.as_usize())?;
         let (rep, state) = self.id().get_mut(store.0).get_mut_by_index(ty, handle)?;
@@ -3723,7 +3716,7 @@ impl Instance {
         };
 
         if result == ReturnCode::Blocked && !self.options(store.0, options).async_ {
-            result = self.wait_for_write(store.0, transmit_handle)?;
+            result = self.wait_for_write(store.0, caller, transmit_handle)?;
         }
 
         if result != ReturnCode::Blocked {
@@ -3754,13 +3747,6 @@ impl Instance {
     ) -> Result<ReturnCode> {
         let count = ItemCount::new(count)?;
 
-        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.check_blocking()?;
-        }
-
         let address = usize::try_from(address)?;
         self.check_bounds(store.0, options, ty, address, count.as_usize())?;
         let (rep, state) = self.id().get_mut(store.0).get_mut_by_index(ty, handle)?;
@@ -3955,7 +3941,7 @@ impl Instance {
         };
 
         if result == ReturnCode::Blocked && !self.options(store.0, options).async_ {
-            result = self.wait_for_read(store.0, transmit_handle)?;
+            result = self.wait_for_read(store.0, caller_instance, transmit_handle)?;
         }
 
         if result != ReturnCode::Blocked {
@@ -3978,10 +3964,11 @@ impl Instance {
     fn wait_for_write(
         self,
         store: &mut StoreOpaque,
+        caller: RuntimeComponentInstanceIndex,
         handle: TableId<TransmitHandle>,
     ) -> Result<ReturnCode> {
         let waitable = Waitable::Transmit(handle);
-        store.wait_for_event(waitable)?;
+        store.wait_for_event(self.runtime_instance(caller), waitable)?;
         let event = waitable.take_event(store.concurrent_state_mut()?)?;
         if let Some(event @ (Event::StreamWrite { code, .. } | Event::FutureWrite { code, .. })) =
             event
@@ -3997,6 +3984,7 @@ impl Instance {
     fn cancel_write(
         self,
         store: &mut StoreOpaque,
+        caller: RuntimeComponentInstanceIndex,
         transmit_id: TableId<TransmitState>,
         async_: bool,
     ) -> Result<ReturnCode> {
@@ -4043,7 +4031,7 @@ impl Instance {
                     .concurrent_state_mut()?
                     .get_mut(transmit_id)?
                     .write_handle;
-                self.wait_for_write(store, handle)?
+                self.wait_for_write(store, caller, handle)?
             }
         } else {
             ReturnCode::Cancelled(ItemCount::ZERO)
@@ -4069,10 +4057,11 @@ impl Instance {
     fn wait_for_read(
         self,
         store: &mut StoreOpaque,
+        caller: RuntimeComponentInstanceIndex,
         handle: TableId<TransmitHandle>,
     ) -> Result<ReturnCode> {
         let waitable = Waitable::Transmit(handle);
-        store.wait_for_event(waitable)?;
+        store.wait_for_event(self.runtime_instance(caller), waitable)?;
         let event = waitable.take_event(store.concurrent_state_mut()?)?;
         if let Some(event @ (Event::StreamRead { code, .. } | Event::FutureRead { code, .. })) =
             event
@@ -4088,6 +4077,7 @@ impl Instance {
     fn cancel_read(
         self,
         store: &mut StoreOpaque,
+        caller: RuntimeComponentInstanceIndex,
         transmit_id: TableId<TransmitState>,
         async_: bool,
     ) -> Result<ReturnCode> {
@@ -4135,7 +4125,7 @@ impl Instance {
                     .concurrent_state_mut()?
                     .get_mut(transmit_id)?
                     .read_handle;
-                self.wait_for_read(store, handle)?
+                self.wait_for_read(store, caller, handle)?
             }
         } else {
             ReturnCode::Cancelled(ItemCount::ZERO)
@@ -4164,17 +4154,11 @@ impl Instance {
     fn guest_cancel_write(
         self,
         store: &mut StoreOpaque,
+        caller: RuntimeComponentInstanceIndex,
         ty: TransmitIndex,
         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.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);
@@ -4189,7 +4173,7 @@ impl Instance {
             TransmitLocalState::Busy => {}
         }
         let transmit_id = store.concurrent_state_mut()?.get_mut(id)?.state;
-        let code = self.cancel_write(store, transmit_id, async_)?;
+        let code = self.cancel_write(store, caller, transmit_id, async_)?;
         if !matches!(code, ReturnCode::Blocked) {
             let state =
                 get_mut_by_index_from(self.id().get_mut(store).table_for_transmit(ty), ty, writer)?
@@ -4205,17 +4189,11 @@ impl Instance {
     fn guest_cancel_read(
         self,
         store: &mut StoreOpaque,
+        caller: RuntimeComponentInstanceIndex,
         ty: TransmitIndex,
         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.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);
@@ -4230,7 +4208,7 @@ impl Instance {
             TransmitLocalState::Busy => {}
         }
         let transmit_id = store.concurrent_state_mut()?.get_mut(id)?.state;
-        let code = self.cancel_read(store, transmit_id, async_)?;
+        let code = self.cancel_read(store, caller, transmit_id, async_)?;
         if !matches!(code, ReturnCode::Blocked) {
             let state =
                 get_mut_by_index_from(self.id().get_mut(store).table_for_transmit(ty), ty, reader)?
@@ -4352,11 +4330,12 @@ impl Instance {
     pub(crate) fn future_cancel_read(
         self,
         store: &mut StoreOpaque,
+        caller: RuntimeComponentInstanceIndex,
         ty: TypeFutureTableIndex,
         async_: bool,
         reader: u32,
     ) -> Result<u32> {
-        self.guest_cancel_read(store, TransmitIndex::Future(ty), async_, reader)
+        self.guest_cancel_read(store, caller, TransmitIndex::Future(ty), async_, reader)
             .map(|v| v.encode())
     }
 
@@ -4364,11 +4343,12 @@ impl Instance {
     pub(crate) fn future_cancel_write(
         self,
         store: &mut StoreOpaque,
+        caller: RuntimeComponentInstanceIndex,
         ty: TypeFutureTableIndex,
         async_: bool,
         writer: u32,
     ) -> Result<u32> {
-        self.guest_cancel_write(store, TransmitIndex::Future(ty), async_, writer)
+        self.guest_cancel_write(store, caller, TransmitIndex::Future(ty), async_, writer)
             .map(|v| v.encode())
     }
 
@@ -4376,11 +4356,12 @@ impl Instance {
     pub(crate) fn stream_cancel_read(
         self,
         store: &mut StoreOpaque,
+        caller: RuntimeComponentInstanceIndex,
         ty: TypeStreamTableIndex,
         async_: bool,
         reader: u32,
     ) -> Result<u32> {
-        self.guest_cancel_read(store, TransmitIndex::Stream(ty), async_, reader)
+        self.guest_cancel_read(store, caller, TransmitIndex::Stream(ty), async_, reader)
             .map(|v| v.encode())
     }
 
@@ -4388,11 +4369,12 @@ impl Instance {
     pub(crate) fn stream_cancel_write(
         self,
         store: &mut StoreOpaque,
+        caller: RuntimeComponentInstanceIndex,
         ty: TypeStreamTableIndex,
         async_: bool,
         writer: u32,
     ) -> Result<u32> {
-        self.guest_cancel_write(store, TransmitIndex::Stream(ty), async_, writer)
+        self.guest_cancel_write(store, caller, TransmitIndex::Stream(ty), async_, writer)
             .map(|v| v.encode())
     }
 
diff --git a/crates/wasmtime/src/runtime/component/concurrent_disabled.rs b/crates/wasmtime/src/runtime/component/concurrent_disabled.rs
index 2d3af7bf5197..98cc17c21251 100644
--- a/crates/wasmtime/src/runtime/component/concurrent_disabled.rs
+++ b/crates/wasmtime/src/runtime/component/concurrent_disabled.rs
@@ -170,10 +170,6 @@ impl StoreOpaque {
         Ok(self.exit_call_not_concurrent())
     }
 
-    pub(crate) fn check_blocking(&mut self) -> crate::Result<()> {
-        Ok(())
-    }
-
     pub(crate) fn may_enter(&mut self, _instance: RuntimeInstance) -> Result<bool> {
         Ok(!self.trapped())
     }
diff --git a/crates/wasmtime/src/runtime/component/func/host.rs b/crates/wasmtime/src/runtime/component/func/host.rs
index 3f38679e840a..f5064e576a71 100644
--- a/crates/wasmtime/src/runtime/component/func/host.rs
+++ b/crates/wasmtime/src/runtime/component/func/host.rs
@@ -283,10 +283,6 @@ where
     T: 'static,
     R: Send + Sync + 'static,
 {
-    /// Whether or not this is `async` function from the perspective of the
-    /// component model.
-    const ASYNC: bool;
-
     /// Performs a type-check to ensure that this host function can be imported
     /// with the provided signature that a component is using.
     fn typecheck(ty: TypeFuncIndex, types: &InstanceType<'_>) -> Result<()>;
@@ -363,13 +359,6 @@ where
         let vminstance = instance.id().get(store.0);
         let async_ = vminstance.component().env_component().options[options].async_;
 
-        // If this is a synchronous-lower of a host-async function, then the
-        // guest is blocking. Test, in the context of the guest task, if that's
-        // allowed.
-        if !async_ && Self::ASYNC {
-            store.0.check_blocking()?;
-        }
-
         if async_ {
             #[cfg(feature = "component-model-async")]
             {
@@ -631,8 +620,6 @@ where
     P: ComponentNamedList + Lift + 'static,
     R: ComponentNamedList + Lower + 'static,
 {
-    const ASYNC: bool = ASYNC;
-
     fn typecheck(ty: TypeFuncIndex, types: &InstanceType<'_>) -> Result<()> {
         let ty = &types.types[ty];
         typecheck_async(ASYNC, ty.async_)?;
@@ -709,8 +696,6 @@ where
     T: 'static,
     F: Fn(StoreContextMut<'_, T>, ComponentFunc, Vec<Val>, usize) -> HostResult<Vec<Val>>,
 {
-    const ASYNC: bool = ASYNC;
-
     /// 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.
diff --git a/crates/wasmtime/src/runtime/component/instance.rs b/crates/wasmtime/src/runtime/component/instance.rs
index 453b6dcf57e7..f8b827c4242d 100644
--- a/crates/wasmtime/src/runtime/component/instance.rs
+++ b/crates/wasmtime/src/runtime/component/instance.rs
@@ -10,13 +10,17 @@ use crate::instance::OwnedImports;
 use crate::linker::DefinitionType;
 use crate::prelude::*;
 use crate::runtime::vm::component::{ComponentInstance, TypedResource, TypedResourceIndex};
-use crate::runtime::vm::{self, VMFuncRef};
+use crate::runtime::vm::{self, VMFuncRef, VMStore};
+#[cfg(feature = "component-model-async")]
+use crate::store::StoreToken;
 use crate::store::{AsStoreOpaque, Asyncness, StoreOpaque};
 use crate::{AsContext, AsContextMut, Engine, Module, StoreContextMut};
 use alloc::sync::Arc;
 use core::marker;
 use core::pin::Pin;
 use core::ptr::NonNull;
+#[cfg(feature = "component-model-async")]
+use futures::channel::oneshot;
 use wasmtime_environ::{EngineOrModuleTypeIndex, component::*};
 use wasmtime_environ::{EntityIndex, EntityType, PrimaryMap};
 
@@ -611,9 +615,6 @@ pub(crate) fn lookup_vmdef(
             // within that store, so it's safe to create a `Func`.
             vm::Export::Function(unsafe { crate::Func::from_vm_func_ref(store.id(), funcref) })
         }
-        CoreDef::TaskMayBlock => vm::Export::Global(crate::Global::from_task_may_block(
-            StoreComponentInstanceId::new(store.id(), id),
-        )),
     }
 }
 
@@ -841,16 +842,74 @@ impl<'a> Instantiator<'a> {
                     // already checked for asyncness and are running on a fiber
                     // if required.
 
-                    let i = unsafe {
-                        crate::Instance::new_started(store, module, imports.as_ref(), asyncness)
+                    let mut instance = {
+                        let (mut limiter, store) = store.0.resource_limiter_and_store_opaque();
+                        unsafe {
+                            crate::Instance::new_raw(
+                                store,
+                                limiter.as_mut(),
+                                module,
+                                imports.as_ref(),
+                            )
                             .await?
+                        }
                     };
 
+                    if instance.id.get_mut(store.0).needs_startup() {
+                        if asyncness == Asyncness::No {
+                            instance.start_raw(store)?;
+                        } else {
+                            #[cfg(feature = "async")]
+                            {
+                                #[cfg(feature = "component-model-async")]
+                                {
+                                    if store.0.concurrency_support() {
+                                        // With concurrency support enabled, we must
+                                        // run the start function inside the store's
+                                        // event loop in case it calls async
+                                        // functions or intrinsics, creates and
+                                        // resumes threads, etc.
+                                        let (tx, rx) = oneshot::channel();
+                                        let token = StoreToken::new(store.as_context_mut());
+                                        store.0.queue_task(move |store| {
+                                            _ = tx.send(
+                                                instance
+                                                    .start_raw(&mut token.as_context_mut(store))
+                                                    .map(|()| instance),
+                                            );
+                                            Ok(())
+                                        })?;
+                                        instance = store
+                                            .as_context_mut()
+                                            .run_concurrent_trap_on_idle(async |_| {
+                                                rx.await.map_err(|_| {
+                                                    format_err!("oneshot channel canceled")
+                                                })
+                                            })
+                                            .await???;
+                                    } else {
+                                        store.on_fiber(|store| instance.start_raw(store)).await??;
+                                    }
+                                }
+                                #[cfg(not(feature = "component-model-async"))]
+                                {
+                                    _ = &mut instance;
+                                    store.on_fiber(|store| instance.start_raw(store)).await??;
+                                }
+                            }
+                            #[cfg(not(feature = "async"))]
+                            {
+                                _ = &mut instance;
+                                unreachable!();
+                            }
+                        }
+                    }
+
                     if exit {
                         store.0.exit_guest_sync_call()?;
                     }
 
-                    self.instance_mut(store.0).push_instance_id(i.id())?;
+                    self.instance_mut(store.0).push_instance_id(instance.id())?;
                 }
 
                 GlobalInitializer::LowerImport { import, index } => {
diff --git a/crates/wasmtime/src/runtime/externals/global.rs b/crates/wasmtime/src/runtime/externals/global.rs
index ede512da35d3..f3bb40ffd734 100644
--- a/crates/wasmtime/src/runtime/externals/global.rs
+++ b/crates/wasmtime/src/runtime/externals/global.rs
@@ -349,17 +349,6 @@ impl Global {
         }
     }
 
-    #[cfg(feature = "component-model")]
-    pub(crate) fn from_task_may_block(
-        instance: crate::component::store::StoreComponentInstanceId,
-    ) -> Global {
-        Global {
-            store: instance.store_id(),
-            instance: instance.instance().as_u32(),
-            kind: VMGlobalKind::TaskMayBlock,
-        }
-    }
-
     pub(crate) fn wasmtime_ty<'a>(&self, store: &'a StoreOpaque) -> &'a wasmtime_environ::Global {
         self.store.assert_belongs_to(store.id());
         match self.kind {
@@ -371,7 +360,7 @@ impl Global {
             }
             VMGlobalKind::Host(index) => unsafe { &store.host_globals()[index].get().as_ref().ty },
             #[cfg(feature = "component-model")]
-            VMGlobalKind::ComponentFlags(_) | VMGlobalKind::TaskMayBlock => {
+            VMGlobalKind::ComponentFlags(_) => {
                 const TY: wasmtime_environ::Global = wasmtime_environ::Global {
                     mutability: true,
                     wasm_ty: wasmtime_environ::WasmValType::I32,
@@ -389,7 +378,7 @@ impl Global {
             }
             VMGlobalKind::Host(_) => None,
             #[cfg(feature = "component-model")]
-            VMGlobalKind::ComponentFlags(_) | VMGlobalKind::TaskMayBlock => {
+            VMGlobalKind::ComponentFlags(_) => {
                 let instance = crate::component::ComponentInstanceId::from_u32(self.instance);
                 Some(
                     VMOpaqueContext::from_vmcomponent(store.component_instance(instance).vmctx())
@@ -421,8 +410,6 @@ impl Global {
             VMGlobalKind::ComponentFlags(idx) => {
                 u64::from(self.instance) << 32 | u64::from(idx.as_u32())
             }
-            #[cfg(feature = "component-model")]
-            VMGlobalKind::TaskMayBlock => u64::from(self.instance) << 32 | u64::from(u32::MAX),
         }
     }
 
@@ -454,12 +441,6 @@ impl Global {
                     .instance_flags(index)
                     .as_raw()
             }
-            #[cfg(feature = "component-model")]
-            VMGlobalKind::TaskMayBlock => store
-                .component_instance(crate::component::ComponentInstanceId::from_u32(
-                    self.instance,
-                ))
-                .task_may_block(),
         }
     }
 }
diff --git a/crates/wasmtime/src/runtime/fiber.rs b/crates/wasmtime/src/runtime/fiber.rs
index 50f086f7d4c8..89c5d49b8258 100644
--- a/crates/wasmtime/src/runtime/fiber.rs
+++ b/crates/wasmtime/src/runtime/fiber.rs
@@ -888,19 +888,6 @@ where
     })
 }
 
-/// Safe wrapper around [`make_fiber_unchecked`] which requires that `S` is
-/// `Send`.
-#[cfg(feature = "component-model-async")]
-pub(crate) fn make_fiber<'a, S>(
-    store: &mut S,
-    fun: impl FnOnce(&mut S) -> Result<()> + Send + Sync + 'a,
-) -> Result<StoreFiber<'a>>
-where
-    S: AsStoreOpaque + Send + ?Sized + 'a,
-{
-    unsafe { make_fiber_unchecked(store, fun) }
-}
-
 /// Run the specified function on a newly-created fiber and `.await` its
 /// completion.
 pub(crate) async fn on_fiber<S, R>(
diff --git a/crates/wasmtime/src/runtime/instance.rs b/crates/wasmtime/src/runtime/instance.rs
index 53fd54faed9b..b46bfa04c68d 100644
--- a/crates/wasmtime/src/runtime/instance.rs
+++ b/crates/wasmtime/src/runtime/instance.rs
@@ -297,7 +297,7 @@ impl Instance {
     /// This method is unsafe because it does not type-check the `imports`
     /// provided. The `imports` provided must be suitable for the module
     /// provided as well.
-    async unsafe fn new_raw(
+    pub(crate) async unsafe fn new_raw(
         store: &mut StoreOpaque,
         mut limiter: Option<&mut StoreResourceLimiter<'_>>,
         module: &Module,
@@ -349,7 +349,7 @@ impl Instance {
         }
     }
 
-    fn start_raw<T>(&self, store: &mut StoreContextMut<'_, T>) -> Result<()> {
+    pub(crate) fn start_raw<T>(&self, store: &mut StoreContextMut<'_, T>) -> Result<()> {
         // If a start function is present, invoke it. Make sure we use all the
         // trap-handling configuration in `store` as well.
         let store_id = store.0.id();
diff --git a/crates/wasmtime/src/runtime/vm/component.rs b/crates/wasmtime/src/runtime/vm/component.rs
index bafaada3d714..e1599ae34f72 100644
--- a/crates/wasmtime/src/runtime/vm/component.rs
+++ b/crates/wasmtime/src/runtime/vm/component.rs
@@ -961,20 +961,6 @@ impl ComponentInstance {
             )
         }
     }
-
-    pub(crate) fn task_may_block(&self) -> NonNull<VMGlobalDefinition> {
-        unsafe { self.vmctx_plus_offset_raw::<VMGlobalDefinition>(self.offsets.task_may_block()) }
-    }
-
-    #[cfg(feature = "component-model-async")]
-    pub(crate) fn get_task_may_block(&self) -> bool {
-        unsafe { *self.task_may_block().as_ref().as_i32() != 0 }
-    }
-
-    #[cfg(feature = "component-model-async")]
-    pub(crate) fn set_task_may_block(self: Pin<&mut Self>, val: bool) {
-        unsafe { *self.task_may_block().as_mut().as_i32_mut() = if val { 1 } else { 0 } }
-    }
 }
 
 // SAFETY: `layout` should describe this accurately and `OwnedVMContext` is the
diff --git a/crates/wasmtime/src/runtime/vm/component/libcalls.rs b/crates/wasmtime/src/runtime/vm/component/libcalls.rs
index 11ec01108492..18b881a2e27c 100644
--- a/crates/wasmtime/src/runtime/vm/component/libcalls.rs
+++ b/crates/wasmtime/src/runtime/vm/component/libcalls.rs
@@ -1,11 +1,13 @@
 //! Implementation of string transcoding required by the component model.
 
+#[cfg(feature = "component-model-async")]
+use crate::bail_bug;
 use crate::component::Instance;
 #[cfg(feature = "component-model-async")]
 use crate::component::concurrent::WaitResult;
 use crate::prelude::*;
 #[cfg(feature = "component-model-async")]
-use crate::runtime::component::concurrent::{ResourcePair, SuspensionTarget};
+use crate::runtime::component::concurrent::{ResourcePair, ResumeThread, SuspensionTarget};
 use crate::runtime::vm::component::{ComponentInstance, VMComponentContext};
 use crate::runtime::vm::{HostResultHasUnwindSentinel, VMStore, VmSafe};
 use core::cell::Cell;
@@ -1017,13 +1019,14 @@ fn future_read(
 fn future_cancel_write(
     store: &mut dyn VMStore,
     instance: Instance,
-    _caller_instance: u32,
+    caller_instance: u32,
     ty: u32,
     async_: u8,
     writer: u32,
 ) -> Result<u32> {
     instance.future_cancel_write(
         store,
+        RuntimeComponentInstanceIndex::from_u32(caller_instance),
         TypeFutureTableIndex::from_u32(ty),
         async_ != 0,
         writer,
@@ -1034,13 +1037,14 @@ fn future_cancel_write(
 fn future_cancel_read(
     store: &mut dyn VMStore,
     instance: Instance,
-    _caller_instance: u32,
+    caller_instance: u32,
     ty: u32,
     async_: u8,
     reader: u32,
 ) -> Result<u32> {
     instance.future_cancel_read(
         store,
+        RuntimeComponentInstanceIndex::from_u32(caller_instance),
         TypeFutureTableIndex::from_u32(ty),
         async_ != 0,
         reader,
@@ -1131,13 +1135,14 @@ fn stream_read(
 fn stream_cancel_write(
     store: &mut dyn VMStore,
     instance: Instance,
-    _caller_instance: u32,
+    caller_instance: u32,
     ty: u32,
     async_: u8,
     writer: u32,
 ) -> Result<u32> {
     instance.stream_cancel_write(
         store,
+        RuntimeComponentInstanceIndex::from_u32(caller_instance),
         TypeStreamTableIndex::from_u32(ty),
         async_ != 0,
         writer,
@@ -1148,13 +1153,14 @@ fn stream_cancel_write(
 fn stream_cancel_read(
     store: &mut dyn VMStore,
     instance: Instance,
-    _caller_instance: u32,
+    caller_instance: u32,
     ty: u32,
     async_: u8,
     reader: u32,
 ) -> Result<u32> {
     instance.stream_cancel_read(
         store,
+        RuntimeComponentInstanceIndex::from_u32(caller_instance),
         TypeStreamTableIndex::from_u32(ty),
         async_ != 0,
         reader,
@@ -1324,13 +1330,16 @@ fn thread_resume_later(
     caller_instance: u32,
     thread_idx: u32,
 ) -> Result<()> {
-    instance.resume_thread(
+    if !instance.resume_thread(
         store,
         RuntimeComponentInstanceIndex::from_u32(caller_instance),
         thread_idx,
-        false,
-        false,
-    )
+        ResumeThread::ResumeLater,
+    )? {
+        bail_bug!("resumed thread should have been ready");
+    }
+
+    Ok(())
 }
 
 #[cfg(feature = "component-model-async")]
@@ -1383,7 +1392,7 @@ fn thread_suspend_then_resume(
             RuntimeComponentInstanceIndex::from_u32(caller),
             cancellable != 0,
             false,
-            SuspensionTarget::SomeSuspended(thread_idx),
+            SuspensionTarget::Resume(thread_idx),
         )
         .map(|r| r == WaitResult::Cancelled)
 }
@@ -1402,7 +1411,7 @@ fn thread_yield_then_resume(
             RuntimeComponentInstanceIndex::from_u32(caller_instance),
             cancellable != 0,
             true,
-            SuspensionTarget::SomeSuspended(thread_idx),
+            SuspensionTarget::Resume(thread_idx),
         )
         .map(|r| r == WaitResult::Cancelled)
 }
@@ -1421,7 +1430,7 @@ fn thread_suspend_then_promote(
             RuntimeComponentInstanceIndex::from_u32(caller),
             cancellable != 0,
             false,
-            SuspensionTarget::Some(thread_idx),
+            SuspensionTarget::Promote(thread_idx),
         )
         .map(|r| r == WaitResult::Cancelled)
 }
@@ -1440,7 +1449,7 @@ fn thread_yield_then_promote(
             RuntimeComponentInstanceIndex::from_u32(caller),
             cancellable != 0,
             true,
-            SuspensionTarget::Some(thread_idx),
+            SuspensionTarget::Promote(thread_idx),
         )
         .map(|r| r == WaitResult::Cancelled)
 }
diff --git a/crates/wasmtime/src/runtime/vm/instance.rs b/crates/wasmtime/src/runtime/vm/instance.rs
index 321f140def87..5be8d457ec74 100644
--- a/crates/wasmtime/src/runtime/vm/instance.rs
+++ b/crates/wasmtime/src/runtime/vm/instance.rs
@@ -744,20 +744,6 @@ impl Instance {
                     index,
                 )
             }
-            #[cfg(feature = "component-model")]
-            VMGlobalKind::TaskMayBlock => {
-                // SAFETY: validity of this `&Instance` means validity of its
-                // imports meaning we can read the id of the vmctx within.
-                let id = unsafe {
-                    let vmctx = super::component::VMComponentContext::from_opaque(
-                        import.vmctx.unwrap().as_non_null(),
-                    );
-                    super::component::ComponentInstance::vmctx_instance_id(vmctx)
-                };
-                crate::Global::from_task_may_block(
-                    crate::component::store::StoreComponentInstanceId::new(store, id),
-                )
-            }
         }
     }
 
diff --git a/crates/wasmtime/src/runtime/vm/vmcontext.rs b/crates/wasmtime/src/runtime/vm/vmcontext.rs
index de5b079425ee..7d02bfb98bbf 100644
--- a/crates/wasmtime/src/runtime/vm/vmcontext.rs
+++ b/crates/wasmtime/src/runtime/vm/vmcontext.rs
@@ -183,8 +183,6 @@ pub enum VMGlobalKind {
     /// Flags for a component instance, stored in `VMComponentContext`.
     #[cfg(feature = "component-model")]
     ComponentFlags(wasmtime_environ::component::RuntimeComponentInstanceIndex),
-    #[cfg(feature = "component-model")]
-    TaskMayBlock,
 }
 
 // SAFETY: the above enum is repr(C) and stores nothing else
diff --git a/tests/component-model b/tests/component-model
index 73b7ad51d3b5..a8d87b22e791 160000
--- a/tests/component-model
+++ b/tests/component-model
@@ -1 +1 @@
-Subproject commit 73b7ad51d3b5d6f1ef53c923d8c585e28b242bcc
+Subproject commit a8d87b22e791fb6730a5a9449bd57d5ba89bf3b1
diff --git a/tests/disas/component-model/sync-adapter-calls-x64.wat b/tests/disas/component-model/sync-adapter-calls-x64.wat
index bed2b5db1987..49e9ec05debd 100644
--- a/tests/disas/component-model/sync-adapter-calls-x64.wat
+++ b/tests/disas/component-model/sync-adapter-calls-x64.wat
@@ -58,39 +58,35 @@
 ;;       movq    0x18(%r10), %r10
 ;;       addq    $0x20, %r10
 ;;       cmpq    %rsp, %r10
-;;       ja      0xe6
+;;       ja      0xd2
 ;;   39: subq    $0x20, %rsp
-;;       movq    0x48(%rdi), %rdi
-;;       movq    0xe8(%rdi), %rax
+;;       movq    0x48(%rdi), %rdx
+;;       movq    0xe8(%rdx), %rax
 ;;       movl    (%rax), %ecx
 ;;       testl   %ecx, %ecx
-;;       je      0xe8
-;;   52: movq    0x100(%rdi), %rdx
-;;       movl    (%rdx), %esi
-;;       movl    $0, (%rdx)
-;;       movq    8(%rdi), %rdi
-;;       movq    0x88(%rdi), %r8
-;;       leaq    (%rsp), %r10
-;;       movq    %r8, (%rsp)
+;;       je      0xd4
+;;   52: movq    8(%rdx), %rdx
+;;       movq    0x88(%rdx), %rsi
+;;       leaq    (%rsp), %r8
+;;       movq    %rsi, (%rsp)
 ;;       movl    $2, 8(%rsp)
 ;;       movl    $0, 0xc(%rsp)
 ;;       movl    $1, 0x10(%rsp)
-;;       movl    0x80(%rdi), %r9d
-;;       movl    %r9d, 0x14(%rsp)
-;;       movl    $0, 0x80(%rdi)
-;;       movl    0x84(%rdi), %r11d
-;;       movl    %r11d, 0x18(%rsp)
-;;       movl    $0, 0x84(%rdi)
-;;       movq    %r10, 0x88(%rdi)
-;;       movq    %r8, 0x88(%rdi)
-;;       movl    %r9d, 0x80(%rdi)
-;;       movl    %r11d, 0x84(%rdi)
+;;       movl    0x80(%rdx), %edi
+;;       movl    %edi, 0x14(%rsp)
+;;       movl    $0, 0x80(%rdx)
+;;       movl    0x84(%rdx), %r9d
+;;       movl    %r9d, 0x18(%rsp)
+;;       movl    $0, 0x84(%rdx)
+;;       movq    %r8, 0x88(%rdx)
+;;       movq    %rsi, 0x88(%rdx)
+;;       movl    %edi, 0x80(%rdx)
+;;       movl    %r9d, 0x84(%rdx)
 ;;       movl    %ecx, (%rax)
-;;       movl    %esi, (%rdx)
 ;;       movl    $0x4fc, %eax
 ;;       addq    $0x20, %rsp
 ;;       movq    %rbp, %rsp
 ;;       popq    %rbp
 ;;       retq
-;;   e6: ud2
-;;   e8: ud2
+;;   d2: ud2
+;;   d4: ud2
diff --git a/tests/disas/component-model/sync-adapter-calls.wat b/tests/disas/component-model/sync-adapter-calls.wat
index b89550743fdf..c12ad4fdd31d 100644
--- a/tests/disas/component-model/sync-adapter-calls.wat
+++ b/tests/disas/component-model/sync-adapter-calls.wat
@@ -99,28 +99,25 @@
 ;;                                     jump block9
 ;;
 ;;                                 block9:
-;;                                     v11 = load.i64 notrap aligned readonly can_move region3 v3+256
-;;                                     v12 = load.i32 notrap aligned region4 v11
+;;                                     v16 = load.i64 notrap aligned readonly can_move region0 v3+8
+;;                                     v17 = load.i64 notrap aligned region5 v16+136
+;;                                     v15 = stack_addr.i64 ss0
+;;                                     store notrap aligned region6 v17, v15
+;;                                     v11 = iconst.i32 2
+;;                                     store notrap aligned region7 v11, v15+8  ; v11 = 2
 ;;                                     v8 = iconst.i32 0
-;;                                     store notrap aligned region4 v8, v11  ; v8 = 0
-;;                                     v20 = load.i64 notrap aligned readonly can_move region0 v3+8
-;;                                     v21 = load.i64 notrap aligned region5 v20+136
-;;                                     v19 = stack_addr.i64 ss0
-;;                                     store notrap aligned region6 v21, v19
-;;                                     v15 = iconst.i32 2
-;;                                     store notrap aligned region7 v15, v19+8  ; v15 = 2
-;;                                     store notrap aligned region8 v8, v19+12  ; v8 = 0
-;;                                     v17 = iconst.i32 1
-;;                                     store notrap aligned region9 v17, v19+16  ; v17 = 1
-;;                                     v22 = load.i32 notrap aligned region10 v20+128
-;;                                     store notrap aligned region11 v22, v19+20
-;;                                     store notrap aligned region10 v8, v20+128  ; v8 = 0
-;;                                     v24 = load.i32 notrap aligned region12 v20+132
-;;                                     store notrap aligned region13 v24, v19+24
-;;                                     store notrap aligned region12 v8, v20+132  ; v8 = 0
-;;                                     store notrap aligned region5 v19, v20+136
-;;                                     v26 = load.i64 notrap aligned readonly can_move region3 v3+208
-;;                                     v27 = load.i32 notrap aligned region4 v26
+;;                                     store notrap aligned region8 v8, v15+12  ; v8 = 0
+;;                                     v13 = iconst.i32 1
+;;                                     store notrap aligned region9 v13, v15+16  ; v13 = 1
+;;                                     v18 = load.i32 notrap aligned region10 v16+128
+;;                                     store notrap aligned region11 v18, v15+20
+;;                                     store notrap aligned region10 v8, v16+128  ; v8 = 0
+;;                                     v20 = load.i32 notrap aligned region12 v16+132
+;;                                     store notrap aligned region13 v20, v15+24
+;;                                     store notrap aligned region12 v8, v16+132  ; v8 = 0
+;;                                     store notrap aligned region5 v15, v16+136
+;;                                     v22 = load.i64 notrap aligned readonly can_move region3 v3+208
+;;                                     v23 = load.i32 notrap aligned region4 v22
 ;;                                     jump block16
 ;;
 ;;                                 block16:
@@ -133,14 +130,13 @@
 ;;                                     jump block12
 ;;
 ;;                                 block12:
-;;                                     store.i64 notrap aligned region5 v21, v20+136
-;;                                     store.i32 notrap aligned region10 v22, v20+128
-;;                                     store.i32 notrap aligned region12 v24, v20+132
+;;                                     store.i64 notrap aligned region5 v17, v16+136
+;;                                     store.i32 notrap aligned region10 v18, v16+128
+;;                                     store.i32 notrap aligned region12 v20, v16+132
 ;;                                     jump block14
 ;;
 ;;                                 block14:
 ;;                                     store.i32 notrap aligned region4 v10, v9
-;;                                     store.i32 notrap aligned region4 v12, v11
 ;;                                     jump block7
 ;;
 ;;                                 block7:
@@ -156,6 +152,6 @@
 ;; @00f0                               jump block1
 ;;
 ;;                                 block1:
-;;                                     v50 = iconst.i32 1276
-;; @00f0                               return v50  ; v50 = 1276
+;;                                     v45 = iconst.i32 1276
+;; @00f0                               return v45  ; v45 = 1276
 ;; }
diff --git a/tests/misc_testsuite/component-model/async/callback-yield-then-exit.wast b/tests/misc_testsuite/component-model/async/callback-yield-then-exit.wast
index c3648c07e559..528aa22bfbeb 100644
--- a/tests/misc_testsuite/component-model/async/callback-yield-then-exit.wast
+++ b/tests/misc_testsuite/component-model/async/callback-yield-then-exit.wast
@@ -47,5 +47,4 @@
   (func (export "run") (alias export $B "run"))
 )
 
-(assert_trap (invoke "run") "wasm trap: cannot block a synchronous task before returning")
-(assert_trap (invoke "run") "wasm trap: cannot enter component instance")
+(assert_return (invoke "run"))
diff --git a/tests/misc_testsuite/component-model/async/fused.wast b/tests/misc_testsuite/component-model/async/fused.wast
index d7b644f594ab..fad18e8b86a5 100644
--- a/tests/misc_testsuite/component-model/async/fused.wast
+++ b/tests/misc_testsuite/component-model/async/fused.wast
@@ -145,4 +145,4 @@
   (func (export "run") (alias export $lowerer "run"))
 )
 
-(assert_trap (invoke "run") "wasm trap: cannot block a synchronous task before returning")
+(assert_return (invoke "run"))
diff --git a/tests/misc_testsuite/component-model/async/future-read.wast b/tests/misc_testsuite/component-model/async/future-read.wast
index 498d2893a513..0f289481a7d3 100644
--- a/tests/misc_testsuite/component-model/async/future-read.wast
+++ b/tests/misc_testsuite/component-model/async/future-read.wast
@@ -122,7 +122,7 @@
   (func (export "run") (alias export $other-child "run"))
 )
 
-(assert_trap (invoke "run") "wasm trap: cannot block a synchronous task before returning")
+(assert_return (invoke "run"))
 
 ;; synchronous future.read; async lift
 (component
diff --git a/tests/misc_testsuite/component-model/async/stackful.wast b/tests/misc_testsuite/component-model/async/stackful.wast
index 30cca1d32ced..6c7712f5fdbb 100644
--- a/tests/misc_testsuite/component-model/async/stackful.wast
+++ b/tests/misc_testsuite/component-model/async/stackful.wast
@@ -104,7 +104,7 @@
   (func (export "run") (alias export $lowerer "run"))
 )
 
-(assert_trap (invoke "run") "wasm trap: cannot block a synchronous task before returning")
+(assert_return (invoke "run"))
 
 ;; waitable-set.wait
 (component