diff --git a/datasette/app.py b/datasette/app.py
index d3626ab2..7499c6f9 100644
--- a/datasette/app.py
+++ b/datasette/app.py
@@ -50,6 +50,7 @@ from .plugins import DEFAULT_PLUGINS, get_plugins, pm
 from .renderer import json_renderer
 from .resources import DatabaseResource, TableResource
 from .telemetry import tracer
+from .telemetry_registry import STARTUP
 from .tokens import TokenInvalid
 from .tracer import AsgiTracer
 from .url_builder import Urls
@@ -788,7 +789,7 @@ class Datasette:
         # A connection warmed lazily later, by a request touching a new
         # database for the first time, nests under that request instead:
         # this span has already ended by then.
-        with tracer.start_as_current_span("datasette.startup"):
+        with tracer.start_as_current_span(STARTUP):
             # Register event classes
             event_classes = []
             for hook in pm.hook.register_events(datasette=self):
diff --git a/datasette/database.py b/datasette/database.py
index 3c14c1c0..22984d6f 100644
--- a/datasette/database.py
+++ b/datasette/database.py
@@ -14,10 +14,32 @@ from pathlib import Path
 
 import sqlite_utils
 from opentelemetry import context as otel_context_api
-from opentelemetry.trace import Link, SpanKind, Status, StatusCode, get_current_span
+from opentelemetry.trace import Link, Status, StatusCode, get_current_span
 
 from .inspect import inspect_hash
 from .telemetry import sql_attribute, sql_operation_name, tracer
+from .telemetry_registry import (
+    DB_COLLECTION_NAME,
+    DB_NAMESPACE,
+    DB_OPERATION_NAME,
+    DB_QUERY,
+    DB_QUERY_EXECUTE,
+    DB_QUERY_TEXT,
+    DB_SYSTEM,
+    DB_WRITE_EXECUTE,
+    DB_WRITE_QUEUE_WAIT,
+    EXECUTEMANY,
+    EXECUTESCRIPT,
+    INTERRUPTED,
+    ISOLATED_CONNECTION,
+    PARAM_COUNT,
+    PARAM_SETS,
+    ROWS_RETURNED,
+    SQL_ERROR_SUPPRESSED,
+    TIME_LIMIT_MS,
+    TRANSACTION,
+    TRUNCATED,
+)
 from .tracer import trace
 from .utils import (
     call_with_supported_arguments,
@@ -268,15 +290,15 @@ class Database:
         with trace(  # noqa: SIM117
             "sql", database=self.name, sql=sql.strip(), params=params
         ):
-            with tracer.start_as_current_span("db.query", kind=SpanKind.CLIENT) as span:
-                span.set_attribute("db.system", "sqlite")
-                span.set_attribute("db.namespace", self.name)
-                span.set_attribute("db.query.text", sql_attribute(sql))
+            with tracer.start_as_current_span(DB_QUERY, kind=DB_QUERY.kind) as span:
+                span.set_attribute(DB_SYSTEM, "sqlite")
+                span.set_attribute(DB_NAMESPACE, self.name)
+                span.set_attribute(DB_QUERY_TEXT, sql_attribute(sql))
                 operation_name = sql_operation_name(sql)
                 if operation_name:
-                    span.set_attribute("db.operation.name", operation_name)
+                    span.set_attribute(DB_OPERATION_NAME, operation_name)
                 if params:
-                    span.set_attribute("datasette.param_count", len(params))
+                    span.set_attribute(PARAM_COUNT, len(params))
                 results = await self.execute_write_fn(
                     _inner, block=block, request=request, transaction=transaction
                 )
@@ -296,11 +318,11 @@ class Database:
             # several semicolon-separated statements, and semantic conventions
             # say the attribute should not be extracted from query text that
             # can hold more than one operation - see sql_operation_name().
-            with tracer.start_as_current_span("db.query", kind=SpanKind.CLIENT) as span:
-                span.set_attribute("db.system", "sqlite")
-                span.set_attribute("db.namespace", self.name)
-                span.set_attribute("db.query.text", sql_attribute(sql))
-                span.set_attribute("datasette.executescript", True)
+            with tracer.start_as_current_span(DB_QUERY, kind=DB_QUERY.kind) as span:
+                span.set_attribute(DB_SYSTEM, "sqlite")
+                span.set_attribute(DB_NAMESPACE, self.name)
+                span.set_attribute(DB_QUERY_TEXT, sql_attribute(sql))
+                span.set_attribute(EXECUTESCRIPT, True)
                 results = await self.execute_write_fn(
                     _inner, block=block, transaction=False, request=request
                 )
@@ -324,22 +346,22 @@ class Database:
         with trace(
             "sql", database=self.name, sql=sql.strip(), executemany=True
         ) as kwargs:
-            with tracer.start_as_current_span("db.query", kind=SpanKind.CLIENT) as span:
-                span.set_attribute("db.system", "sqlite")
-                span.set_attribute("db.namespace", self.name)
-                span.set_attribute("db.query.text", sql_attribute(sql))
-                span.set_attribute("datasette.executemany", True)
+            with tracer.start_as_current_span(DB_QUERY, kind=DB_QUERY.kind) as span:
+                span.set_attribute(DB_SYSTEM, "sqlite")
+                span.set_attribute(DB_NAMESPACE, self.name)
+                span.set_attribute(DB_QUERY_TEXT, sql_attribute(sql))
+                span.set_attribute(EXECUTEMANY, True)
                 # A single statement run with many parameter sets, so unlike
                 # execute_write_script() there is exactly one operation to name.
                 operation_name = sql_operation_name(sql)
                 if operation_name:
-                    span.set_attribute("db.operation.name", operation_name)
+                    span.set_attribute(DB_OPERATION_NAME, operation_name)
                 results, count = await self.execute_write_fn(
                     _inner, block=block, request=request
                 )
                 # count is the number of parameter *sets* consumed by
                 # executemany(), not a row count - executemany returns no rows.
-                span.set_attribute("datasette.param_sets", count)
+                span.set_attribute(PARAM_SETS, count)
             kwargs["count"] = count
         return results
 
@@ -588,7 +610,7 @@ class Database:
                 # waiting in the queue (enqueue -> dequeue), not the near-
                 # zero time spent constructing/ending the span object here.
                 tracer.start_span(
-                    "db.write.queue_wait",
+                    DB_WRITE_QUEUE_WAIT,
                     start_time=task.enqueued_at_ns,
                     **write_span_kwargs,
                 ).end(end_time=time.time_ns())
@@ -599,15 +621,13 @@ class Database:
                 elif task.isolated_connection:
                     try:
                         with tracer.start_as_current_span(
-                            "db.write.execute", **write_span_kwargs
+                            DB_WRITE_EXECUTE, **write_span_kwargs
                         ) as span:
                             span.set_attribute(
-                                "datasette.isolated_connection",
+                                ISOLATED_CONNECTION,
                                 task.isolated_connection,
                             )
-                            span.set_attribute(
-                                "datasette.transaction", task.transaction
-                            )
+                            span.set_attribute(TRANSACTION, task.transaction)
                             isolated_connection = self.connect(write=True)
                             try:
                                 result = task.fn(isolated_connection)
@@ -628,15 +648,13 @@ class Database:
                 else:
                     try:
                         with tracer.start_as_current_span(
-                            "db.write.execute", **write_span_kwargs
+                            DB_WRITE_EXECUTE, **write_span_kwargs
                         ) as span:
                             span.set_attribute(
-                                "datasette.isolated_connection",
+                                ISOLATED_CONNECTION,
                                 task.isolated_connection,
                             )
-                            span.set_attribute(
-                                "datasette.transaction", task.transaction
-                            )
+                            span.set_attribute(TRANSACTION, task.transaction)
                             if task.transaction:
                                 with conn:
                                     conn.execute("BEGIN IMMEDIATE")
@@ -720,7 +738,7 @@ class Database:
             # db.query span in execute(). Without this, facet suggestion marks
             # two spans per text column as failed on every table page.
             with tracer.start_as_current_span(
-                "db.query.execute",
+                DB_QUERY_EXECUTE,
                 record_exception=log_sql_errors,
                 set_status_on_exception=log_sql_errors,
             ):
@@ -764,27 +782,27 @@ class Database:
             # manager's defaults, so that callers passing log_sql_errors=False
             # can be honoured - see the comment on the generic handler below.
             with tracer.start_as_current_span(
-                "db.query",
-                kind=SpanKind.CLIENT,
+                DB_QUERY,
+                kind=DB_QUERY.kind,
                 record_exception=False,
                 set_status_on_exception=False,
             ) as span:
-                span.set_attribute("db.system", "sqlite")
-                span.set_attribute("db.namespace", self.name)
-                span.set_attribute("db.query.text", sql_attribute(sql))
-                span.set_attribute("datasette.time_limit_ms", time_limit_ms)
+                span.set_attribute(DB_SYSTEM, "sqlite")
+                span.set_attribute(DB_NAMESPACE, self.name)
+                span.set_attribute(DB_QUERY_TEXT, sql_attribute(sql))
+                span.set_attribute(TIME_LIMIT_MS, time_limit_ms)
                 operation_name = sql_operation_name(sql)
                 if operation_name:
-                    span.set_attribute("db.operation.name", operation_name)
+                    span.set_attribute(DB_OPERATION_NAME, operation_name)
                 if table:
-                    span.set_attribute("db.collection.name", table)
+                    span.set_attribute(DB_COLLECTION_NAME, table)
                 if params:
-                    span.set_attribute("datasette.param_count", len(params))
+                    span.set_attribute(PARAM_COUNT, len(params))
                 try:
                     results = await self.execute_fn(sql_operation_in_thread)
                 except QueryInterrupted as e:
                     span.set_status(Status(StatusCode.ERROR, str(e)))
-                    span.set_attribute("datasette.interrupted", True)
+                    span.set_attribute(INTERRUPTED, True)
                     span.record_exception(e)
                     raise
                 except Exception as e:
@@ -799,10 +817,10 @@ class Database:
                         span.record_exception(e)
                         span.set_status(Status(StatusCode.ERROR, str(e)))
                     else:
-                        span.set_attribute("datasette.sql_error_suppressed", True)
+                        span.set_attribute(SQL_ERROR_SUPPRESSED, True)
                     raise
-                span.set_attribute("datasette.truncated", results.truncated)
-                span.set_attribute("datasette.rows_returned", len(results.rows))
+                span.set_attribute(TRUNCATED, results.truncated)
+                span.set_attribute(ROWS_RETURNED, len(results.rows))
         return results
 
     @property
diff --git a/datasette/telemetry_registry.py b/datasette/telemetry_registry.py
new file mode 100644
index 00000000..40b32cc5
--- /dev/null
+++ b/datasette/telemetry_registry.py
@@ -0,0 +1,269 @@
+"""
+The single source of truth for every span and span attribute that Datasette
+core emits.
+
+Three things read this module, which is the point of it existing:
+
+1. **The instrumentation itself.** `Attribute` and `SpanName` subclass `str`,
+   so a registry entry *is* the string OpenTelemetry wants. Call sites pass
+   `DB_NAMESPACE` where they used to pass `"db.namespace"` - no wrapper API
+   over the OTel calls, no parallel structure to keep in step, and a typo is
+   now an `ImportError` instead of a silently misnamed attribute.
+
+2. **The documentation.** `docs/telemetry_doc.py` renders the span reference
+   in `docs/internals.rst` from these definitions using cog, and
+   `cog --check` runs in CI - so the docs cannot drift from the code.
+
+3. **A conformance test.** `tests/test_telemetry_registry.py` makes real
+   requests, collects every span and attribute actually emitted, and compares
+   both directions: emitted-but-unregistered catches instrumentation added
+   without documentation, registered-but-never-emitted catches documentation
+   describing something that no longer exists. Neither the type system nor
+   the generated docs can catch that second case.
+"""
+
+from opentelemetry.trace import SpanKind
+
+
+class Attribute(str):
+    """
+    A span attribute key, carrying its own documentation.
+
+    Subclasses `str` so it can be handed straight to `set_attribute()`.
+    """
+
+    __slots__ = ("description", "optional")
+
+    def __new__(cls, name, description, optional=False):
+        self = super().__new__(cls, name)
+        self.description = description
+        self.optional = optional
+        return self
+
+    def __repr__(self):
+        return f"Attribute({str(self)!r})"
+
+
+class SpanName(str):
+    "A span name, carrying its documentation and the attributes it may set."
+
+    __slots__ = ("attributes", "description", "kind", "prefix")
+
+    def __new__(
+        cls, name, description, attributes=(), prefix=False, kind=SpanKind.INTERNAL
+    ):
+        self = super().__new__(cls, name)
+        self.description = description
+        self.attributes = tuple(attributes)
+        # True for a span family whose emitted names carry a variable suffix,
+        # so the conformance test matches by prefix rather than equality.
+        # Nothing sets it yet.
+        self.prefix = prefix
+        # SpanKind.INTERNAL by default - every span Datasette emits describes
+        # its own internal work. db.query is the one exception: it is a real
+        # database call, so semantic conventions (and trace UIs, which key
+        # their database styling off this) expect SpanKind.CLIENT.
+        self.kind = kind
+        return self
+
+    def __repr__(self):
+        return f"SpanName({str(self)!r})"
+
+
+# --- Attributes -----------------------------------------------------------
+#
+# Shared attributes are defined once and referenced by every span that sets
+# them, so "which spans carry db.namespace?" is answerable by grep.
+
+DB_SYSTEM = Attribute("db.system", "Always ``sqlite``.")
+DB_NAMESPACE = Attribute("db.namespace", "Name of the database being queried.")
+DB_QUERY_TEXT = Attribute(
+    "db.query.text",
+    "The SQL, truncated to 2048 characters. Never the parameter values.",
+)
+DB_OPERATION_NAME = Attribute(
+    "db.operation.name",
+    "The statement's leading keyword - ``SELECT``, ``INSERT``, ``CREATE``, and "
+    "so on - matched against a small fixed allowlist. Omitted rather than set "
+    "to an arbitrary value: the allowlist exists because this attribute is a "
+    "candidate dimension for a query-duration metric in a later phase, and "
+    "echoing an unrecognised first token from user-supplied SQL would be an "
+    "unbounded-cardinality hazard. Also omitted for "
+    "``execute_write_script()``, which runs multiple statements - per "
+    "semantic conventions, the operation name should not be extracted from "
+    "query text that can contain more than one operation. Note that a "
+    "statement beginning with a CTE reports ``WITH``, not the operation "
+    "inside it - a substantial share of Datasette's own reads take that "
+    "form. Resolving it further would mean parsing.",
+    optional=True,
+)
+DB_COLLECTION_NAME = Attribute(
+    "db.collection.name",
+    "The primary table, set only where the view already knows it - the table "
+    "and row pages. Omitted for arbitrary ``?sql=`` queries, where determining "
+    "the table would mean parsing the query.",
+    optional=True,
+)
+
+PARAM_COUNT = Attribute(
+    "datasette.param_count",
+    "Number of bound parameters. Recorded instead of the values themselves.",
+    optional=True,
+)
+PARAM_SETS = Attribute(
+    "datasette.param_sets",
+    "Number of parameter sets consumed by ``execute_write_many()``. Not a row "
+    "count - ``executemany()`` returns no rows. The parameter values "
+    "themselves are never recorded: that sequence can hold thousands of rows.",
+    optional=True,
+)
+TIME_LIMIT_MS = Attribute(
+    "datasette.time_limit_ms",
+    "The :ref:`setting_sql_time_limit_ms` value this query ran under. Set on "
+    "reads, which are the queries that time limit applies to.",
+    optional=True,
+)
+ROWS_RETURNED = Attribute(
+    "datasette.rows_returned",
+    "Number of rows a read returned. Set on the read path only, and only when "
+    "the read succeeded.",
+    optional=True,
+)
+TRUNCATED = Attribute(
+    "datasette.truncated",
+    "True if the result was cut short by :ref:`setting_max_returned_rows`.",
+    optional=True,
+)
+INTERRUPTED = Attribute(
+    "datasette.interrupted",
+    "True if the query was cancelled for exceeding the time limit. The span "
+    "status is also set to ``ERROR``.",
+    optional=True,
+)
+SQL_ERROR_SUPPRESSED = Attribute(
+    "datasette.sql_error_suppressed",
+    "True when the query failed but the caller passed ``log_sql_errors=False``, "
+    "meaning it was probing and treats failure as an expected answer. Facet "
+    "suggestion does this against every column.",
+    optional=True,
+)
+EXECUTESCRIPT = Attribute(
+    "datasette.executescript",
+    "True for ``execute_write_script()``, which runs multiple statements.",
+    optional=True,
+)
+EXECUTEMANY = Attribute(
+    "datasette.executemany",
+    "True for ``execute_write_many()``, which runs one statement against many "
+    "parameter sets.",
+    optional=True,
+)
+ISOLATED_CONNECTION = Attribute(
+    "datasette.isolated_connection",
+    "True if the write ran on its own connection rather than the shared write "
+    "connection.",
+)
+TRANSACTION = Attribute(
+    "datasette.transaction",
+    "False for statements such as ``VACUUM`` that cannot run inside a transaction.",
+)
+
+
+# --- Spans ----------------------------------------------------------------
+
+DB_QUERY = SpanName(
+    "db.query",
+    "A SQL operation issued by Datasette, covering the full round trip "
+    "including any time spent queued for a thread.",
+    (
+        DB_SYSTEM,
+        DB_NAMESPACE,
+        DB_QUERY_TEXT,
+        DB_OPERATION_NAME,
+        DB_COLLECTION_NAME,
+        PARAM_COUNT,
+        PARAM_SETS,
+        TIME_LIMIT_MS,
+        ROWS_RETURNED,
+        TRUNCATED,
+        INTERRUPTED,
+        SQL_ERROR_SUPPRESSED,
+        EXECUTESCRIPT,
+        EXECUTEMANY,
+    ),
+    kind=SpanKind.CLIENT,
+)
+
+DB_QUERY_EXECUTE = SpanName(
+    "db.query.execute",
+    "The read executing inside a SQL worker thread. Child of ``db.query``; the "
+    "gap between the two is time spent waiting for a thread.",
+)
+
+DB_WRITE_QUEUE_WAIT = SpanName(
+    "db.write.queue_wait",
+    "Time a write spent waiting in its database's write queue before the write "
+    "thread picked it up. Child of ``db.query`` for a ``block=True`` write, "
+    "where the caller awaits the write and containment is accurate. For a "
+    "``block=False`` write the caller does not await it - the enqueueing "
+    "request *caused* the write without *containing* it, and the write's "
+    "spans can outlive the request's own - so this is a root span instead, "
+    "carrying an OpenTelemetry link back to the enqueueing span rather than "
+    "a parent. A link records causation without asserting containment, which "
+    "is exactly the distinction here.",
+)
+
+DB_WRITE_EXECUTE = SpanName(
+    "db.write.execute",
+    "The write executing on the write thread. Child of ``db.query`` for a "
+    "``block=True`` write; for ``block=False`` a root span with a link back "
+    "to the enqueueing span instead - see ``db.write.queue_wait`` above.",
+    (ISOLATED_CONNECTION, TRANSACTION),
+)
+
+STARTUP = SpanName(
+    "datasette.startup",
+    "``invoke_startup()`` running: ``register_events``, ``register_actions``, "
+    "``register_column_types``, ``prepare_jinja2_environment``, internal-database "
+    "schema catalog refresh (including the ``prepare_connection`` warm-up this "
+    "triggers for each database touched for the first time), saved queries, "
+    "column type config and the ``startup`` hook. Runs once per process, before "
+    "any request exists, so without this span every child it creates would be "
+    "its own orphan root trace. A connection warmed later - lazily, the first "
+    "time a *request* touches a new database or thread - nests under that "
+    "request's own span instead, not under this one, since this span has "
+    "already ended by then.",
+)
+
+SPANS = (
+    DB_QUERY,
+    DB_QUERY_EXECUTE,
+    DB_WRITE_QUEUE_WAIT,
+    DB_WRITE_EXECUTE,
+    STARTUP,
+)
+
+
+def span_for(emitted_name):
+    """
+    Resolve an emitted span name to its registry entry, or None.
+
+    Handles span families whose emitted names carry a suffix that is not
+    knowable in advance - `prefix=True` entries. Phase 1 has none, but the
+    lookup is what the conformance test calls, so it lives here rather than
+    in the test.
+    """
+    for span in SPANS:
+        if span.prefix:
+            if emitted_name.startswith(span):
+                return span
+        elif emitted_name == span:
+            return span
+    return None
+
+
+def attribute_allowed(span, emitted_key):
+    "Whether `emitted_key` is a registered attribute of `span`."
+    if span is None:
+        return False
+    return emitted_key in span.attributes
diff --git a/docs/internals.rst b/docs/internals.rst
index d2bd46ef..b2215d9d 100644
--- a/docs/internals.rst
+++ b/docs/internals.rst
@@ -2313,6 +2313,78 @@ The ``Database`` class also provides properties and methods for introspecting th
           }
         }
 
+.. _internals_telemetry:
+
+OpenTelemetry
+=============
+
+Datasette core depends on `opentelemetry-api <https://pypi.org/project/opentelemetry-api/>`__ only. It never creates a ``TracerProvider``, never configures an exporter and never sets a sampler. With no OpenTelemetry SDK provider installed, every span described below is a no-op ``NonRecordingSpan`` - the overhead is close to zero and nothing is recorded or exported anywhere.
+
+Turning tracing on is entirely an operational decision made outside of Datasette itself: run Datasette under the standard ``opentelemetry-instrument`` agent, or embed Datasette inside a host application that installs its own provider.
+
+Everything Datasette emits carries the instrumentation scope ``datasette``, versioned with the running Datasette version and declaring the `semantic conventions schema <https://opentelemetry.io/docs/specs/otel/schemas/>`__ its attribute names follow.
+
+Span reference
+--------------
+
+Attribute names use the ``datasette.*`` prefix for Datasette-specific data, alongside standard OpenTelemetry attributes such as ``db.system``.
+
+This reference is generated from ``datasette/telemetry_registry.py``, the single source of truth for every span and attribute Datasette emits. A conformance test makes real requests and compares what is actually emitted against that registry in both directions, so nothing here is hand-maintained and nothing can silently drift out of date.
+
+Spans are ``SpanKind.INTERNAL`` unless a kind is listed below. Only ``db.query`` is ``CLIENT``: it is the one span that represents a call to a database rather than Datasette's own work, and trace UIs use the kind to decide whether to render a span as a database call. Its children stay ``INTERNAL`` because they are Datasette's decomposition of that one query - marking them ``CLIENT`` too would make a single query look like several database calls to anything counting by kind.
+
+.. [[[cog
+    from telemetry_doc import spans
+    spans(cog)
+.. ]]]
+
+``db.query``
+    A SQL operation issued by Datasette, covering the full round trip including any time spent queued for a thread.
+
+    Kind: ``CLIENT``.
+
+    Attributes:
+
+    - ``db.system`` - Always ``sqlite``.
+    - ``db.namespace`` - Name of the database being queried.
+    - ``db.query.text`` - The SQL, truncated to 2048 characters. Never the parameter values.
+    - ``db.operation.name`` *(optional)* - The statement's leading keyword - ``SELECT``, ``INSERT``, ``CREATE``, and so on - matched against a small fixed allowlist. Omitted rather than set to an arbitrary value: the allowlist exists because this attribute is a candidate dimension for a query-duration metric in a later phase, and echoing an unrecognised first token from user-supplied SQL would be an unbounded-cardinality hazard. Also omitted for ``execute_write_script()``, which runs multiple statements - per semantic conventions, the operation name should not be extracted from query text that can contain more than one operation. Note that a statement beginning with a CTE reports ``WITH``, not the operation inside it - a substantial share of Datasette's own reads take that form. Resolving it further would mean parsing.
+    - ``db.collection.name`` *(optional)* - The primary table, set only where the view already knows it - the table and row pages. Omitted for arbitrary ``?sql=`` queries, where determining the table would mean parsing the query.
+    - ``datasette.param_count`` *(optional)* - Number of bound parameters. Recorded instead of the values themselves.
+    - ``datasette.param_sets`` *(optional)* - Number of parameter sets consumed by ``execute_write_many()``. Not a row count - ``executemany()`` returns no rows. The parameter values themselves are never recorded: that sequence can hold thousands of rows.
+    - ``datasette.time_limit_ms`` *(optional)* - The :ref:`setting_sql_time_limit_ms` value this query ran under. Set on reads, which are the queries that time limit applies to.
+    - ``datasette.rows_returned`` *(optional)* - Number of rows a read returned. Set on the read path only, and only when the read succeeded.
+    - ``datasette.truncated`` *(optional)* - True if the result was cut short by :ref:`setting_max_returned_rows`.
+    - ``datasette.interrupted`` *(optional)* - True if the query was cancelled for exceeding the time limit. The span status is also set to ``ERROR``.
+    - ``datasette.sql_error_suppressed`` *(optional)* - True when the query failed but the caller passed ``log_sql_errors=False``, meaning it was probing and treats failure as an expected answer. Facet suggestion does this against every column.
+    - ``datasette.executescript`` *(optional)* - True for ``execute_write_script()``, which runs multiple statements.
+    - ``datasette.executemany`` *(optional)* - True for ``execute_write_many()``, which runs one statement against many parameter sets.
+
+``db.query.execute``
+    The read executing inside a SQL worker thread. Child of ``db.query``; the gap between the two is time spent waiting for a thread.
+
+    No attributes.
+
+``db.write.queue_wait``
+    Time a write spent waiting in its database's write queue before the write thread picked it up. Child of ``db.query`` for a ``block=True`` write, where the caller awaits the write and containment is accurate. For a ``block=False`` write the caller does not await it - the enqueueing request *caused* the write without *containing* it, and the write's spans can outlive the request's own - so this is a root span instead, carrying an OpenTelemetry link back to the enqueueing span rather than a parent. A link records causation without asserting containment, which is exactly the distinction here.
+
+    No attributes.
+
+``db.write.execute``
+    The write executing on the write thread. Child of ``db.query`` for a ``block=True`` write; for ``block=False`` a root span with a link back to the enqueueing span instead - see ``db.write.queue_wait`` above.
+
+    Attributes:
+
+    - ``datasette.isolated_connection`` - True if the write ran on its own connection rather than the shared write connection.
+    - ``datasette.transaction`` - False for statements such as ``VACUUM`` that cannot run inside a transaction.
+
+``datasette.startup``
+    ``invoke_startup()`` running: ``register_events``, ``register_actions``, ``register_column_types``, ``prepare_jinja2_environment``, internal-database schema catalog refresh (including the ``prepare_connection`` warm-up this triggers for each database touched for the first time), saved queries, column type config and the ``startup`` hook. Runs once per process, before any request exists, so without this span every child it creates would be its own orphan root trace. A connection warmed later - lazily, the first time a *request* touches a new database or thread - nests under that request's own span instead, not under this one, since this span has already ended by then.
+
+    No attributes.
+
+.. [[[end]]]
+
 .. _internals_csrf:
 
 CSRF protection
diff --git a/docs/telemetry_doc.py b/docs/telemetry_doc.py
new file mode 100644
index 00000000..3551ef0e
--- /dev/null
+++ b/docs/telemetry_doc.py
@@ -0,0 +1,37 @@
+"""
+Render the span reference in ``internals.rst`` from
+``datasette/telemetry_registry.py``.
+
+Driven by cog, and ``cog --check docs/*.rst`` runs in CI - so adding a span
+without documenting it, or documenting one that no longer exists, is a build
+failure rather than something a reader discovers later.
+"""
+
+
+def _attribute_lines(cog, attributes):
+    if not attributes:
+        cog.out("    No attributes.\n\n")
+        return
+    cog.out("    Attributes:\n\n")
+    for attribute in attributes:
+        suffix = " *(optional)*" if attribute.optional else ""
+        cog.out(f"    - ``{attribute}``{suffix} - {attribute.description}\n")
+    cog.out("\n")
+
+
+def spans(cog):
+    from opentelemetry.trace import SpanKind
+
+    from datasette.telemetry_registry import SPANS
+
+    cog.out("\n")
+    for span in SPANS:
+        title = f"{span}*" if span.prefix else str(span)
+        cog.out(f"``{title}``\n")
+        cog.out(f"    {span.description}\n\n")
+        # INTERNAL is the default and the overwhelming majority of spans -
+        # printing it on every one would be noise. Only the exceptional case,
+        # a real database call, is worth calling out.
+        if span.kind != SpanKind.INTERNAL:
+            cog.out(f"    Kind: ``{span.kind.name}``.\n\n")
+        _attribute_lines(cog, span.attributes)
diff --git a/tests/test_telemetry_registry.py b/tests/test_telemetry_registry.py
new file mode 100644
index 00000000..335eb946
--- /dev/null
+++ b/tests/test_telemetry_registry.py
@@ -0,0 +1,309 @@
+"""
+Two-way conformance between `datasette/telemetry_registry.py` and what
+Datasette actually emits.
+
+This is the test that makes the generated documentation trustworthy. cog
+guarantees the docs match the registry; this guarantees the registry matches
+the code. Without it, both could agree with each other and be wrong.
+
+It checks both directions, and the second one is the one nothing else catches:
+
+- **emitted but not registered** - instrumentation was added without
+  documenting it, so the reference page silently omits it.
+- **registered but never emitted** - the reference page describes a span or
+  attribute that no longer exists, which is worse than omitting it, because a
+  reader will build a dashboard on it.
+
+Both of those directions compare the code against the registry. Neither can
+catch a *rename*, because the call sites now take their names from the
+registry - move `DB_NAMESPACE` to `"db.namespace2"` and code and registry
+still agree with each other, while every existing dashboard breaks. So the
+literal names live here too, spelled out, and are asserted against both the
+registry and the wire. That is the one comparison in this file that is not
+made against a value derived from the registry itself.
+"""
+
+import itertools
+
+import pytest
+import pytest_asyncio
+
+pytest.importorskip("opentelemetry.sdk")
+
+from datasette import telemetry_registry as reg
+from datasette.app import Datasette
+from datasette.database import QueryInterrupted
+from datasette.utils.sqlite import sqlite3
+
+# The names as they appear on the wire, written out rather than read from the
+# registry. If a change to the registry makes one of these fail, that change
+# is renaming something a user's dashboards and saved queries depend on -
+# which is a decision to take deliberately, here, not a line to re-derive.
+EXPECTED_ATTRIBUTES = {
+    "db.query": {
+        "db.system",
+        "db.namespace",
+        "db.query.text",
+        "db.operation.name",
+        "db.collection.name",
+        "datasette.param_count",
+        "datasette.param_sets",
+        "datasette.time_limit_ms",
+        "datasette.rows_returned",
+        "datasette.truncated",
+        "datasette.interrupted",
+        "datasette.sql_error_suppressed",
+        "datasette.executescript",
+        "datasette.executemany",
+    },
+    "db.query.execute": set(),
+    "db.write.queue_wait": set(),
+    "db.write.execute": {
+        "datasette.isolated_connection",
+        "datasette.transaction",
+    },
+    "datasette.startup": set(),
+}
+EXPECTED_SPANS = set(EXPECTED_ATTRIBUTES)
+
+# Named in-memory databases are shared-cache, so two Datasette instances using
+# the same name share one SQLite database - and the second `create table`
+# fails. Every workload below therefore gets its own name.
+_names = itertools.count()
+
+
+def _unique(prefix):
+    return f"{prefix}{next(_names)}"
+
+
+async def exercise():
+    """
+    Drive enough of Datasette to emit every span and attribute the registry
+    claims exists.
+
+    Each call is here because it is the only thing that produces some span or
+    attribute - see the comments. If you add instrumentation on a path this
+    does not reach, add the path rather than loosening the assertions.
+
+    Returns the instance so the caller can close it; startup happens inside
+    so that the `datasette.startup` span lands in the collected set.
+    """
+    name = _unique("registry")
+    ds = Datasette(memory=True)
+    ds.add_memory_database(name)
+    # datasette.startup - and the internal catalog work nested under it
+    await ds.invoke_startup()
+    db = ds.get_database(name)
+
+    # Writes: db.write.queue_wait, db.write.execute, db.query
+    await db.execute_write("create table t (id integer primary key, v text)")
+    # datasette.executemany, datasette.param_sets
+    await db.execute_write_many(
+        "insert into t (id, v) values (?, ?)", [[i, f"v{i}"] for i in range(30)]
+    )
+    # datasette.executescript
+    await db.execute_write_script("create table t2 (id integer); drop table t2;")
+    # datasette.transaction=False - VACUUM cannot run inside a transaction
+    await db.execute_write("vacuum", transaction=False)
+    # datasette.isolated_connection=True
+    await db.execute_isolated_fn(lambda conn: conn.execute("select 1").fetchone())
+
+    # Reads: db.query.execute, datasette.rows_returned, datasette.truncated,
+    # datasette.param_count, datasette.time_limit_ms
+    await db.execute("select * from t where id > :n", {"n": 5})
+    await db.execute("select * from t", truncate=True)
+
+    # datasette.sql_error_suppressed - the caller is probing and treats
+    # failure as an expected answer
+    with pytest.raises(sqlite3.OperationalError):
+        await db.execute("select nope from t", log_sql_errors=False)
+
+    # datasette.interrupted - only ever set when a query exceeds its time
+    # limit, so the workload has to force one rather than exempt it. An
+    # unbounded recursive CTE cannot finish, so 1ms is always exceeded.
+    with pytest.raises(QueryInterrupted):
+        await db.execute(
+            "with recursive c(x) as (select 0 union all select x+1 from c) "
+            "select * from c",
+            custom_time_limit=1,
+        )
+
+    # db.collection.name - set only by views that already know their table
+    assert (await ds.client.get(f"/{name}/t?_facet=v")).status_code == 200
+    assert (await ds.client.get(f"/{name}/t/1.json")).status_code == 200
+    return ds
+
+
+@pytest_asyncio.fixture
+async def emitted(otel_spans):
+    "Every span name and (span name, attribute key) pair a broad workload emits."
+    # otel_spans has already cleared the exporter, and nothing is cleared
+    # after this point: the workload's own startup emits datasette.startup.
+    ds = await exercise()
+    spans = otel_spans.get_finished_spans()
+    assert spans, "no spans captured - the fixture is not exercising anything"
+    names = set()
+    pairs = set()
+    for span in spans:
+        # str() because span.name is the registry's SpanName instance, and a
+        # set of those would compare equal to literals but read confusingly
+        # in a failure message.
+        names.add(str(span.name))
+        for key in span.attributes or {}:
+            pairs.add((str(span.name), str(key)))
+    ds.close()
+    return {"names": names, "pairs": pairs}
+
+
+def _keys_by_span(pairs):
+    by_span = {}
+    for span_name, key in pairs:
+        by_span.setdefault(span_name, set()).add(key)
+    return by_span
+
+
+@pytest.mark.asyncio
+async def test_workload_emits_exactly_the_expected_names(emitted):
+    """
+    The wire format, pinned to literals.
+
+    Not derived from the registry, so this is what catches a rename that the
+    registry and the call sites make together.
+    """
+    assert emitted["names"] == EXPECTED_SPANS
+    by_span = _keys_by_span(emitted["pairs"])
+    assert {name: by_span.get(name, set()) for name in emitted["names"]} == (
+        EXPECTED_ATTRIBUTES
+    )
+
+
+def test_registry_matches_the_expected_names():
+    "The other half of the rename check: the registry against the same literals."
+    assert {str(span) for span in reg.SPANS} == EXPECTED_SPANS
+    for span in reg.SPANS:
+        assert {str(attribute) for attribute in span.attributes} == EXPECTED_ATTRIBUTES[
+            str(span)
+        ], f"{span} attributes have drifted"
+
+
+@pytest.mark.asyncio
+async def test_every_emitted_span_is_registered(emitted):
+    "A span added without a registry entry would be missing from the docs."
+    unregistered = sorted(
+        name for name in emitted["names"] if reg.span_for(name) is None
+    )
+    assert (
+        not unregistered
+    ), f"these spans are emitted but not in telemetry_registry.SPANS: {unregistered}"
+
+
+@pytest.mark.asyncio
+async def test_every_emitted_attribute_is_registered(emitted):
+    "An attribute added without a registry entry would be missing from the docs."
+    unregistered = sorted(
+        f"{span_name} -> {key}"
+        for span_name, key in emitted["pairs"]
+        if not reg.attribute_allowed(reg.span_for(span_name), key)
+    )
+    assert (
+        not unregistered
+    ), "these span attributes are emitted but not registered: " + ", ".join(
+        unregistered
+    )
+
+
+@pytest.mark.asyncio
+async def test_every_registered_span_is_emitted(emitted):
+    """
+    The direction nothing else catches: the docs must not describe a span that
+    no longer exists.
+    """
+    missing = sorted(
+        str(span)
+        for span in reg.SPANS
+        if not any(reg.span_for(name) is span for name in emitted["names"])
+    )
+    assert not missing, (
+        f"these spans are documented but never emitted by the workload: {missing}. "
+        "Either the instrumentation was removed, or exercise() no longer reaches it."
+    )
+
+
+@pytest.mark.asyncio
+async def test_every_registered_attribute_is_emitted(emitted):
+    """
+    Every registered attribute, optional or not, must actually be set at least
+    once by the workload.
+
+    `optional` describes whether a reader should expect it on every span, not
+    whether the code still sets it - so an attribute deleted from the code but
+    left in the docs has to fail here even when it is marked optional. If a
+    new attribute only appears in some rare case, extend exercise() to reach
+    that case.
+    """
+    by_span = _keys_by_span(emitted["pairs"])
+    missing = []
+    for span in reg.SPANS:
+        emitted_keys = by_span.get(str(span), set())
+        for attribute in span.attributes:
+            if attribute not in emitted_keys:
+                missing.append(f"{span} -> {attribute}")
+    assert not missing, (
+        "these attributes are documented but never emitted by the workload: "
+        + ", ".join(sorted(missing))
+    )
+
+
+def test_registry_has_no_duplicate_names():
+    assert len(set(reg.SPANS)) == len(reg.SPANS)
+    for span in reg.SPANS:
+        assert len(set(span.attributes)) == len(
+            span.attributes
+        ), f"{span} lists an attribute twice"
+
+
+def test_registry_entries_are_documented():
+    "Every entry carries a description - the docs are generated from these."
+    for span in reg.SPANS:
+        assert span.description.strip(), f"{span} has no description"
+        for attribute in span.attributes:
+            assert attribute.description.strip(), f"{span} -> {attribute} has none"
+
+
+def test_registry_entries_are_usable_as_plain_strings():
+    "The str subclassing is the whole reason call sites need no wrapper API."
+    assert isinstance(reg.DB_QUERY, str)
+    assert isinstance(reg.DB_NAMESPACE, str)
+    assert reg.DB_QUERY == "db.query"
+    assert reg.DB_NAMESPACE == "db.namespace"
+    assert f"{reg.DB_QUERY}.execute" == "db.query.execute"
+
+
+def test_span_and_attribute_lookup():
+    assert reg.span_for("db.query") is reg.DB_QUERY
+    assert reg.span_for("datasette.startup") is reg.STARTUP
+    assert reg.span_for("not.a.datasette.span") is None
+    assert reg.attribute_allowed(reg.DB_QUERY, "db.namespace")
+    assert not reg.attribute_allowed(reg.DB_QUERY, "db.namespace.extra")
+    assert not reg.attribute_allowed(reg.DB_QUERY, "datasette.isolated_connection")
+    assert not reg.attribute_allowed(None, "db.namespace")
+
+
+def test_prefix_span_lookup():
+    """
+    `prefix=True` matching, exercised directly.
+
+    Phase 1 registers no prefix spans, so without this the branch in
+    `span_for()` would be untested code that the conformance tests silently
+    never reach.
+    """
+    hook = reg.SpanName("datasette.hook.", "A hypothetical span family", prefix=True)
+    original = reg.SPANS
+    reg.SPANS = original + (hook,)
+    try:
+        assert reg.span_for("datasette.hook.render_cell") is hook
+        assert reg.span_for("datasette.hook.anything") is hook
+        assert reg.span_for("datasette.hookish") is None
+        assert reg.span_for("db.query") is reg.DB_QUERY
+    finally:
+        reg.SPANS = original