Add /-/tasks cross-references and the 1.0a39 changelog

Rolled down from the stack's docs-only tip PR: introspection docs now
link to the background-task API sections, the internals docs mention
/-/tasks where relevant, and the full 1.0a39 changelog (including the
asgi_wrapper migration guide) lands here at the top of the stack.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_012U7coQfVu8nK2R4q2mCULA

datasette/__init__.py

@@ -1,6 +1,7 @@
 from datasette.permissions import Permission  # noqa
 from datasette.version import __version_info__, __version__  # noqa
 from datasette.events import Event  # noqa
+from datasette.background_tasks import BackgroundTask, BackgroundTaskSupervisor  # noqa
 from datasette.tokens import TokenHandler, TokenInvalid, TokenRestrictions  # noqa
 from datasette.utils.asgi import (  # noqa
     Forbidden,

datasette/app.py

@@ -42,6 +42,7 @@ from jinja2.exceptions import TemplateNotFound
 from markupsafe import Markup, escape
 
 from . import stored_queries, write_sql
+from .background_tasks import BackgroundTask, BackgroundTaskSupervisor
 from .column_types import SQLiteType
 from .csrf import CrossOriginProtectionMiddleware
 from .database import Database, QueryInterrupted
@@ -422,6 +423,7 @@ class Datasette:
         default_deny=False,
     ):
         self._startup_invoked = False
+        self._shutdown_invoked = False
         self._closed = False
         assert config_dir is None or isinstance(
             config_dir, Path
@@ -454,6 +456,7 @@ class Datasette:
         self.actions = {}  # .invoke_startup() will populate this
         self._column_types = {}  # .invoke_startup() will populate this
         self._setup_db_done = False
+        self._suppress_background_tasks = False
         try:
             self._refresh_schemas_lock = asyncio.Lock()
             self._startup_lock = asyncio.Lock()
@@ -467,6 +470,7 @@ class Datasette:
                 self._startup_lock = asyncio.Lock()
             else:
                 raise
+        self._background_tasks = BackgroundTaskSupervisor(self)
         self.crossdb = crossdb
         self.nolock = nolock
         if memory or crossdb or not self.files:
@@ -2281,6 +2285,21 @@ class Datasette:
         )
         return d
 
+    def _tasks(self):
+        return {
+            "tasks": [
+                {
+                    "name": t.name,
+                    "state": t.state,
+                    "plugin": t.plugin,
+                    "started_at": t.started_at,
+                    "exception": repr(t.exception) if t.exception else None,
+                }
+                for t in self._background_tasks.tasks()
+            ],
+            "launched": self._background_tasks.launched,
+        }
+
     def _actor(self, request):
         return {"actor": request.actor}
 
@@ -2569,6 +2588,12 @@ class Datasette:
             ),
             r"/-/threads(\.(?P<format>json))?$",
         )
