diff --git a/CHANGELOG.md b/CHANGELOG.md
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,5 +1,39 @@
 # Revision history for pgmq-hasql
 
+## 0.7.0.0 -- 2026-10-26
+
+### New Features
+
+- Add non-destructive inspection reads, exported from `Pgmq` and
+  `Pgmq.Hasql.Sessions`: `peekMessages`, `peekArchivedMessages`, `lookupMessage`,
+  `lookupArchivedMessage` and `queueMetricsUnvalidated`. Paging uses ascending
+  `msg_id` with an exclusive cursor and a row limit, never `OFFSET`; lookups return
+  `Nothing` for missing messages. Reads leave `vt`, `read_ct` and `last_read_at`
+  untouched.
+- `PeekMessages` and `LookupMessage` accept plain `Text` names, including foreign and
+  mixed-case names that `parseQueueName` rejects. Physical tables are resolved with
+  `pgmq.format_table_name` and quoted as identifiers; missing queues retain server
+  error `42P01`.
+- Add `Pgmq.Hasql.Statements.Inspection` with `formatTableName`, `quoteIdentifier`
+  and four `unpreparable` statement builders, plus `archivedMessageDecoder` in
+  `Pgmq.Hasql.Decoders`.
+- Add `ToJSON QueueMetrics` with snake_case fields and explicit nulls. Total depth
+  (`queue_length`) stays distinct from available depth (`queue_visible_length`);
+  null `default_partition_length` means unavailable/inapplicable, not zero.
+  Goldens and parity properties pin the published contract in design note 020.
+
+### Other Changes
+
+- Document partition notification silence and cover native/stock partition channels,
+  native aliases and ordinary throttling.
+- Inspection tests cover late lower-ID commits and late archival, propagate consumer
+  failures and exercise 100,000-row ordinary/unlogged and numeric/time-partitioned
+  live/archive layouts. Design note 019 records query plans, moving-view semantics,
+  finite database deadlines and separate inspection capacity requirements. Row
+  limits do not bound scan work or payload bytes.
+- Ephemeral PostgreSQL roots are created/validated as owner-only (`0700`), existing
+  owned roots are tightened, unsafe paths are rejected and registry contents survive.
+
 ## 0.6.1.1 -- 2026-09-25
 
 Coordinated family version bump; the library is unchanged from 0.6.1.0.
diff --git a/pgmq-hasql.cabal b/pgmq-hasql.cabal
--- a/pgmq-hasql.cabal
+++ b/pgmq-hasql.cabal
@@ -1,6 +1,6 @@
 cabal-version: 3.4
 name: pgmq-hasql
-version: 0.6.1.1
+version: 0.7.0.0
 synopsis: Hasql-based client for PGMQ (PostgreSQL Message Queue)
 description:
   A Haskell client library for PGMQ (PostgreSQL Message Queue) built
@@ -17,7 +17,9 @@
 category: Database
 build-type: Simple
 extra-doc-files: CHANGELOG.md
