feat: lots of improvements to pipeline runner and converted all tasks to new system

README.md

@@ -20,6 +20,7 @@
 
 ## ✅ TODO
 
+- [ ] show time taken for op in view
 - [ ] package meta index (package.lua cache for all packages)
 - [ ] migrate from Packer
 - [ ] auto lazy-loading of lua modules

lua/lazy/core/plugin.lua

@@ -24,6 +24,7 @@ M.dirty = false
 ---@field updated? {from:string, to:string}
 ---@field is_local? boolean
 ---@field is_symlink? boolean
+---@field cloned? boolean
 
 ---@class LazyPluginRef
 ---@field branch? string
@@ -139,7 +140,9 @@ function M.update_state(check_clean)
   for _, plugin in pairs(Config.plugins) do
     plugin._ = plugin._ or {}
     plugin[1] = plugin["1"] or plugin[1]
-    plugin.opt = plugin.opt == nil and Config.options.opt or plugin.opt
+    if plugin.opt == nil then
+      plugin.opt = Config.options.opt
+    end
     local opt = plugin.opt and "opt" or "start"
     plugin.dir = Config.options.package_path .. "/" .. opt .. "/" .. plugin.name
     plugin._.is_local = plugin.uri:sub(1, 4) ~= "http" and plugin.uri:sub(1, 3) ~= "git"

lua/lazy/manage/git.lua

@@ -3,7 +3,10 @@ local Semver = require("lazy.manage.semver")
 
 local M = {}
 
+---@alias GitInfo {branch?:string, commit?:string, tag?:string, version?:Semver}
+
 ---@param details? boolean
+---@return GitInfo?
 function M.info(repo, details)
   local line = Util.head(repo .. "/.git/HEAD")
   if line then
@@ -23,7 +26,6 @@ function M.info(repo, details)
         end
       end)
     end
-
     return ret
   end
 end
@@ -48,6 +50,7 @@ function M.get_versions(repo, spec)
 end
 
 ---@param plugin LazyPlugin