+        add_route(
+            JsonDataView.as_view(
+                self, "tasks.json", self._tasks, permission="permissions-debug"
+            ),
+            r"/-/tasks(\.(?P<format>json))?$",
+        )
         add_route(
             JsonDataView.as_view(
                 self,
@@ -2836,24 +2861,123 @@ class Datasette:
                 self._setup_db_done = True
             await self.invoke_startup()
 
+    def add_background_task(self, func, name=None) -> BackgroundTask:
+        """Register a piece of supervised background work, typically from
+        a plugin's ``startup`` hook.
+
+        ``func`` must be a coroutine function taking one positional
+        argument, the ``Datasette`` instance - core calls ``func(self)``.
+        Callable any time after ``__init__``: if background tasks haven't
+        launched yet (the common case - most callers are ``startup`` hooks,
+        which run before launch), this buffers the registration until they
+        do; if they've already launched (e.g. called from a request
+        handler after the server is up), the task starts immediately.
+
+        Returns a :class:`~datasette.background_tasks.BackgroundTask`
+        handle (``.name``, ``.state``, ``.task``, ``.exception``,
+        ``.started_at``, ``.plugin``, ``.cancel()``).
+
+        ``name`` defaults to ``func.__qualname__``; on a name collision a
+        ``-2``, ``-3``, ... suffix is appended, since names are how
+        ``/-/tasks`` and log messages identify work.
+        """
+        return self._background_tasks.add(func, name=name)
+
+    async def start_background_tasks(self):
+        """Run startup (if it hasn't run yet) and launch every registered
+        background task.
+
+        Public entry point for tests, embedders, and headless CLIs (the
+        ``datasette-rss``-style ``fetch --due`` shape) that want supervised
+        background tasks without running a server - equivalent to what
+        happens automatically via ASGI lifespan / the first-request
+        fallback in a served deployment.
+        """
+        await self.invoke_startup()
+        await self._background_tasks.launch_all()
+
+    async def _launch_background_tasks(self):
+        """Idempotently launch every registered background task. Private:
+        this is the entry point wired into the lifecycle trigger lists
+        (the second entry in both ``AsgiLifespan`` and
+        ``AsgiRunOnFirstRequest``'s ``on_startup``, after
+        ``_startup_sequence``) - not something plugins or embedders should
+        call directly; use ``add_background_task`` /
+        ``start_background_tasks`` instead.
+
+        Positioned after ``_startup_sequence`` in both trigger lists so
+        launch always happens once every plugin's ``startup`` hook has had
+        a chance to register work - the ordering guarantee that makes
+        ``add_background_task`` useful. No-ops when
+        ``_suppress_background_tasks`` is set (the ``--get`` CLI path: its
+        one-shot TestClient request flows through the full ASGI stack,
+        including the first-request fallback, but must never launch
+        long-lived background work).
+        """
+        if self._suppress_background_tasks:
+            return
+        await self._background_tasks.launch_all()
+
+    async def invoke_shutdown(self):
+        """Run the graceful teardown sequence: plugin ``shutdown`` hooks,
+        then cancel and drain supervised background tasks, then close
+        every database.
+
+        Idempotent (guarded by ``_shutdown_invoked``) and safe to call more
+        than once - a second ``lifespan.shutdown`` message from a
+        misbehaving ASGI host, or any future caller, must not re-run
+        teardown. A ``shutdown`` hook that raises is logged and swallowed
+        rather than propagated, so one broken plugin can't skip another
+        plugin's cleanup, or skip task cancellation / ``close()``
+        altogether.
+
+        Order matters: hooks run first, while background tasks are still
+        alive, so a plugin can coordinate with its own task (e.g. tell a
+        queue consumer to stop pulling new work) before that task gets
+        cancelled; ``close()`` runs last so both the hooks and the
+        cancelled tasks still have working database connections to write
+        any final state.
+        """
+        if self._shutdown_invoked:
+            return
+        self._shutdown_invoked = True
+        for hook in pm.hook.shutdown(datasette=self):
+            try:
+                await await_me_maybe(hook)
+            except Exception:
+                logging.getLogger("datasette").exception("shutdown hook failed")
+        await self._background_tasks.cancel_all(grace=5.0)
+        self.close()
+
     def app(self):
         """Returns an ASGI app function that serves the whole of Datasette"""
         routes = self._routes()
 
-        async def _close_on_shutdown():
-            self.close()
-
         asgi = CrossOriginProtectionMiddleware(DatasetteRouter(self, routes), self)
         if self.setting("trace_debug"):
             asgi = AsgiTracer(asgi)
         asgi = AsgiLifespan(
             asgi,
-            on_startup=[self._startup_sequence],
-            on_shutdown=[_close_on_shutdown],
+            on_startup=[self._startup_sequence, self._launch_background_tasks],
+            on_shutdown=[self.invoke_shutdown],
         )
-        asgi = AsgiRunOnFirstRequest(asgi, on_startup=[self._startup_sequence])
+        # Plugin asgi_wrapper middleware sits INSIDE AsgiRunOnFirstRequest
+        # (below it, i.e. closer to the app) but OUTSIDE AsgiLifespan (above
+        # it). That gives wrappers a single, simple contract: every http/
+        # websocket scope they see has already been through
+        # AsgiRunOnFirstRequest, so startup (including plugin migrations
+        # against the internal database) is guaranteed to have completed -
+        # even for a wrapper that short-circuits and never calls the inner
+        # app, and even on hosts that never send ASGI lifespan events.
+        # "lifespan" scopes are untouched by this reorder: AsgiRunOnFirstRequest
+        # ignores them and passes them straight through to the wrappers (and
+        # from there down to AsgiLifespan), exactly as before.
         for wrapper in pm.hook.asgi_wrapper(datasette=self):
             asgi = wrapper(asgi)
+        asgi = AsgiRunOnFirstRequest(
+            asgi,
+            on_startup=[self._startup_sequence, self._launch_background_tasks],
+        )
         return asgi
 
 

datasette/background_tasks.py

@@ -0,0 +1,263 @@
+"""
+Supervised background-task registration for Datasette core.
+
+Plugins that need long-lived background work (a polling loop, a queue
+consumer, a scheduled job runner) register it with
+``datasette.add_background_task(func, name=None)`` - typically from a
+``startup`` plugin hook - instead of fire-and-forgetting their own
+``asyncio.create_task()``. Core owns:
+
+- **references**: every launched ``asyncio.Task`` is kept alive on a
+  :class:`BackgroundTaskSupervisor`, so it can never be silently garbage
+  collected the way an unreferenced ``create_task()`` call can be;
+- **launch timing**: registered work is buffered until
+  :meth:`BackgroundTaskSupervisor.launch_all` runs, which core arranges to
+  happen only after *every* plugin's ``startup`` hook has finished - so
+  a task that depends on another plugin having registered something first
+  doesn't need ``tryfirst=True`` ordering tricks;
+- **crash surfacing**: an unhandled exception in a background task is
+  logged with its full traceback to the ``datasette.background_tasks``
+  logger and recorded on the handle, instead of becoming an "Task
+  exception was never retrieved" warning nobody sees;
+- **cancellation**: :meth:`BackgroundTaskSupervisor.cancel_all` cancels
+  every task still running and waits (with a grace period) for them to
+  actually stop.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import datetime
+import functools
+import inspect
+import logging
+from collections.abc import Awaitable, Callable
+
+logger = logging.getLogger("datasette.background_tasks")
+
+
+def _utcnow_iso() -> str:
+    return datetime.datetime.now(datetime.timezone.utc).isoformat()
+
+
+def _resolve_plugin_name(func: Callable) -> str | None:
+    """Best-effort, cheap attempt to work out which registered plugin a
+    background-task function belongs to, for the ``.plugin`` field on
+    :class:`BackgroundTask` (used by ``/-/tasks`` and logs).
+
+    This matches ``func``'s module against every currently-registered
+    pluggy plugin's module - the same module a plugin's ``startup`` hook
+    implementation lives in, in the overwhelmingly common case where
+    ``add_background_task`` is called directly from (or a couple of
+    frames below) that hook. It deliberately does *not* walk the call
+    stack or otherwise try harder: this is a nice-to-have for
+    introspection, not something worth building heavy machinery for, and
+    returning ``None`` when it can't tell is a fine fallback.
+    """
+    try:
+        from .plugins import pm
+
+        module = inspect.getmodule(func)
+        if module is None:
+            return None
+        module_name = getattr(module, "__name__", None)
+        if not module_name:
+            return None
+        for plugin in pm.get_plugins():
+            plugin_module = (
+                plugin if inspect.ismodule(plugin) else inspect.getmodule(plugin)
+            )
+            if plugin_module is None:
+                continue
+            plugin_module_name = getattr(plugin_module, "__name__", None)
+            if not plugin_module_name:
+                continue
+            if module_name == plugin_module_name or module_name.startswith(
+                plugin_module_name + "."
+            ):
+                return pm.get_name(plugin)
+    except Exception:  # noqa: BLE001
+        # Never let plugin-name resolution break task registration.
+        return None
+    return None
+
+
+class BackgroundTask:
+    """A handle to a single piece of supervised background work.
+
+    States: ``registered`` (added but not yet launched) -> ``running`` ->
+    one of ``completed`` (returned cleanly), ``crashed`` (raised an
+    exception other than ``CancelledError`` - see ``.exception``), or
+    ``cancelled`` (``.cancel()`` was called, or it was still running at
+    shutdown).
+    """
+
+    def __init__(
+        self,
+        name: str,
+        func: Callable[[object], Awaitable[None]],
+        plugin: str | None = None,
+    ):
+        self.name = name
+        self.state = "registered"
+        self.task: asyncio.Task | None = None
+        self.exception: BaseException | None = None
+        self.started_at: str | None = None
+        self.plugin = plugin
+        self._func = func
+        self._supervisor: BackgroundTaskSupervisor | None = None
+
+    def cancel(self) -> None:
+        """Cancel this task.
+
+        If it has already been launched, cancels the underlying
+        ``asyncio.Task`` - its state becomes ``cancelled`` once the
+        cancellation is observed (asynchronously, via the task's done
+        callback). If it has not been launched yet, this is a no-op as
+        far as asyncio is concerned (there's no task to cancel) but it
+        deregisters the handle from its supervisor so it never runs.
+        """
+        if self.task is not None:
+            self.task.cancel()
+        elif self._supervisor is not None:
+            self._supervisor._deregister(self)
+
+    def __repr__(self) -> str:
+        return f"<BackgroundTask name={self.name!r} state={self.state!r}>"
+
+
+class BackgroundTaskSupervisor:
+    """Owns registration and launch of every :class:`BackgroundTask` for a
+    single ``Datasette`` instance.
+
+    Registration (:meth:`add`) is separate from launch
+    (:meth:`launch_all`): plugins register work whenever convenient
+    (typically from a ``startup`` hook, but request handlers can register
+    dynamic per-job work too), and it either sits buffered until
+    :meth:`launch_all` runs, or - if :meth:`launch_all` has already run -
+    starts immediately.
+
+    Strong references to every :class:`BackgroundTask` (and its
+    ``asyncio.Task``) are kept for the life of the instance, by design -
+    that's what makes the enrichments-style "fire-and-forget task gets
+    garbage collected mid-flight" bug impossible here. There is currently
+    no pruning of completed/crashed/cancelled tasks, so a plugin that
+    dynamically registers many short-lived tasks over a long process
+    lifetime (a per-job registration pattern, e.g. one task per queued
+    job) will grow this list without bound. That's an accepted v1
+    trade-off in favour of full introspection (``/-/tasks``); revisit
+    with a pruning or capping policy if unbounded growth is reported in
+    practice.
+    """
+
+    def __init__(self, datasette):
+        self._datasette = datasette
+        self._tasks: list[BackgroundTask] = []
+        self._names = set()
+        self._launched = False
+        self._lock = asyncio.Lock()
+
+    def add(self, func, name=None) -> BackgroundTask:
+        base_name = name or getattr(func, "__qualname__", None) or repr(func)
+        actual_name = self._unique_name(base_name)
+        plugin = _resolve_plugin_name(func)
+        handle = BackgroundTask(actual_name, func, plugin=plugin)
+        handle._supervisor = self
+        self._tasks.append(handle)
+        self._names.add(actual_name)
+        if self._launched:
+            self._launch_one(handle)
+        return handle
+
+    def _unique_name(self, base_name: str) -> str:
+        if base_name not in self._names:
+            return base_name
+        n = 2
+        while f"{base_name}-{n}" in self._names:
+            n += 1
+        return f"{base_name}-{n}"
+
+    def _deregister(self, handle: BackgroundTask) -> None:
+        try:
+            self._tasks.remove(handle)
+        except ValueError:
+            pass
+        self._names.discard(handle.name)
+
+    def _launch_one(self, handle: BackgroundTask) -> None:
+        handle.state = "running"
+        handle.started_at = _utcnow_iso()
+        handle.task = asyncio.create_task(
+            handle._func(self._datasette), name=handle.name
+        )
+        handle.task.add_done_callback(functools.partial(_on_task_done, handle))
+
+    async def launch_all(self) -> None:
+        """Launch every currently-registered task that hasn't launched
+        yet. Idempotent and safe to call concurrently: subsequent (or
+        racing) calls are no-ops once the first has set ``self._launched``.
+        """
+        if self._launched:
+            return
+        async with self._lock:
+            if self._launched:
+                return
+            self._launched = True
+            for handle in list(self._tasks):
+                if handle.task is None:
+                    self._launch_one(handle)
+
+    async def cancel_all(self, grace: float = 5.0) -> None:
+        """Cancel every task that isn't already done, then wait up to
+        ``grace`` seconds for them to actually finish. Stragglers still
+        running after that are logged by name (but left to finish or not
+        on their own - this does not forcibly kill them, asyncio has no
+        mechanism for that).
+        """
+        handles_by_task = {
+            handle.task: handle for handle in self._tasks if handle.task is not None
+        }
+        pending = [task for task in handles_by_task if not task.done()]
+        for task in pending:
+            task.cancel()
+        if not pending:
+            return
+        _done, not_done = await asyncio.wait(pending, timeout=grace)
+        if not_done:
+            names = sorted(handles_by_task[task].name for task in not_done)
+            logger.warning(
+                "%d background task(s) did not finish within the %.1fs grace "
+                "period after cancellation: %s",
+                len(names),
+                grace,
+                ", ".join(names),
+            )
+
+    def tasks(self) -> list[BackgroundTask]:
+        """Return every registered :class:`BackgroundTask`, launched or
+        not, in registration order. Used by the ``/-/tasks`` debug
+        endpoint.
+        """
+        return list(self._tasks)
+
+    @property
+    def launched(self) -> bool:
+        """Whether :meth:`launch_all` has run yet - lets ``/-/tasks``
+        distinguish "no tasks registered" from "tasks registered but
+        nothing has armed the launch yet" without reaching for the
+        private ``_launched`` attribute.
+        """
+        return self._launched
+
+
+def _on_task_done(handle: BackgroundTask, task: asyncio.Task) -> None:
+    if task.cancelled():
+        handle.state = "cancelled"
+        return
+    exc = task.exception()
+    if exc is not None:
+        handle.state = "crashed"
+        handle.exception = exc
+        logger.error("Background task %r crashed", handle.name, exc_info=exc)
+        return
+    handle.state = "completed"

datasette/cli.py

@@ -678,6 +678,11 @@ def serve(
         except StartupError as e:
             raise click.ClickException(e.args[0])
 
+        # --get never launches background tasks: TestClient's request below
+        # flows through the full ASGI stack, including the
+        # AsgiRunOnFirstRequest fallback, which would otherwise launch them.
+        ds._suppress_background_tasks = True
+
         client = TestClient(ds)
         request_headers = {}
         if token:

datasette/hookspecs.py

@@ -9,6 +9,11 @@ def startup(datasette):
     """Fires directly after Datasette first starts running"""
 
 
+@hookspec
+def shutdown(datasette):
+    """Called once when the Datasette server is shutting down"""
+
+
 @hookspec
 def asgi_wrapper(datasette):
     """Returns an ASGI middleware callable to wrap our ASGI application with"""

docs/changelog.rst

@@ -4,6 +4,46 @@
 Changelog
 =========
 
+.. _v1_0_a39:
+
+1.0a39 (unreleased)
+-------------------
+
+This alpha gives plugins a real process lifecycle. Previously, ``datasette serve`` ran ``invoke_startup()`` on a temporary event loop that was closed before the server's own loop was created, so anything a ``startup`` hook scheduled with ``asyncio.create_task()`` - or any loop-bound primitive it created - could silently die before it ever ran. Every non-CLI deployment was worse off still: startup, including populating the internal database's table catalog, didn't run at all until the *first* HTTP request arrived, so plugins compensated with ``asgi_wrapper`` bootstrap shims, ``tryfirst=True`` ordering hacks and hand-rolled "has this started yet" flags. This release fixes all three problems together: one event loop for the whole process, startup wired into ASGI lifespan, and a new supervised background-task API plus a ``shutdown`` hook, so plugins no longer need to build any of that scaffolding themselves. See :ref:`datasette_lifecycle` for the full guarantee.
+
+- ``datasette serve`` now runs startup and the server on a single ``asyncio`` event loop, instead of a temporary loop that was discarded before ``uvicorn.run()`` created the loop that actually serves requests. A ``startup`` hook can now safely call ``asyncio.create_task()``, or create loop-bound primitives such as ``asyncio.Lock``, ``asyncio.Queue`` or ``asyncio.Event``, and expect them to still be alive once the server starts handling requests.
+- Startup - the internal database's table catalog, canned queries and column type configuration, and every :ref:`plugin_hook_startup` hook - is now wired into the ASGI ``lifespan.startup`` event via ``Datasette.app()``. A spec-compliant ASGI server (uvicorn, hypercorn, and others) completes ``lifespan.startup`` before delivering any request, so startup is now guaranteed to have finished before the first request in every deployment, not only ``datasette serve``; previously the table catalog in particular only populated on the first request, even when running under ``datasette serve``. A failing ``startup`` hook now surfaces as ``lifespan.startup.failed`` with the exception message, instead of leaving the ASGI host to hang or crash ambiguously.
+- The pre-existing first-request fallback is preserved as a safety net for hosts that never send ASGI lifespan events at all - some ASGI mounts, bare ``app()`` embedding, ``datasette.client``/test clients - and is idempotent alongside the lifespan path, so it's safe for both to fire.
+- New :ref:`datasette_add_background_task` API: plugins register supervised, long-lived background work - typically from a ``startup`` hook - and core owns launching it, once every ``startup`` hook has run. Core keeps a strong reference for the life of the process (no more silently garbage-collected fire-and-forget tasks), logs crashes with a full traceback to the ``datasette.background_tasks`` logger instead of a silent "Task exception was never retrieved", and cancels every task with a five-second grace period on shutdown. There is no automatic restart of a crashed task in this release. Registration returns a :ref:`BackgroundTask <BackgroundTask>` handle (``.name``, ``.state``, ``.task``, ``.exception``, ``.cancel()``). New :ref:`await datasette.start_background_tasks() <datasette_start_background_tasks>` method lets tests and headless embedders launch registered tasks explicitly, without running a server.
+- New ``/-/tasks`` JSON debug endpoint lists every supervised background task and its state, in the style of ``/-/threads``. See :ref:`JsonDataView_tasks`. It requires the ``permissions-debug`` permission, since a crashed task's recorded exception can reveal internal details such as file paths.
+- New :ref:`plugin_hook_shutdown` plugin hook, called during graceful shutdown (Ctrl-C, ``SIGTERM``) before background tasks are cancelled and before database connections are closed, so a plugin can tell its own background work to stop gracefully while a database connection is still available to write out final state. Exceptions raised by a ``shutdown`` hook are logged, not raised, so one plugin's broken teardown code cannot block another plugin's cleanup or Datasette's own database close. It is not called on a hard kill (``SIGKILL``).
+- Plugin ``asgi_wrapper`` middleware now always runs *after* startup has completed, on every deployment path including the first-request fallback - a wrapper that short-circuits and never calls the wrapped app (an auth check returning a 401, a CORS preflight response) can no longer defer startup indefinitely. ``lifespan`` scopes are unaffected by this change and continue to flow through plugin wrappers exactly as before.
+- The ``uvicorn`` dependency floor is now ``uvicorn>=0.29``, up from ``uvicorn>=0.11``.
+- ``datasette serve --headers`` and ``--token`` are only valid alongside ``--get``; that usage error is now raised immediately after the ``Datasette`` instance is constructed and before startup runs, instead of after ``invoke_startup()`` - and therefore every plugin's ``startup`` hook - had already executed.
+
+Migrating away from ``asgi_wrapper`` bootstrap hacks
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+If your plugin uses ``asgi_wrapper`` purely to detect "is this the first request" so that it can lazily start some background work, you can delete that code:
+
+.. list-table::
+   :header-rows: 1
+
+   * - Before
+     - After
+   * - An ``asgi_wrapper`` that checks a module-level flag and calls ``asyncio.create_task()`` (or awaits an ``async def`` closure) the first time it sees a request scope
+     - Call ``datasette.add_background_task()`` from a :ref:`plugin_hook_startup` hook
+   * - A hand-rolled ``_ensure_started`` / ``_started`` flag guarded by a lock, to avoid starting the work twice
+     - Not needed - registration and launch are both idempotent and safe to call from multiple places
+   * - An ``asgi_wrapper`` that sniffs the ``lifespan.shutdown`` message in its receive callable to run cleanup
+     - Implement the :ref:`plugin_hook_shutdown` hook instead
+   * - A fire-and-forget ``asyncio.create_task()`` with no reference kept, plus a README caveat like "no traffic, no runs" or "ping the server to keep the scheduler alive"
+     - ``datasette.add_background_task()`` - core keeps a strong reference and launches the task once, as soon as startup finishes, whether or not any request ever arrives
+   * - ``tryfirst=True`` on a ``startup`` hook, to make sure it runs before another plugin's task-starting code
+     - Not needed - ``add_background_task()`` launch happens only after *every* ``startup`` hook across every plugin has completed, so registration order between plugins doesn't matter
+
+`datasette-cron <https://datasette.io/plugins/datasette-cron>`__ and `datasette-enrichments <https://datasette.io/plugins/datasette-enrichments>`__ are being migrated to this pattern as worked examples of the mapping above.
+
 .. _v1_0_a38:
 
 1.0a38 (2026-08-06)

docs/internals.rst

@@ -1403,7 +1403,141 @@ Release all resources held by this ``Datasette`` instance. This calls :ref:`data
 
 If a call to ``Database.close()`` on one of the attached databases raises an exception, ``Datasette.close()`` will continue trying to close the remaining databases and will re-raise the first exception after every database has been processed.
 
-When Datasette is being served over ASGI the ``close()`` method is wired up to the lifespan shutdown event, so resources are released cleanly on ``SIGTERM`` / ``SIGINT``.
+When Datasette is being served over ASGI the ``close()`` method is wired up to the lifespan shutdown event, so resources are released cleanly on ``SIGTERM`` / ``SIGINT``. See :ref:`datasette_lifecycle` for where ``close()`` fits into the full startup-to-shutdown sequence.
+
+.. _datasette_lifecycle:
+
+Application lifecycle
+---------------------
+
+Datasette guarantees a fixed sequence of events between the moment a ``Datasette`` instance is constructed and the moment its resources are released:
+
+1. ``Datasette(...)`` — the constructor runs synchronously and does not run plugin hooks.
+2. **Startup** — ``await datasette.invoke_startup()`` runs once: it populates the internal database's catalog of table schemas (:ref:`internals_internal`), loads canned queries and column type configuration, then calls every registered :ref:`plugin_hook_startup` hook, in plugin registration order. When Datasette is being served, table-count precomputation for immutable databases runs immediately before this, as part of the same startup sequence.
+3. **Background-task launch** — once *every* ``startup`` hook has finished (not before), every task registered with :ref:`datasette_add_background_task` — by any plugin — is launched. A task registered by one plugin's ``startup`` hook can safely depend on state set up by another plugin's ``startup`` hook, because launch only happens after the whole round of hooks completes.
+4. **Serving** — the instance handles requests (or, for headless or CLI use, does whatever the embedding program does with it).
+5. **Shutdown** — triggered by the ASGI ``lifespan.shutdown`` event (Ctrl-C, ``SIGTERM``) or the end of a ``datasette serve`` process: every :ref:`plugin_hook_shutdown` hook runs first, while background tasks are still alive, so a plugin can tell its own task to wind down gracefully; every still-running background task is then cancelled and given a five-second grace period to actually stop; finally every database connection is released via :ref:`datasette_close`.
+
+.. admonition:: Startup hooks run on the event loop that serves requests
+
+   In every trigger path below, ``startup`` hooks run on the same ``asyncio`` event loop that goes on to accept connections. It is safe to create loop-bound primitives — ``asyncio.Lock``, ``asyncio.Queue``, ``asyncio.Event``, a raw ``asyncio.create_task()`` call — inside a ``startup`` hook, and to register long-lived background work with :ref:`datasette_add_background_task` there. This was not always true: older Datasette versions ran startup on a temporary event loop in the CLI that was closed before the server's own loop was created, which could silently kill anything scheduled on it.
+
+Three trigger paths
+~~~~~~~~~~~~~~~~~~~
+
+- **``datasette serve`` (CLI)** — startup and ``uvicorn.Server.serve()`` both run inside a single ``asyncio.run()`` call, so there is exactly one event loop for the whole life of the process.
+- **ASGI lifespan** — ``Datasette.app()`` wires startup and background-task launch into the ``on_startup`` list, and shutdown into the ``on_shutdown`` list, of an internal ``AsgiLifespan`` wrapper. A spec-compliant ASGI server (uvicorn, hypercorn, and others) sends the ``lifespan.startup`` message and waits for ``lifespan.startup.complete`` before delivering any ``http`` or ``websocket`` scope, so startup — including every plugin's own internal-database migrations — is guaranteed to have finished before any request reaches Datasette, including requests seen by plugin :ref:`asgi_wrapper <plugin_asgi_wrapper>` middleware. If a ``startup`` hook raises, ``AsgiLifespan`` sends ``lifespan.startup.failed`` with the exception message instead of hanging or crashing ambiguously, so the host can abort the boot cleanly.
+- **First-request fallback** — an internal ``AsgiRunOnFirstRequest`` wrapper runs the same startup work as a safety net for hosts that never send ASGI lifespan events at all: some ASGI mounts, a bare ``app()`` embedded inside another framework, and :ref:`datasette.client <internals_datasette_client>` / test clients, which drive requests directly over ``httpx.ASGITransport`` without ever emitting ``lifespan.startup``. It runs startup exactly once, the first time any non-lifespan scope arrives, guarded by a lock so that concurrent early requests can't run it twice.
+
+All three paths call the same idempotent internal methods, so it is safe for more than one of them to fire — lifespan startup completing and then a first request arriving afterwards is a no-op the second time. A host that never sends lifespan events and never goes through the CLI degrades to first-request timing: startup runs on the first request instead of before it, exactly as Datasette always worked prior to this lifecycle guarantee. This is a deliberate fallback rather than a regression — see :ref:`datasette_add_background_task` for how to opt out of launching background tasks (the ``--get`` CLI path) or drive startup and launch explicitly (tests, headless embedders).
+
+.. _datasette_add_background_task:
+
+.add_background_task(func, name=None)
+-------------------------------------
+
+``func`` - async callable
+    A coroutine function taking one positional argument, the ``Datasette`` instance. Core calls ``await func(datasette)``.
+
+``name`` - string, optional
+    A name for the task, used to identify it in the ``/-/tasks`` introspection endpoint (:ref:`JsonDataView_tasks`) and in log messages. Defaults to ``func.__qualname__``. If the resulting name collides with an already-registered task, a ``-2``, ``-3``, ... suffix is appended.
+
+Registers a piece of supervised, long-lived background work — typically called from a :ref:`plugin_hook_startup` hook, though it can be called at any point after the instance exists, including from a request handler. Returns a :ref:`BackgroundTask <BackgroundTask>` handle.
+
+Registration is separate from launch. Calling this from a ``startup`` hook — the common case — buffers the task; core launches every registered task once *all* ``startup`` hooks have completed, as described in :ref:`datasette_lifecycle`. Calling it after launch has already happened — for example from a request handler, to start a per-job task dynamically — starts the task immediately instead.
+
+.. code-block:: python
+
+    import asyncio
+    from datasette import hookimpl
+
+
+    async def poll_for_updates(datasette):
+        while True:
+            await do_one_poll(datasette)
+            await asyncio.sleep(60)
+
+
+    @hookimpl
+    def startup(datasette):
+        datasette.add_background_task(
+            poll_for_updates, name="my-plugin-poller"
+        )
+
+Core owns the task for the rest of the process's life:
+
+- **A strong reference is kept forever**, so the task can never be silently garbage collected the way an unreferenced ``asyncio.create_task()`` call can be.
+- **A crash is logged, not swallowed.** If ``func`` raises anything other than ``asyncio.CancelledError``, the exception (with its traceback) is logged to the ``datasette.background_tasks`` logger and recorded on the handle's ``.exception``, and the task's ``.state`` becomes ``crashed``. **There is no automatic restart in v1** — a long-running loop should catch and log its own transient errors internally if it wants to keep running after one.
+- **Cancellation is coordinated.** On shutdown, every task that is still running is cancelled and given a grace period to stop — see :ref:`datasette_lifecycle`.
+
+Raw ``asyncio.create_task()`` inside a ``startup`` hook now works correctly, because ``startup`` hooks run on the serving event loop (see the admonition in :ref:`datasette_lifecycle`) — the bug that made this unsafe is fixed. But a task created that way is unsupervised: nothing keeps a reference to it, nothing logs its exceptions, nothing cancels it on shutdown, and it will not show up in ``/-/tasks``. Prefer ``add_background_task()`` for anything long-lived.
+
+Launch matrix
+~~~~~~~~~~~~~
+
+Whether registered tasks actually launch depends on how the instance is being run:
+
+.. list-table::
+   :header-rows: 1
+
+   * - Trigger
+     - Launches registered tasks?
+   * - ASGI lifespan (real server deployments)
+     - Yes, after ``lifespan.startup`` completes
+   * - First-request fallback (lifespan-less hosts)
+     - Yes, on the first request — parity with the lifespan case
+   * - ``datasette serve --get``
+     - Never
+   * - Tests / headless embedders
+     - Only if you call :ref:`datasette_start_background_tasks` explicitly
+
+``datasette --get`` never launches background tasks, even though its one-shot request flows through the same first-request fallback as everything else: it sets an internal flag before making that request specifically to suppress the launch, since a one-shot CLI invocation has no server loop left running afterwards to keep any launched tasks alive.
+
+.. _BackgroundTask:
+
+BackgroundTask objects
+~~~~~~~~~~~~~~~~~~~~~~
+
+``add_background_task()`` returns a ``BackgroundTask`` handle with the following attributes:
+
+``.name`` - string
+    The task's (unique) name.
+
+``.state`` - string
+    One of ``registered`` (added but not yet launched), ``running``, ``completed`` (returned cleanly), ``crashed`` (raised an exception) or ``cancelled``.
+
+``.task`` - ``asyncio.Task`` or ``None``
+    The underlying ``asyncio.Task``, once launched. ``None`` while still ``registered``.
+
+``.exception`` - ``BaseException`` or ``None``
+    The exception that crashed the task, if ``.state`` is ``crashed``.
+
+``.started_at`` - string or ``None``
+    ISO 8601 UTC timestamp of when the task was launched.
+
+``.plugin`` - string or ``None``
+    Best-effort name of the plugin that registered the task, resolved from the module ``func`` was defined in. Used by ``/-/tasks`` and log messages; ``None`` if it cannot be determined.
+
+``.cancel()``
+    Cancel the task. If it has already launched, this cancels the underlying ``asyncio.Task`` — ``.state`` becomes ``cancelled`` once the cancellation is observed. If it has not launched yet, it is removed from the queue so it never runs.
+
+This is also the shape of each entry returned by the ``/-/tasks`` JSON introspection endpoint — see :ref:`JsonDataView_tasks`.
+
+.. _datasette_start_background_tasks:
+
+await .start_background_tasks()
+-------------------------------
+
+Runs startup (if it has not already run) and launches every task registered with :ref:`datasette_add_background_task`. This is the explicit equivalent of what happens automatically via ASGI lifespan or the first-request fallback in a served deployment — the entry point for tests and headless embedders (a cron-style CLI command that wants supervised background work without running a server) that need background tasks without going through either of those paths.
+
+.. code-block:: python
+
+    datasette = Datasette(memory=True)
+    await datasette.start_background_tasks()
+
+.. note::
+
+   ``start_background_tasks()`` calls ``invoke_startup()`` internally, **not** the fuller startup sequence a served instance uses — so calling it directly, without a prior request through ``datasette.client``, skips the immutable-database table-count precompute that a real server performs as part of startup. This only matters if your code inspects table counts before any request has been made; if you also exercise the instance via ``datasette.client`` (which arms the first-request fallback, and therefore the full startup sequence including table counts), or don't care about table counts up front, there is nothing to worry about.
 
 .. _datasette_track_event:
 

docs/introspection.rst

@@ -9,7 +9,7 @@ Each of these pages can be viewed in your browser. Add ``.json`` to the URL to g
 
 JSON responses that return an object include an ``"ok": true`` key, consistent with the rest of the :ref:`JSON API <json_api>`.
 
-The introspection endpoints documented on this page are covered by the :ref:`JSON API stability promise <json_api_stability>`, with the exception of the debug endpoints ``/-/threads`` and ``/-/actions``, whose shapes may change in future releases.
+The introspection endpoints documented on this page are covered by the :ref:`JSON API stability promise <json_api_stability>`, with the exception of the debug endpoints ``/-/threads``, ``/-/tasks`` and ``/-/actions``, whose shapes may change in future releases.
 
 .. _JsonDataView_metadata:
 
@@ -278,6 +278,57 @@ Shows details of threads and ``asyncio`` tasks. This endpoint requires the ``per
         ]
     }
 
+.. _JsonDataView_tasks:
+
+/-/tasks
+--------
+
+Shows the state of every supervised background task registered with
+:ref:`datasette.add_background_task() <datasette_add_background_task>`; see also
+:ref:`BackgroundTask <BackgroundTask>` for what each field below means, and
+:ref:`datasette_lifecycle` for when tasks are launched. This endpoint requires
+the ``permissions-debug`` permission, since a crashed task's ``exception``
+field can reveal internals such as file paths or query text:
+
+.. code-block:: json
+
+    {
+        "ok": true,
+        "tasks": [
+            {
+                "name": "my_plugin.poll_for_updates",
+                "state": "running",
+                "plugin": "my-plugin",
+                "started_at": "2026-07-30T12:00:00+00:00",
+                "exception": null
+            },
+            {
+                "name": "my_plugin.broken_task",
+                "state": "crashed",
+                "plugin": "my-plugin",
+                "started_at": "2026-07-30T12:00:00+00:00",
+                "exception": "ValueError('something went wrong')"
+            }
+        ],
+        "launched": true
+    }
+
+Each entry's ``state`` is one of ``registered`` (added but not yet
+launched), ``running``, ``completed``, ``crashed`` or ``cancelled``.
+``exception`` is a one-line ``repr()`` of the exception for a ``crashed``
+task, or ``null`` otherwise - the full traceback is written to the
+``datasette.background_tasks`` logger instead, to keep this payload
+skimmable.
+
+The top-level ``launched`` flag reports whether the instance has run its
+one-time background task launch (after ``startup`` hooks finish, or via
+lifespan/first-request/:ref:`start_background_tasks() <datasette_start_background_tasks>`).
+It distinguishes "no tasks have been registered" (``tasks`` is empty either
+way) from "tasks are registered but nothing has armed the launch yet"
+(``launched`` is ``false`` and every task's ``state`` is still
+``registered``) - useful when debugging a host that never triggers
+Datasette's lifespan events.
+
 .. _JsonDataView_actor:
 
 /-/actor

docs/json_api.rst

@@ -48,6 +48,7 @@ Some JSON endpoints are **exempt** from this promise:
   debug playground.
 - Debug and support endpoints are documented so you can use them, but their
   JSON shapes are not frozen: :ref:`/-/threads <JsonDataView_threads>`,
+  :ref:`/-/tasks <JsonDataView_tasks>`,
   :ref:`/-/actions <JsonDataView_actions>`,
   the :ref:`permission debug endpoints <PermissionsDebugView>`
   (``/-/allowed``, ``/-/rules``, ``/-/check``) and the

docs/plugin_hooks.rst

@@ -1157,7 +1157,7 @@ Examples: `datasette-cors <https://datasette.io/plugins/datasette-cors>`__, `dat
 startup(datasette)
 ------------------
 
-This hook fires when the Datasette application server first starts up.
+This hook fires when the Datasette application server first starts up. It runs on the same event loop that goes on to serve requests, so it is safe to create loop-bound primitives and register background work here — see :ref:`datasette_lifecycle` for the full guarantee and the three ways startup can be triggered.
 
 Here is an example that validates required plugin configuration. The server will fail to start and show an error if the validation check fails:
 
@@ -1195,6 +1195,7 @@ Potential use-cases:
 * Create database tables that a plugin needs on startup
 * Validate the configuration for a plugin on startup, and raise an error if it is invalid
 * Raise a ``datasette.utils.StartupError("message")`` exception to prevent Datasette from starting and display that message to the user.
+* Register supervised long-lived background work using :ref:`datasette_add_background_task`, which core launches once every plugin's ``startup()`` hook has finished.
 
 .. note::
 
@@ -1211,6 +1212,31 @@ Potential use-cases:
 
 Examples: `datasette-saved-queries <https://datasette.io/plugins/datasette-saved-queries>`__, `datasette-init <https://datasette.io/plugins/datasette-init>`__
 
+.. _plugin_hook_shutdown:
+
+shutdown(datasette)
+-------------------
+
+This hook fires once, when the Datasette application server is shutting down gracefully - triggered by the ASGI ``lifespan.shutdown`` event, which includes pressing Ctrl-C or sending ``SIGTERM`` to a ``datasette serve`` process. It is not called on a hard kill (``SIGKILL``), since there is no opportunity to run any code in that case.
+
+Like ``startup()``, this can be a regular function or it can return an async function to be awaited.
+
+It runs before Datasette cancels any background tasks it is supervising (see :ref:`datasette_add_background_task`) and before it closes its database connections, so you can use it to tell your plugin's own background work to stop gracefully while a database connection is still available to write out any final state. See :ref:`datasette_lifecycle` for exactly where this fits into the full startup-to-shutdown sequence:
+
+.. code-block:: python
+
+    @hookimpl
+    def shutdown(datasette):
+        async def inner():
+            db = datasette.get_database()
+            await db.execute_write(
+                "insert into shutdown_log (at) values (datetime('now'))"
+            )
+
+        return inner
+
+If your ``shutdown()`` hook raises an exception it will be logged but not re-raised, so one plugin's broken shutdown code cannot prevent other plugins - or Datasette itself - from finishing their own teardown.
+
 .. _plugin_hook_actor_from_request:
 
 actor_from_request(datasette, request)

docs/testing_plugins.rst

@@ -78,9 +78,19 @@ Creating a ``Datasette()`` instance like this as useful shortcut in tests, but t
     datasette = Datasette(memory=True)
     await datasette.invoke_startup()
 
-This method registers any :ref:`plugin_hook_startup` or :ref:`plugin_hook_prepare_jinja2_environment` plugins that might themselves need to make async calls.
+This method registers any :ref:`plugin_hook_startup` or :ref:`plugin_hook_prepare_jinja2_environment` plugins that might themselves need to make async calls. It runs on the same event loop that runs your test, matching the guarantee described in :ref:`datasette_lifecycle`.
 
-If you are using ``await datasette.client.get()`` and similar methods then you don't need to worry about this - Datasette automatically calls ``invoke_startup()`` the first time it handles a request.
+If you are using ``await datasette.client.get()`` and similar methods then you don't need to worry about this - Datasette automatically calls ``invoke_startup()`` the first time it handles a request, via the first-request fallback described in :ref:`datasette_lifecycle`.
+
+If your plugin also registers work with :ref:`datasette_add_background_task` (typically from a ``startup`` hook) and your test needs that work to actually run, call ``await datasette.start_background_tasks()`` as well - ``invoke_startup()`` alone only runs ``startup`` hooks, it does not launch anything they registered:
+
+.. code-block:: python
+
+    datasette = Datasette(memory=True)
+    await datasette.start_background_tasks()
+    # Any tasks registered by a startup() hook are now running
+
+A request made through ``datasette.client`` arms both startup and background-task launch automatically, since they're both part of the same first-request fallback - ``start_background_tasks()`` is for tests that need tasks running without making an HTTP request first.
 
 .. _testing_plugins_datasette_fixtures_database:
 

tests/test_background_tasks.py

@@ -0,0 +1,379 @@
+"""
+Tests for datasette.add_background_task() / start_background_tasks() and the
+BackgroundTask / BackgroundTaskSupervisor machinery in
+datasette/background_tasks.py.
+"""
+
+import asyncio
+import contextlib
+import logging
+
+import httpx
+import pytest
+
+from datasette import hookimpl
+from datasette.app import Datasette
+from datasette.plugins import pm
+
+
+async def _drive_lifespan_startup(app):
+    """Send a single lifespan.startup message into app's ASGI lifespan loop
+    and return the list of messages sent back, without ever sending
+    lifespan.shutdown. Copied from tests/test_lifespan.py's helper of the
+    same name - mirrors what a real server does: after startup completes
+    it parks waiting for the next event, and we cancel that wait once
+    we've observed the startup response.
+    """
+    messages_sent = []
+    startup_responded = asyncio.Event()
+    delivered = False
+
+    async def receive():
+        nonlocal delivered
+        if not delivered:
+            delivered = True
+            return {"type": "lifespan.startup"}
+        await asyncio.Event().wait()
+
+    async def send(message):
+        messages_sent.append(message)
+        startup_responded.set()
+
+    task = asyncio.create_task(app({"type": "lifespan"}, receive, send))
+    try:
+        await asyncio.wait_for(startup_responded.wait(), timeout=5)
+    finally:
+        task.cancel()
+        with contextlib.suppress(asyncio.CancelledError):
+            await task
+    return messages_sent
+
+
+@pytest.mark.asyncio
+async def test_tasks_registered_in_startup_hook_run_after_lifespan_startup():
+    # Two tasks registered by one plugin's startup hook - order preserved,
+    # both running after lifespan startup completes, and no HTTP request
+    # of any kind is issued anywhere in this test.
+    events = []
+
+    async def task_one(datasette):
+        events.append("task_one")
+        await asyncio.Event().wait()
+
+    async def task_two(datasette):
+        events.append("task_two")
+        await asyncio.Event().wait()
+
+    class TwoTaskPlugin:
+        __name__ = "TwoTaskPlugin"
+
+        @hookimpl
+        def startup(self, datasette):
+            async def inner():
+                datasette.add_background_task(task_one, name="task-one")
+                datasette.add_background_task(task_two, name="task-two")
+
+            return inner
+
+    ds = Datasette(memory=True)
+    pm.register(TwoTaskPlugin(), name="two_task_plugin")
+    try:
+        app = ds.app()
+        messages = await _drive_lifespan_startup(app)
+        assert {"type": "lifespan.startup.complete"} in messages
+
+        handles = ds._background_tasks.tasks()
+        assert [h.name for h in handles] == ["task-one", "task-two"]
+
+        # Let both tasks run their first line of code.
+        await asyncio.sleep(0)
+        assert handles[0].state == "running"
+        assert handles[1].state == "running"
+        assert events == ["task_one", "task_two"]
+    finally:
+        pm.unregister(name="two_task_plugin")
+        await ds._background_tasks.cancel_all(grace=1.0)
+
+
+@pytest.mark.asyncio
+async def test_launch_waits_for_every_startup_hook_before_running_any_task():
+    # PluginA registers a task from its startup hook; PluginB does the
+    # same from ITS startup hook, which runs after PluginA's (forced with
+    # tryfirst=True on A). Even though A's registration happens first,
+    # A's task body must not actually execute until every startup hook -
+    # including B's - has finished, since launch only happens after
+    # invoke_startup() completes. This is the ordering guarantee that
+    # dissolves datasette-cron's tryfirst=True launch hack.
+    hook_call_order = []
+    seen_names_when_a_ran = {}
+
+    async def task_a(datasette):
+        seen_names_when_a_ran["names"] = [
+            h.name for h in datasette._background_tasks.tasks()
+        ]
+
+    async def task_b(datasette):
+        pass
+
+    class PluginA:
+        __name__ = "PluginA"
+
+        @hookimpl(tryfirst=True)
+        def startup(self, datasette):
+            async def inner():
+                hook_call_order.append("A")
+                datasette.add_background_task(task_a, name="task-a")
+
+            return inner
+
+    class PluginB:
+        __name__ = "PluginB"
+
+        @hookimpl
+        def startup(self, datasette):
+            async def inner():
+                hook_call_order.append("B")
+                datasette.add_background_task(task_b, name="task-b")
+
+            return inner
+
+    ds = Datasette(memory=True)
+    pm.register(PluginA(), name="plugin_a")
+    pm.register(PluginB(), name="plugin_b")
+    try:
+        await ds.start_background_tasks()
+        # Confirm A's startup hook really did run (and register task-a)
+        # strictly before B's startup hook ran.
+        assert hook_call_order == ["A", "B"]
+
+        handles = ds._background_tasks.tasks()
+        await asyncio.wait_for(asyncio.gather(*[h.task for h in handles]), timeout=5)
+        # Yet by the time task-a's own body executed (after launch, which
+        # only happens once every startup hook - including B's - has
+        # finished), task-b was already registered.
+        assert "task-b" in seen_names_when_a_ran["names"]
+    finally:
+        pm.unregister(name="plugin_a")
+        pm.unregister(name="plugin_b")
+
+
+@pytest.mark.asyncio
+async def test_concurrent_first_requests_launch_background_tasks_exactly_once():
+    launch_count = {"n": 0}
+
+    async def counting_task(datasette):
+        launch_count["n"] += 1
+
+    class CountingTaskPlugin:
+        __name__ = "CountingTaskPlugin"
+
+        @hookimpl
+        def startup(self, datasette):
+            async def inner():
+                datasette.add_background_task(counting_task, name="counting-task")
+
+            return inner
+
+    ds = Datasette(memory=True)
+    pm.register(CountingTaskPlugin(), name="counting_task_plugin")
+    try:
+        app = ds.app()
+        transport = httpx.ASGITransport(app=app)
+        async with httpx.AsyncClient(
+            transport=transport, base_url="http://localhost"
+        ) as client:
+            responses = await asyncio.gather(
+                *[client.get("/-/versions.json") for _ in range(10)]
+            )
+        assert all(response.status_code == 200 for response in responses)
+
+        handles = ds._background_tasks.tasks()
+        assert len(handles) == 1
+        await asyncio.wait_for(handles[0].task, timeout=5)
+        assert launch_count["n"] == 1
+    finally:
+        pm.unregister(name="counting_task_plugin")
+
+
+@pytest.mark.asyncio
+async def test_post_launch_registration_starts_immediately_and_cancel_works():
+    ds = Datasette(memory=True)
+    await ds.start_background_tasks()  # nothing registered yet, but launched
+
+    started = asyncio.Event()
+
+    async def long_running(datasette):
+        started.set()
+        await asyncio.Event().wait()
+
+    handle = ds.add_background_task(long_running, name="dynamic-task")
+    # Registered after launch: starts immediately rather than sitting in
+    # "registered" limbo.
+    assert handle.state == "running"
+    assert handle.task is not None
+
+    await asyncio.wait_for(started.wait(), timeout=5)
+    assert handle.state == "running"
+
+    handle.cancel()
+    with pytest.raises(asyncio.CancelledError):
+        await handle.task
+    await asyncio.sleep(0)
+    assert handle.state == "cancelled"
+
+
+@pytest.mark.asyncio
+async def test_pre_launch_registration_starts_as_registered():
+    ds = Datasette(memory=True)
+
+    async def task(datasette):
+        pass
+
+    handle = ds.add_background_task(task, name="buffered-task")
+    assert handle.state == "registered"
+    assert handle.task is None
+
+    handle.cancel()  # not yet launched: deregisters instead of cancelling
+    assert handle not in ds._background_tasks.tasks()
+
+
+@pytest.mark.asyncio
+async def test_crashing_task_logs_traceback_and_state_is_crashed(caplog):
+    ds = Datasette(memory=True)
+    await ds.start_background_tasks()
+
+    survivor_ran = asyncio.Event()
+
+    async def crashing_task(datasette):
+        raise RuntimeError("kaboom")
+
+    async def survivor(datasette):
+        survivor_ran.set()
+
+    with caplog.at_level(logging.ERROR, logger="datasette.background_tasks"):
+        crash_handle = ds.add_background_task(crashing_task, name="crashing_task")
+        survivor_handle = ds.add_background_task(survivor, name="survivor")
+        await asyncio.wait_for(
+            asyncio.gather(
+                crash_handle.task, survivor_handle.task, return_exceptions=True
+            ),
+            timeout=5,
+        )
+
+    assert crash_handle.state == "crashed"
+    assert isinstance(crash_handle.exception, RuntimeError)
+    assert str(crash_handle.exception) == "kaboom"
+
+    # The crash must not affect any other task.
+    assert survivor_ran.is_set()
+    assert survivor_handle.state == "completed"
+
+    assert "crashing_task" in caplog.text
+    assert "kaboom" in caplog.text
+    assert "Traceback" in caplog.text
+    assert "RuntimeError" in caplog.text
+
+
+def test_name_collisions_get_suffixed_and_explicit_names_are_respected():
+    ds = Datasette(memory=True)
+
+    async def noop(datasette):
+        pass
+
+    async def another_noop(datasette):
+        pass
+
+    h1 = ds.add_background_task(noop, name="dup")
+    h2 = ds.add_background_task(another_noop, name="dup")
+    h3 = ds.add_background_task(noop, name="dup")
+    assert [h1.name, h2.name, h3.name] == ["dup", "dup-2", "dup-3"]
+
+    h_explicit = ds.add_background_task(noop, name="explicit-name")
+    assert h_explicit.name == "explicit-name"
+
+    h_default = ds.add_background_task(noop)
+    assert h_default.name == noop.__qualname__
+
+
+@pytest.mark.asyncio
+async def test_start_background_tasks_on_bare_datasette():
+    # The headless-CLI path (datasette-rss's `fetch --due` shape): no
+    # server, no lifespan, no first HTTP request - just an explicit call.
+    ran = asyncio.Event()
+
+    async def task(datasette):
+        ran.set()
+
+    ds = Datasette([])
+    assert ds._startup_invoked is False
+
+    handle = ds.add_background_task(task, name="headless-task")
+    assert handle.state == "registered"
+
+    await ds.start_background_tasks()
+
+    assert ds._startup_invoked is True
+    await asyncio.wait_for(ran.wait(), timeout=5)
+    await asyncio.wait_for(handle.task, timeout=5)
+    # handle.task being done only guarantees the coroutine has returned,
+    # not that our done-callback (which updates handle.state) has run yet -
+    # asyncio schedules done-callbacks via call_soon, and awaiting an
+    # already-done future/task returns immediately without giving the loop
+    # a chance to drain its ready queue. Yield once to let it run.
+    await asyncio.sleep(0)
+    assert handle.state == "completed"
+
+
+@pytest.mark.asyncio
+async def test_cancel_all_cancels_running_tasks_and_leaves_completed_alone():
+    ds = Datasette(memory=True)
+    await ds.start_background_tasks()
+
+    async def forever(datasette):
+        await asyncio.Event().wait()
+
+    async def quick(datasette):
+        return "done"
+
+    forever_handle = ds.add_background_task(forever, name="forever")
+    quick_handle = ds.add_background_task(quick, name="quick")
+    await asyncio.wait_for(quick_handle.task, timeout=5)
+    assert quick_handle.state == "completed"
+
+    await ds._background_tasks.cancel_all(grace=1.0)
+
+    assert forever_handle.state == "cancelled"
+    assert quick_handle.state == "completed"
+
+
+@pytest.mark.asyncio
+async def test_cancel_all_logs_stragglers_that_outlive_the_grace_period(caplog):
+    ds = Datasette(memory=True)
+    await ds.start_background_tasks()
+
+    async def stubborn(datasette):
+        with contextlib.suppress(asyncio.CancelledError):
+            await asyncio.sleep(10)
+        # Swallowing CancelledError above and returning normally simulates
+        # a task that ignores cancellation for longer than the grace period.
+        await asyncio.sleep(10)
+
+    handle = ds.add_background_task(stubborn, name="stubborn-task")
+    # Let the task actually start running and reach its first sleep (inside
+    # the CancelledError-suppressing block) before cancelling it - a task
+    # cancelled before it has ever run its first step never enters that
+    # block at all (the throw happens before the coroutine body starts),
+    # so it would finish cancelling immediately instead of behaving like a
+    # straggler.
+    await asyncio.sleep(0)
+
+    with caplog.at_level(logging.WARNING, logger="datasette.background_tasks"):
+        await ds._background_tasks.cancel_all(grace=0.1)
+
+    assert "stubborn-task" in caplog.text
+
+    # Clean up: actually cancel it now that the test has made its
+    # assertion, so it doesn't leak past the end of the test.
+    handle.task.cancel()
+    with contextlib.suppress(asyncio.CancelledError):
+        await handle.task

tests/test_cli_serve_get.py

@@ -52,6 +52,56 @@ def test_serve_with_get(tmp_path_factory):
     pm.unregister(to_unregister)
 
 
+def test_serve_with_get_does_not_launch_background_tasks(tmp_path_factory):
+    # --get must never launch background tasks, even though its TestClient
+    # request
+    # flows through the full ASGI stack (including the AsgiRunOnFirstRequest
+    # fallback that would otherwise launch them). The plugin's startup hook
+    # itself still runs (registration happens) - only the launch is
+    # suppressed, so the sentinel file the background task would write must
+    # never appear.
+    plugins_dir = tmp_path_factory.mktemp("plugins_for_get_background_tasks")
+    sentinel = plugins_dir / "sentinel.txt"
+    (plugins_dir / "bg_task_for_get.py").write_text(
+        textwrap.dedent(
+            f"""
+        from datasette import hookimpl
+
+        @hookimpl
+        def startup(datasette):
+            async def inner():
+                async def task(datasette):
+                    with open("{sentinel!s}", "w") as fp:
+                        fp.write("ran")
+
+                datasette.add_background_task(task, name="get-sentinel-task")
+
+            return inner
+    """,
+        ),
+        "utf-8",
+    )
+    runner = CliRunner()
+    result = runner.invoke(
+        cli,
+        [
+            "serve",
+            "--memory",
+            "--plugins-dir",
+            str(plugins_dir),
+            "--get",
+            "/_memory/-/query.json?sql=select+1",
+        ],
+    )
+    assert result.exit_code == 0, result.output
+    assert not sentinel.exists()
+
+    to_unregister = next(
+        p for p in pm.get_plugins() if p.__name__ == "bg_task_for_get.py"
+    )
+    pm.unregister(to_unregister)
+
+
 def test_serve_with_get_headers():
     runner = CliRunner()
     result = runner.invoke(

tests/test_cli_serve_server.py

@@ -1,4 +1,6 @@
+import signal
 import socket
+import subprocess
 import time
 
 import httpx
@@ -145,3 +147,83 @@ def test_startup_error_fails_fast_before_port_binds(serve_with_plugins):
         socket.create_connection(("127.0.0.1", port), timeout=0.2),
     ):
         pass
+
+
+# Proves the ticket-01 serve path (asyncio.run(_serve_async()) wrapping
+# uvicorn.Server.serve()) actually delivers a graceful signal through to
+# uvicorn's lifespan.shutdown, which now runs Datasette.invoke_shutdown()
+# and therefore every plugin's `shutdown` hook. The plugin below writes a
+# sentinel file from inside its shutdown hook so this can be checked from
+# outside the subprocess after it exits.
+SHUTDOWN_SENTINEL_PLUGIN_TEMPLATE = """
+import pathlib
+from datasette import hookimpl
+
+SENTINEL_PATH = {sentinel_path!r}
+
+
+@hookimpl
+def shutdown(datasette):
+    pathlib.Path(SENTINEL_PATH).write_text("shutdown ran", "utf-8")
+"""
+
+
+def _start_serve_with_shutdown_sentinel(serve_with_plugins, tmp_path):
+    sentinel_path = tmp_path / "shutdown-sentinel.txt"
+    proc, _ = serve_with_plugins(
+        {
+            "shutdown_sentinel_plugin": SHUTDOWN_SENTINEL_PLUGIN_TEMPLATE.format(
+                sentinel_path=str(sentinel_path)
+            )
+        }
+    )
+    return proc, sentinel_path
+
+
+@pytest.mark.serial
+def test_sigterm_runs_shutdown_hooks(serve_with_plugins, tmp_path):
+    ds_proc, sentinel_path = _start_serve_with_shutdown_sentinel(
+        serve_with_plugins, tmp_path
+    )
+    assert not sentinel_path.exists()
+    ds_proc.send_signal(signal.SIGTERM)
+    try:
+        ds_proc.wait(timeout=10)
+    except subprocess.TimeoutExpired:
+        ds_proc.kill()
+        ds_proc.wait()
+        raise AssertionError(
+            "datasette serve did not exit within 10s of SIGTERM\n"
+            + ds_proc.stdout.read().decode("utf-8")
+        )
+    output = ds_proc.stdout.read().decode("utf-8")
+    assert sentinel_path.exists(), (
+        "shutdown hook never wrote its sentinel file after SIGTERM\n" + output
+    )
+    assert sentinel_path.read_text("utf-8") == "shutdown ran"
+
+
+@pytest.mark.serial
+@pytest.mark.skipif(
+    not hasattr(signal, "SIGINT"), reason="Requires signal.SIGINT support"
+)
+def test_sigint_runs_shutdown_hooks(serve_with_plugins, tmp_path):
+    ds_proc, sentinel_path = _start_serve_with_shutdown_sentinel(
+        serve_with_plugins, tmp_path
+    )
+    assert not sentinel_path.exists()
+    ds_proc.send_signal(signal.SIGINT)
+    try:
+        ds_proc.wait(timeout=10)
+    except subprocess.TimeoutExpired:
+        ds_proc.kill()
+        ds_proc.wait()
+        raise AssertionError(
+            "datasette serve did not exit within 10s of SIGINT\n"
+            + ds_proc.stdout.read().decode("utf-8")
+        )
+    output = ds_proc.stdout.read().decode("utf-8")
+    assert sentinel_path.exists(), (
+        "shutdown hook never wrote its sentinel file after SIGINT\n" + output
+    )
+    assert sentinel_path.read_text("utf-8") == "shutdown ran"

tests/test_lifespan.py

@@ -7,6 +7,13 @@ These exercise Datasette._startup_sequence() via three different callers:
 - AsgiRunOnFirstRequest, the fallback for hosts that never send lifespan
   events (this is what DatasetteClient / plain httpx.ASGITransport uses)
 - Both at once, to prove startup hooks run at most once
+
+Also covers the asgi_wrapper reorder:
+plugin asgi_wrapper middleware runs INSIDE AsgiRunOnFirstRequest (below it)
+but OUTSIDE AsgiLifespan (above it), so wrappers only ever see http/
+websocket scopes after startup has completed, in both the lifespan and
+fallback paths - while lifespan scopes still flow through wrappers
+unchanged.
 """
 
 import asyncio
@@ -257,3 +264,144 @@ async def test_setup_db_still_runs_when_invoke_startup_ran_first(tmp_path, monke
     # Idempotency: a second call must not recompute.
     await ds._startup_sequence()
     assert call_count["n"] == 1
+
+
+@pytest.mark.asyncio
+async def test_asgi_wrapper_runs_after_startup_fallback_path():
+    # Ticket 05: plugin asgi_wrapper middleware must never see an http scope
+    # before startup has completed - even on the fallback path (no ASGI
+    # lifespan events at all), which is what plain httpx.ASGITransport /
+    # bare app() embedding exercises. Before the app() reorder in this
+    # ticket, the wrapper loop ran OUTSIDE (above) AsgiRunOnFirstRequest, so
+    # this assertion could see _startup_invoked is False on request #1 -
+    # this test fails on the pre-reorder app().
+    class AssertStartupPlugin:
+        __name__ = "AssertStartupPlugin"
+
+        @hookimpl
+        def asgi_wrapper(self, datasette):
+            def wrap(app):
+                async def check_startup(scope, receive, send):
+                    if scope["type"] == "http":
+                        assert (
+                            datasette._startup_invoked is True
+                        ), "asgi_wrapper saw an http scope before startup completed"
+                    await app(scope, receive, send)
+
+                return check_startup
+
+            return wrap
+
+    ds = Datasette(memory=True)
+    pm.register(AssertStartupPlugin(), name="assert_startup_plugin")
+    try:
+        assert ds._startup_invoked is False
+        app = ds.app()
+        transport = httpx.ASGITransport(app=app)
+        async with httpx.AsyncClient(
+            transport=transport, base_url="http://localhost"
+        ) as client:
+            response = await client.get("/-/versions.json")
+            assert response.status_code == 200
+    finally:
+        pm.unregister(name="assert_startup_plugin")
+
+    assert ds._startup_invoked is True
+
+
+@pytest.mark.asyncio
+async def test_short_circuit_wrapper_no_longer_defers_startup():
+    # Ticket 05: a wrapper that short-circuits (returns a response without
+    # ever calling the inner app - the shape of a 401/403/CORS-preflight
+    # responder) used to mean startup never ran for that request, because
+    # the wrapper sat OUTSIDE AsgiRunOnFirstRequest. Now that
+    # AsgiRunOnFirstRequest is outermost, it arms startup before the
+    # wrapper (or anything else) ever sees the scope.
+    class ShortCircuitPlugin:
+        __name__ = "ShortCircuitPlugin"
+
+        @hookimpl
+        def asgi_wrapper(self, datasette):
+            def wrap(app):
+                async def forbidden(scope, receive, send):
+                    if scope["type"] != "http":
+                        await app(scope, receive, send)
+                        return
+                    await send(
+                        {
+                            "type": "http.response.start",
+                            "status": 403,
+                            "headers": [[b"content-type", b"text/plain"]],
+                        }
+                    )
+                    await send(
+                        {
+                            "type": "http.response.body",
+                            "body": b"Forbidden",
+                        }
+                    )
+                    # Deliberately never call app(...): this is the
+                    # short-circuiting shape (auth-passwords, auth-tailscale,
+                    # datasette-cors preflight).
+
+                return forbidden
+
+            return wrap
+
+    ds = Datasette(memory=True)
+    pm.register(ShortCircuitPlugin(), name="short_circuit_plugin")
+    try:
+        assert ds._startup_invoked is False
+        app = ds.app()
+        transport = httpx.ASGITransport(app=app)
+        async with httpx.AsyncClient(
+            transport=transport, base_url="http://localhost"
+        ) as client:
+            response = await client.get("/-/versions.json")
+            assert response.status_code == 403
+    finally:
+        pm.unregister(name="short_circuit_plugin")
+
+    assert ds._startup_invoked is True
+
+
+@pytest.mark.asyncio
+async def test_asgi_wrapper_still_sees_lifespan_scopes():
+    # Ticket 05: the reorder only moves AsgiRunOnFirstRequest outside the
+    # wrapper loop - AsgiLifespan stays inside it, so a wrapper that
+    # inspects/wraps lifespan scopes still sees identical message flow.
+    # Pin that lifespan scopes keep reaching wrappers alongside http scopes.
+    seen_types = []
+
+    class RecordingPlugin:
+        __name__ = "RecordingPlugin"
+
+        @hookimpl
+        def asgi_wrapper(self, datasette):
+            def wrap(app):
+                async def record(scope, receive, send):
+                    seen_types.append(scope["type"])
+                    await app(scope, receive, send)
+
+                return record
+
+            return wrap
+
+    ds = Datasette(memory=True)
+    pm.register(RecordingPlugin(), name="recording_plugin")
+    try:
+        app = ds.app()
+        messages = await _drive_lifespan_startup(app)
+        assert {"type": "lifespan.startup.complete"} in messages
+
+        transport = httpx.ASGITransport(app=app)
+        async with httpx.AsyncClient(
+            transport=transport, base_url="http://localhost"
+        ) as client:
+            response = await client.get("/-/versions.json")
+            assert response.status_code == 200
+    finally:
+        pm.unregister(name="recording_plugin")
+
+    assert "lifespan" in seen_types
+    assert "http" in seen_types

tests/test_permissions.py

@@ -520,6 +520,7 @@ def view_instance_client():
         "/-/plugins",
         "/-/settings",
         "/-/threads",
+        "/-/tasks",
         "/-/databases",
         "/-/permissions",
         "/-/messages",

tests/test_plugins.py

@@ -53,6 +53,13 @@ def test_hook_jump_items_sql():
     assert "jump_items_sql" in dir(pm.hook)
 
 
+def test_hook_shutdown():
+    # Detailed behavior (ordering against background-task cancellation and
+    # close(), idempotency, exception handling, sync vs async support) is
+    # covered in tests/test_shutdown.py.
+    assert "shutdown" in dir(pm.hook)
+
+
 @pytest.mark.asyncio
 async def test_hook_plugins_dir_plugin_prepare_connection(ds_client):
     response = await ds_client.get(

tests/test_shutdown.py

@@ -0,0 +1,318 @@
+"""
+Tests for the shutdown(datasette) plugin hook and Datasette.invoke_shutdown().
+
+Order under test: plugin `shutdown` hooks run first (while background tasks
+are still alive) -> supervised background tasks are cancelled and drained
+(fixed 5s grace) -> databases are closed. Hook exceptions are logged, never
+propagated, and never skip a later step. The whole sequence is idempotent,
+so a second lifespan.shutdown message (or any other second caller) is a
+no-op.
+"""
+
+import asyncio
+import contextlib
+import logging
+
+import pytest
+
+from datasette import hookimpl
+from datasette.app import Datasette
+from datasette.plugins import pm
+
+
+async def _drive_lifespan(app, messages):
+    """Drive a single ASGI lifespan connection against `app`, feeding
+    `messages` to it in order via receive(). Returns once `app` itself
+    returns, which - per AsgiLifespan - happens after it has processed a
+    lifespan.shutdown message and sent back .complete or .failed. Returns
+    the list of messages `app` sent via send().
+
+    `messages` must end with a lifespan.shutdown (or a startup that fails)
+    or this will hang forever waiting for a message that never comes,
+    since AsgiLifespan only returns after handling shutdown.
+    """
+    sent = []
+    idx = 0
+
+    async def receive():
+        nonlocal idx
+        if idx < len(messages):
+            message = messages[idx]
+            idx += 1
+            return message
+        # Real servers park here waiting for lifespan.shutdown; nothing
+        # left to deliver in this test, so just block - the caller is
+        # expected to have already gotten what it needs via a preceding
+        # lifespan.shutdown in `messages`.
+        await asyncio.Event().wait()
+
+    async def send(message):
+        sent.append(message)
+
+    await app({"type": "lifespan"}, receive, send)
+    return sent
+
+
+@pytest.mark.asyncio
+async def test_shutdown_hook_runs_before_task_cancellation_then_closes():
+    # Ordering proof: the shutdown hook observes the background task still
+    # "running" (hooks run BEFORE cancellation); after the whole lifespan
+    # drive completes, the task is "cancelled" and the Datasette instance
+    # is closed.
+    events = []
+    task_state_seen_in_hook = {}
+
+    async def bg_task(datasette):
+        await asyncio.Event().wait()
+
+    class LifecyclePlugin:
+        __name__ = "LifecyclePlugin"
+
+        @hookimpl
+        def startup(self, datasette):
+            async def inner():
+                datasette.add_background_task(bg_task, name="bg-task")
+
+            return inner
+
+        @hookimpl
+        def shutdown(self, datasette):
+            async def inner():
+                events.append("shutdown-hook-ran")
+                handle = datasette._background_tasks.tasks()[0]
+                task_state_seen_in_hook["state"] = handle.state
+
+            return inner
+
+    ds = Datasette(memory=True)
+    pm.register(LifecyclePlugin(), name="lifecycle_plugin")
+    try:
+        app = ds.app()
+        messages = await _drive_lifespan(
+            app, [{"type": "lifespan.startup"}, {"type": "lifespan.shutdown"}]
+        )
+    finally:
+        pm.unregister(name="lifecycle_plugin")
+
+    assert {"type": "lifespan.startup.complete"} in messages
+    assert {"type": "lifespan.shutdown.complete"} in messages
+    assert events == ["shutdown-hook-ran"]
+    assert task_state_seen_in_hook["state"] == "running"
+
+    handle = ds._background_tasks.tasks()[0]
+    assert handle.state == "cancelled"
+    assert ds._closed is True
+    assert ds._shutdown_invoked is True
+
+
+@pytest.mark.asyncio
+async def test_second_lifespan_shutdown_does_not_double_invoke():
+    call_count = {"n": 0}
+
+    class CountingShutdownPlugin:
+        __name__ = "CountingShutdownPlugin"
+
+        @hookimpl
+        def shutdown(self, datasette):
+            async def inner():
+                call_count["n"] += 1
+
+            return inner
+
+    ds = Datasette(memory=True)
+    pm.register(CountingShutdownPlugin(), name="counting_shutdown_plugin")
+    try:
+        app = ds.app()
+        messages = await _drive_lifespan(
+            app, [{"type": "lifespan.startup"}, {"type": "lifespan.shutdown"}]
+        )
+        assert {"type": "lifespan.shutdown.complete"} in messages
+        assert call_count["n"] == 1
+
+        # A second, separate lifespan connection (a misbehaving host, or a
+        # second embedder driving the same Datasette instance) sends
+        # lifespan.shutdown again. AsgiLifespan itself has no memory of
+        # the earlier connection, so this exercises invoke_shutdown()'s
+        # own idempotency guard, not anything ASGI-layer.
+        messages2 = await _drive_lifespan(app, [{"type": "lifespan.shutdown"}])
+        assert {"type": "lifespan.shutdown.complete"} in messages2
+        assert call_count["n"] == 1
+    finally:
+        pm.unregister(name="counting_shutdown_plugin")
+
+    assert ds._closed is True
+
+
+@pytest.mark.asyncio
+async def test_raising_shutdown_hook_is_logged_and_does_not_block_the_rest(caplog):
+    events = []
+
+    async def bg_task(datasette):
+        await asyncio.Event().wait()
+
+    class RaisingShutdownPlugin:
+        __name__ = "RaisingShutdownPlugin"
+
+        @hookimpl
+        def shutdown(self, datasette):
+            async def inner():
+                raise RuntimeError("boom from shutdown hook")
+
+            return inner
+
+    class WellBehavedPlugin:
+        __name__ = "WellBehavedPlugin"
+
+        @hookimpl
+        def startup(self, datasette):
+            async def inner():
+                datasette.add_background_task(bg_task, name="bg-task")
+
+            return inner
+
+        @hookimpl
+        def shutdown(self, datasette):
+            async def inner():
+                events.append("well-behaved-ran")
+
+            return inner
+
+    ds = Datasette(memory=True)
+    pm.register(RaisingShutdownPlugin(), name="raising_shutdown_plugin")
+    pm.register(WellBehavedPlugin(), name="well_behaved_plugin")
+    try:
+        app = ds.app()
+        with caplog.at_level(logging.ERROR, logger="datasette"):
+            messages = await _drive_lifespan(
+                app, [{"type": "lifespan.startup"}, {"type": "lifespan.shutdown"}]
+            )
+    finally:
+        pm.unregister(name="raising_shutdown_plugin")
+        pm.unregister(name="well_behaved_plugin")
+
+    # The exception must never turn into lifespan.shutdown.failed - it's
+    # swallowed inside invoke_shutdown(), logged, and teardown continues.
+    assert {"type": "lifespan.shutdown.complete"} in messages
+    assert events == ["well-behaved-ran"]
+    assert "shutdown hook failed" in caplog.text
+    assert "boom from shutdown hook" in caplog.text
+
+    handle = ds._background_tasks.tasks()[0]
+    assert handle.state == "cancelled"
+    assert ds._closed is True
+
+
+@pytest.mark.asyncio
+async def test_sync_shutdown_hook_variant_works():
+    events = []
+
+    class SyncShutdownPlugin:
+        __name__ = "SyncShutdownPlugin"
+
+        @hookimpl
+        def shutdown(self, datasette):
+            # Deliberately not returning a coroutine/callable - a plain
+            # sync hookimpl, same as `def startup(datasette): ...` is
+            # supported via await_me_maybe.
+            events.append("sync-shutdown-ran")
+
+    ds = Datasette(memory=True)
+    pm.register(SyncShutdownPlugin(), name="sync_shutdown_plugin")
+    try:
+        app = ds.app()
+        messages = await _drive_lifespan(
+            app, [{"type": "lifespan.startup"}, {"type": "lifespan.shutdown"}]
+        )
+    finally:
+        pm.unregister(name="sync_shutdown_plugin")
+
+    assert {"type": "lifespan.shutdown.complete"} in messages
+    assert events == ["sync-shutdown-ran"]
+    assert ds._closed is True
+
+
+@pytest.mark.asyncio
+async def test_shutdown_logs_stragglers_that_outlive_the_grace_period(
+    caplog, monkeypatch
+):
+    async def stubborn(datasette):
+        with contextlib.suppress(asyncio.CancelledError):
+            await asyncio.sleep(10)
+        # Swallowing CancelledError above and returning normally simulates
+        # a task that ignores cancellation for longer than the grace
+        # period - same shape as test_background_tasks.py's equivalent
+        # test.
+        await asyncio.sleep(10)
+
+    class StubbornTaskPlugin:
+        __name__ = "StubbornTaskPlugin"
+
+        @hookimpl
+        def startup(self, datasette):
+            async def inner():
+                datasette.add_background_task(stubborn, name="stubborn-task")
+
+            return inner
+
+    ds = Datasette(memory=True)
+    pm.register(StubbornTaskPlugin(), name="stubborn_task_plugin")
+
+    # invoke_shutdown() calls self._background_tasks.cancel_all(grace=5.0)
+    # with a grace hardcoded in app.py, per the ticket. Rather than
+    # sleeping for a real 5s in this test, monkeypatch the
+    # BackgroundTaskSupervisor *instance's* cancel_all to a wrapper that
+    # ignores the caller-supplied grace and substitutes a small one - this
+    # is the cleanest seam because it requires no production-code changes
+    # (no grace= setting/attribute to add) and leaves invoke_shutdown's
+    # own code under test untouched.
+    real_cancel_all = ds._background_tasks.cancel_all
+
+    async def fast_cancel_all(grace=5.0):
+        return await real_cancel_all(grace=0.1)
+
+    monkeypatch.setattr(ds._background_tasks, "cancel_all", fast_cancel_all)
+
+    try:
+        app = ds.app()
+
+        sent = []
+        queue = asyncio.Queue()
+        startup_complete = asyncio.Event()
+
+        async def receive():
+            return await queue.get()
+
+        async def send(message):
+            sent.append(message)
+            if message.get("type") == "lifespan.startup.complete":
+                startup_complete.set()
+
+        task = asyncio.create_task(app({"type": "lifespan"}, receive, send))
+        await queue.put({"type": "lifespan.startup"})
+        await asyncio.wait_for(startup_complete.wait(), timeout=5)
+
+        # Let the stubborn background task actually start running and
+        # reach its CancelledError-suppressing sleep before shutdown
+        # cancels it - a task cancelled before it has ever run its first
+        # step never enters that block at all, so it would finish
+        # cancelling immediately instead of behaving like a straggler.
+        await asyncio.sleep(0.05)
+
+        with caplog.at_level(logging.WARNING, logger="datasette.background_tasks"):
+            await queue.put({"type": "lifespan.shutdown"})
+            await asyncio.wait_for(task, timeout=5)
+
+        assert {"type": "lifespan.shutdown.complete"} in sent
+        assert "stubborn-task" in caplog.text
+        assert ds._closed is True
+    finally:
+        pm.unregister(name="stubborn_task_plugin")
+        # Clean up: the stubborn task ignored cancellation and is still
+        # sleeping past the shrunk grace period; actually cancel and await
+        # it now that the test has made its assertions, so it doesn't leak
+        # past the end of the test.
+        handles = ds._background_tasks.tasks()
+        if handles and handles[0].task and not handles[0].task.done():
+            handles[0].task.cancel()
+            with contextlib.suppress(asyncio.CancelledError):
+                await handles[0].task

tests/test_success_envelope.py

@@ -59,6 +59,7 @@ async def test_success_object_has_ok_true(ds_client, path):
         "/-/rules.json?action=view-instance",
         "/-/check.json?action=view-instance",
         "/-/threads.json",
+        "/-/tasks.json",
     ),
 )
 async def test_permission_debug_success_has_ok_true(ds_envelope, path):

tests/test_tasks_endpoint.py

@@ -0,0 +1,112 @@
+"""
+Tests for the /-/tasks introspection endpoint.
+
+/-/tasks exposes datasette._background_tasks (see tests/test_background_tasks.py
+for the supervisor machinery itself) the same way /-/threads exposes threading
+internals: gated behind the permissions-debug permission, JSON-only.
+"""
+
+import asyncio
+import contextlib
+
+import pytest
+
+from datasette.app import Datasette
+
+
+@pytest.mark.asyncio
+async def test_tasks_requires_permissions_debug():
+    ds = Datasette(memory=True)
+    ds.root_enabled = True
+
+    denied = await ds.client.get("/-/tasks.json")
+    assert denied.status_code == 403
+
+    allowed = await ds.client.get("/-/tasks.json", actor={"id": "root"})
+    assert allowed.status_code == 200
+    data = allowed.json()
+    assert data["ok"] is True
+    assert "tasks" in data
+    assert "launched" in data
+
+
+@pytest.mark.asyncio
+async def test_running_and_crashed_task_states():
+    ds = Datasette(memory=True)
+    ds.root_enabled = True
+
+    async def long_running(datasette):
+        await asyncio.Event().wait()
+
+    async def crashing_task(datasette):
+        raise RuntimeError("kaboom")
+
+    long_handle = ds.add_background_task(long_running, name="long-runner")
+    crash_handle = ds.add_background_task(crashing_task, name="crashing_task")
+
+    await ds.start_background_tasks()
+
+    # Let the crashing_task run to completion and its done-callback (which sets
+    # handle.state = "crashed") actually fire before we read state back out.
+    await asyncio.wait_for(
+        asyncio.gather(crash_handle.task, return_exceptions=True), timeout=5
+    )
+    await asyncio.sleep(0)
+
+    try:
+        response = await ds.client.get("/-/tasks.json", actor={"id": "root"})
+        assert response.status_code == 200
+        data = response.json()
+        assert data["launched"] is True
+
+        by_name = {t["name"]: t for t in data["tasks"]}
+        assert by_name["long-runner"]["state"] == "running"
+        assert by_name["long-runner"]["exception"] is None
+        assert by_name["long-runner"]["started_at"] is not None
+
+        crashed = by_name["crashing_task"]
+        assert crashed["state"] == "crashed"
+        assert crashed["exception"] is not None
+        assert isinstance(crashed["exception"], str)
+        assert "kaboom" in crashed["exception"]
+        assert "RuntimeError" in crashed["exception"]
+    finally:
+        long_handle.cancel()
+        with contextlib.suppress(asyncio.CancelledError):
+            await long_handle.task
+
+
+@pytest.mark.asyncio
+async def test_cold_instance_launched_false_and_task_registered():
+    ds = Datasette(memory=True)
+    ds.root_enabled = True
+
+    async def task(datasette):
+        pass
+
+    ds.add_background_task(task, name="cold-task")
+
+    # ds.client / httpx.ASGITransport routes through the full ASGI app,
+    # including AsgiRunOnFirstRequest - the first-request fallback that
+    # itself launches background tasks (_launch_background_tasks) so hosts
+    # without lifespan support still get supervised tasks running. That
+    # means a plain HTTP request here would launch "cold-task" before we
+    # ever get a response, making the "never launched" state impossible to
+    # observe over HTTP. _suppress_background_tasks is the same switch the
+    # `datasette --get` CLI path sets to stop its one-shot request from
+    # launching long-lived work (see _launch_background_tasks's docstring
+    # in datasette/app.py) - setting it here keeps this one request from
+    # arming the launch, so we can still exercise the real permission-gated
+    # HTTP endpoint while asserting on a genuinely pre-launch snapshot.
+    ds._suppress_background_tasks = True
+
+    response = await ds.client.get("/-/tasks.json", actor={"id": "root"})
+    assert response.status_code == 200
+    data = response.json()
+    assert data["launched"] is False
+    assert len(data["tasks"]) == 1
+    task_data = data["tasks"][0]
+    assert task_data["name"] == "cold-task"
+    assert task_data["state"] == "registered"
+    assert task_data["started_at"] is None
+    assert task_data["exception"] is None