diff --git a/build.sbt b/build.sbt
index c24ca7b8..0c015d25 100644
--- a/build.sbt
+++ b/build.sbt
@@ -231,7 +231,8 @@ lazy val tests = crossProject(JVMPlatform, JSPlatform, NativePlatform)
   )
   .jvmSettings(
     Test / fork := true,
-    javaOptions += "-Dotel.service.name=SkunkTests"
+    javaOptions += "-Dotel.service.name=SkunkTests",
+    libraryDependencies += "org.typelevel" %% "otel4s-sdk-testkit" % otel4sSdkVersion % Test,
   )
   .jsSettings(
     scalaJSLinkerConfig ~= { _.withESFeatures(_.withESVersion(org.scalajs.linker.interface.ESVersion.ES2018)) },
diff --git a/modules/core/shared/src/main/scala/Command.scala b/modules/core/shared/src/main/scala/Command.scala
index e9c6294d..2b8c4830 100644
--- a/modules/core/shared/src/main/scala/Command.scala
+++ b/modules/core/shared/src/main/scala/Command.scala
@@ -5,6 +5,7 @@
 package skunk
 
 import cats.Contravariant
+import org.typelevel.otel4s.{Attribute, Attributes}
 import org.typelevel.twiddles.Iso
 import skunk.util.Origin
 import skunk.util.Twiddler
@@ -34,16 +35,32 @@ import skunk.util.Twiddler
 final case class Command[A](
   override val sql:     String,
   override val origin:  Origin,
-  override val encoder: Encoder[A]
+  override val encoder: Encoder[A],
+  override val telemetry: Statement.Telemetry = Statement.Telemetry.empty,
 ) extends Statement[A] {
 
+  /** Attaches a low-cardinality summary used as `db.query.summary` and as the span name. Statement
+    * summaries take precedence over summaries returned by a configured query analyzer. The
+    * method is a shortcut for `addAttributes(DbAttributes.DbQuerySummary(summary))`.
+    */
+  def withQuerySummary(summary: String): Command[A] =
+    copy(telemetry = telemetry.withQuerySummary(summary))
+
+  /** Replaces the additional attributes exported on the logical database span. */
+  def withAttributes(attributes: Attributes): Command[A] =
+    copy(telemetry = telemetry.withAttributes(attributes))
+
+  /** Adds or replaces additional logical database span attributes by key. */
+  def addAttributes(attributes: Attribute[_]*): Command[A] =
+    copy(telemetry = telemetry.addAttributes(attributes: _*))
+
   /**
    * Command is a [[https://typelevel.org/cats/typeclasses/contravariant.html contravariant
    * functor]].
    * @group Transformations
    */
   def contramap[B](f: B => A): Command[B] =
-    Command(sql, origin, encoder.contramap(f))
+    Command(sql, origin, encoder.contramap(f), telemetry)
 
   @deprecated("Use .to[CaseClass] instead of .gcontramap[CaseClass]", "0.6")
   def gcontramap[B](implicit ev: Twiddler.Aux[B, A]): Command[B] =
diff --git a/modules/core/shared/src/main/scala/Query.scala b/modules/core/shared/src/main/scala/Query.scala
index 31a698d4..ba5fb85c 100644
--- a/modules/core/shared/src/main/scala/Query.scala
+++ b/modules/core/shared/src/main/scala/Query.scala
@@ -5,6 +5,7 @@
 package skunk
 
 import cats.arrow.Profunctor
+import org.typelevel.otel4s.{Attribute, Attributes}
 import org.typelevel.twiddles.Iso
 import skunk.util.Origin
 import skunk.util.Twiddler
@@ -40,15 +41,31 @@ final case class Query[A, B](
   override val origin:  Origin,
   override val encoder: Encoder[A],
   decoder: Decoder[B],
-  isDynamic: Boolean = false
+  isDynamic: Boolean = false,
+  override val telemetry: Statement.Telemetry = Statement.Telemetry.empty,
 ) extends Statement[A] {
 
+  /** Attaches a low-cardinality summary used as `db.query.summary` and as the span name. Statement
+    * summaries take precedence over summaries returned by a configured query analyzer. The
+    * method is a shortcut for `addAttributes(DbAttributes.DbQuerySummary(summary))`.
+    */
+  def withQuerySummary(summary: String): Query[A, B] =
+    copy(telemetry = telemetry.withQuerySummary(summary))
+
+  /** Replaces the additional attributes exported on the logical database span. */
+  def withAttributes(attributes: Attributes): Query[A, B] =
+    copy(telemetry = telemetry.withAttributes(attributes))
+
+  /** Adds or replaces additional logical database span attributes by key. */
+  def addAttributes(attributes: Attribute[_]*): Query[A, B] =
+    copy(telemetry = telemetry.addAttributes(attributes: _*))
+
   /**
    * Query is a profunctor.
    * @group Transformations
    */
   def dimap[C, D](f: C => A)(g: B => D): Query[C, D] =
-    Query(sql, origin, encoder.contramap(f), decoder.map(g), isDynamic)
+    Query(sql, origin, encoder.contramap(f), decoder.map(g), isDynamic, telemetry)
 
   /**
    * Query is a contravariant functor in `A`.
diff --git a/modules/core/shared/src/main/scala/Session.scala b/modules/core/shared/src/main/scala/Session.scala
index 0b43e149..4de40e88 100644
--- a/modules/core/shared/src/main/scala/Session.scala
+++ b/modules/core/shared/src/main/scala/Session.scala
@@ -8,15 +8,12 @@ import cats._
 import cats.effect._
 import cats.effect.std.Console
 import cats.syntax.all._
-import com.comcast.ip4s.*
+import com.comcast.ip4s._
 import fs2.concurrent.Signal
-import fs2.io.net.{ Network, Socket, SocketOption }
+import fs2.io.net.{Network, Socket, SocketOption}
 import fs2.Pipe
 import fs2.Stream
-import org.typelevel.otel4s.metrics.Meter
 import org.typelevel.otel4s.metrics.MeterProvider
-import org.typelevel.otel4s.metrics.Histogram
-import org.typelevel.otel4s.trace.Tracer
 import org.typelevel.otel4s.trace.TracerProvider
 import skunk.codec.all.bool
 import skunk.data._
@@ -27,6 +24,7 @@ import skunk.net.SSLNegotiation
 import skunk.net.protocol.Describe
 import scala.concurrent.duration.Duration
 import skunk.net.protocol.Parse
+import skunk.telemetry.{ConnectionInfo, Telemetry, TelemetryConfig}
 
 /**
  * Represents a live connection to a Postgres database. Operations provided here are safe to use
@@ -491,6 +489,7 @@ object Session {
     val debug: Boolean,
     val typingStrategy: TypingStrategy,
     val redactionStrategy: RedactionStrategy,
+    val telemetryConfig: TelemetryConfig,
     val ssl: SSL,
     val connectionParameters: Map[String, String],
     val socketOptions: List[SocketOption],
@@ -511,6 +510,7 @@ object Session {
       debug: Boolean = self.debug,
       typingStrategy: TypingStrategy = self.typingStrategy,
       redactionStrategy: RedactionStrategy = self.redactionStrategy,
+      telemetryConfig: TelemetryConfig = self.telemetryConfig,
       ssl: SSL = self.ssl,
       connectionParameters: Map[String, String] = self.connectionParameters,
       socketOptions: List[SocketOption] = self.socketOptions,
@@ -519,7 +519,7 @@ object Session {
       queryCacheSize: Int = self.queryCacheSize,
       parseCacheSize: Int = self.parseCacheSize,
     ): Builder[F] =
-      new Builder(connectionType, host, port, unixSocketAddress, unixSocketDirectory, credentials, database, debug, typingStrategy, redactionStrategy, ssl, connectionParameters, socketOptions, readTimeout, commandCacheSize, queryCacheSize, parseCacheSize)
+      new Builder(connectionType, host, port, unixSocketAddress, unixSocketDirectory, credentials, database, debug, typingStrategy, redactionStrategy, telemetryConfig, ssl, connectionParameters, socketOptions, readTimeout, commandCacheSize, queryCacheSize, parseCacheSize)
 
     /** Configures the connection type. */
     def withConnectionType(newConnectionType: ConnectionType): Builder[F] =
@@ -588,6 +588,10 @@ object Session {
     def withRedactionStrategy(newRedactionStrategy: RedactionStrategy): Builder[F] =
       copy(redactionStrategy = newRedactionStrategy)
 
+    /** Configures query capture, query analysis, protocol spans, and pool spans. */
+    def withTelemetryConfig(newTelemetryConfig: TelemetryConfig): Builder[F] =
+      copy(telemetryConfig = newTelemetryConfig)
+
     def withSSL(newSSL: SSL): Builder[F] =
       copy(ssl = newSSL)
 
@@ -633,26 +637,23 @@ object Session {
      */
     def pooled(max: Int): Resource[F, Resource[F, Session[F]]] =
       for {
-        tracer <- Resource.eval(TracerProvider[F].tracer("org.typelevel.skunk").withVersion(BuildInfo.version).get)
-        meter  <- Resource.eval(MeterProvider[F].meter("org.typelevel.skunk").withVersion(BuildInfo.version).get)
-        pool   <- pooledWithTracer(max)(meter)
-      } yield pool(tracer)
+        telemetry <- Resource.eval(Telemetry.create(telemetryConfig, connectionInfo(database.getOrElse(""))))
+        pool      <- pooledWithTelemetry(max)
+      } yield pool(telemetry)
 
-    private def pooledWithTracer(max: Int)(implicit M: Meter[F]): Resource[F, Tracer[F] => Resource[F, Session[F]]] = {
+    private def pooledWithTelemetry(max: Int): Resource[F, Telemetry[F] => Resource[F, Session[F]]] = {
       val logger: String => F[Unit] = s => Console[F].println(s"TLS: $s")
       for {
         dc      <- Resource.eval(Describe.Cache.empty[F](commandCacheSize, queryCacheSize))
         sslOp   <- ssl.toSSLNegotiationOptions(if (debug) logger.some else none)
-        opDuration <- Resource.eval(Otel.OpDurationHistogram[F])
-        pool    <- Pool.ofF({implicit T: Tracer[F] => sessions(sslOp, dc, opDuration)}, max)(Recyclers.full)
+        pool    <- Pool.ofF({implicit T: Telemetry[F] => sessions(sslOp, dc)}, max)(Recyclers.full)
       } yield pool
     }
 
     private def sessions(
       sslOptions:    Option[SSLNegotiation.Options[F]],
       describeCache: Describe.Cache[F],
-      opDuration:    Histogram[F, Double]
-    )(implicit T: Tracer[F]): Resource[F, Session[F]] = {
+    )(implicit T: Telemetry[F]): Resource[F, Session[F]] = {
       val sockets = connectionType match {
         case ConnectionType.TCP =>
           val address = SocketAddress(host, port)
@@ -663,23 +664,45 @@ object Session {
           val filteredSocketOptions = socketOptions.filter(o => o.key != SocketOption.NoDelay)
           Network[F].connect(address, filteredSocketOptions)
       }
-      fromSockets(sockets, sslOptions, describeCache, opDuration)
+
+      for {
+        resolvedCredentials <- Resource.eval(credentials)
+        telemetry = T.withConnection(connectionInfo(database.getOrElse(resolvedCredentials.user)))
+        session <- fromSockets(sockets, sslOptions, describeCache, resolvedCredentials)(telemetry)
+      } yield session
     }
 
     private def fromSockets(
       sockets:           Resource[F, Socket[F]],
       sslOptions:        Option[SSLNegotiation.Options[F]],
       describeCache:     Describe.Cache[F],
-      opDuration:        Histogram[F, Double]
-    )(implicit T: Tracer[F]): Resource[F, Session[F]] =
+      creds:             Credentials,
+    )(implicit T: Telemetry[F]): Resource[F, Session[F]] =
        for {
         namer <- Resource.eval(Namer[F])
         pc    <- Resource.eval(Parse.Cache.empty[F](parseCacheSize))
-        proto <- Protocol[F](debug, namer, sockets, sslOptions, describeCache, pc, readTimeout, redactionStrategy, opDuration)
-        creds <- Resource.eval(credentials)
+        proto <- Protocol[F](debug, namer, sockets, sslOptions, describeCache, pc, readTimeout, redactionStrategy)
         _     <- Resource.eval(proto.startup(creds.user, database.getOrElse(creds.user), creds.password, connectionParameters))
         sess  <- Resource.make(fromProtocol(proto, namer, typingStrategy, redactionStrategy))(_ => proto.cleanup)
       } yield sess
+
+
+    private def connectionInfo(databaseName: String): ConnectionInfo =
+      connectionType match {
+        case ConnectionType.TCP =>
+          ConnectionInfo(
+            databaseName,
+            host.toString,
+            Option.when(port.value != 5432)(port.value.toLong),
+          )
+        case ConnectionType.Unix =>
+          ConnectionInfo(
+            databaseName,
+            unixSocketAddress.fold(s"$unixSocketDirectory/.s.PGSQL.$port")(_.path),
+            None,
+          )
+      }
+
   }
 
   /**
@@ -714,6 +737,7 @@ object Session {
         debug = false,
         typingStrategy = TypingStrategy.BuiltinsOnly,
         redactionStrategy = RedactionStrategy.OptIn,
+        telemetryConfig = TelemetryConfig.default,
         ssl = SSL.None,
         connectionParameters = DefaultConnectionParameters,
         socketOptions = DefaultSocketOptions,
diff --git a/modules/core/shared/src/main/scala/Statement.scala b/modules/core/shared/src/main/scala/Statement.scala
index aabfe62d..92fec87d 100644
--- a/modules/core/shared/src/main/scala/Statement.scala
+++ b/modules/core/shared/src/main/scala/Statement.scala
@@ -4,6 +4,8 @@
 
 package skunk
 
+import org.typelevel.otel4s.{Attribute, Attributes}
+import org.typelevel.otel4s.semconv.attributes.DbAttributes
 import skunk.util.Origin
 import skunk.data.Type
 
@@ -12,10 +14,49 @@ trait Statement[A] {
   def origin:   Origin
   def encoder:  Encoder[A]
   def cacheKey: Statement.CacheKey
+  def telemetry: Statement.Telemetry
 }
 
 object Statement {
 
+  /** Explicit, typed telemetry metadata attached to a statement. */
+  sealed trait Telemetry {
+    /** The low-cardinality query summary, if defined. */
+    def querySummary: Option[String]
+
+    /** Additional attributes exported on the logical database span. */
+    def attributes: Attributes
+
+    /** Shortcut for `addAttributes(DbAttributes.DbQuerySummary(summary))`. */
+    def withQuerySummary(summary: String): Telemetry
+
+    /** Replaces the additional logical database span attributes. */
+    def withAttributes(attributes: Attributes): Telemetry
+
+    /** Adds or replaces additional logical database span attributes by key. */
+    def addAttributes(attributes: Attribute[_]*): Telemetry
+  }
+
+  object Telemetry {
+    val empty: Telemetry = Impl(Attributes.empty)
+
+    private final case class Impl(attributes: Attributes) extends Telemetry {
+      def querySummary: Option[String] =
+        attributes.get(DbAttributes.DbQuerySummary).map(_.value)
+
+      def withQuerySummary(summary: String): Telemetry =
+        copy(attributes = attributes.added(DbAttributes.DbQuerySummary(summary)))
+
+      def withAttributes(attributes: Attributes): Telemetry =
+        copy(attributes = attributes)
+
+      def addAttributes(values: Attribute[_]*): Telemetry =
+        withAttributes(attributes ++ values)
+
+      override def toString: String = s"Telemetry($attributes)"
+    }
+  }
+
   /**
    * A digest of a `Statement`, consisting only of the SQL statement and asserted input/output
    * types. This data type has lawful universal equality/hashing so we can use it as a hash key,
@@ -23,4 +64,4 @@ object Statement {
    */
   final case class CacheKey(sql: String, encodedTypes: List[Type], decodedTypes: List[Type])
 
-}
\ No newline at end of file
+}
diff --git a/modules/core/shared/src/main/scala/net/Protocol.scala b/modules/core/shared/src/main/scala/net/Protocol.scala
index b880e43a..2d1cdb6f 100644
--- a/modules/core/shared/src/main/scala/net/Protocol.scala
+++ b/modules/core/shared/src/main/scala/net/Protocol.scala
@@ -13,13 +13,12 @@ import skunk.{ Command, Query, Statement, ~, Void, RedactionStrategy }
 import skunk.data._
 import skunk.util.{ Namer, Origin }
 import skunk.util.Typer
-import org.typelevel.otel4s.trace.Tracer
 import fs2.io.net.Socket
 import skunk.net.protocol.Describe
 import scala.concurrent.duration.Duration
 import skunk.net.protocol.Exchange
 import skunk.net.protocol.Parse
-import org.typelevel.otel4s.metrics.Histogram
+import skunk.telemetry.Telemetry
 
 /**
  * Interface for a Postgres database, expressed through high-level operations that rely on exchange
@@ -209,7 +208,7 @@ object Protocol {
     def execute(maxRows: Int): F[List[B] ~ Boolean]
   }
 
-  def apply[F[_]: Temporal: Tracer: Console](
+  def apply[F[_]: Temporal: Telemetry: Console](
     debug:             Boolean,
     nam:               Namer[F],
     sockets:           Resource[F, Socket[F]],
@@ -218,20 +217,18 @@ object Protocol {
     parseCache:        Parse.Cache[F],
     readTimeout:       Duration,
     redactionStrategy: RedactionStrategy,
-    opDuration:        Histogram[F, Double]
   ): Resource[F, Protocol[F]] =
     for {
       bms <- BufferedMessageSocket[F](256, debug, sockets, sslOptions, readTimeout) // TODO: should we expose the queue size?
-      p   <- Resource.eval(fromMessageSocket(bms, nam, describeCache, parseCache, redactionStrategy, opDuration))
+      p   <- Resource.eval(fromMessageSocket(bms, nam, describeCache, parseCache, redactionStrategy))
     } yield p
 
-  def fromMessageSocket[F[_]: Concurrent: Tracer](
+  def fromMessageSocket[F[_]: Concurrent: Telemetry](
     bms: BufferedMessageSocket[F],
     nam: Namer[F],
     dc:  Describe.Cache[F],
     pc:  Parse.Cache[F],
     redactionStrategy: RedactionStrategy,
-    opDuration: Histogram[F, Double]
   ): F[Protocol[F]] =
     Exchange[F].map { ex =>
       new Protocol[F] {
@@ -249,38 +246,38 @@ object Protocol {
           bms.parameters
 
         override def prepare[A](command: Command[A], ty: Typer): F[PreparedCommand[F, A]] =
-          protocol.Prepare[F](describeCache, parseCache, redactionStrategy, opDuration).apply(command, ty)
+          protocol.Prepare[F](describeCache, parseCache, redactionStrategy).apply(command, ty)
 
         override def prepare[A, B](query: Query[A, B], ty: Typer): F[PreparedQuery[F, A, B]] =
-          protocol.Prepare[F](describeCache, parseCache, redactionStrategy, opDuration).apply(query, ty)
+          protocol.Prepare[F](describeCache, parseCache, redactionStrategy).apply(query, ty)
 
         override def prepareR[A](command: Command[A], ty: Typer): Resource[F, Protocol.PreparedCommand[F, A]] = {
           val acquire = Parse.Cache.empty[F](1).flatMap { pc =>
-            protocol.Prepare[F](describeCache, pc, redactionStrategy, opDuration).apply(command, ty)
+            protocol.Prepare[F](describeCache, pc, redactionStrategy).apply(command, ty)
           }
-          Resource.make(acquire)(pc => protocol.Close[F](opDuration).apply(pc.id))
+          Resource.make(acquire)(pc => protocol.Close[F].apply(pc.id))
         }
 
         override def prepareR[A, B](query: Query[A, B], ty: Typer): Resource[F, Protocol.PreparedQuery[F, A, B]] = {
           val acquire = Parse.Cache.empty[F](1).flatMap { pc =>
-            protocol.Prepare[F](describeCache, pc, redactionStrategy, opDuration).apply(query, ty)
+            protocol.Prepare[F](describeCache, pc, redactionStrategy).apply(query, ty)
           }
-          Resource.make(acquire)(pq => protocol.Close[F](opDuration).apply(pq.id))
+          Resource.make(acquire)(pq => protocol.Close[F].apply(pq.id))
         }
 
         override def execute(command: Command[Void]): F[Completion] =
-          protocol.Query[F](redactionStrategy, opDuration).apply(command)
+          protocol.Query[F](redactionStrategy).apply(command)
 
         override def execute[B](query: Query[Void, B], ty: Typer): F[List[B]] =
-          protocol.Query[F](redactionStrategy, opDuration).apply(query, ty)
+          protocol.Query[F](redactionStrategy).apply(query, ty)
 
-        override def executeDiscard(statement: Statement[Void]): F[Unit] = protocol.Query[F](redactionStrategy, opDuration).applyDiscard(statement)
+        override def executeDiscard(statement: Statement[Void]): F[Unit] = protocol.Query[F](redactionStrategy).applyDiscard(statement)
 
         override def startup(user: String, database: String, password: Option[String], parameters: Map[String, String]): F[Unit] =
-          protocol.Startup[F](opDuration).apply(user, database, password, parameters)
+          protocol.Startup[F].apply(user, database, password, parameters)
 
         override def cleanup: F[Unit] =
-          parseCache.value.values.flatMap(_.traverse_(protocol.Close[F](opDuration).apply))
+          parseCache.value.values.flatMap(_.traverse_(protocol.Close[F].apply))
       
         override def transactionStatus: Signal[F, TransactionStatus] =
           bms.transactionStatus
@@ -292,10 +289,8 @@ object Protocol {
          pc
 
         override def closeEvictedPreparedStatements: F[Unit] = 
-          pc.value.clearEvicted.flatMap(_.traverse_(protocol.Close[F](opDuration).apply))
+          pc.value.clearEvicted.flatMap(_.traverse_(protocol.Close[F].apply))
       }
     }
 
 }
-
-
diff --git a/modules/core/shared/src/main/scala/net/protocol/Bind.scala b/modules/core/shared/src/main/scala/net/protocol/Bind.scala
index 33e905b4..c1c98232 100644
--- a/modules/core/shared/src/main/scala/net/protocol/Bind.scala
+++ b/modules/core/shared/src/main/scala/net/protocol/Bind.scala
@@ -12,10 +12,8 @@ import skunk.net.message.{ Bind => BindMessage, Close => _, _ }
 import skunk.net.MessageSocket
 import skunk.net.Protocol.{ PreparedStatement, PortalId }
 import skunk.util.{ Origin, Namer }
-import org.typelevel.otel4s.Attribute
-import org.typelevel.otel4s.trace.{Span, Tracer}
 import skunk.RedactionStrategy
-import org.typelevel.otel4s.metrics.Histogram
+import skunk.telemetry.{SkunkAttributes, Telemetry}
 
 trait Bind[F[_]] {
 
@@ -30,7 +28,7 @@ trait Bind[F[_]] {
 
 object Bind {
 
-  def apply[F[_]: Exchange: MessageSocket: Namer: Tracer](opDuration: Histogram[F, Double])(
+  def apply[F[_]: Exchange: MessageSocket: Namer: Telemetry](
     implicit ev: MonadCancel[F, Throwable]
   ): Bind[F] =
     new Bind[F] {
@@ -42,13 +40,13 @@ object Bind {
         redactionStrategy: RedactionStrategy
       ): Resource[F, PortalId] =
         Resource.make {
-          exchange("bind", opDuration) { (span: Span[F]) =>
+          exchange("bind") {
             for {
               pn <- nextName("portal").map(PortalId(_))
               ea  = statement.statement.encoder.encode(args) // encoded args
-              _  <- span.addAttributes(
-                Attribute("arguments", redactionStrategy.redactArguments(ea).map(_.orNull).mkString(",")),
-                Attribute("portal-id", pn.value)
+              _  <- Telemetry[F].addProtocolAttributes(
+                SkunkAttributes.portalId(pn.value),
+                SkunkAttributes.statementId(statement.id.value)
               )
               _  <- send(BindMessage(pn.value, statement.id.value, ea.map(_.map(_.value))))
               _  <- send(Flush)
@@ -71,7 +69,7 @@ object Bind {
                     }
             } yield pn
           }
-        } { Close[F](opDuration).apply }
+        } { Close[F].apply }
 
     }
 
diff --git a/modules/core/shared/src/main/scala/net/protocol/BindExecute.scala b/modules/core/shared/src/main/scala/net/protocol/BindExecute.scala
index 63456ecd..2a82c6e2 100644
--- a/modules/core/shared/src/main/scala/net/protocol/BindExecute.scala
+++ b/modules/core/shared/src/main/scala/net/protocol/BindExecute.scala
@@ -13,14 +13,11 @@ import skunk.net.message.{ Bind => BindMessage, Execute => ExecuteMessage, Close
 import skunk.net.MessageSocket
 import skunk.net.Protocol.PortalId
 import skunk.util.{ Origin, Namer }
-import org.typelevel.otel4s.Attribute
-import org.typelevel.otel4s.trace.{Span, Tracer}
 import skunk.RedactionStrategy
 import skunk.net.Protocol
 import skunk.data.Completion
-import skunk.net.protocol.exchange
 import cats.effect.kernel.Deferred
-import org.typelevel.otel4s.metrics.Histogram
+import skunk.telemetry.{SkunkAttributes, Telemetry}
 
 trait BindExecute[F[_]] {
 
@@ -42,7 +39,7 @@ trait BindExecute[F[_]] {
 
 object BindExecute {
   
-  def apply[F[_]: Exchange: MessageSocket: Namer: Tracer](opDuration: Histogram[F, Double])(
+  def apply[F[_]: Exchange: MessageSocket: Namer: Telemetry](
     implicit ev: Concurrent[F]
   ): BindExecute[F] =
     new Unroll[F] with BindExecute[F] {
@@ -50,17 +47,16 @@ object BindExecute {
       def bindExchange[A](
         statement: Protocol.PreparedStatement[F, A],
         args: A,
-        argsOrigin: Origin,
-        redactionStrategy: RedactionStrategy
-      ):(Span[F] => F[PortalId], F[Unit]) = {
+        argsOrigin: Origin
+      ): (F[PortalId], F[Unit]) = {
         val ea  = statement.statement.encoder.encode(args) // encoded args
         
-        def preBind(span: Span[F]): F[PortalId] = for {
+        val preBind: F[PortalId] = for {
               pn <- nextName("portal").map(PortalId(_))
-              _  <- span.addAttributes(
-                Attribute("arguments", redactionStrategy.redactArguments(ea).map(_.orNull).mkString(",")),
-                Attribute("portal-id", pn.value)
-              )
+              _  <- Telemetry[F].addAttributes(
+               SkunkAttributes.portalId(pn.value),
+               SkunkAttributes.statementId(statement.id.value)
+             )
               _  <- send(BindMessage(pn.value, statement.id.value, ea.map(_.map(_.value))))
         } yield pn
 
@@ -91,7 +87,7 @@ object BindExecute {
         redactionStrategy: RedactionStrategy
       ): Resource[F, Protocol.CommandPortal[F, A]] = {
 
-        val (preBind, postBind) = bindExchange(statement, args, argsOrigin, redactionStrategy)
+        val (preBind, postBind) = bindExchange(statement, args, argsOrigin)
 
         val postExec: F[Completion] = flatExpect {
           case CommandComplete(c) => send(Sync) *> expect { case ReadyForQuery(_) => c } // https://github.com/tpolecat/skunk/issues/210
@@ -134,9 +130,15 @@ object BindExecute {
         }
 
         Resource.make {
-          exchange("bind+execute", opDuration){ (span: Span[F]) =>
+          database(
+            "bind+execute",
+            statement.statement,
+            statement.statement.encoder.encode(args),
+            redactionStrategy,
+          ) {
+
             for {
-              pn <- preBind(span)
+              pn <- preBind
               _  <- send(ExecuteMessage(pn.value, 0))
               _  <- send(Flush)
               _  <- postBind
@@ -145,7 +147,7 @@ object BindExecute {
               def execute: F[Completion] = c.pure
             }
           }
-        } { portal => Close[F](opDuration).apply(portal.id)}
+        } { portal => Close[F].apply(portal.id)}
 
       }
 
@@ -156,16 +158,18 @@ object BindExecute {
         redactionStrategy: RedactionStrategy,
         initialSize: Int
       ): Resource[F, Protocol.QueryPortal[F, A, B]] = {
-        val (preBind, postBind) = bindExchange(statement, args, argsOrigin, redactionStrategy)
+        val (preBind, postBind) = bindExchange(statement, args, argsOrigin)
         Resource.eval(Deferred[F, Unit]).flatMap { prefetch =>
           Resource.make {
-            exchange("bind+execute", opDuration){ (span: Span[F]) =>
+            database(
+              "bind+execute",
+              statement.statement,
+              statement.statement.encoder.encode(args),
+              redactionStrategy,
+            ) {
               for {
-                pn <- preBind(span)
-                _  <- span.addAttributes(
-                        Attribute("max-rows",  initialSize.toLong),
-                        Attribute("portal-id", pn.value)
-                      )
+                pn <- preBind
+                _  <- Telemetry[F].addAttributes(SkunkAttributes.fetchMaxRows(initialSize.toLong))
                 _  <- send(ExecuteMessage(pn.value, initialSize))
                 _  <- send(Flush)
                 _  <- postBind
@@ -174,11 +178,11 @@ object BindExecute {
                 def execute(maxRows: Int): F[List[B] ~ Boolean] = 
                   prefetch.tryGet.flatMap {
                     case None => rs.pure <* prefetch.complete(())
-                    case Some(()) => Execute[F](opDuration).apply(this, maxRows)
+                    case Some(()) => Execute[F].apply(this, maxRows)
                   }
               }
             }
-          } { portal => Close[F](opDuration).apply(portal.id)}
+          } { portal => Close[F].apply(portal.id)}
         }
       }
   }
diff --git a/modules/core/shared/src/main/scala/net/protocol/Close.scala b/modules/core/shared/src/main/scala/net/protocol/Close.scala
index b7ae019c..f9ad02ae 100644
--- a/modules/core/shared/src/main/scala/net/protocol/Close.scala
+++ b/modules/core/shared/src/main/scala/net/protocol/Close.scala
@@ -8,10 +8,7 @@ package protocol
 import cats.effect.MonadCancelThrow
 import cats.syntax.all._
 import skunk.net.message.{ Close => CloseMessage, Flush, CloseComplete }
-import org.typelevel.otel4s.Attribute
-import org.typelevel.otel4s.trace.Span
-import org.typelevel.otel4s.trace.Tracer
-import org.typelevel.otel4s.metrics.Histogram
+import skunk.telemetry.{SkunkAttributes, Telemetry}
 
 trait Close[F[_]] {
   def apply(portalId: Protocol.PortalId): F[Unit]
@@ -20,18 +17,18 @@ trait Close[F[_]] {
 
 object Close {
 
-  def apply[F[_]: MonadCancelThrow: Exchange: MessageSocket: Tracer](opDuration: Histogram[F, Double]): Close[F] =
+  def apply[F[_]: MonadCancelThrow: Exchange: MessageSocket: Telemetry]: Close[F] =
     new Close[F] {
 
       override def apply(portalId: Protocol.PortalId): F[Unit] =
-        exchange("close-portal", opDuration) { (span: Span[F]) =>
-          span.addAttribute(Attribute("portal", portalId.value)) *>
+        exchange("close-portal") {
+          Telemetry[F].addProtocolAttributes(SkunkAttributes.portalId(portalId.value)) *>
           close(CloseMessage.portal(portalId.value))
         }
 
       override def apply(statementId: Protocol.StatementId): F[Unit] =
-        exchange("close-statement", opDuration) { (span: Span[F]) =>
-          span.addAttribute(Attribute("statement", statementId.value)) *>
+        exchange("close-statement") {
+          Telemetry[F].addProtocolAttributes(SkunkAttributes.statementId(statementId.value)) *>
           close(CloseMessage.statement(statementId.value))
         }
 
diff --git a/modules/core/shared/src/main/scala/net/protocol/Execute.scala b/modules/core/shared/src/main/scala/net/protocol/Execute.scala
index 51fa3a38..3c5dac72 100644
--- a/modules/core/shared/src/main/scala/net/protocol/Execute.scala
+++ b/modules/core/shared/src/main/scala/net/protocol/Execute.scala
@@ -9,10 +9,7 @@ import cats.effect.MonadCancel
 import skunk.~
 import skunk.net.{ Protocol, MessageSocket }
 import skunk.net.message.{ Execute => ExecuteMessage, _ }
-import org.typelevel.otel4s.Attribute
-import org.typelevel.otel4s.trace.Span
-import org.typelevel.otel4s.trace.Tracer
-import org.typelevel.otel4s.metrics.Histogram
+import skunk.telemetry.{SkunkAttributes, Telemetry}
 
 trait Execute[F[_]] {
   def apply[A, B](portal: Protocol.QueryPortal[F, A, B], maxRows: Int): F[List[B] ~ Boolean]
@@ -20,22 +17,28 @@ trait Execute[F[_]] {
 
 object Execute {
 
-  def apply[F[_]: Exchange: MessageSocket: Tracer](opDuration: Histogram[F, Double])(
+  def apply[F[_]: Exchange: MessageSocket: Telemetry](
     implicit ev: MonadCancel[F, Throwable]
   ): Execute[F] =
     new Unroll[F] with Execute[F] {
 
       override def apply[A, B](portal: Protocol.QueryPortal[F, A, B], maxRows: Int): F[List[B] ~ Boolean] =
-        exchange("execute", opDuration) { (span: Span[F]) =>
-          for {
-            _  <- span.addAttributes(
-                    Attribute("max-rows",    maxRows.toLong),
-                    Attribute("portal-id", portal.id.value)
-                  )
-            _  <- send(ExecuteMessage(portal.id.value, maxRows))
-            _  <- send(Flush)
-            rs <- unroll(portal)
-          } yield rs
+        database(
+          "execute",
+          portal.preparedStatement.statement,
+          portal.preparedStatement.statement.encoder.encode(portal.arguments),
+          portal.redactionStrategy,
+        ) {
+            for {
+              _  <- Telemetry[F].addAttributes(
+                      SkunkAttributes.fetchMaxRows(maxRows.toLong),
+                      SkunkAttributes.portalId(portal.id.value),
+                      SkunkAttributes.statementId(portal.preparedStatement.id.value)
+                    )
+              _  <- send(ExecuteMessage(portal.id.value, maxRows))
+              _  <- send(Flush)
+              rs <- unroll(portal)
+            } yield rs
         }
 
     }
diff --git a/modules/core/shared/src/main/scala/net/protocol/ParseDescribe.scala b/modules/core/shared/src/main/scala/net/protocol/ParseDescribe.scala
index 477df3c0..23c73b98 100644
--- a/modules/core/shared/src/main/scala/net/protocol/ParseDescribe.scala
+++ b/modules/core/shared/src/main/scala/net/protocol/ParseDescribe.scala
@@ -10,16 +10,13 @@ import skunk.net.Protocol.StatementId
 import skunk.net.message.{ Describe => DescribeMessage, Parse => ParseMessage, _ }
 import skunk.util.Typer
 import skunk.data.TypedRowDescription
-import org.typelevel.otel4s.Attribute
-import org.typelevel.otel4s.trace.Span
-import org.typelevel.otel4s.trace.Tracer
 import cats.data.OptionT
 import cats.effect.MonadCancel
 import skunk.Statement
 import skunk.exception.*
 import skunk.util.Namer
 import skunk.net.protocol.exchange
-import org.typelevel.otel4s.metrics.Histogram
+import skunk.telemetry.{SkunkAttributes, Telemetry}
 
 trait ParseDescribe[F[_]] {
   def command[A](cmd: skunk.Command[A], ty: Typer): F[StatementId]
@@ -28,7 +25,7 @@ trait ParseDescribe[F[_]] {
 
 object ParseDescribe {
 
-  def apply[F[_]: Exchange: MessageSocket: Tracer: Namer](cache: Describe.Cache[F], parseCache: Parse.Cache[F], opDuration: Histogram[F, Double])(
+  def apply[F[_]: Exchange: MessageSocket: Telemetry: Namer](cache: Describe.Cache[F], parseCache: Parse.Cache[F])(
     implicit ev: MonadCancel[F, Throwable]
   ): ParseDescribe[F] =
     new ParseDescribe[F] {
@@ -45,7 +42,7 @@ object ParseDescribe {
                 ).raiseError[F, Unit]
         } yield a
 
-      def parseExchange(stmt: Statement[_], ty: Typer)(span: Span[F]): F[(F[StatementId], StatementId => F[Unit])] = 
+      def parseExchange(stmt: Statement[_], ty: Typer): F[(F[StatementId], StatementId => F[Unit])] =
         stmt.encoder.oids(ty) match {
 
           case Right(os) if os.length > Short.MaxValue =>
@@ -54,10 +51,11 @@ object ParseDescribe {
           case Right(os) =>
 
             def addStatement(id: StatementId): F[Unit] =
-              span.addAttributes(
-                Attribute("statement-name", id.value),
-                Attribute("statement-sql",  stmt.sql),
-                Attribute("statement-parameter-types", os.map(n => ty.typeForOid(n, -1).fold(n.toString)(_.toString)).mkString("[", ", ", "]"))
+              Telemetry[F].addProtocolAttributes(
+                SkunkAttributes.statementId(id.value),
+                SkunkAttributes.statementParameterTypes(
+                  os.map(n => ty.typeForOid(n, -1).fold(n.toString)(_.toString)).mkString("[", ", ", "]")
+                )
               )
 
             OptionT(parseCache.value.get(stmt)).map(id => (addStatement(id).as(id), (_:StatementId) => ().pure)).getOrElse {
@@ -83,13 +81,9 @@ object ParseDescribe {
 
       override def command[A](cmd: skunk.Command[A], ty: Typer): F[StatementId] = {
 
-        def describeExchange(span: Span[F]): F[(StatementId => F[Unit], F[Unit])] = {
-          def addStatementId(id: StatementId): F[Unit] =
-            span.addAttribute(Attribute("statement-id", id.value))
-
-          OptionT(cache.commandCache.get(cmd)).as(((id: StatementId) => addStatementId(id), ().pure[F])).getOrElse {
+        def describeExchange: F[(StatementId => F[Unit], F[Unit])] = {
+          OptionT(cache.commandCache.get(cmd)).as(((_: StatementId) => ().pure[F], ().pure[F])).getOrElse {
             val pre = (id: StatementId) => for {
-              _  <- addStatementId(id)
               _  <- send(DescribeMessage.statement(id.value))
             } yield ()
 
@@ -116,9 +110,9 @@ object ParseDescribe {
           }
         }
 
-        exchange("parse+describe", opDuration) { (span: Span[F]) =>
-          parseExchange(cmd, ty)(span).flatMap { case (preParse, postParse) =>
-            describeExchange(span).flatMap { case (preDesc, postDesc) =>
+        exchange("parse+describe") {
+          parseExchange(cmd, ty).flatMap { case (preParse, postParse) =>
+            describeExchange.flatMap { case (preDesc, postDesc) =>
               for {
                 id <- preParse
                 _  <- preDesc(id)
@@ -134,21 +128,19 @@ object ParseDescribe {
 
       override def apply[A, B](query: skunk.Query[A, B], ty: Typer): F[(StatementId, TypedRowDescription)] = {
 
-        def describeExchange(span: Span[F]): F[(StatementId => F[Unit], F[TypedRowDescription])] = {
-          def addStatementId(id: StatementId): F[Unit] =
-            span.addAttribute(Attribute("statement-id", id.value))
-
+        def describeExchange: F[(StatementId => F[Unit], F[TypedRowDescription])] = {
           def addColumnTypes(td: TypedRowDescription): F[Unit] =
-            span.addAttribute(Attribute("column-types", td.fields.map(_.tpe).mkString("[", ", ", "]")))
+            Telemetry[F].addProtocolAttributes(
+              SkunkAttributes.resultColumnTypes(td.fields.map(_.tpe).mkString("[", ", ", "]"))
+            )
 
           OptionT(cache.queryCache.get(query)).map { rd =>
-            val pre  = (id: StatementId) => addStatementId(id)
+            val pre  = (_: StatementId) => ().pure[F]
             val post = addColumnTypes(rd).as(rd)
             (pre, post)
           }.getOrElse {
             val pre = (id: StatementId) =>
               for {
-                _  <- addStatementId(id)
                 _  <- send(DescribeMessage.statement(id.value))
               } yield ()
 
@@ -172,9 +164,9 @@ object ParseDescribe {
         }
 
         
-        exchange("parse+describe", opDuration) { (span: Span[F]) => 
-          parseExchange(query, ty)(span).flatMap { case (preParse, postParse) =>
-            describeExchange(span).flatMap { case (preDesc, postDesc) =>
+        exchange("parse+describe") {
+          parseExchange(query, ty).flatMap { case (preParse, postParse) =>
+            describeExchange.flatMap { case (preDesc, postDesc) =>
               for {
                 id <- preParse
                 _  <- preDesc(id)
diff --git a/modules/core/shared/src/main/scala/net/protocol/Prepare.scala b/modules/core/shared/src/main/scala/net/protocol/Prepare.scala
index 39edde7f..8f9dc1c8 100644
--- a/modules/core/shared/src/main/scala/net/protocol/Prepare.scala
+++ b/modules/core/shared/src/main/scala/net/protocol/Prepare.scala
@@ -13,8 +13,7 @@ import skunk.net.MessageSocket
 import skunk.net.Protocol.{ PreparedCommand, PreparedQuery, CommandPortal, QueryPortal }
 import skunk.util.{ Origin, Namer }
 import skunk.util.Typer
-import org.typelevel.otel4s.trace.Tracer
-import org.typelevel.otel4s.metrics.Histogram
+import skunk.telemetry.Telemetry
 
 trait Prepare[F[_]] {
   def apply[A](command: skunk.Command[A], ty: Typer): F[PreparedCommand[F, A]]
@@ -23,31 +22,31 @@ trait Prepare[F[_]] {
 
 object Prepare {
 
-  def apply[F[_]: Exchange: MessageSocket: Namer: Tracer](describeCache: Describe.Cache[F], parseCache: Parse.Cache[F], redactionStrategy: RedactionStrategy, opDuration: Histogram[F, Double])(
+  def apply[F[_]: Exchange: MessageSocket: Namer: Telemetry](describeCache: Describe.Cache[F], parseCache: Parse.Cache[F], redactionStrategy: RedactionStrategy)(
     implicit ev: Concurrent[F]
   ): Prepare[F] =
     new Prepare[F] {
 
       override def apply[A](command: skunk.Command[A], ty: Typer): F[PreparedCommand[F, A]] =
-        ParseDescribe[F](describeCache, parseCache, opDuration).command(command, ty).map { id =>
+        ParseDescribe[F](describeCache, parseCache).command(command, ty).map { id =>
           new PreparedCommand[F, A](id, command) { pc =>
             def bind(args: A, origin: Origin): Resource[F, CommandPortal[F, A]] =
-              BindExecute[F](opDuration).command(this, args, origin, redactionStrategy)
+              BindExecute[F].command(this, args, origin, redactionStrategy)
           }
         }
 
       override def apply[A, B](query: skunk.Query[A, B], ty: Typer): F[PreparedQuery[F, A, B]] =
-        ParseDescribe[F](describeCache, parseCache, opDuration).apply(query, ty).map { case (id, rd) =>
+        ParseDescribe[F](describeCache, parseCache).apply(query, ty).map { case (id, rd) =>
           new PreparedQuery[F, A, B](id, query, rd) { pq =>
             def bind(args: A, origin: Origin): Resource[F, QueryPortal[F, A, B]] =
-              Bind[F](opDuration).apply(this, args, origin, redactionStrategy).map {
+              Bind[F].apply(this, args, origin, redactionStrategy).map {
                new QueryPortal[F, A, B](_, pq, args, origin, redactionStrategy) {
                   def execute(maxRows: Int): F[List[B] ~ Boolean] =
-                   Execute[F](opDuration).apply(this, maxRows)
+                   Execute[F].apply(this, maxRows)
                 }
               }
             def bindSized(args: A, origin: Origin, maxRows: Int): Resource[F, QueryPortal[F, A, B]] =
-              BindExecute[F](opDuration).query(this, args, origin, redactionStrategy, maxRows)
+              BindExecute[F].query(this, args, origin, redactionStrategy, maxRows)
           }
         }
 
diff --git a/modules/core/shared/src/main/scala/net/protocol/Query.scala b/modules/core/shared/src/main/scala/net/protocol/Query.scala
index 2cd42fea..50096e8e 100644
--- a/modules/core/shared/src/main/scala/net/protocol/Query.scala
+++ b/modules/core/shared/src/main/scala/net/protocol/Query.scala
@@ -12,11 +12,8 @@ import skunk.exception._
 import skunk.net.message.{ Query => QueryMessage, _ }
 import skunk.net.MessageSocket
 import skunk.util.Typer
-import org.typelevel.otel4s.semconv.attributes.DbAttributes
-import org.typelevel.otel4s.trace.Span
-import org.typelevel.otel4s.trace.Tracer
 import skunk.Statement
-import org.typelevel.otel4s.metrics.Histogram
+import skunk.telemetry.Telemetry
 
 trait Query[F[_]] {
   def apply(command: Command[Void]): F[Completion]
@@ -26,7 +23,7 @@ trait Query[F[_]] {
 
 object Query {
 
-  def apply[F[_]: Exchange: MessageSocket: Tracer](redactionStrategy: RedactionStrategy, opDuration: Histogram[F, Double])(
+  def apply[F[_]: Exchange: MessageSocket: Telemetry](redactionStrategy: RedactionStrategy)(
     implicit ev: MonadCancel[F, Throwable]
   ): Query[F] =
     new Unroll[F] with Query[F] {
@@ -71,10 +68,8 @@ object Query {
         }
 
       override def apply[B](query: skunk.Query[Void, B], ty: Typer): F[List[B]] =
-        exchange("query", opDuration) { (span: Span[F]) =>
-          span.addAttribute(
-            DbAttributes.DbQueryText(query.sql)
-          ) *> send(QueryMessage(query.sql)) *> flatExpect {
+        database("query", query, Nil, redactionStrategy) {
+          send(QueryMessage(query.sql)) *> flatExpect {
 
             // If we get a RowDescription back it means we have a valid query as far as Postgres is
             // concerned, and we will soon receive zero or more RowData followed by CommandComplete.
@@ -163,10 +158,8 @@ object Query {
         }
 
       override def apply(command: Command[Void]): F[Completion] =
-        exchange("query", opDuration) { (span: Span[F]) =>
-          span.addAttribute(
-            DbAttributes.DbQueryText(command.sql)
-          ) *> send(QueryMessage(command.sql)) *> flatExpect {
+        database("query", command, Nil, redactionStrategy) {
+          send(QueryMessage(command.sql)) *> flatExpect {
 
             case CommandComplete(c) =>
               finishUp(command).as(c)
@@ -246,10 +239,8 @@ object Query {
         }
 
       override def applyDiscard(statement: Statement[Void]): F[Unit] =
-        exchange("query", opDuration) { (span: Span[F]) =>
-          span.addAttribute(
-            DbAttributes.DbQueryText(statement.sql)
-          ) *> send(QueryMessage(statement.sql)) *> finishUpDiscard(statement, None)
+        database("query", statement, Nil, redactionStrategy) {
+          send(QueryMessage(statement.sql)) *> finishUpDiscard(statement, None)
         }
     }
 }
diff --git a/modules/core/shared/src/main/scala/net/protocol/Startup.scala b/modules/core/shared/src/main/scala/net/protocol/Startup.scala
index 81aea2ae..16c23d75 100644
--- a/modules/core/shared/src/main/scala/net/protocol/Startup.scala
+++ b/modules/core/shared/src/main/scala/net/protocol/Startup.scala
@@ -6,9 +6,6 @@ package skunk.net.protocol
 
 import cats.{ApplicativeError, MonadError, MonadThrow}
 import cats.syntax.all._
-import org.typelevel.otel4s.Attribute
-import org.typelevel.otel4s.trace.Span
-import org.typelevel.otel4s.trace.Tracer
 import scala.util.control.NonFatal
 import scodec.bits.ByteVector
 import skunk.net.MessageSocket
@@ -21,7 +18,7 @@ import skunk.exception.{
   UnsupportedAuthenticationSchemeException,
   UnsupportedSASLMechanismsException
 }
-import org.typelevel.otel4s.metrics.Histogram
+import skunk.telemetry.Telemetry
 import cats.effect.MonadCancel
 
 trait Startup[F[_]] {
@@ -30,18 +27,14 @@ trait Startup[F[_]] {
 
 object Startup {
 
-  def apply[F[_]: Exchange: MessageSocket: Tracer](opDuration: Histogram[F, Double])(
+  def apply[F[_]: Exchange: MessageSocket: Telemetry](
     implicit ev: MonadCancel[F, Throwable]
   ): Startup[F] =
     new Startup[F] {
       override def apply(user: String, database: String, password: Option[String], parameters: Map[String, String]): F[Unit] =
-        exchange("startup", opDuration) { (span: Span[F]) =>
+        exchange("startup") {
           val sm = StartupMessage(user, database, parameters)
           for {
-            _ <- span.addAttributes(
-              Attribute("user", user),
-              Attribute("database", database)
-            )
             _ <- send(sm)
             _ <- flatExpectStartup(sm) {
                     case AuthenticationOk                => ().pure[F]
@@ -63,13 +56,13 @@ object Startup {
     }
 
   // already inside an exchange
-  private def authenticationCleartextPassword[F[_]: MessageSocket: Tracer](
+  private def authenticationCleartextPassword[F[_]: MessageSocket: Telemetry](
     sm:       StartupMessage,
     password: Option[String]
   )(
     implicit ev: MonadError[F, Throwable]
   ): F[Unit] =
-    Tracer[F].span("authenticationCleartextPassword").surround {
+    Telemetry[F].internalSpan("authenticationCleartextPassword") {
       requirePassword[F](sm, password).flatMap { pw =>
         for {
           _ <- send(PasswordMessage.cleartext(pw))
@@ -78,14 +71,14 @@ object Startup {
       }
     }
 
-  private def authenticationMD5Password[F[_]: MessageSocket: Tracer](
+  private def authenticationMD5Password[F[_]: MessageSocket: Telemetry](
     sm:       StartupMessage,
     password: Option[String],
     salt:     Array[Byte]
   )(
     implicit ev: MonadError[F, Throwable]
   ): F[Unit] =
-    Tracer[F].span("authenticationMD5Password").surround {
+    Telemetry[F].internalSpan("authenticationMD5Password") {
       requirePassword[F](sm, password).flatMap { pw =>
         for {
           _ <- send(PasswordMessage.md5(sm.user, pw, salt))
@@ -94,12 +87,12 @@ object Startup {
       }
     }
 
-  private def authenticationSASL[F[_]: MonadThrow: MessageSocket: Tracer](
+  private def authenticationSASL[F[_]: MonadThrow: MessageSocket: Telemetry](
     sm:         StartupMessage,
     password:   Option[String],
     mechanisms: List[String]
   ): F[Unit] =
-    Tracer[F].span("authenticationSASL").surround {
+    Telemetry[F].internalSpan("authenticationSASL") {
       if (mechanisms.contains(Scram.SaslMechanism)) {
         for {
           pw <- requirePassword[F](sm, password)
diff --git a/modules/core/shared/src/main/scala/net/protocol/Unroll.scala b/modules/core/shared/src/main/scala/net/protocol/Unroll.scala
index c8d26766..e43affe9 100644
--- a/modules/core/shared/src/main/scala/net/protocol/Unroll.scala
+++ b/modules/core/shared/src/main/scala/net/protocol/Unroll.scala
@@ -14,16 +14,15 @@ import skunk.net.Protocol.QueryPortal
 import skunk.net.Protocol.PreparedQuery
 import skunk.util.Origin
 import skunk.data.TypedRowDescription
-import org.typelevel.otel4s.Attribute
-import org.typelevel.otel4s.trace.Tracer
 import skunk.exception.PostgresErrorException
+import skunk.telemetry.{SkunkAttributes, Telemetry}
 import scala.util.control.NonFatal
 
 /**
  * Superclass for `Query` and `Execute` sub-protocols, both of which need a way to accumulate
  * results in a `List` and report errors when decoding fails.
  */
-private[protocol] class Unroll[F[_]: MessageSocket: Tracer](
+private[protocol] class Unroll[F[_]: MessageSocket: Telemetry](
   implicit ev: MonadError[F, Throwable]
 ) {
 
@@ -103,17 +102,17 @@ private[protocol] class Unroll[F[_]: MessageSocket: Tracer](
       }
 
     val rows: F[(List[List[Option[String]]], Boolean)] =
-      Tracer[F].span("read").use { span =>
-        accumulate(Nil).flatTap { case (rows, bool) =>
-          span.addAttributes(
-            Attribute("row-count", rows.length.toLong),
-            Attribute("more-rows", bool)
+      Telemetry[F].internalSpan("read") {
+        accumulate(Nil).flatTap { case (rows, moreRows) =>
+          Telemetry[F].addProtocolAttributes(
+            SkunkAttributes.responseRowCount(rows.length.toLong),
+            SkunkAttributes.responseMoreRows(moreRows)
           )
         }
       }
 
     rows.flatMap { case (rows, bool) =>
-      Tracer[F].span("decode").surround {
+      Telemetry[F].internalSpan("decode") {
         rows.traverse { data =>
 
           // https://github.com/tpolecat/skunk/issues/129
diff --git a/modules/core/shared/src/main/scala/net/protocol/package.scala b/modules/core/shared/src/main/scala/net/protocol/package.scala
index e34b9dba..07018b51 100644
--- a/modules/core/shared/src/main/scala/net/protocol/package.scala
+++ b/modules/core/shared/src/main/scala/net/protocol/package.scala
@@ -7,29 +7,27 @@ package skunk.net
 import skunk.net.message._
 import skunk.util.Namer
 import skunk.util.Origin
-import skunk.util.Otel
-import org.typelevel.otel4s.trace.Span
-import org.typelevel.otel4s.trace.Tracer
-import org.typelevel.otel4s.trace.SpanKind
-import org.typelevel.otel4s.metrics.Histogram
-import java.util.concurrent.TimeUnit
-import cats.effect.MonadCancel
+import skunk.RedactionStrategy
+import skunk.Statement
+import skunk.data.Encoded
+import skunk.telemetry.Telemetry
 
 package object protocol {
 
-  def exchange[F[_]: Tracer, A](label: String, opDuration: Histogram[F, Double])(f: Span[F] => F[A])(
-    implicit exchange: Exchange[F], ev: MonadCancel[F, Throwable]
+  def exchange[F[_]: Telemetry, A](label: String)(fa: F[A])(
+    implicit exchange: Exchange[F]
   ): F[A] =
-    Tracer[F].spanBuilder(label)
-      .withSpanKind(SpanKind.Client)
-      .addAttribute(Otel.DbSystemName)
-      .withFinalizationStrategy(Otel.PostgresStrategy)
-      .build
-      .use{span =>
-        opDuration.recordDuration(TimeUnit.SECONDS, Otel.opDurationAttributes(_)).surround {
-          exchange(f(span))
-        }
-      }
+    Telemetry[F].internalSpan(label)(exchange(fa))
+
+  def database[F[_]: Telemetry, A](
+    label: String,
+    statement: Statement[_],
+    arguments: List[Option[Encoded]],
+    redactionStrategy: RedactionStrategy,
+  )(fa: F[A])(
+    implicit exchange: Exchange[F]
+  ): F[A] =
+    Telemetry[F].databaseSpan(label, statement, arguments, redactionStrategy)(exchange(fa))
 
   def receive[F[_]](implicit ev: MessageSocket[F]): F[BackendMessage] =
     ev.receive
diff --git a/modules/core/shared/src/main/scala/telemetry/ConnectionInfo.scala b/modules/core/shared/src/main/scala/telemetry/ConnectionInfo.scala
new file mode 100644
index 00000000..8f65c583
--- /dev/null
+++ b/modules/core/shared/src/main/scala/telemetry/ConnectionInfo.scala
@@ -0,0 +1,11 @@
+// Copyright (c) 2018-2024 by Rob Norris and Contributors
+// This software is licensed under the MIT License (MIT).
+// For more information see LICENSE or https://opensource.org/licenses/MIT
+
+package skunk.telemetry
+
+private[skunk] final case class ConnectionInfo(
+    database: String,
+    serverAddress: String,
+    serverPort: Option[Long]
+)
diff --git a/modules/core/shared/src/main/scala/telemetry/QueryAnalyzer.scala b/modules/core/shared/src/main/scala/telemetry/QueryAnalyzer.scala
new file mode 100644
index 00000000..447da64b
--- /dev/null
+++ b/modules/core/shared/src/main/scala/telemetry/QueryAnalyzer.scala
@@ -0,0 +1,82 @@
+// Copyright (c) 2018-2024 by Rob Norris and Contributors
+// This software is licensed under the MIT License (MIT).
+// For more information see LICENSE or https://opensource.org/licenses/MIT
+
+package skunk.telemetry
+
+/** Analyzes SQL text and returns semantic query metadata.
+  *
+  * Implementations must only return `queryText` when literals and other
+  * sensitive values have been removed. The summary must be low-cardinality and
+  * must not contain dynamic or sensitive values.
+  *
+  * In particular, analyzers may derive `querySummary` suitable for
+  * `db.query.summary` span naming/cardinality rules. Custom analyzers are
+  * created with [[QueryAnalyzer.apply]].
+  *
+  * @see
+  *   [[https://opentelemetry.io/docs/specs/semconv/db/database-spans/#generating-a-summary-of-the-query]]
+  */
+sealed trait QueryAnalyzer { self =>
+
+  /** Returns metadata for `sql`, or `None` when this analyzer cannot analyze it. */
+  def analyze(sql: String): Option[QueryAnalyzer.Analysis]
+
+  /** Uses `fallback` when this analyzer cannot analyze the query. */
+  final def orElse(fallback: QueryAnalyzer): QueryAnalyzer =
+    QueryAnalyzer { sql =>
+      self.analyze(sql).orElse(fallback.analyze(sql))
+    }
+
+}
+
+object QueryAnalyzer {
+
+  /** Query metadata. All fields are optional because analyzers may produce
+    * partial information.
+    */
+  sealed trait Analysis {
+
+    /** Sanitized SQL text with literals and other sensitive values removed. */
+    def queryText: Option[String]
+
+    /** Stored procedure name for procedure-style operations. */
+    def storedProcedureName: Option[String]
+
+    /** Low-cardinality summary suitable for `db.query.summary` span naming. */
+    def querySummary: Option[String]
+  }
+
+  object Analysis {
+
+    /** Creates query metadata. An analyzer may leave any field empty. */
+    def apply(
+        queryText: Option[String],
+        storedProcedureName: Option[String],
+        querySummary: Option[String]
+    ): Analysis =
+      QueryMetadataImpl(
+        queryText = queryText,
+        storedProcedureName = storedProcedureName,
+        querySummary = querySummary
+      )
+
+    private final case class QueryMetadataImpl(
+        queryText: Option[String],
+        storedProcedureName: Option[String],
+        querySummary: Option[String]
+    ) extends Analysis
+  }
+
+  /** Creates an analyzer from a function. */
+  def apply(f: String => Option[Analysis]): QueryAnalyzer =
+    Impl(f)
+
+  /** An analyzer that never produces query metadata. */
+  val noop: QueryAnalyzer = Impl(_ => None)
+
+  private final case class Impl(f: String => Option[Analysis]) extends QueryAnalyzer {
+    def analyze(sql: String): Option[Analysis] = f(sql)
+  }
+
+}
diff --git a/modules/core/shared/src/main/scala/telemetry/QueryCaptureConfig.scala b/modules/core/shared/src/main/scala/telemetry/QueryCaptureConfig.scala
new file mode 100644
index 00000000..7f11f0b8
--- /dev/null
+++ b/modules/core/shared/src/main/scala/telemetry/QueryCaptureConfig.scala
@@ -0,0 +1,86 @@
+// Copyright (c) 2018-2024 by Rob Norris and Contributors
+// This software is licensed under the MIT License (MIT).
+// For more information see LICENSE or https://opensource.org/licenses/MIT
+
+package skunk.telemetry
+
+/** Controls capture of query text and bound parameters. */
+sealed trait QueryCaptureConfig {
+  def queryTextPolicy: QueryCaptureConfig.QueryTextPolicy
+  def queryParametersPolicy: QueryCaptureConfig.QueryParametersPolicy
+
+  def withQueryTextPolicy(
+      queryTextPolicy: QueryCaptureConfig.QueryTextPolicy
+  ): QueryCaptureConfig
+
+  def withQueryParametersPolicy(
+      queryParametersPolicy: QueryCaptureConfig.QueryParametersPolicy
+  ): QueryCaptureConfig
+}
+
+object QueryCaptureConfig {
+
+  def apply(
+      queryTextPolicy: QueryTextPolicy,
+      queryParametersPolicy: QueryParametersPolicy
+  ): QueryCaptureConfig =
+    Impl(queryTextPolicy, queryParametersPolicy)
+
+  sealed trait QueryTextPolicy
+
+  object QueryTextPolicy {
+
+    /** Never record query text. */
+    case object None extends QueryTextPolicy
+
+    /**
+      * Record parameterized query text as-is. Record non-parameterized query text only when a
+      * configured [[QueryAnalyzer]] supplies a sanitized value.
+      */
+    case object SemconvRecommended extends QueryTextPolicy
+
+    /** Record all query text as-is. This may expose sensitive literal values. */
+    case object UnsafeAlways extends QueryTextPolicy
+
+  }
+
+  sealed trait QueryParametersPolicy
+
+  object QueryParametersPolicy {
+
+    /** Never record query parameters. */
+    case object None extends QueryParametersPolicy
+
+    /** Record all query parameters. */
+    case object All extends QueryParametersPolicy
+
+  }
+
+  /** Semantic-conventions-oriented defaults: safe query text and no parameter values. */
+  val recommended: QueryCaptureConfig =
+    QueryCaptureConfig(
+      QueryTextPolicy.SemconvRecommended,
+      QueryParametersPolicy.None
+    )
+
+  /** Query text and parameter capture are both disabled. */
+  val disabled: QueryCaptureConfig =
+    QueryCaptureConfig(
+      QueryTextPolicy.None,
+      QueryParametersPolicy.None
+    )
+
+  private final case class Impl(
+      queryTextPolicy: QueryTextPolicy,
+      queryParametersPolicy: QueryParametersPolicy
+  ) extends QueryCaptureConfig {
+    def withQueryTextPolicy(queryTextPolicy: QueryTextPolicy): QueryCaptureConfig =
+      copy(queryTextPolicy = queryTextPolicy)
+
+    def withQueryParametersPolicy(queryParametersPolicy: QueryParametersPolicy): QueryCaptureConfig =
+      copy(queryParametersPolicy = queryParametersPolicy)
+
+    override def toString: String =
+      s"QueryCaptureConfig($queryTextPolicy,$queryParametersPolicy)"
+  }
+}
diff --git a/modules/core/shared/src/main/scala/telemetry/SkunkAttributes.scala b/modules/core/shared/src/main/scala/telemetry/SkunkAttributes.scala
new file mode 100644
index 00000000..6c1f8fa3
--- /dev/null
+++ b/modules/core/shared/src/main/scala/telemetry/SkunkAttributes.scala
@@ -0,0 +1,30 @@
+// Copyright (c) 2018-2024 by Rob Norris and Contributors
+// This software is licensed under the MIT License (MIT).
+// For more information see LICENSE or https://opensource.org/licenses/MIT
+
+package skunk.telemetry
+
+import org.typelevel.otel4s.{Attribute, AttributeKey}
+
+private[skunk] object SkunkAttributes {
+
+  object Keys {
+    val OperationName = AttributeKey[String]("skunk.operation.name")
+    val PortalId = AttributeKey[String]("skunk.portal.id")
+    val StatementId = AttributeKey[String]("skunk.statement.id")
+    val StatementParameterTypes = AttributeKey[String]("skunk.statement.parameter_types")
+    val ResultColumnTypes = AttributeKey[String]("skunk.result.column_types")
+    val FetchMaxRows = AttributeKey[Long]("skunk.fetch.max_rows")
+    val ResponseRowCount = AttributeKey[Long]("skunk.response.row_count")
+    val ResponseMoreRows = AttributeKey[Boolean]("skunk.response.more_rows")
+  }
+
+  def operationName(value: String): Attribute[String] = Keys.OperationName(value)
+  def portalId(value: String): Attribute[String] = Keys.PortalId(value)
+  def statementId(value: String): Attribute[String] = Keys.StatementId(value)
+  def statementParameterTypes(value: String): Attribute[String] = Keys.StatementParameterTypes(value)
+  def resultColumnTypes(value: String): Attribute[String] = Keys.ResultColumnTypes(value)
+  def fetchMaxRows(value: Long): Attribute[Long] = Keys.FetchMaxRows(value)
+  def responseRowCount(value: Long): Attribute[Long] = Keys.ResponseRowCount(value)
+  def responseMoreRows(value: Boolean): Attribute[Boolean] = Keys.ResponseMoreRows(value)
+}
diff --git a/modules/core/shared/src/main/scala/telemetry/Telemetry.scala b/modules/core/shared/src/main/scala/telemetry/Telemetry.scala
new file mode 100644
index 00000000..bb436bc1
--- /dev/null
+++ b/modules/core/shared/src/main/scala/telemetry/Telemetry.scala
@@ -0,0 +1,290 @@
+// Copyright (c) 2018-2024 by Rob Norris and Contributors
+// This software is licensed under the MIT License (MIT).
+// For more information see LICENSE or https://opensource.org/licenses/MIT
+
+package skunk.telemetry
+
+import java.util.concurrent.TimeUnit
+
+import cats.arrow.FunctionK
+import cats.effect.{MonadCancelThrow, Resource}
+import cats.syntax.flatMap._
+import cats.syntax.functor._
+import cats.syntax.semigroup._
+import cats.~>
+import org.typelevel.otel4s.{Attribute, Attributes}
+import org.typelevel.otel4s.metrics.{BucketBoundaries, Histogram, Meter, MeterProvider}
+import org.typelevel.otel4s.semconv.attributes.{DbAttributes, ErrorAttributes, ServerAttributes}
+import org.typelevel.otel4s.semconv.metrics.DbMetrics
+import org.typelevel.otel4s.trace.{SpanFinalizer, SpanKind, StatusCode, Tracer, TracerProvider}
+import skunk.data.Encoded
+import skunk.exception.PostgresErrorException
+import skunk.{BuildInfo, RedactionStrategy, Statement}
+
+import scala.util.control.NonFatal
+
+sealed trait Telemetry[F[_]] {
+
+  private[skunk] def withConnection(connection: ConnectionInfo): Telemetry[F]
+
+  private[skunk] def poolSpan[A](name: String)(fa: F[A]): F[A]
+
+  private[skunk] def internalSpan[A](label: String)(fa: F[A]): F[A]
+
+  private[skunk] def databaseSpan[A](
+      operationName: String,
+      statement: Statement[_],
+      arguments: List[Option[Encoded]],
+      redactionStrategy: RedactionStrategy
+  )(fa: F[A]): F[A]
+
+  private[skunk] def addAttributes(attributes: Attribute[_]*): F[Unit]
+
+  private[skunk] def addProtocolAttributes(attributes: Attribute[_]*): F[Unit]
+
+}
+
+object Telemetry {
+
+  private val DbSystemName =
+    DbAttributes.DbSystemName(DbAttributes.DbSystemNameValue.Postgresql)
+
+  private val opDurationBoundaries =
+    BucketBoundaries(0.001d, 0.005d, 0.01d, 0.05d, 0.1d, 0.5d, 1d, 5d, 10d)
+
+  private[skunk] final case class ResolvedOperation(
+      spanName: String,
+      spanAttributes: Attributes,
+      metricAttributes: Attributes
+  )
+
+  def apply[F[_]](implicit ev: Telemetry[F]): Telemetry[F] = ev
+
+  def create[F[_]: MonadCancelThrow: TracerProvider: MeterProvider](
+      config: TelemetryConfig,
+      connection: ConnectionInfo
+  ): F[Telemetry[F]] =
+    MeterProvider[F].meter("org.typelevel.skunk").withVersion(BuildInfo.version).get.flatMap { implicit meter: Meter[F] =>
+      TracerProvider[F].tracer("org.typelevel.skunk").withVersion(BuildInfo.version).get.flatMap { implicit tracer: Tracer[F] =>
+        for {
+          operationDuration <- DbMetrics.ClientOperationDuration.create[F, Double](opDurationBoundaries)
+        } yield new Impl(config, connection, operationDuration)
+      }
+    }
+
+  private[skunk] final class Impl[F[_]: Tracer: MonadCancelThrow](
+      config: TelemetryConfig,
+      connection: ConnectionInfo,
+      operationDuration: Histogram[F, Double]
+  ) extends Telemetry[F] {
+
+    def withConnection(connection: ConnectionInfo): Telemetry[F] =
+      new Impl(config, connection, operationDuration)
+
+    private val finalizationStrategy: SpanFinalizer.Strategy = {
+      case Resource.ExitCase.Errored(e: PostgresErrorException) =>
+        val builder = Attributes.newBuilder
+
+        builder += DbAttributes.DbResponseStatusCode(e.code)
+        builder += ErrorAttributes.ErrorType(e.code)
+        builder ++= DbAttributes.DbCollectionName.maybe(e.tableName)
+
+        SpanFinalizer.recordException(e) |+|
+          SpanFinalizer.setStatus(StatusCode.Error) |+|
+          SpanFinalizer.addAttributes(builder.result())
+
+      case Resource.ExitCase.Errored(e) =>
+        SpanFinalizer.recordException(e) |+|
+          SpanFinalizer.setStatus(StatusCode.Error) |+|
+          SpanFinalizer.addAttribute(
+            ErrorAttributes.ErrorType(e.getClass.getName)
+          )
+
+      case Resource.ExitCase.Canceled =>
+        SpanFinalizer.setStatus(StatusCode.Error, "canceled") |+|
+          SpanFinalizer.addAttribute(ErrorAttributes.ErrorType("canceled"))
+
+    }
+
+    private val poolSpanF: String => F ~> F =
+      config.poolSpans match {
+        case TelemetryConfig.PoolSpans.Internal =>
+          label =>
+            FunctionK.liftFunction[F, F](
+              Tracer[F]
+                .spanBuilder(label)
+                .withSpanKind(SpanKind.Internal)
+                .build
+                .surround
+            )
+
+        case TelemetryConfig.PoolSpans.Disabled =>
+          Function.const(FunctionK.id[F])(_)
+      }
+
+    private val internalSpanF: String => F ~> F =
+      config.protocolSpans match {
+        case TelemetryConfig.ProtocolSpans.Internal =>
+          label =>
+            FunctionK.liftFunction[F, F](
+              Tracer[F]
+                .spanBuilder(label)
+                .withSpanKind(SpanKind.Internal)
+                .build
+                .surround
+            )
+
+        case TelemetryConfig.ProtocolSpans.Disabled =>
+          Function.const(FunctionK.id[F])(_)
+      }
+
+    def poolSpan[A](name: String)(fa: F[A]): F[A] =
+      poolSpanF(name)(fa)
+
+    def internalSpan[A](label: String)(fa: F[A]): F[A] =
+      internalSpanF(label)(fa)
+
+    def databaseSpan[A](
+        operationName: String,
+        statement: Statement[_],
+        arguments: List[Option[Encoded]],
+        redactionStrategy: RedactionStrategy
+    )(fa: F[A]): F[A] = {
+      val resolved = resolveOperation(
+        operationName,
+        statement,
+        arguments,
+        redactionStrategy,
+        config,
+        connection
+      )
+
+      Tracer[F]
+        .spanBuilder(resolved.spanName)
+        .withSpanKind(SpanKind.Client)
+        .addAttributes(resolved.spanAttributes)
+        .withFinalizationStrategy(finalizationStrategy)
+        .build
+        .surround {
+          val attributes = operationDurationAttributes(resolved)(_)
+          operationDuration
+            .recordDuration(TimeUnit.SECONDS, attributes)
+            .surround(fa)
+        }
+    }
+
+    private[skunk] def addAttributes(attributes: Attribute[_]*): F[Unit] =
+      Tracer[F].withCurrentSpanOrNoop(_.addAttributes(attributes))
+
+    private[skunk] def addProtocolAttributes(attributes: Attribute[_]*): F[Unit] =
+      config.protocolSpans match {
+        case TelemetryConfig.ProtocolSpans.Internal => addAttributes(attributes: _*)
+        case TelemetryConfig.ProtocolSpans.Disabled => MonadCancelThrow[F].unit
+      }
+
+    private def operationDurationAttributes(
+        operation: ResolvedOperation
+    )(exitCase: Resource.ExitCase): Attributes = {
+      val builder = Attributes.newBuilder
+
+      builder ++= operation.metricAttributes
+
+      exitCase match {
+        case Resource.ExitCase.Succeeded =>
+
+        case Resource.ExitCase.Errored(e: PostgresErrorException) =>
+          builder += ErrorAttributes.ErrorType(e.code)
+          builder += DbAttributes.DbResponseStatusCode(e.code)
+          builder ++= DbAttributes.DbCollectionName.maybe(e.tableName)
+
+        case Resource.ExitCase.Errored(e) =>
+          builder += ErrorAttributes.ErrorType(e.getClass().getName())
+
+        case Resource.ExitCase.Canceled =>
+          builder += ErrorAttributes.ErrorType("canceled")
+
+      }
+
+      builder.result()
+    }
+  }
+
+  private[skunk] def resolveOperation(
+      operationName: String,
+      statement: Statement[_],
+      arguments: List[Option[Encoded]],
+      redactionStrategy: RedactionStrategy,
+      config: TelemetryConfig,
+      connection: ConnectionInfo
+  ): ResolvedOperation = {
+    val analysis =
+      try config.queryAnalyzer.analyze(statement.sql)
+      catch { case NonFatal(_) => None }
+
+    val summary =
+      statement.telemetry.querySummary
+        .orElse(
+          analysis
+            .flatMap(_.querySummary)
+            .filter(_.trim.nonEmpty)
+            .map(truncateSummary)
+        )
+
+    val storedProcedure =
+      analysis.flatMap(_.storedProcedureName).filter(_.nonEmpty)
+
+    val common = Attributes.newBuilder
+    common += DbSystemName
+    common += DbAttributes.DbNamespace(connection.database)
+    common += SkunkAttributes.operationName(operationName)
+    common += ServerAttributes.ServerAddress(connection.serverAddress)
+    connection.serverPort.foreach(p => common += ServerAttributes.ServerPort(p))
+    summary.foreach(s => common += DbAttributes.DbQuerySummary(s))
+    storedProcedure.foreach(p =>
+      common += DbAttributes.DbStoredProcedureName(p)
+    )
+
+    val metricAttributes = common.result()
+    val span = Attributes.newBuilder
+    span ++= statement.telemetry.attributes
+    span ++= metricAttributes
+
+    val isParameterized = statement.encoder.types.nonEmpty
+    val queryTextPolicy = config.captureQuery.queryTextPolicy
+    if (queryTextPolicy == QueryCaptureConfig.QueryTextPolicy.SemconvRecommended) {
+      val queryText =
+        if (isParameterized) Some(statement.sql)
+        else analysis.flatMap(_.queryText)
+      queryText
+        .filter(_.nonEmpty)
+        .foreach(q => span += DbAttributes.DbQueryText(q))
+    } else if (queryTextPolicy == QueryCaptureConfig.QueryTextPolicy.UnsafeAlways) {
+      span += DbAttributes.DbQueryText(statement.sql)
+    }
+
+    config.captureQuery.queryParametersPolicy match {
+      case QueryCaptureConfig.QueryParametersPolicy.All =>
+        redactionStrategy.redactArguments(arguments).zipWithIndex.foreach {
+          case (argument, index) =>
+            val value = argument.fold("NULL")(_.toString)
+            span.addOne(s"db.query.parameter.$index", value)
+        }
+      case _ =>
+    }
+
+    ResolvedOperation(
+      spanName = summary.orElse(storedProcedure).getOrElse(operationName),
+      spanAttributes = span.result(),
+      metricAttributes = metricAttributes
+    )
+  }
+
+  private def truncateSummary(summary: String): String =
+    if (summary.length <= 255) summary
+    else {
+      val lastSpace = summary.lastIndexOf(' ', 255)
+      if (lastSpace > 0) summary.substring(0, lastSpace)
+      else summary.substring(0, 255)
+    }
+
+}
diff --git a/modules/core/shared/src/main/scala/telemetry/TelemetryConfig.scala b/modules/core/shared/src/main/scala/telemetry/TelemetryConfig.scala
new file mode 100644
index 00000000..5bb619d1
--- /dev/null
+++ b/modules/core/shared/src/main/scala/telemetry/TelemetryConfig.scala
@@ -0,0 +1,94 @@
+// Copyright (c) 2018-2024 by Rob Norris and Contributors
+// This software is licensed under the MIT License (MIT).
+// For more information see LICENSE or https://opensource.org/licenses/MIT
+
+package skunk.telemetry
+
+/** Configures traces and metrics emitted by Skunk.
+  *
+  * Start with [[TelemetryConfig.default]] and use the `with` methods to change individual settings.
+  */
+sealed trait TelemetryConfig {
+  def captureQuery: QueryCaptureConfig
+  def queryAnalyzer: QueryAnalyzer
+  def poolSpans: TelemetryConfig.PoolSpans
+  def protocolSpans: TelemetryConfig.ProtocolSpans
+
+  /** Changes query text and parameter capture. */
+  def withCaptureQuery(captureQuery: QueryCaptureConfig): TelemetryConfig
+
+  /** Changes the analyzer used to derive sanitized query metadata. */
+  def withQueryAnalyzer(queryAnalyzer: QueryAnalyzer): TelemetryConfig
+
+  /** Enables or disables connection-pool spans. */
+  def withPoolSpans(poolSpans: TelemetryConfig.PoolSpans): TelemetryConfig
+
+  /** Enables or disables PostgreSQL wire-protocol spans. */
+  def withProtocolSpans(protocolSpans: TelemetryConfig.ProtocolSpans): TelemetryConfig
+}
+
+object TelemetryConfig {
+
+  /** Controls spans emitted by Skunk's connection pool. */
+  sealed trait PoolSpans
+  object PoolSpans {
+
+    /** Emit connection-pool operations as `INTERNAL` spans. */
+    case object Internal extends PoolSpans
+
+    /** Do not export connection-pool spans. */
+    case object Disabled extends PoolSpans
+  }
+
+  /** Controls spans emitted for PostgreSQL wire-protocol operations. */
+  sealed trait ProtocolSpans
+  object ProtocolSpans {
+
+    /** Emit PostgreSQL wire-protocol details as `INTERNAL` spans. */
+    case object Internal extends ProtocolSpans
+
+    /** Do not export PostgreSQL wire-protocol spans. */
+    case object Disabled extends ProtocolSpans
+  }
+
+  /** Recommended defaults. Query capture is safe, protocol spans are enabled, pool spans are
+    * disabled, and no query analyzer is installed.
+    */
+  val default: TelemetryConfig = TelemetryConfig(
+    QueryCaptureConfig.recommended,
+    QueryAnalyzer.noop,
+    PoolSpans.Disabled,
+    ProtocolSpans.Internal
+  )
+
+  /** Creates a telemetry configuration with explicit settings. */
+  def apply(
+      captureQuery: QueryCaptureConfig,
+      queryAnalyzer: QueryAnalyzer,
+      poolSpans: PoolSpans,
+      protocolSpans: ProtocolSpans
+  ): TelemetryConfig =
+    Impl(captureQuery, queryAnalyzer, poolSpans, protocolSpans)
+
+  private final case class Impl(
+      captureQuery: QueryCaptureConfig,
+      queryAnalyzer: QueryAnalyzer,
+      poolSpans: PoolSpans,
+      protocolSpans: ProtocolSpans
+  ) extends TelemetryConfig {
+    def withCaptureQuery(captureQuery: QueryCaptureConfig): TelemetryConfig =
+      copy(captureQuery = captureQuery)
+
+    def withQueryAnalyzer(queryAnalyzer: QueryAnalyzer): TelemetryConfig =
+      copy(queryAnalyzer = queryAnalyzer)
+
+    def withPoolSpans(poolSpans: PoolSpans): TelemetryConfig =
+      copy(poolSpans = poolSpans)
+
+    def withProtocolSpans(protocolSpans: ProtocolSpans): TelemetryConfig =
+      copy(protocolSpans = protocolSpans)
+
+    override def toString: String =
+      s"TelemetryConfig($captureQuery, $queryAnalyzer, $poolSpans, $protocolSpans)"
+  }
+}
diff --git a/modules/core/shared/src/main/scala/util/Otel.scala b/modules/core/shared/src/main/scala/util/Otel.scala
deleted file mode 100644
index e77a6a05..00000000
--- a/modules/core/shared/src/main/scala/util/Otel.scala
+++ /dev/null
@@ -1,75 +0,0 @@
-// Copyright (c) 2018-2024 by Rob Norris and Contributors
-// This software is licensed under the MIT License (MIT).
-// For more information see LICENSE or https://opensource.org/licenses/MIT
-
-package skunk.util
-
-import cats.effect.Resource
-import cats.syntax.semigroup._
-import org.typelevel.otel4s.Attributes
-import org.typelevel.otel4s.semconv.attributes.DbAttributes
-import org.typelevel.otel4s.trace.SpanFinalizer
-import org.typelevel.otel4s.trace.StatusCode
-import skunk.exception.PostgresErrorException
-import org.typelevel.otel4s.semconv.attributes.ErrorAttributes
-import org.typelevel.otel4s.metrics.Meter
-import org.typelevel.otel4s.semconv.metrics.DbMetrics
-import org.typelevel.otel4s.metrics.BucketBoundaries
-import org.typelevel.otel4s.metrics.Histogram
-
-object Otel {
-
-  val DbSystemName = DbAttributes.DbSystemName(DbAttributes.DbSystemNameValue.Postgresql)
-
-  // Similar to the default reportAbnormal strategy but records some
-  // postgresql specific attributes in case it is a postgres error
-  val PostgresStrategy: SpanFinalizer.Strategy = {
-    case Resource.ExitCase.Errored(e: PostgresErrorException) =>
-      val builder = Attributes.newBuilder
-
-      builder += DbAttributes.DbResponseStatusCode(e.code)
-      builder ++= DbAttributes.DbCollectionName.maybe(e.tableName)
-      builder ++= DbAttributes.DbNamespace.maybe(e.schemaName)
-
-      SpanFinalizer.recordException(e) |+|
-        SpanFinalizer.setStatus(StatusCode.Error) |+|
-        SpanFinalizer.addAttributes(builder.result())
-
-    case Resource.ExitCase.Errored(e) =>
-      SpanFinalizer.recordException(e) |+| SpanFinalizer.setStatus(StatusCode.Error)
-
-    case Resource.ExitCase.Canceled =>
-      SpanFinalizer.setStatus(StatusCode.Error, "canceled")
-
-  }
-
-  private val opDurationBoundaries = BucketBoundaries(0.001d, 0.005d, 0.01d, 0.05d, 0.1d, 0.5d, 1d, 5d, 10d)
-
-  def OpDurationHistogram[F[_]: Meter]: F[Histogram[F, Double]] =
-    DbMetrics.ClientOperationDuration.create[F, Double](opDurationBoundaries)
-
-  def opDurationAttributes(exitCase: Resource.ExitCase): Attributes = {
-    val builder = Attributes.newBuilder
-
-    builder += DbSystemName
-
-    exitCase match {
-      case Resource.ExitCase.Succeeded =>
-
-      case Resource.ExitCase.Errored(e: PostgresErrorException) =>
-        builder += ErrorAttributes.ErrorType(e.getClass().getName())
-        builder += DbAttributes.DbResponseStatusCode(e.code)
-        builder ++= DbAttributes.DbCollectionName.maybe(e.tableName)
-        builder ++= DbAttributes.DbNamespace.maybe(e.schemaName)
-
-      case Resource.ExitCase.Errored(e) =>
-        builder += ErrorAttributes.ErrorType(e.getClass().getName())
-
-      case Resource.ExitCase.Canceled =>
-
-    }
-
-    builder.result()
-  }
-
-}
diff --git a/modules/core/shared/src/main/scala/util/Pool.scala b/modules/core/shared/src/main/scala/util/Pool.scala
index 6f0191ec..4a02ee78 100644
--- a/modules/core/shared/src/main/scala/util/Pool.scala
+++ b/modules/core/shared/src/main/scala/util/Pool.scala
@@ -12,7 +12,7 @@ import cats.effect.implicits._
 import cats.effect.Resource
 import cats.syntax.all._
 import skunk.exception.SkunkException
-import org.typelevel.otel4s.trace.Tracer
+import skunk.telemetry.Telemetry
 
 object Pool {
 
@@ -45,11 +45,11 @@ object Pool {
 
   // Preserved for previous use, and specifically simpler use for
   // Tracer systems that are universal rather than shorter scoped.
-  def of[F[_]: Concurrent: Tracer, A](
+  def of[F[_]: Concurrent: Telemetry, A](
     rsrc: Resource[F, A],
     size:  Int)(
     recycler: Recycler[F, A]
-  ): Resource[F, Resource[F, A]] = ofF({(_: Tracer[F]) => rsrc}, size)(recycler).map(_.apply(Tracer[F]))
+  ): Resource[F, Resource[F, A]] = ofF({(_: Telemetry[F]) => rsrc}, size)(recycler).map(_.apply(Telemetry[F]))
 
   /**
    * A pooled resource (which is itself a managed resource).
@@ -59,10 +59,10 @@ object Pool {
    *   yielding false here means the element should be freed and removed from the pool.
    */
   def ofF[F[_]: Concurrent, A](
-    rsrc:  Tracer[F] => Resource[F, A],
+    rsrc:  Telemetry[F] => Resource[F, A],
     size:  Int)(
     recycler: Recycler[F, A]
-  ): Resource[F, Tracer[F] => Resource[F, A]] = {
+  ): Resource[F, Telemetry[F] => Resource[F, A]] = {
 
     // Just in case.
     assert(size > 0, s"Pool size must be positive (you passed $size).")
@@ -78,14 +78,14 @@ object Pool {
     )
 
     // We can construct a pool given a Ref containing our initial state.
-    def poolImpl(ref: Ref[F, State])(implicit T: Tracer[F]): Resource[F, A] = {
+    def poolImpl(ref: Ref[F, State])(implicit T: Telemetry[F]): Resource[F, A] = {
 
       // To give out an alloc we create a deferral first, which we will need if there are no slots
       // available. If there is a filled slot, remove it and yield its alloc. If there is an empty
       // slot, remove it and allocate. If there are no slots, enqueue the deferral and wait on it,
       // which will [semantically] block the caller until an alloc is returned to the pool.
       def give(poll: Poll[F]): F[Alloc] =
-        Tracer[F].span("pool.allocate").surround {
+        Telemetry[F].poolSpan("pool.allocate") {
           Deferred[F, Either[Throwable, Alloc]].flatMap { d =>
 
             // If allocation fails for any reason then there's no resource to return to the pool
@@ -99,7 +99,7 @@ object Pool {
             // all (defer and wait).
             ref.modify {
               case (Some(a) :: os, ds) => ((os, ds), a.pure[F])
-              case (None    :: os, ds) => ((os, ds), Concurrent[F].onError(rsrc(Tracer[F]).allocated)(restore))
+              case (None    :: os, ds) => ((os, ds), Concurrent[F].onError(rsrc(Telemetry[F]).allocated)(restore))
               case (Nil,           ds) =>
                 val cancel = ref.flatModify { // try to remove our deferred
                   case (os, ds) =>
@@ -126,7 +126,7 @@ object Pool {
       // there are a bunch of error conditions to consider. This operation is a finalizer and
       // cannot be canceled, so we don't need to worry about that case here.
       def take(a: Alloc): F[Unit] =
-        Tracer[F].span("pool.free").surround {
+        Telemetry[F].poolSpan("pool.free") {
           recycler(a._1).onError { case _ => dispose(a) } flatMap {
             case true  => recycle(a)
             case false => dispose(a)
@@ -136,7 +136,7 @@ object Pool {
       // Return `a` to the pool. If there are awaiting deferrals, complete the next one. Otherwise
       // push a filled slot into the queue.
       def recycle(a: Alloc): F[Unit] =
-        Tracer[F].span("recycle").surround {
+        Telemetry[F].poolSpan("recycle") {
           ref.modify {
             case (os, d :: ds) => ((os, ds), d.complete(a.asRight).void)  // hand it back out
             case (os, Nil)     => ((Some(a) :: os, Nil), ().pure[F]) // return to pool
@@ -148,10 +148,10 @@ object Pool {
       // of `a`. If there are deferrals, remove the next one and complete it (failures in allocation
       // are handled by the awaiting deferral in `give` above). Always finalize `a`
       def dispose(a: Alloc): F[Unit] =
-        Tracer[F].span("dispose").surround {
+        Telemetry[F].poolSpan("dispose") {
           ref.modify {
             case (os, Nil) =>  ((os :+ None, Nil), ().pure[F]) // new empty slot
-            case (os, d :: ds) =>  ((os, ds), Concurrent[F].attempt(rsrc(Tracer[F]).allocated).flatMap(d.complete).void) // alloc now!
+            case (os, d :: ds) =>  ((os, ds), Concurrent[F].attempt(rsrc(Telemetry[F]).allocated).flatMap(d.complete).void) // alloc now!
           }.flatMap(next => a._2.guarantee(next)) // first finalize the original alloc then potentially do new alloc
         }
 
@@ -184,7 +184,7 @@ object Pool {
 
       }
 
-    Resource.make(alloc)(free).map(a => {implicit T: Tracer[F] => poolImpl(a)})
+    Resource.make(alloc)(free).map(a => {implicit T: Telemetry[F] => poolImpl(a)})
 
   }
 
diff --git a/modules/docs/src/main/laika/tutorial/Telemetry.md b/modules/docs/src/main/laika/tutorial/Telemetry.md
index d8fe22be..ac958d91 100644
--- a/modules/docs/src/main/laika/tutorial/Telemetry.md
+++ b/modules/docs/src/main/laika/tutorial/Telemetry.md
@@ -1,29 +1,205 @@
+```scala mdoc:invisible
+import cats.effect._
+import org.typelevel.otel4s.metrics.MeterProvider
+import org.typelevel.otel4s.trace.TracerProvider
+import skunk._
+import skunk.codec.all._
+import skunk.implicits._
+
+implicit val tracerProvider: TracerProvider[IO] = TracerProvider.noop
+implicit val meterProvider: MeterProvider[IO] = MeterProvider.noop
+```
+
 # Telemetry
 
-Skunk uses [OpenTelemetry](https://opentelemetry.io/), using the [otel4s](https://github.com/typelevel/otel4s). Session construction requires a `TracerProvider` and a `MeterProvider`; Skunk uses them to acquire its own versioned tracer and meter.
+Skunk emits OpenTelemetry database spans and metrics through
+[otel4s](https://github.com/typelevel/otel4s). The instrumentation scope is
+`org.typelevel.skunk`, with the Skunk version recorded as the scope version.
+
+`Session.Builder` uses `TelemetryConfig.default` unless another configuration is supplied with
+`withTelemetryConfig`. The default is the recommended starting point.
+
+## Default behavior
+
+The default configuration provides the following behavior:
+
+- Simple statements, prepared statements, and cursor fetches emit semantic database `CLIENT` spans.
+- Each database span has a matching
+  [`db.client.operation.duration`](https://opentelemetry.io/docs/specs/semconv/db/database-metrics/#metric-dbclientoperationduration)
+  histogram observation.
+- PostgreSQL wire-protocol operations such as parse, bind, read, decode, and close emit diagnostic
+  `INTERNAL` spans.
+- Connection-pool spans are disabled.
+- Parameterized query text is recorded because it does not contain bound values.
+- Literal query text and query parameter values are not recorded.
+- Query analysis is disabled. Skunk uses `QueryAnalyzer.noop` by default.
+
+The configured `TracerProvider` and `MeterProvider` still decide whether signals are recorded or
+exported. Supplying no-op providers disables tracing and metrics without changing
+`TelemetryConfig`.
+
+## Database spans and metrics
+
+A database span and its duration measurement cover the same logical operation. Both include the
+usual PostgreSQL attributes, such as `db.system.name`, `db.namespace`, and `server.address`.
+
+Skunk also retains its low-cardinality execution label as `skunk.operation.name` (`query`,
+`bind+execute`, or `execute`). These values describe Skunk's execution path, so Skunk does not
+record them as `db.operation.name`.
+
+Skunk records the connected database as the PostgreSQL namespace. It does not issue an additional
+query to discover the session's current schema.
+
+## Set a database span name
+
+The most direct way to name a database span is to add a summary to a `Query` or `Command`:
+
+```scala mdoc:silent
+val findCountry =
+  sql"SELECT name FROM country WHERE code = $varchar"
+    .query(varchar)
+    .withQuerySummary("SELECT country")
+```
+
+The resulting span is named `SELECT country` and has `db.query.summary` set to the same value. The
+summary should be low-cardinality and stable across calls. Do not include IDs, argument values, or
+other request-specific data.
+
+`Command` supports the same method:
+
+```scala mdoc:silent
+val updateCountry =
+  sql"UPDATE country SET name = $varchar WHERE code = $varchar"
+    .command
+    .withQuerySummary("UPDATE country")
+```
+
+A summary attached to a statement takes precedence over a summary returned by a
+`QueryAnalyzer`. If neither is available, Skunk uses its execution label as the span name.
+
+## Add statement attributes
+
+Statements can carry additional attributes for the logical database span:
+
+```scala mdoc:silent
+import org.typelevel.otel4s.AttributeKey
+
+val QueryCategory = AttributeKey[String]("app.query.category")
+
+val findCountryWithMetadata =
+  findCountry.addAttributes(QueryCategory("lookup"))
+```
+
+Application-defined attributes are not added to `db.client.operation.duration`. Avoid sensitive or
+high-cardinality values. A string `db.query.summary` attribute is also used for span naming and the
+operation duration metric. Attributes are accepted as-is; attributes managed by Skunk may overwrite
+values with the same keys when the span is resolved.
+
+## Configure auxiliary spans
+
+Protocol spans are enabled by default. They describe Skunk's work below the logical database
+operation and use `SpanKind.INTERNAL`. Disable them when traces should contain only database client
+operations:
+
+```scala mdoc:silent
+import skunk.telemetry.TelemetryConfig
+
+val protocolTelemetry = TelemetryConfig.default
+  .withProtocolSpans(TelemetryConfig.ProtocolSpans.Disabled)
+```
+
+Pool spans are disabled by default. They can be enabled when investigating connection acquisition
+or pool cleanup:
+
+```scala mdoc:silent
+val poolTelemetry = TelemetryConfig.default
+  .withPoolSpans(TelemetryConfig.PoolSpans.Internal)
+```
+
+To state explicitly that neither category should be emitted:
+
+```scala mdoc:silent
+val minimalTelemetry = TelemetryConfig.default
+  .withProtocolSpans(TelemetryConfig.ProtocolSpans.Disabled)
+  .withPoolSpans(TelemetryConfig.PoolSpans.Disabled)
+
+val sessions =
+  Session.Builder[IO]
+    .withTelemetryConfig(minimalTelemetry)
+```
+
+Disabling these spans does not disable semantic database spans or metrics.
+
+## Configure query capture
+
+`QueryCaptureConfig.recommended` is part of the default telemetry configuration:
+
+- Parameterized query text is recorded because bound values are not present in it.
+- Non-parameterized query text is omitted unless a configured analyzer supplies sanitized text.
+- Query parameter values are never recorded by default.
+
+Parameter values require an explicit global opt-in:
+
+```scala mdoc:silent
+import skunk.telemetry.{ QueryCaptureConfig, TelemetryConfig }
+
+val parameterCaptureTelemetry = TelemetryConfig.default.withCaptureQuery(
+  QueryCaptureConfig.recommended.withQueryParametersPolicy(
+    QueryCaptureConfig.QueryParametersPolicy.All
+  )
+)
 
-## Metrics
+val session =
+  Session.Builder[IO]
+    .withTelemetryConfig(parameterCaptureTelemetry)
+```
 
-Skunk provides the [`db.client.operation.duration`](https://opentelemetry.io/docs/specs/semconv/db/database-metrics/#metric-dbclientoperationduration) histogram, that records the duration of all operations interacting with the postgresql server.
+This records all bound parameters for every statement using the configuration. Captured values
+still respect the session's `RedactionStrategy` and encoder-level redaction. Parameter attributes
+are emitted only on spans, never as metric dimensions. Enable this only in controlled environments
+where exporting parameter values is acceptable.
 
-### I Don't Care
+Disable query text and parameter capture together with `QueryCaptureConfig.disabled`:
 
-If you don't care about metrics you can use the **no-op meter** to disable metrics entirely (`org.typelevel.otel4s.metrics.MeterProvider.noop`).
+```scala mdoc:silent
+val disabledCaptureTelemetry = TelemetryConfig.default
+  .withCaptureQuery(QueryCaptureConfig.disabled)
+```
 
-## Tracing
+`QueryTextPolicy.UnsafeAlways` records SQL containing literal values. Use it only when that data is
+safe to export.
 
-### I Don't Care
+## Custom query analyzers
 
-If you don't care about tracing you have two choices:
+Applications can provide an analyzer that supplies sanitized query text, a low-cardinality summary,
+and a stored-procedure name. For example, an application with a small set of known statements can
+define explicit metadata:
 
-- Use the **no-op tracer** to disable tracing entirely (`org.typelevel.otel4s.TracerProvider.noop`).
-- Use the **log tracer** to log completed traces to a [log4cats](https://typelevel.org/log4cats/) logger.
+```scala mdoc:silent
+import skunk.telemetry.QueryAnalyzer
+import skunk.telemetry.TelemetryConfig
 
-### Tracing with Jaeger
+val analyzer = QueryAnalyzer {
+  case "SELECT name FROM country WHERE code = 'GBR'" =>
+    Some(QueryAnalyzer.Analysis(
+      queryText = Some("SELECT name FROM country WHERE code = ?"),
+      storedProcedureName = None,
+      querySummary = Some("SELECT country")
+    ))
+  case _ =>
+    None
+}
 
-Easy because there's a docker container.
+val analyzerTelemetry = TelemetryConfig.default.withQueryAnalyzer(analyzer)
+```
 
-### Tracing Example
+Only return `queryText` after removing literals and other sensitive values. Query summaries must be
+low-cardinality and must not contain request data. Analyzer failures are ignored so they cannot
+fail a database operation.
 
-... program that adds its own events to the trace
+## Skunk-specific attributes
 
+Logical database spans may include `skunk.portal.id`, `skunk.statement.id`, and
+`skunk.fetch.max_rows`. Protocol spans may include statement parameter types, result column types,
+row counts, and whether more rows are available. Protocol spans do not contain raw SQL or encoded
+argument values.
diff --git a/modules/tests/jvm/src/test/scala/TelemetryIntegrationTest.scala b/modules/tests/jvm/src/test/scala/TelemetryIntegrationTest.scala
new file mode 100644
index 00000000..65c32914
--- /dev/null
+++ b/modules/tests/jvm/src/test/scala/TelemetryIntegrationTest.scala
@@ -0,0 +1,424 @@
+// Copyright (c) 2018-2024 by Rob Norris and Contributors
+// This software is licensed under the MIT License (MIT).
+// For more information see LICENSE or https://opensource.org/licenses/MIT
+
+package skunk
+
+import cats.effect.IO
+import cats.effect.Resource
+import munit.CatsEffectSuite
+import org.typelevel.otel4s.Attribute
+import org.typelevel.otel4s.metrics.BucketBoundaries
+import org.typelevel.otel4s.metrics.MeterProvider
+import org.typelevel.otel4s.sdk.metrics.data.MetricData
+import org.typelevel.otel4s.sdk.testkit.InstrumentationScopeExpectation
+import org.typelevel.otel4s.sdk.testkit.OpenTelemetrySdkTestkit
+import org.typelevel.otel4s.sdk.testkit.metrics.MetricExpectation
+import org.typelevel.otel4s.sdk.testkit.metrics.MetricExpectations
+import org.typelevel.otel4s.sdk.testkit.metrics.PointExpectation
+import org.typelevel.otel4s.sdk.testkit.trace.SpanExpectation
+import org.typelevel.otel4s.sdk.testkit.trace.StatusExpectation
+import org.typelevel.otel4s.sdk.testkit.trace.TraceExpectation
+import org.typelevel.otel4s.sdk.testkit.trace.TraceExpectations
+import org.typelevel.otel4s.sdk.testkit.trace.TraceForestExpectation
+import org.typelevel.otel4s.sdk.trace.data.SpanData
+import org.typelevel.otel4s.trace.TracerProvider
+import skunk.codec.all.int4
+import skunk.codec.all.varchar
+import skunk.implicits._
+import skunk.telemetry.QueryCaptureConfig
+import skunk.telemetry.TelemetryConfig
+
+class TelemetryIntegrationTest extends CatsEffectSuite {
+
+  private val ScopeName = "org.typelevel.skunk"
+  private val ScopeVersion = BuildInfo.version
+  private val DurationMetricName = "db.client.operation.duration"
+  private val DurationBoundaries =
+    BucketBoundaries(0.001d, 0.005d, 0.01d, 0.05d, 0.1d, 0.5d, 1d, 5d, 10d)
+
+  private val scope =
+    InstrumentationScopeExpectation
+      .name(ScopeName)
+      .version(ScopeVersion)
+      .attributesEmpty
+
+  private def session(
+      config: TelemetryConfig
+  ): Resource[IO, (Session[IO], OpenTelemetrySdkTestkit[IO])] =
+    OpenTelemetrySdkTestkit.inMemory[IO]().flatMap { testkit =>
+      implicit val T: TracerProvider[IO] = testkit.tracerProvider
+      implicit val M: MeterProvider[IO] = testkit.meterProvider
+      Session.Builder[IO]
+        .withUserAndPassword("jimmy", "banana")
+        .withDatabase("world")
+        .withTelemetryConfig(config)
+        .single
+        .map((_, testkit))
+    }
+
+  private def commonAttributes(
+      summary: String,
+      operationName: String
+  ): List[Attribute[_]] =
+    List(
+      Attribute("db.system.name", "postgresql"),
+      Attribute("db.namespace", "world"),
+      Attribute("server.address", "localhost"),
+      Attribute("db.query.summary", summary),
+      Attribute("skunk.operation.name", operationName)
+    )
+
+  private def attributeNames(span: SpanData): Set[String] =
+    span.attributes.elements.iterator.map(_.key.name).toSet
+
+  private def stringAttribute(span: SpanData, name: String): Option[String] =
+    span.attributes.elements.get[String](name).map(_.value)
+
+  private def longAttribute(span: SpanData, name: String): Option[Long] =
+    span.attributes.elements.get[Long](name).map(_.value)
+
+  private def hasNonEmptyString(span: SpanData, name: String): Boolean =
+    stringAttribute(span, name).exists(_.nonEmpty)
+
+  private def successfulClient(
+      name: String,
+      attributes: List[Attribute[_]]
+  ): SpanExpectation =
+    SpanExpectation
+      .client(name)
+      .noParentSpanContext
+      .attributesExact(attributes: _*)
+      .scope(scope)
+      .status(StatusExpectation.unset)
+      .eventCount(0)
+      .linkCount(0)
+
+  private def durationPoint(
+      attributes: List[Attribute[_]]
+  ): PointExpectation.Histogram =
+    PointExpectation.histogram
+      .count(1L)
+      .boundaries(DurationBoundaries)
+      .attributesExact(attributes: _*)
+
+  private def durationMetric(
+      first: PointExpectation.Histogram,
+      rest: PointExpectation.Histogram*
+  ): MetricExpectation =
+    MetricExpectation
+      .histogram(DurationMetricName)
+      .unit("s")
+      .scope(scope)
+      .exactlyPoints(first, rest: _*)
+
+  private def assertTrace(
+      spans: List[SpanData],
+      relevantNames: Set[String],
+      expectation: TraceForestExpectation
+  ): Unit = {
+    val relevantSpans = spans.filter(span => relevantNames(span.name))
+    TraceExpectations.check(relevantSpans, expectation) match {
+      case Right(_)         => ()
+      case Left(mismatches) => fail(TraceExpectations.format(mismatches))
+    }
+  }
+
+  private def assertMetrics(
+      metrics: List[MetricData],
+      expectation: MetricExpectation
+  ): Unit = {
+    val operationMetrics = metrics.filter(_.name == DurationMetricName)
+    assertEquals(operationMetrics.length, 1)
+    MetricExpectations.checkAll(operationMetrics, expectation) match {
+      case Right(_)         => ()
+      case Left(mismatches) => fail(MetricExpectations.format(mismatches))
+    }
+  }
+
+  test("simple execution emits one safe semantic span and one matching metric point") {
+    val config = TelemetryConfig.default
+      .withProtocolSpans(TelemetryConfig.ProtocolSpans.Disabled)
+    session(config).use { case (session, testkit) =>
+      val query = sql"SELECT 'secret'::varchar".query(varchar).withQuerySummary("SELECT constant")
+      val attributes = commonAttributes("SELECT constant", "query")
+      for {
+        result <- session.unique(query)
+        spans <- testkit.finishedSpans
+        metrics <- testkit.collectMetrics
+      } yield {
+        assertEquals(result, "secret")
+        assertTrace(
+          spans,
+          Set("SELECT constant"),
+          TraceForestExpectation.unordered(
+            TraceExpectation.leaf(successfulClient("SELECT constant", attributes))
+          )
+        )
+        assertMetrics(metrics, durationMetric(durationPoint(attributes)))
+      }
+    }
+  }
+
+  test("prepared and cursor executions capture all parameters at each logical boundary") {
+    val config = TelemetryConfig.default
+      .withCaptureQuery(
+        QueryCaptureConfig.recommended.withQueryParametersPolicy(
+          QueryCaptureConfig.QueryParametersPolicy.All
+        )
+      )
+      .withProtocolSpans(TelemetryConfig.ProtocolSpans.Disabled)
+    session(config).use { case (session, testkit) =>
+      val command =
+        sql"UPDATE country SET name = name WHERE code = $varchar".command
+          .withQuerySummary("UPDATE country")
+      val query =
+        sql"SELECT name FROM country WHERE code = $varchar".query(varchar)
+          .withQuerySummary("SELECT country")
+      val commandMetricAttributes = commonAttributes("UPDATE country", "bind+execute")
+      val queryMetricAttributes = commonAttributes("SELECT country", "execute")
+      val commandStaticAttributes = commandMetricAttributes ++ List(
+        Attribute("db.query.text", "UPDATE country SET name = name WHERE code = $1"),
+        Attribute("db.query.parameter.0", "GBR")
+      )
+      val queryStaticAttributes = queryMetricAttributes ++ List(
+        Attribute("db.query.text", "SELECT name FROM country WHERE code = $1"),
+        Attribute("db.query.parameter.0", "GBR"),
+        Attribute("skunk.fetch.max_rows", 1L)
+      )
+      val commandKeys = commandStaticAttributes.map(_.key.name).toSet ++
+        Set("skunk.portal.id", "skunk.statement.id")
+      val queryKeys = queryStaticAttributes.map(_.key.name).toSet ++
+        Set("skunk.portal.id", "skunk.statement.id")
+
+      val commandSpan =
+        SpanExpectation
+          .client("UPDATE country")
+          .noParentSpanContext
+          .attributesSubset(commandStaticAttributes: _*)
+          .scope(scope)
+          .status(StatusExpectation.unset)
+          .eventCount(0)
+          .linkCount(0)
+          .where("exact dynamic command attributes") { span =>
+            attributeNames(span) == commandKeys &&
+            hasNonEmptyString(span, "skunk.portal.id") &&
+            hasNonEmptyString(span, "skunk.statement.id")
+          }
+
+      val querySpan =
+        SpanExpectation
+          .client("SELECT country")
+          .noParentSpanContext
+          .attributesSubset(queryStaticAttributes: _*)
+          .scope(scope)
+          .status(StatusExpectation.unset)
+          .eventCount(0)
+          .linkCount(0)
+          .where("exact dynamic cursor attributes") { span =>
+            attributeNames(span) == queryKeys &&
+            hasNonEmptyString(span, "skunk.portal.id") &&
+            hasNonEmptyString(span, "skunk.statement.id")
+          }
+
+      for {
+        _ <- session.execute(command)("GBR")
+        rows <- session.prepare(query).flatMap(_.cursor("GBR").use(_.fetch(1)))
+        spans <- testkit.finishedSpans
+        metrics <- testkit.collectMetrics
+      } yield {
+        assertEquals(rows._1, List("United Kingdom"))
+        assertTrace(
+          spans,
+          Set("UPDATE country", "SELECT country"),
+          TraceForestExpectation.unordered(
+            TraceExpectation.leaf(commandSpan),
+            TraceExpectation.leaf(querySpan)
+          )
+        )
+        assertMetrics(
+          metrics,
+          durationMetric(
+            durationPoint(commandMetricAttributes),
+            durationPoint(queryMetricAttributes)
+          )
+        )
+      }
+    }
+  }
+
+  test("prepare failures emit a protocol span when protocol tracing is enabled") {
+    val config = TelemetryConfig.default
+      .withProtocolSpans(TelemetryConfig.ProtocolSpans.Internal)
+    session(config).use { case (session, testkit) =>
+      val query =
+        sql"SELECT name FROM telemetry_missing_table WHERE code = $varchar"
+          .query(varchar)
+          .withQuerySummary("SELECT missing table")
+      val span =
+        SpanExpectation
+          .internal("parse+describe")
+          .noParentSpanContext
+          .scope(scope)
+          .status(StatusExpectation.error)
+          .eventCount(1)
+          .linkCount(0)
+
+      for {
+        _ <- session.prepare(query).attempt
+        spans <- testkit.finishedSpans
+        metrics <- testkit.collectMetrics
+      } yield {
+        assertTrace(
+          spans,
+          Set("parse+describe"),
+          TraceForestExpectation.unordered(TraceExpectation.leaf(span))
+        )
+        assertEquals(metrics.filter(_.name == DurationMetricName), Nil)
+      }
+    }
+  }
+
+  test("PostgreSQL errors use SQLSTATE on the semantic span and metric") {
+    val config = TelemetryConfig.default
+      .withProtocolSpans(TelemetryConfig.ProtocolSpans.Disabled)
+    session(config).use { case (session, testkit) =>
+      val query = sql"SELECT 1 / 0".query(int4).withQuerySummary("divide by zero")
+      val attributes = commonAttributes("divide by zero", "query") ++ List(
+        Attribute("error.type", "22012"),
+        Attribute("db.response.status_code", "22012")
+      )
+      val errorSpan =
+        SpanExpectation
+          .client("divide by zero")
+          .noParentSpanContext
+          .attributesExact(attributes: _*)
+          .scope(scope)
+          .status(StatusExpectation.error)
+          .eventCount(1)
+          .linkCount(0)
+
+      for {
+        _ <- session.unique(query).attempt
+        spans <- testkit.finishedSpans
+        metrics <- testkit.collectMetrics
+      } yield {
+        assertTrace(
+          spans,
+          Set("divide by zero"),
+          TraceForestExpectation.unordered(TraceExpectation.leaf(errorSpan))
+        )
+        assertMetrics(metrics, durationMetric(durationPoint(attributes)))
+      }
+    }
+  }
+
+  test("protocol spans retain namespaced Skunk details without SQL or arguments") {
+    val config = TelemetryConfig.default
+      .withProtocolSpans(TelemetryConfig.ProtocolSpans.Internal)
+    session(config).use { case (session, testkit) =>
+      val query =
+        sql"SELECT name FROM country WHERE code = $varchar".query(varchar)
+          .withQuerySummary("SELECT country")
+      val clientStaticAttributes = commonAttributes("SELECT country", "execute") ++ List(
+        Attribute("db.query.text", "SELECT name FROM country WHERE code = $1"),
+        Attribute("skunk.fetch.max_rows", 1L)
+      )
+      val clientKeys = clientStaticAttributes.map(_.key.name).toSet ++
+        Set("skunk.portal.id", "skunk.statement.id")
+
+      val prepareKeys = Set(
+        "skunk.statement.id",
+        "skunk.statement.parameter_types",
+        "skunk.result.column_types"
+      )
+
+      val prepare =
+        SpanExpectation
+          .internal("parse+describe")
+          .noParentSpanContext
+          .scope(scope)
+          .status(StatusExpectation.unset)
+          .eventCount(0)
+          .linkCount(0)
+          .where("exact prepare attributes") { span =>
+            attributeNames(span) == prepareKeys &&
+            hasNonEmptyString(span, "skunk.statement.id") &&
+            hasNonEmptyString(span, "skunk.statement.parameter_types") &&
+            hasNonEmptyString(span, "skunk.result.column_types")
+          }
+
+      val client =
+        SpanExpectation
+          .client("SELECT country")
+          .noParentSpanContext
+          .attributesSubset(clientStaticAttributes: _*)
+          .scope(scope)
+          .status(StatusExpectation.unset)
+          .eventCount(0)
+          .linkCount(0)
+          .where("exact dynamic cursor attributes") { span =>
+            attributeNames(span) == clientKeys &&
+            hasNonEmptyString(span, "skunk.portal.id") &&
+            hasNonEmptyString(span, "skunk.statement.id")
+          }
+
+      val read =
+        SpanExpectation
+          .internal("read")
+          .attributesSubset(Attribute("skunk.response.row_count", 1L))
+          .scope(scope)
+          .status(StatusExpectation.unset)
+          .eventCount(0)
+          .linkCount(0)
+          .where("exact read response attributes") { span =>
+            attributeNames(span) == Set(
+              "skunk.response.row_count",
+              "skunk.response.more_rows"
+            ) && longAttribute(span, "skunk.response.row_count").contains(1L)
+          }
+
+      val decode =
+        SpanExpectation
+          .internal("decode")
+          .attributesEmpty
+          .scope(scope)
+          .status(StatusExpectation.unset)
+          .eventCount(0)
+          .linkCount(0)
+
+      val close =
+        SpanExpectation
+          .internal("close-portal")
+          .noParentSpanContext
+          .scope(scope)
+          .status(StatusExpectation.unset)
+          .eventCount(0)
+          .linkCount(0)
+          .where("exact close portal attributes") { span =>
+            attributeNames(span) == Set("skunk.portal.id") &&
+            hasNonEmptyString(span, "skunk.portal.id")
+          }
+
+      for {
+        rows <- session.prepare(query).flatMap(_.cursor("GBR").use(_.fetch(1)))
+        spans <- testkit.finishedSpans
+      } yield {
+        assertEquals(rows._1, List("United Kingdom"))
+        assertTrace(
+          spans,
+          Set("parse+describe", "SELECT country", "read", "decode", "close-portal"),
+          TraceForestExpectation.unordered(
+            TraceExpectation.leaf(prepare),
+            TraceExpectation.unordered(
+              client,
+              TraceExpectation.leaf(read),
+              TraceExpectation.leaf(decode)
+            ),
+            TraceExpectation.leaf(close)
+          )
+        )
+      }
+    }
+  }
+}
diff --git a/modules/tests/jvm/src/test/scala/TelemetryPoolConfigTest.scala b/modules/tests/jvm/src/test/scala/TelemetryPoolConfigTest.scala
new file mode 100644
index 00000000..f4ad140f
--- /dev/null
+++ b/modules/tests/jvm/src/test/scala/TelemetryPoolConfigTest.scala
@@ -0,0 +1,70 @@
+// Copyright (c) 2018-2024 by Rob Norris and Contributors
+// This software is licensed under the MIT License (MIT).
+// For more information see LICENSE or https://opensource.org/licenses/MIT
+
+package skunk
+
+import cats.effect.IO
+import munit.CatsEffectSuite
+import org.typelevel.otel4s.metrics.MeterProvider
+import org.typelevel.otel4s.sdk.testkit.InstrumentationScopeExpectation
+import org.typelevel.otel4s.sdk.testkit.OpenTelemetrySdkTestkit
+import org.typelevel.otel4s.sdk.testkit.trace.SpanExpectation
+import org.typelevel.otel4s.sdk.testkit.trace.StatusExpectation
+import org.typelevel.otel4s.sdk.testkit.trace.TraceExpectation
+import org.typelevel.otel4s.sdk.testkit.trace.TraceExpectations
+import org.typelevel.otel4s.sdk.testkit.trace.TraceForestExpectation
+import org.typelevel.otel4s.sdk.trace.data.SpanData
+import org.typelevel.otel4s.trace.TracerProvider
+import skunk.telemetry.ConnectionInfo
+import skunk.telemetry.Telemetry
+import skunk.telemetry.TelemetryConfig
+
+class TelemetryPoolConfigTest extends CatsEffectSuite {
+
+  private val scope =
+    InstrumentationScopeExpectation
+      .name("org.typelevel.skunk")
+      .version(BuildInfo.version)
+      .attributesEmpty
+
+  private def poolSpans(config: TelemetryConfig): IO[List[SpanData]] =
+    OpenTelemetrySdkTestkit.inMemory[IO]().use { testkit =>
+      implicit val tracerProvider: TracerProvider[IO] = testkit.tracerProvider
+      implicit val meterProvider: MeterProvider[IO] = testkit.meterProvider
+
+      Telemetry
+        .create[IO](config, ConnectionInfo("world", "localhost", None))
+        .flatMap(_.poolSpan("pool.allocate")(IO.unit)) *>
+        testkit.finishedSpans
+    }
+
+  test("pool spans are disabled by default") {
+    poolSpans(TelemetryConfig.default).map(spans => assertEquals(spans, Nil))
+  }
+
+  test("pool spans can be emitted as internal spans") {
+    poolSpans(
+      TelemetryConfig.default.withPoolSpans(TelemetryConfig.PoolSpans.Internal)
+    ).map { spans =>
+      val expectation =
+        TraceForestExpectation.unordered(
+          TraceExpectation.leaf(
+            SpanExpectation
+              .internal("pool.allocate")
+              .noParentSpanContext
+              .attributesEmpty
+              .scope(scope)
+              .status(StatusExpectation.unset)
+              .eventCount(0)
+              .linkCount(0)
+          )
+        )
+
+      TraceExpectations.check(spans, expectation) match {
+        case Right(_)         => ()
+        case Left(mismatches) => fail(TraceExpectations.format(mismatches))
+      }
+    }
+  }
+}
diff --git a/modules/tests/jvm/src/test/scala/TelemetryQueryAnalyzerTest.scala b/modules/tests/jvm/src/test/scala/TelemetryQueryAnalyzerTest.scala
new file mode 100644
index 00000000..98bb4a5e
--- /dev/null
+++ b/modules/tests/jvm/src/test/scala/TelemetryQueryAnalyzerTest.scala
@@ -0,0 +1,156 @@
+// Copyright (c) 2018-2024 by Rob Norris and Contributors
+// This software is licensed under the MIT License (MIT).
+// For more information see LICENSE or https://opensource.org/licenses/MIT
+
+package skunk
+
+import cats.effect.IO
+import munit.CatsEffectSuite
+import org.typelevel.otel4s.Attribute
+import org.typelevel.otel4s.metrics.MeterProvider
+import org.typelevel.otel4s.sdk.testkit.InstrumentationScopeExpectation
+import org.typelevel.otel4s.sdk.testkit.OpenTelemetrySdkTestkit
+import org.typelevel.otel4s.sdk.testkit.trace.SpanExpectation
+import org.typelevel.otel4s.sdk.testkit.trace.StatusExpectation
+import org.typelevel.otel4s.sdk.testkit.trace.TraceExpectation
+import org.typelevel.otel4s.sdk.testkit.trace.TraceExpectations
+import org.typelevel.otel4s.sdk.testkit.trace.TraceForestExpectation
+import org.typelevel.otel4s.sdk.trace.data.SpanData
+import org.typelevel.otel4s.trace.TracerProvider
+import skunk.telemetry.ConnectionInfo
+import skunk.telemetry.QueryAnalyzer
+import skunk.telemetry.SkunkAttributes
+import skunk.telemetry.Telemetry
+import skunk.telemetry.TelemetryConfig
+import skunk.util.Origin
+import skunk.exception.PostgresErrorException
+
+class TelemetryQueryAnalyzerTest extends CatsEffectSuite {
+
+  private def stringAttribute(span: SpanData, name: String): Option[String] =
+    span.attributes.elements.get[String](name).map(_.value)
+
+  private val scope =
+    InstrumentationScopeExpectation
+      .name("org.typelevel.skunk")
+      .version(BuildInfo.version)
+      .attributesEmpty
+
+  test("disabled protocol spans do not add protocol attributes to the ambient span") {
+    val config = TelemetryConfig.default
+      .withProtocolSpans(TelemetryConfig.ProtocolSpans.Disabled)
+
+    OpenTelemetrySdkTestkit.inMemory[IO]().use { testkit =>
+      implicit val tracerProvider: TracerProvider[IO] = testkit.tracerProvider
+      implicit val meterProvider: MeterProvider[IO] = testkit.meterProvider
+
+      for {
+        telemetry <- Telemetry.create[IO](config, ConnectionInfo("world", "localhost", None))
+        tracer <- tracerProvider.tracer("test").get
+        _ <- tracer.span("ambient").surround {
+               telemetry.internalSpan("protocol") {
+                 telemetry.addProtocolAttributes(SkunkAttributes.statementId("statement"))
+               }
+             }
+        spans <- testkit.finishedSpans
+      } yield {
+        val span = spans.find(_.name == "ambient").getOrElse(fail("ambient span not found"))
+        assertEquals(stringAttribute(span, "skunk.statement.id"), None)
+      }
+    }
+  }
+
+  test("dummy analyzer values determine semantic attributes and span name") {
+    val analyzer = QueryAnalyzer { _ =>
+      Some(
+        QueryAnalyzer.Analysis(
+          queryText = Some("CALL find_country(?)"),
+          storedProcedureName = Some("find_country"),
+          querySummary = Some("CALL find_country")
+        )
+      )
+    }
+    val config = TelemetryConfig.default.withQueryAnalyzer(analyzer)
+    val statement = Command("CALL find_country('GBR')", Origin.unknown, Void.codec)
+      .addAttributes(Attribute("app.query.category", "lookup"))
+
+    OpenTelemetrySdkTestkit.inMemory[IO]().use { testkit =>
+      implicit val tracerProvider: TracerProvider[IO] = testkit.tracerProvider
+      implicit val meterProvider: MeterProvider[IO] = testkit.meterProvider
+
+      for {
+        telemetry <- Telemetry.create[IO](config, ConnectionInfo("world", "localhost", None))
+        _ <- telemetry.databaseSpan("query", statement, Nil, RedactionStrategy.OptIn)(IO.unit)
+        spans <- testkit.finishedSpans
+      } yield {
+        val span =
+          SpanExpectation
+            .client("CALL find_country")
+            .noParentSpanContext
+            .attributesExact(
+              Attribute("db.system.name", "postgresql"),
+              Attribute("db.namespace", "world"),
+              Attribute("skunk.operation.name", "query"),
+              Attribute("server.address", "localhost"),
+              Attribute("db.query.summary", "CALL find_country"),
+              Attribute("db.stored_procedure.name", "find_country"),
+              Attribute("db.query.text", "CALL find_country(?)"),
+              Attribute("app.query.category", "lookup")
+            )
+            .scope(scope)
+            .status(StatusExpectation.unset)
+            .eventCount(0)
+            .linkCount(0)
+
+        val expectation =
+          TraceForestExpectation.unordered(TraceExpectation.leaf(span))
+
+        TraceExpectations.check(spans, expectation) match {
+          case Right(_)         => ()
+          case Left(mismatches) => fail(TraceExpectations.format(mismatches))
+        }
+      }
+    }
+  }
+
+  test("session telemetry keeps the connection namespace when an error names another schema") {
+    val statement = Command("INSERT INTO country VALUES ('GBR')", Origin.unknown, Void.codec)
+    val error = new PostgresErrorException(
+      sql = statement.sql,
+      sqlOrigin = Some(statement.origin),
+      info = Map(
+        'S' -> "ERROR",
+        'C' -> "23505",
+        'M' -> "duplicate key",
+        's' -> "archive",
+        't' -> "country",
+      ),
+      history = Nil,
+    )
+
+    OpenTelemetrySdkTestkit.inMemory[IO]().use { testkit =>
+      implicit val tracerProvider: TracerProvider[IO] = testkit.tracerProvider
+      implicit val meterProvider: MeterProvider[IO] = testkit.meterProvider
+
+      for {
+        poolTelemetry <- Telemetry.create[IO](
+                           TelemetryConfig.default,
+                           ConnectionInfo("", "localhost", None),
+                         )
+        sessionTelemetry = poolTelemetry.withConnection(
+                             ConnectionInfo("world", "localhost", None)
+                           )
+        _ <- sessionTelemetry
+               .databaseSpan("query", statement, Nil, RedactionStrategy.OptIn)(
+                 IO.raiseError[Unit](error)
+               )
+               .attempt
+        spans <- testkit.finishedSpans
+      } yield {
+        val span = spans.headOption.getOrElse(fail("database client span not found"))
+        assertEquals(stringAttribute(span, "db.namespace"), Some("world"))
+        assertEquals(stringAttribute(span, "db.collection.name"), Some("country"))
+      }
+    }
+  }
+}
diff --git a/modules/tests/shared/src/test/scala/PoolTest.scala b/modules/tests/shared/src/test/scala/PoolTest.scala
index 6a7044f9..6fad98f7 100644
--- a/modules/tests/shared/src/test/scala/PoolTest.scala
+++ b/modules/tests/shared/src/test/scala/PoolTest.scala
@@ -18,9 +18,13 @@ import skunk.util.Pool.ShutdownException
 import org.typelevel.otel4s.trace.Tracer
 import skunk.util.Recycler
 import cats.effect.testkit.TestControl
+import skunk.telemetry.Telemetry
 
 class PoolTest extends FTest {
 
+  implicit def telemetry(implicit tracer: Tracer[IO]): Telemetry[IO] =
+    skunk.TestTelemetry("pool-test")
+
   case class UserFailure() extends Exception("user failure")
   case class AllocFailure() extends Exception("allocation failure")
   case class FreeFailure() extends Exception("free failure")
@@ -47,18 +51,18 @@ class PoolTest extends FTest {
   // This test leaks
   tracedTestWithTracer("error in alloc is rethrown to caller (immediate)") { implicit tracer: Tracer[IO] =>
     val rsrc = Resource.make(IO.raiseError[String](AllocFailure()))(_ => IO.unit)
-    val pool = Pool.ofF({(_: Tracer[IO]) => rsrc}, 42)(Recycler.success)
-    pool.use(_(Tracer[IO]).use(_ => IO.unit)).assertFailsWith[AllocFailure]
+    val pool = Pool.ofF({(_: Telemetry[IO]) => rsrc}, 42)(Recycler.success)
+    pool.use(_(Telemetry[IO]).use(_ => IO.unit)).assertFailsWith[AllocFailure]
   }
 
   tracedTestWithTracer("error in alloc is rethrown to caller (deferral completion following errored cleanup)") { implicit tracer: Tracer[IO] =>
     resourceYielding(IO(1), IO.raiseError(AllocFailure())).flatMap { r =>
-      val p = Pool.ofF({(_: Tracer[IO]) => r}, 1)(Recycler[IO, Int](_ => IO.raiseError(ResetFailure())))
+      val p = Pool.ofF({(_: Telemetry[IO]) => r}, 1)(Recycler[IO, Int](_ => IO.raiseError(ResetFailure())))
       p.use { r =>
         for {
           d  <- Deferred[IO, Unit]
-          f1 <- r(Tracer[IO]).use(n => assertEqual("n should be 1", n, 1) *> d.get).assertFailsWith[ResetFailure].start
-          f2 <- r(Tracer[IO]).use(_ => fail[Int]("should never get here")).assertFailsWith[AllocFailure].start
+          f1 <- r(Telemetry[IO]).use(n => assertEqual("n should be 1", n, 1) *> d.get).assertFailsWith[ResetFailure].start
+          f2 <- r(Telemetry[IO]).use(_ => fail[Int]("should never get here")).assertFailsWith[AllocFailure].start
           _  <- d.complete(())
           _  <- f1.join
           _  <- f2.join
@@ -69,12 +73,12 @@ class PoolTest extends FTest {
 
   tracedTestWithTracer("error in alloc is rethrown to caller (deferral completion following failed cleanup)") { implicit tracer: Tracer[IO] =>
     resourceYielding(IO(1), IO.raiseError(AllocFailure())).flatMap { r =>
-      val p = Pool.ofF({(_: Tracer[IO]) => r}, 1)(Recycler.failure)
+      val p = Pool.ofF({(_: Telemetry[IO]) => r}, 1)(Recycler.failure)
       p.use { r =>
         for {
           d  <- Deferred[IO, Unit]
-          f1 <- r(tracer).use(n => assertEqual("n should be 1", n, 1) *> d.get).start
-          f2 <- r(tracer).use(_ => fail[Int]("should never get here")).assertFailsWith[AllocFailure].start
+          f1 <- r(Telemetry[IO]).use(n => assertEqual("n should be 1", n, 1) *> d.get).start
+          f2 <- r(Telemetry[IO]).use(_ => fail[Int]("should never get here")).assertFailsWith[AllocFailure].start
           _  <- d.complete(())
           _  <- f1.join
           _  <- f2.join
@@ -85,23 +89,23 @@ class PoolTest extends FTest {
 
   tracedTestWithTracer("error in finalizer does not prevent cleanup of deferreds") { implicit tracer: Tracer[IO] =>
     val r = Resource.make(IO(1))(_ => IO.raiseError(ResetFailure()))
-    val p = Pool.ofF({(_: Tracer[IO]) => r}, 1)(Recycler.failure)
+    val p = Pool.ofF({(_: Telemetry[IO]) => r}, 1)(Recycler.failure)
     p.use { r =>
-      val tx = r(Tracer[IO]).use(_ => IO.unit)
+      val tx = r(Telemetry[IO]).use(_ => IO.unit)
       List(tx, tx).parSequence
     }.assertFailsWith[ResetFailure]
   }
 
   tracedTestWithTracer("provoke dangling deferral cancellation") { implicit tracer: Tracer[IO] =>
     ints.flatMap { r =>
-      val p = Pool.ofF({(_: Tracer[IO]) => r}, 1)(Recycler.failure)
+      val p = Pool.ofF({(_: Telemetry[IO]) => r}, 1)(Recycler.failure)
       Deferred[IO, Either[Throwable, Int]].flatMap { d1 =>
         p.use { r =>
           for {
             d <- Deferred[IO, Unit]
-            _ <- r(tracer).use(_ => d.complete(()) *> IO.never).start // leaked forever
+            _ <- r(Telemetry[IO]).use(_ => d.complete(()) *> IO.never).start // leaked forever
             _ <- d.get // make sure the resource has been allocated
-            f <- r(tracer).use(_ => fail[Int]("should never get here")).attempt.flatMap(d1.complete).start // defer
+            f <- r(Telemetry[IO]).use(_ => fail[Int]("should never get here")).attempt.flatMap(d1.complete).start // defer
             _ <- IO.sleep(100.milli) // ensure that the fiber has a chance to run
           } yield f
         } .assertFailsWith[ResourceLeak].flatMap {
@@ -113,23 +117,23 @@ class PoolTest extends FTest {
 
   tracedTestWithTracer("error in free is rethrown to caller") { implicit tracer: Tracer[IO] =>
     val rsrc = Resource.make("foo".pure[IO])(_ => IO.raiseError(FreeFailure()))
-    val pool = Pool.ofF({(_: Tracer[IO]) => rsrc}, 42)(Recycler.success)
-    pool.use(_(tracer).use(_ => IO.unit)).assertFailsWith[FreeFailure]
+    val pool = Pool.ofF({(_: Telemetry[IO]) => rsrc}, 42)(Recycler.success)
+    pool.use(_(Telemetry[IO]).use(_ => IO.unit)).assertFailsWith[FreeFailure]
   }
 
   tracedTestWithTracer("error in reset is rethrown to caller") { implicit tracer: Tracer[IO] =>
     val rsrc = Resource.make("foo".pure[IO])(_ => IO.unit)
-    val pool = Pool.ofF({(_: Tracer[IO]) => rsrc}, 42)(Recycler[IO, String](_ => IO.raiseError(ResetFailure())))
-    pool.use(_(tracer).use(_ => IO.unit)).assertFailsWith[ResetFailure]
+    val pool = Pool.ofF({(_: Telemetry[IO]) => rsrc}, 42)(Recycler[IO, String](_ => IO.raiseError(ResetFailure())))
+    pool.use(_(Telemetry[IO]).use(_ => IO.unit)).assertFailsWith[ResetFailure]
   }
 
   tracedTestWithTracer("reuse on serial access") { implicit tracer: Tracer[IO] =>
-    ints.map(a => Pool.ofF({(_: Tracer[IO]) => a}, 3)(Recycler.success)).flatMap { factory =>
+    ints.map(a => Pool.ofF({(_: Telemetry[IO]) => a}, 3)(Recycler.success)).flatMap { factory =>
       factory.use { pool =>
-        pool(tracer).use { n =>
+        pool(Telemetry[IO]).use { n =>
           assertEqual("first num should be 1", n, 1)
         } *>
-        pool(tracer).use { n =>
+        pool(Telemetry[IO]).use { n =>
           assertEqual("we should get it again", n, 1)
         }
       }
@@ -137,14 +141,14 @@ class PoolTest extends FTest {
   }
 
   tracedTestWithTracer("allocation on nested access") { implicit tracer: Tracer[IO] =>
-    ints.map(a => Pool.ofF({(_: Tracer[IO]) => a}, 3)(Recycler.success)).flatMap { factory =>
+    ints.map(a => Pool.ofF({(_: Telemetry[IO]) => a}, 3)(Recycler.success)).flatMap { factory =>
       factory.use { pool =>
-        pool(tracer).use { n =>
+        pool(Telemetry[IO]).use { n =>
           assertEqual("first num should be 1", n, 1) *>
-          pool(tracer).use { n =>
+          pool(Telemetry[IO]).use { n =>
             assertEqual("but this one should be 2", n, 2)
           } *>
-          pool(tracer).use { n =>
+          pool(Telemetry[IO]).use { n =>
             assertEqual("and again", n, 2)
           }
         }
@@ -153,9 +157,9 @@ class PoolTest extends FTest {
   }
 
   tracedTestWithTracer("allocated resource can cause a leak, which will be detected on finalization") { implicit tracer: Tracer[IO] =>
-    ints.map(a => Pool.ofF({(_: Tracer[IO]) => a}, 3)(Recycler.success)).flatMap { factory =>
+    ints.map(a => Pool.ofF({(_: Telemetry[IO]) => a}, 3)(Recycler.success)).flatMap { factory =>
       factory.use { pool =>
-        pool(tracer).allocated
+        pool(Telemetry[IO]).allocated
       } .assertFailsWith[ResourceLeak].flatMap {
         case ResourceLeak(expected, actual, _) =>
           assert("expected 1 leakage", expected - actual == 1)
@@ -164,9 +168,9 @@ class PoolTest extends FTest {
   }
 
   tracedTestWithTracer("unmoored fiber can cause a leak, which will be detected on finalization") { implicit tracer: Tracer[IO] =>
-    ints.map(a => Pool.ofF({(_: Tracer[IO]) => a}, 3)(Recycler.success)).flatMap { factory =>
+    ints.map(a => Pool.ofF({(_: Telemetry[IO]) => a}, 3)(Recycler.success)).flatMap { factory =>
       factory.use { pool =>
-        pool(tracer).use(_ => IO.never).start *>
+        pool(Telemetry[IO]).use(_ => IO.never).start *>
         IO.sleep(100.milli) // ensure that the fiber has a chance to run
       } .assertFailsWith[ResourceLeak].flatMap {
         case ResourceLeak(expected, actual, _) =>
@@ -183,10 +187,10 @@ class PoolTest extends FTest {
   val shortRandomDelay = IO((Random.nextInt() % 100).abs.milliseconds)
 
   tracedTestWithTracer("progress and safety with many fibers") { implicit tracer: Tracer[IO] =>
-    ints.map(a => Pool.ofF({(_: Tracer[IO]) => a}, PoolSize)(Recycler.success)).flatMap { factory =>
+    ints.map(a => Pool.ofF({(_: Telemetry[IO]) => a}, PoolSize)(Recycler.success)).flatMap { factory =>
       (1 to ConcurrentTasks).toList.parTraverse_{ _ =>
         factory.use { p =>
-          p(tracer).use { _ =>
+          p(Telemetry[IO]).use { _ =>
             for {
               t <- shortRandomDelay
               _ <- IO.sleep(t)
@@ -198,12 +202,12 @@ class PoolTest extends FTest {
   }
 
   tracedTestWithTracer("progress and safety with many fibers and cancellation") { implicit tracer: Tracer[IO] =>
-    ints.map(a => Pool.ofF({(_: Tracer[IO]) => a}, PoolSize)(Recycler.success)).flatMap { factory =>
+    ints.map(a => Pool.ofF({(_: Telemetry[IO]) => a}, PoolSize)(Recycler.success)).flatMap { factory =>
       factory.use { pool =>
         (1 to ConcurrentTasks).toList.parTraverse_{_ =>
           for {
             t <- shortRandomDelay
-            f <- pool(tracer).use(_ => IO.sleep(t)).start
+            f <- pool(Telemetry[IO]).use(_ => IO.sleep(t)).start
             _ <- if (t > 50.milliseconds) f.join else f.cancel
           } yield ()
         }
@@ -212,10 +216,10 @@ class PoolTest extends FTest {
   }
 
   tracedTestWithTracer("progress and safety with many fibers and user failures") { implicit tracer: Tracer[IO] =>
-    ints.map(a => Pool.ofF({(_: Tracer[IO]) => a}, PoolSize)(Recycler.success)).flatMap { factory =>
+    ints.map(a => Pool.ofF({(_: Telemetry[IO]) => a}, PoolSize)(Recycler.success)).flatMap { factory =>
       factory.use { pool =>
         (1 to ConcurrentTasks).toList.parTraverse_{ _ =>
-          pool(tracer).use { _ =>
+          pool(Telemetry[IO]).use { _ =>
             for {
               t <- shortRandomDelay
               _ <- IO.sleep(t)
@@ -233,9 +237,9 @@ class PoolTest extends FTest {
       case false => IO.raiseError(AllocFailure())
     }
     val rsrc = Resource.make(alloc)(_ => IO.unit)
-    Pool.ofF({(_: Tracer[IO]) => rsrc}, PoolSize)(Recycler.success).use { pool =>
+    Pool.ofF({(_: Telemetry[IO]) => rsrc}, PoolSize)(Recycler.success).use { pool =>
       (1 to ConcurrentTasks).toList.parTraverse_{ _ =>
-        pool(tracer).use { _ =>
+        pool(Telemetry[IO]).use { _ =>
           IO.unit
         } .attempt
       }
@@ -248,9 +252,9 @@ class PoolTest extends FTest {
       case false => IO.raiseError(FreeFailure())
     }
     val rsrc  = Resource.make(IO.unit)(_ => free)
-    Pool.ofF({(_: Tracer[IO]) => rsrc}, PoolSize)(Recycler.success).use { pool =>
+    Pool.ofF({(_: Telemetry[IO]) => rsrc}, PoolSize)(Recycler.success).use { pool =>
       (1 to ConcurrentTasks).toList.parTraverse_{ _ =>
-        pool(tracer).use { _ =>
+        pool(Telemetry[IO]).use { _ =>
           IO.unit
         } .attempt
       }
@@ -268,9 +272,9 @@ class PoolTest extends FTest {
       case 2 => IO.raiseError(ResetFailure())
     }
     val rsrc  = Resource.make(IO.unit)(_ => IO.unit)
-    Pool.ofF({(_: Tracer[IO]) => rsrc}, PoolSize)(Recycler(_ => recycle)).use { pool =>
+    Pool.ofF({(_: Telemetry[IO]) => rsrc}, PoolSize)(Recycler(_ => recycle)).use { pool =>
       (1 to ConcurrentTasks).toList.parTraverse_{ _ =>
-        pool(tracer).use { _ =>
+        pool(Telemetry[IO]).use { _ =>
           IO.unit
         } handleErrorWith {
           case ResetFailure() => IO.unit
diff --git a/modules/tests/shared/src/test/scala/TelemetryConfigTest.scala b/modules/tests/shared/src/test/scala/TelemetryConfigTest.scala
new file mode 100644
index 00000000..e88193f7
--- /dev/null
+++ b/modules/tests/shared/src/test/scala/TelemetryConfigTest.scala
@@ -0,0 +1,244 @@
+// Copyright (c) 2018-2024 by Rob Norris and Contributors
+// This software is licensed under the MIT License (MIT).
+// For more information see LICENSE or https://opensource.org/licenses/MIT
+
+package skunk
+
+import munit.FunSuite
+import org.typelevel.otel4s.{Attribute, Attributes}
+import org.typelevel.otel4s.semconv.attributes.DbAttributes
+import skunk.codec.all.varchar
+import skunk.data.Encoded
+import skunk.telemetry.ConnectionInfo
+import skunk.telemetry.QueryAnalyzer
+import skunk.telemetry.QueryCaptureConfig
+import skunk.telemetry.Telemetry
+import skunk.telemetry.TelemetryConfig
+import skunk.util.Origin
+
+object TestTelemetry {
+  def apply(database: String)(implicit tracer: org.typelevel.otel4s.trace.Tracer[cats.effect.IO]): Telemetry[cats.effect.IO] =
+    new Telemetry.Impl(
+      TelemetryConfig.default,
+      ConnectionInfo(database, "simulated", None),
+      org.typelevel.otel4s.metrics.Histogram.noop[cats.effect.IO, Double],
+    )
+}
+
+class TelemetryConfigTest extends FunSuite {
+
+  private val connection =
+    ConnectionInfo("world", "localhost", None)
+
+  private def attribute(attributes: org.typelevel.otel4s.Attributes, name: String): Option[String] =
+    attributes.get[String](name).map(_.value)
+
+  test("telemetry records use abstract with-methods while analysis uses apply") {
+    val capture = QueryCaptureConfig.recommended
+      .withQueryTextPolicy(QueryCaptureConfig.QueryTextPolicy.None)
+      .withQueryParametersPolicy(QueryCaptureConfig.QueryParametersPolicy.All)
+    val analyzer = QueryAnalyzer.noop
+    val config = TelemetryConfig.default
+      .withCaptureQuery(capture)
+      .withQueryAnalyzer(analyzer)
+      .withPoolSpans(TelemetryConfig.PoolSpans.Internal)
+      .withProtocolSpans(TelemetryConfig.ProtocolSpans.Disabled)
+    val analysis = QueryAnalyzer.Analysis(
+      queryText = Some("SELECT ?"),
+      storedProcedureName = Some("find_country"),
+      querySummary = Some("SELECT country")
+    )
+    val statementTelemetry = Statement.Telemetry.empty.withQuerySummary("SELECT nation")
+
+    assertEquals(capture.queryTextPolicy, QueryCaptureConfig.QueryTextPolicy.None)
+    assertEquals(capture.queryParametersPolicy, QueryCaptureConfig.QueryParametersPolicy.All)
+    assertEquals(config.captureQuery, capture)
+    assertEquals(config.queryAnalyzer, analyzer)
+    assertEquals(config.poolSpans, TelemetryConfig.PoolSpans.Internal)
+    assertEquals(config.protocolSpans, TelemetryConfig.ProtocolSpans.Disabled)
+    assertEquals(analysis.queryText, Some("SELECT ?"))
+    assertEquals(analysis.storedProcedureName, Some("find_country"))
+    assertEquals(analysis.querySummary, Some("SELECT country"))
+    assertEquals(statementTelemetry.querySummary, Some("SELECT nation"))
+  }
+
+  test("pool spans are disabled by default") {
+    assertEquals(TelemetryConfig.default.poolSpans, TelemetryConfig.PoolSpans.Disabled)
+  }
+
+  test("QueryAnalyzer.noop and fallback") {
+    val fallback = QueryAnalyzer(_ => Some(QueryAnalyzer.Analysis(None, None, Some("SELECT country"))))
+    assertEquals(QueryAnalyzer.noop.analyze("SELECT 1"), None)
+    assertEquals(
+      QueryAnalyzer.noop.orElse(fallback).analyze("SELECT 1").flatMap(_.querySummary),
+      Some("SELECT country"),
+    )
+  }
+
+  test("safe defaults do not capture literal query text") {
+    val statement = Command("SELECT 'secret'", Origin.unknown, Void.codec)
+    val operation = Telemetry.resolveOperation(
+      "query",
+      statement,
+      Nil,
+      RedactionStrategy.OptIn,
+      TelemetryConfig.default,
+      connection,
+    )
+
+    assertEquals(attribute(operation.spanAttributes, "db.query.text"), None)
+    assertEquals(attribute(operation.spanAttributes, "skunk.operation.name"), Some("query"))
+    assertEquals(attribute(operation.spanAttributes, "db.operation.name"), None)
+    assertEquals(operation.spanName, "query")
+  }
+
+  test("safe defaults capture parameterized query text but not values") {
+    val statement = Command("SELECT $1::varchar", Origin.unknown, varchar)
+    val operation = Telemetry.resolveOperation(
+      "bind+execute",
+      statement,
+      List(Some(Encoded("secret"))),
+      RedactionStrategy.OptIn,
+      TelemetryConfig.default,
+      connection,
+    )
+
+    assertEquals(attribute(operation.spanAttributes, "db.query.text"), Some(statement.sql))
+    assertEquals(attribute(operation.spanAttributes, "db.query.parameter.0"), None)
+    assertEquals(attribute(operation.metricAttributes, "skunk.operation.name"), Some("bind+execute"))
+  }
+
+  test("all parameter capture respects redaction and remains span-only") {
+    val statement = Command("SELECT $1::varchar, $2::varchar", Origin.unknown, varchar ~ varchar)
+    val config = TelemetryConfig.default.withCaptureQuery(
+      QueryCaptureConfig.recommended.withQueryParametersPolicy(
+        QueryCaptureConfig.QueryParametersPolicy.All
+      )
+    )
+    val operation = Telemetry.resolveOperation(
+      "bind+execute",
+      statement,
+      List(Some(Encoded("secret", redacted = true)), Some(Encoded("public"))),
+      RedactionStrategy.OptIn,
+      config,
+      connection,
+    )
+
+    assertEquals(attribute(operation.spanAttributes, "db.query.parameter.0"), Some(Encoded.RedactedText))
+    assertEquals(attribute(operation.spanAttributes, "db.query.parameter.1"), Some("public"))
+    assertEquals(attribute(operation.metricAttributes, "db.query.parameter.0"), None)
+    assertEquals(attribute(operation.metricAttributes, "db.query.parameter.1"), None)
+  }
+
+  test("analyzed text, typed summary, and connection attributes are resolved at span creation") {
+    val analyzer = QueryAnalyzer(_ => Some(QueryAnalyzer.Analysis(
+      queryText = Some("SELECT * FROM country WHERE code = ?"),
+      storedProcedureName = None,
+      querySummary = Some("ignored analyzer summary"),
+    )))
+    val statement =
+      Command("SELECT * FROM country WHERE code = 'GBR'", Origin.unknown, Void.codec)
+        .withQuerySummary("get country by code")
+    val operation = Telemetry.resolveOperation(
+      "query",
+      statement,
+      Nil,
+      RedactionStrategy.OptIn,
+      TelemetryConfig.default.withQueryAnalyzer(analyzer),
+      connection.copy(serverPort = Some(6432L)),
+    )
+
+    assertEquals(operation.spanName, "get country by code")
+    assertEquals(attribute(operation.spanAttributes, "db.system.name"), Some("postgresql"))
+    assertEquals(attribute(operation.spanAttributes, "db.namespace"), Some("world"))
+    assertEquals(attribute(operation.spanAttributes, "server.address"), Some("localhost"))
+    assertEquals(operation.spanAttributes.get[Long]("server.port").map(_.value), Some(6432L))
+    assertEquals(attribute(operation.spanAttributes, "db.query.summary"), Some("get country by code"))
+    assertEquals(attribute(operation.metricAttributes, "db.query.summary"), Some("get country by code"))
+    assertEquals(
+      attribute(operation.spanAttributes, "db.query.text"),
+      Some("SELECT * FROM country WHERE code = ?"),
+    )
+    assertEquals(attribute(operation.metricAttributes, "db.query.text"), None)
+  }
+
+  test("a failing analyzer cannot fail a database operation") {
+    val analyzer = QueryAnalyzer(_ => throw new IllegalStateException("broken analyzer"))
+    val statement = Command("SELECT 'secret'", Origin.unknown, Void.codec)
+    val operation = Telemetry.resolveOperation(
+      "query",
+      statement,
+      Nil,
+      RedactionStrategy.OptIn,
+      TelemetryConfig.default.withQueryAnalyzer(analyzer),
+      connection,
+    )
+
+    assertEquals(operation.spanName, "query")
+    assertEquals(attribute(operation.spanAttributes, "db.query.text"), None)
+  }
+
+  test("statement summaries survive transformations and are accepted as-is") {
+    val command = Command("SELECT $1::varchar", Origin.unknown, varchar)
+      .withQuerySummary("select value")
+      .contramap[Int](_.toString)
+    val query = Query("SELECT $1::varchar", Origin.unknown, varchar, varchar)
+      .withQuerySummary("select value")
+      .dimap[Int, Int](_.toString)(_.length)
+
+    assertEquals(command.telemetry.querySummary, Some("select value"))
+    assertEquals(query.telemetry.querySummary, Some("select value"))
+    assertEquals(command.withQuerySummary("").telemetry.querySummary, Some(""))
+    assertEquals(command.withQuerySummary(" " * 2).telemetry.querySummary, Some(" " * 2))
+    assertEquals(command.withQuerySummary("x" * 256).telemetry.querySummary, Some("x" * 256))
+  }
+
+  test("db.query.summary can be supplied as a statement attribute") {
+    val statement = Command("SELECT 1", Origin.unknown, Void.codec)
+      .addAttributes(DbAttributes.DbQuerySummary("SELECT constant"))
+    val operation = Telemetry.resolveOperation(
+      "query",
+      statement,
+      Nil,
+      RedactionStrategy.OptIn,
+      TelemetryConfig.default,
+      connection,
+    )
+
+    assertEquals(statement.telemetry.querySummary, Some("SELECT constant"))
+    assertEquals(operation.spanName, "SELECT constant")
+    assertEquals(attribute(operation.spanAttributes, "db.query.summary"), Some("SELECT constant"))
+    assertEquals(attribute(operation.metricAttributes, "db.query.summary"), Some("SELECT constant"))
+  }
+
+  test("statement attributes are span-only and survive transformations") {
+    val attributes = Attributes(Attribute("app.query.category", "lookup"))
+    val statement = Command("SELECT $1::varchar", Origin.unknown, varchar)
+      .withAttributes(attributes)
+      .addAttributes(Attribute("db.collection.name", "country"))
+      .contramap[Int](_.toString)
+    val operation = Telemetry.resolveOperation(
+      "bind+execute",
+      statement,
+      Nil,
+      RedactionStrategy.OptIn,
+      TelemetryConfig.default,
+      connection,
+    )
+
+    assertEquals(attribute(statement.telemetry.attributes, "app.query.category"), Some("lookup"))
+    assertEquals(attribute(operation.spanAttributes, "app.query.category"), Some("lookup"))
+    assertEquals(attribute(operation.spanAttributes, "db.collection.name"), Some("country"))
+    assertEquals(attribute(operation.metricAttributes, "app.query.category"), None)
+    assertEquals(attribute(operation.metricAttributes, "db.collection.name"), None)
+
+    val unrestricted = Statement.Telemetry.empty.addAttributes(
+      Attribute("db.query.summary", 1L),
+      Attribute("db.query.text", "secret"),
+      Attribute("skunk.custom", "value"),
+    )
+    assertEquals(unrestricted.attributes.get[Long]("db.query.summary").map(_.value), Some(1L))
+    assertEquals(attribute(unrestricted.attributes, "db.query.text"), Some("secret"))
+    assertEquals(attribute(unrestricted.attributes, "skunk.custom"), Some("value"))
+  }
+}
diff --git a/modules/tests/shared/src/test/scala/simulation/SimTest.scala b/modules/tests/shared/src/test/scala/simulation/SimTest.scala
index 108edd90..f66a1b2d 100644
--- a/modules/tests/shared/src/test/scala/simulation/SimTest.scala
+++ b/modules/tests/shared/src/test/scala/simulation/SimTest.scala
@@ -9,7 +9,6 @@ import cats.effect._
 import ffstest.FTest
 import fs2.concurrent.Signal
 import org.typelevel.otel4s.trace.Tracer
-import org.typelevel.otel4s.metrics.Histogram
 import skunk.{Session, RedactionStrategy, TypingStrategy}
 import skunk.data.Notification
 import skunk.data.TransactionStatus
@@ -21,10 +20,12 @@ import skunk.util.Namer
 import skunk.util.Origin
 import skunk.net.protocol.Describe
 import skunk.net.protocol.Parse
+import skunk.telemetry.Telemetry
 
 trait SimTest extends FTest with SimMessageSocket.DSL {
 
   implicit val tracer: Tracer[IO] = Tracer.noop
+  implicit val telemetry: Telemetry[IO] = skunk.TestTelemetry("simulated")
 
   private class SimulatedBufferedMessageSocket(ms: MessageSocket[IO]) extends BufferedMessageSocket[IO] {
     def receive: IO[BackendMessage] = ms.receive
@@ -45,7 +46,7 @@ trait SimTest extends FTest with SimMessageSocket.DSL {
       nam <- Namer[IO]
       dc  <- Describe.Cache.empty[IO](1024, 1024)
       pc  <- Parse.Cache.empty[IO](1024)
-      pro <- Protocol.fromMessageSocket(bms, nam, dc, pc, RedactionStrategy.None, Histogram.noop[IO, Double])
+      pro <- Protocol.fromMessageSocket(bms, nam, dc, pc, RedactionStrategy.None)
       _   <- pro.startup(user, database, password, Session.DefaultConnectionParameters)
       ses <- Session.fromProtocol(pro, nam, TypingStrategy.BuiltinsOnly, RedactionStrategy.None)
     } yield ses