diff --git a/datasette/app.py b/datasette/app.py
index 56f9954d..1ed5ce4f 100644
--- a/datasette/app.py
+++ b/datasette/app.py
@@ -52,8 +52,10 @@ from .telemetry import (
     TelemetryMiddleware,
     _in_datasette_client,
     clamp_http_method,
+    register_datasette,
     request_span,
     tracer,
+    unregister_datasette,
 )
 from .telemetry_registry import HTTP_ROUTE, STARTUP
 from .tokens import TokenInvalid
@@ -646,6 +648,10 @@ class Datasette:
         self.root_enabled = False
         self.default_deny = default_deny
         self.client = DatasetteClient(self)
+        # Last, so that the observable-gauge callbacks - which may fire on the
+        # SDK's collection thread the instant this returns - never see a
+        # half-built instance.
+        register_datasette(self)
 
     async def apply_metadata_json(self):
         # Apply any metadata entries from metadata.json to the internal tables
@@ -985,6 +991,10 @@ class Datasette:
         if self._closed:
             return
         self._closed = True
+        # Stop reporting gauges before tearing anything down, so a collection
+        # cycle landing mid-close cannot observe a half-closed instance. The
+        # WeakSet would drop it eventually anyway; this makes it immediate.
+        unregister_datasette(self)
         first_exception = None
         dbs = list(self.databases.values()) + [self._internal_database]
         for db in dbs:
diff --git a/datasette/database.py b/datasette/database.py
index 22c4b5f2..3b4d64dd 100644
--- a/datasette/database.py
+++ b/datasette/database.py
@@ -17,7 +17,15 @@ from opentelemetry import context as otel_context_api
 from opentelemetry.trace import Link, Status, StatusCode, get_current_span
 
 from .inspect import inspect_hash
-from .telemetry import callback_name, sql_attribute, sql_operation_name, tracer
+from .telemetry import (
+    callback_name,
+    record_operation_duration,
+    record_query_interrupted,
+    record_write_queue_wait,
+    sql_attribute,
+    sql_operation_name,
+    tracer,
+)
 from .telemetry_registry import (
     CALLBACK,
     DB_COLLECTION_NAME,
@@ -313,9 +321,10 @@ class Database:
                     span.set_attribute(DB_OPERATION_NAME, operation_name)
                 if params:
                     span.set_attribute(PARAM_COUNT, len(params))
-                results = await self._execute_write_fn(
-                    _inner, block=block, request=request, transaction=transaction
-                )
+                with record_operation_duration(self.name, "write"):
+                    results = await self._execute_write_fn(
+                        _inner, block=block, request=request, transaction=transaction
+                    )
         return results
 
     async def execute_write_script(self, sql, block=True, request=None):
@@ -337,9 +346,10 @@ class Database:
                 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
-                )
+                with record_operation_duration(self.name, "write"):
+                    results = await self._execute_write_fn(
+                        _inner, block=block, transaction=False, request=request
+                    )
         return results
 
     async def execute_write_many(self, sql, params_seq, block=True, request=None):
@@ -370,9 +380,10 @@ class Database:
                 operation_name = sql_operation_name(sql)
                 if operation_name:
                     span.set_attribute(DB_OPERATION_NAME, operation_name)
-                results, count = await self._execute_write_fn(
-                    _inner, block=block, request=request
-                )
+                with record_operation_duration(self.name, "write"):
+                    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(PARAM_SETS, count)
@@ -662,11 +673,15 @@ class Database:
                 # this span's duration is the time the task actually spent
                 # waiting in the queue (enqueue -> dequeue), not the near-
                 # zero time spent constructing/ending the span object here.
+                dequeued_at_ns = time.time_ns()
                 tracer.start_span(
                     DB_WRITE_QUEUE_WAIT,
                     start_time=task.enqueued_at_ns,
                     **write_span_kwargs,
-                ).end(end_time=time.time_ns())
+                ).end(end_time=dequeued_at_ns)
+                record_write_queue_wait(
+                    self.name, dequeued_at_ns - task.enqueued_at_ns
+                )
                 if conn_exception is not None:
                     # fn never runs in this branch, so there is nothing to
                     # wrap in a db.write.execute span.
@@ -924,7 +939,8 @@ class Database:
                 if params:
                     span.set_attribute(PARAM_COUNT, len(params))
                 try:
-                    results = await self._execute_fn(sql_operation_in_thread)
+                    with record_operation_duration(self.name, "read"):
+                        results = await self._execute_fn(sql_operation_in_thread)
                 except QueryInterrupted as e:
                     # datasette.interrupted is set either way - it is the
                     # signal worth having. Only the ERROR status is
@@ -933,6 +949,14 @@ class Database:
                     if not timeout_expected:
                         span.set_status(Status(StatusCode.ERROR, str(e)))
                         span.record_exception(e)
+                        # A counter rather than only a span, because this is the
+                        # one thing an operator wants a rate and an alert on, and
+                        # spans under a 1% sampler cannot provide either. An
+                        # expected timeout - a caller that opted into a shorter
+                        # budget, like facet suggestion - is not counted, for the
+                        # same reason it is not a span error: it fires routinely
+                        # by design and would drown the signal this exists for.
+                        record_query_interrupted(self.name)
                     raise
                 except Exception as e:
                     # log_sql_errors=False means the caller is probing and
diff --git a/datasette/telemetry.py b/datasette/telemetry.py
index 19462585..bf4d8e65 100644
--- a/datasette/telemetry.py
+++ b/datasette/telemetry.py
@@ -2,27 +2,36 @@
 OpenTelemetry integration for Datasette core.
 
 Core depends on `opentelemetry-api` only. It never creates a
-`TracerProvider`, never configures an exporter, and never touches
-sampling - that is the responsibility of whoever is running Datasette
-(an `opentelemetry-instrument` agent, a future plugin, or a test
+`TracerProvider` or a `MeterProvider`, never configures an exporter, and
+never touches sampling - that is the responsibility of whoever is running
+Datasette (an `opentelemetry-instrument` agent, a future plugin, or a test
 harness).
 
 With no provider installed every span produced here is a
 `NonRecordingSpan`. That is not free - a table page emits ~100 spans -
 but end-to-end page benchmarks put the overhead below their own
 run-to-run variation. Installing an SDK provider is what costs
-something measurable.
+something measurable. Every metric instrument is likewise a no-op
+without a provider, and the observable-gauge callbacks are never
+invoked at all.
 """
 
 import contextvars
 import re
+import threading
+import time
+import weakref
+from contextlib import contextmanager
 
+from opentelemetry import metrics as otel_metrics
 from opentelemetry import trace as otel_trace
 from opentelemetry.propagate import extract
 from opentelemetry.propagators.textmap import Getter
 from opentelemetry.trace import SpanKind, Status, StatusCode
 
 from .telemetry_registry import (
+    DB_NAMESPACE,
+    DB_SYSTEM,
     ERROR_TYPE,
     HTTP_REQUEST_METHOD,
     HTTP_RESPONSE_STATUS_CODE,
@@ -64,6 +73,7 @@ _in_datasette_client = contextvars.ContextVar("in_datasette_client", default=Fal
 SCHEMA_URL = "https://opentelemetry.io/schemas/1.29.0"
 
 tracer = otel_trace.get_tracer("datasette", __version__, schema_url=SCHEMA_URL)
+meter = otel_metrics.get_meter("datasette", __version__, schema_url=SCHEMA_URL)
 
 MAX_SQL_LENGTH = 2048
 
@@ -359,3 +369,249 @@ class TelemetryMiddleware:
                     if status >= 500 and not escaped:
                         span.set_status(Status(StatusCode.ERROR))
                         span.set_attribute(ERROR_TYPE, str(status))
+
+
+# --- Metrics --------------------------------------------------------------
+#
+# Spans answer "what happened during this request". They cannot answer "am I
+# saturating my 3 SQL threads right now", because that is a gauge: a level
+# sampled at collection time, not an event with a duration. It is also the
+# single most useful operational question about a Datasette deployment, since
+# num_sql_threads defaults to 3 and every read query in the process competes
+# for those threads.
+#
+# Two shapes are used here:
+#
+#   Observable gauges - a callback the SDK invokes on its own collection
+#   cycle. Nothing is computed unless something is collecting, so the default
+#   no-provider install pays literally nothing for them.
+#
+#   Synchronous histograms/counters - recorded inline on the query path. These
+#   survive trace sampling, which spans do not: an operator sampling 1% of
+#   traces still gets 100% of the latency distribution and the interrupted
+#   count.
+#
+# Note a real difference from tracing: `_ProxyMeter` and its instruments
+# forward to a provider installed *after* they were created, whereas
+# `ProxyTracer` permanently caches the concrete tracer it first resolves. So
+# module-level instruments here are safe, and tests do not need a provider
+# installed before this module is imported.
+
+
+def _duration_attributes(database_name, operation):
+    return {
+        DB_SYSTEM: "sqlite",
+        DB_NAMESPACE: database_name,
+        "datasette.operation": operation,
+    }
+
+
+sql_operation_duration = meter.create_histogram(
+    "db.client.operation.duration",
+    unit="s",
+    description="Duration of a SQL operation issued by Datasette",
+)
+
+write_queue_wait = meter.create_histogram(
+    "datasette.write.queue_wait",
+    unit="s",
+    description=(
+        "Time a write spent queued behind the single write thread for its database"
+    ),
+)
+
+queries_interrupted = meter.create_counter(
+    "datasette.sql.queries.interrupted",
+    unit="{query}",
+    description=(
+        "Queries cancelled for exceeding sql_time_limit_ms. Not derivable from "
+        "spans under sampling, and the signal that a time limit is too tight"
+    ),
+)
+
+
+@contextmanager
+def record_operation_duration(database_name, operation):
+    """
+    Record `db.client.operation.duration` for one SQL operation.
+
+    `error.type` is set from the exception class on failure, per semconv, so a
+    latency distribution can be split by success and failure. For a
+    `block=False` write this measures the enqueue, not the write - the same
+    caveat that applies to the surrounding span.
+    """
+    attributes = _duration_attributes(database_name, operation)
+    started = time.perf_counter()
+    try:
+        yield
+    except BaseException as exception:
+        attributes[ERROR_TYPE] = type(exception).__qualname__
+        raise
+    finally:
+        sql_operation_duration.record(time.perf_counter() - started, attributes)
+
+
+def record_write_queue_wait(database_name, waited_ns):
+    write_queue_wait.record(waited_ns / 1e9, {DB_NAMESPACE: database_name})
+
+
+def record_query_interrupted(database_name):
+    queries_interrupted.add(1, {DB_NAMESPACE: database_name})
+
+
+# Live Datasette instances, weakly held so that instrumenting an instance
+# never keeps it alive. Guarded by a lock because the gauge callbacks run on
+# the SDK's collection thread while the event loop may be building or closing
+# a Datasette.
+#
+# Known limitation: the pool gauges below carry no attribute identifying which
+# Datasette produced them, so if a single process runs more than one instance
+# their observations collide and last-one-wins. Production runs one instance
+# per process; adding an instance id to make the test suite's hundreds of
+# instances distinguishable would mean unbounded attribute cardinality in
+# exchange for fixing a case that does not occur in production.
+_live_datasettes = weakref.WeakSet()
+_live_datasettes_lock = threading.Lock()
+
+
+def register_datasette(ds):
+    "Start reporting pool/queue gauges for this Datasette instance."
+    with _live_datasettes_lock:
+        _live_datasettes.add(ds)
+
+
+def unregister_datasette(ds):
+    "Stop reporting gauges for an instance that has been closed."
+    with _live_datasettes_lock:
+        _live_datasettes.discard(ds)
+
+
+def _live_instances():
+    with _live_datasettes_lock:
+        return list(_live_datasettes)
+
+
+def _databases_of(ds):
+    """
+    Every Database attached to an instance, including the internal database.
+
+    The internal database is deliberately included: permission checks run SQL
+    against it on essentially every request, so its queue depth and connection
+    count are as operationally interesting as any user database's.
+    """
+    databases = list(ds.databases.values())
+    internal = getattr(ds, "_internal_database", None)
+    if internal is not None:
+        databases.append(internal)
+    return databases
+
+
+# Each callback is a plain generator function so it can be unit-tested
+# directly, without standing up an SDK provider and a metric reader.
+
+
+def observe_sql_thread_limit(options=None):
+    "Size of the shared read-query thread pool (the num_sql_threads setting)."
+    for ds in _live_instances():
+        if ds.executor is None:
+            # num_sql_threads=0 - queries run on the event loop, no pool.
+            continue
+        yield otel_metrics.Observation(ds.setting("num_sql_threads"), {})
+
+
+def observe_sql_thread_queue_depth(options=None):
+    """
+    Read queries waiting for a free thread in the shared pool.
+
+    This is the saturation signal: sustained above zero means requests are
+    queueing on num_sql_threads. `_work_queue` is a private attribute of
+    ThreadPoolExecutor, so its absence is tolerated rather than fatal - a
+    missing gauge is much better than a crashed collection cycle.
+    """
+    for ds in _live_instances():
+        if ds.executor is None:
+            continue
+        work_queue = getattr(ds.executor, "_work_queue", None)
+        if work_queue is None:
+            continue
+        yield otel_metrics.Observation(work_queue.qsize(), {})
+
+
+def observe_pending_queries(options=None):
+    """
+    Read queries submitted to the pool and not yet finished, per database.
+
+    Summed across databases and compared against the thread limit, this is the
+    utilisation half of the saturation picture. `len()` is deliberately taken
+    without `_pending_execute_futures_lock`: it is atomic, and taking a lock
+    held on the request path from the collection thread would let telemetry
+    add latency to queries.
+    """
+    for ds in _live_instances():
+        for db in _databases_of(ds):
+            yield otel_metrics.Observation(
+                len(db._pending_execute_futures), {DB_NAMESPACE: db.name}
+            )
+
+
+def observe_write_queue_depth(options=None):
+    """
+    Writes queued behind the single write thread, per database.
+
+    Every database serialises its writes through one thread, so this is
+    unbounded backpressure that no amount of num_sql_threads will relieve.
+    """
+    for ds in _live_instances():
+        for db in _databases_of(ds):
+            write_queue = db._write_queue
+            if write_queue is None:
+                # No write has ever been queued for this database.
+                continue
+            yield otel_metrics.Observation(
+                write_queue.qsize(), {DB_NAMESPACE: db.name}
+            )
+
+
+def observe_open_connections(options=None):
+    "Open SQLite file connections tracked for closing, per database."
+    for ds in _live_instances():
+        for db in _databases_of(ds):
+            yield otel_metrics.Observation(
+                len(db._all_file_connections), {DB_NAMESPACE: db.name}
+            )
+
+
+sql_thread_limit_gauge = meter.create_observable_gauge(
+    "datasette.sql.threads.limit",
+    callbacks=[observe_sql_thread_limit],
+    unit="{thread}",
+    description="Maximum concurrent read queries (the num_sql_threads setting)",
+)
+
+sql_thread_queue_depth_gauge = meter.create_observable_gauge(
+    "datasette.sql.threads.queue_depth",
+    callbacks=[observe_sql_thread_queue_depth],
+    unit="{query}",
+    description="Read queries waiting for a free thread in the shared SQL pool",
+)
+
+pending_queries_gauge = meter.create_observable_gauge(
+    "datasette.sql.queries.pending",
+    callbacks=[observe_pending_queries],
+    unit="{query}",
+    description="Read queries submitted to the pool and not yet complete",
+)
+
+write_queue_depth_gauge = meter.create_observable_gauge(
+    "datasette.write.queue_depth",
+    callbacks=[observe_write_queue_depth],
+    unit="{write}",
+    description="Writes queued behind a database's single write thread",
+)
+
+open_connections_gauge = meter.create_observable_gauge(
+    "datasette.connections.open",
+    callbacks=[observe_open_connections],
+    unit="{connection}",
+    description="Open SQLite file connections tracked for closing",
+)
diff --git a/docs/changelog.rst b/docs/changelog.rst
index 9d1e6ad6..379a0279 100644
--- a/docs/changelog.rst
+++ b/docs/changelog.rst
@@ -13,6 +13,7 @@ Unreleased
 - Datasette's database layer now emits `OpenTelemetry <https://opentelemetry.io/>`__ spans: one per query, covering the full round trip including time spent waiting for a SQL worker thread, plus separate child spans for the execution itself and for time spent in the write queue. Callback-style calls - :ref:`db.execute_fn() <database_execute_fn>`, :ref:`db.execute_write_fn() <database_execute_write_fn>` and ``db.execute_isolated_fn()``, the documented way for plugins to run arbitrary SQL - are covered too, carrying ``datasette.callback`` in place of the SQL text. Datasette core depends on ``opentelemetry-api`` only and never installs an SDK provider, an exporter or a sampler, so there is no effect and no measurable overhead unless tracing is switched on externally - normally with the standard ``opentelemetry-instrument`` agent. See :ref:`internals_telemetry`. (:issue:`1730`)
 - :ref:`db.execute(sql, ..., table=None) <database_execute>` has a new optional ``table=`` parameter, naming the table a query is about so it can be recorded on that query's OpenTelemetry span. It has no effect on query execution, and Datasette never derives it from the SQL. (:issue:`1730`)
 - Every HTTP request now gets an OpenTelemetry ``SERVER`` span, named after the request method and matched route, carrying ``http.route``, the response status and W3C trace context extracted from inbound headers - so every database span has a request to belong to, and Datasette joins distributed traces started by a proxy or calling service. The query string is never recorded. See :ref:`internals_telemetry_requests`. (:issue:`1730`)
+- Datasette core now also emits OpenTelemetry **metrics** covering SQL thread pool saturation, per-database write queue depth, open connections, query latency and time-limit interruptions. These answer operational questions that spans structurally cannot - "am I saturating my :ref:`setting_num_sql_threads` threads?" is a level, not an event - and they survive trace sampling. As with spans, core installs no ``MeterProvider``, so there is no cost unless metrics are collected externally. See :ref:`internals_telemetry`. (:issue:`1730`)
 
 Nothing is removed by the OpenTelemetry work: the ``?_trace=1`` query string parameter, the ``trace_debug`` setting and the :ref:`internals_tracer` module all continue to work as before.
 
diff --git a/tests/conftest.py b/tests/conftest.py
index 4ea97d6b..9b005090 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -114,6 +114,106 @@ def otel_spans():
     yield _otel_span_exporter
 
 
+_otel_metric_reader = None
+
+
+@pytest.fixture(scope="session", autouse=True)
+def _otel_meter_provider():
+    """
+    Install a real OTel SDK MeterProvider + InMemoryMetricReader once per
+    process.
+
+    Unlike the tracer, ordering is not load-bearing here: `_ProxyMeter` and
+    the `_ProxyInstrument`s it hands out forward to a provider installed
+    *after* they were created, whereas `ProxyTracer` permanently caches the
+    first concrete tracer it resolves. This fixture is still session-scoped
+    and autouse for symmetry, and so that a single reader collects for the
+    whole run.
+
+    DELTA temporality is chosen for counters and histograms so that each
+    collection reports only what happened since the previous one. With the
+    SDK default of CUMULATIVE, every metrics test would see every query run
+    by every earlier test in the session.
+    """
+    global _otel_metric_reader
+    try:
+        from opentelemetry import metrics as otel_metrics
+        from opentelemetry.sdk.metrics import Counter, Histogram, MeterProvider
+        from opentelemetry.sdk.metrics.export import (
+            AggregationTemporality,
+            InMemoryMetricReader,
+        )
+    except ImportError:
+        return
+    reader = InMemoryMetricReader(
+        preferred_temporality={
+            Counter: AggregationTemporality.DELTA,
+            Histogram: AggregationTemporality.DELTA,
+        }
+    )
+    otel_metrics.set_meter_provider(MeterProvider(metric_readers=[reader]))
+    _otel_metric_reader = reader
+
+
+class MetricsCollector:
+    """
+    Thin reader over an `InMemoryMetricReader`.
+
+    `collect()` runs a collection cycle - which is what invokes the observable
+    gauge callbacks - and snapshots the result. Queries then run against that
+    snapshot rather than re-collecting, so a test that inspects several
+    metrics sees one consistent moment and does not drain delta state twice.
+    """
+
+    def __init__(self, reader):
+        self.reader = reader
+        self.snapshot = {}
+
+    def collect(self):
+        self.snapshot = {}
+        data = self.reader.get_metrics_data()
+        if data is None:
+            return self.snapshot
+        for resource_metrics in data.resource_metrics:
+            for scope_metrics in resource_metrics.scope_metrics:
+                for metric in scope_metrics.metrics:
+                    self.snapshot.setdefault(metric.name, []).extend(
+                        metric.data.data_points
+                    )
+        return self.snapshot
+
+    def points(self, name, attributes=None):
+        "Data points for `name` whose attributes are a superset of `attributes`."
+        found = []
+        for point in self.snapshot.get(name, []):
+            point_attributes = dict(point.attributes or {})
+            if all(point_attributes.get(k) == v for k, v in (attributes or {}).items()):
+                found.append(point)
+        return found
+
+    def point(self, name, attributes=None):
+        "The single matching data point, asserting there is exactly one."
+        found = self.points(name, attributes)
+        assert len(found) == 1, (
+            f"expected exactly one {name} point matching {attributes}, "
+            f"got {len(found)}: {found}"
+        )
+        return found[0]
+
+
+@pytest.fixture
+def otel_metrics():
+    """
+    Function-scoped metrics collector. Drains any delta state accumulated by
+    earlier tests before yielding, so counts start from zero.
+    """
+    pytest.importorskip("opentelemetry.sdk")
+    if _otel_metric_reader is None:
+        pytest.skip("OpenTelemetry SDK meter provider was not installed")
+    _otel_metric_reader.get_metrics_data()
+    yield MetricsCollector(_otel_metric_reader)
+
+
 @pytest.fixture
 def bare_ds():
     """
diff --git a/tests/test_telemetry_metrics.py b/tests/test_telemetry_metrics.py
new file mode 100644
index 00000000..6d1efd1f
--- /dev/null
+++ b/tests/test_telemetry_metrics.py
@@ -0,0 +1,323 @@
+"""
+Tests for the OpenTelemetry metrics Datasette core emits.
+
+Two layers are tested separately and deliberately:
+
+- The gauge callbacks are plain generator functions, so they are called
+  directly for exact-value assertions. Going through the SDK for those would
+  be unreliable: the pool gauges carry no attribute identifying which
+  Datasette produced them, and a pytest session has many live instances, so
+  the SDK's last-value aggregation would report whichever one happened to be
+  observed last.
+
+- The SDK pipeline (instrument -> reader -> data points) is tested through
+  the `otel_metrics` fixture, using metrics that carry `db.namespace` - a
+  uniquely named in-memory database is enough to isolate those from every
+  other instance alive in the session.
+"""
+
+import asyncio
+
+import pytest
+
+from datasette import telemetry
+from datasette.app import Datasette
+from datasette.database import Database
+from datasette.utils.sqlite import sqlite3
+
+pytestmark = pytest.mark.filterwarnings("ignore::ResourceWarning")
+
+
+def observations(callback, datasette=None):
+    """
+    Run a gauge callback, optionally keeping only observations produced by one
+    Datasette's databases. Returns a list of (attributes dict, value).
+    """
+    results = []
+    names = None
+    if datasette is not None:
+        names = {db.name for db in telemetry._databases_of(datasette)}
+    for observation in callback():
+        attributes = dict(observation.attributes or {})
+        namespace = attributes.get("db.namespace")
+        if names is not None and namespace is not None and namespace not in names:
+            continue
+        results.append((attributes, observation.value))
+    return results
+
+
+@pytest.fixture
+def metrics_ds():
+    "A Datasette with a distinctive thread count and a uniquely named database."
+    ds = Datasette(
+        memory=True,
+        settings={"num_sql_threads": 7},
+    )
+    ds.add_memory_database("metrics_test_db")
+    try:
+        yield ds
+    finally:
+        ds.close()
+
+
+@pytest.mark.asyncio
+async def test_sql_thread_limit_gauge_reports_num_sql_threads(metrics_ds):
+    values = [value for _, value in observations(telemetry.observe_sql_thread_limit)]
+    # Other instances are alive in this session, so assert membership rather
+    # than uniqueness - 7 is distinctive enough to only come from metrics_ds.
+    assert 7 in values
+
+
+@pytest.mark.asyncio
+async def test_no_thread_gauges_in_non_threaded_mode():
+    """
+    num_sql_threads=0 means there is no pool at all, so the pool gauges must
+    skip the instance rather than report a bogus limit of 0.
+
+    The pool gauges carry no attributes, so the assertion is that adding this
+    instance produces no *additional* observations.
+    """
+    before_limits = len(list(telemetry.observe_sql_thread_limit()))
+    before_depths = len(list(telemetry.observe_sql_thread_queue_depth()))
+    ds = Datasette(memory=True, settings={"num_sql_threads": 0})
+    try:
+        assert ds.executor is None
+        assert len(list(telemetry.observe_sql_thread_limit())) == before_limits
+        assert len(list(telemetry.observe_sql_thread_queue_depth())) == before_depths
+        # Per-database gauges are unaffected - they do not depend on the pool.
+        assert observations(telemetry.observe_pending_queries, ds)
+    finally:
+        ds.close()
+
+
+@pytest.mark.asyncio
+async def test_pending_queries_gauge_tracks_in_flight_queries(metrics_ds):
+    db = metrics_ds.get_database("metrics_test_db")
+    attributes = {"db.namespace": "metrics_test_db"}
+
+    def value():
+        points = [
+            v
+            for a, v in observations(telemetry.observe_pending_queries, metrics_ds)
+            if a == attributes
+        ]
+        assert len(points) == 1
+        return points[0]
+
+    assert value() == 0
+
+    # sqlite3.sleep is not a thing, so block the worker thread on an event we
+    # control from the event loop and sample the gauge while it is held.
+    release = asyncio.Event()
+    loop = asyncio.get_running_loop()
+    entered = asyncio.Event()
+
+    def blocking_fn(conn):
+        loop.call_soon_threadsafe(entered.set)
+        asyncio.run_coroutine_threadsafe(release.wait(), loop).result()
+        return "done"
+
+    task = asyncio.ensure_future(db.execute_fn(blocking_fn))
+    await entered.wait()
+    assert value() == 1, "a query occupying a pool thread must be counted as pending"
+    release.set()
+    assert await task == "done"
+    assert value() == 0, "the count must drop once the query completes"
+
+
+@pytest.mark.asyncio
+async def test_write_queue_depth_gauge(metrics_ds):
+    db = metrics_ds.get_database("metrics_test_db")
+    attributes = {"db.namespace": "metrics_test_db"}
+
+    def depths():
+        return [
+            v
+            for a, v in observations(telemetry.observe_write_queue_depth, metrics_ds)
+            if a == attributes
+        ]
+
+    # No write has ever been queued, so there is no queue and no observation -
+    # rather than a fabricated zero for a queue that does not exist.
+    assert depths() == []
+
+    await db.execute_write("create table t (id integer primary key)")
+    assert depths() == [0], "an idle write queue reports zero, not nothing"
+
+
+@pytest.mark.asyncio
+async def test_open_connections_gauge(metrics_ds, tmp_path):
+    path = str(tmp_path / "conns.db")
+    sqlite3.connect(path).execute("create table t (id integer primary key)")
+    db = metrics_ds.add_database(Database(metrics_ds, path=path), name="conns_db")
+    attributes = {"db.namespace": "conns_db"}
+
+    def open_connections():
+        points = [
+            v
+            for a, v in observations(telemetry.observe_open_connections, metrics_ds)
+            if a == attributes
+        ]
+        assert len(points) == 1
+        return points[0]
+
+    assert open_connections() == 0
+    await db.execute("select 1")
+    assert open_connections() >= 1, "executing a query opens a tracked file connection"
+
+
+@pytest.mark.asyncio
+async def test_operation_duration_histogram_read(otel_metrics):
+    ds = Datasette(memory=True)
+    ds.add_memory_database("duration_read_db")
+    try:
+        db = ds.get_database("duration_read_db")
+        await db.execute("select 1")
+        otel_metrics.collect()
+        point = otel_metrics.point(
+            "db.client.operation.duration",
+            {"db.namespace": "duration_read_db", "datasette.operation": "read"},
+        )
+        assert point.count == 1
+        assert point.sum > 0
+        assert dict(point.attributes)["db.system"] == "sqlite"
+        assert "error.type" not in dict(point.attributes)
+    finally:
+        ds.close()
+
+
+@pytest.mark.asyncio
+async def test_operation_duration_histogram_write(otel_metrics):
+    ds = Datasette(memory=True)
+    ds.add_memory_database("duration_write_db")
+    try:
+        db = ds.get_database("duration_write_db")
+        await db.execute_write("create table t (id integer primary key)")
+        otel_metrics.collect()
+        point = otel_metrics.point(
+            "db.client.operation.duration",
+            {"db.namespace": "duration_write_db", "datasette.operation": "write"},
+        )
+        assert point.count == 1
+        assert point.sum > 0
+    finally:
+        ds.close()
+
+
+@pytest.mark.asyncio
+async def test_operation_duration_records_error_type(otel_metrics):
+    "A failed query is still timed, and is separable from a successful one."
+    ds = Datasette(memory=True)
+    ds.add_memory_database("duration_error_db")
+    try:
+        db = ds.get_database("duration_error_db")
+        with pytest.raises(sqlite3.OperationalError):
+            await db.execute("select * from nope")
+        otel_metrics.collect()
+        point = otel_metrics.point(
+            "db.client.operation.duration",
+            {"db.namespace": "duration_error_db", "datasette.operation": "read"},
+        )
+        assert point.count == 1
+        assert dict(point.attributes)["error.type"] == "OperationalError"
+    finally:
+        ds.close()
+
+
+@pytest.mark.asyncio
+async def test_write_queue_wait_histogram(otel_metrics):
+    ds = Datasette(memory=True)
+    ds.add_memory_database("queue_wait_db")
+    try:
+        db = ds.get_database("queue_wait_db")
+        await db.execute_write("create table t (id integer primary key)")
+        await db.execute_write("insert into t (id) values (1)")
+        otel_metrics.collect()
+        point = otel_metrics.point(
+            "datasette.write.queue_wait", {"db.namespace": "queue_wait_db"}
+        )
+        assert point.count == 2, "one measurement per write dequeued"
+        assert point.sum >= 0
+    finally:
+        ds.close()
+
+
+@pytest.mark.asyncio
+async def test_interrupted_queries_counter(otel_metrics):
+    "The count of time-limit kills, which sampled traces cannot provide."
+    ds = Datasette(memory=True, settings={"sql_time_limit_ms": 1})
+    ds.add_memory_database("interrupted_db")
+    try:
+        db = ds.get_database("interrupted_db")
+        from datasette.database import QueryInterrupted
+
+        with pytest.raises(QueryInterrupted):
+            await db.execute("""
+                with recursive counter(x) as (
+                    select 0 union all select x + 1 from counter
+                )
+                select * from counter
+                """)
+        otel_metrics.collect()
+        point = otel_metrics.point(
+            "datasette.sql.queries.interrupted", {"db.namespace": "interrupted_db"}
+        )
+        assert point.value == 1
+    finally:
+        ds.close()
+
+
+@pytest.mark.asyncio
+async def test_metrics_are_reported_through_the_sdk_for_gauges(otel_metrics):
+    "End-to-end: a gauge callback reaches the reader as a data point."
+    ds = Datasette(memory=True)
+    ds.add_memory_database("gauge_pipeline_db")
+    try:
+        await ds.get_database("gauge_pipeline_db").execute("select 1")
+        otel_metrics.collect()
+        point = otel_metrics.point(
+            "datasette.sql.queries.pending", {"db.namespace": "gauge_pipeline_db"}
+        )
+        assert point.value == 0
+        assert otel_metrics.points("datasette.sql.threads.limit")
+    finally:
+        ds.close()
+
+
+def test_closed_datasette_stops_being_observed():
+    ds = Datasette(memory=True)
+    ds.add_memory_database("closed_db")
+    assert observations(telemetry.observe_pending_queries, ds)
+    ds.close()
+    names = [
+        attributes.get("db.namespace")
+        for attributes, _ in observations(telemetry.observe_pending_queries)
+    ]
+    assert "closed_db" not in names
+
+
+def test_registry_holds_instances_weakly():
+    """
+    Registering an instance must never be the thing that keeps it alive.
+
+    A stand-in object is used rather than a real Datasette because a Datasette
+    with a temp-disk internal database is pinned for the life of the process
+    by `Database.__init__`'s `atexit.register(self._cleanup_temp_file)`, which
+    holds the Database, which holds the Datasette. That is pre-existing and
+    unrelated to telemetry; what is tested here is that this registry adds no
+    reference of its own.
+    """
+    import gc
+    import weakref
+
+    class FakeDatasette:
+        pass
+
+    fake = FakeDatasette()
+    telemetry.register_datasette(fake)
+    assert fake in telemetry._live_instances()
+    ref = weakref.ref(fake)
+    del fake
+    gc.collect()
+    assert ref() is None
+    assert not any(isinstance(ds, FakeDatasette) for ds in telemetry._live_instances())