+---@return {branch:string, commit?:string}?
 function M.get_branch(plugin)
   if plugin.branch then
     return {
@@ -69,22 +72,36 @@ function M.get_branch(plugin)
 end
 
 ---@param plugin LazyPlugin
+---@return GitInfo?
 function M.get_target(plugin)
   local branch = M.get_branch(plugin)
 
   if plugin.commit then
-    return { branch = branch, commit = plugin.commit }
+    return {
+      branch = branch and branch.branch,
+      commit = plugin.commit,
+    }
   end
   if plugin.tag then
-    return { branch = branch, tag = plugin.tag, commit = M.ref(plugin.dir, "tags/" .. plugin.tag) }
+    return {
+      branch = branch and branch.branch,
+      tag = plugin.tag,
+      commit = M.ref(plugin.dir, "tags/" .. plugin.tag),
+    }
   end
   if plugin.version then
     local last = Semver.last(M.get_versions(plugin.dir, plugin.version))
     if last then
-      return { branch = branch, version = last, tag = last.tag, commit = M.ref(plugin.dir, "tags/" .. last.tag) }
+      return {
+        branch = branch and branch.branch,
+        version = last,
+        tag = last.tag,
+        commit = M.ref(plugin.dir, "tags/" .. last.tag),
+      }
     end
   end
-  return { branch = branch, commit = branch.commit }
+  ---@diagnostic disable-next-line: return-type-mismatch
+  return branch
 end
 
 function M.ref(repo, ref)

lua/lazy/manage/init.lua

@@ -16,9 +16,6 @@ function M.run(ropts, opts)
   if opts.interactive == nil then
     opts.interactive = Config.options.interactive
   end
-  if ropts.interactive == nil then
-    ropts.interactive = opts.interactive
-  end
 
   if opts.clear then
     M.clear()
@@ -47,7 +44,14 @@ end
 ---@param opts? ManagerOpts
 function M.install(opts)
   M.run({
-    pipeline = { "git.install", "plugin.docs", "plugin.run" },
+    pipeline = {
+      "fs.symlink",
+      "git.clone",
+      "git.checkout",
+      "plugin.docs",
+      "wait",
+      "plugin.run",
+    },
     plugins = function(plugin)
       return plugin.uri and not plugin._.installed
     end,
@@ -57,7 +61,16 @@ end
 ---@param opts? ManagerOpts
 function M.update(opts)
   M.run({
-    pipeline = { "git.update", "plugin.docs", "plugin.run", "wait", "git.log" },
+    pipeline = {
+      "fs.symlink",
+      "git.branch",
+      "git.fetch",
+      "git.checkout",
+      "plugin.docs",
+      "plugin.run",
+      "wait",
+      { "git.log", updated = true },
+    },
     plugins = function(plugin)
       return plugin.uri and plugin._.installed
     end,
@@ -78,7 +91,7 @@ end
 function M.clean(opts)
   Plugin.update_state(true)
   M.run({
-    pipeline = { "plugin.clean" },
+    pipeline = { "fs.clean" },
     plugins = Config.to_clean,
   }, opts)
 end

lua/lazy/manage/runner.lua

@@ -1,19 +1,18 @@
 local Task = require("lazy.manage.task")
 local Config = require("lazy.core.config")
 
----@alias LazyPipeline TaskType[]
-
 ---@class RunnerOpts
----@field pipeline LazyPipeline
+---@field pipeline (string|{[1]:string, [string]:any})[]
----@field interactive? boolean
 ---@field plugins? LazyPlugin[]|fun(plugin:LazyPlugin):any?
 
+---@alias PipelineStep {task:string, opts?:TaskOptions}
+---@alias LazyRunnerTask {co:thread, status: {task?:LazyTask, waiting?:boolean}}
+
 ---@class Runner
----@field _tasks LazyTask[]
 ---@field _plugins LazyPlugin[]
----@field _running boolean
+---@field _running LazyRunnerTask[]
+---@field _pipeline PipelineStep[]
 ---@field _on_done fun()[]
----@field _waiting fun()[]
 ---@field _opts RunnerOpts
 local Runner = {}
 
@@ -21,7 +20,6 @@ local Runner = {}
 function Runner.new(opts)
   local self = setmetatable({}, { __index = Runner })
   self._opts = opts or {}
-  self._tasks = {}
 
   local plugins = self._opts.plugins
   if type(plugins) == "function" then
@@ -29,71 +27,58 @@ function Runner.new(opts)
   else
     self._plugins = plugins or Config.plugins
   end
-  self._running = false
+  self._running = {}
   self._on_done = {}
-  self._waiting = {}
+
+  ---@param step string|(TaskOptions|{[1]:string})
+  self._pipeline = vim.tbl_map(function(step)
+    return type(step) == "string" and { task = step } or { task = step[1], opts = step }
+  end, self._opts.pipeline)
+
   return self
 end
 
----@param plugin LazyPlugin
+---@param entry LazyRunnerTask
----@param pipeline LazyPipeline
+function Runner:_resume(entry)
-function Runner:_run(plugin, pipeline)
+  if entry.status.task and not entry.status.task:is_done() then
-  ---@type TaskType
+    return true
-  local op = table.remove(pipeline, 1)
-  if op == "wait" then
-    return table.insert(self._waiting, function()
-      self:_run(plugin, pipeline)
-    end)
   end
-  self:queue(plugin, op, function(task)
+  local ok, status = coroutine.resume(entry.co)
-    if not (task and task.error) and #pipeline > 0 then
+  entry.status = ok and status
-      self:_run(plugin, pipeline)
+  return entry.status ~= nil
-    end
-  end)
 end
 
----@param plugin LazyPlugin
+function Runner:resume(waiting)
----@param task_type TaskType
+  local running = false
----@param on_done fun(task?:LazyTask)
+  for _, entry in ipairs(self._running) do
----@return LazyTask?
+    if entry.status then
-function Runner:queue(plugin, task_type, on_done)
+      if waiting and entry.status.waiting then
-  local def = vim.split(task_type, ".", { plain = true })
+        entry.status.waiting = false
-  assert(#def == 2)
-  ---@type LazyTaskDef
-  local task_def = require("lazy.manage.task." .. def[1])[def[2]]
-  assert(task_def)
-  if not (task_def.skip and task_def.skip(plugin, self._opts)) then
-    local task = Task.new(plugin, def[2], task_def.run, { on_done = on_done })
-    table.insert(self._tasks, task)
-    task:start()
-  else
-    on_done()
       end
+      if not entry.status.waiting and self:_resume(entry) then
+        running = true
+      end
+    end
+  end
+  return running or (not waiting and self:resume(true))
 end
 
 function Runner:start()
   for _, plugin in pairs(self._plugins) do
-    self:_run(plugin, vim.deepcopy(self._opts.pipeline))
+    local co = coroutine.create(self.run_pipeline)
+    local ok, status = coroutine.resume(co, self, plugin)
+    if ok then
+      table.insert(self._running, { co = co, status = status })
+    end
   end
-  self._running = true
-  local check = vim.loop.new_check()
 
+  local check = vim.loop.new_check()
   check:start(function()
-    for _, task in ipairs(self._tasks) do
+    if self:resume() then
-      if task:is_running() then
-        return
-      end
-    end
-    if #self._waiting > 0 then
-      local waiting = self._waiting
-      self._waiting = {}
-      for _, cb in ipairs(waiting) do
-        cb()
-      end
       return
     end
     check:stop()
-    self._running = false
+    self._running = {}
     for _, cb in ipairs(self._on_done) do
       vim.schedule(cb)
     end
@@ -101,23 +86,47 @@ function Runner:start()
   end)
 end
 
----@return LazyPlugin[]
+---@async
-function Runner:plugins()
+---@param plugin LazyPlugin
-  ---@param task LazyTask
+function Runner:run_pipeline(plugin)
-  return vim.tbl_map(function(task)
+  for _, step in ipairs(self._pipeline) do
-    return task.plugin
+    if step.task == "wait" then
-  end, self._tasks)
+      coroutine.yield({ waiting = true })
+    else
+      local task = self:queue(plugin, step.task, step.opts)
+      if task then
+        coroutine.yield({ task = task })
+        assert(task:is_done())
+        if task.error then
+          return
+        end
+      end
+    end
+  end
 end
 
-function Runner:tasks()
+---@param plugin LazyPlugin
-  return self._tasks
+---@param task_type string
+---@param task_opts? TaskOptions
+---@return LazyTask?
+function Runner:queue(plugin, task_type, task_opts)
+  assert(self._running)
+  local def = vim.split(task_type, ".", { plain = true })
+  ---@type LazyTaskDef
+  local task_def = require("lazy.manage.task." .. def[1])[def[2]]
+  assert(task_def)
+  if not (task_def.skip and task_def.skip(plugin, task_opts)) then
+    local task = Task.new(plugin, def[2], task_def.run, task_opts)
+    task:start()
+    return task
+  end
 end
 
 -- Execute the callback async when done.
 -- When no callback is specified, this will wait sync
 ---@param cb? fun()
 function Runner:wait(cb)
-  if #self._tasks == 0 or not self._running then
+  if #self._running == 0 then
     return cb and cb()
   end
 
@@ -125,8 +134,8 @@ function Runner:wait(cb)
     table.insert(self._on_done, cb)
   else
     -- sync wait
-    while self._running do
+    while #self._running > 0 do
-      vim.wait(100)
+      vim.wait(10)
     end
   end
 end

lua/lazy/manage/task/fs.lua

@@ -0,0 +1,50 @@
+local Util = require("lazy.util")
+
+---@type table<string, LazyTaskDef>
+local M = {}
+
+M.clean = {
+  run = function(self)
+    local dir = self.plugin.dir:gsub("/+$", "")
+    local stat = vim.loop.fs_lstat(dir)
+
+    if stat.type == "directory" then
+      Util.walk(dir, function(path, _, type)
+        if type == "directory" then
+          vim.loop.fs_rmdir(path)
+        else
+          vim.loop.fs_unlink(path)
+        end
+      end)
+      vim.loop.fs_rmdir(dir)
+    else
+      vim.loop.fs_unlink(dir)
+    end
+
+    self.plugin._.installed = false
+  end,
+}
+
+M.symlink = {
+  skip = function(plugin)
+    if not plugin._.is_local then
+      return true
+    end
+    return not plugin._.is_symlink and plugin._.installed
+  end,
+  run = function(self)
+    local stat = vim.loop.fs_lstat(self.plugin.dir)
+    if stat then
+      assert(stat.type == "link")
+      if vim.loop.fs_realpath(self.plugin.uri) == vim.loop.fs_realpath(self.plugin.dir) then
+        return
+      else
+        vim.loop.fs_unlink(self.plugin.dir)
+      end
+    end
+    vim.loop.fs_symlink(self.plugin.uri, self.plugin.dir, { dir = true })
+    vim.opt.runtimepath:append(self.plugin.uri)
+  end,
+}
+
+return M

lua/lazy/manage/task/git.lua

@@ -5,13 +5,15 @@ local Git = require("lazy.manage.git")
 local M = {}
 
 M.log = {
+  ---@param opts {since?: string, updated?:boolean}
   skip = function(plugin, opts)
-    if not (opts.interactive and Util.file_exists(plugin.dir .. "/.git")) then
+    if opts.updated and not (plugin._.updated and plugin._.updated.from ~= plugin._.updated.to) then
-      return false
+      return true
     end
-    return plugin._.updated and plugin._.updated.from == plugin._.updated.to
+    return not Util.file_exists(plugin.dir .. "/.git")
   end,
-  run = function(self)
+  ---@param opts {since?: string, updated?:boolean}
+  run = function(self, opts)
     local args = {
       "log",
       "--pretty=format:%h %s (%cr)",
@@ -21,10 +23,10 @@ M.log = {
       "--color=never",
     }
 
-    if self.plugin._.updated then
+    if opts.updated then
       table.insert(args, self.plugin._.updated.from .. ".." .. (self.plugin._.updated.to or "HEAD"))
     else
-      table.insert(args, "--since=7 days ago")
+      table.insert(args, "--since=" .. (opts.since or "7 days ago"))
     end
 
     self:spawn("git", {
@@ -34,53 +36,11 @@ M.log = {
   end,
 }
 
-M.update = {
+M.clone = {
-  run = function(self)
+  skip = function(plugin)
-    if self.plugin._.is_local ~= self.plugin._.is_symlink then
+    return plugin._.installed or plugin._.is_local
-      -- FIXME: should change here and in install
-      error("incorrect local")
-    end
-    if self.plugin._.is_local then
-      if vim.loop.fs_realpath(self.plugin.uri) ~= vim.loop.fs_realpath(self.plugin.dir) then
-        vim.loop.fs_unlink(self.plugin.dir)
-        vim.loop.fs_symlink(self.plugin.uri, self.plugin.dir, {
-          dir = true,
-        })
-        vim.opt.runtimepath:append(self.plugin.uri)
-      end
-    else
-      local args = {
-        "pull",
-        "--recurse-submodules",
-        "--update-shallow",
-        "--progress",
-      }
-      local git = assert(Git.info(self.plugin.dir))
-
-      self:spawn("git", {
-        args = args,
-        cwd = self.plugin.dir,
-        on_exit = function(ok)
-          if ok then
-            local git_new = assert(Git.info(self.plugin.dir))
-            self.plugin._.updated = {
-              from = git.commit,
-              to = git_new.commit,
-            }
-            self.plugin._.dirty = not vim.deep_equal(git, git_new)
-          end
   end,
-      })
-    end
-  end,
-}
-
-M.install = {
   run = function(self)
-    if self.plugin._.is_local then
-      vim.loop.fs_symlink(self.plugin.uri, self.plugin.dir, { dir = true })
-      vim.opt.runtimepath:append(self.plugin.uri)
-    else
     local args = {
       "clone",
       self.plugin.uri,
@@ -101,12 +61,100 @@ M.install = {
       args = args,
       on_exit = function(ok)
         if ok then
+          self.plugin._.cloned = true
           self.plugin._.installed = true
           self.plugin._.dirty = true
         end
       end,
     })
+  end,
+}
+
+M.branch = {
+  skip = function(plugin)
+    if not plugin._.installed or plugin._.is_local then
+      return true
     end
+    local branch = assert(Git.get_branch(plugin))
+    return branch and branch.commit
+  end,
+  run = function(self)
+    local branch = assert(Git.get_branch(self.plugin))
+    local args = {
+      "remote",
+      "set-branches",
+      "--add",
+      "origin",
+      branch.branch,
+    }
+
+    self:spawn("git", {
+      args = args,
+      cwd = self.plugin.dir,
+    })
+  end,
+}
+
+M.fetch = {
+  skip = function(plugin)
+    return not plugin._.installed or plugin._.is_local
+  end,
+  run = function(self)
+    local args = {
+      "fetch",
+      "--recurse-submodules",
+      "--update-shallow",
+      "--progress",
+    }
+
+    self:spawn("git", {
+      args = args,
+      cwd = self.plugin.dir,
+    })
+  end,
+}
+
+M.checkout = {
+  skip = function(plugin)
+    return not plugin._.installed or plugin._.is_local
+  end,
+  run = function(self)
+    local info = assert(Git.info(self.plugin.dir))
+    local target = assert(Git.get_target(self.plugin))
+
+    if not self.plugin._.cloned and info.commit == target.commit then
+      return
+    end
+
+    local args = {
+      "checkout",
+      "--progress",
+    }
+
+    if target.tag then
+      table.insert(args, "tags/" .. target.tag)
+    elseif self.plugin.commit then
+      table.insert(args, self.plugin.commit)
+    elseif target.branch then
+      table.insert(args, target.branch)
+    end
+
+    self:spawn("git", {
+      args = args,
+      cwd = self.plugin.dir,
+      on_exit = function(ok)
+        if ok then
+          local new_info = assert(Git.info(self.plugin.dir))
+          if not self.plugin._.cloned then
+            self.plugin._.updated = {
+              from = info.commit,
+              to = new_info.commit,
+            }
+          end
+          self.plugin._.dirty = true
+        end
+      end,
+    })
   end,
 }
 return M

lua/lazy/manage/task/init.lua

@@ -1,42 +1,42 @@
 local Process = require("lazy.manage.process")
 
 ---@class LazyTaskDef
----@field skip? fun(plugin:LazyPlugin, opts:RunnerOpts):any?
+---@field skip? fun(plugin:LazyPlugin, opts?:TaskOptions):any?
----@field run fun(task:LazyTask)
+---@field run fun(task:LazyTask, opts:TaskOptions)
 
 ---@alias LazyTaskState fun():boolean?
 
 ---@class LazyTask
 ---@field plugin LazyPlugin
----@field type TaskType
+---@field name string
+---@field type string
 ---@field output string
 ---@field status string
 ---@field error? string
 ---@field private _task fun(task:LazyTask)
 ---@field private _running LazyPluginState[]
----@field private _started boolean
+---@field private _started? number
+---@field private _ended? number
 ---@field private _opts TaskOptions
 local Task = {}
 
----@alias TaskType "update"|"install"|"run"|"clean"|"log"|"docs"
+---@class TaskOptions: {[string]:any}
-
----@class TaskOptions
 ---@field on_done? fun(task:LazyTask)
 
 ---@param plugin LazyPlugin
----@param type TaskType
+---@param name string
 ---@param opts? TaskOptions
 ---@param task fun(task:LazyTask)
-function Task.new(plugin, type, task, opts)
+function Task.new(plugin, name, task, opts)
   local self = setmetatable({}, {
     __index = Task,
   })
   self._opts = opts or {}
   self._running = {}
   self._task = task
-  self._started = false
+  self._started = nil
   self.plugin = plugin
-  self.type = type
+  self.name = name
   self.output = ""
   self.status = ""
   plugin._.tasks = plugin._.tasks or {}
@@ -45,7 +45,7 @@ function Task.new(plugin, type, task, opts)
 end
 
 function Task:has_started()
-  return self._started
+  return self._started ~= nil
 end
 
 function Task:is_done()
@@ -62,9 +62,14 @@ function Task:is_running()
 end
 
 function Task:start()
-  self._started = true
+  if vim.in_fast_event() then
+    return vim.schedule(function()
+      self:start()
+    end)
+  end
+  self._started = vim.loop.hrtime()
   ---@type boolean, string|any
-  local ok, err = pcall(self._task, self)
+  local ok, err = pcall(self._task, self, self._opts)
   if not ok then
     self.error = err or "failed"
   end
@@ -76,16 +81,27 @@ function Task:_check()
   if self:is_running() then
     return
   end
+  self._ended = vim.loop.hrtime()
   if self._opts.on_done then
     self._opts.on_done(self)
   end
   vim.cmd("do User LazyRender")
   vim.api.nvim_exec_autocmds("User", {
-    pattern = "LazyPlugin" .. self.type:sub(1, 1):upper() .. self.type:sub(2),
+    pattern = "LazyPlugin" .. self.name:sub(1, 1):upper() .. self.name:sub(2),
     data = { plugin = self.plugin.name },
   })
 end
 
+function Task:time()
+  if not self:has_started() then
+    return 0
+  end
+  if not self:is_done() then
+    return (vim.loop.hrtime() - self._started) / 1e6
+  end
+  return (self._ended - self._started) / 1e6
+end
+
 ---@param fn fun()
 function Task:schedule(fn)
   local done = false

lua/lazy/manage/task/plugin.lua

@@ -29,28 +29,6 @@ M.run = {
   end,
 }
 
-M.clean = {
-  run = function(self)
-    local dir = self.plugin.dir:gsub("/+$", "")
-    local stat = vim.loop.fs_lstat(dir)
-
-    if stat.type == "directory" then
-      Util.walk(dir, function(path, _, type)
-        if type == "directory" then
-          vim.loop.fs_rmdir(path)
-        else
-          vim.loop.fs_unlink(path)
-        end
-      end)
-      vim.loop.fs_rmdir(dir)
-    else
-      vim.loop.fs_unlink(dir)
-    end
-
-    self.plugin._.installed = false
-  end,
-}
-
 M.docs = {
   skip = function(plugin)
     return not plugin._.dirty

lua/lazy/view/init.lua

@@ -69,10 +69,12 @@ function M.show()
 
   local render = Render.new(buf, win, 2)
   local update = Util.throttle(30, function()
+    if buf and vim.api.nvim_buf_is_valid(buf) then
       vim.bo[buf].modifiable = true
       render:update()
       vim.bo[buf].modifiable = false
       vim.cmd.redraw()
+    end
   end)
 
   local function get_plugin()

lua/lazy/view/render.lua

@@ -219,11 +219,11 @@ function M:diagnostics(plugin)
     if task:is_running() then
       self:diagnostic({
         severity = vim.diagnostic.severity.WARN,
-        message = task.type .. (task.status == "" and "" or (": " .. task.status)),
+        message = task.name .. (task.status == "" and "" or (": " .. task.status)),
       })
     elseif task.error then
       self:diagnostic({
-        message = task.type .. " failed",
+        message = task.name .. " failed",
         severity = vim.diagnostic.severity.ERROR,
       })
     end
@@ -250,7 +250,12 @@ end
 ---@param plugin LazyPlugin
 function M:tasks(plugin)
   for _, task in ipairs(plugin._.tasks or {}) do
-    if task.type == "log" and not task.error then
+    if self._details == plugin.name then
+      self:append("✔ [task] ", "Title", { indent = 4 }):append(task.name)
+      self:append(" " .. math.floor((task:time()) * 100) / 100 .. "ms", "Bold")
+      self:nl()
+    end
+    if task.name == "log" and not task.error then
       self:log(task)
     elseif task.error or self._details == plugin.name then
       if task.error then

lua/lazy/view/sections.lua

@@ -33,7 +33,7 @@ return {
   {
     filter = function(plugin)
       return has_task(plugin, function(task)
-        if task.type ~= "log" then
+        if task.name ~= "log" then
           return
         end
         local lines = vim.split(task.output, "\n")
@@ -53,10 +53,17 @@ return {
     end,
     title = "Updated",
   },
+  {
+    ---@param plugin LazyPlugin
+    filter = function(plugin)
+      return plugin._.cloned
+    end,
+    title = "Installed",
+  },
   {
     filter = function(plugin)
       return has_task(plugin, function(task)
-        return task.type == "log" and vim.trim(task.output) ~= ""
+        return task.name == "log" and vim.trim(task.output) ~= ""
       end)
     end,
     title = "Log",

tests/manage/runner_spec.lua

@@ -19,33 +19,65 @@ describe("runner", function()
     package.loaded["lazy.manage.task.test"]["test" .. i] = {
       ---@param task LazyTask
       run = function(task)
-        table.insert(runs, { plugin = task.plugin.name, task = task.type })
+        table.insert(runs, { plugin = task.plugin.name, task = task.name })
       end,
     }
     package.loaded["lazy.manage.task.test"]["error" .. i] = {
       ---@param task LazyTask
       run = function(task)
-        table.insert(runs, { plugin = task.plugin.name, task = task.type })
+        table.insert(runs, { plugin = task.plugin.name, task = task.name })
         error("error" .. i)
       end,
     }
+    package.loaded["lazy.manage.task.test"]["async" .. i] = {
+      ---@param task LazyTask
+      run = function(task)
+        task:schedule(function()
+          table.insert(runs, { plugin = task.plugin.name, task = task.name })
+        end)
+      end,
+    }
   end
 
   it("runs the pipeline", function()
     local runner = Runner.new({ plugins = plugins, pipeline = { "test.test1", "test.test2" } })
     runner:start()
+    runner:wait()
+    assert.equal(4, #runs)
+  end)
+
+  it("waits", function()
+    local runner = Runner.new({ plugins = plugins, pipeline = { "test.test1", "wait", "test.test2" } })
+    runner:start()
+    runner:wait()
+    assert.equal(4, #runs)
+  end)
+
+  it("handles async", function()
+    local runner = Runner.new({ plugins = plugins, pipeline = { "test.async1", "wait", "test.async2" } })
+    runner:start()
+    runner:wait()
     assert.equal(4, #runs)
   end)
 
   it("handles skips", function()
     local runner = Runner.new({ plugins = plugins, pipeline = { "test.test1", "test.skip", "test.test2" } })
     runner:start()
+    runner:wait()
+    assert.equal(4, #runs)
+  end)
+
+  it("handles opts", function()
+    local runner = Runner.new({ plugins = plugins, pipeline = { "test.test1", { "test.test2", foo = "bar" } } })
+    runner:start()
+    runner:wait()
     assert.equal(4, #runs)
   end)
 
   it("aborts on error", function()
     local runner = Runner.new({ plugins = plugins, pipeline = { "test.test1", "test.error1", "test.test2" } })
     runner:start()
+    runner:wait()
     assert.equal(4, #runs)
   end)
 end)