@@ -340,6 +340,9 @@ canon    ::= 0x00 0x00 f:<core:funcidx> opts:<opts> ft:<typeidx> => (canon lift
            | 0x2b 0x00                                           => (canon thread.yield-then-resume (core func)) 🧵
            | 0x2c 0x00                                           => (canon thread.suspend-then-promote (core func)) 🧵
            | 0x2d 0x00                                           => (canon thread.yield-then-promote (core func)) 🧵
+           | 0x30                                                => (canon thread.get-task (core func)) 🧵
+           | 0x31                                                => (canon thread.set-task (core func)) 🧵
+           | 0x32                                                => (canon task.drop (core func)) 🧵
            | 0x40 shared?:<sh?> ft:<core:typeidx>                => (canon thread.spawn-ref shared? ft (core func)) 🧵②
            | 0x41 shared?:<sh?> ft:<core:typeidx> tbl:<core:tableidx> => (canon thread.spawn-indirect shared? ft tbl (core func)) 🧵②
            | 0x42 shared?:<sh?>                                  => (canon thread.available-parallelism shared? (core func)) 🧵②
 
@@ -65,6 +65,9 @@ specified here.
   * [`canon thread.yield-then-resume`](#-canon-threadyield-then-resume) 🧵
   * [`canon thread.suspend-then-promote`](#-canon-threadsuspend-then-promote) 🧵
   * [`canon thread.yield-then-promote`](#-canon-threadyield-then-promote) 🧵
+  * [`canon thread.get-task`](#-canon-threadget-task) 🧵
+  * [`canon thread.set-task`](#-canon-threadset-task) 🧵
+  * [`canon task.drop`](#-canon-taskdrop) 🧵
   * [`canon error-context.new`](#-canon-error-contextnew) 📝
   * [`canon error-context.debug-message`](#-canon-error-contextdebug-message) 📝
   * [`canon error-context.drop`](#-canon-error-contextdrop) 📝
@@ -119,7 +122,7 @@ into the `handles` or `threads` fields of `ComponentInstance`.
 ```python
 class ComponentInstance:
   store: Store
-  handles: Table[ResourceHandle | Waitable | WaitableSet | ErrorContext]
+  handles: Table[ResourceHandle | Waitable | WaitableSet | ErrorContext | Task]
   threads: Table[Thread]
   may_leave: bool
   backpressure: int
@@ -609,8 +612,7 @@ class Task:
   on_resolve: OnResolve
   state: State
   num_borrows: int
-  implicit_thread: Optional[Thread]
-  threads: list[Thread]
+  waiting_to_enter: Optional[Thread]
 
   def __init__(self, ft, opts, inst, on_start, on_resolve):
     self.ft = ft
@@ -620,22 +622,12 @@ class Task:
     self.on_resolve = on_resolve
     self.state = Task.State.INITIAL
     self.num_borrows = 0
-    self.implicit_thread = None
-    self.threads = []
+    self.waiting_to_enter = None
 ```
 
 The `Task.needs_exclusive` predicate returns whether this task's implicit thread
-(`Task.implicit_thread`) has *not* opted in to multiple concurrent linear memory
-shadow stacks (via "stackful" lift) and thus, according to [Component Invariant]
-#2, requires serialization with all the other implicit threads in the component
-instance that have similarly not opted in. This question only applies to
-`async`-typed functions, since synchronous functions can't block and thus can
-always execute in a LIFO fashion using a single linear memory shadow stack. When
-`needs_exclusive` is true, core wasm execution is gated on acquiring the
-`ComponentInstance.exclusive_thread` lock. Due to cooperativity, the
-`exclusive_thread` "lock" is simply a mutable field holding either `None`, when
-unlocked, or, when locked, a reference to the `Task.implicit_thread` currently
-holding the lock.
+requires the `ComponentInstance.exclusive_thread` lock to acquired and released
+in order to enforce [Component Invariant] #2.
 ```python
   def needs_exclusive(self):
     assert(self.ft.async_)
@@ -668,29 +660,24 @@ exports.
 ```python
   def enter_implicit_thread(self):
     assert(self.state == Task.State.INITIAL)
-    self.implicit_thread = current_thread()
     if self.ft.async_:
       def has_backpressure():
         return (self.inst.backpressure > 0 or
                 (self.needs_exclusive() and self.inst.exclusive_thread is not None))
       if has_backpressure() or self.inst.num_waiting_to_enter > 0:
+        self.waiting_to_enter = current_thread()
         self.inst.num_waiting_to_enter += 1
-        self.implicit_thread.wait_until(lambda: not has_backpressure())
+        current_thread().wait_until(lambda: not has_backpressure())
         self.inst.num_waiting_to_enter -= 1
+        self.waiting_to_enter = None
         if self.deliver_pending_cancel():
           self.cancel()
           return False
       if self.needs_exclusive():
         assert(self.inst.exclusive_thread is None)
-        self.inst.exclusive_thread = self.implicit_thread
-    self.register_thread(self.implicit_thread)
+        self.inst.exclusive_thread = current_thread()
+    current_thread().index = self.inst.threads.add(current_thread())
     return True
-
-  def register_thread(self, thread):
-    assert(thread not in self.threads and thread.task is self)
-    self.threads.append(thread)
-    assert(thread.index is None)
-    thread.index = self.inst.threads.add(thread)
 ```
 Since the order in which suspended threads are resumed is nondeterministic (see
 `Store.tick` below), once `Task.enter_implicit_thread` suspends the task's
@@ -704,58 +691,44 @@ above definition ensures the following properties:
   backpressure (i.e., disabling backpressure never unleashes an unstoppable
   thundering herd of pending tasks).
 
-Once a task's implicit thread has cleared the backpressure gate, it is added to
-the lists of threads running inside the current task and component instance by
-`Task.register_thread()` (which is also called by `thread.new-indirect`, below).
+As shown above, only once a task has cleared the backpressure gate is its
+implicit thread visibly added to the component-instance-wide `threads` table.
 
 Symmetrically, the `Task.exit_implicit_thread` method is called before a task's
 implicit thread returns to reverse the effects of `Task.enter_implicit_thread`.
-In particular, if the `exclusive_thread` lock was acquired, it is released.
-`Task.unregister_thread` (which is also called by `thread.new-indirect`, below)
-traps if the task's last thread is unregistered and the task has not yet
-returned a value to its caller.
 ```python
   def exit_implicit_thread(self):
-    assert(current_thread() is self.implicit_thread)
-    self.unregister_thread(self.implicit_thread)
+    self.inst.threads.remove(current_thread().index)
     if self.ft.async_ and self.needs_exclusive():
-      assert(self.inst.exclusive_thread is self.implicit_thread)
+      assert(self.inst.exclusive_thread is current_thread())
       self.inst.exclusive_thread = None
-
-  def unregister_thread(self, thread):
-    assert(thread in self.threads and thread.task is self)
-    self.threads.remove(thread)
-    if len(self.threads) == 0:
-      trap_if(self.state != Task.State.RESOLVED)
-      assert(self.num_borrows == 0)
-    assert(thread.index is not None)
-    self.inst.threads.remove(thread.index)
 ```
 
 The `Task.request_cancellation` method implements the `OnCancel` callback that
 is called by the `subtask.cancel` built-in. If a task's implicit thread is
 waiting to start (in `Task.enter_implicit_thread`, defined above) due to
 backpressure, then it is immediately cancelled without running any guest code.
-Otherwise, if the task's implicit thread is `cancellable` and `ready`, it is
-resumed, allowing it to execute until returning or blocking. Currently, only
-`callback` threads that have returned to their event loop are `cancellable`.
-When resumed, the `callback` thread will call `Task.deliver_pending_cancel`,
-which will return `True`, leading to the `TASK_CANCELLED` event code being
-passed to the `callback` function. Ideally, the `callback` function will
-promptly call `task.cancel` (or `task.return`) to resolve the task before the
-`OnCancel` callback returns, allowing `subtask.cancel` to return without
-blocking.
+Otherwise, if the task contains one or more threads that is `cancellable` and
+`ready`, one nondeterministically picked and resumed to allow it to execute
+until returning or blocking. Currently, only `callback` threads that have
+returned to their event loop are `cancellable`. When resumed, the `callback`
+thread will call `Task.deliver_pending_cancel`, which will return `True`,
+leading to the `TASK_CANCELLED` event code being passed to the `callback`
+function. Ideally, the `callback` function will promptly call `task.cancel` (or
+`task.return`) to resolve the task before the `OnCancel` callback returns,
+allowing `subtask.cancel` to return without blocking.
 ```python
   def request_cancellation(self):
     if self.state == Task.State.INITIAL:
       self.state = Task.State.PENDING_CANCEL
-      self.implicit_thread.resume()
+      self.waiting_to_enter.resume()
       assert(self.state == Task.State.RESOLVED)
     else:
       assert(self.state == Task.State.STARTED)
       self.state = Task.State.PENDING_CANCEL
-      if self.implicit_thread.cancellable and self.implicit_thread.ready():
-        self.implicit_thread.resume()
+      candidates = { t for t in self.inst.threads if t.task is self and t.cancellable and t.ready() }
+      if candidates:
+        random.choice(list(candidates)).resume()
 
   def has_pending_cancel(self):
     return self.state == Task.State.PENDING_CANCEL
@@ -3374,10 +3347,12 @@ are lifted according to `lift_flat_values` above, the optional `post-return`
 function (specified as a `canonopt` immediate of `canon lift`) is called,
 passing the same core wasm results as parameters so that the `post-return`
 function can free any associated allocations.
+TODO: copy from set-task-old (and also alll the other stuff in Concurrency.md)
 ```python
     if not opts.async_:
       flat_results = call_and_trap_on_throw(callee, flat_args)
       assert(types_match_values(flat_ft.results, flat_results))
+      trap_if(thread.task is not task)
       result = lift_flat_values(cx, MAX_FLAT_RESULTS, CoreValueIter(flat_results), ft.result_type())
       task.return_(result)
       if opts.post_return is not None:
@@ -3394,11 +3369,11 @@ functions can always be implemented by a plain synchronous function call
 without the need for fibers which would otherwise be necessary if the
 `post-return` function performed a blocking operation.
 
-In both of the `async` cases below (with or without `callback`), the
-`task.return` built-in must be called, providing the return value as core wasm
-*parameters* to the `task.return` built-in (rather than as core function
-results as in the synchronous case). If `task.return` is *not* called by the
-time the `Task`'s last `Thread` exits, there is a trap (in `Task.unregister_thread`).
+In both of the `async` cases below (with or without `callback`), the return
+value is provided by calling the `task.return` built-in, passing the return
+value as core wasm *parameters* (rather than as core function results as in
+the synchronous case). If `task.return` never ends up being called, the
+task will never complete for the caller.
 
 In the `async` non-`callback` ("stackful async") case, there is a single call
 to the core wasm callee which must return empty core results. Waiting for async
@@ -3426,7 +3401,7 @@ function (specified as a `funcidx` immediate in `canon lift`) until the
       if thread.task.deliver_pending_cancel():
         event = (EventCode.TASK_CANCELLED, 0, 0)
       else:
-        assert(inst.exclusive_thread is task.implicit_thread)
+        assert(inst.exclusive_thread is thread)
         inst.exclusive_thread = None
         thread.cancellable = True
         match code:
@@ -3444,7 +3419,7 @@ function (specified as a `funcidx` immediate in `canon lift`) until the
             trap()
         assert(inst.exclusive_thread is None)
         thread.cancellable = False
-        inst.exclusive_thread = task.implicit_thread
+        inst.exclusive_thread = thread
       event_code, p1, p2 = event
       [packed] = call_and_trap_on_throw(opts.callback, [event_code, p1, p2])
       code,si = unpack_callback_result(packed)
@@ -4490,17 +4465,17 @@ class CoreFuncRef:
   callee: Callable[[list[CoreValType]], list[CoreValType]]
 
 def canon_thread_new_indirect(ft, ftbl: Table[CoreFuncRef], fi, c):
-  task = current_task()
-  trap_if(not task.inst.may_leave)
+  inst = current_instance()
+  trap_if(not inst.may_leave)
   f = ftbl.get(fi)
   assert(ft == CoreFuncType(['i32'], []) or ft == CoreFuncType(['i64'], []))
   trap_if(f.t != ft)
   def thread_func():
     [] = call_and_trap_on_throw(f.callee, [c])
-    task.unregister_thread(new_thread)
-  new_thread = Thread(task, thread_func)
+    inst.threads.remove(new_thread.index)
+  new_thread = Thread(current_task(), thread_func)
   assert(new_thread.suspended())
-  task.register_thread(new_thread)
+  new_thread.index = inst.threads.add(new_thread)
   return [new_thread.index]
 ```
 The newly-created thread starts out in a "suspended" state and so, to
@@ -4679,6 +4654,66 @@ def canon_thread_yield_then_promote(i):
 ```
 
 
+###### 🧵 `canon thread.get-task`
+
+For a canonical definition:
+```wat
+(canon thread.get-task (core func $thread.get-task))
+```
+validation specifies:
+* `$thread.get-task` is given type `(func (result i32))`
+
+Calling `$thread.get-task` invokes the following function which TODO
+```python
+def canon_thread_get_task():
+  thread = current_thread()
+  trap_if(not thread.task.inst.may_leave)
+  taski = thread.task.inst.handles.add(thread.task)
+  return [taski]
+```
+
+
+###### 🧵 `canon thread.set-task`
+
+For a canonical definition:
+```wat
+(canon thread.set-task (core func $thread.set-task))
+```
+validation specifies:
+* `$thread.set-task` is given type `(func (param i32))`
+
+Calling `$thread.set-task` invokes the following function which TODO
+```python
+def canon_thread_set_task(taski):
+  thread = current_thread()
+  trap_if(not thread.task.inst.may_leave)
+  new_task = thread.task.inst.handles.get(taski)
+  trap_if(not isinstance(new_task, Task))
+  thread.task = new_task
+  return []
+```
+
+
+###### 🧵 `canon task.drop`
+
+For a canonical definition:
+```wat
+(canon thread.drop (core func $task.drop))
+```
+validation specifies:
+* `$thread.drop` is given type `(func (param i32))`
+
+Calling `$task.drop` invokes the following function which TODO
+```python
+def canon_task_drop(taski):
+  inst = current_instance()
+  trap_if(not inst.may_leave)
+  task = inst.handles.remove(taski)
+  trap_if(not isinstance(task, Task))
+  return []
+```
+
+
 ### 📝 `canon error-context.new`
 
 For a canonical definition: