diff --git a/README.md b/README.md
index 9b9df840..5acc1a54 100644
--- a/README.md
+++ b/README.md
@@ -2,10 +2,10 @@
 
 This repository is where the component model is being standardized. For a more user-focused explanation, take a look at the **[Component Model Documentation]**.
 
-This repository describes the high-level [goals], [use cases], [design choices]
-and [FAQ] of the component model as well as a more-detailed [assembly-level explainer], [IDL],
-[binary format] and [ABI] covering the initial Minimum Viable Product (MVP)
-release.
+This repository contains the high-level [goals], [use cases], [design choices]
+and [FAQ] of the component model as well as a more-detailed, low-level
+[AST explainer], [async explainer], [IDL], [binary format] and [ABI] covering
+the initial Minimum Viable Product (MVP) release.
 
 In the future, this repository will additionally contain a [formal spec],
 reference interpreter and test suite.
@@ -27,7 +27,8 @@ To contribute to any of these repositories, see the Community Group's
 [use cases]: design/high-level/UseCases.md
 [design choices]: design/high-level/Choices.md
 [FAQ]: design/high-level/FAQ.md
-[assembly-level explainer]: design/mvp/Explainer.md
+[AST explainer]: design/mvp/Explainer.md
+[async explainer]: design/mvp/Async.md
 [IDL]: design/mvp/WIT.md
 [binary format]: design/mvp/Binary.md
 [ABI]: design/mvp/CanonicalABI.md