-extra-source-files: test/fixtures/pgmq-1.12.0.sql
+extra-source-files:
+  test/fixtures/pgmq-1.12.0.sql
+  test/golden/*.json
 
 common warnings
   ghc-options:
@@ -42,6 +44,7 @@
     Pgmq.Hasql.Quasi
     Pgmq.Hasql.Sessions
     Pgmq.Hasql.Statements
+    Pgmq.Hasql.Statements.Inspection
     Pgmq.Hasql.Statements.Message
     Pgmq.Hasql.Statements.QueueManagement
     Pgmq.Hasql.Statements.QueueObservability
@@ -64,7 +67,7 @@
     hasql ^>=1.10,
     hasql-transaction ^>=1.2,
     lens ^>=5.3,
-    pgmq-core >=0.6 && <0.7,
+    pgmq-core >=0.7 && <0.8,
     template-haskell >=2.20 && <3,
     text ^>=2.1,
     time ^>=1.14,
@@ -91,6 +94,10 @@
     DecoderValidationSpec
     EphemeralDb
     Generators
+    InspectionCostSpec
+    InspectionForeignNameSpec
+    InspectionSpec
+    JsonGoldenSpec
     MessageSpec
     MetricsSpec
     MixedCaseRemediationSpec
@@ -98,7 +105,9 @@
     NotifyRaceSpec
     NullBodySpec
     NullSemanticsSpec
+    PartitionedNotifySpec
     QueueSpec
+    RootPermissionsSpec
     RoundTripSpec
     SchemaSpec
     TestUtils
@@ -112,19 +121,22 @@
   build-depends:
     aeson,
     base >=4.18 && <5,
+    bytestring >=0.11 && <0.13,
+    containers,
     directory,
-    ephemeral-pg >=0.3.1.0,
+    ephemeral-pg >=0.3.1 && <0.4,
     hasql,
     hasql-pool ^>=1.4,
     hedgehog ^>=1.5,
     pg-migrate,
-    pgmq-core >=0.6 && <0.7,
-    pgmq-hasql >=0.6 && <0.7,
-    pgmq-migration >=0.6 && <0.7,
+    pgmq-core >=0.7 && <0.8,
+    pgmq-hasql >=0.7 && <0.8,
+    pgmq-migration >=0.7 && <0.8,
     postgresql-libpq >=0.10.1 && <0.12,
     random ^>=1.2,
     scientific ^>=0.3,
     tasty ^>=1.5,
+    tasty-golden ^>=2.3.6,
     tasty-hedgehog ^>=1.4,
     tasty-hunit ^>=0.10,
     text,
diff --git a/src/Pgmq.hs b/src/Pgmq.hs
--- a/src/Pgmq.hs
+++ b/src/Pgmq.hs
@@ -56,6 +56,21 @@
     ReadGrouped (..),
     ReadGroupedWithPoll (..),
 
+    -- * Non-destructive Inspection
+
+    -- | Reads that observe a queue or its archive without leasing anything:
+    -- @vt@ and @read_ct@ are untouched. They accept any server-accepted name
+    -- (pass 'queueNameToText' for a validated one) and page by exclusive
+    -- @msg_id@ cursor, never @OFFSET@. See
+    -- @docs/design/019-non-destructive-inspection-reads.md@.
+    peekMessages,
+    peekArchivedMessages,
+    lookupMessage,
+    lookupArchivedMessage,
+    queueMetricsUnvalidated,
+    PeekMessages (..),
+    LookupMessage (..),
+
     -- * Topic Routing (pgmq 1.11.0+)
 
     -- ** Topic Management
@@ -84,6 +99,7 @@
     MessageHeaders (..),
     MessageId (..),
     Message (..),
+    ArchivedMessage (..),
     Queue (..),
     UnvalidatedQueue (..),
     QueueName,
@@ -173,8 +189,13 @@
     listQueuesUnvalidated,
     listTopicBindings,
     listTopicBindingsForQueue,
+    lookupArchivedMessage,
+    lookupMessage,
+    peekArchivedMessages,
+    peekMessages,
     pop,
     queueMetrics,
+    queueMetricsUnvalidated,
     readGrouped,
     readGroupedHead,
     readGroupedHeadWithPoll,
@@ -211,7 +232,9 @@
     BindTopic (..),
     CreatePartitionedQueue (..),
     EnableNotifyInsert (..),
+    LookupMessage (..),
     MessageQuery (..),
+    PeekMessages (..),
     PopMessage (..),
     QueueMetrics (..),
     ReadGrouped (..),
@@ -230,7 +253,8 @@
     VisibilityTimeoutQuery (..),
   )
 import Pgmq.Types
-  ( Message (..),
+  ( ArchivedMessage (..),
+    Message (..),
     MessageBody (..),
     MessageHeaders (..),
     MessageId (..),
diff --git a/src/Pgmq/Hasql/Decoders.hs b/src/Pgmq/Hasql/Decoders.hs
--- a/src/Pgmq/Hasql/Decoders.hs
+++ b/src/Pgmq/Hasql/Decoders.hs
@@ -1,6 +1,7 @@
 module Pgmq.Hasql.Decoders
   ( messageDecoder,
     messageIdDecoder,
+    archivedMessageDecoder,
     queueDecoder,
     unvalidatedQueueDecoder,
     queueMetricsDecoder,
@@ -19,7 +20,8 @@
 import Hasql.Decoders qualified as D
 import Pgmq.Hasql.Statements.Types (QueueMetrics (..))
 import Pgmq.Types
-  ( Message (..),
+  ( ArchivedMessage (..),
+    Message (..),
     MessageBody (..),
     MessageId (..),
     NotifyInsertThrottle (..),
@@ -65,6 +67,16 @@
 
 messageIdDecoder :: D.Row MessageId
 messageIdDecoder = MessageId <$> D.column (D.nonNullable D.int8)
+
+-- | Decoder for an archive-table row projected as the seven message columns in
+-- 'messageDecoder' order followed by @archived_at@. The inspection statements
+-- list the columns explicitly in exactly this order; the archive table's own
+-- column order (which puts @archived_at@ before @vt@) is irrelevant.
+archivedMessageDecoder :: D.Row ArchivedMessage
+archivedMessageDecoder =
+  ArchivedMessage
+    <$> messageDecoder
+    <*> D.column (D.nonNullable D.timestamptz) -- archived_at
 
 -- | Decoder for pgmq.queue_record type
 -- Column order: queue_name (varchar), is_partitioned (bool), is_unlogged (bool), created_at (timestamptz)
diff --git a/src/Pgmq/Hasql/Sessions.hs b/src/Pgmq/Hasql/Sessions.hs
--- a/src/Pgmq/Hasql/Sessions.hs
+++ b/src/Pgmq/Hasql/Sessions.hs
@@ -33,6 +33,12 @@
     allQueueMetrics,
     readMessage,
     readWithPoll,
+    -- Non-destructive inspection (no upstream function)
+    peekMessages,
+    peekArchivedMessages,
+    lookupMessage,
+    lookupArchivedMessage,
+    queueMetricsUnvalidated,
     -- FIFO read functions (pgmq 1.8.0+)
     readGrouped,
     readGroupedWithPoll,
@@ -69,6 +75,7 @@
 import Hasql.Session (Session, statement)
 import Pgmq.Hasql.Prelude
 import Pgmq.Hasql.Statements qualified as Stmt
+import Pgmq.Hasql.Statements.Inspection qualified as Inspect
 import Pgmq.Hasql.Statements.Message qualified as Msg
 import Pgmq.Hasql.Statements.TopicManagement qualified as Topic
 import Pgmq.Hasql.Statements.Types
@@ -86,7 +93,9 @@
     BindTopic,
     CreatePartitionedQueue,
     EnableNotifyInsert,
+    LookupMessage (..),
     MessageQuery,
+    PeekMessages (..),
     PopMessage,
     QueueMetrics,
     ReadGrouped,
@@ -105,7 +114,8 @@
     VisibilityTimeoutQuery,
   )
 import Pgmq.Types
-  ( Message,
+  ( ArchivedMessage,
+    Message,
     MessageId,
     NotifyInsertThrottle,
     Queue,
@@ -216,6 +226,10 @@
 detachArchive q = statement q Stmt.detachArchive
 
 -- | Enable insert notifications for a queue (pgmq 1.7.0+)
+--
+-- Normal partitioned queues receive no insert notifications on their documented
+-- parent channel. See @docs/design/015-notification-delivery-contract.md@.
+-- The reconciler reports partitioned notification declarations as unsupported.
 enableNotifyInsert :: EnableNotifyInsert -> Session ()
 enableNotifyInsert config = statement config Stmt.enableNotifyInsert
 
@@ -232,6 +246,49 @@
 
 allQueueMetrics :: Session [QueueMetrics]
 allQueueMetrics = statement () Stmt.allQueueMetrics
+
+-- | Metrics for a queue named by plain text, including names
+-- 'Pgmq.Types.parseQueueName' rejects. Same projection as 'queueMetrics'.
+queueMetricsUnvalidated :: Text -> Session QueueMetrics
+queueMetricsUnvalidated q = statement q Stmt.queueMetricsUnvalidated
+
+-- Non-destructive inspection ---------------------------------------------------
+--
+-- Each read resolves the physical table on the server first, then runs an
+-- unprepared SELECT against it. Neither statement modifies a row: vt, read_ct,
+-- and last_read_at are exactly what they were. A missing queue surfaces as the
+-- server's undefined_table error (SQLSTATE 42P01); a missing message is Nothing.
+
+-- | A keyset page of a queue table in ascending @msg_id@ order, starting
+-- strictly after 'afterMessageId', without leasing anything. Each page has its
+-- own snapshot; refresh from the beginning to find lower IDs committed later.
+-- The row limit does not bound scan work: partitioned queues may scan and sort.
+-- Use finite database deadlines and capacity separate from worker connections.
+peekMessages :: PeekMessages -> Session (Vector Message)
+peekMessages PeekMessages {unvalidatedQueueName, afterMessageId, limit} = do
+  table <- statement (unvalidatedQueueName, "q") Inspect.formatTableName
+  statement (afterMessageId, limit) (Inspect.peekStatement table)
+
+-- | A keyset page of an archive table, each row with its archival timestamp.
+-- An older ID archived after the cursor advances requires a first-page refresh.
+-- The same cost and isolation limits as 'peekMessages' apply.
+peekArchivedMessages :: PeekMessages -> Session (Vector ArchivedMessage)
+peekArchivedMessages PeekMessages {unvalidatedQueueName, afterMessageId, limit} = do
+  table <- statement (unvalidatedQueueName, "a") Inspect.formatTableName
+  statement (afterMessageId, limit) (Inspect.peekArchivedStatement table)
+
+-- | One message of a queue table by id, or 'Nothing' when it is not there
+-- (deleted, archived, popped, or never sent).
+lookupMessage :: LookupMessage -> Session (Maybe Message)
+lookupMessage LookupMessage {unvalidatedQueueName, messageId} = do
+  table <- statement (unvalidatedQueueName, "q") Inspect.formatTableName
+  statement messageId (Inspect.lookupStatement table)
+
+-- | One message of an archive table by id, or 'Nothing'.
+lookupArchivedMessage :: LookupMessage -> Session (Maybe ArchivedMessage)
+lookupArchivedMessage LookupMessage {unvalidatedQueueName, messageId} = do
+  table <- statement (unvalidatedQueueName, "a") Inspect.formatTableName
+  statement messageId (Inspect.lookupArchivedStatement table)
 
 readMessage :: ReadMessage -> Session (Vector Message)
 readMessage query = statement query Stmt.readMessage
diff --git a/src/Pgmq/Hasql/Statements.hs b/src/Pgmq/Hasql/Statements.hs
--- a/src/Pgmq/Hasql/Statements.hs
+++ b/src/Pgmq/Hasql/Statements.hs
@@ -1,11 +1,13 @@
 module Pgmq.Hasql.Statements
-  ( module Pgmq.Hasql.Statements.Message,
+  ( module Pgmq.Hasql.Statements.Inspection,
+    module Pgmq.Hasql.Statements.Message,
     module Pgmq.Hasql.Statements.QueueManagement,
     module Pgmq.Hasql.Statements.QueueObservability,
     module Pgmq.Hasql.Statements.TopicManagement,
   )
 where
 
+import Pgmq.Hasql.Statements.Inspection
 import Pgmq.Hasql.Statements.Message
 import Pgmq.Hasql.Statements.QueueManagement
 import Pgmq.Hasql.Statements.QueueObservability
diff --git a/src/Pgmq/Hasql/Statements/Inspection.hs b/src/Pgmq/Hasql/Statements/Inspection.hs
new file mode 100644
--- /dev/null
+++ b/src/Pgmq/Hasql/Statements/Inspection.hs
@@ -0,0 +1,110 @@
+-- | Non-destructive inspection statements over queue and archive tables.
+--
+-- Upstream pgmq offers no read that does not lease (@pgmq.read@,
+-- @read_with_poll@, and @pop@ all set @vt@ and bump @read_ct@), no read of an
+-- archive table, and no fetch by id. These statements are plain @SELECT@s over
+-- the tables the upstream schema defines; they shadow no @pgmq.*@ function and
+-- add no migration, which is exactly the kind of hand-written statement
+-- @docs/design/012-vendor-upstream-pgmq-sql.md@ permits. None of them modifies
+-- a row.
+--
+-- The physical table is resolved on the server by 'formatTableName' (upstream's
+-- @pgmq.format_table_name@, which lowercases and rejects @$@, @;@, @--@, and
+-- @'@) and spliced into the SQL as a double-quoted identifier by
+-- 'quoteIdentifier'. Because the SQL text differs per table, every builder uses
+-- 'unpreparable': hasql caches prepared statements by SQL text, and a prepared
+-- statement per queue would grow the per-connection cache without bound.
+--
+-- Pages are keyset pages: @msg_id > cursor@, ordered by @msg_id@, bounded by
+-- @LIMIT@, never @OFFSET@. A limit bounds returned rows, not scan work;
+-- partitioned layouts may scan and sort. Separate calls do not hold a shared
+-- snapshot: refresh from the first page for lower IDs committed or archived
+-- after a cursor advances. Use finite statement/lock deadlines and separate
+-- inspection capacity. See @docs/design/019-non-destructive-inspection-reads.md@.
+module Pgmq.Hasql.Statements.Inspection
+  ( formatTableName,
+    quoteIdentifier,
+    peekStatement,
+    peekArchivedStatement,
+    lookupStatement,
+    lookupArchivedStatement,
+  )
+where
+
+import Data.Text qualified as T
+import Hasql.Decoders qualified as D
+import Hasql.Encoders qualified as E
+import Hasql.Statement (Statement, preparable, unpreparable)
+import Pgmq.Hasql.Decoders (archivedMessageDecoder, messageDecoder)
+import Pgmq.Hasql.Encoders (messageIdValue)
+import Pgmq.Hasql.Prelude
+import Pgmq.Types (ArchivedMessage, Message, MessageId)
+
+-- | @select pgmq.format_table_name($1, $2)@: the physical table name for a
+-- queue name and a prefix (@"q"@ for the queue table, @"a"@ for the archive).
+-- Raises the server's own error for names containing @$@, @;@, @--@, or @'@.
+formatTableName :: Statement (Text, Text) Text
+formatTableName = preparable sql encoder decoder
+  where
+    sql = "select pgmq.format_table_name($1, $2)"
+    encoder =
+      (fst >$< E.param (E.nonNullable E.text))
+        <> (snd >$< E.param (E.nonNullable E.text))
+    decoder = D.singleRow (D.column (D.nonNullable D.text))
+
+-- | Double-quote a SQL identifier, doubling any embedded double quote, so a
+-- physical table name such as @q_odd-name@ or @q_myqueue@ can be spliced into
+-- SQL text safely.
+quoteIdentifier :: Text -> Text
+quoteIdentifier ident = "\"" <> T.replace "\"" "\"\"" ident <> "\""
+
+-- | The seven columns 'messageDecoder' expects, in its order.
+messageColumns :: Text
+messageColumns = "msg_id, read_ct, enqueued_at, last_read_at, vt, message, headers"
+
+-- | 'messageColumns' followed by the archive table's @archived_at@.
+archivedColumns :: Text
+archivedColumns = messageColumns <> ", archived_at"
+
+pageEncoder :: E.Params (Maybe MessageId, Int32)
+pageEncoder =
+  (fst >$< E.param (E.nullable messageIdValue))
+    <> (snd >$< E.param (E.nonNullable E.int4))
+
+messageIdEncoder :: E.Params MessageId
+messageIdEncoder = E.param (E.nonNullable messageIdValue)
+
+-- | A keyset page of a queue table. The argument is the already-resolved
+-- physical table name (from 'formatTableName' with prefix @"q"@). Returns at
+-- most @limit@ rows with @msg_id@ strictly greater than the cursor, in
+-- ascending @msg_id@ order, and modifies nothing.
+peekStatement :: Text -> Statement (Maybe MessageId, Int32) (Vector Message)
+peekStatement table = unpreparable (pageSql messageColumns table) pageEncoder (D.rowVector messageDecoder)
+
+-- | A keyset page of an archive table (from 'formatTableName' with prefix
+-- @"a"@), each row carrying @archived_at@. Same paging rules as
+-- 'peekStatement'.
+peekArchivedStatement :: Text -> Statement (Maybe MessageId, Int32) (Vector ArchivedMessage)
+peekArchivedStatement table = unpreparable (pageSql archivedColumns table) pageEncoder (D.rowVector archivedMessageDecoder)
+
+-- | One row of a queue table by id, or 'Nothing'. Modifies nothing.
+lookupStatement :: Text -> Statement MessageId (Maybe Message)
+lookupStatement table = unpreparable (lookupSql messageColumns table) messageIdEncoder (D.rowMaybe messageDecoder)
+
+-- | One row of an archive table by id, or 'Nothing'.
+lookupArchivedStatement :: Text -> Statement MessageId (Maybe ArchivedMessage)
+lookupArchivedStatement table = unpreparable (lookupSql archivedColumns table) messageIdEncoder (D.rowMaybe archivedMessageDecoder)
+
+-- | @$1@ is the exclusive cursor (nullable), @$2@ the page size. The explicit
+-- null test keeps the no-cursor case independent of where ids start.
+pageSql :: Text -> Text -> Text
+pageSql columns table =
+  "select "
+    <> columns
+    <> " from pgmq."
+    <> quoteIdentifier table
+    <> " where ($1::bigint is null or msg_id > $1) order by msg_id asc limit $2"
+
+lookupSql :: Text -> Text -> Text
+lookupSql columns table =
+  "select " <> columns <> " from pgmq." <> quoteIdentifier table <> " where msg_id = $1"
diff --git a/src/Pgmq/Hasql/Statements/QueueManagement.hs b/src/Pgmq/Hasql/Statements/QueueManagement.hs
--- a/src/Pgmq/Hasql/Statements/QueueManagement.hs
+++ b/src/Pgmq/Hasql/Statements/QueueManagement.hs
@@ -74,6 +74,10 @@
 -- NOTIFY is fire-and-forget: notifications are not queued for disconnected
 -- listeners, and a configured throttle interval suppresses them by design. Every
 -- consumer needs a poll fallback in addition to LISTEN.
+--
+-- Normal partitioned queues receive no insert notifications on their documented
+-- parent channel. See @docs/design/015-notification-delivery-contract.md@.
+-- The reconciler reports partitioned notification declarations as unsupported.
 enableNotifyInsert :: Statement EnableNotifyInsert ()
 enableNotifyInsert = preparable sql enableNotifyInsertEncoder D.noResult
   where
diff --git a/src/Pgmq/Hasql/Statements/QueueObservability.hs b/src/Pgmq/Hasql/Statements/QueueObservability.hs
--- a/src/Pgmq/Hasql/Statements/QueueObservability.hs
+++ b/src/Pgmq/Hasql/Statements/QueueObservability.hs
@@ -3,6 +3,7 @@
     listQueuesUnvalidated,
     listFifoIndexQueueNames,
     queueMetrics,
+    queueMetricsUnvalidated,
     allQueueMetrics,
   )
 where
@@ -50,18 +51,26 @@
       \where schemaname = 'pgmq' and indexname ~ '^q_.*_fifo_idx$'"
     decoder = D.rowList (D.column (D.nonNullable D.text))
 
+-- | The metrics projection with a stable shape on pgmq 1.12 and 1.13. The JSON
+-- record lookup preserves SQL NULL for the attribute 1.12 lacks.
+queueMetricsSql :: Text
+queueMetricsSql =
+  "select m.queue_name, m.queue_length, m.newest_msg_age_sec, m.oldest_msg_age_sec, \
+  \m.total_messages, m.scrape_time, m.queue_visible_length, \
+  \(to_jsonb(m)->>'default_partition_length')::bigint \
+  \from pgmq.metrics($1) as m"
+
 -- | Metrics with a stable projection on pgmq 1.12 and 1.13.
--- The JSON record lookup preserves SQL NULL for the missing 1.12 attribute.
 -- | https://pgmq.github.io/pgmq/api/sql/functions/#metrics
 queueMetrics :: Statement QueueName QueueMetrics
-queueMetrics = preparable sql queueNameEncoder decoder
-  where
-    sql =
-      "select m.queue_name, m.queue_length, m.newest_msg_age_sec, m.oldest_msg_age_sec, \
-      \m.total_messages, m.scrape_time, m.queue_visible_length, \
-      \(to_jsonb(m)->>'default_partition_length')::bigint \
-      \from pgmq.metrics($1) as m"
-    decoder = D.singleRow queueMetricsDecoder
+queueMetrics = preparable queueMetricsSql queueNameEncoder (D.singleRow queueMetricsDecoder)
+
+-- | 'queueMetrics' for a queue named by plain text: any name the server
+-- accepts, including names 'Pgmq.Types.parseQueueName' rejects, so an
+-- inspection surface can report metrics for foreign queues. Same SQL, same
+-- decoder, same nullable eighth column.
+queueMetricsUnvalidated :: Statement Text QueueMetrics
+queueMetricsUnvalidated = preparable queueMetricsSql (E.param (E.nonNullable E.text)) (D.singleRow queueMetricsDecoder)
 
 -- | https://pgmq.github.io/pgmq/api/sql/functions/#metrics_all
 allQueueMetrics :: Statement () [QueueMetrics]
diff --git a/src/Pgmq/Hasql/Statements/Types.hs b/src/Pgmq/Hasql/Statements/Types.hs
--- a/src/Pgmq/Hasql/Statements/Types.hs
+++ b/src/Pgmq/Hasql/Statements/Types.hs
@@ -23,6 +23,9 @@
     -- FIFO read types (pgmq 1.8.0+)
     ReadGrouped (..),
     ReadGroupedWithPoll (..),
+    -- Non-destructive inspection (no upstream function)
+    PeekMessages (..),
+    LookupMessage (..),
     -- Topic types (pgmq 1.11.0+)
     BindTopic (..),
     UnbindTopic (..),
@@ -36,8 +39,8 @@
   )
 where
 
-import Data.Aeson (Value)
-import Pgmq.Hasql.Prelude
+import Data.Aeson (KeyValue, Value, object, pairs, (.=))
+import Pgmq.Hasql.Prelude hiding ((.=))
 import Pgmq.Types (MessageBody, MessageHeaders, MessageId, QueueName, RoutingKey, TopicPattern)
 
 type Delay = Int32
@@ -318,3 +321,66 @@
     throttleIntervalMs :: !Int32
   }
   deriving stock (Generic)
+
+-- | A keyset page over a queue table (@pgmq.q_\<name\>@) or, through
+-- @peekArchivedMessages@, an archive table (@pgmq.a_\<name\>@).
+--
+-- @unvalidatedQueueName@ is any name the server accepts, including names
+-- 'Pgmq.Types.parseQueueName' rejects: an inspection surface must show what
+-- exists. Pass 'Pgmq.Types.queueNameToText' when you hold a validated name.
+-- The physical table is resolved by @pgmq.format_table_name@ on the server.
+--
+-- @afterMessageId@ is an /exclusive/ cursor: the page starts strictly after
+-- it; 'Nothing' starts at the beginning. @limit@ is passed straight to SQL
+-- @LIMIT@, so it must be positive (@0@ returns nothing; a negative value is a
+-- server error). Pages are ordered by @msg_id@ ascending and never use
+-- @OFFSET@, so deleting earlier rows does not shift the page. Each call has
+-- its own snapshot: lower IDs committed or archived later require a refresh
+-- from the first page. This is not a replay feed. @LIMIT@ bounds rows returned,
+-- not scan work or payload bytes; partitioned layouts may scan and sort. Use
+-- finite statement/lock deadlines and inspection capacity separate from workers.
+data PeekMessages = PeekMessages
+  { unvalidatedQueueName :: !Text,
+    afterMessageId :: !(Maybe MessageId),
+    limit :: !Int32
+  }
+  deriving stock (Generic)
+
+-- | One message by id in a queue table or, through @lookupArchivedMessage@, an
+-- archive table. The name rules are those of t'PeekMessages'.
+data LookupMessage = LookupMessage
+  { unvalidatedQueueName :: !Text,
+    messageId :: !MessageId
+  }
+  deriving stock (Generic)
+
+-- JSON wire encodings ---------------------------------------------------------
+
+-- | Wire keys are the @pgmq.metrics_result@ columns. Two of them are easy to
+-- confuse and must not be: @queue_length@ is the total number of rows in the
+-- queue table, and @queue_visible_length@ is the subset whose visibility
+-- timeout has expired and are therefore available to a reader (see
+-- @docs/design/002-queue-visible-length.md@). @default_partition_length@ is
+-- @null@ when the metric is unavailable (PGMQ 1.12) or inapplicable (a
+-- non-partitioned queue); @null@ never means zero. Encode-only. The encoding is
+-- a published compatibility surface; see
+-- @docs/design/020-json-wire-encodings.md@.
+--
+-- The pattern is positional because @queueName@ is a duplicate record field in
+-- this module and its selector is ambiguous under GHC 9.12.
+queueMetricsFields :: (KeyValue e kv) => QueueMetrics -> [kv]
+queueMetricsFields (QueueMetrics qn len newest oldest total scrape visible dpl) =
+  [ "queue_name" .= qn,
+    "queue_length" .= len,
+    "newest_msg_age_sec" .= newest,
+    "oldest_msg_age_sec" .= oldest,
+    "total_messages" .= total,
+    "scrape_time" .= scrape,
+    "queue_visible_length" .= visible,
+    "default_partition_length" .= dpl
+  ]
+
+-- | JSON wire contract: @docs/design/020-json-wire-encodings.md@.
+instance ToJSON QueueMetrics where
+  toJSON = object . queueMetricsFields
+  toEncoding = pairs . mconcat . queueMetricsFields
diff --git a/test/AliasingSpec.hs b/test/AliasingSpec.hs
--- a/test/AliasingSpec.hs
+++ b/test/AliasingSpec.hs
@@ -40,6 +40,7 @@
     migrationPlan,
     runMigrationPlan,
   )
+import EphemeralDb (ephemeralConfig)
 import EphemeralPg qualified as Pg
 import Hasql.Decoders qualified as D
 import Hasql.Pool qualified as Pool
@@ -169,7 +170,8 @@
 -- installed, exactly as @EphemeralDb@ does for the shared one.
 acquireDb :: IO (Pg.Database, Pool.Pool)
 acquireDb = do
-  started <- Pg.startCached Pg.defaultConfig Pg.defaultCacheConfig
+  config <- ephemeralConfig
+  started <- Pg.startCached config Pg.defaultCacheConfig
   db <- either (\err -> error ("could not start a dedicated PostgreSQL: " <> show err)) pure started
   component <- either (error . ("Invalid PGMQ migration component: " <>) . show) pure Migration.pgmqMigrations
   plan <- either (error . ("Invalid PGMQ migration plan: " <>) . show) pure (migrationPlan (component :| []))
diff --git a/test/EphemeralDb.hs b/test/EphemeralDb.hs
--- a/test/EphemeralDb.hs
+++ b/test/EphemeralDb.hs
@@ -6,6 +6,7 @@
     withPgmqDb,
     withPgmqPool,
     ephemeralConfig,
+    ensureOwnerOnlyRoot,
 
     -- * Test fixtures
     TestFixture (..),
@@ -17,7 +18,8 @@
   )
 where
 
-import Control.Monad (filterM, when)
+import Control.Monad (filterM, unless, when)
+import Data.Bits ((.&.))
 import Data.List.NonEmpty (NonEmpty (..))
 import Data.Monoid (Last (..))
 import Data.Text qualified as T
@@ -42,8 +44,11 @@
 import Hasql.Session qualified as Session
 import Pgmq.Migration qualified as Migration
 import Pgmq.Types (QueueName, parseQueueName)
-import System.Directory (createDirectoryIfMissing, doesFileExist)
+import System.Directory (doesFileExist)
 import System.Environment (lookupEnv)
+import System.IO.Error (catchIOError, isAlreadyExistsError)
+import System.Posix.Directory qualified as Posix
+import System.Posix.Files (deviceID, fileID, fileMode, fileOwner, getSymbolicLinkStatus, isDirectory, setFileMode)
 import System.Posix.User (getEffectiveUserID)
 import System.Random (randomRIO)
 
@@ -62,11 +67,31 @@
   uid <- getEffectiveUserID
   pure ("/tmp/ephpg-pgmq-hs-" <> show uid)
 
+-- | Establish a real effective-uid-owned root without deleting its contents.
+ensureOwnerOnlyRoot :: FilePath -> IO ()
+ensureOwnerOnlyRoot root = do
+  catchIOError (Posix.createDirectory root 0o700) $ \err ->
+    unless (isAlreadyExistsError err) (ioError err)
+  uid <- getEffectiveUserID
+  before <- getSymbolicLinkStatus root
+  unless (isDirectory before && fileOwner before == uid) $
+    ioError (userError ("Unsafe ephemeral PostgreSQL root: " <> root))
+  setFileMode root 0o700
+  after <- getSymbolicLinkStatus root
+  unless
+    ( isDirectory after
+        && fileOwner after == uid
+        && deviceID before == deviceID after
+        && fileID before == fileID after
+        && fileMode after .&. 0o7777 == 0o700
+    )
+    (ioError (userError ("Ephemeral PostgreSQL root changed or is not owner-only: " <> root)))
+
 -- | Cached-startup configuration pinned to 'ephemeralRoot'.
 ephemeralConfig :: IO Config
 ephemeralConfig = do
   root <- ephemeralRoot
-  createDirectoryIfMissing True root
+  ensureOwnerOnlyRoot root
   pure defaultConfig {temporaryRoot = Last (Just root)}
 
 -- | Run an action with a temporary PostgreSQL database that has pgmq schema installed.
diff --git a/test/InspectionCostSpec.hs b/test/InspectionCostSpec.hs
new file mode 100644
--- /dev/null
+++ b/test/InspectionCostSpec.hs
@@ -0,0 +1,120 @@
+{-# LANGUAGE OverloadedStrings #-}
+
+-- | Complete projections on real upstream layouts, without installing indexes.
+-- Set PGMQ_INSPECTION_PLANS to an absolute directory to retain JSON EXPLAINs.
+module InspectionCostSpec (tests) where
+
+import Control.Exception (finally)
+import Control.Monad (forM_, when)
+import Data.Aeson (Value, object, (.=))
+import Data.Aeson qualified as Aeson
+import Data.Int (Int64)
+import Data.Text (Text)
+import Data.Text qualified as T
+import Data.Vector qualified as V
+import EphemeralDb (TestFixture (..), withTestFixture)
+import Hasql.Decoders qualified as D
+import Hasql.Pool qualified as Pool
+import Hasql.Session (Session, statement)
+import Hasql.Session qualified as Session
+import Hasql.Statement (toSql, unpreparable)
+import Pgmq.Hasql.Sessions qualified as Sessions
+import Pgmq.Hasql.Statements.Inspection qualified as Inspect
+import Pgmq.Hasql.Statements.Types (CreatePartitionedQueue (..), LookupMessage (LookupMessage), PeekMessages (PeekMessages))
+import Pgmq.Types (ArchivedMessage (..), Message (..), MessageId (..), queueNameToText)
+import System.Directory (createDirectoryIfMissing)
+import System.Environment (lookupEnv)
+import Test.Tasty (TestTree, testGroup)
+import Test.Tasty.HUnit (assertBool, assertFailure, testCase, (@?=))
+import TestUtils (assertSession, cleanupQueue)
+
+tests :: Pool.Pool -> TestTree
+tests pool =
+  testGroup "Inspection cost and partition layouts" $
+    map (layoutTest pool) ["ordinary", "unlogged", "numeric", "time"]
+
+layoutTest :: Pool.Pool -> Text -> TestTree
+layoutTest p layout = testCase (T.unpack layout <> ": 100000 live and archive rows, complete projections") $ do
+  available <- assertSession p (scalarBool "select exists (select from pg_extension where extname = 'pg_partman')")
+  required <- (== Just "1") <$> lookupEnv "PGMQ_REQUIRE_PARTMAN"
+  let partitioned = layout == "numeric" || layout == "time"
+  if partitioned && not available
+    then if required then assertFailure "required pg_partman is absent" else putStrLn "    SKIPPED: pg_partman is not installed"
+    else withTestFixture p $ \TestFixture {pool, queueName} -> flip finally (cleanupQueue pool queueName) $ do
+      case layout of
+        "ordinary" -> assertSession pool (Sessions.createQueue queueName)
+        "unlogged" -> assertSession pool (Sessions.createUnloggedQueue queueName)
+        "numeric" -> assertSession pool (Sessions.createPartitionedQueue (CreatePartitionedQueue queueName "10000" "1000000"))
+        "time" -> assertSession pool (Sessions.createPartitionedQueue (CreatePartitionedQueue queueName "1 day" "1000 days"))
+        _ -> assertFailure "unknown test layout"
+      let q = queueNameToText queueName
+      forM_ ["q", "a"] $ \prefix -> do
+        table <- assertSession pool (statement (q, prefix) Inspect.formatTableName)
+        let qualified = "pgmq." <> Inspect.quoteIdentifier table
+            timestamp = if layout == "time" then "date_trunc('day', now()) + case n % 3 when 0 then interval '-1 day' when 1 then interval '0 days' else interval '100 years' end" else "now()"
+            columns = "msg_id, vt, enqueued_at, message, headers" <> if prefix == "a" then ", archived_at" else ""
+            values = "n, now(), " <> timestamp <> ", jsonb_build_object('n', n, 'body', repeat('x', 128)), '{\"fixture\":true}'::jsonb" <> if prefix == "a" then ", " <> timestamp else ""
+        assertSession pool $
+          Session.script $
+            "insert into " <> qualified <> " (" <> columns <> ") overriding system value select " <> values <> " from generate_series(1, 100000) n; analyze " <> qualified
+        when partitioned $ do
+          (leaves, hasDefault) <-
+            assertSession pool $
+              statement () $
+                unpreparable
+                  ("select count(distinct t.tableoid), bool_or(pg_get_expr(c.relpartbound, c.oid) = 'DEFAULT') from " <> qualified <> " t join pg_class c on c.oid = t.tableoid")
+                  mempty
+                  (D.singleRow ((,) <$> D.column (D.nonNullable D.int8) <*> D.column (D.nonNullable D.bool)))
+          assertBool "rows occupy multiple leaves" (leaves >= (3 :: Int64))
+          assertBool "rows occupy the default partition" hasDefault
+        verifyReads pool q prefix
+        let pageSql = if prefix == "q" then toSql (Inspect.peekStatement table) else toSql (Inspect.peekArchivedStatement table)
+            lookupSql = if prefix == "q" then toSql (Inspect.lookupStatement table) else toSql (Inspect.lookupArchivedStatement table)
+            queries :: [(Text, Text)]
+            queries =
+              [ ("first", T.replace "$2" "50" (T.replace "$1" "NULL" pageSql)),
+                ("late", T.replace "$2" "50" (T.replace "$1" "99950" pageSql)),
+                ("present", T.replace "$1" "99999" lookupSql),
+                ("absent", T.replace "$1" "100001" lookupSql)
+              ]
+        plans <-
+          mapM
+            ( \(label, sql) -> do
+                plan <- assertSession pool (explain sql)
+                pure (object ["case" .= label, "sql" .= sql, "explain" .= plan])
+            )
+            queries
+        output <- lookupEnv "PGMQ_INSPECTION_PLANS"
+        case output of
+          Nothing -> pure ()
+          Just directory -> do
+            createDirectoryIfMissing True directory
+            version <- maybe "1.13.0" id <$> lookupEnv "PGMQ_TEST_SCHEMA_VERSION"
+            postgres <- assertSession pool (scalarText "select version()")
+            partman <- assertSession pool (scalarText "select coalesce((select extversion from pg_extension where extname = 'pg_partman'), 'absent')")
+            Aeson.encodeFile (directory <> "/" <> version <> "-" <> T.unpack layout <> "-" <> T.unpack prefix <> ".json") $
+              object ["schema" .= version, "postgres" .= postgres, "pg_partman" .= partman, "layout" .= layout, "table" .= table, "rows" .= (100000 :: Int), "plans" .= plans]
+
+verifyReads :: Pool.Pool -> Text -> Text -> IO ()
+verifyReads pool queue prefix = do
+  let page cursor =
+        if prefix == "q"
+          then map messageId . V.toList <$> assertSession pool (Sessions.peekMessages (PeekMessages queue cursor 50))
+          else map (messageId . archivedMessage) . V.toList <$> assertSession pool (Sessions.peekArchivedMessages (PeekMessages queue cursor 50))
+      lookupId mid =
+        if prefix == "q"
+          then fmap messageId <$> assertSession pool (Sessions.lookupMessage (LookupMessage queue mid))
+          else fmap (messageId . archivedMessage) <$> assertSession pool (Sessions.lookupArchivedMessage (LookupMessage queue mid))
+  page Nothing >>= (@?= map MessageId [1 .. 50])
+  page (Just (MessageId 99950)) >>= (@?= map MessageId [99951 .. 100000])
+  lookupId (MessageId 99999) >>= (@?= Just (MessageId 99999))
+  lookupId (MessageId 100001) >>= (@?= Nothing)
+
+explain :: Text -> Session Value
+explain sql = statement () $ unpreparable ("explain (analyze, buffers, format json) " <> sql) mempty (D.singleRow (D.column (D.nonNullable D.json)))
+
+scalarBool :: Text -> Session Bool
+scalarBool sql = statement () (unpreparable sql mempty (D.singleRow (D.column (D.nonNullable D.bool))))
+
+scalarText :: Text -> Session Text
+scalarText sql = statement () (unpreparable sql mempty (D.singleRow (D.column (D.nonNullable D.text))))
diff --git a/test/InspectionForeignNameSpec.hs b/test/InspectionForeignNameSpec.hs
new file mode 100644
--- /dev/null
+++ b/test/InspectionForeignNameSpec.hs
@@ -0,0 +1,119 @@
+{-# LANGUAGE OverloadedStrings #-}
+
+-- | IR-1 acceptance 5: every inspection read works for queues whose names
+-- 'parseQueueName' rejects, because an inspection surface sees whatever exists
+-- in the database, not only what this library created.
+--
+-- The names are created and fed through raw SQL (the Haskell API rightly
+-- refuses them) and read back through the new sessions. Runs on a dedicated
+-- PostgreSQL instance: a mixed-case row in pgmq.meta poisons the typed
+-- listQueues decoding for every test sharing a database.
+module InspectionForeignNameSpec (tests) where
+
+import Control.Monad (void)
+import Data.Int (Int64)
+import Data.List.NonEmpty (NonEmpty (..))
+import Data.Text (Text)
+import Data.Text qualified as T
+import Data.Vector qualified as V
+import Data.Word (Word32)
+import Database.PostgreSQL.Migrate (defaultRunOptions, migrationPlan, runMigrationPlan)
+import EphemeralDb (ephemeralConfig)
+import EphemeralPg qualified as Pg
+import Hasql.Decoders qualified as D
+import Hasql.Pool qualified as Pool
+import Hasql.Pool.Config qualified as PoolConfig
+import Hasql.Session (Session, statement)
+import Hasql.Statement (unpreparable)
+import Pgmq.Hasql.Sessions qualified as Sessions
+import Pgmq.Hasql.Statements.Types (LookupMessage (LookupMessage), PeekMessages (PeekMessages), QueueMetrics (..))
+import Pgmq.Migration qualified as Migration
+import Pgmq.Types (ArchivedMessage (..), Message (..), MessageId (..), parseQueueName)
+import System.Random (randomRIO)
+import Test.Tasty (TestTree, testGroup, withResource)
+import Test.Tasty.HUnit (assertEqual, assertFailure, testCase, (@?=))
+
+tests :: TestTree
+tests =
+  withResource acquireDb releaseDb $ \getDb ->
+    testGroup
+      "Inspection of foreign queue names"
+      [ testForeignName getDb "mixed-case" ("MyQueue_" <>),
+        testForeignName getDb "hyphenated" ("odd-name_" <>)
+      ]
+
+testForeignName :: IO (Pg.Database, Pool.Pool) -> String -> (Text -> Text) -> TestTree
+testForeignName getDb label mkName = testCase (label <> " name: peek, lookup, archive, and metrics work") $ do
+  (_, pool) <- getDb
+  suffix <- T.pack . show <$> randomRIO (10000 :: Word32, 99999)
+  let name = mkName suffix
+  -- The whole point: this library's validator refuses the name.
+  case parseQueueName name of
+    Left _ -> pure ()
+    Right _ -> assertFailure ("expected parseQueueName to reject " <> T.unpack name)
+  assertSession pool (rawUnit ("select pgmq.create('" <> name <> "')"))
+  ids <- mapM (\i -> MessageId <$> assertSession pool (rawId ("select pgmq.send('" <> name <> "', '{\"i\":" <> T.pack (show i) <> "}'::jsonb)"))) [1 .. 3 :: Int]
+  firstId <- case ids of
+    x : _ -> pure x
+    [] -> assertFailure "expected three ids"
+  page <- assertSession pool (Sessions.peekMessages (PeekMessages name Nothing 10))
+  map messageId (V.toList page) @?= ids
+  one <- assertSession pool (Sessions.lookupMessage (LookupMessage name firstId))
+  fmap messageId one @?= Just firstId
+  metrics <- assertSession pool (Sessions.queueMetricsUnvalidated name)
+  queueLength metrics @?= 3
+  assertEqual "metrics report the name as given" name (queueName metrics)
+  void $ assertSession pool (rawBool ("select pgmq.archive('" <> name <> "', " <> T.pack (show (unMessageId firstId)) <> "::bigint)"))
+  archived <- assertSession pool (Sessions.peekArchivedMessages (PeekMessages name Nothing 10))
+  map (messageId . archivedMessage) (V.toList archived) @?= [firstId]
+  found <- assertSession pool (Sessions.lookupArchivedMessage (LookupMessage name firstId))
+  fmap (messageId . archivedMessage) found @?= Just firstId
+  remaining <- assertSession pool (Sessions.peekMessages (PeekMessages name Nothing 10))
+  map messageId (V.toList remaining) @?= drop 1 ids
+  void $ assertSession pool (rawBool ("select pgmq.drop_queue('" <> name <> "')"))
+
+-- Dedicated database plumbing (as in MixedCaseRemediationSpec) -----------------
+
+acquireDb :: IO (Pg.Database, Pool.Pool)
+acquireDb = do
+  config <- ephemeralConfig
+  started <- Pg.startCached config Pg.defaultCacheConfig
+  db <- either (\err -> error ("could not start a dedicated PostgreSQL: " <> show err)) pure started
+  component <- either (error . ("Invalid PGMQ migration component: " <>) . show) pure Migration.pgmqMigrations
+  plan <- either (error . ("Invalid PGMQ migration plan: " <>) . show) pure (migrationPlan (component :| []))
+  installResult <- runMigrationPlan defaultRunOptions (Pg.connectionSettings db) plan
+  case installResult of
+    Left migrationErr -> error $ "Migration failed: " <> show migrationErr
+    Right _ -> pure ()
+  pool <-
+    Pool.acquire $
+      PoolConfig.settings
+        [ PoolConfig.size 2,
+          PoolConfig.staticConnectionSettings (Pg.connectionSettings db)
+        ]
+  pure (db, pool)
+
+releaseDb :: (Pg.Database, Pool.Pool) -> IO ()
+releaseDb (db, pool) = do
+  Pool.release pool
+  Pg.stop db
+
+-- Raw statement helpers. Names are spliced into SQL text because these tests
+-- must construct names the Haskell API refuses; every spliced value is built
+-- above from [A-Za-z0-9_-], so splicing is safe here.
+
+rawUnit :: Text -> Session ()
+rawUnit sqlText = statement () (unpreparable sqlText mempty D.noResult)
+
+rawBool :: Text -> Session Bool
+rawBool sqlText = statement () (unpreparable sqlText mempty (D.singleRow (D.column (D.nonNullable D.bool))))
+
+rawId :: Text -> Session Int64
+rawId sqlText = statement () (unpreparable sqlText mempty (D.singleRow (D.column (D.nonNullable D.int8))))
+
+assertSession :: Pool.Pool -> Session a -> IO a
+assertSession pool session = do
+  result <- Pool.use pool session
+  case result of
+    Left err -> assertFailure $ "Session failed: " <> show err
+    Right a -> pure a
diff --git a/test/InspectionSpec.hs b/test/InspectionSpec.hs
new file mode 100644
--- /dev/null
+++ b/test/InspectionSpec.hs
@@ -0,0 +1,364 @@
+{-# LANGUAGE OverloadedStrings #-}
+
+-- | IR-1: reads that observe a queue or archive without leasing anything.
+--
+-- The non-destructive guarantee is asserted two ways: the raw cells every
+-- lease touches (vt, read_ct, last_read_at) are byte-identical before and
+-- after a sequence of peeks, and a consumer polling the queue concurrently
+-- with a loop of peeks still receives every message exactly once with
+-- read_ct = 1. Paging is keyset-only: the statements' SQL contains no OFFSET,
+-- and paging by the last id seen visits every message in a quiescent fixture
+-- exactly once. Late commits and late archival require a first-page refresh.
+module InspectionSpec (tests) where
+
+import Control.Concurrent (forkFinally, killThread, threadDelay)
+import Control.Concurrent.MVar (newEmptyMVar, putMVar, takeMVar)
+import Control.Exception (IOException, bracket, finally, mask, throwIO, try)
+import Control.Monad (forM_, replicateM_, void)
+import Data.Aeson (object, (.=))
+import Data.Int (Int32, Int64)
+import Data.List (sort)
+import Data.Set qualified as Set
+import Data.Text (Text)
+import Data.Text qualified as T
+import Data.Time (UTCTime)
+import Data.Vector qualified as V
+import EphemeralDb (Database, TestFixture (..), withTestFixture)
+import EphemeralPg qualified as Pg
+import Hasql.Connection qualified as Connection
+import Hasql.Decoders qualified as D
+import Hasql.Encoders qualified as E
+import Hasql.Pool qualified as Pool
+import Hasql.Session (Session, statement)
+import Hasql.Session qualified as Session
+import Hasql.Statement (preparable, toSql, unpreparable)
+import Pgmq.Hasql.Sessions qualified as Sessions
+import Pgmq.Hasql.Statements.Inspection qualified as Inspect
+import Pgmq.Hasql.Statements.Types
+  ( BatchMessageQuery (..),
+    BatchSendMessage (..),
+    CreatePartitionedQueue (..),
+    LookupMessage (LookupMessage),
+    PeekMessages (PeekMessages),
+    QueueMetrics (..),
+    ReadMessage (..),
+    SendMessage (..),
+  )
+import Pgmq.Types (ArchivedMessage (..), Message (..), MessageBody (..), MessageId (..), QueueName, queueNameToText)
+import System.Environment (lookupEnv)
+import System.Timeout (timeout)
+import Test.Tasty (TestTree, testGroup)
+import Test.Tasty.HUnit (assertBool, assertEqual, assertFailure, testCase, (@?=))
+import TestUtils (assertSession, cleanupQueue)
+
+tests :: Pool.Pool -> Database -> TestTree
+tests p db =
+  testGroup
+    "Inspection (non-destructive reads)"
+    [ testPeekLeavesRowsUntouched p,
+      testPeekDoesNotDisturbConsumer p,
+      testConsumerFailure,
+      testLateCommit p db,
+      testLateArchive p,
+      testKeysetPagingVisitsEachOnce p,
+      testNoOffset,
+      testArchiveReads p,
+      testLookups p,
+      testLimitSemantics p,
+      testMissingQueue p,
+      testMetricsParity p,
+      testPartitionedQueue p
+    ]
+
+-- | The cells a lease would change, for every row, in id order.
+rowImages :: Text -> Session [(Int64, Int32, Maybe UTCTime, UTCTime)]
+rowImages table = statement () (unpreparable sql mempty decoder)
+  where
+    -- Test queue names are generated from [a-z0-9_], so splicing is safe here.
+    sql = "select msg_id, read_ct, last_read_at, vt from pgmq.q_" <> table <> " order by msg_id"
+    decoder =
+      D.rowList
+        ( (,,,)
+            <$> D.column (D.nonNullable D.int8)
+            <*> D.column (D.nonNullable D.int4)
+            <*> D.column (D.nullable D.timestamptz)
+            <*> D.column (D.nonNullable D.timestamptz)
+        )
+
+sendN :: Pool.Pool -> QueueName -> Int -> IO [MessageId]
+sendN pool q n =
+  assertSession pool $
+    Sessions.batchSendMessage
+      BatchSendMessage
+        { queueName = q,
+          messageBodies = [MessageBody (object ["n" .= i]) | i <- [1 .. n]],
+          delay = Nothing
+        }
+
+peek :: Pool.Pool -> QueueName -> Maybe MessageId -> Int32 -> IO [Message]
+peek pool q cursor n =
+  V.toList <$> assertSession pool (Sessions.peekMessages (PeekMessages (queueNameToText q) cursor n))
+
+lastId :: [Message] -> IO MessageId
+lastId [] = assertFailure "expected a non-empty page"
+lastId page = pure (messageId (last page))
+
+testPeekLeavesRowsUntouched :: Pool.Pool -> TestTree
+testPeekLeavesRowsUntouched p = testCase "peek leaves vt, read_ct, and last_read_at byte-identical" $
+  withTestFixture p $ \TestFixture {pool, queueName} -> flip finally (cleanupQueue pool queueName) $ do
+    assertSession pool (Sessions.createQueue queueName)
+    _ <- sendN pool queueName 3
+    -- Two more with a delay, so vt differs between rows.
+    _ <- assertSession pool (Sessions.sendMessage (SendMessage queueName (MessageBody "late") (Just 60)))
+    _ <- assertSession pool (Sessions.sendMessage (SendMessage queueName (MessageBody "later") (Just 120)))
+    -- Lease one through the real read so a non-trivial read_ct and last_read_at exist.
+    leased <- assertSession pool (Sessions.readMessage (ReadMessage queueName 30 (Just 1) Nothing))
+    V.length leased @?= 1
+    before <- assertSession pool (rowImages (queueNameToText queueName))
+    replicateM_ 5 $ do
+      first <- peek pool queueName Nothing 2
+      cursor <- lastId first
+      void (peek pool queueName (Just cursor) 50)
+    _ <- assertSession pool (Sessions.lookupMessage (LookupMessage (queueNameToText queueName) (messageId (V.head leased))))
+    after <- assertSession pool (rowImages (queueNameToText queueName))
+    assertEqual "rows are identical after peeks and lookups" before after
+    length before @?= 5
+
+testPeekDoesNotDisturbConsumer :: Pool.Pool -> TestTree
+testPeekDoesNotDisturbConsumer p = testCase "a concurrently polling consumer sees every message exactly once" $
+  withTestFixture p $ \TestFixture {pool, queueName} -> flip finally (cleanupQueue pool queueName) $ do
+    assertSession pool (Sessions.createQueue queueName)
+    sent <- sendN pool queueName 50
+    -- The bracketed worker propagates failures and is cancelled if peeking fails.
+    let consume seen empties
+          | Set.size seen >= 50 || empties >= (200 :: Int) = pure seen
+          | otherwise = do
+              batch <- assertSession pool (Sessions.readMessage (ReadMessage queueName 60 (Just 5) Nothing))
+              if V.null batch
+                then threadDelay 5_000 >> consume seen (empties + 1)
+                else consume (foldr (Set.insert . messageId) seen (V.toList batch)) empties
+    seen <-
+      withWorker (consume Set.empty 0) $
+        replicateM_ 200 (void (peek pool queueName Nothing 50))
+    assertEqual "the consumer received exactly the fifty sent ids" (Set.fromList sent) seen
+    rows <- assertSession pool (rowImages (queueNameToText queueName))
+    length rows @?= 50
+    forM_ rows $ \(mid, readCt, _, _) ->
+      assertEqual ("read_ct of " <> show mid <> " was bumped only by the consumer") 1 readCt
+    -- Everything is leased for sixty seconds now; a further read returns nothing,
+    -- which a destructive "peek" would have made impossible to predict.
+    more <- assertSession pool (Sessions.readMessage (ReadMessage queueName 60 (Just 50) Nothing))
+    V.length more @?= 0
+
+-- | Base-only structured concurrency: every exit cancels the worker, and its
+-- exception is rethrown in the parent rather than leaving an empty result MVar.
+withWorker :: IO a -> IO () -> IO a
+withWorker work during = mask $ \restore -> do
+  result <- newEmptyMVar
+  bracket
+    (forkFinally (restore work) (putMVar result))
+    killThread
+    (\_ -> restore (during >> takeMVar result >>= either throwIO pure))
+
+testConsumerFailure :: TestTree
+testConsumerFailure = testCase "consumer exceptions reach the parent promptly" $ do
+  outcome <-
+    timeout 1_000_000 $
+      try @IOException (withWorker (ioError (userError "forced consumer failure")) (pure ()))
+  case outcome of
+    Just (Left err) -> assertBool "original worker failure" ("forced consumer failure" `T.isInfixOf` T.pack (show err))
+    Just (Right ()) -> assertFailure "worker failure was swallowed"
+    Nothing -> assertFailure "parent hung waiting for a failed consumer"
+
+testLateCommit :: Pool.Pool -> Database -> TestTree
+testLateCommit p db = testCase "a late lower-id commit needs a first-page refresh" $
+  withTestFixture p $ \TestFixture {pool, queueName} -> flip finally (cleanupQueue pool queueName) $ do
+    assertSession pool (Sessions.createQueue queueName)
+    -- Hold the first producer transaction on its own connection. The pool's
+    -- second producer commits before the first, without timing-based sleeps.
+    bracket
+      (Connection.acquire (Pg.connectionSettings db) >>= either (assertFailure . show) pure)
+      Connection.release
+      $ \producer -> do
+        let run :: Session a -> IO a
+            run session = Connection.use producer session >>= either (assertFailure . show) pure
+        run (Session.script "BEGIN")
+        older <- run (Sessions.sendMessage (SendMessage queueName (MessageBody "older") Nothing))
+        newer <- assertSession pool (Sessions.sendMessage (SendMessage queueName (MessageBody "newer") Nothing))
+        assertBool "allocation order differs from commit order" (older < newer)
+        first <- peek pool queueName Nothing 10
+        map messageId first @?= [newer]
+        run (Session.script "COMMIT")
+        continued <- peek pool queueName (Just newer) 10
+        map messageId continued @?= []
+        refreshed <- peek pool queueName Nothing 10
+        map messageId refreshed @?= [older, newer]
+
+testLateArchive :: Pool.Pool -> TestTree
+testLateArchive p = testCase "archiving an older id needs a first-page archive refresh" $
+  withTestFixture p $ \TestFixture {pool, queueName} -> flip finally (cleanupQueue pool queueName) $ do
+    assertSession pool (Sessions.createQueue queueName)
+    ids <- sendN pool queueName 2
+    (older, newer) <- case ids of
+      [a, b] -> pure (a, b)
+      other -> assertFailure ("expected two ids: " <> show other)
+    let q = queueNameToText queueName
+        archive mid = void (assertSession pool (Sessions.batchArchiveMessages (BatchMessageQuery queueName [mid])))
+        page cursor = map (messageId . archivedMessage) . V.toList <$> assertSession pool (Sessions.peekArchivedMessages (PeekMessages q cursor 10))
+    archive newer
+    page Nothing >>= (@?= [newer])
+    archive older
+    page (Just newer) >>= (@?= [])
+    page Nothing >>= (@?= [older, newer])
+
+testKeysetPagingVisitsEachOnce :: Pool.Pool -> TestTree
+testKeysetPagingVisitsEachOnce p = testCase "paging by the last id visits every message exactly once" $
+  withTestFixture p $ \TestFixture {pool, queueName} -> flip finally (cleanupQueue pool queueName) $ do
+    assertSession pool (Sessions.createQueue queueName)
+    sent <- concat <$> mapM (const (sendN pool queueName 100)) [1 .. 10 :: Int]
+    length sent @?= 1000
+    before <- assertSession pool (rowImages (queueNameToText queueName))
+    let go cursor pages acc = do
+          page <- peek pool queueName cursor 7
+          let acc' = acc <> map messageId page
+          if length page < 7
+            then pure (pages + 1 :: Int, acc')
+            else do
+              next <- lastId page
+              go (Just next) (pages + 1) acc'
+    (pages, visited) <- go Nothing 0 []
+    assertEqual "143 pages of at most seven" 143 pages
+    assertEqual "every id exactly once, in order" (sort sent) visited
+    assertEqual "no duplicates" (length visited) (Set.size (Set.fromList visited))
+    after <- assertSession pool (rowImages (queueNameToText queueName))
+    assertEqual "paging modified nothing" before after
+
+testNoOffset :: TestTree
+testNoOffset = testCase "inspection statements never use OFFSET" $ do
+  let sqls =
+        [ toSql (Inspect.peekStatement "q_x"),
+          toSql (Inspect.peekArchivedStatement "a_x"),
+          toSql (Inspect.lookupStatement "q_x"),
+          toSql (Inspect.lookupArchivedStatement "a_x")
+        ]
+  forM_ sqls $ \sql -> do
+    assertBool ("no offset in " <> T.unpack sql) (not ("offset" `T.isInfixOf` T.toLower sql))
+    assertBool ("quoted identifier in " <> T.unpack sql) ("pgmq.\"" `T.isInfixOf` sql)
+  assertBool "pages are ordered by msg_id" ("order by msg_id asc limit $2" `T.isInfixOf` toSql (Inspect.peekStatement "q_x"))
+  Inspect.quoteIdentifier "q_odd-name" @?= "\"q_odd-name\""
+  Inspect.quoteIdentifier "we\"ird" @?= "\"we\"\"ird\""
+
+testArchiveReads :: Pool.Pool -> TestTree
+testArchiveReads p = testCase "archive reads return archived messages with their archival timestamp" $
+  withTestFixture p $ \TestFixture {pool, queueName} -> flip finally (cleanupQueue pool queueName) $ do
+    assertSession pool (Sessions.createQueue queueName)
+    sent <- sendN pool queueName 3
+    archived <- assertSession pool (Sessions.batchArchiveMessages (BatchMessageQuery queueName (take 2 sent)))
+    sort archived @?= sort (take 2 sent)
+    let q = queueNameToText queueName
+    rows <- V.toList <$> assertSession pool (Sessions.peekArchivedMessages (PeekMessages q Nothing 10))
+    map (messageId . archivedMessage) rows @?= take 2 sent
+    forM_ rows $ \row ->
+      assertBool "archived_at is not before enqueued_at" (archivedAt row >= enqueuedAt (archivedMessage row))
+    live <- peek pool queueName Nothing 10
+    map messageId live @?= drop 2 sent
+    -- The archive pages by the same exclusive cursor.
+    second <- V.toList <$> assertSession pool (Sessions.peekArchivedMessages (PeekMessages q (Just (sent !! 0)) 10))
+    map (messageId . archivedMessage) second @?= [sent !! 1]
+
+testLookups :: Pool.Pool -> TestTree
+testLookups p = testCase "lookup finds a present message and answers Nothing for an absent one, in both tables" $
+  withTestFixture p $ \TestFixture {pool, queueName} -> flip finally (cleanupQueue pool queueName) $ do
+    assertSession pool (Sessions.createQueue queueName)
+    sent <- sendN pool queueName 2
+    (liveId, archivedId) <- case sent of
+      [a, b] -> pure (a, b)
+      other -> assertFailure ("expected two ids, got " <> show other)
+    _ <- assertSession pool (Sessions.batchArchiveMessages (BatchMessageQuery queueName [archivedId]))
+    let q = queueNameToText queueName
+    live <- assertSession pool (Sessions.lookupMessage (LookupMessage q liveId))
+    fmap messageId live @?= Just liveId
+    gone <- assertSession pool (Sessions.lookupMessage (LookupMessage q archivedId))
+    fmap messageId gone @?= Nothing
+    inArchive <- assertSession pool (Sessions.lookupArchivedMessage (LookupMessage q archivedId))
+    fmap (messageId . archivedMessage) inArchive @?= Just archivedId
+    notInArchive <- assertSession pool (Sessions.lookupArchivedMessage (LookupMessage q liveId))
+    fmap (messageId . archivedMessage) notInArchive @?= Nothing
+    never <- assertSession pool (Sessions.lookupMessage (LookupMessage q (MessageId 999_999_999)))
+    fmap messageId never @?= Nothing
+
+testLimitSemantics :: Pool.Pool -> TestTree
+testLimitSemantics p = testCase "limit bounds the page and a cursor excludes its own row" $
+  withTestFixture p $ \TestFixture {pool, queueName} -> flip finally (cleanupQueue pool queueName) $ do
+    assertSession pool (Sessions.createQueue queueName)
+    sent <- sendN pool queueName 5
+    three <- peek pool queueName Nothing 3
+    map messageId three @?= take 3 sent
+    rest <- peek pool queueName (Just (sent !! 2)) 3
+    map messageId rest @?= drop 3 sent
+    none <- peek pool queueName (Just (sent !! 4)) 3
+    map messageId none @?= []
+    zero <- peek pool queueName Nothing 0
+    map messageId zero @?= []
+
+testMissingQueue :: Pool.Pool -> TestTree
+testMissingQueue p = testCase "a missing queue fails with undefined_table (42P01)" $ do
+  result <- Pool.use p (Sessions.peekMessages (PeekMessages "no_such_queue_for_inspection" Nothing 1))
+  case result of
+    Right _ -> assertFailure "peeking a missing queue should fail"
+    Left err -> assertBool ("expected 42P01, got " <> show err) ("42P01" `T.isInfixOf` T.pack (show err))
+
+testMetricsParity :: Pool.Pool -> TestTree
+testMetricsParity p = testCase "queueMetricsUnvalidated matches queueMetrics for a validated name" $
+  withTestFixture p $ \TestFixture {pool, queueName} -> flip finally (cleanupQueue pool queueName) $ do
+    assertSession pool (Sessions.createQueue queueName)
+    _ <- sendN pool queueName 4
+    typed <- assertSession pool (Sessions.queueMetrics queueName)
+    lenient <- assertSession pool (Sessions.queueMetricsUnvalidated (queueNameToText queueName))
+    queueLength lenient @?= queueLength typed
+    queueVisibleLength lenient @?= queueVisibleLength typed
+    totalMessages lenient @?= totalMessages typed
+    defaultPartitionLength lenient @?= defaultPartitionLength typed
+    queueLength lenient @?= 4
+
+-- | A partitioned queue's physical table is a partitioned parent; a SELECT on
+-- it sees every partition, so the reads need no special case. Needs pg_partman.
+testPartitionedQueue :: Pool.Pool -> TestTree
+testPartitionedQueue p = testCase "peek and lookup work on a partitioned queue (needs pg_partman)" $
+  withPartman p $
+    withTestFixture p $ \TestFixture {pool, queueName} -> flip finally (cleanupQueue pool queueName) $ do
+      assertSession pool (Sessions.createPartitionedQueue (CreatePartitionedQueue queueName "10" "100"))
+      sent <- sendN pool queueName 12
+      page <- peek pool queueName Nothing 20
+      map messageId page @?= sent
+      one <- assertSession pool (Sessions.lookupMessage (LookupMessage (queueNameToText queueName) (sent !! 11)))
+      fmap messageId one @?= Just (sent !! 11)
+      -- The interval-10 archive also spans multiple leaves and supports the
+      -- same first/late cursor and by-id paths as the larger diagnostic fixture.
+      archivedIds <- assertSession pool (Sessions.batchArchiveMessages (BatchMessageQuery queueName sent))
+      sort archivedIds @?= sent
+      let q = queueNameToText queueName
+          archivePage cursor = map (messageId . archivedMessage) . V.toList <$> assertSession pool (Sessions.peekArchivedMessages (PeekMessages q cursor 5))
+      archivePage Nothing >>= (@?= take 5 sent)
+      archivePage (Just (sent !! 10)) >>= (@?= drop 11 sent)
+      found <- assertSession pool (Sessions.lookupArchivedMessage (LookupMessage q (sent !! 11)))
+      fmap (messageId . archivedMessage) found @?= Just (sent !! 11)
+      missing <- assertSession pool (Sessions.lookupArchivedMessage (LookupMessage q (MessageId 999999)))
+      fmap (messageId . archivedMessage) missing @?= Nothing
+
+withPartman :: Pool.Pool -> IO () -> IO ()
+withPartman pool action = do
+  available <-
+    assertSession pool $
+      statement () $
+        preparable
+          "select exists (select 1 from pg_extension where extname = 'pg_partman')"
+          E.noParams
+          (D.singleRow (D.column (D.nonNullable D.bool)))
+  required <- (== Just "1") <$> lookupEnv "PGMQ_REQUIRE_PARTMAN"
+  if available
+    then action
+    else
+      if required
+        then assertFailure "PGMQ_REQUIRE_PARTMAN=1 but pg_partman is not installed"
+        else putStrLn "    SKIPPED: pg_partman is not installed"
diff --git a/test/JsonGoldenSpec.hs b/test/JsonGoldenSpec.hs
new file mode 100644
--- /dev/null
+++ b/test/JsonGoldenSpec.hs
@@ -0,0 +1,76 @@
+{-# LANGUAGE OverloadedStrings #-}
+
+-- | Pins the wire shape of 'QueueMetrics' and proves that 'Message' survives
+-- encode-then-decode for arbitrary jsonb-storable bodies and headers. The
+-- property lives here rather than in pgmq-core because this suite already
+-- owns the hedgehog generators PostgreSQL accepts.
+module JsonGoldenSpec (tests) where
+
+import Data.Aeson (ToJSON (..), Value (Null), decode, encode)
+import Data.Time (UTCTime (..), fromGregorian)
+import Data.Time.Clock.POSIX (posixSecondsToUTCTime)
+import Generators (genJsonValue)
+import Hedgehog (Gen, forAll, property, (===))
+import Hedgehog.Gen qualified as Gen
+import Hedgehog.Range qualified as Range
+import Pgmq.Hasql.Statements.Types (QueueMetrics (..))
+import Pgmq.Types (ArchivedMessage (..), Message (..), MessageBody (..), MessageId (..))
+import Test.Tasty (TestTree, localOption, testGroup)
+import Test.Tasty.Golden (NoCreateFile (..), goldenVsString)
+import Test.Tasty.Hedgehog (testProperty)
+
+tests :: TestTree
+tests =
+  localOption (NoCreateFile True) $
+    testGroup
+      "JSON wire encodings"
+      [ goldenVsString "queue_metrics" "test/golden/queue_metrics.json" (pure (encode sampleMetrics <> "\n")),
+        goldenVsString "queue_metrics_partitioned" "test/golden/queue_metrics_partitioned.json" (pure (encode partitionedMetrics <> "\n")),
+        testProperty "Message round-trips modulo explicit-null header normalization" $ property $ do
+          m <- forAll genMessage
+          let normalized = m {headers = case headers m of Just Null -> Nothing; other -> other}
+          decode (encode m) === Just normalized
+          encode normalized === encode m
+          decode (encode m) === Just (toJSON m)
+          at <- forAll genTime
+          let archived = ArchivedMessage m at
+          decode (encode archived) === Just (ArchivedMessage normalized at)
+          decode (encode archived) === Just (toJSON archived)
+          encode (ArchivedMessage normalized at) === encode archived,
+        testProperty "QueueMetrics encoder parity" $ property $ do
+          newest <- forAll (Gen.maybe (Gen.int32 (Range.linear 0 maxBound)))
+          oldest <- forAll (Gen.maybe (Gen.int32 (Range.linear 0 maxBound)))
+          estimate <- forAll (Gen.maybe (Gen.int64 (Range.linear 0 maxBound)))
+          let m = QueueMetrics "orders" 120 newest oldest 9000 scrape 115 estimate
+          decode (encode m) === Just (toJSON m)
+      ]
+
+scrape :: UTCTime
+scrape = UTCTime (fromGregorian 2026 8 19) 45296.123
+
+-- | An empty queue: both ages are null, the partition estimate is null.
+sampleMetrics :: QueueMetrics
+sampleMetrics = QueueMetrics "orders" 0 Nothing Nothing 0 scrape 0 Nothing
+
+-- | A partitioned queue on PGMQ 1.13 with a planner estimate.
+partitionedMetrics :: QueueMetrics
+partitionedMetrics = QueueMetrics "events" 120 (Just 2) (Just 3600) 9000 scrape 115 (Just 7)
+
+-- | Whole milliseconds so the timestamp survives jsonb-independent rendering
+-- exactly; aeson itself round-trips picoseconds, but the property is about
+-- the record, not the clock.
+genTime :: Gen UTCTime
+genTime = do
+  millis <- Gen.integral (Range.linear 0 (4_102_444_800_000 :: Integer))
+  pure (posixSecondsToUTCTime (fromIntegral millis / 1000))
+
+genMessage :: Gen Message
+genMessage =
+  Message
+    <$> (MessageId <$> Gen.integral (Range.linear 1 maxBound))
+    <*> genTime
+    <*> genTime
+    <*> Gen.maybe genTime
+    <*> Gen.integral (Range.linear 0 1000)
+    <*> (MessageBody <$> genJsonValue)
+    <*> Gen.maybe genJsonValue
diff --git a/test/Main.hs b/test/Main.hs
--- a/test/Main.hs
+++ b/test/Main.hs
@@ -6,7 +6,11 @@
 import AliasingSpec qualified
 import AllFunctionsDecoderSpec qualified
 import DecoderValidationSpec qualified
-import EphemeralDb (withPgmqDb)
+import EphemeralDb (ensureOwnerOnlyRoot, withPgmqDb)
+import InspectionCostSpec qualified
+import InspectionForeignNameSpec qualified
+import InspectionSpec qualified
+import JsonGoldenSpec qualified
 import MessageSpec qualified
 import MetricsSpec qualified
 import MixedCaseRemediationSpec qualified
@@ -14,7 +18,9 @@
 import NotifyRaceSpec qualified
 import NullBodySpec qualified
 import NullSemanticsSpec qualified
+import PartitionedNotifySpec qualified
 import QueueSpec qualified
+import RootPermissionsSpec qualified
 import RoundTripSpec qualified
 import SchemaSpec qualified
 import System.Environment (lookupEnv)
@@ -30,8 +36,12 @@
     let tree =
           testGroup
             "pgmq-hasql"
-            ( [ QueueSpec.tests pool,
+            ( [ RootPermissionsSpec.tests ensureOwnerOnlyRoot,
+                JsonGoldenSpec.tests,
+                QueueSpec.tests pool,
                 MessageSpec.tests pool,
+                InspectionSpec.tests pool db,
+                InspectionCostSpec.tests pool,
                 MetricsSpec.tests,
                 AdvancedOpsSpec.tests pool,
                 NullSemanticsSpec.tests pool,
@@ -43,9 +53,14 @@
                 -- sweeps every mixed-case row in its database.
                 AliasingSpec.tests,
                 MixedCaseRemediationSpec.tests,
+                -- Creates mixed-case (and hyphenated) queues through raw SQL to
+                -- prove the inspection reads accept foreign names, so it too
+                -- runs on its own instance.
+                InspectionForeignNameSpec.tests,
                 -- Needs the Database handle: LISTEN/NOTIFY has no hasql API, so
                 -- the round-trip test opens a raw libpq connection.
                 NotifyChannelSpec.tests pool db,
+                PartitionedNotifySpec.tests,
                 SchemaSpec.tests pool,
                 RoundTripSpec.tests pool,
                 DecoderValidationSpec.tests pool,
diff --git a/test/NotifyChannelSpec.hs b/test/NotifyChannelSpec.hs
--- a/test/NotifyChannelSpec.hs
+++ b/test/NotifyChannelSpec.hs
@@ -10,7 +10,7 @@
 -- This module pins the contract from both sides: a real notification arrives on
 -- exactly the channel the helper computes, and a listener on the old documented
 -- name receives nothing.
-module NotifyChannelSpec (tests) where
+module NotifyChannelSpec (tests, withListener, awaitNotify) where
 
 import Control.Concurrent (threadDelay)
 import Control.Exception (bracket)
@@ -120,7 +120,9 @@
 awaitNotify conn attempts
   | attempts <= 0 = pure Nothing
   | otherwise = do
-      _ <- LibPQ.consumeInput conn
+      ok <- LibPQ.consumeInput conn
+      unless ok (assertFailure "libpq consumeInput failed")
+      LibPQ.status conn >>= (@?= LibPQ.ConnectionOk)
       pending <- LibPQ.notifies conn
       case pending of
         Just n -> pure (Just n)
diff --git a/test/PartitionedNotifySpec.hs b/test/PartitionedNotifySpec.hs
new file mode 100644
--- /dev/null
+++ b/test/PartitionedNotifySpec.hs
@@ -0,0 +1,101 @@
+{-# LANGUAGE OverloadedStrings #-}
+
+-- | Partition silence, including native metadata aliases, with a live control.
+module PartitionedNotifySpec (tests) where
+
+import Control.Exception (finally)
+import Control.Monad (forM_, unless)
+import Data.Text (Text)
+import Data.Text qualified as T
+import Data.Text.Encoding qualified as TE
+import Database.PostgreSQL.LibPQ qualified as PQ
+import EphemeralDb (withPgmqDb)
+import Hasql.Pool qualified as Pool
+import Hasql.Session qualified as Session
+import NotifyChannelSpec (awaitNotify, withListener)
+import System.Environment (lookupEnv)
+import Test.Tasty (TestTree, testGroup)
+import Test.Tasty.HUnit (assertFailure, testCase, (@?=))
+
+tests :: TestTree
+tests =
+  testGroup
+    "PartitionedNotifySpec"
+    [ testCase "partition inserts are silent and leave throttle at epoch" (partitionCase False False),
+      testCase "native leaf metadata alias stays silent" (nativeOnly (partitionCase True False)),
+      testCase "native leaf throttle alias stays silent and unchanged" (nativeOnly (partitionCase True True)),
+      testCase "ordinary nonzero throttle suppresses second committed insert" ordinaryCase,
+      testCase "native mixed-case missing lowercase metadata stays silent" (nativeOnly mixedCase)
+    ]
+
+nativeOnly :: IO () -> IO ()
+nativeOnly action = do
+  stock <- (== Just "1.12.0") <$> lookupEnv "PGMQ_TEST_SCHEMA_VERSION"
+  if stock then putStrLn "SKIPPED: native alias hardening" else action
+
+isolated :: (Pool.Pool -> (Text -> (PQ.Connection -> IO ()) -> IO ()) -> IO ()) -> IO ()
+isolated action = do
+  result <- withPgmqDb $ \pool db -> action pool (withListener db) `finally` Pool.release pool
+  either (assertFailure . show) pure result
+
+sql :: Pool.Pool -> Text -> IO ()
+sql pool command = Pool.use pool (Session.script command) >>= either (assertFailure . show) pure
+
+-- A separate listener subscription for each leaf, on the same live connection.
+listen :: PQ.Connection -> Text -> IO ()
+listen conn channel = do
+  result <- PQ.exec conn (TE.encodeUtf8 ("LISTEN \"" <> T.replace "\"" "\"\"" channel <> "\""))
+  case result of
+    Nothing -> assertFailure "LISTEN returned no result"
+    Just res -> PQ.resultStatus res >>= (@?= PQ.CommandOk)
+
+partitionCase :: Bool -> Bool -> IO ()
+partitionCase alias throttle = isolated $ \pool listener -> do
+  available <- Pool.use pool (Session.script "DO $$ BEGIN IF NOT EXISTS (SELECT FROM pg_extension WHERE extname='pg_partman') THEN RAISE EXCEPTION 'pg_partman unavailable'; END IF; END $$")
+  required <- (== Just "1") <$> lookupEnv "PGMQ_REQUIRE_PARTMAN"
+  case available of
+    Left err -> if required then assertFailure (show err) else putStrLn "SKIPPED: pg_partman unavailable"
+    Right () -> do
+      sql pool "SELECT pgmq.create_partitioned('pq','10','100'); SELECT pgmq.enable_notify_insert('pq',0); SELECT pgmq.create('control'); SELECT pgmq.enable_notify_insert('control',0)"
+      if alias then sql pool "SELECT pgmq.create('pq_p0'); DO $$ BEGIN IF NOT (SELECT relispartition FROM pg_class WHERE oid='pgmq.q_pq_p0'::regclass) THEN RAISE EXCEPTION 'alias changed relation'; END IF; END $$" else pure ()
+      if throttle then sql pool "INSERT INTO pgmq.notify_insert_throttle(queue_name,throttle_interval_ms,last_notified_at) VALUES ('pq_p0',0,to_timestamp(0))" else pure ()
+      listener "pgmq.q_control.INSERT" $ \conn -> do
+        result <- PQ.exec conn "SELECT 'pgmq.' || c.relname || '.INSERT' FROM pg_inherits i JOIN pg_class c ON c.oid=i.inhrelid WHERE i.inhparent='pgmq.q_pq'::regclass"
+        case result of
+          Nothing -> assertFailure "partition query failed"
+          Just res -> do
+            PQ.resultStatus res >>= (@?= PQ.TuplesOk)
+            rows <- PQ.ntuples res
+            forM_ [0 .. rows - 1] $ \row -> do
+              channel <- PQ.getvalue res row 0
+              maybe (assertFailure "null partition channel") (listen conn . TE.decodeUtf8) channel
+        listen conn "pgmq.q_pq.INSERT"
+        sql pool "SELECT pgmq.send('control','{}')"
+        control <- awaitNotify conn 20
+        fmap PQ.notifyRelname control @?= Just "pgmq.q_control.INSERT"
+        forM_ [1 :: Int .. 3] $ \_ -> sql pool "SELECT pgmq.send('pq','{}')"
+        received <- awaitNotify conn 10
+        fmap PQ.notifyRelname received @?= Nothing
+      sql pool "DO $$ BEGIN IF (SELECT count(*) FROM pgmq.q_pq) <> 3 THEN RAISE EXCEPTION 'messages not stored'; END IF; IF EXISTS (SELECT FROM pgmq.notify_insert_throttle WHERE queue_name IN ('pq','pq_p0') AND last_notified_at <> to_timestamp(0)) THEN RAISE EXCEPTION 'partition throttle advanced'; END IF; END $$"
+
+ordinaryCase :: IO ()
+ordinaryCase = isolated $ \pool listener -> do
+  sql pool "SELECT pgmq.create('ordinary'); SELECT pgmq.enable_notify_insert('ordinary',3600000)"
+  listener "pgmq.q_ordinary.INSERT" $ \conn -> do
+    sql pool "SELECT pgmq.send('ordinary','{}')"
+    first <- awaitNotify conn 20
+    fmap PQ.notifyRelname first @?= Just "pgmq.q_ordinary.INSERT"
+    sql pool "CREATE TABLE public.throttle_before AS SELECT last_notified_at FROM pgmq.notify_insert_throttle WHERE queue_name='ordinary'"
+    sql pool "SELECT pgmq.send('ordinary','{}')"
+    second <- awaitNotify conn 10
+    fmap PQ.notifyRelname second @?= Nothing
+    sql pool "DO $$ BEGIN IF (SELECT last_notified_at FROM pgmq.notify_insert_throttle WHERE queue_name='ordinary') <> (SELECT last_notified_at FROM public.throttle_before) THEN RAISE EXCEPTION 'suppressed insert advanced timestamp'; END IF; END $$"
+    sql pool "DO $$ BEGIN IF (SELECT count(*) FROM pgmq.q_ordinary) <> 2 THEN RAISE EXCEPTION 'messages not stored'; END IF; END $$"
+
+mixedCase :: IO ()
+mixedCase = isolated $ \pool listener -> do
+  sql pool "SELECT pgmq.create('MixedNotify'); SELECT pgmq.enable_notify_insert('MixedNotify',0)"
+  listener "pgmq.q_mixednotify.INSERT" $ \conn -> do
+    sql pool "SELECT pgmq.send('MixedNotify','{}')"
+    received <- awaitNotify conn 10
+    unless (fmap PQ.notifyRelname received == Nothing) (assertFailure "mixed-case queue notified without matching metadata")
diff --git a/test/RootPermissionsSpec.hs b/test/RootPermissionsSpec.hs
new file mode 100644
--- /dev/null
+++ b/test/RootPermissionsSpec.hs
@@ -0,0 +1,63 @@
+-- | Check each suite's local root helper using private disposable fixtures.
+module RootPermissionsSpec (tests) where
+
+import Control.Exception (IOException, bracket, try)
+import Data.Bits ((.&.))
+import System.Directory (getTemporaryDirectory, removePathForcibly)
+import System.Posix.Directory qualified as Posix
+import System.Posix.Files (createSymbolicLink, fileMode, getSymbolicLinkStatus, isDirectory, setFileMode)
+import System.Posix.Temp (mkdtemp)
+import Test.Tasty (TestTree, testGroup)
+import Test.Tasty.HUnit (assertBool, assertFailure, testCase, (@?=))
+
+tests :: (FilePath -> IO ()) -> TestTree
+tests ensure =
+  testGroup
+    "RootPermissionsSpec"
+    [ testCase "missing root is a real 0700 directory" $ private $ \parent -> do
+        let root = parent <> "/root"
+        ensure root
+        status <- getSymbolicLinkStatus root
+        assertBool "real directory" (isDirectory status)
+        fileMode status .&. 0o7777 @?= 0o700,
+      testCase "owned 0755 directory is tightened without losing contents" $ private $ \parent -> do
+        let root = parent <> "/root"
+        Posix.createDirectory root 0o755
+        setFileMode root 0o755
+        writeFile (root <> "/sentinel") "preserve"
+        ensure root
+        mode root >>= (@?= 0o700)
+        readFile (root <> "/sentinel") >>= (@?= "preserve"),
+      testCase "repeated call preserves contents and permissions" $ private $ \parent -> do
+        let root = parent <> "/root"
+        ensure root
+        writeFile (root <> "/sentinel") "preserve"
+        ensure root
+        mode root >>= (@?= 0o700)
+        readFile (root <> "/sentinel") >>= (@?= "preserve"),
+      testCase "regular file is rejected unchanged" $ private $ \parent -> do
+        let root = parent <> "/root"
+        writeFile root "preserve"
+        setFileMode root 0o644
+        rejected (ensure root)
+        mode root >>= (@?= 0o644)
+        readFile root >>= (@?= "preserve"),
+      testCase "symlink is rejected without chmod of its target" $ private $ \parent -> do
+        let root = parent <> "/root"
+            target = parent <> "/target"
+        Posix.createDirectory target 0o755
+        setFileMode target 0o755
+        writeFile (target <> "/sentinel") "preserve"
+        createSymbolicLink target root
+        rejected (ensure root)
+        mode target >>= (@?= 0o755)
+        readFile (target <> "/sentinel") >>= (@?= "preserve")
+    ]
+  where
+    mode path = (.&. 0o7777) . fileMode <$> getSymbolicLinkStatus path
+    rejected action = do
+      result <- try action :: IO (Either IOException ())
+      case result of
+        Left _ -> pure ()
+        Right () -> assertFailure "unsafe root accepted"
+    private action = bracket (getTemporaryDirectory >>= \tmp -> mkdtemp (tmp <> "/pgmq-root-test-")) removePathForcibly action
diff --git a/test/UmbrellaExportsSpec.hs b/test/UmbrellaExportsSpec.hs
--- a/test/UmbrellaExportsSpec.hs
+++ b/test/UmbrellaExportsSpec.hs
@@ -4,9 +4,10 @@
 {-# LANGUAGE TypeOperators #-}
 
 -- Compile witnesses: this module intentionally uses only the public umbrella.
-module UmbrellaExportsSpec (grouped, polling, creation, arguments, metric) where
+module UmbrellaExportsSpec (grouped, polling, creation, arguments, metric, inspection, lenientMetrics, inspectionArguments) where
 
 import Data.Int (Int32, Int64)
+import Data.Text (Text)
 import Data.Vector (Vector)
 import Hasql.Session (Session)
 import Pgmq
@@ -25,3 +26,12 @@
 
 metric :: QueueMetrics -> Maybe Int64
 metric = defaultPartitionLength
+
+inspection :: PeekMessages -> LookupMessage -> (Session (Vector Message), Session (Vector ArchivedMessage), Session (Maybe Message), Session (Maybe ArchivedMessage))
+inspection page one = (peekMessages page, peekArchivedMessages page, lookupMessage one, lookupArchivedMessage one)
+
+lenientMetrics :: Text -> Session QueueMetrics
+lenientMetrics = queueMetricsUnvalidated
+
+inspectionArguments :: Text -> (PeekMessages, LookupMessage)
+inspectionArguments q = (PeekMessages q Nothing 50, LookupMessage q (MessageId 1))
diff --git a/test/golden/queue_metrics.json b/test/golden/queue_metrics.json
new file mode 100644
--- /dev/null
+++ b/test/golden/queue_metrics.json
@@ -0,0 +1,1 @@
+{"queue_name":"orders","queue_length":0,"newest_msg_age_sec":null,"oldest_msg_age_sec":null,"total_messages":0,"scrape_time":"2026-08-19T12:34:56.123Z","queue_visible_length":0,"default_partition_length":null}
diff --git a/test/golden/queue_metrics_partitioned.json b/test/golden/queue_metrics_partitioned.json
new file mode 100644
--- /dev/null
+++ b/test/golden/queue_metrics_partitioned.json
@@ -0,0 +1,1 @@
+{"queue_name":"events","queue_length":120,"newest_msg_age_sec":2,"oldest_msg_age_sec":3600,"total_messages":9000,"scrape_time":"2026-08-19T12:34:56.123Z","queue_visible_length":115,"default_partition_length":7}