feat(lua): add vim.async

Problem: Nvim has many Lua APIs that start callback-driven work: timers,
jobs, libuv handles, and other event-loop tasks. Callers that need to
sequence or cancel that work have to build their own coroutine wrappers,
task bookkeeping, and cleanup rules. This makes async control flow hard
to share, test, and document.

Solution: Add `vim.async`, a structured-concurrency module vendored from
async.nvim. It provides task handles, await/pawait helpers,
sleep/timeout helpers, completion-order iteration, and semaphores on top
of Nvim's event loop.

The API follows the same broad model as Trio: async work has an owner,
tasks are awaited explicitly, and cancellation is cooperative. Include
generated vimdoc with an introductory overview and examples, a news
entry, and functional tests for the new module.

AI-assisted

10 files changed, 1957 insertions, 0 deletions

diff --git a/runtime/lua/vim/_core/editor.lua b/runtime/lua/vim/_core/editor.lua
index 66cc57f3d2..3fde7b4ef3 100644
--- a/runtime/lua/vim/_core/editor.lua
+++ b/runtime/lua/vim/_core/editor.lua
@@ -21,6 +21,7 @@ for k, v in pairs({
   secure = true,
   snippet = true,
   pack = true,
+  async = true,
   _watch = true,
   net = true,
   pos = true,
diff --git a/runtime/lua/vim/_core/util.lua b/runtime/lua/vim/_core/util.lua
index 02e5e6af04..08cbf17f04 100644
--- a/runtime/lua/vim/_core/util.lua
+++ b/runtime/lua/vim/_core/util.lua
@@ -2,6 +2,27 @@
 
 local M = {}
 
+-- Generated from async.nvim/lua/async/_errors.lua: start
+local nil_error = 'error(nil)'
+
+--- Normalize a failed Lua operation for error slots where `nil` means success.
+--- @param err any
+--- @return any
+--- @private
+function M._normalize_error(err)
+  return err == nil and nil_error or err
+end
+
+--- Convert an error to a string without letting its metamethod interrupt cleanup.
+--- @param err any
+--- @return string
+--- @private
+function M._stringify_error(err)
+  local ok, message = pcall(tostring, err)
+  return ok and message or '<unprintable error>'
+end
+-- Generated from async.nvim/lua/async/_errors.lua: end
+
 --- Adds one or more blank lines above or below the cursor.
 --- @param above? boolean Place blank line(s) above the cursor
 local function add_blank(above)
diff --git a/runtime/lua/vim/_meta.lua b/runtime/lua/vim/_meta.lua
index 18dbe49c85..e285cd5a12 100644
--- a/runtime/lua/vim/_meta.lua
+++ b/runtime/lua/vim/_meta.lua
@@ -66,6 +66,8 @@ vim.uv = ...
 
 vim.F = require('vim.F')
 vim._watch = require('vim._watch')
+---@diagnostic disable-next-line: no-unknown
+vim.async = require('vim.async')
 vim.diagnostic = require('vim.diagnostic')
 vim.filetype = require('vim.filetype')
 vim.fs = require('vim.fs')
diff --git a/runtime/lua/vim/async.lua b/runtime/lua/vim/async.lua
new file mode 100644
index 0000000000..8483b45add
--- /dev/null
+++ b/runtime/lua/vim/async.lua
@@ -0,0 +1,317 @@
+local core = require('vim.async._core')
+local validate = vim.validate
+local new_queue = require('vim.async._queue')
+local runtime = require('vim.async._runtime')
+local F = vim.F
+
+--- Structured async API for Lua code that waits on event-loop work.
+---
+--- `vim.async` lets Lua code wait for timers, callbacks, and other tasks
+--- without blocking Nvim's event loop. Async work runs inside tasks, which can
+--- pause at checkpoints and manage child tasks created while they are running.
+---
+--- Start async work with [vim.async.run()]. Inside a task, use
+--- [vim.async.await()] to wait for callback-style APIs or other tasks without
+--- blocking the event loop. Use [vim.async.pawait()] when an awaited operation
+--- can fail and the current task should continue.
+---
+--- Examples in this help use `local async = vim.async` for brevity.
+---
+--- Example: run async work without blocking Nvim:
+---
+--- ```lua
+--- local async = vim.async
+---
+--- async.run(function()
+---   vim.notify('waiting...')
+---   async.sleep(1000)
+---   vim.notify('done')
+--- end)
+--- ```
+---
+--- Example: await a callback-style API. Callback results are returned
+--- unchanged, so an error-first callback still returns `err, value`:
+---
+--- ```lua
+--- local async = vim.async
+---
+--- async.run(function()
+---   local err, stat = async.await(2, fs_stat, 'notes.txt')
+---   if err then
+---     error(err, 0)
+---   end
+---   print(('notes.txt is %d bytes'):format(stat.size))
+--- end)
+--- ```
+---
+--- A task has two roles:
+---
+--- - it is a handle that can be awaited, waited for, or closed
+--- - it is a scope for child tasks created while the task is running
+---
+--- [vim.async.run()] creates a task. A top-level task starts immediately. A
+--- task created while another task is running becomes a child of that task, and
+--- its function starts when the parent reaches its next checkpoint. A parent
+--- task finishes only after its attached children finish. If a child fails
+--- without being handled, the parent fails and closes the remaining children.
+---
+--- Use [Task:detach()] for background work that should keep running after the
+--- current task finishes. A detached task becomes top-level work; the original
+--- parent no longer waits for it or closes it.
+---
+--- Awaiting a task observes that task's result; it does not attach the task to
+--- the awaiter or change which task owns it. Ownership is decided when the task
+--- is created.
+---
+--- Scheduling is cooperative. When a task awaits a timer, I/O operation,
+--- callback, or another task, `vim.async` saves the Lua stack and returns
+--- control to the event loop. Other callbacks can run while the task is paused.
+--- Nothing interrupts synchronous Lua code in the middle of a stack frame.
+---
+--- Checkpoints are the places where a task can pause, start pending child
+--- tasks, observe cancellation, and receive unhandled child failures. Inside a
+--- task, these operations are checkpoints:
+---
+--- - `vim.async.await(...)`
+--- - `vim.async.pawait(...)`
+--- - `vim.async.checkpoint()`
+--- - successful return from the task function, which is the final checkpoint
+---   for child management
+---
+--- Convenience APIs such as [vim.async.sleep()] and [vim.async.timeout()] can
+--- also checkpoint because they call checkpointing APIs internally.
+---
+--- Closing a task closes its attached children. Cancellation is cooperative:
+--- [Task:close()] marks a task as closing, and the task observes that state at
+--- a checkpoint. If a task is suspended on a closable operation such as a timer
+--- or child task, `vim.async` closes that operation before reporting the
+--- cancellation.
+---
+--- Use [vim.async.await()] inside a task to suspend until work completes. It
+--- accepts a task, a callback-taking function, or an argument position plus a
+--- callback-taking function. `await(task)` returns the task result or raises
+--- the task failure. [vim.async.pawait()] is the async counterpart to `pcall()`
+--- for recoverable awaited-operation failures; it returns `ok, ...` instead of
+--- failing the current task for that awaited operation. It does not suppress
+--- cancellation or a failure already pending on the current task.
+---
+--- From synchronous code, use [Task:wait()] or [Task:pwait()] to pump the event
+--- loop until the task completes. Use [Task:on_complete()] to observe
+--- completion without blocking.
+---
+--- Coordination helpers work with task handles. [vim.async.iter()] yields
+--- completed task handles in completion order, [vim.async.timeout()] awaits a
+--- task with a deadline, and `vim.async.semaphore(permits)` creates a
+--- [vim.async.Semaphore] that limits how many tasks can hold a permit for a
+--- section at once.
+--- @class vim.async: vim.async._core
+local M = setmetatable({}, { __index = core })
+
+M.semaphore = require('vim.async._semaphore')
+
+--- @param unsubscribe fun()[]
+local function unsubscribe_all(unsubscribe)
+  for _, unsub in ipairs(unsubscribe) do
+    unsub()
+  end
+end
+
+--- Create an async function from a callback-style function.
+---
+--- The callback is inserted at argument position `argc`. If `func` returns a
+--- closable handle, it is closed when the awaiting task is cancelled.
+---
+--- This is a reusable wrapper around [vim.async.await()]. Use it when the same
+--- callback API is awaited from more than one place.
+---
+--- ```lua
+--- local async = vim.async
+--- local fs_stat = async.wrap(2, vim.uv.fs_stat)
+---
+--- async.run(function()
+---   local err, stat = fs_stat(vim.api.nvim_buf_get_name(0))
+---   if not err and stat then
+---     print(stat.size)
+---   end
+--- end)
+--- ```
+---
+--- @generic T, R
+--- @param argc integer
+--- @param func fun(...: T..., callback: fun(...: R...)): vim.async.Closable?
+--- @return async fun(...: T...): R...
+function M.wrap(argc, func)
+  validate('argc', argc, 'number')
+  validate('func', func, 'callable')
+  --- @async
+  return function(...)
+    return M.await(argc, func, ...)
+  end
+end
+
+--- Iterate completed tasks in completion order.
+---
+--- The iterator yields task handles, not task results. Use
+--- [vim.async.await()] or [vim.async.pawait()] to retrieve each result. The
+--- tasks are observed in the order they complete, regardless of the order in
+--- the input list.
+---
+--- ```lua
+--- local async = vim.async
+---
+--- async.run(function()
+---   local tasks = {
+---     async.run(function() return 'cache', read_cache() end):detach(),
+---     async.run(function() return 'disk', read_file() end):detach(),
+---   }
+---
+---   for task in async.iter(tasks) do
+---     local ok, source, text = async.pawait(task)
+---     if ok then
+---       for _, other in ipairs(tasks) do
+---         if other ~= task then
+---           other:close()
+---         end
+---       end
+---       print(('loaded from %s'):format(source))
+---       return text
+---     end
+---   end
+--- end)
+--- ```
+---
+--- If code must support PUC Lua 5.1, use the direct-call form instead of a
+--- generic `for` loop. The iterator may need to suspend while waiting for the
+--- next completed task, and PUC Lua 5.1 cannot yield from a generic-for
+--- iterator call.
+---
+--- ```lua
+--- local next_task = async.iter(tasks)
+--- while true do
+---   local task = next_task()
+---   if task == nil then
+---     break
+---   end
+---   async.await(task)
+--- end
+--- ```
+--- @async
+--- @generic R
+--- @param tasks vim.async.Task<R>[] A list of tasks to wait for and iterate over.
+--- @return async fun(): vim.async.Task<R>? iterator that yields each completed task.
+function M.iter(tasks)
+  validate('tasks', tasks, 'table')
+
+  local remaining = #tasks
+  local queue = new_queue()
+  local unsubscribe = {} --- @type fun()[]
+
+  if remaining == 0 then
+    queue:put_nowait()
+  else
+    for _, task in ipairs(tasks) do
+      unsubscribe[#unsubscribe + 1] = task:on_complete(function()
+        remaining = remaining - 1
+        queue:put_nowait(task)
+        if remaining == 0 then
+          queue:put_nowait()
+        end
+      end)
+    end
+  end
+
+  local proxy = newproxy(true)
+  getmetatable(proxy).__gc = function()
+    unsubscribe_all(unsubscribe)
+  end
+
+  --- @async
+  return function()
+    local _ = proxy -- Keep the GC proxy alive with the iterator.
+    return queue:get()
+  end
+end
+
+--- Asynchronously sleep for a given duration.
+---
+--- Suspends the current task for the given duration without blocking the event
+--- loop. After the delay and timer cleanup complete, `sleep()` returns to its
+--- caller through the runtime's `schedule` hook.
+---
+--- ```lua
+--- vim.async.run(function()
+---   vim.async.sleep(100)
+---   vim.notify('resumed later')
+--- end)
+--- ```
+--- @async
+--- @param duration integer ms
+function M.sleep(duration)
+  validate('duration', duration, 'number')
+  M.await(function(callback)
+    local timer = runtime.new_timer()
+    timer:start(duration, 0, callback)
+    return timer
+  end)
+  -- Timer cleanup resumes this function directly from its close callback.
+  -- Yield once more so M.sleep() returns through the runtime scheduler.
+  M.await(runtime.schedule)
+end
+
+--- Await a task with a timeout.
+---
+--- If the task completes first, returns or raises the task result as
+--- [vim.async.await()] would. If the deadline wins, closes the task and raises
+--- `"timeout"` after the target task finishes cancellation cleanup.
+---
+--- ```lua
+--- local async = vim.async
+---
+--- async.run(function()
+---   local task = async.run(read_file, 'notes.txt')
+---   local text = async.timeout(5000, task)
+---   show_buffer(text)
+--- end)
+--- ```
+--- @async
+--- @generic R
+--- @param duration integer Timeout duration in milliseconds
+--- @param task vim.async.Task<R>
+--- @return R...
+function M.timeout(duration, task)
+  validate('duration', duration, 'number')
+  validate('task', task, 'table')
+
+  local timed_out = false
+  local timer = M.run('__timeout', function()
+    M.sleep(duration)
+    timed_out = true
+    task:close()
+  end)
+  --- @diagnostic disable-next-line: invisible
+  timer._hidden = true
+
+  local result = F.pack_len(M.pawait(task))
+  timer:close()
+  M.pawait(timer)
+
+  if timed_out then
+    error('timeout', 0)
+  end
+
+  if not result[1] then
+    error(result[2], 0)
+  end
+
+  return unpack(result, 2, result.n)
+end
+
+if type(vim) == 'table' then
+  runtime.config({
+    wait = vim.wait,
+    schedule = vim.schedule,
+    new_timer = vim.uv.new_timer,
+  })
+end
+
+return M
diff --git a/runtime/lua/vim/async/_core.lua b/runtime/lua/vim/async/_core.lua
new file mode 100644
index 0000000000..25837598a4
--- /dev/null
+++ b/runtime/lua/vim/async/_core.lua
@@ -0,0 +1,1128 @@
+-- LuaLS cannot model the generic annotations used by this vendored implementation.
+---@diagnostic disable: no-unknown, undefined-doc-name, luadoc-miss-symbol, missing-return, missing-return-value, param-type-mismatch, return-type-mismatch, redundant-return-value, undefined-field, need-check-nil, await-in-sync
+
+local util = require('vim._core.util')
+local future = require('vim.async._future')
+local runtime = require('vim.async._runtime')
+
+local is_callable = vim.is_callable
+local validate = vim.validate
+local pcall = pcall
+local coroutine_running = coroutine.running
+do
+  local ok, coxpcall = pcall(require, 'coxpcall')
+  if ok and type(coxpcall) == 'table' then
+    if type(coxpcall.pcall) == 'function' then
+      pcall = coxpcall.pcall
+    end
+    if type(coxpcall.running) == 'function' then
+      coroutine_running = coxpcall.running
+    end
+  end
+end
+local maxint = 2 ^ 32 - 1
+local pack_len = vim.F.pack_len
+local unpack_len = vim.F.unpack_len
+
+--- @class vim.async._core
+--- @nodoc
+local M = {}
+
+--- Weak table to keep track of running tasks
+--- @type table<thread, vim.async.Task<any>?>
+local threads = setmetatable({}, { __mode = 'k' })
+
+--- Returns the currently running task.
+--- @return vim.async.Task<any>?
+local function running()
+  --- @diagnostic disable-next-line: invisible, undefined-field
+  local task = threads[coroutine_running()]
+  if task and not task:completed() then
+    return task
+  end
+end
+
+--- Internal marker used to identify that a yielded value is an asynchronous yielding.
+local yield_marker = {}
+local resume_marker = {}
+
+local resume_error = 'Unexpected coroutine.resume()'
+local yield_error = 'Unexpected coroutine.yield()'
+
+--- @return vim.async.Task<any>
+local function current_task()
+  return (assert(running(), 'Not in async context'))
+end
+
+--- Checks the arguments of a `coroutine.resume`.
+--- This is used to ensure that a resume is expected.
+--- @generic T
+--- @param marker any
+--- @param err? any
+--- @param ... T...
+--- @return T...
+local function check_yield(marker, err, ...)
+  if marker ~= resume_marker then
+    current_task():_raise(resume_error)
+    -- Return an error to the caller. This will also leave the task in a dead
+    -- and unfinished state.
+    error(resume_error, 0)
+  elseif err ~= nil then
+    error(err, 0)
+  end
+  return ...
+end
+
+--- @class vim.async.Closable
+--- @field close fun(self, callback?: fun())
+--- @field is_closing? fun(self): boolean
+
+--- A coroutine-backed async operation and concurrency scope.
+---
+--- Use [vim.async.run()] to create tasks. A task may be awaited by more than
+--- one waiter. When a task is created inside another task, it is attached to
+--- that parent and becomes part of the parent's concurrency scope.
+---
+--- @class vim.async.Task<R>: vim.async.Closable
+--- @field package _thread thread
+--- @field package _future vim.async.Future<R>
+--- @field package _closing boolean
+--- @field package _error? any
+--- @field package _finalizing_children boolean
+--- @field package _started boolean
+---
+--- Reference to parent to handle attaching/detaching.
+--- @field package _parent? vim.async.Task<any>
+--- @field package _parent_children_idx? integer
+---
+--- Name of the task
+--- @field name? string
+---
+--- Hide implementation tasks from user-facing inspection output.
+--- @field package _hidden? boolean
+---
+--- The source line that created this task, used for inspect().
+--- @field package _caller? string
+---
+--- Maintain children as an array to preserve closure order.
+--- @field package _children table<integer, vim.async.Task<any>?>
+---
+--- Pointer to last child in children
+--- @field package _children_idx integer
+---
+--- Tasks can await other async functions (task of callback functions)
+--- when we are waiting on a child, we store the handle to it here so we can
+--- close it.
+--- @field package _awaiting? vim.async.Task<any> | vim.async.Closable
+--- Removes the completion callback when a Task await is abandoned.
+--- @field package _awaiting_unsubscribe? fun()
+local Task = {}
+
+--- @return_cast x vim.async.Task<any>
+local function is_task(x)
+  return getmetatable(x) == Task
+end
+
+do --- Task
+  Task.__index = Task
+
+  --- @package
+  --- @param name? string
+  --- @param func async fun(...: any)
+  --- @return vim.async.Task<any>
+  function Task._new(name, func, ...)
+    local func_args = pack_len(...) --[[@as any[]? ]]
+    local thread = coroutine.create(function(marker, err)
+      -- Drop the packed vararg table before user code can suspend; otherwise
+      -- the coroutine closure retains it for the task lifetime.
+      local args = func_args
+      func_args = nil
+      check_yield(marker, err)
+      return func(unpack_len(args))
+    end)
+
+    local self = setmetatable({
+      name = name,
+      _closing = false,
+      _finalizing_children = false,
+      _started = false,
+      _thread = thread,
+      _future = future(),
+      _children = {},
+      _children_idx = 0,
+    }, Task)
+
+    threads[thread] = self
+
+    return self
+  end
+
+  --- Returns whether the Task has completed.
+  --- @return boolean
+  function Task:completed()
+    return self._future:completed()
+  end
+
+  --- Add a callback to be run when the Task has completed.
+  ---
+  --- If the Task is already done when this method is called, the callback is
+  --- called immediately with the results.
+  ---
+  --- This only observes completion. It does not start a pending task.
+  --- @param callback fun(err?: any, ...: R...)
+  --- @return fun() unsubscribe
+  function Task:on_complete(callback)
+    validate('callback', callback, 'callable')
+    return self._future:on_complete(callback)
+  end
+
+  --- Synchronously wait for the Task to complete.
+  ---
+  --- If a timeout is provided, waits for the given time in milliseconds before
+  --- failing with `"timeout"`. With no timeout, waits indefinitely.
+  ---
+  --- This is for synchronous code. Inside a task, prefer [vim.async.await()] so
+  --- the current task suspends instead of pumping the event loop itself.
+  ---
+  --- ```lua
+  --- local result = task:wait(10) -- wait for 10ms or raise "timeout"
+  ---
+  --- local result = task:wait() -- wait indefinitely
+  --- ```
+  --- @param timeout integer?
+  --- @return R...
+  function Task:wait(timeout)
+    validate('timeout', timeout, 'number', true)
+    self:_start()
+
+    if not runtime.wait(timeout or maxint, function()
+      return self:completed()
+    end) then
+      error('timeout', 2)
+    end
+    local res = pack_len(self._future:result())
+
+    assert(self:status() == 'completed' or res[2] == yield_error)
+
+    if not res[1] then
+      error(res[2], 2)
+    end
+
+    return unpack(res, 2, res.n)
+  end
+
+  --- Protected-call version of [Task:wait()].
+  ---
+  --- Equivalent to `pcall(task.wait, task, timeout)`.
+  ---
+  --- ```lua
+  --- local ok, result_or_err = task:pwait(1000)
+  --- if not ok then
+  ---   vim.notify(tostring(result_or_err), vim.log.levels.ERROR)
+  --- end
+  --- ```
+  --- @param timeout integer?
+  --- @return boolean, R...
+  function Task:pwait(timeout)
+    validate('timeout', timeout, 'number', true)
+    return pcall(self.wait, self, timeout)
+  end
+
+  --- @package
+  --- @param parent? vim.async.Task<any>
+  function Task:_attach(parent)
+    if parent then
+      -- Attach to parent
+      parent._children_idx = parent._children_idx + 1
+      parent._children[parent._children_idx] = self
+
+      -- Keep track of the parent and this tasks index so we can detach
+      self._parent = parent
+      self._parent_children_idx = parent._children_idx
+    end
+  end
+
+  --- Remove this task from its parent without changing execution state.
+  --- @private
+  --- @return boolean removed
+  function Task:_detach()
+    if not self._parent then
+      return false
+    end
+
+    self._parent._children[self._parent_children_idx] = nil
+    self._parent = nil
+    self._parent_children_idx = nil
+    return true
+  end
+
+  --- Detach a task from its parent.
+  ---
+  --- The task becomes a top-level task.
+  --- If it was waiting for a parent checkpoint, it is scheduled to start.
+  ---
+  --- Use this for background work that should not be cancelled when the
+  --- current task finishes. Detached task failures no longer fail the original
+  --- parent, so observe them explicitly with [Task:on_complete()],
+  --- [Task:wait()], or [vim.async.await()].
+  ---
+  --- ```lua
+  --- vim.async.run(function()
+  ---   while true do
+  ---     refresh_index()
+  ---     vim.async.sleep(1000)
+  ---   end
+  --- end):detach()
+  --- ```
+  --- @return vim.async.Task<R>
+  function Task:detach()
+    local should_start = self._parent and not self._started and not self:completed()
+    self:_detach()
+    if should_start then
+      runtime.schedule(function()
+        self:_start()
+      end)
+    end
+    return self
+  end
+
+  --- Get the traceback of a task when it is not active.
+  --- Will also get the traceback of nested tasks.
+  ---
+  --- @param msg? string
+  --- @param level? integer
+  --- @return string traceback
+  function Task:traceback(msg, level)
+    level = level or 0
+
+    local thread = '[' .. tostring(self._thread) .. '] '
+
+    local awaiting = self._awaiting
+    if is_task(awaiting) then
+      msg = awaiting:traceback(msg, level + 1)
+    end
+
+    local tblvl = is_task(awaiting) and 2 or nil
+    local tb = debug.traceback(self._thread, '', tblvl) or ''
+    msg = (msg == nil and '' or tostring(msg)) .. tb:gsub('\n\t', '\n\t' .. thread)
+
+    if level == 0 then
+      --- @type string
+      msg = msg
+        :gsub('\nstack traceback:\n', '\nSTACK TRACEBACK:\n', 1)
+        :gsub('\nstack traceback:\n', '\n')
+        :gsub('\nSTACK TRACEBACK:\n', '\nstack traceback:\n', 1)
+    end
+
+    return msg
+  end
+
+  --- Raise this task's error when it completes.
+  ---
+  --- Use this for detached or top-level fire-and-forget tasks whose completion
+  --- will not otherwise be observed. Attached task errors already propagate to
+  --- their parent.
+  ---
+  --- Detached tasks do not raise errors automatically. Detaching changes
+  --- ownership only; their completion can still be handled with
+  --- [vim.async.await()], [Task:wait()], [Task:pwait()], or [Task:on_complete()].
+  --- @return vim.async.Task<R> self
+  function Task:raise_on_error()
+    self:on_complete(function(err)
+      if err ~= nil then
+        error(self:traceback(err), 0)
+      end
+    end)
+    return self
+  end
+
+  --- @package
+  function Task:_start()
+    if self._started or self:completed() then
+      return
+    end
+
+    self._started = true
+    self:_resume()
+  end
+
+  --- Start children whose first resume was deferred by `run()`.
+  ---
+  --- Deferring child start lets `await()` or `pawait()` claim the task boundary
+  --- before child code runs. At await checkpoints and successful parent finish,
+  --- any remaining pending children must start so implicit waits, cancellation,
+  --- and inspection see the full task tree.
+  --- @package
+  function Task:_start_pending_children()
+    for i = 1, self._children_idx do
+      local child = self._children[i]
+      if child then
+        child:_start()
+      end
+    end
+  end
+
+  --- @private
+  function Task:_close_children()
+    for i = 1, self._children_idx do
+      local child = self._children[i]
+      if child then
+        child:close()
+      end
+    end
+  end
+
+  --- Keep the first task error. The error can be any non-nil Lua value.
+  --- @package
+  --- @param err any
+  --- @return any
+  function Task:_set_error(err)
+    if self._error == nil then
+      self._error = err
+    end
+    return self._error
+  end
+
+  --- @package
+  --- @param err any
+  function Task:_raise(err)
+    if self:status() == 'running' then
+      -- A running coroutine cannot be resumed recursively, so deliver the
+      -- error on a later event-loop turn after the current stack unwinds.
+      runtime.schedule(function()
+        if not self:completed() then
+          self:_resume(err)
+        end
+      end)
+    else
+      self:_resume(err)
+    end
+  end
+
+  --- Request cooperative close for the task and all of its children.
+  ---
+  --- The optional callback observes task completion and may run immediately if
+  --- the task has already completed.
+  ---
+  --- Closing is cooperative. The task observes the close request at a
+  --- checkpoint such as [vim.async.await()] or [vim.async.checkpoint()]. If the
+  --- task is suspended on an owned closable operation, that operation is closed
+  --- before the task reports `"closed"`.
+  ---
+  --- @param callback? fun()
+  function Task:close(callback)
+    if not self:completed() and not self._closing then
+      self._closing = true
+      self:_raise('closed')
+    end
+    if callback then
+      self:on_complete(function()
+        callback()
+      end)
+    end
+  end
+
+  --- Record a child failure on this task and either deliver it to a live
+  --- parent or close sibling children during finalization.
+  --- @package
+  --- @param child vim.async.Task<any>
+  --- @param err any
+  --- @param child_was_awaited boolean?
+  function Task:_child_failed(child, err, child_was_awaited)
+    -- A parent close turns child "closed" results into cleanup, not failure.
+    if child._closing or child_was_awaited or (self._closing and err == 'closed') then
+      return
+    end
+
+    local task_err = self:_set_error('child error: ' .. util._stringify_error(err))
+    if self._finalizing_children then
+      -- The parent coroutine is already dead, so sibling cleanup must wake the
+      -- finalizer instead of trying to resume the parent.
+      self:_close_children()
+    else
+      self:_raise(task_err)
+    end
+  end
+
+  --- Checks if an object is closable, i.e., has a `close` method.
+  --- @param obj any
+  --- @return boolean
+  --- @return_cast obj vim.async.Closable
+  local function is_closable(obj)
+    local ty = type(obj)
+    return (ty == 'table' or ty == 'userdata') and is_callable(obj.close)
+  end
+
+  do -- Task:_resume()
+    --- Complete this task with an error and propagate it to the parent if the
+    --- parent did not explicitly await this task.
+    --- @param parent? vim.async.Task<any>
+    --- @param err any
+    function Task:_finish_error(parent, err)
+      if err == nil then
+        err = self._error
+      end
+      err = util._normalize_error(err)
+      if parent then
+        parent:_child_failed(self, err, parent._awaiting == self)
+      end
+      self._future:complete(err)
+    end
+
+    --- @private
+    --- @param stat boolean
+    --- @param ... R... result
+    function Task:_finish(stat, ...)
+      if self:completed() then
+        return
+      end
+
+      local parent = self._parent
+      self:_detach()
+
+      threads[self._thread] = nil
+
+      if not stat then
+        self:_finish_error(parent, ...)
+      else
+        if self._error ~= nil then
+          self:_finish_error(parent, self._error)
+        else
+          self._future:complete(nil, ...)
+        end
+      end
+    end
+
+    --- @package
+    --- @param stat boolean
+    --- @param ... R... result
+    function Task:_finalize(stat, ...)
+      if next(self._children) == nil then
+        self:_finish(stat, ...)
+        return
+      end
+
+      local finish_args = pack_len(stat, ...)
+      self._finalizing_children = true
+      -- Only spawn the helper after the no-child path; an empty helper task
+      -- would otherwise finalize by recursively spawning another helper.
+      local await_children = Task._new('await_children', function()
+        -- TODO(lewis6991): should we collect all errors?
+        local close_remaining = not stat
+
+        if close_remaining then
+          self:_close_children()
+        else
+          self:_start_pending_children()
+        end
+
+        for i = 1, self._children_idx do
+          local child = self._children[i]
+          if child then
+            -- Child failures are recorded on `self`. Protect this helper so one
+            -- failed child cannot stop it awaiting the remaining cleanup.
+            local ok, err = M.pawait(child)
+            -- A close can arrive while normal finalization is awaiting
+            -- children; from that point child errors are cleanup results.
+            if not close_remaining and not self._closing and not ok and not child._closing then
+              self:_set_error('child error: ' .. util._stringify_error(err))
+              close_remaining = true
+              self:_close_children()
+            end
+          end
+        end
+
+        self._finalizing_children = false
+        if stat and self._closing and self._error == nil then
+          self:_finish(false, 'closed')
+        else
+          self:_finish(unpack_len(finish_args))
+        end
+      end)
+      await_children._hidden = true
+      await_children:_start()
+    end
+
+    --- Resume a task with the raw or protected result of an await.
+    --- @param task vim.async.Task<any>
+    --- @param yielded vim.async.Task<any>|fun(callback: fun(err?: any, ...: any)): vim.async.Closable?
+    --- @param protected boolean?
+    --- @param err? any
+    --- @param ... any
+    local function resume_from_await(task, yielded, protected, err, ...)
+      -- An error from `await(task)` also marks the waiting task as failed, even
+      -- if the error raised by `await()` is caught.
+      if not protected and is_task(yielded) and err ~= nil then
+        task:_set_error(err)
+      end
+
+      if protected then
+        if err ~= nil then
+          return task:_resume(nil, false, err)
+        end
+        return task:_resume(nil, true, ...)
+      end
+
+      return task:_resume(err, ...)
+    end
+
+    --- Begin waiting on a yielded awaitab