diff --git a/design/mvp/Async.md b/design/mvp/Async.md
new file mode 100644
index 00000000..45db28b4
--- /dev/null
+++ b/design/mvp/Async.md
@@ -0,0 +1,485 @@
+# 🔀 Async Explainer
+
+*This explainer describes a feature that is part of the forthcoming "Preview 3"
+release of the Component Model. The relevant parts of the [AST explainer],
+[binary format] and [Canonical ABI explainer] are gated by the 🔀 emoji.*
+
+This explainer provides a high-level summary of the native async support in the
+Component Model. For a detailed presentation of the runtime semantics, see the
+[Canonical ABI explainer]. See also the [Wasm I/O 2024 presentation] for a
+summary of the motivation and animated sketch of the design in action.
+
+* [Goals](#goals)
+* [High-level Approach](#high-level-approach)
+* [Concepts](#concepts)
+  * [Sync and Async Functions](#sync-and-async-functions)
+  * [Task](#task)
+  * [Current task](#current-task)
+  * [Subtask and Supertask](#subtask-and-supertask)
+  * [Structured concurrency](#structured-concurrency)
+  * [Waiting](#waiting)
+  * [Starting](#starting)
+  * [Returning](#returning)
+* [Examples](#examples)
+* [Interaction with multi-threading](#interaction-with-multi-threading)
+* [Interaction with the start function](#interaction-with-the-start-function)
+* [TODO](#todo)
+
+
+## Goals
+
+Given only *synchronous* functions with values and resources, when a component
+needs to do concurrent (i.e., overlapping, interleaved, streaming) I/O, the
+resulting [WIT] interfaces and implementations end up being complex,
+hard to compose, and less efficient. By extending the Component Model with
+built-in **asynchronous** support, these pain points can be addressed.
+
+The Component Model's [goals] and intended [use cases] suggest the following
+additional goals and requirements for native async support:
+
+* Be independent-of but complementary-to the Core WebAssembly [stack-switching]
+  proposal; don't depend on this proposal being fully standard or implemented
+  (just like [JSPI]).
+* Be independent-of but complementary-to the Core WebAssembly
+  [shared-everything-threads] proposal; don't depend on this proposal being
+  fully standard or implemented and ensure that components can achieve a high
+  degree of concurrency using only one.
+* Avoid partitioning interfaces and components into separate strata; don't give
+  functions (or components) a [color].
+* Enable tight integration (e.g., automatic bindings generation) with a wide
+  variety of source languages' built-in concurrency features.
+* Maintain meaningful cross-language call stacks (for the benefit of debugging,
+  logging and tracing).
+* Provide mechanisms for applying and observing backpressure.
+
+
+## High-level Approach
+
+Based on the above goals, the Component Model's approach to native async starts
+by allowing components to import and export "async" functions which abstract
+over and can literally be implemented in terms of:
+* `async` functions in languages like C#, JS, Python, Rust and Swift
+* stackful coroutines in languages like Kotlin, Perl, PHP and (recently) C++
+* green threads as-if running on a single OS thread in languages like Go and
+  (initially and recently again) Java
+
+The Component Model supports this wide variety of language features by
+specifying a common low-level "async" ABI which the different languages'
+compilers and runtimes can bind their different language-level concurrency
+features to. This is similar to how a native OS exposes APIs for concurrency
+(such as `epoll`, `io_uring`, `kqueue` or Overlapped I/O) to which most of
+these languages' concurrency features are already bound (making the Component
+Model "just another OS" from the language toolchains' perspective).
+
+Moreover, this async ABI does not require components to use preemptive
+multi-threading ([`thread.spawn`]) in order to achieve concurrency. Instead,
+concurrency can be achieved by cooperatively switching between different
+logical tasks running on a single thread. This switching may require the use of
+[fibers] or a [CPS transform], but may also be avoided entirely when a
+component's producer toolchain is engineered to always return to an
+[event loop].
+
+To avoid partitioning the world along sync/async lines as mentioned in the
+Goals section, the Component Model allows *every* component-level function type
+to be both implemented and called in either a synchronous or asynchronous
+manner. Thus, function types do not dictate synchrony or asynchrony and all 4
+combinations of {sync, async} x {caller, callee} are supported and given a
+well-defined behavior. Specifically, the caller and callee can independently
+specify `async` as an immediate flags on the [lift and lower definitions] used
+to define their imports and exports.
+
+To propagate backpressure, it's necessary for a component to be able to say
+"there are too many async export calls already in progress, don't start any
+more until I let some of them complete". Thus, the low-level async ABI provides
+a way to apply and release backpressure.
+
+With this backpressure protocol in place, there is a natural way for sync and
+async code to interoperate:
+1. If an async component calls a sync component and the sync component blocks,
+   execution is immediately returned to the async component, effectively
+   suspending the sync component.
+2. If anyone tries to reenter the now-suspended sync component, the Component
+   Model automatically signals backpressure on the suspended component's
+   behalf.
+
+Thus, backpressure combined with the partitioning of low-level state provided
+by the Component Model enables sync and async code to interoperate while
+preserving the expectations of both.
+
+[TODO](#todo): `future` and `stream` types that can be used in function
+signatures will be added next.
+
+
+## Concepts
+
+The following concepts are defined as part of the Component Model's native
+async support.
+
+### Sync and Async Functions
+
+The distinction between sync and async functions does not appear in the
+component-level function type (nor in WIT). Rather, an "async" function is a
+component-level function that has been [lifted] from Core WebAssembly with the
+`async` option set. Symmetrically, a "sync" function is a component-level
+function that does not have the `async` option set (which is the default and
+only option prior to Preview 3). Thus, the sync/async distinction appears
+only independently in how a component-level function is *implemented* or
+*called*.
+
+### Task
+
+Every time a lifted function is called (e.g., when a component's export is
+called by the outside world), a new **task** is created that logically contains
+all the transitive control-flow state of the export call and will be destroyed
+when the export call finishes. When all of a component's exports are lifted
+synchronously, there will be at most one task alive at any one time. However,
+when a component exports asynchronously-lifted functions, there can be multiple
+tasks alive at once.
+
+In the Canonical ABI explainer, a "task" is represented with the Python
+[`Task`] class, which is created by each call to [`canon_lift`] and has
+`SyncTask` and `AsyncTask` subclasses to factor out state and behavior only
+needed for sync- or async-lifted functions, resp.
+
+### Current Task
+
+At any point in time when executing in Core WebAssembly code, there is a
+single, unambiguous **current task**. Thus, whenever a [canonical built-in] is
+called by Core WebAssembly code, it is meaningful for the built-in to work in
+terms "the current task".
+
+The "current task" is modelled in the Canonical ABI's Python code
+by implicitly threading the `Task` object created by [`canon_lift`] through all
+the `async def` Python functions transitively called by `canon_lift`. Thus,
+although there can be multiple live `Task` objects in a component instance,
+"the current one" is always clear: it's the one passed to the current function
+as a parameter.
+
+### Subtask and Supertask
+
+Each component-to-component call necessarily creates a new task in the callee.
+The callee task is a **subtask** of the calling task (and, conversely, the
+calling task is a **supertask** of the callee task. This sub/super relationship
+is immutable and doesn't change over time (until the callee task completes and
+is destroyed).
+
+The Canonical ABI's Python code represents the subtask relationship between a
+caller `Task` and a callee `Task` via the Python [`Subtask`] class. Whereas a
+`Task` object is created by each call to `canon_lift`, a `Subtask` object is
+created by each call to [`canon_lower`]. This allows `Subtask`s to store the
+state that enforces the caller side of the Canonical ABI rules.
+
+### Structured concurrency
+
+To realize the above goals of always having a well-defined cross-component
+async callstack, the Component Model's Canonical ABI enforces [Structured
+Concurrency] by dynamically requiring that a task waits for all its subtasks to
+finish before the task itself is allowed to finish. This means that a subtask
+cannot be orphaned and there will always be an async callstack rooted at an
+invocation of an export by the host. Moreover, at any one point in time, the
+set of tasks active in a linked component graph form a forest of async call
+trees which e.g., can be visualized using a traditional flamegraph.
+
+The Canonical ABI's Python code enforces Structured Concurrency by maintaining
+a simple per-[`Task`] `num_async_subtasks` counter that traps if not zero when
+the `Task` finishes.
+
+### Waiting
+
+When a component asynchronously lowers an import, it is explicitly requesting
+that, if the import blocks, control flow be returned back to the calling task
+so that it can do something else. Eventually though a task may run out of other
+things to do and will need to **wait** for progress on one of the task's
+subtasks. While a task is waiting, the runtime can switch to other running
+tasks or start new tasks by invoking exports.
+
+The Canonical ABI provides two ways for a task to wait:
+* The task can call the [`task.wait`] built-in to synchronously wait for
+  progress. This is specified in the Canonical ABI by the [`canon_task_wait`]
+  function.
+* The task can specify a `callback` function (in the `canon lift` definition)
+  and return to the event loop to wait for notification of progress by a call
+  to the `callback` function. This is specified in the Canonical ABI by
+  the `opts.callback` case in [`canon_lift`].
+
+While the two approaches have significant runtime implementation differences
+(the former requires [fibers] or a [CPS transform] while the latter only
+requires storing a small `i32` "context" in the task), semantically they do the
+same thing which, in the Canonical ABI Python code, is factored out into
+[`Task`]'s `wait` method. Thus, the difference between `callback` and
+non-`callback` is mostly one of optimization, not expressivity.
+
+### Starting
+
+When a component asynchronously lifts a function, instead of the function
+eagerly receiving its lowered parameters (as a synchronous function would) the
+asynchronously-lifted Core WebAssembly function is passed an empty list of
+arguments and must instead call an imported [`task.start`] built-in to lower
+and receive its arguments.
+
+The main reason to have `task.start` is so that an overloaded component
+instance can [wait](#waiting) for existing tasks to finish before starting a
+new task. Once a component instance waits to start a task, the Component Model
+will automatically apply backpressure, preventing new tasks from starting.
+See the [`AsyncTask`] class and [`canon_task_start`] function in the Canonical
+ABI explainer for more details.
+
+Before a task has called `task.start`, it is considered in the "starting"
+state. After calling `task.start`, the task is in a "started" state.
+
+### Returning
+
+Symmetric to starting, the way a Core WebAssembly function returns its value is
+by calling [`task.return`], passing the core values that are to be lifted.
+
+The main reason to have `task.return` is so that a task can continue execution
+after returning its value. This is useful for various finalization tasks (such
+as logging, billing or metrics) that don't need to be on the critical path of
+returning a value to the caller. (It also subsumes and generalizes the
+Canonical ABI's `post_return` function used by synchronous functions to release
+memory.)
+
+A task may not call `task.return` unless it is in the "started" state. Once
+`task.return` is called, the task is in the "returned" state. A task can only
+finish once it is in the "returned" state. See the [`canon_task_return`]
+function in the Canonical ABI explainer for more details.
+
+
+## Examples
+
+With that background, we can sketch the shape of an async component that lifts
+and lowers its imports and exports with `async`. The meat of this component is
+replaced with `...` to focus on the overall flow of function calls.
+```wat
+(component
+  (import "fetch" (func $fetch (param "url" string) (result (list u8))))
+  (core module $Main
+    (import "" "fetch" (func $fetch (param i32 i32) (result i32)))
+    (import "" "task.start" (func $task_start (param i32)))
+    (import "" "task.return" (func $task_return (param i32)))
+    (import "" "task.wait" (func $wait (param i32) (result i32)))
+    (func (export "summarize")
+      ...
+      call $task_start   ;; receive the list of strings arguments
+      ...
+      loop
+        ...
+        call $fetch      ;; pass a pointer-to-string and pointer-to-list-of-bytes outparam
+        ...              ;; ... and receive the index of a new async subtask
+      end
+      loop               ;; loop as long as there are any subtasks
+        ...
+        call $task_wait  ;; wait for a subtask to make progress
+        ...
+      end
+      ...
+      call $task_return  ;; return the string result
+      ...
+    )
+  )
+  (canon lower $fetch async (core func $fetch'))
+  (canon task.start (core func $task_start))
+  (canon task.return (core func $task_return))
+  (canon task.wait (core func $task_wait))
+  (core instance $main (instantiate $Main (with "" (instance
+    (export "fetch" (func $fetch'))
+    (export "task.start" (func $task_start))
+    (export "task.return" (func $task_return))
+    (export "task.wait" (func $task_wait))
+  ))))
+  (canon lift (core func $main "summarize") async
+    (func $summarize (param "urls" (list string)) (result string)))
+  (export "summarize" (func $summarize))
+)
+```
+Because the imported `fetch` function is `canon lower`ed with `async`, its core
+function type (shown in the first import of `$Main`) takes pointers to the
+parameter and results (which are asynchronously read-from and written-to) and
+returns the index of a new subtask. `summarize` calls `task.wait` repeatedly
+until all `fetch` subtasks have finished, noting that `task.wait` can return
+intermediate progress (as subtasks transition from "starting" to "started" to
+"returned" to "done") which tell the surrounding core wasm code that it can
+reclaim the memory passed arguments or use the results that have now been
+written to the outparam memory.
+
+Because the `summarize` function is `canon lift`ed with `async`, its core
+function type has no params or results, since parameters are passed in via
+`task.start` and results are passed out via `task.return`. It also means that
+multiple `summarize` calls can be active at once: once the first call to
+`task.wait` blocks, the runtime will suspend its callstack (fiber) and start a
+new stack for the new call to `summarize`. Thus, `summarize` must be careful to
+allocate a separate linear-memory stack in its entry point, if one is needed,
+and to save and restore this before and after calling `task.wait`.
+
+(Note that, for brevity this example ignores the `memory` and `realloc`
+immediates required by `canon lift` and `canon lower` to allocate the `list`
+param and `string` result, resp.)
+
+This same example can be re-written to use the `callback` immediate (thereby
+avoiding the need for fibers) as follows. Note that the internal structure of
+this component is almost the same as the previous one (other than that
+`summarize` is now lifted from *two* core wasm functions instead of one) and
+the public signature of this component is the exact same. Thus, the difference
+is just about whether the stack is cleared by the core wasm code between events,
+not externally-visible behavior.
+```wat
+(component
+  (import "fetch" (func $fetch (param "url" string) (result (list u8))))
+  (core module $Main
+    (import "" "fetch" (func $fetch (param i32 i32) (result i32)))
+    (import "" "task.start" (func $task_start (param i32)))
+    (import "" "task.return" (func $task_return (param i32)))
+    (import "" "task.wait" (func $wait (param i32) (result i32)))
+    (func (export "summarize") (result i32)
+      ...
+      call $task_start        ;; receive the list of strings arguments
+      ...
+      loop
+        ...
+        call $fetch           ;; pass a pointer-to-string and pointer-to-list-of-bytes outparam
+        ...                   ;; ... and receive the index of a new async subtask
+      end
+      ...                     ;; return a non-zero "cx" value passed to the next call to "cb"
+    )
+    (func (export "cb") (param $cx i32) (param $event i32) (param $payload i32)
+      ...
+      if ... subtasks remain ...
+        get_local $cx
+        return                ;; wait for another subtask to make progress
+      end
+      ...
+      call $task_return       ;; return the string result
+      ...
+      i32.const 0             ;; return zero to signal that this task is done
+    )
+  )
+  (canon lower $fetch async (core func $fetch'))
+  (canon task.start (core func $task_start))
+  (canon task.return (core func $task_return))
+  (canon task.wait (core func $task_wait))
+  (core instance $main (instantiate $Main (with "" (instance
+    (export "fetch" (func $fetch'))
+    (export "task.start" (func $task_start))
+    (export "task.return" (func $task_return))
+    (export "task.wait" (func $task_wait))
+  ))))
+  (canon lift (core func $main "summarize") async (callback (core func $main "cb"))
+    (func $summarize (param "urls" (list string)) (result string)))
+  (export "summarize" (func $summarize))
+)
+```
+While this example spawns all the subtasks in the initial call to `summarize`,
+subtasks can also be spawned from `cb` (even after the call to `task.return`).
+It's also possible for `summarize` to wait to call `task.start` from `cb` or,
+conversely, for `task.return` to be called eagerly in the initial call to
+`summarize`.
+
+The `$event` and `$payload` parameters passed to `cb` are the same as the return
+values from `task.wait` in the previous example. The precise meaning of these
+values is defined by the Canonical ABI.
+
+
+## Interaction with multi-threading
+
+For now, the integration between multi-threading (via [`thread.spawn`]) and
+native async is limited. In particular, because all [lift and lower
+definitions] produce non-`shared` functions, any threads spawned by a component
+via `thread.spawn` will not be able to directly call imports (synchronously
+*or* asynchronously) and will thus have to use Core WebAssembly `atomics.*`
+instructions to switch back to a non-`shared` function running on the "main"
+thread (i.e., whichever thread was used to call the component's exports).
+
+However, a future addition to this proposal (in the [TODO](#todo)s below) would
+be to allow lifting and lowering with `async` + `shared`. What's exciting about
+this approach is that a non-`shared` component-level function could be safely
+lowered with `async shared`. In the case that the lifted function being lowered
+was also `async shared`, the entire call could happen on the non-main thread
+without a context switch. But if the lifting side was non-`shared`, then the
+Component Model could automatically handle the synchronization of enqueing a
+call to the export (as in the backpressure case mentioned above), returning a
+subtask for the async caller to wait on as usual. Thus, the sync+async
+composition story described above could naturally be extended to a
+sync+async+shared composition story, continuing to avoid the "what color is
+your function" problem (where `shared` is the [color]).
+
+Even without any use of `thread.new`, native async provides an opportunity to
+achieve some automatic parallelism "for free". In particular, due to the
+shared-nothing nature of components, each component instance could be given a
+separate thread on which to interleave all tasks executing in that instance.
+Thus, in a cross-component call from `C1` to `C2`, `C2`'s task can run in a
+separate thread that is automatically synchronized with `C1` by the runtime.
+This is analogous to how traditional OS processes can run in separate threads,
+except that the component model is *allowing*, but not *requiring* the separate
+threads. While it's unclear how much parallelism this would unlock in practice,
+it does present interesting opportunities to experiment with optimizations over
+time as applications are built with more components.
+
+
+## Interaction with the start function
+
+Since component-level start functions can be any component-level function (with
+type `[] -> []`), async functions can be start functions. This raises some
+interesting questions concerning how much concurrency during instantiation (of
+a whole component DAG) is allowed and how parent components can control this.
+For now, this remains a [TODO](#todo) and validation will reject `async`-lifted
+`start` functions.
+
+
+## TODO
+
+Native async support is being proposed in progressive chunks. The following
+features will be added in future chunks to complete "async" in Preview 3:
+* `future`/`stream`/`error`: add for use in function types for finer-grained
+  concurrency
+* `subtask.cancel`: allow a supertask to signal to a subtask that its result is
+  no longer wanted and to please wrap it up promptly
+* `subtask.drop`: allow tail-calling a subtask so that the current wasm
+  instance can be torn down eagerly
+* `task.index`+`task.wake`: allow tasks in the same instance to wait on and
+  wake each other (async condvar-style)
+* `nonblocking` function type attribute: allow a function to declare in its
+  type that it will not transitively do anything blocking
+* `recursive` function type attribute: allow a function to be reentered
+  recursively (instead of trapping)
+* enable `async` `start` functions
+* integrate with `shared`: define how to lift and lower functions `async` *and*
+  `shared`
+
+
+[Wasm I/O 2024 presentation]: https://www.youtube.com/watch?v=y3x4-nQeXxc
+[Color]: https://journal.stuffwithstuff.com/2015/02/01/what-color-is-your-function/
+
+[Fibers]: https://en.wikipedia.org/wiki/Fiber_(computer_science)
+[CPS Transform]: https://en.wikipedia.org/wiki/Continuation-passing_style
+[Event Loop]: https://en.wikipedia.org/wiki/Event_loop
+[Structured Concurrency]: https://en.wikipedia.org/wiki/Structured_concurrency
+
+[AST Explainer]: Explainer.md
+[Lift and Lower Definitions]: Explainer.md#canonical-definitions
+[Lifted]: Explainer.md#canonical-definitions
+[Canonical Built-in]: Explainer.md#canonical-built-ins
+[`task.start`]: Explainer.md#-async-built-ins
+[`task.return`]: Explainer.md#-async-built-ins
+[`task.wait`]: Explainer.md#-async-built-ins
+[`thread.spawn`]: Explainer.md#-threading-built-ins
+[ESM-integration]: Explainer.md#ESM-integration
+
+[Canonical ABI Explainer]: CanonicalABI.md
+[`canon_lift`]: CanonicalABI.md#canon-lift
+[`canon_lower`]: CanonicalABI.md#canon-lower
+[`canon_lower`]: CanonicalABI.md#canon-task-wait
+[`canon_task_wait`]: CanonicalABI.md#-canon-taskwait
+[`canon_task_start`]: CanonicalABI.md#-canon-taskstart
+[`canon_task_return`]: CanonicalABI.md#-canon-taskreturn
+[`Task`]: CanonicalABI.md#runtime-state
+[`Subtask`]: CanonicalABI.md#runtime-state
+[`AsyncTask`]: CanonicalABI.md#runtime-state
+
+[Binary Format]: Binary.md
+[WIT]: WIT.md
+[Goals]: ../high-level/Goals.md
+[Use Cases]: ../high-level/UseCases.md
+
+[stack-switching]: https://github.com/WebAssembly/stack-switching/
+[JSPI]: https://github.com/WebAssembly/js-promise-integration/
+[shared-everything-threads]: https://github.com/webAssembly/shared-everything-threads
diff --git a/design/mvp/Binary.md b/design/mvp/Binary.md
index 2192ca53..9008fe0c 100644
--- a/design/mvp/Binary.md
+++ b/design/mvp/Binary.md
@@ -200,6 +200,7 @@ label'        ::= len:<u32> l:<label>                     => l    (if len = |l|)
 valtype       ::= i:<typeidx>                             => i
                 | pvt:<primvaltype>                       => pvt
 resourcetype  ::= 0x3f 0x7f f?:<funcidx>?                 => (resource (rep i32) (dtor f)?)
+                | 0x3e 0x7f f:<funcidx> cb?:<funcidx>?    => (resource (rep i32) (dtor async f (callback cb)?))
 functype      ::= 0x40 ps:<paramlist> rs:<resultlist>     => (func ps rs)
 paramlist     ::= lt*:vec(<labelvaltype>)                 => (param lt)*
 resultlist    ::= 0x00 t:<valtype>                        => (result t)
@@ -270,9 +271,15 @@ canon    ::= 0x00 0x00 f:<core:funcidx> opts:<opts> ft:<typeidx> => (canon lift
            | 0x01 0x00 f:<funcidx> opts:<opts>                   => (canon lower f opts (core func))
            | 0x02 rt:<typeidx>                                   => (canon resource.new rt (core func))
            | 0x03 rt:<typeidx>                                   => (canon resource.drop rt (core func))
+           | 0x07 rt:<typdidx>                                   => (canon resource.drop rt async (core func))
            | 0x04 rt:<typeidx>                                   => (canon resource.rep rt (core func))
            | 0x05 ft:<typeidx>                                   => (canon thread.spawn ft (core func))
            | 0x06                                                => (canon thread.hw_concurrency (core func))
+           | 0x08 ft:<core:typeidx>                              => (canon task.start ft (core func))
+           | 0x09 ft:<core:typeidx>                              => (canon task.return ft (core func))
+           | 0x0a                                                => (canon task.wait (core func))
+           | 0x0b                                                => (canon task.poll (core func))
+           | 0x0c                                                => (canon task.yield (core func))
 opts     ::= opt*:vec(<canonopt>)                                => opt*
 canonopt ::= 0x00                                                => string-encoding=utf8
            | 0x01                                                => string-encoding=utf16
@@ -280,6 +287,8 @@ canonopt ::= 0x00                                                => string-encod
            | 0x03 m:<core:memidx>                                => (memory m)
            | 0x04 f:<core:funcidx>                               => (realloc f)
            | 0x05 f:<core:funcidx>                               => (post-return f)
+           | 0x06                                                => async
+           | 0x07 f:<core:funcidx>                               => (callback f)
 ```
 Notes:
 * The second `0x00` byte in `canon` stands for the `func` sort and thus the
diff --git a/design/mvp/CanonicalABI.md b/design/mvp/CanonicalABI.md
index a50e4abf..3277a30d 100644
--- a/design/mvp/CanonicalABI.md
+++ b/design/mvp/CanonicalABI.md
@@ -2,7 +2,10 @@
 
 This document defines the Canonical ABI used to convert between the values and
 functions of components in the Component Model and the values and functions
-of modules in Core WebAssembly.
+of modules in Core WebAssembly. See the [AST explainer](Explainer.md) for a
+walkthrough of the static structure of a component and the
+[async explainer](Async.md) for a high-level description of the async model
+being specified here.
 
 * [Supporting definitions](#supporting-definitions)
   * [Despecialization](#despecialization)
@@ -21,6 +24,11 @@ of modules in Core WebAssembly.
   * [`canon resource.new`](#canon-resourcenew)
   * [`canon resource.drop`](#canon-resourcedrop)
   * [`canon resource.rep`](#canon-resourcerep)
+  * [`canon task.start`](#-canon-taskstart) 🔀
+  * [`canon task.return`](#-canon-taskreturn) 🔀
+  * [`canon task.wait`](#-canon-taskwait) 🔀
+  * [`canon task.poll`](#-canon-taskpoll) 🔀
+  * [`canon task.yield`](#-canon-taskyield) 🔀
 
 
 ## Supporting definitions
@@ -53,7 +61,7 @@ out-of-memory conditions (such as `memory.grow` returning `-1` from within
 `realloc`) will trap (via `unreachable`). This significantly simplifies the
 Canonical ABI by avoiding the need to support the complicated protocols
 necessary to support recovery in the middle of nested allocations. In the MVP,
-for large allocations that can OOM, [streams](Explainer.md#TODO) would usually
+for large allocations that can OOM, [streams](Async.md#todo) would usually
 be the appropriate type to use and streams will be able to explicitly express
 failure in their type. Post-MVP, [adapter functions] would allow fully custom
 OOM handling for all component-level types, allowing a toolchain to
@@ -235,6 +243,8 @@ class CanonicalOptions:
   string_encoding: Optional[str] = None
   realloc: Optional[Callable] = None
   post_return: Optional[Callable] = None
+  sync: bool = True # = !canonopt.async
+  callback: Optional[Callable] = None
 ```
 
 The `inst` field of `CallContext` points to the component instance which the
@@ -245,22 +255,53 @@ Canonical ABI:
 class ComponentInstance:
   # core module instance state
   may_leave: bool
-  may_enter: bool
+  may_enter_sync: bool
+  may_enter_async: bool
+  pending_sync_tasks: list[asyncio.Future]
+  pending_async_tasks: list[asyncio.Future]
   handles: HandleTables
+  async_subtasks: Table[AsyncSubtask]
+  thread: asyncio.Lock
 
   def __init__(self):
     self.may_leave = True
-    self.may_enter = True
+    self.may_enter_sync = True
+    self.may_enter_async = True
+    self.pending_sync_tasks = []
+    self.pending_async_tasks = []
     self.handles = HandleTables()
-```
-The `may_enter` and `may_leave` fields are used to enforce the [component
-invariants]: `may_leave` indicates whether the instance may call out to an
-import and the `may_enter` state indicates whether the instance may be called
-from the outside world through an export.
-
-The `handles` field of `ComponentInstance` contains a mapping from
-`ResourceType` to `Table`s of `HandleElem`s (defined next), establishing a
-separate `i32`-indexed array per resource type.
+    self.async_subtasks = Table[AsyncSubtask]()
+    self.thread = asyncio.Lock()
+```
+The `may_leave` field is used below to track whether the instance may call a
+lowered import to prevent optimization-breaking cases of reentrance during
+lowering.
+
+The `may_enter_(sync|async)` and `pending_(sync|async)_tasks` fields
+are used below to implement backpressure that is applied when new
+sync|async-lifted export calls try to enter this `ComponentInstance`.
+
+The `handles` field contains a mapping from `ResourceType` to `Table`s of
+`HandleElem`s (defined next), establishing a separate `i32`-indexed array per
+resource type.
+
+The `async_subtasks` field is used below to track and assign an `i32` index to
+each active async-lowered call in progress that has been made by this
+`ComponentInstance`.
+
+Finally, the `thread` field is used below to restrict the switching of Python
+coroutines (`async def` functions) to only occur at specific points (such as
+when a task blocks on `task.wait` or when an `async callback`-lifted export
+call returns to its event loop to wait for an event. Thus, the calls to
+`thread.acquire()` and `thread.release()` in the Python code below point to
+where the runtime may switch between concurrent tasks. Without this
+`asyncio.Lock`, Python's normal `asyncio` semantics would allow switching
+between concurrent tasks at *every* spec-internal `await`, which would lead to
+multi-threading-like interleaving between concurrent tasks. (Alternatively, if
+Python had standard-library fibers, they could have been used instead of
+`asyncio`, obviating the need for this `Lock`.)
+
+One `HandleTables` object is stored per `ComponentInstance` and is defined as:
 ```python
 class HandleTables:
   rt_to_table: MutableMapping[ResourceType, Table[HandleElem]]
@@ -297,10 +338,14 @@ corresponds to resource type equality, as defined by [type checking] rules.
 class ResourceType(Type):
   impl: ComponentInstance
   dtor: Optional[Callable]
+  dtor_sync: bool
+  dtor_callback: Optional[Callable]
 
-  def __init__(self, impl, dtor = None):
+  def __init__(self, impl, dtor = None, dtor_sync = True, dtor_callback = None):
     self.impl = impl
     self.dtor = dtor
+    self.dtor_sync = dtor_sync
+    self.dtor_callback = dtor_callback
 ```
 The `Table` class, used by `HandleTables` above, encapsulates a single
 mutable, growable array of generic elements, indexed by Core WebAssembly
@@ -360,7 +405,7 @@ stored in `HandleTables`:
 class HandleElem:
   rep: int
   own: bool
-  scope: Optional[ExportCall]
+  scope: Optional[Task]
   lend_count: int
 
   def __init__(self, rep, own, scope = None):
@@ -375,10 +420,10 @@ fixed to be an `i32`) passed to `resource.new`.
 The `own` field indicates whether this element was created from an `own` type
 (or, if false, a `borrow` type).
 
-The `scope` field stores the `ExportCall` that created the borrowed handle.
-Until async is added to the Component Model, because of the non-reentrancy of
-components, there is at most one `ExportCall` alive for a given component at a
-time and thus this field does not actually need to be stored per `HandleElem`.
+The `scope` field stores the `Task` that created the borrowed handle. When a
+component only uses sync-lifted exports, due to lack of reentrance, there is at
+most one `Task` alive in a component instance at any time and thus an
+optimizing implementation doesn't need to store the `Task` per `HandleElem`.
 
 The `lend_count` field maintains a conservative approximation of the number of
 live handles that were lent from this `own` handle (by calls to `borrow`-taking
@@ -391,43 +436,187 @@ table only contains `own` or `borrow` handles and then, based on this,
 statically eliminate the `own` and the `lend_count` xor `scope` fields,
 and guards thereof.
 
-One `CallContext` is created for each call for both the component caller and
-callee (in `canon_lower` and `canon_lift`, resp., as defined below). Thus, a
-cross-component call will create 2 `CallContext` objects for the call, while a
-component-to-host or host-to-component call will create a single `CallContext`
-for the component caller or callee, resp.
-
-Additional per-call state is required to check that callers and callees uphold
-their respective parts of the call contract:
-
-The `ExportCall` subclass of `CallContext` tracks the number of borrowed
-handles that were passed as parameters to the export that have not yet been
-dropped (which might dangle if the caller destroys the resource after the
-call):
-```python
-class ExportCall(CallContext):
+Additional runtime state is required to implement the canonical built-ins and
+check that callers and callees uphold their respective parts of the call
+contract. This additional call state derives from `CallContext`, adding extra
+mutable fields. There are two subclasses of `CallContext`: `Task`, which is
+created by `canon_lift` and `Subtask`, which is created by `canon_lower`.
+Additional sync-/async-specialized mutable state is added by the `SyncTask`,
+`AsyncTask` and `AsyncSubtask` subclasses.
+
+The `Task` class and its subclasses depend on the following two enums:
+```python
+class AsyncCallState(IntEnum):
+  STARTING = 0
+  STARTED = 1
+  RETURNED = 2
+  DONE = 3
+
+class EventCode(IntEnum):
+  CALL_STARTING = AsyncCallState.STARTING
+  CALL_STARTED = AsyncCallState.STARTED
+  CALL_RETURNED = AsyncCallState.RETURNED
+  CALL_DONE = AsyncCallState.DONE
+  YIELDED = 4
+```
+The `AsyncCallState` enum describes the linear sequence of states that an async
+call necessarily transitions through: [`STARTING`](Async.md#starting),
+`STARTED`, [`RETURNING`](Async.md#returning) and `DONE`. The `EventCode` enum
+shares common code values with `AsyncCallState` to define the set of integer
+event codes that are delivered to [waiting](Async.md#waiting) or polling tasks.
+
+A `Task` object is created for each call to `canon_lift` and is implicitly
+threaded through all core function calls. This implicit `Task` parameter
+specifies a concept of [the current task](Async.md#current-task) and inherently
+scopes execution of all core wasm (including `canon`-defined core functions) to
+a `Task`.
+```python
+class Task(CallContext):
+  caller: Optional[Task]
   borrow_count: int
+  events: asyncio.Queue[AsyncSubtask]
+  num_async_subtasks: int
 
-  def __init__(self, opts, inst):
+  def __init__(self, opts, inst, caller):
     super().__init__(opts, inst)
+    self.caller = caller
     self.borrow_count = 0
-
+    self.events = asyncio.Queue[AsyncSubtask]()
+    self.num_async_subtasks = 0
+```
+The fields of `Task` are only accessed by the methods of `Task` and are
+introduced in groups of related `Task`-methods next. Using a conservative
+syntactic analysis of the component-level definitions of a linked component
+DAG, an optimizing implementation can statically eliminate these fields when
+the particular feature (`borrow` handles, `async` imports) is not used.
+
+The `caller` field is immutable and is either `None`, when a `Task` is created
+for a component export called directly by the host, or else the current task
+when the calling component called into this component. The `caller` field is
+used by the following two methods to prevent a component from being reentered
+(enforcing the [component invariant]) in a way that is well-defined even in the
+presence of async calls). (The `thread.acquire()` call in `enter()` is
+described above and here ensures that concurrent export calls do not
+arbitrarily interleave.)
+```python
+  async def enter(self):
+    await self.inst.thread.acquire()
+    self.trap_if_on_the_stack(self.inst)
+
+  def trap_if_on_the_stack(self, inst):
+    c = self.caller
+    while c is not None:
+      trap_if(c.inst is int)
+      c = c.caller
+```
+By analyzing a linked component DAG, an optimized implementation can avoid the
+O(n) loop in `trap_if_on_the_stack`:
+* Reentrance by a child component can (often) be statically ruled out when the
+  parent component doesn't both lift and lower the child's imports and exports
+  (i.e., "donut wrapping").
+* Reentrance of the root component by the host can either be asserted not to
+  happen or be tracked in a per-root-component-instance flag.
+* When a potentially-reenterable child component only lifts and lowers
+  synchronously, reentrance can be tracked in a per-component-instance flag.
+* For the remaining cases, the live instances on the stack can be maintained in
+  a packed bit-vector (assigning each potentially-reenterable async component
+  instance a static bit position) that is passed by copy from caller to callee.
+
+The `borrow_count` field is used by the following methods to track the number
+of borrowed handles that were passed as parameters to the export that have not
+yet been dropped (and thus might dangle if the caller destroys the resource
+after this export call finishes):
+```python
   def create_borrow(self):
     self.borrow_count += 1
 
   def drop_borrow(self):
     assert(self.borrow_count > 0)
     self.borrow_count -= 1
-
+```
+The `exit` defined below traps if `borrow_count` is not zero when the lifted
+call completes.
+
+All `Task`s (whether lifted `async` or not) are allowed to call `async`-lowered
+imports. Calling an `async`-lowered import creates an `AsyncSubtask` (defined
+below) which is stored in the current component instance's `async_subtasks`
+table and tracked by the current task's `num_async_subtasks` counter, which is
+guarded to be `0` in `Task.exit` (below) to ensure [structured concurrency].
+```python
+  def add_async_subtask(self, subtask):
+    assert(subtask.supertask is None and subtask.index is None)
+    subtask.supertask = self
+    subtask.index = self.inst.async_subtasks.add(subtask)
+    self.num_async_subtasks += 1
+    return subtask.index
+
+  def async_subtask_made_progress(self, subtask):
+    assert(subtask.supertask is self)
+    if subtask.enqueued:
+      return
+    subtask.enqueued = True
+    self.events.put_nowait(subtask)
+
+  async def wait(self):
+    self.inst.thread.release()
+    subtask = await self.events.get()
+    await self.inst.thread.acquire()
+    return self.process_event(subtask)
+
+  def process_event(self, subtask):
+    assert(subtask.supertask is self)
+    subtask.enqueued = False
+    if subtask.state == AsyncCallState.DONE:
+      self.inst.async_subtasks.remove(subtask.index)
+      self.num_async_subtasks -= 1
+    return (EventCode(subtask.state), subtask.index)
+```
+While a task is running, it may call `wait` (via `canon task.wait` or, when a
+`callback` is present, by returning to the event loop) to block until there is
+progress on one of the task's async subtasks. Although the Python code above
+uses an `asyncio.Queue` to coordinate async events, an optimized implementation
+should not have to create an actual queue; instead it should be possible to
+embed a "next ready" linked list in the elements of the `async_subtasks` table
+(noting the `enqueued` guard above ensures that a subtask can be enqueued at
+most once). The implementation of `wait` releases and reacquires the
+instance-wide `thread` lock to specify that `wait` is a point where the runtime
+can switch to another task running in the same component instance or start a
+new task in response to an incoming export call.
+
+Alternatively, the current task can call `poll` (via `canon task.poll`, defined
+below), which does not block and does not allow the runtime to switch to
+another task:
+```python
+  def poll(self):
+    if self.events.empty():
+      return None
+    return self.process_event(self.events.get_nowait())
+```
+
+A task may also cooperatively yield the current thread, explicitly allowing
+the runtime to switch to another ready task, but without blocking on I/O (as
+emulated in the Python code here by awaiting a `sleep(0)`).
+```python
+  async def yield_(self):
+    self.inst.thread.release()
+    await asyncio.sleep(0)
+    await self.inst.thread.acquire()
+```
+
+Lastly, when a task exists, the runtime enforces the guard conditions mentioned
+above and releases the `thread` lock, allowing other tasks to start or make
+progress.
+```python
   def exit(self):
+    assert(self.events.empty())
     trap_if(self.borrow_count != 0)
+    trap_if(self.num_async_subtasks != 0)
+    self.inst.thread.release()
 ```
 
-The `ImportCall` subclass of `CallContext` tracks the owned handles that have
-been lent for the duration of an import call, ensuring that they aren't dropped
-during the call (which might create a dangling borrowed handle):
+While `canon_lift` creates `Task`s, `canon_lower` creates `Subtask` objects:
 ```python
-class ImportCall(CallContext):
+class Subtask(CallContext):
   lenders: list[HandleElem]
 
   def __init__(self, opts, inst):
@@ -439,14 +628,167 @@ class ImportCall(CallContext):
     lending_handle.lend_count += 1
     self.lenders.append(lending_handle)
 
-  def exit(self):
+  def finish(self):
     for h in self.lenders:
       h.lend_count -= 1
 ```
-Note, the `lenders` list usually has a fixed size (in all cases except when a
-function signature has `borrow`s in `list`s) and thus can be stored inline in
-the native stack frame.
+A `Subtask` tracks the owned handles that have been lent for the duration of
+the call, ensuring that the caller doesn't drop them during the call (which
+might create a dangling borrowed handle in the callee). Note, the `lenders`
+list usually has a fixed size (in all cases except when a function signature
+has `borrow`s in `list`s) and thus can be stored inline in the native stack
+frame.
+
+The following `SyncTask`/`AsyncTask`/`AsyncSubtask` classes extend the
+preceding `Task`/`Subtask` classes with additional state and methods that apply
+only to the sync or async case.
+
+The `SyncTask` classes overrides the `enter` and `exit` methods to additionally
+enforce the rule that there only ever at most one synchronous task running in a
+given component instance at a given time.
+```python
+class SyncTask(Task):
+  async def enter(self):
+    if not self.inst.may_enter_sync:
+      f = asyncio.Future()
+      self.inst.pending_sync_tasks.append(f)
+      await f
+      assert(self.inst.may_enter_sync)
+    self.inst.may_enter_sync = False
+    await super().enter()
 
+  def exit(self):
+    super().exit()
+    assert(not self.inst.may_enter_sync)
+    self.inst.may_enter_sync = True
+    if self.inst.pending_sync_tasks:
+      self.inst.pending_sync_tasks.pop(0).set_result(None)
+```
+Thus, after one sync task starts running, any subsequent attempts to call into
+the same component instance before the first sync task finishes will wait in a
+LIFO queue until the sync task ahead of them in line completes. An optimized
+implementation should be able to avoid separately allocating
+`pending_sync_tasks` by instead embedding a "next pending" linked list in the
+`Subtask` table element of the caller.
+
+The first 3 fields of `AsyncTask` are simply immutable copies of
+arguments/immediates passed to `canon_lift` that are used later on. The last 2
+fields are used to check the above-mentioned state machine transitions and also
+specify an async version of backpressure. In particular, the rules apply
+backpressure if a task blocks (calling `wait`) while still in the `STARTING`
+state, which signals that the component instance isn't ready to take on any new
+async calls (until some active calls finish):
+```python
+class AsyncTask(Task):
+  ft: FuncType
+  start_thunk: Callable
+  return_thunk: Callable
+  state: AsyncCallState
+  unblock_next_pending: bool
+
+  def __init__(self, opts, inst, caller, ft, start_thunk, return_thunk):
+    super().__init__(opts, inst, caller)
+    self.ft = ft
+    self.start_thunk = start_thunk
+    self.return_thunk = return_thunk
+    self.state = AsyncCallState.STARTING
+    self.unblock_next_pending = False
+
+  async def enter(self):
+    if not self.inst.may_enter_async or self.inst.pending_async_tasks:
+      f = asyncio.Future()
+      self.inst.pending_async_tasks.append(f)
+      await f
+      assert(self.inst.may_enter_async)
+      self.unblock_next_pending = len(self.inst.pending_async_tasks) > 0
+    await super().enter()
+
+  async def wait(self):
+    if self.state == AsyncCallState.STARTING:
+      self.inst.may_enter_async = False
+      self.inst.unblock_next_pending = False
+    else:
+      self.maybe_unblock_next_pending()
+    return await super().wait()
+
+  def maybe_unblock_next_pending(self):
+    if self.unblock_next_pending:
+      self.unblock_next_pending = False
+      assert(self.inst.may_enter_async)
+      self.inst.pending_async_tasks.pop(0).set_result(None)
+
+  def start(self):
+    trap_if(self.state != AsyncCallState.STARTING)
+    self.state = AsyncCallState.STARTED
+    if not self.inst.may_enter_async:
+      self.inst.may_enter_async = True
+      self.unblock_next_pending = len(self.inst.pending_async_tasks) > 0
+
+  def return_(self):
+    trap_if(self.state != AsyncCallState.STARTED)
+    self.state = AsyncCallState.RETURNED
+
+  def exit(self):
+    super().exit()
+    trap_if(self.state != AsyncCallState.RETURNED)
+    self.state = AsyncCallState.DONE
+    self.maybe_unblock_next_pending()
+```
+The above rules are careful to release pending async calls from the queue one
+at a time (rather than unblocking all of them at once). This ensures that, in
+all cases, every new task has a chance to apply backpressure before the next
+new task starts.
+
+Note that the backpressure rules described above apply independently to sync
+and async tasks and thus if a component exports both sync- *and* async-lifted
+functions, async functions may execute concurrently with sync functions.
+
+Finally, the `AsyncSubtask` class extends `Subtask` with fields that are used
+by the methods of `Task`, as shown above. `AsyncSubtask`s have the same linear
+state machine as `AsyncTask`s, except that the state transitions are guaranteed
+by the Canonical ABI to happen in the right order. Each time an async subtask
+advances a state, it notifies its "supertask", which was the current task when
+the async-lowered function was first called.
+```python
+class AsyncSubtask(Subtask):
+  state: AsyncCallState
+  supertask: Optional[Task]
+  index: Optional[int]
+  enqueued: bool
+
+  def __init__(self, opts, inst):
+    super().__init__(opts, inst)
+    self.state = AsyncCallState.STARTING
+    self.supertask = None
+    self.index = None
+    self.enqueued = False
+
+  def start(self):
+    assert(self.state == AsyncCallState.STARTING)
+    self.state = AsyncCallState.STARTED
+    if self.supertask is not None:
+      self.supertask.async_subtask_made_progress(self)
+
+  def return_(self):
+    assert(self.state == AsyncCallState.STARTED)
+    self.state = AsyncCallState.RETURNED
+    if self.supertask is not None:
+      self.supertask.async_subtask_made_progress(self)
+
+  def finish(self):
+    super().finish()
+    assert(self.state == AsyncCallState.RETURNED)
+    self.state = AsyncCallState.DONE
+    if self.supertask is not None:
+      self.supertask.async_subtask_made_progress(self)
+```
+The `supertask` and `index` fields will be `None` when a subtask first starts
+executing, before it blocks and gets added to the `async_subtasks` table (by
+`canon_lower`, below). If a subtask advances all the way to the `DONE` state
+before blocking, the `async`-lowered call will indicate to the caller that the
+callee completed synchronously, avoiding the overhead of adding an
+`AsyncSubtask` altogether. Thus, progress events don't need to be delivered
+until the subtask has passed this "possibly synchronous early return" phase.
 
 ### Loading
 
@@ -693,7 +1035,7 @@ from the source handle, leaving the source handle intact in the current
 component instance's handle table:
 ```python
 def lift_borrow(cx, i, t):
-  assert(isinstance(cx, ImportCall))
+  assert(isinstance(cx, Subtask))
   h = cx.inst.handles.get(t.rt, i)
   if h.own:
     cx.track_owning_lend(h)
@@ -1097,7 +1439,7 @@ def lower_own(cx, rep, t):
   return cx.inst.handles.add(t.rt, h)
 
 def lower_borrow(cx, rep, t):
-  assert(isinstance(cx, ExportCall))
+  assert(isinstance(cx, Task))
   if cx.inst is t.rt.impl:
     return rep
   h = HandleElem(rep, own=False, scope=cx)
@@ -1141,25 +1483,38 @@ Given all this, the top-level definition of `flatten` is:
 MAX_FLAT_PARAMS = 16
 MAX_FLAT_RESULTS = 1
 
-def flatten_functype(ft, context):
-  flat_params = flatten_types(ft.param_types())
-  if len(flat_params) > MAX_FLAT_PARAMS:
-    flat_params = ['i32']
-
-  flat_results = flatten_types(ft.result_types())
-  if len(flat_results) > MAX_FLAT_RESULTS:
+def flatten_functype(opts, ft, context):
+  if opts.sync:
+    flat_params = flatten_types(ft.param_types())
+    if len(flat_params) > MAX_FLAT_PARAMS:
+      flat_params = ['i32']
+
+    flat_results = flatten_types(ft.result_types())
+    if len(flat_results) > MAX_FLAT_RESULTS:
+      match context:
+        case 'lift':
+          flat_results = ['i32']
+        case 'lower':
+          flat_params += ['i32']
+          flat_results = []
+
+    return CoreFuncType(flat_params, flat_results)
+  else:
     match context:
       case 'lift':
-        flat_results = ['i32']
-      case 'lower':
-        flat_params += ['i32']
+        flat_params = []
         flat_results = []
-
-  return CoreFuncType(flat_params, flat_results)
+      case 'lower':
+        flat_params = ['i32', 'i32']
+        flat_results = ['i32']
+    return CoreFuncType(flat_params, flat_results)
 
 def flatten_types(ts):
   return [ft for t in ts for ft in flatten_type(t)]
 ```
+As shown here, the core signatures `async` functions are fixed and don't vary
+based on the function type (parameters and results are passed through memory
+pointed to by the fixed `i32` parameters).
 
 Presenting the definition of `flatten_type` piecewise, we start with the
 top-level case analysis:
@@ -1463,59 +1818,89 @@ def lower_flat_flags(v, labels):
 
 ### Lifting and Lowering Values
 
-The `lift_values` function defines how to lift a list of at most `max_flat`
-core parameters or results given by the `CoreValueIter` `vi` into a tuple of
-values with types `ts`:
+The `lift_(sync|async)_values` functions define how to lift a list of core
+parameters or results (given by the `CoreValueIter` `vi`) into a tuple of
+component-level values with types `ts`. The sync and async variants differ in
+how much they can pass in scalar "registers" before falling back to passing
+values through linear memory: sync functions use up to `max_flat` scalars
+whereas async functions have a single fixed `i32` that is either a single
+scalar value or a pointer into linear memory:
 ```python
-def lift_values(cx, max_flat, vi, ts):
+def lift_sync_values(cx, max_flat, vi, ts):
   flat_types = flatten_types(ts)
   if len(flat_types) > max_flat:
-    ptr = vi.next('i32')
-    tuple_type = Tuple(ts)
-    trap_if(ptr != align_to(ptr, alignment(tuple_type)))
-    trap_if(ptr + elem_size(tuple_type) > len(cx.opts.memory))
-    return list(load(cx, ptr, tuple_type).values())
+    return lift_heap_values(cx, vi, ts)
   else:
     return [ lift_flat(cx, vi, t) for t in ts ]
+
+def lift_async_values(cx, vi, ts):
+  if len(ts) == 0:
+    _ = vi.next('i32')
+    return []
+  flat_types = flatten_types(ts)
+  if len(flat_types) == 1 and flat_types[0] == 'i32':
+    assert(len(ts) == 1)
+    return [ lift_flat(cx, vi, ts[0]) ]
+  else:
+    return lift_heap_values(cx, vi, ts)
+
+def lift_heap_values(cx, vi, ts):
+  ptr = vi.next('i32')
+  tuple_type = Tuple(ts)
+  trap_if(ptr != align_to(ptr, alignment(tuple_type)))
+  trap_if(ptr + elem_size(tuple_type) > len(cx.opts.memory))
+  return list(load(cx, ptr, tuple_type).values())
 ```
 
-The `lower_values` function defines how to lower a list of component-level
-values `vs` of types `ts` into a list of at most `max_flat` core values. As
-already described for [`flatten`](#flattening) above, lowering handles the
+Symmetrically, the `lower_(sync|async)_values` functions define how to lower a
+list of component-level values `vs` of types `ts` into a list of core values.
+As already described for [`flatten`](#flattening) above, lowering handles the
 greater-than-`max_flat` case by either allocating storage with `realloc` or
 accepting a caller-allocated buffer as an out-param:
 ```python
-def lower_values(cx, max_flat, vs, ts, out_param = None):
+def lower_sync_values(cx, max_flat, vs, ts, out_param = None):
   inst = cx.inst
   assert(inst.may_leave)
   inst.may_leave = False
-
   flat_types = flatten_types(ts)
   if len(flat_types) > max_flat:
-    tuple_type = Tuple(ts)
-    tuple_value = {str(i): v for i,v in enumerate(vs)}
-    if out_param is None:
-      ptr = cx.opts.realloc(0, 0, alignment(tuple_type), elem_size(tuple_type))
-    else:
-      ptr = out_param.next('i32')
-    trap_if(ptr != align_to(ptr, alignment(tuple_type)))
-    trap_if(ptr + elem_size(tuple_type) > len(cx.opts.memory))
-    store(cx, tuple_value, tuple_type, ptr)
-    flat_vals = [ptr]
+    flat_vals = lower_heap_values(cx, vs, ts, out_param)
   else:
     flat_vals = []
     for i in range(len(vs)):
       flat_vals += lower_flat(cx, vs[i], ts[i])
+  inst.may_leave = True
+  return flat_vals
 
+def lower_async_values(cx, vs, ts, out_param):
+  if len(ts) == 0:
+    _ = out_param.next('i32')
+    return
+  inst = cx.inst
+  assert(inst.may_leave)
+  inst.may_leave = False
+  lower_heap_values(cx, vs, ts, out_param)
   inst.may_leave = True
+
+def lower_heap_values(cx, vs, ts, out_param):
+  tuple_type = Tuple(ts)
+  tuple_value = {str(i): v for i,v in enumerate(vs)}
+  if out_param is None:
+    ptr = cx.opts.realloc(0, 0, alignment(tuple_type), elem_size(tuple_type))
+    flat_vals = [ptr]
+  else:
+    ptr = out_param.next('i32')
+    flat_vals = []
+  trap_if(ptr != align_to(ptr, alignment(tuple_type)))
+  trap_if(ptr + elem_size(tuple_type) > len(cx.opts.memory))
+  store(cx, tuple_value, tuple_type, ptr)
   return flat_vals
 ```
-The `may_leave` flag is used by `canon_lower` below to prevent a component from
-calling out of the component while in the middle of lowering, ensuring that the
-relative ordering of the side effects of `lift_values` and `lower_values`
-cannot be observed and thus an implementation may reliably fuse `lift_values`
-with `lower_values` when making a cross-component call, avoiding any
-intermediate copy.
+The `may_leave` flag is guarded by `canon_lower` below to prevent a component
+from calling out of the component while in the middle of lowering, ensuring
+that the relative ordering of the side effects of lifting followed by lowering
+cannot be observed and thus an implementation may reliably fuse lifting with
+lowering when making a cross-component call to avoid the intermediate copy.
 
 ## Canonical Definitions
 
@@ -1530,62 +1915,112 @@ For a canonical definition:
 (canon lift $callee:<funcidx> $opts:<canonopt>* (func $f (type $ft)))
 ```
 validation specifies:
-* `$callee` must have type `flatten_functype($ft, 'lift')`
+* `$callee` must have type `flatten_functype($opts, $ft, 'lift')`
 * `$f` is given type `$ft`
 * a `memory` is present if required by lifting and is a subtype of `(memory 1)`
 * a `realloc` is present if required by lifting and has type `(func (param i32 i32 i32 i32) (result i32))`
-* if a `post-return` is present, it has type `(func (param flatten_functype($ft).results))`
+* if a `post-return` is present, it has type `(func (param flatten_functype({}, $ft, 'lift').results))`
 
 When instantiating component instance `$inst`:
 * Define `$f` to be the partially-bound closure `canon_lift($opts, $inst, $callee, $ft)`
 
-Thus, `$f` captures `$opts`, `$inst`, `$callee` and `$ft` in a closure which
-can be subsequently exported or passed into a child instance (via `with`). If
-`$f` ends up being called by the host, the host is responsible for, in a
-host-defined manner, conjuring up component values suitable for passing into
-`lower` and, conversely, consuming the component values produced by `lift`. For
-example, if the host is a native JS runtime, the [JavaScript embedding] would
-specify how native JavaScript values are converted to and from component
-values. Alternatively, if the host is a Unix CLI that invokes component exports
-directly from the command line, the CLI could choose to automatically parse
-`argv` into component-level values according to the declared types of the
-export. In any case, `canon lift` specifies how these variously-produced values
-are consumed as parameters (and produced as results) by a *single host-agnostic
-component*.
-
-Given the above closure arguments, `canon_lift` is defined:
-```python
-def canon_lift(opts, inst, callee, ft, start_thunk, return_thunk):
-  export_call = ExportCall(opts, inst)
-  trap_if(not inst.may_enter)
-
-  flat_args = lower_values(export_call, MAX_FLAT_PARAMS, start_thunk(), ft.param_types())
-  flat_results = call_and_trap_on_throw(callee, flat_args)
-  return_thunk(lift_values(export_call, MAX_FLAT_RESULTS, CoreValueIter(flat_results), ft.result_types()))
+The resulting function `$f` takes 3 runtime arguments:
+* `caller`: the caller's `Task` or, if this lifted function is being called by
+  the host, `None`
+* `start_thunk`: a nullary function that must be called to return the caller's
+  arguments as a list of component-level values
+* `return_thunk`: a unary function that must be called after `start_thunk`,
+  passing the list of component-level return values
+
+The indirection of `start_thunk` and `return_thunk` are used to model the
+interleaving of reading arguments out of the caller's stack and memory and
+writing results back into the caller's stack and memory, which will vary in
+async calls.
+
+If `$f` ends up being called by the host, the host is responsible for, in a
+host-defined manner, conjuring up component-level values suitable for passing
+into `lower` and, conversely, consuming the component values produced by
+`lift`. For example, if the host is a native JS runtime, the [JavaScript
+embedding] would specify how native JavaScript values are converted to and from
+component values. Alternatively, if the host is a Unix CLI that invokes
+component exports directly from the command line, the CLI could choose to
+automatically parse `argv` into component-level values according to the
+declared types of the export. In any case, `canon lift` specifies how these
+variously-produced values are consumed as parameters (and produced as results)
+by a *single host-agnostic component*.
+
+Based on this, `canon_lift` is defined:
+```python
+async def canon_lift(opts, inst, callee, ft, caller, start_thunk, return_thunk):
+  if opts.sync:
+    task = SyncTask(opts, inst, caller)
+    await task.enter()
+
+    flat_args = lower_sync_values(task, MAX_FLAT_PARAMS, start_thunk(), ft.param_types())
+    flat_results = await call_and_trap_on_throw(callee, task, flat_args)
+    return_thunk(lift_sync_values(task, MAX_FLAT_RESULTS, CoreValueIter(flat_results), ft.result_types()))
+
+    if opts.post_return is not None:
+      [] = await call_and_trap_on_throw(opts.post_return, task, flat_results)
+
+    task.exit()
+  else:
+    task = AsyncTask(opts, inst, caller, ft, start_thunk, return_thunk)
+    await task.enter()
 
-  if opts.post_return is not None:
-    call_and_trap_on_throw(opts.post_return, flat_results)
+    if not opts.callback:
+      [] = await call_and_trap_on_throw(callee, task, [])
+    else:
+      [packed_ctx] = await call_and_trap_on_throw(callee, task, [])
+      while packed_ctx != 0:
+        is_yield = bool(packed_ctx & 1)
+        ctx = packed_ctx & ~1
+        if is_yield:
+          await task.yield_()
+          event, payload = (EventCode.YIELDED, 0)
+        else:
+          event, payload = await task.wait()
+        [packed_ctx] = await call_and_trap_on_throw(opts.callback, task, [ctx, event, payload])
 
-  export_call.exit()
+    assert(opts.post_return is None)
+    task.exit()
 
-def call_and_trap_on_throw(callee, args):
+async def call_and_trap_on_throw(callee, task, args):
   try:
-    return callee(args)
+    return await callee(task, args)
   except CoreWebAssemblyException:
     trap()
 ```
+The only fundamental difference between sync and async lifting is whether
+parameters/results are automatically lowered/lifted (with `canon_lift` calling
+`start_thunk` and `return_thunk`) or whether the `callee` explicitly triggers
+`start_thunk`/`return_thunk` via `task.start`/`task.return` (defined below).
+The latter gives the callee the ability to explicitly apply backpressure (by
+waiting before calling `task.start`) whereas the former applies "backpressure"
+immediately if a sync call is already running. In both cases, backpressure is
+exerted by the `enter()` function which is `await`ed by `canon_lift`.
+
+In a sync call, after the results have been copied from the callee's memory
+into the caller's memory, the callee's `post_return` function is called to
+allow the callee to reclaim any memory. An async call doesn't need a
+`post_return` function, since the callee can keep running after calling
+`task.return`.
+
+Within the async case, there are two sub-cases depending on whether the
+`callback` `canonopt` was set. When `callback` is present, waiting happens in
+an "event loop" inside `canon_lift` which also allows yielding (i.e., allowing
+other tasks to run without blocking) by setting the LSB of the returned `i32`.
+Otherwise, waiting must happen by calling `task.wait` (defined below), which
+potentially requires the runtime implementation to use a fiber (aka. stackful
+coroutine) to switch to another task. Thus, `callback` is an optimization for
+avoiding fiber creation for async languages that don't need it (e.g., JS,
+Python, C# and Rust).
+
 Uncaught Core WebAssembly [exceptions] result in a trap at component
 boundaries. Thus, if a component wishes to signal an error, it must use some
 sort of explicit type such as `result` (whose `error` case particular language
 bindings may choose to map to and from exceptions).
 
-The `start_thunk` and `return_thunk` are used to model the interleaving of
-reading arguments out of the caller's stack and memory and writing results
-back into the caller's stack and memory. After the results have been copied
-from the callee's memory into the caller's memory, the callee's `post_return`
-function is called to allow the callee to reclaim any memory.
-
-
 ### `canon lower`
 
 For a canonical definition:
@@ -1593,42 +2028,103 @@ For a canonical definition:
 (canon lower $callee:<funcidx> $opts:<canonopt>* (core func $f))
 ```
 where `$callee` has type `$ft`, validation specifies:
-* `$f` is given type `flatten_functype($ft, 'lower')`
+* `$f` is given type `flatten_functype($opts, $ft, 'lower')`
 * a `memory` is present if required by lifting and is a subtype of `(memory 1)`
 * a `realloc` is present if required by lifting and has type `(func (param i32 i32 i32 i32) (result i32))`
 * there is no `post-return` in `$opts`
 
 When instantiating component instance `$inst`:
-* Define `$f` to be the partially-bound closure: `canon_lower($opts, $inst, $callee, $ft)`
+* Define `$f` to be the partially-bound closure: `canon_lower($opts, $callee, $ft)`
 
-where `canon_lower` is defined:
+The resulting function `$f` takes 2 runtime arguments:
+* `task`: the `Task` that was created by `canon_lift` when entering the current
+  component instance
+* `flat_args`: the list of core values passed by the core function calller
+
+Given this, `canon_lower` is defined:
 ```python
-def canon_lower(opts, inst, callee, calling_import, ft, flat_args):
-  import_call = ImportCall(opts, inst)
+async def canon_lower(opts, callee, ft, task, flat_args):
+  inst = task.inst
   trap_if(not inst.may_leave)
+  flat_args = CoreValueIter(flat_args)
 
-  assert(inst.may_enter)
-  if calling_import:
-    inst.may_enter = False
+  if opts.sync:
+    subtask = Subtask(opts, inst)
 
-  flat_args = CoreValueIter(flat_args)
-  flat_results = None
+    def start_thunk():
+      return lift_sync_values(subtask, MAX_FLAT_PARAMS, flat_args, ft.param_types())
+
+    flat_results = None
+    def return_thunk(results):
+      nonlocal flat_results
+      flat_results = lower_sync_values(subtask, MAX_FLAT_RESULTS, results, ft.result_types(), flat_args)
 
-  def start_thunk():
-    return lift_values(import_call, MAX_FLAT_PARAMS, flat_args, ft.param_types())
+    await callee(task, start_thunk, return_thunk)
 
-  def return_thunk(results):
-    nonlocal flat_results
-    flat_results = lower_values(import_call, MAX_FLAT_RESULTS, results, ft.result_types(), flat_args)
+    subtask.finish()
+  else:
+    subtask = AsyncSubtask(opts, inst)
 
-  callee(start_thunk, return_thunk)
+    async def do_call():
+      def start_thunk():
+        subtask.start()
+        return lift_async_values(subtask, flat_args, ft.param_types())
 
-  if calling_import:
-    inst.may_enter = True
+      def return_thunk(results):
+        subtask.return_()
+        lower_async_values(subtask, results, ft.result_types(), flat_args)
+
+      await callee(task, start_thunk, return_thunk)
+      subtask.finish()
+
+    asyncio.create_task(do_call())
+    await asyncio.sleep(0) # start do_call eagerly
+
+    if subtask.state == AsyncCallState.DONE:
+      flat_results = [0]
+    else:
+      i = task.add_async_subtask(subtask)
+      assert(0 < i < 2**30)
+      assert(0 <= int(subtask.state) < 2**2)
+      flat_results = [ i | (int(subtask.state) << 30) ]
 
-  import_call.exit()
   return flat_results
 ```
+In the async case, `asyncio.create_task` followed by `await asyncio.sleep(0)`
+are used together to achieve the effect of eagerly executing the `do_call`
+Python coroutine without `await`ing it. Following the `sleep(0)`, the coroutine
+has either completed eagerly (with `return_thunk` having been called to write
+the return values into the caller-supplied memory buffer), in which case the
+lowered function can simply return `0` to indicate "done". Otherwise, the
+coroutine is still running, in which case it is added to an instance-wide table
+of active async subtasks, returning the table index packed with the current
+state of the subtask (so that the caller can know whether it can reclaim the
+parameter and result memory buffers) and delivering subsequent progress events
+to the calling task via the `AsyncSubtask` `start` and `return_` methods
+(defined above).
+
+Note that the async case does *not* release or reacquire the `thread` lock
+since (due to the `trap_if_on_stack` reentrance guard in `Task.enter`) the
+`callee` is necessarily in another component instance (which has a separate
+`thread` lock). This allows fine-grained inter-component task interleaving (up
+to and including preemptive multithreading) which doesn't break regular async
+codes' assumptions due to the component shared-nothing model. This also means
+that an async import calls are *not* allowed to switch to another task in the
+same component instance (switching can happen later, when waiting for a subtask
+to make progress).
+
+The above definitions of sync/async `canon_lift`/`canon_lower` ensure that a
+sync-or-async `canon_lift` may call a sync-or-async `canon_lower`, with all
+combinations working. This is why the `Task` base class (derived by `SyncTask`
+and `AsyncTask`) contains the code for handling async-lowered subtasks. As
+mentioned above, conservative syntactic analysis of all `canon` definitions in
+a component can statically rule out combinations so that, e.g., a DAG of
+all-sync components use a plain synchronous callstack and a DAG of all `async
+callback` components use only an event loop without fibers. It's only when
+`async` (without a `callback`) or various compositions of async and sync
+components are used that fibers (or [Asyncify]) are required to implement the
+above async rules.
+
 Since any cross-component call necessarily transits through a statically-known
 `canon_lower`+`canon_lift` call pair, an AOT compiler can fuse `canon_lift` and
 `canon_lower` into a single, efficient trampoline. In the future this may allow
@@ -1638,17 +2134,10 @@ as post-MVP [adapter functions].
 
 By clearing `may_enter` for the duration of calls to imports, the `may_enter`
 guard in `canon_lift` ensures that components cannot be externally reentered,
-which is part of the [component invariants]. The `calling_import` condition
-allows a parent component to call into a child component (which is, by
-definition, not a call to an import) and for the child to then reenter the
-parent through a function the parent explicitly supplied to the child's
-`instantiate`. This form of internal reentrance allows the parent to fully
-virtualize the child's imports.
-
-Because `may_enter` is not cleared on the exceptional exit path taken by
-`trap()`, if there is a trap during Core WebAssembly execution of lifting or
-lowering, the component is left permanently un-enterable, ensuring the
-lockdown-after-trap [component invariant].
+which is part of the [component invariants]. Because `may_enter` is not cleared
+on the exceptional exit path taken by `trap()`, if there is a trap during Core
+WebAssembly execution of lifting or lowering, the component is left permanently
+un-enterable, ensuring the lockdown-after-trap [component invariant].
 
 ### `canon resource.new`
 
@@ -1665,9 +2154,9 @@ Calling `$f` invokes the following function, which adds an owning handle
 containing the given resource representation in the current component
 instance's handle table:
 ```python
-def canon_resource_new(inst, rt, rep):
+async def canon_resource_new(rt, task, rep):
   h = HandleElem(rep, own=True)
-  i = inst.handles.add(rt, h)
+  i = task.inst.handles.add(rt, h)
   return [i]
 ```
 
@@ -1685,20 +2174,42 @@ Calling `$f` invokes the following function, which removes the handle from the
 current component instance's handle table and, if the handle was owning, calls
 the resource's destructor.
 ```python
-def canon_resource_drop(inst, rt, i):
+async def canon_resource_drop(rt, sync, task, i):
+  inst = task.inst
   h = inst.handles.remove(rt, i)
+  flat_results = [] if sync else [0]
   if h.own:
     assert(h.scope is None)
     trap_if(h.lend_count != 0)
-    trap_if(inst is not rt.impl and not rt.impl.may_enter)
-    if rt.dtor:
-      rt.dtor(h.rep)
+    if inst is rt.impl:
+      if rt.dtor:
+        await rt.dtor(h.rep)
+    else:
+      if rt.dtor:
+        caller_opts = CanonicalOptions(sync = sync)
+        callee_opts = CanonicalOptions(sync = rt.dtor_sync, callback = rt.dtor_callback)
+        ft = FuncType([U32()],[])
+        callee = partial(canon_lift, callee_opts, rt.impl, rt.dtor, ft)
+        flat_results = await canon_lower(caller_opts, callee, ft, task, [h.rep, 0])
+      else:
+        task.trap_if_on_the_stack(rt.impl)
   else:
     h.scope.drop_borrow()
-  return []
+  return flat_results
 ```
-The `may_enter` guard ensures the non-reentrance [component invariant], since
-a destructor call is analogous to a call to an export.
+In general, the call to a resource's destructor is treated like a
+cross-component call (as-if the destructor was exported by the component
+defining the resource type). This means that cross-component destructor calls
+follow the same concurrency rules as normal exports. However, since there are
+valid reasons to call `resource.drop` in the same component instance that
+defined the resource, which would otherwise trap at the reentrance guard of
+`Task.enter`, an exception is made when the resource type's
+implementation-instance is the same as the current instance (which is
+statically known for any given `canon resource.drop`).
+
+When a destructor isn't present, the rules still perform a reentrance check
+since this is the caller's responsibility and the presence or absence of a
+destructor is an encapsualted implementation detail of the resource type.
 
 ### `canon resource.rep`
 
@@ -1714,13 +2225,145 @@ validation specifies:
 Calling `$f` invokes the following function, which extracts the resource
 representation from the handle.
 ```python
-def canon_resource_rep(inst, rt, i):
-  h = inst.handles.get(rt, i)
+async def canon_resource_rep(rt, task, i):
+  h = task.inst.handles.get(rt, i)
   return [h.rep]
 ```
 Note that the "locally-defined" requirement above ensures that only the
 component instance defining a resource can access its representation.
 
+### 🔀 `canon task.start`
+
+For a canonical definition:
+```wasm
+(canon task.start $ft (core func $f))
+```
+validation specifies:
+* `$f` is given type `$ft`, which validation requires to be a (core) function type
+
+Calling `$f` invokes the following function which extracts the arguments from the
+caller and lowers them into the current instance:
+```python
+async def canon_task_start(task, core_ft, flat_args):
+  assert(len(core_ft.params) == len(flat_args))
+  trap_if(task.opts.sync)
+  trap_if(core_ft != flatten_functype(CanonicalOptions(), FuncType([], task.ft.params), 'lower'))
+  task.start()
+  args = task.start_thunk()
+  flat_results = lower_sync_values(task, MAX_FLAT_RESULTS, args, task.ft.param_types(), CoreValueIter(flat_args))
+  assert(len(core_ft.results) == len(flat_results))
+  return flat_results
+```
+An expected implementation of `task.start` would generate a core wasm function
+for each lowering of an `async`-lifted export that performs the fused copy of
+the arguments into the caller, storing the index of this function in the `Task`
+structure and using `call_indirect` to perform the function-type-equality check
+required here. The call to `Task.start` (defined above) ensures that `canon
+task.start` is called exactly once, before `canon task.return`, before an async
+call finishes.
+
+### 🔀 `canon task.return`
+
+For a canonical definition:
+```wasm
+(canon task.return $ft (core func $f))
+```
+validation specifies:
+* `$f` is given type `$ft`, which validation requires to be a (core) function type
+
+Calling `$f` invokes the following function which lifts the results from the
+current instance and passes them to the caller:
+```python
+async def canon_task_return(task, core_ft, flat_args):
+  assert(len(core_ft.params) == len(flat_args))
+  trap_if(task.opts.sync)
+  trap_if(core_ft != flatten_functype(CanonicalOptions(), FuncType(task.ft.results, []), 'lower'))
+  task.return_()
+  results = lift_sync_values(task, MAX_FLAT_PARAMS, CoreValueIter(flat_args), task.ft.result_types())
+  task.return_thunk(results)
+  assert(len(core_ft.results) == 0)
+  return []
+```
+An expected implementation of `task.return` would generate a core wasm function
+for each lowering of an `async`-lifted export that performs the fused copy of
+the results into the caller, storing the index of this function in the `Task`
+structure and using `call_indirect` to perform the function-type-equality check
+required here. The call to `Task.return_` (defined above) ensures that `canon
+task.return` is called exactly once, after `canon task.start`, before an async
+call finishes.
+
+### 🔀 `canon task.wait`
+
+For a canonical definition:
+```wasm
+(canon task.wait (core func $f))
+```
+validation specifies:
+* `$f` is given type `(func (param i32) (result i32))`
+
+Calling `$f` waits for progress to be made in a subtask of the current task,
+returning the event (which is currently simply an `AsyncCallState` value)
+and writing the subtask index as an outparam:
+```python
+async def canon_task_wait(task, ptr):
+  trap_if(task.opts.callback is not None)
+  event, payload = await task.wait()
+  store(task, payload, U32(), ptr)
+  return [event]
+```
+The `trap_if` ensures that, when a component uses a `callback` all events flow
+through the event loop at the base of the stack.
+
+Note that `task.wait` releases and reacquires the `thread` lock and thus
+`canon_task_wait` allows the runtime to switch to another active task in the
+current component instance. Note also that `task.wait` can be called from a
+sync-lifted `SyncTask` so that even fully synchronous code can make concurrent
+import calls. In these fully-synchrohous cases, though, the automatic
+backpressure (applied by `SyncTask.enter`) will ensure there is only ever at
+most once task executing in the component instance and thus `task.wait` will be
+statically guaranteed to never switch tasks.
+
+### 🔀 `canon task.poll`
+
+For a canonical definition:
+```wasm
+(canon task.poll (core func $f))
+```
+validation specifies:
+* `$f` is given type `(func (param i32) (result i32))`
+
+Calling `$f` does a non-blocking check for whether an event is already
+available, returning whether or not there was such an event as a boolean and,
+if there was an event, storing the `i32` event+payload pair as an outparam.
+```python
+async def canon_task_poll(task, ptr):
+  ret = task.poll()
+  if ret is None:
+    return [0]
+  store(task, ret, Tuple([U32(), U32()]), ptr)
+  return [1]
+```
+Note that there is no `await` of `poll` and thus no possible task switching.
+
+### 🔀 `canon task.yield`
+
+For a canonical definition:
+```wasm
+(canon task.yield (core func $f))
+```
+validation specifies:
+* `$f` is given type `(func)`
+
+Calling `$f` simply releases and reacquires the `thread` lock, using a
+Python `asyncio.sleep(0)` in the middle to make it clear that other
+coroutines are allowed to acquire the `lock` and execute.
+```python
+async def canon_task_yield(task):
+  trap_if(task.opts.callback is not None)
+  await task.yield_()
+  return []
+```
+
 ### 🧵 `canon thread.spawn`
 
 For a canonical definition:
@@ -1798,6 +2441,7 @@ def canon_thread_hw_concurrency():
 [JavaScript Embedding]: Explainer.md#JavaScript-embedding
 [Adapter Functions]: FutureFeatures.md#custom-abis-via-adapter-functions
 [Shared-Everything Dynamic Linking]: examples/SharedEverythingDynamicLinking.md
+[Structured Concurrency]: Async.md#structured-concurrency
 
 [Administrative Instructions]: https://webassembly.github.io/spec/core/exec/runtime.html#syntax-instr-admin
 [Implementation Limits]: https://webassembly.github.io/spec/core/appendix/implementation.html
@@ -1817,6 +2461,7 @@ def canon_thread_hw_concurrency():
 [Unicode Code Point]: https://unicode.org/glossary/#code_point
 [Surrogate]: https://unicode.org/faq/utf_bom.html#utf16-2
 [Name Mangling]: https://en.wikipedia.org/wiki/Name_mangling
+[Asyncify]: https://emscripten.org/docs/porting/asyncify.html
 
 [`import_name`]: https://clang.llvm.org/docs/AttributeReference.html#import-name
 [`export_name`]: https://clang.llvm.org/docs/AttributeReference.html#export-name
diff --git a/design/mvp/Explainer.md b/design/mvp/Explainer.md
index d70d5ffc..461b41e0 100644
--- a/design/mvp/Explainer.md
+++ b/design/mvp/Explainer.md
@@ -1,8 +1,8 @@
-# Component Model Explainer
+# Component Model AST Explainer
 
-This explainer walks through the assembly-level definition of a
-[component](../high-level) and the proposed embedding of components into native
-JavaScript runtimes. For a more user-focused explanation, take a look at the
+This explainer walks through the grammar of a [component](../high-level) and
+the proposed embedding of components into native JavaScript runtimes. For a
+more user-focused explanation, take a look at the
 **[Component Model Documentation]**.
 
 * [Gated features](#gated-features)
@@ -21,17 +21,19 @@ JavaScript runtimes. For a more user-focused explanation, take a look at the
     * [Declarators](#declarators)
     * [Type checking](#type-checking)
   * [Canonical definitions](#canonical-definitions)
-    * [Canonical ABI](#canonical-built-ins)
+    * [Canonical ABI](#canonical-abi)
     * [Canonical built-ins](#canonical-built-ins)
-  * [Value definitions](#value-definitions)
-  * [Start definitions](#start-definitions)
+      * [Resource built-ins](#resource-built-ins)
+      * [Async built-ins](#-async-built-ins)
+      * [Threading built-ins](#-threading-built-ins)
+  * [Value definitions](#-value-definitions)
+  * [Start definitions](#-start-definitions)
   * [Import and export definitions](#import-and-export-definitions)
 * [Component invariants](#component-invariants)
 * [JavaScript embedding](#JavaScript-embedding)
   * [JS API](#JS-API)
   * [ESM-integration](#ESM-integration)
 * [Examples](#examples)
-* [TODO](#TODO)
 
 ## Gated Features
 
@@ -43,6 +45,7 @@ emoji symbols listed below; these emojis will be removed once they are
 implemented, considered stable and included in a future milestone:
 * 🪙: value imports/exports and component-level start function
 * 🪺: nested namespaces and packages in import/export names
+* 🔀: async
 * 🧵: threading built-ins
 
 (Based on the previous [scoping and layering] proposal to the WebAssembly CG,
@@ -548,7 +551,7 @@ defvaltype    ::= bool
                 | (borrow <typeidx>)
 valtype       ::= <typeidx>
                 | <defvaltype>
-resourcetype  ::= (resource (rep i32) (dtor <funcidx>)?)
+resourcetype  ::= (resource (rep i32) (dtor async? <funcidx> (callback <funcidx>)?)?)
 functype      ::= (func <paramlist> <resultlist>)
 paramlist     ::= (param "<label>" <valtype>)*
 resultlist    ::= (result "<label>" <valtype>)*
@@ -705,7 +708,9 @@ is currently fixed to `i32`, but will be relaxed in the future (to at least
 include `i64`, but also potentially other types). When the last handle to a
 resource is dropped, the resource's destructor function specified by the `dtor`
 immediate will be called (if present), allowing the implementing component to
-perform clean-up like freeing linear memory allocations.
+perform clean-up like freeing linear memory allocations. Destructors can be
+declared `async`, with the same meaning for the `async` and `callback`
+immediates as described below for `canon lift`.
 
 The `instance` type constructor describes a list of named, typed definitions
 that can be imported or exported by a component. Informally, instance types
@@ -1170,6 +1175,8 @@ canonopt ::= string-encoding=utf8
            | (memory <core:memidx>)
            | (realloc <core:funcidx>)
            | (post-return <core:funcidx>)
+           | async 🔀
+           | (callback <core:funcidx>) 🔀
 ```
 While the production `externdesc` accepts any `sort`, the validation rules
 for `canon lift` would only allow the `func` sort. In the future, other sorts
@@ -1208,9 +1215,25 @@ destructors called. This immediate is always optional but, if present, is
 validated to have parameters matching the callee's return type and empty
 results.
 
-Based on this description of the AST, the [Canonical ABI explainer][Canonical
-ABI] gives a detailed walkthrough of the static and dynamic semantics of `lift`
-and `lower`.
+🔀 The `async` option specifies that the component wants to make (for imports)
+or support (for exports) multiple concurrent (asynchronous) calls. This option
+can be applied to any component-level function type and changes the derived
+Canonical ABI significantly. See the [async explainer](Async.md) for more
+details.
+
+🔀 The `(callback ...)` option may only be present in `canon lift` when the
+`async` option has also been set and specifies a core function that is
+validated to have the following core function type:
+```wasm
+(func (param $ctx i32)
+      (param $event i32)
+      (param $payload i32)
+      (result $done i32))
+```
+Again, see the [async explainer](Async.md) for more details.
+
+Based on this description of the AST, the [Canonical ABI explainer] gives a
+detailed walkthrough of the static and dynamic semantics of `lift` and `lower`.
 
 One high-level consequence of the dynamic semantics of `canon lift` given in
 the Canonical ABI explainer is that component functions are different from core
@@ -1288,22 +1311,30 @@ async is added to the proposal, [tasks][Future and Stream Types]).
 ```ebnf
 canon ::= ...
         | (canon resource.new <typeidx> (core func <id>?))
-        | (canon resource.drop <typeidx> (core func <id>?))
+        | (canon resource.drop <typeidx> async? (core func <id>?))
         | (canon resource.rep <typeidx> (core func <id>?))
+        | (canon task.start <core:typeidx> (core func <id>?)) 🔀
+        | (canon task.return <core:typeidx> (core func <id>?)) 🔀
+        | (canon task.wait (core func <id>?)) 🔀
+        | (canon task.poll (core func <id>?)) 🔀
+        | (canon task.yield (core func <id>?)) 🔀
         | (canon thread.spawn <typeidx> (core func <id>?)) 🧵
         | (canon thread.hw_concurrency (core func <id>?)) 🧵
 ```
 
-##### Resources
+##### Resource built-ins
 
 The `resource.new` built-in has type `[i32] -> [i32]` and creates a new
 resource (with resource type `typeidx`) with the given `i32` value as its
 representation and returning the `i32` index of a new handle pointing to this
 resource.
 
-The `resource.drop` built-in has type `[i32] -> []` and drops a resource handle
-(with resource type `typeidx`) at the given `i32` index. If the dropped handle
-owns the resource, the resource's `dtor` is called, if present.
+The `resource.drop` drops a resource handle (with resource type `typeidx`) at
+a given `i32` index. If the dropped handle owns the resource, the resource's
+`dtor` is called, if present. If `async` is not specified, the core function
+type of `resource.drop` is `[i32] -> []`. Otherwise, the core function type
+of `async` `drop` is `[i32] -> [i32]`, where the returned `i32` is either `0`
+(if the drop completed eagerly) or the index of the in-progress drop.
 
 The `resource.rep` built-in has type `[i32] -> [i32]` and returns the `i32`
 representation of the resource (with resource type `typeidx`) pointed to by the
@@ -1338,7 +1369,50 @@ Here, the `i32` returned by `resource.new`, which is an index into the
 component's handle-table, is immediately returned by `make_R`, thereby
 transferring ownership of the newly-created resource to the export's caller.
 
-##### 🧵 Threads
+##### 🔀 Async built-ins
+
+See the [async explainer](Async.md) for high-level context and terminology
+and the [Canonical ABI explainer] for detailed runtime semantics.
+
+The `task.start` built-in returns the arguments to the currently-executing
+task. This built-in must be called from an `async`-lifted export exactly once
+per export activation. Delaying the call to `task.start` allows the async
+callee to exert *backpressure* on the caller. The `canon task.start` definition
+takes the type index of a core function type and produces a core function with
+exactly that type. When called, the declared core function type is checked
+to match the lowered function type of a component-level function returning the
+parameter types of the current task. (See also [Starting](Async.md#starting) in
+the async explainer and [`canon_task_start`] in the Canonical ABI explainer.)
+
+The `task.return` built-in takes as parameters the result values of the
+currently-executing task. This built-in must be called from an `async`-lifted
+export exactly once per export activation after `task.start`. After calling
+`task.return`, the callee can continue executing for an arbitrary amount of
+time before returning to the caller. The `canon task.return` definition takes
+the type index of a core function type and produces a core function with
+exactly that type. When called, the declared core function type is checked
+to match the lowered function type of a component-level function taking the
+result types of the current task. (See also [Returning](Async.md#returning) in
+the async explainer and [`canon_task_return`] in the Canonical ABI explainer.)
+
+The `task.wait` built-in has type `[i32] -> [i32]`, returning an event and
+storing the 4-byte payload of the event at the address passed as parameter.
+`task.wait` can be called whether or not `async` was present, allowing any sort
+of code to synchronously wait for progress on any of the currently-executing
+subtasks. (See also [Waiting](Async.md#waiting) in the async explainer and
+[`canon_task_wait`] in the Canonical ABI explainer.)
+
+The `task.poll` built-in has type `[i32] -> [i32]`, returning whether or not
+an event was immediately available as a boolean and, if true, storing the
+event and its payload in the buffer pointed to by the `i32` parameter.
+(See also [`canon_task_poll`] in the Canonical ABI explainer.)
+
+The `task.yield` built-in has type `[] -> []` and simply allows the runtime to
+switch to another task, allowing a long-running computation to cooperatively
+interleave with other tasks. (See also [`canon_task_yield`] in the Canonical
+ABI explainer.)
+
+##### 🧵 Threading built-ins
 
 The [shared-everything-threads] proposal adds component model built-ins for
 thread management. These are specified as built-ins and not core WebAssembly
@@ -1831,8 +1905,7 @@ three runtime invariants:
    the interim. This establishes a clear contract between separate components
    that both prevents obscure composition-time bugs and also enables
    more-efficient non-reentrant runtime glue code (particularly in the middle
-   of the [Canonical ABI](CanonicalABI.md)). This implies that components by
-   default don't allow concurrency and multi-threaded access will trap.
+   of the [Canonical ABI](CanonicalABI.md)).
 
 
 ## JavaScript Embedding
@@ -2069,16 +2142,6 @@ For some use-case-focused, worked examples, see:
 * [Component Examples presentation](https://docs.google.com/presentation/d/11lY9GBghZJ5nCFrf4MKWVrecQude0xy_buE--tnO9kQ)
 
 
-## TODO
-
-The following features are needed to address the [MVP Use Cases](../high-level/UseCases.md)
-and will be added over the coming months to complete the MVP proposal:
-* concurrency support ([slides][Future And Stream Types])
-* optional imports, definitions and exports (subsuming
-  [WASI Optional Imports](https://github.com/WebAssembly/WASI/blob/main/legacy/optional-imports.md)
-  and maybe [conditional-sections](https://github.com/WebAssembly/conditional-sections/issues/22))
-
-
 
 [Structure Section]: https://webassembly.github.io/spec/core/syntax/index.html
 [Text Format Section]: https://webassembly.github.io/spec/core/text/index.html
@@ -2173,7 +2236,12 @@ and will be added over the coming months to complete the MVP proposal:
 [WASI Preview 2]: https://github.com/WebAssembly/WASI/tree/main/preview2
 
 [Adapter Functions]: FutureFeatures.md#custom-abis-via-adapter-functions
-[Canonical ABI]: CanonicalABI.md
+[Canonical ABI explainer]: CanonicalABI.md
+[`canon_task_start`]: CanonicalABI.md#-canon-taskstart
+[`canon_task_return`]: CanonicalABI.md#-canon-taskreturn
+[`canon_task_wait`]: CanonicalABI.md#-canon-taskwait
+[`canon_task_poll`]: CanonicalABI.md#-canon-taskpoll
+[`canon_task_yield`]: CanonicalABI.md#-canon-taskyield
 [Shared-Nothing]: ../high-level/Choices.md
 [Use Cases]: ../high-level/UseCases.md
 [Host Embeddings]: ../high-level/UseCases.md#hosts-embedding-components
diff --git a/design/mvp/canonical-abi/definitions.py b/design/mvp/canonical-abi/definitions.py
index 59299eb7..350ac404 100644
--- a/design/mvp/canonical-abi/definitions.py
+++ b/design/mvp/canonical-abi/definitions.py
@@ -6,10 +6,13 @@
 
 from __future__ import annotations
 from dataclasses import dataclass
+from functools import partial
 from typing import Optional, Callable, MutableMapping, TypeVar, Generic
+from enum import IntEnum
 import math
 import struct
 import random
+import asyncio
 
 class Trap(BaseException): pass
 class CoreWebAssemblyException(BaseException): pass
@@ -46,6 +49,8 @@ class ModuleType(ExternType):
 class CoreFuncType(CoreExternType):
   params: list[str]
   results: list[str]
+  def __eq__(self, other):
+    return self.params == other.params and self.results == other.results
 
 @dataclass
 class CoreMemoryType(CoreExternType):
@@ -282,17 +287,29 @@ class CanonicalOptions:
   string_encoding: Optional[str] = None
   realloc: Optional[Callable] = None
   post_return: Optional[Callable] = None
+  sync: bool = True # = !canonopt.async
+  callback: Optional[Callable] = None
 
 class ComponentInstance:
   # core module instance state
   may_leave: bool
-  may_enter: bool
+  may_enter_sync: bool
+  may_enter_async: bool
+  pending_sync_tasks: list[asyncio.Future]
+  pending_async_tasks: list[asyncio.Future]
   handles: HandleTables
+  async_subtasks: Table[AsyncSubtask]
+  thread: asyncio.Lock
 
   def __init__(self):
     self.may_leave = True
-    self.may_enter = True
+    self.may_enter_sync = True
+    self.may_enter_async = True
+    self.pending_sync_tasks = []
+    self.pending_async_tasks = []
     self.handles = HandleTables()
+    self.async_subtasks = Table[AsyncSubtask]()
+    self.thread = asyncio.Lock()
 
 class HandleTables:
   rt_to_table: MutableMapping[ResourceType, Table[HandleElem]]
@@ -315,10 +332,14 @@ def remove(self, rt, i):
 class ResourceType(Type):
   impl: ComponentInstance
   dtor: Optional[Callable]
+  dtor_sync: bool
+  dtor_callback: Optional[Callable]
 
-  def __init__(self, impl, dtor = None):
+  def __init__(self, impl, dtor = None, dtor_sync = True, dtor_callback = None):
     self.impl = impl
     self.dtor = dtor
+    self.dtor_sync = dtor_sync
+    self.dtor_callback = dtor_callback
 
 ElemT = TypeVar('ElemT')
 class Table(Generic[ElemT]):
@@ -354,7 +375,7 @@ def remove(self, i):
 class HandleElem:
   rep: int
   own: bool
-  scope: Optional[ExportCall]
+  scope: Optional[Task]
   lend_count: int
 
   def __init__(self, rep, own, scope = None):
@@ -363,12 +384,41 @@ def __init__(self, rep, own, scope = None):
     self.scope = scope
     self.lend_count = 0
 
-class ExportCall(CallContext):
+class AsyncCallState(IntEnum):
+  STARTING = 0
+  STARTED = 1
+  RETURNED = 2
+  DONE = 3
+
+class EventCode(IntEnum):
+  CALL_STARTING = AsyncCallState.STARTING
+  CALL_STARTED = AsyncCallState.STARTED
+  CALL_RETURNED = AsyncCallState.RETURNED
+  CALL_DONE = AsyncCallState.DONE
+  YIELDED = 4
+
+class Task(CallContext):
+  caller: Optional[Task]
   borrow_count: int
+  events: asyncio.Queue[AsyncSubtask]
+  num_async_subtasks: int
 
-  def __init__(self, opts, inst):
+  def __init__(self, opts, inst, caller):
     super().__init__(opts, inst)
+    self.caller = caller
     self.borrow_count = 0
+    self.events = asyncio.Queue[AsyncSubtask]()
+    self.num_async_subtasks = 0
+
+  async def enter(self):
+    await self.inst.thread.acquire()
+    self.trap_if_on_the_stack(self.inst)
+
+  def trap_if_on_the_stack(self, inst):
+    c = self.caller
+    while c is not None:
+      trap_if(c.inst is int)
+      c = c.caller
 
   def create_borrow(self):
     self.borrow_count += 1
@@ -377,10 +427,51 @@ def drop_borrow(self):
     assert(self.borrow_count > 0)
     self.borrow_count -= 1
 
+  def add_async_subtask(self, subtask):
+    assert(subtask.supertask is None and subtask.index is None)
+    subtask.supertask = self
+    subtask.index = self.inst.async_subtasks.add(subtask)
+    self.num_async_subtasks += 1
+    return subtask.index
+
+  def async_subtask_made_progress(self, subtask):
+    assert(subtask.supertask is self)
+    if subtask.enqueued:
+      return
+    subtask.enqueued = True
+    self.events.put_nowait(subtask)
+
+  async def wait(self):
+    self.inst.thread.release()
+    subtask = await self.events.get()
+    await self.inst.thread.acquire()
+    return self.process_event(subtask)
+
+  def process_event(self, subtask):
+    assert(subtask.supertask is self)
+    subtask.enqueued = False
+    if subtask.state == AsyncCallState.DONE:
+      self.inst.async_subtasks.remove(subtask.index)
+      self.num_async_subtasks -= 1
+    return (EventCode(subtask.state), subtask.index)
+
+  def poll(self):
+    if self.events.empty():
+      return None
+    return self.process_event(self.events.get_nowait())
+
+  async def yield_(self):
+    self.inst.thread.release()
+    await asyncio.sleep(0)
+    await self.inst.thread.acquire()
+
   def exit(self):
+    assert(self.events.empty())
     trap_if(self.borrow_count != 0)
+    trap_if(self.num_async_subtasks != 0)
+    self.inst.thread.release()
 
-class ImportCall(CallContext):
+class Subtask(CallContext):
   lenders: list[HandleElem]
 
   def __init__(self, opts, inst):
@@ -392,10 +483,114 @@ def track_owning_lend(self, lending_handle):
     lending_handle.lend_count += 1
     self.lenders.append(lending_handle)
 
-  def exit(self):
+  def finish(self):
     for h in self.lenders:
       h.lend_count -= 1
 
+class SyncTask(Task):
+  async def enter(self):
+    if not self.inst.may_enter_sync:
+      f = asyncio.Future()
+      self.inst.pending_sync_tasks.append(f)
+      await f
+      assert(self.inst.may_enter_sync)
+    self.inst.may_enter_sync = False
+    await super().enter()
+
+  def exit(self):
+    super().exit()
+    assert(not self.inst.may_enter_sync)
+    self.inst.may_enter_sync = True
+    if self.inst.pending_sync_tasks:
+      self.inst.pending_sync_tasks.pop(0).set_result(None)
+
+class AsyncTask(Task):
+  ft: FuncType
+  start_thunk: Callable
+  return_thunk: Callable
+  state: AsyncCallState
+  unblock_next_pending: bool
+
+  def __init__(self, opts, inst, caller, ft, start_thunk, return_thunk):
+    super().__init__(opts, inst, caller)
+    self.ft = ft
+    self.start_thunk = start_thunk
+    self.return_thunk = return_thunk
+    self.state = AsyncCallState.STARTING
+    self.unblock_next_pending = False
+
+  async def enter(self):
+    if not self.inst.may_enter_async or self.inst.pending_async_tasks:
+      f = asyncio.Future()
+      self.inst.pending_async_tasks.append(f)
+      await f
+      assert(self.inst.may_enter_async)
+      self.unblock_next_pending = len(self.inst.pending_async_tasks) > 0
+    await super().enter()
+
+  async def wait(self):
+    if self.state == AsyncCallState.STARTING:
+      self.inst.may_enter_async = False
+      self.inst.unblock_next_pending = False
+    else:
+      self.maybe_unblock_next_pending()
+    return await super().wait()
+
+  def maybe_unblock_next_pending(self):
+    if self.unblock_next_pending:
+      self.unblock_next_pending = False
+      assert(self.inst.may_enter_async)
+      self.inst.pending_async_tasks.pop(0).set_result(None)
+
+  def start(self):
+    trap_if(self.state != AsyncCallState.STARTING)
+    self.state = AsyncCallState.STARTED
+    if not self.inst.may_enter_async:
+      self.inst.may_enter_async = True
+      self.unblock_next_pending = len(self.inst.pending_async_tasks) > 0
+
+  def return_(self):
+    trap_if(self.state != AsyncCallState.STARTED)
+    self.state = AsyncCallState.RETURNED
+
+  def exit(self):
+    super().exit()
+    trap_if(self.state != AsyncCallState.RETURNED)
+    self.state = AsyncCallState.DONE
+    self.maybe_unblock_next_pending()
+
+class AsyncSubtask(Subtask):
+  state: AsyncCallState
+  supertask: Optional[Task]
+  index: Optional[int]
+  enqueued: bool
+
+  def __init__(self, opts, inst):
+    super().__init__(opts, inst)
+    self.state = AsyncCallState.STARTING
+    self.supertask = None
+    self.index = None
+    self.enqueued = False
+
+  def start(self):
+    assert(self.state == AsyncCallState.STARTING)
+    self.state = AsyncCallState.STARTED
+    if self.supertask is not None:
+      self.supertask.async_subtask_made_progress(self)
+
+  def return_(self):
+    assert(self.state == AsyncCallState.STARTED)
+    self.state = AsyncCallState.RETURNED
+    if self.supertask is not None:
+      self.supertask.async_subtask_made_progress(self)
+
+  def finish(self):
+    super().finish()
+    assert(self.state == AsyncCallState.RETURNED)
+    self.state = AsyncCallState.DONE
+    if self.supertask is not None:
+      self.supertask.async_subtask_made_progress(self)
+
 ### Loading
 
 def load(cx, ptr, t):
@@ -563,7 +758,7 @@ def lift_own(cx, i, t):
   return h.rep
 
 def lift_borrow(cx, i, t):
-  assert(isinstance(cx, ImportCall))
+  assert(isinstance(cx, Subtask))
   h = cx.inst.handles.get(t.rt, i)
   if h.own:
     cx.track_owning_lend(h)
@@ -842,7 +1037,7 @@ def lower_own(cx, rep, t):
   return cx.inst.handles.add(t.rt, h)
 
 def lower_borrow(cx, rep, t):
-  assert(isinstance(cx, ExportCall))
+  assert(isinstance(cx, Task))
   if cx.inst is t.rt.impl:
     return rep
   h = HandleElem(rep, own=False, scope=cx)
@@ -854,21 +1049,31 @@ def lower_borrow(cx, rep, t):
 MAX_FLAT_PARAMS = 16
 MAX_FLAT_RESULTS = 1
 
-def flatten_functype(ft, context):
-  flat_params = flatten_types(ft.param_types())
-  if len(flat_params) > MAX_FLAT_PARAMS:
-    flat_params = ['i32']
-
-  flat_results = flatten_types(ft.result_types())
-  if len(flat_results) > MAX_FLAT_RESULTS:
+def flatten_functype(opts, ft, context):
+  if opts.sync:
+    flat_params = flatten_types(ft.param_types())
+    if len(flat_params) > MAX_FLAT_PARAMS:
+      flat_params = ['i32']
+
+    flat_results = flatten_types(ft.result_types())
+    if len(flat_results) > MAX_FLAT_RESULTS:
+      match context:
+        case 'lift':
+          flat_results = ['i32']
+        case 'lower':
+          flat_params += ['i32']
+          flat_results = []
+
+    return CoreFuncType(flat_params, flat_results)
+  else:
     match context:
       case 'lift':
-        flat_results = ['i32']
-      case 'lower':
-        flat_params += ['i32']
+        flat_params = []
         flat_results = []
-
-  return CoreFuncType(flat_params, flat_results)
+      case 'lower':
+        flat_params = ['i32', 'i32']
+        flat_results = ['i32']
+    return CoreFuncType(flat_params, flat_results)
 
 def flatten_types(ts):
   return [ft for t in ts for ft in flatten_type(t)]
@@ -1089,114 +1294,242 @@ def lower_flat_flags(v, labels):
 
 ### Lifting and Lowering Values
 
-def lift_values(cx, max_flat, vi, ts):
+def lift_sync_values(cx, max_flat, vi, ts):
   flat_types = flatten_types(ts)
   if len(flat_types) > max_flat:
-    ptr = vi.next('i32')
-    tuple_type = Tuple(ts)
-    trap_if(ptr != align_to(ptr, alignment(tuple_type)))
-    trap_if(ptr + elem_size(tuple_type) > len(cx.opts.memory))
-    return list(load(cx, ptr, tuple_type).values())
+    return lift_heap_values(cx, vi, ts)
   else:
     return [ lift_flat(cx, vi, t) for t in ts ]
 
-def lower_values(cx, max_flat, vs, ts, out_param = None):
+def lift_async_values(cx, vi, ts):
+  if len(ts) == 0:
+    _ = vi.next('i32')
+    return []
+  flat_types = flatten_types(ts)
+  if len(flat_types) == 1 and flat_types[0] == 'i32':
+    assert(len(ts) == 1)
+    return [ lift_flat(cx, vi, ts[0]) ]
+  else:
+    return lift_heap_values(cx, vi, ts)
+
+def lift_heap_values(cx, vi, ts):
+  ptr = vi.next('i32')
+  tuple_type = Tuple(ts)
+  trap_if(ptr != align_to(ptr, alignment(tuple_type)))
+  trap_if(ptr + elem_size(tuple_type) > len(cx.opts.memory))
+  return list(load(cx, ptr, tuple_type).values())
+
+def lower_sync_values(cx, max_flat, vs, ts, out_param = None):
   inst = cx.inst
   assert(inst.may_leave)
   inst.may_leave = False
-
   flat_types = flatten_types(ts)
   if len(flat_types) > max_flat:
-    tuple_type = Tuple(ts)
-    tuple_value = {str(i): v for i,v in enumerate(vs)}
-    if out_param is None:
-      ptr = cx.opts.realloc(0, 0, alignment(tuple_type), elem_size(tuple_type))
-    else:
-      ptr = out_param.next('i32')
-    trap_if(ptr != align_to(ptr, alignment(tuple_type)))
-    trap_if(ptr + elem_size(tuple_type) > len(cx.opts.memory))
-    store(cx, tuple_value, tuple_type, ptr)
-    flat_vals = [ptr]
+    flat_vals = lower_heap_values(cx, vs, ts, out_param)
   else:
     flat_vals = []
     for i in range(len(vs)):
       flat_vals += lower_flat(cx, vs[i], ts[i])
+  inst.may_leave = True
+  return flat_vals
 
+def lower_async_values(cx, vs, ts, out_param):
+  if len(ts) == 0:
+    _ = out_param.next('i32')
+    return
+  inst = cx.inst
+  assert(inst.may_leave)
+  inst.may_leave = False
+  lower_heap_values(cx, vs, ts, out_param)
   inst.may_leave = True
+
+def lower_heap_values(cx, vs, ts, out_param):
+  tuple_type = Tuple(ts)
+  tuple_value = {str(i): v for i,v in enumerate(vs)}
+  if out_param is None:
+    ptr = cx.opts.realloc(0, 0, alignment(tuple_type), elem_size(tuple_type))
+    flat_vals = [ptr]
+  else:
+    ptr = out_param.next('i32')
+    flat_vals = []
+  trap_if(ptr != align_to(ptr, alignment(tuple_type)))
+  trap_if(ptr + elem_size(tuple_type) > len(cx.opts.memory))
+  store(cx, tuple_value, tuple_type, ptr)
   return flat_vals
 
 ### `canon lift`
 
-def canon_lift(opts, inst, callee, ft, start_thunk, return_thunk):
-  export_call = ExportCall(opts, inst)
-  trap_if(not inst.may_enter)
+async def canon_lift(opts, inst, callee, ft, caller, start_thunk, return_thunk):
+  if opts.sync:
+    task = SyncTask(opts, inst, caller)
+    await task.enter()
+
+    flat_args = lower_sync_values(task, MAX_FLAT_PARAMS, start_thunk(), ft.param_types())
+    flat_results = await call_and_trap_on_throw(callee, task, flat_args)
+    return_thunk(lift_sync_values(task, MAX_FLAT_RESULTS, CoreValueIter(flat_results), ft.result_types()))
 
-  flat_args = lower_values(export_call, MAX_FLAT_PARAMS, start_thunk(), ft.param_types())
-  flat_results = call_and_trap_on_throw(callee, flat_args)
-  return_thunk(lift_values(export_call, MAX_FLAT_RESULTS, CoreValueIter(flat_results), ft.result_types()))
+    if opts.post_return is not None:
+      [] = await call_and_trap_on_throw(opts.post_return, task, flat_results)
 
-  if opts.post_return is not None:
-    call_and_trap_on_throw(opts.post_return, flat_results)
+    task.exit()
+  else:
+    task = AsyncTask(opts, inst, caller, ft, start_thunk, return_thunk)
+    await task.enter()
+
+    if not opts.callback:
+      [] = await call_and_trap_on_throw(callee, task, [])
+    else:
+      [packed_ctx] = await call_and_trap_on_throw(callee, task, [])
+      while packed_ctx != 0:
+        is_yield = bool(packed_ctx & 1)
+        ctx = packed_ctx & ~1
+        if is_yield:
+          await task.yield_()
+          event, payload = (EventCode.YIELDED, 0)
+        else:
+          event, payload = await task.wait()
+        [packed_ctx] = await call_and_trap_on_throw(opts.callback, task, [ctx, event, payload])
 
-  export_call.exit()
+    assert(opts.post_return is None)
+    task.exit()
 
-def call_and_trap_on_throw(callee, args):
+async def call_and_trap_on_throw(callee, task, args):
   try:
-    return callee(args)
+    return await callee(task, args)
   except CoreWebAssemblyException:
     trap()
 
 ### `canon lower`
 
-def canon_lower(opts, inst, callee, calling_import, ft, flat_args):
-  import_call = ImportCall(opts, inst)
+async def canon_lower(opts, callee, ft, task, flat_args):
+  inst = task.inst
   trap_if(not inst.may_leave)
+  flat_args = CoreValueIter(flat_args)
 
-  assert(inst.may_enter)
-  if calling_import:
-    inst.may_enter = False
+  if opts.sync:
+    subtask = Subtask(opts, inst)
 
-  flat_args = CoreValueIter(flat_args)
-  flat_results = None
+    def start_thunk():
+      return lift_sync_values(subtask, MAX_FLAT_PARAMS, flat_args, ft.param_types())
+
+    flat_results = None
+    def return_thunk(results):
+      nonlocal flat_results
+      flat_results = lower_sync_values(subtask, MAX_FLAT_RESULTS, results, ft.result_types(), flat_args)
+
+    await callee(task, start_thunk, return_thunk)
 
-  def start_thunk():
-    return lift_values(import_call, MAX_FLAT_PARAMS, flat_args, ft.param_types())
+    subtask.finish()
+  else:
+    subtask = AsyncSubtask(opts, inst)
+
+    async def do_call():
+      def start_thunk():
+        subtask.start()
+        return lift_async_values(subtask, flat_args, ft.param_types())
 
-  def return_thunk(results):
-    nonlocal flat_results
-    flat_results = lower_values(import_call, MAX_FLAT_RESULTS, results, ft.result_types(), flat_args)
+      def return_thunk(results):
+        subtask.return_()
+        lower_async_values(subtask, results, ft.result_types(), flat_args)
 
-  callee(start_thunk, return_thunk)
+      await callee(task, start_thunk, return_thunk)
+      subtask.finish()
 
-  if calling_import:
-    inst.may_enter = True
+    asyncio.create_task(do_call())
+    await asyncio.sleep(0) # start do_call eagerly
+
+    if subtask.state == AsyncCallState.DONE:
+      flat_results = [0]
+    else:
+      i = task.add_async_subtask(subtask)
+      assert(0 < i < 2**30)
+      assert(0 <= int(subtask.state) < 2**2)
+      flat_results = [ i | (int(subtask.state) << 30) ]
 
-  import_call.exit()
   return flat_results
 
 ### `canon resource.new`
 
-def canon_resource_new(inst, rt, rep):
+async def canon_resource_new(rt, task, rep):
   h = HandleElem(rep, own=True)
-  i = inst.handles.add(rt, h)
+  i = task.inst.handles.add(rt, h)
   return [i]
 
 ### `canon resource.drop`
 
-def canon_resource_drop(inst, rt, i):
+async def canon_resource_drop(rt, sync, task, i):
+  inst = task.inst
   h = inst.handles.remove(rt, i)
+  flat_results = [] if sync else [0]
   if h.own:
     assert(h.scope is None)
     trap_if(h.lend_count != 0)
-    trap_if(inst is not rt.impl and not rt.impl.may_enter)
-    if rt.dtor:
-      rt.dtor(h.rep)
+    if inst is rt.impl:
+      if rt.dtor:
+        await rt.dtor(h.rep)
+    else:
+      if rt.dtor:
+        caller_opts = CanonicalOptions(sync = sync)
+        callee_opts = CanonicalOptions(sync = rt.dtor_sync, callback = rt.dtor_callback)
+        ft = FuncType([U32()],[])
+        callee = partial(canon_lift, callee_opts, rt.impl, rt.dtor, ft)
+        flat_results = await canon_lower(caller_opts, callee, ft, task, [h.rep, 0])
+      else:
+        task.trap_if_on_the_stack(rt.impl)
   else:
     h.scope.drop_borrow()
-  return []
+  return flat_results
 
 ### `canon resource.rep`
 
-def canon_resource_rep(inst, rt, i):
-  h = inst.handles.get(rt, i)
+async def canon_resource_rep(rt, task, i):
+  h = task.inst.handles.get(rt, i)
   return [h.rep]
+
+### `canon task.start`
+
+async def canon_task_start(task, core_ft, flat_args):
+  assert(len(core_ft.params) == len(flat_args))
+  trap_if(task.opts.sync)
+  trap_if(core_ft != flatten_functype(CanonicalOptions(), FuncType([], task.ft.params), 'lower'))
+  task.start()
+  args = task.start_thunk()
+  flat_results = lower_sync_values(task, MAX_FLAT_RESULTS, args, task.ft.param_types(), CoreValueIter(flat_args))
+  assert(len(core_ft.results) == len(flat_results))
+  return flat_results
+
+### `canon task.return`
+
+async def canon_task_return(task, core_ft, flat_args):
+  assert(len(core_ft.params) == len(flat_args))
+  trap_if(task.opts.sync)
+  trap_if(core_ft != flatten_functype(CanonicalOptions(), FuncType(task.ft.results, []), 'lower'))
+  task.return_()
+  results = lift_sync_values(task, MAX_FLAT_PARAMS, CoreValueIter(flat_args), task.ft.result_types())
+  task.return_thunk(results)
+  assert(len(core_ft.results) == 0)
+  return []
+
+### `canon task.wait`
+
+async def canon_task_wait(task, ptr):
+  trap_if(task.opts.callback is not None)
+  event, payload = await task.wait()
+  store(task, payload, U32(), ptr)
+  return [event]
+
+### `canon task.poll`
+
+async def canon_task_poll(task, ptr):
+  ret = task.poll()
+  if ret is None:
+    return [0]
+  store(task, ret, Tuple([U32(), U32()]), ptr)
+  return [1]
+
+### `canon task.yield`
+
+async def canon_task_yield(task):
+  trap_if(task.opts.callback is not None)
+  await task.yield_()
+  return []
diff --git a/design/mvp/canonical-abi/run_tests.py b/design/mvp/canonical-abi/run_tests.py
index a04503ad..4b7e2c6a 100644
--- a/design/mvp/canonical-abi/run_tests.py
+++ b/design/mvp/canonical-abi/run_tests.py
@@ -1,5 +1,4 @@
 import definitions
-from functools import partial
 from definitions import *
 
 def equal_modulo_string_encoding(s, t):
@@ -39,6 +38,8 @@ def mk_opts(memory = bytearray(), encoding = 'utf8', realloc = None, post_return
   opts.string_encoding = encoding
   opts.realloc = realloc
   opts.post_return = post_return
+  opts.sync = True
+  opts.callback = None
   return opts
 
 def mk_cx(memory = bytearray(), encoding = 'utf8', realloc = None, post_return = None):
@@ -313,13 +314,13 @@ def test_flatten(t, params, results):
 
   if len(results) > definitions.MAX_FLAT_RESULTS:
     expect.results = ['i32']
-  got = flatten_functype(t, 'lift')
+  got = flatten_functype(CanonicalOptions(), t, 'lift')
   assert(got == expect)
 
   if len(results) > definitions.MAX_FLAT_RESULTS:
     expect.params += ['i32']
     expect.results = []
-  got = flatten_functype(t, 'lower')
+  got = flatten_functype(CanonicalOptions(), t, 'lower')
   assert(got == expect)
 
 test_flatten(FuncType([U8(),F32(),F64()],[]), ['i32','f32','f64'], [])
@@ -336,33 +337,34 @@ def test_roundtrip(t, v):
   definitions.MAX_FLAT_RESULTS = 16
 
   ft = FuncType([t],[t])
-  callee = lambda x: x
+  async def callee(task, x):
+    return x
 
   callee_heap = Heap(1000)
-  callee_opts = mk_opts(callee_heap.memory, 'utf8', callee_heap.realloc, lambda x: () )
+  callee_opts = mk_opts(callee_heap.memory, 'utf8', callee_heap.realloc)
   callee_inst = ComponentInstance()
   lifted_callee = partial(canon_lift, callee_opts, callee_inst, callee, ft)
 
   caller_heap = Heap(1000)
   caller_opts = mk_opts(caller_heap.memory, 'utf8', caller_heap.realloc)
   caller_inst = ComponentInstance()
-  caller_cx = CallContext(caller_opts, caller_inst)
+  caller_task = Task(caller_opts, caller_inst, None)
 
   return_in_heap = len(flatten_types([t])) > definitions.MAX_FLAT_RESULTS
 
-  flat_args = lower_values(caller_cx, definitions.MAX_FLAT_PARAMS, [v], [t])
+  asyncio.run(caller_task.enter())
+  flat_args = lower_sync_values(caller_task, definitions.MAX_FLAT_PARAMS, [v], [t])
   if return_in_heap:
     flat_args += [ caller_heap.realloc(0, 0, alignment(t), elem_size(t)) ]
-  flat_results = canon_lower(caller_opts, caller_inst, lifted_callee, True, ft, flat_args)
+  flat_results = asyncio.run(canon_lower(caller_opts, lifted_callee, ft, caller_task, flat_args))
   if return_in_heap:
     flat_results = [ flat_args[-1] ]
-  [got] = lift_values(caller_cx, definitions.MAX_FLAT_PARAMS, CoreValueIter(flat_results), [t])
+  [got] = lift_sync_values(caller_task, definitions.MAX_FLAT_PARAMS, CoreValueIter(flat_results), [t])
+  caller_task.exit()
 
   if got != v:
     fail("test_roundtrip({},{}) got {}".format(t, v, got))
 
-  assert(caller_inst.may_leave and caller_inst.may_enter)
-  assert(callee_inst.may_leave and callee_inst.may_enter)
   definitions.MAX_FLAT_RESULTS = before
 
 test_roundtrip(S8(), -1)
@@ -379,24 +381,24 @@ def test_handles():
   definitions.MAX_FLAT_RESULTS = 16
 
   dtor_value = None
-  def dtor(x):
+  async def dtor(task, args):
     nonlocal dtor_value
-    dtor_value = x
+    assert(len(args) == 1)
+    dtor_value = args[0]
 
   rt = ResourceType(ComponentInstance(), dtor) # usable in imports and exports
-
   inst = ComponentInstance()
   rt2 = ResourceType(inst, dtor) # only usable in exports
   opts = mk_opts()
 
-  def host_import(start_thunk, return_thunk):
+  async def host_import(entered, start_thunk, return_thunk):
     args = start_thunk()
     assert(len(args) == 2)
     assert(args[0] == 42)
     assert(args[1] == 44)
     return_thunk([45])
 
-  def core_wasm(args):
+  async def core_wasm(task, args):
     nonlocal dtor_value
 
     assert(len(args) == 4)
@@ -406,9 +408,9 @@ def core_wasm(args):
     assert(args[1] == 2)
     assert(args[2] == 3)
     assert(args[3] == 13)
-    assert(canon_resource_rep(inst, rt, 1)[0] == 42)
-    assert(canon_resource_rep(inst, rt, 2)[0] == 43)
-    assert(canon_resource_rep(inst, rt, 3)[0] == 44)
+    assert((await canon_resource_rep(rt, task, 1))[0] == 42)
+    assert((await canon_resource_rep(rt, task, 2))[0] == 43)
+    assert((await canon_resource_rep(rt, task, 3))[0] == 44)
 
     host_ft = FuncType([
       Borrow(rt),
@@ -420,26 +422,26 @@ def core_wasm(args):
       1,
       3
     ]
-    results = canon_lower(opts, inst, host_import, True, host_ft, args)
+    results = await canon_lower(opts, host_import, host_ft, task, args)
     assert(len(results) == 1)
     assert(results[0] == 4)
-    assert(canon_resource_rep(inst, rt, 4)[0] == 45)
+    assert((await canon_resource_rep(rt, task, 4))[0] == 45)
 
     dtor_value = None
-    canon_resource_drop(inst, rt, 1)
+    await canon_resource_drop(rt, True, task, 1)
     assert(dtor_value == 42)
     assert(len(inst.handles.table(rt).array) == 5)
     assert(inst.handles.table(rt).array[1] is None)
     assert(len(inst.handles.table(rt).free) == 1)
 
-    h = canon_resource_new(inst, rt, 46)[0]
+    h = (await canon_resource_new(rt, task, 46))[0]
     assert(h == 1)
     assert(len(inst.handles.table(rt).array) == 5)
     assert(inst.handles.table(rt).array[1] is not None)
     assert(len(inst.handles.table(rt).free) == 0)
 
     dtor_value = None
-    canon_resource_drop(inst, rt, 3)
+    await canon_resource_drop(rt, True, task, 3)
     assert(dtor_value is None)
     assert(len(inst.handles.table(rt).array) == 5)
     assert(inst.handles.table(rt).array[3] is None)
@@ -466,7 +468,7 @@ def return_thunk(results):
     nonlocal got
     got = results
 
-  canon_lift(opts, inst, core_wasm, ft, arg_thunk, return_thunk)
+  asyncio.run(canon_lift(opts, inst, core_wasm, ft, None, arg_thunk, return_thunk))
 
   assert(len(got) == 3)
   assert(got[0] == 46)
@@ -479,4 +481,290 @@ def return_thunk(results):
 
 test_handles()
 
+async def test_async_to_async():
+  producer_heap = Heap(10)
+  producer_opts = mk_opts(producer_heap.memory)
+  producer_opts.sync = False
+
+  producer_inst = ComponentInstance()
+
+  eager_ft = FuncType([], [U8()])
+  async def core_eager_producer(task, args):
+    assert(len(args) == 0)
+    [] = await canon_task_start(task, CoreFuncType([],[]), [])
+    [] = await canon_task_return(task, CoreFuncType(['i32'],[]), [43])
+    return []
+  eager_callee = partial(canon_lift, producer_opts, producer_inst, core_eager_producer, eager_ft)
+
+  fut1, fut2, fut3 = asyncio.Future(), asyncio.Future(), asyncio.Future()
+  blocking_ft = FuncType([U8(),U8()], [U8()])
+  async def blocking_callee(entered, start_thunk, return_thunk):
+    await fut1
+    [x,y] = start_thunk()
+    assert(x == 83)
+    assert(y == 84)
+    await fut2
+    return_thunk([44])
+    await fut3
+
+  consumer_heap = Heap(10)
+  consumer_opts = mk_opts(consumer_heap.memory)
+  consumer_opts.sync = False
+
+  async def consumer(task, args):
+    assert(len(args) == 0)
+
+    [b] = await canon_task_start(task, CoreFuncType([],['i32']), [])
+
+    ptr = consumer_heap.realloc(0, 0, 1, 1)
+    [ret] = await canon_lower(consumer_opts, eager_callee, eager_ft, task, [0, ptr])
+    assert(ret == 0)
+    u8 = consumer_heap.memory[ptr]
+    assert(u8 == 43)
+
+    argp = consumer_heap.realloc(0, 0, 1, 2)
+    retp = ptr
+    consumer_heap.memory[argp] = 13
+    consumer_heap.memory[argp+1] = 14
+    consumer_heap.memory[retp] = 15
+    [ret] = await canon_lower(consumer_opts, blocking_callee, blocking_ft, task, [argp, retp])
+    assert(consumer_heap.memory[retp] == 15)
+    assert(task.num_async_subtasks == 1)
+    assert(ret == 1)
+    consumer_heap.memory[argp] = 83
+    consumer_heap.memory[argp+1] = 84
+    fut1.set_result(None)
+    event, callidx = await task.wait()
+    assert(event == EventCode.CALL_STARTED)
+    assert(callidx == 1)
+    assert(consumer_heap.memory[retp] == 15)
+    fut2.set_result(None)
+    event, callidx = await task.wait()
+    assert(event == EventCode.CALL_RETURNED)
+    assert(callidx == 1)
+    assert(consumer_heap.memory[retp] == 44)
+    fut3.set_result(None)
+    assert(task.num_async_subtasks == 1)
+    event, callidx = await task.wait()
+    assert(event == EventCode.CALL_DONE)
+    assert(callidx == 1)
+    assert(task.num_async_subtasks == 0)
+
+    dtor_fut = asyncio.Future()
+    dtor_value = None
+    async def dtor(task, args):
+      nonlocal dtor_value
+      assert(len(args) == 1)
+      await dtor_fut
+      dtor_value = args[0]
+    rt = ResourceType(producer_inst, dtor)
+
+    [i] = await canon_resource_new(rt, task, 50)
+    assert(i == 1)
+    assert(dtor_value is None)
+    [ret] = await canon_resource_drop(rt, False, task, 1)
+    assert(ret == (1 | (AsyncCallState.STARTED << 30)))
+    assert(task.num_async_subtasks == 1)
+    assert(dtor_value is None)
+    dtor_fut.set_result(None)
+    event, callidx = await task.wait()
+    assert(event == AsyncCallState.DONE)
+    assert(callidx == 1)
+    assert(task.num_async_subtasks == 0)
+
+    [] = await canon_task_return(task, CoreFuncType(['i32'],[]), [42])
+    return []
+
+  ft = FuncType([Bool()],[U8()])
+
+  def arg_thunk():
+    return [ True ]
+
+  got = None
+  def return_thunk(results):
+    nonlocal got
+    got = results
+
+  consumer_inst = ComponentInstance()
+  await canon_lift(consumer_opts, consumer_inst, consumer, ft, None, arg_thunk, return_thunk)
+  assert(len(got) == 1)
+  assert(got[0] == 42)
+
+asyncio.run(test_async_to_async())
+
+async def test_async_callback():
+  producer_inst = ComponentInstance()
+
+  producer_ft = FuncType([], [])
+  async def producer(fut, entered, start_thunk, return_thunk):
+    start_thunk()
+    await fut
+    return_thunk([])
+
+  fut1 = asyncio.Future()
+  producer1 = partial(producer, fut1)
+  fut2 = asyncio.Future()
+  producer2 = partial(producer, fut2)
+
+  consumer_ft = FuncType([],[U32()])
+  async def consumer(task, args):
+    assert(len(args) == 0)
+    [] = await canon_task_start(task, CoreFuncType([],[]), [])
+
+    [ret] = await canon_lower(opts, producer1, producer_ft, task, [0, 0])
+    assert(ret == (1 | (AsyncCallState.STARTED << 30)))
+
+    [ret] = await canon_lower(opts, producer2, producer_ft, task, [0, 0])
+    assert(ret == (2 | (AsyncCallState.STARTED << 30)))
+
+    fut1.set_result(None)
+    return [42]
+
+  async def callback(task, args):
+    assert(len(args) == 3)
+    if args[0] == 42:
+      assert(args[1] == EventCode.CALL_DONE)
+      assert(args[2] == 1)
+      return [53]
+    elif args[0] == 52:
+      assert(args[1] == EventCode.YIELDED)
+      assert(args[2] == 0)
+      fut2.set_result(None)
+      return [62]
+    else:
+      assert(args[0] == 62)
+      assert(args[1] == EventCode.CALL_DONE)
+      assert(args[2] == 2)
+      [] = await canon_task_return(task, CoreFuncType(['i32'],[]), [83])
+      return [0]
+
+  consumer_inst = ComponentInstance()
+  def arg_thunk(): return []
+
+  got = None
+  def return_thunk(results):
+    nonlocal got
+    got = results
+
+  opts = mk_opts()
+  opts.sync = False
+  opts.callback = callback
+
+  await canon_lift(opts, consumer_inst, consumer, consumer_ft, None, arg_thunk, return_thunk)
+  assert(got[0] == 83)
+
+asyncio.run(test_async_callback())
+
+async def test_async_to_sync():
+  producer_opts = CanonicalOptions()
+  producer_inst = ComponentInstance()
+
+  producer_ft = FuncType([],[])
+  fut = asyncio.Future()
+  producer1_done = False
+  async def producer1_core(task, args):
+    nonlocal producer1_done
+    assert(len(args) == 0)
+    await fut
+    producer1_done = True
+    return []
+
+  producer2_done = False
+  async def producer2_core(task, args):
+    nonlocal producer2_done
+    assert(len(args) == 0)
+    assert(producer1_done == True)
+    producer2_done = True
+    return []
+
+  producer1 = partial(canon_lift, producer_opts, producer_inst, producer1_core, producer_ft)
+  producer2 = partial(canon_lift, producer_opts, producer_inst, producer2_core, producer_ft)
+
+  consumer_opts = mk_opts()
+  consumer_opts.sync = False
+
+  consumer_ft = FuncType([],[U8()])
+  async def consumer(task, args):
+    assert(len(args) == 0)
+
+    [ret] = await canon_lower(consumer_opts, producer1, producer_ft, task, [0, 0])
+    assert(ret == (1 | (AsyncCallState.STARTED << 30)))
+
+    [ret] = await canon_lower(consumer_opts, producer2, producer_ft, task, [0, 0])
+    assert(ret == (2 | (AsyncCallState.STARTING << 30)))
+
+    assert(task.poll() is None)
+
+    fut.set_result(None)
+    assert(producer1_done == False)
+    event, callidx = await task.wait()
+    assert(event == EventCode.CALL_DONE)
+    assert(callidx == 1)
+    assert(producer1_done == True)
+
+    assert(producer2_done == False)
+    await canon_task_yield(task)
+    assert(producer2_done == True)
+    event, callidx = task.poll()
+    assert(event == EventCode.CALL_DONE)
+    assert(callidx == 2)
+    assert(producer2_done == True)
+
+    assert(task.poll() is None)
+
+    await canon_task_start(task, CoreFuncType([],[]), [])
+    await canon_task_return(task, CoreFuncType(['i32'],[]), [83])
+    return []
+
+  consumer_inst = ComponentInstance()
+  def arg_thunk(): return []
+
+  got = None
+  def return_thunk(results):
+    nonlocal got
+    got = results
+
+  await canon_lift(consumer_opts, consumer_inst, consumer, consumer_ft, None, arg_thunk, return_thunk)
+  assert(got[0] == 83)
+
+asyncio.run(test_async_to_sync())
+
+async def test_sync_using_wait():
+  inst = ComponentInstance()
+  ft = FuncType([], [])
+
+  async def hostcall(fut, entered, start_thunk, return_thunk):
+    start_thunk()
+    await fut
+    return_thunk([])
+  fut1 = asyncio.Future()
+  hostcall1 = partial(hostcall, fut1)
+  fut2 = asyncio.Future()
+  hostcall2 = partial(hostcall, fut2)
+
+  lower_opts = mk_opts()
+  lower_opts.sync = False
+
+  async def core_func(task, args):
+    [ret] = await canon_lower(lower_opts, hostcall1, ft, task, [0,0])
+    assert(ret == (1 | (AsyncCallState.STARTED << 30)))
+    [ret] = await canon_lower(lower_opts, hostcall2, ft, task, [0,0])
+    assert(ret == (2 | (AsyncCallState.STARTED << 30)))
+
+    fut1.set_result(None)
+    event, callidx = await task.wait()
+    assert(event == EventCode.CALL_DONE)
+    assert(callidx == 1)
+    fut2.set_result(None)
+    event, callidx = await task.wait()
+    assert(event == EventCode.CALL_DONE)
+    assert(callidx == 2)
+    return []
+
+  def arg_thunk(): return []
+  def return_thunk(results): pass
+  await canon_lift(mk_opts(), inst, core_func, ft, None, arg_thunk, return_thunk)
+
+asyncio.run(test_sync_using_wait())
+
 print("All tests passed")