diff --git a/CHANGELOG.md b/CHANGELOG.md
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,5 +1,41 @@
 # Changelog
 
+## 0.18.0.0 — 2026-10-27
+
+### Breaking Changes
+
+- Require [`shibuya-core`](mori://shinzui/shibuya/packages/shibuya-core) `^>=0.12.0.0`,
+  adopting its extended configuration, processor metrics, paused-state API, and
+  `Ingested.deliveryStatus` field. PGMQ deliveries report `DeliveryCurrent`,
+  preserving handler and acknowledgement behavior.
+- Require the [`pgmq-hs`](mori://shinzui/pgmq-hs) 0.7.0.0 package family
+  (`pgmq-core`, `pgmq-hasql`, and `pgmq-effectful`; tests also use `pgmq-migration`).
+
+### New Features
+
+- Expose `Shibuya.Adapter.Pgmq.DeadLetter` for explicitly registered,
+  non-destructive DLQ browse/fetch and default-disabled transactional redrive.
+  Redrive authorizes the source queue and atomically claims each DLQ row before
+  sending its original payload; concurrent calls and uncertain-commit retries
+  cannot duplicate that physical row's redrive.
+- Add `mkDlqPayloadForQueue` and include `original_queue` in adapter-produced
+  DLQ envelopes in both metadata modes. Preserve `mkDlqPayload` and existing
+  reason fields; historical and malformed envelopes remain inspectable.
+
+### Other Changes
+
+- Use the published dependency releases without the transitional `dlq-admin` flag.
+- Extend PostgreSQL coverage for managed DLQs and consumer trace context;
+  preserve core configuration defaults in lifecycle fixtures and use stable
+  per-user temporary roots for ephemeral PostgreSQL cleanup.
+- Add reproducible worker, administration, and encoding performance comparisons.
+  Performance acceptance remains incomplete: 196 of 198 numeric gates pass;
+  FIFO acknowledgement p95 upper bounds are 1.106857 against the original
+  control and 1.172465 against the matched control, exceeding the 1.10 limit.
+- Defer performance acceptance for this release by explicit user approval;
+  retain the failed bounds and unchanged numeric limits.
+- Consolidate the unpublished 0.17.0.0 changes into this release.
+
 ## 0.16.1.0 — 2026-09-21
 
 ### Reliability
diff --git a/shibuya-pgmq-adapter.cabal b/shibuya-pgmq-adapter.cabal
--- a/shibuya-pgmq-adapter.cabal
+++ b/shibuya-pgmq-adapter.cabal
@@ -1,6 +1,6 @@
 cabal-version: 3.12
 name: shibuya-pgmq-adapter
-version: 0.16.1.0
+version: 0.18.0.0
 synopsis: PGMQ adapter for the Shibuya queue processing framework
 description:
   A Shibuya adapter that integrates with pgmq (PostgreSQL Message Queue)
@@ -23,8 +23,10 @@
     Shibuya.Adapter.Pgmq
     Shibuya.Adapter.Pgmq.Config
     Shibuya.Adapter.Pgmq.Convert
+    Shibuya.Adapter.Pgmq.DeadLetter
 
   other-modules:
+    Shibuya.Adapter.Pgmq.DeadLetter.Internal
     Shibuya.Adapter.Pgmq.Internal
 
   default-extensions:
@@ -46,10 +48,10 @@
     hasql ^>=1.10,
     hasql-pool ^>=1.4,
     hasql-transaction ^>=1.2,
-    pgmq-core ^>=0.6,
-    pgmq-effectful ^>=0.6,
-    pgmq-hasql ^>=0.6,
-    shibuya-core ^>=0.10.0.0,
+    pgmq-core ^>=0.7.0.0,
+    pgmq-effectful ^>=0.7.0.0,
+    pgmq-hasql ^>=0.7.0.0,
+    shibuya-core ^>=0.12.0.0,
     stm ^>=2.5,
     streamly ^>=0.11,
     streamly-core ^>=0.3,
@@ -75,6 +77,12 @@
     src
 
   main-is: Main.hs
+  other-modules:
+    DeadLetterExample
+    Shibuya.Adapter.Pgmq.DeadLetter
+    Shibuya.Adapter.Pgmq.DeadLetter.Internal
+    Shibuya.Adapter.Pgmq.DeadLetterSpec
+
   default-extensions:
     DeriveAnyClass
     DerivingStrategies
@@ -107,19 +115,21 @@
     base ^>=4.21.0.0,
     bytestring,
     effectful-core (>=2.6.1 && <2.7) || (>=2.7.1.1 && <2.8),
-    ephemeral-pg,
+    ephemeral-pg >=0.3.1 && <0.4, directory, unix,
     hasql ^>=1.10,
     hasql-pool ^>=1.4,
     hasql-transaction ^>=1.2,
     hspec ^>=2.11,
+    hs-opentelemetry-api ^>=1.0,
+    hs-opentelemetry-exporter-in-memory ^>=1.0,
     pg-migrate >=1.1 && <1.3,
-    pgmq-core ^>=0.6,
-    pgmq-effectful ^>=0.6,
-    pgmq-hasql ^>=0.6,
-    pgmq-migration ^>=0.6,
+    pgmq-core ^>=0.7.0.0,
+    pgmq-effectful ^>=0.7.0.0,
+    pgmq-hasql ^>=0.7.0.0,
+    pgmq-migration ^>=0.7.0.0,
     quickcheck-instances ^>=0.3,
     random,
-    shibuya-core ^>=0.10.0.0,
+    shibuya-core ^>=0.12.0.0,
     shibuya-pgmq-adapter,
     stm,
     streamly ^>=0.11,
diff --git a/src/Shibuya/Adapter/Pgmq/Convert.hs b/src/Shibuya/Adapter/Pgmq/Convert.hs
--- a/src/Shibuya/Adapter/Pgmq/Convert.hs
+++ b/src/Shibuya/Adapter/Pgmq/Convert.hs
@@ -13,6 +13,7 @@
 
     -- * DLQ Payload
     mkDlqPayload,
+    mkDlqPayloadForQueue,
   )
 where
 
@@ -169,3 +170,13 @@
             "original_headers" .= msg.headers
           ]
       | otherwise = []
+
+-- | Create an adapter-managed DLQ envelope with an authoritative source queue.
+-- The original helper remains byte-for-byte compatible; source identity is
+-- present independently of the optional delivery metadata.
+mkDlqPayloadForQueue :: Pgmq.QueueName -> Pgmq.Message -> DeadLetterReason -> Bool -> Pgmq.MessageBody
+mkDlqPayloadForQueue queue msg reason includeMetadata =
+  case mkDlqPayload msg reason includeMetadata of
+    Pgmq.MessageBody (Object fields) ->
+      Pgmq.MessageBody (Object (KeyMap.insert "original_queue" (String (Pgmq.queueNameToText queue)) fields))
+    body -> body
diff --git a/src/Shibuya/Adapter/Pgmq/DeadLetter.hs b/src/Shibuya/Adapter/Pgmq/DeadLetter.hs
new file mode 100644
--- /dev/null
+++ b/src/Shibuya/Adapter/Pgmq/DeadLetter.hs
@@ -0,0 +1,39 @@
+-- | In-process administration for explicitly registered adapter-managed DLQs.
+--
+-- Supply a separate administrative pool connected to the workers' database.
+-- Constructing a handle performs no I/O and starts no background work. The host
+-- owns and closes the pool. Separate connections do not isolate database CPU,
+-- I/O or locks. Host transports must enforce their own caller permissions;
+-- 'redriveEnabled' is a mutation gate, not authentication.
+--
+-- Pages are moving views ordered by queue-local message ID. Restart at the first
+-- page to discover lower IDs committed late. Limits bound rows, not JSON bytes.
+-- Archive recovery and queue discovery/creation are outside this API.
+module Shibuya.Adapter.Pgmq.DeadLetter
+  ( DeadLetterAdmin,
+    DeadLetterAdminConfig (..),
+    defaultDeadLetterAdminConfig,
+    DlqRegistration (..),
+    registrationFromAdapterConfig,
+    DlqConfigError (..),
+    mkDeadLetterAdmin,
+    DlqCursor (..),
+    DlqPageRequest (..),
+    defaultDlqPageRequest,
+    DeadLetterPage (..),
+    DeadLetterItem (..),
+    DeadLetterEntry (..),
+    ReasonRepresentation (..),
+    SourceResolution (..),
+    DlqDecodeError (..),
+    DlqDecodeProblem (..),
+    DlqPageError (..),
+    DlqError (..),
+    RedriveOutcome (..),
+    browseDeadLetters,
+    fetchDeadLetter,
+    redriveDeadLetter,
+  )
+where
+
+import Shibuya.Adapter.Pgmq.DeadLetter.Internal
diff --git a/src/Shibuya/Adapter/Pgmq/DeadLetter/Internal.hs b/src/Shibuya/Adapter/Pgmq/DeadLetter/Internal.hs
new file mode 100644
--- /dev/null
+++ b/src/Shibuya/Adapter/Pgmq/DeadLetter/Internal.hs
@@ -0,0 +1,354 @@
+-- | Implementation of adapter-owned DLQ semantics and transaction composition.
+module Shibuya.Adapter.Pgmq.DeadLetter.Internal where
+
+import Control.Monad (unless, void, when)
+import Data.Aeson (FromJSON, Result (..), Value (..), fromJSON)
+import Data.Aeson.Key qualified as Key
+import Data.Aeson.KeyMap qualified as KM
+import Data.Functor.Contravariant (contramap)
+import Data.Int (Int32, Int64)
+import Data.List (nub)
+import Data.Text (Text)
+import Data.Text qualified as Text
+import Data.Time (UTCTime)
+import Data.Vector (Vector)
+import Data.Vector qualified as Vector
+import Hasql.Decoders qualified as D
+import Hasql.Encoders qualified as E
+import Hasql.Pool qualified as Pool
+import Hasql.Statement qualified as Statement
+import Hasql.Transaction qualified as Tx
+import Hasql.Transaction.Sessions qualified as TxSession
+import Pgmq.Effectful (PgmqRuntimeError, fromUsageError)
+import Pgmq.Hasql.Decoders (messageDecoder)
+import Pgmq.Hasql.Encoders (messageIdValue)
+import Pgmq.Hasql.Statements.Inspection qualified as Inspection
+import Pgmq.Hasql.Statements.Message qualified as Msg
+import Pgmq.Hasql.Statements.Types qualified as Q
+import Pgmq.Types (Message (..), MessageBody (..), MessageHeaders (..), MessageId (..), QueueName, parseQueueName, queueNameToText)
+import Shibuya.Adapter.Pgmq.Config (DeadLetterConfig (..), DeadLetterTarget (..), PgmqAdapterConfig (..))
+
+-- | One explicitly managed physical DLQ and the sources it may return to.
+data DlqRegistration = DlqRegistration
+  { deadLetterQueue :: !QueueName,
+    allowedSourceQueues :: ![QueueName]
+  }
+  deriving stock (Eq, Show)
+
+-- | Administrative settings, independent of worker configuration.
+data DeadLetterAdminConfig = DeadLetterAdminConfig
+  { registrations :: ![DlqRegistration],
+    redriveEnabled :: !Bool,
+    statementTimeoutMs :: !Int32,
+    lockTimeoutMs :: !Int32
+  }
+  deriving stock (Eq, Show)
+
+-- | Disable mutation; bound each statement to 5 seconds and lock waits to 1.
+defaultDeadLetterAdminConfig :: [DlqRegistration] -> DeadLetterAdminConfig
+defaultDeadLetterAdminConfig rs = DeadLetterAdminConfig rs False 5000 1000
+
+-- | Invalid configuration is rejected without acquiring a connection.
+data DlqConfigError
+  = DuplicateManagedQueue !QueueName
+  | EmptySourceQueues !QueueName
+  | DuplicateSourceQueues !QueueName
+  | SelfMapping !QueueName
+  | InvalidStatementTimeout !Int32
+  | InvalidLockTimeout !Int32
+  | NoDirectDeadLetterTarget
+  deriving stock (Eq, Show)
+
+-- | Borrowed pool and validated configuration; no threads, queries or ownership.
+data DeadLetterAdmin = DeadLetterAdmin !Pool.Pool !DeadLetterAdminConfig
+
+-- | Validate registrations and deadlines without performing I/O. The supplied
+-- pool is borrowed; the host remains responsible for its lifetime.
+mkDeadLetterAdmin :: Pool.Pool -> DeadLetterAdminConfig -> Either DlqConfigError DeadLetterAdmin
+mkDeadLetterAdmin pool config = do
+  when (config.statementTimeoutMs <= 0) $ Left (InvalidStatementTimeout config.statementTimeoutMs)
+  when (config.lockTimeoutMs <= 0) $ Left (InvalidLockTimeout config.lockTimeoutMs)
+  let validate seen registration = do
+        let queue = registration.deadLetterQueue
+            sources = registration.allowedSourceQueues
+        when (queue `elem` seen) $ Left (DuplicateManagedQueue queue)
+        when (null sources) $ Left (EmptySourceQueues queue)
+        when (sources /= nub sources) $ Left (DuplicateSourceQueues queue)
+        when (queue `elem` sources) $ Left (SelfMapping queue)
+        pure (queue : seen)
+      go _ [] = pure ()
+      go seen (r : rs) = validate seen r >>= \next -> go next rs
+  go [] config.registrations
+  pure (DeadLetterAdmin pool config)
+
+-- | Derive the single-source registration for a direct adapter target.
+-- Topic destinations require explicit physical registrations.
+registrationFromAdapterConfig :: PgmqAdapterConfig -> Either DlqConfigError DlqRegistration
+registrationFromAdapterConfig config = case config.deadLetterConfig of
+  Just DeadLetterConfig {dlqTarget = DirectQueue queue}
+    | queue /= config.queueName -> Right (DlqRegistration queue [config.queueName])
+    | otherwise -> Left (SelfMapping queue)
+  _ -> Left NoDirectDeadLetterTarget
+
+-- | Exclusive cursor. IDs have meaning only within this physical queue's lifetime.
+data DlqCursor = DlqCursor
+  { deadLetterQueue :: !QueueName,
+    afterMessageId :: !MessageId
+  }
+  deriving stock (Eq, Show)
+
+-- | Positive row limit (at most 500), with an optional exclusive cursor.
+data DlqPageRequest = DlqPageRequest
+  { pageLimit :: !Int32,
+    afterCursor :: !(Maybe DlqCursor)
+  }
+  deriving stock (Eq, Show)
+
+-- | Start at the first ID and return at most 50 rows.
+defaultDlqPageRequest :: DlqPageRequest
+defaultDlqPageRequest = DlqPageRequest 50 Nothing
+
+-- | Whether reason fields were stored explicitly or inferred from old text.
+data ReasonRepresentation = StructuredReason | LegacyReason
+  deriving stock (Eq, Show)
+
+-- | How the source was resolved. A multi-source historical row remains unknown.
+data SourceResolution = PayloadSource | ConfiguredFallback | SourceUnavailable
+  deriving stock (Eq, Show)
+
+-- | Stable, field-specific reasons an envelope cannot be interpreted safely.
+data DlqDecodeProblem
+  = MissingField
+  | WrongType !Text
+  | InvalidSourceName !Text
+  | SourceNotAllowed !QueueName
+  deriving stock (Eq, Show)
+
+-- | The envelope field that could not be interpreted and its typed problem.
+data DlqDecodeError = DlqDecodeError
+  { field :: !Text,
+    problem :: !DlqDecodeProblem
+  }
+  deriving stock (Eq, Show)
+
+-- | Original delivery metadata and current DLQ metadata are deliberately separate.
+-- Missing 'originalHeaders' is 'Nothing'; explicit JSON null is @Just Null@.
+data DeadLetterEntry = DeadLetterEntry
+  { originalPayload :: !Value,
+    reasonCode :: !Text,
+    reasonDetail :: !Value,
+    reasonRepresentation :: !ReasonRepresentation,
+    resolvedSource :: !(Maybe QueueName),
+    sourceResolution :: !SourceResolution,
+    originalMessageId :: !(Maybe MessageId),
+    originalReadCount :: !(Maybe Int64),
+    originalEnqueuedAt :: !(Maybe UTCTime),
+    originalLastReadAt :: !(Maybe UTCTime),
+    originalHeaders :: !(Maybe Value),
+    dlqMessageId :: !MessageId,
+    dlqReadCount :: !Int64,
+    dlqEnqueuedAt :: !UTCTime,
+    dlqLastReadAt :: !(Maybe UTCTime)
+  }
+  deriving stock (Eq, Show)
+
+-- | Corrupt envelopes stay in their original position and retain their raw row.
+data DeadLetterItem = DeadLetterItem
+  { deadLetterQueue :: !QueueName,
+    rawMessage :: !Message,
+    decodedEntry :: !(Either DlqDecodeError DeadLetterEntry)
+  }
+  deriving stock (Eq, Show)
+
+-- | A moving view, not a shared snapshot. Refresh page one for late lower IDs.
+-- The row limit does not bound JSON bytes or database scan work.
+data DeadLetterPage = DeadLetterPage
+  { items :: !(Vector DeadLetterItem),
+    nextCursor :: !(Maybe DlqCursor)
+  }
+  deriving stock (Eq, Show)
+
+-- | Invalid row limit or a cursor bound to a different physical DLQ.
+data DlqPageError = InvalidPageLimit !Int32 | CursorQueueMismatch !QueueName !QueueName
+  deriving stock (Eq, Show)
+
+-- | Database failures never become empty pages or successful absence.
+data DlqError
+  = RedriveDisabled
+  | UnknownManagedDlq !QueueName
+  | InvalidPageRequest !DlqPageError
+  | EntryNotFound !QueueName !MessageId
+  | MalformedEnvelope !DlqDecodeError
+  | UnresolvedSource
+  | DisallowedSource !QueueName
+  | DatabaseError !PgmqRuntimeError
+  deriving stock (Eq, Show)
+
+-- | Absence cannot establish which actor removed the row.
+data RedriveOutcome = Redriven !MessageId | AlreadyAbsent
+  deriving stock (Eq, Show)
+
+findRegistration :: DeadLetterAdminConfig -> QueueName -> Either DlqError DlqRegistration
+findRegistration config queue = case filter (\r -> r.deadLetterQueue == queue) config.registrations of
+  [registration] -> Right registration
+  _ -> Left (UnknownManagedDlq queue)
+
+decodeEntry :: DlqRegistration -> Message -> Either DlqDecodeError DeadLetterEntry
+decodeEntry registration msg = case unMessageBody msg.body of
+  Object fields -> do
+    originalPayload <- required fields "original_message"
+    (reasonCode, reasonDetail, reasonRepresentation) <- decodeReason fields
+    (resolvedSource, sourceResolution) <- case KM.lookup "original_queue" fields of
+      Nothing -> pure $ case registration.allowedSourceQueues of
+        [source] -> (Just source, ConfiguredFallback)
+        _ -> (Nothing, SourceUnavailable)
+      Just value -> do
+        name <- typed "original_queue" value
+        source <- either (const $ bad "original_queue" (InvalidSourceName name)) pure (parseQueueName name)
+        unless (source `elem` registration.allowedSourceQueues) $ bad "original_queue" (SourceNotAllowed source)
+        pure (Just source, PayloadSource)
+    originalMessageId <- fmap MessageId <$> optional fields "original_message_id"
+    originalReadCount <- optional fields "read_count"
+    originalEnqueuedAt <- optional fields "original_enqueued_at"
+    originalLastReadAt <- case KM.lookup "last_read_at" fields of
+      Nothing -> pure Nothing
+      Just Null -> pure Nothing
+      Just value -> Just <$> typed "last_read_at" value
+    let originalHeaders = KM.lookup "original_headers" fields
+    pure
+      DeadLetterEntry
+        { originalPayload,
+          reasonCode,
+          reasonDetail,
+          reasonRepresentation,
+          resolvedSource,
+          sourceResolution,
+          originalMessageId,
+          originalReadCount,
+          originalEnqueuedAt,
+          originalLastReadAt,
+          originalHeaders,
+          dlqMessageId = msg.messageId,
+          dlqReadCount = msg.readCount,
+          dlqEnqueuedAt = msg.enqueuedAt,
+          dlqLastReadAt = msg.lastReadAt
+        }
+  _ -> bad "$" (WrongType "object")
+  where
+    decodeReason fields
+      | KM.member "dead_letter_reason_code" fields || KM.member "dead_letter_reason_detail" fields = do
+          code <- required fields "dead_letter_reason_code" >>= typed "dead_letter_reason_code"
+          detail <- required fields "dead_letter_reason_detail"
+          pure (code, detail, StructuredReason)
+      | otherwise = do
+          rendered <- required fields "dead_letter_reason" >>= typed "dead_letter_reason"
+          let (code, suffix) = Text.breakOn ": " rendered
+          pure (code, if Text.null suffix then Null else String (Text.drop 2 suffix), LegacyReason)
+    required fields key = maybe (bad (Key.toText key) MissingField) pure (KM.lookup key fields)
+    optional fields key = traverse (typed (Key.toText key)) (KM.lookup key fields)
+
+bad :: Text -> DlqDecodeProblem -> Either DlqDecodeError a
+bad field problem = Left DlqDecodeError {field, problem}
+
+typed :: (FromJSON a) => Text -> Value -> Either DlqDecodeError a
+typed field value = case fromJSON value of
+  Error err -> bad field (WrongType (Text.pack err))
+  Success result -> Right result
+
+inspectItem :: DlqRegistration -> Message -> DeadLetterItem
+inspectItem registration msg = DeadLetterItem registration.deadLetterQueue msg (decodeEntry registration msg)
+
+-- | Transaction-local settings are reset by PostgreSQL at commit or rollback.
+setTimeouts :: DeadLetterAdminConfig -> Tx.Transaction ()
+setTimeouts config =
+  void $
+    Tx.statement
+      (Text.pack (show config.statementTimeoutMs) <> "ms", Text.pack (show config.lockTimeoutMs) <> "ms")
+      ( Statement.preparable
+          "select set_config('statement_timeout', $1, true), set_config('lock_timeout', $2, true)"
+          (contramap fst (E.param (E.nonNullable E.text)) <> contramap snd (E.param (E.nonNullable E.text)))
+          (D.singleRow ((,) <$> D.column (D.nonNullable D.text) <*> D.column (D.nonNullable D.text)))
+      )
+
+runTransaction :: DeadLetterAdmin -> TxSession.Mode -> Tx.Transaction (Either DlqError a) -> IO (Either DlqError a)
+runTransaction (DeadLetterAdmin pool config) mode tx = do
+  result <- Pool.use pool $ TxSession.transactionNoRetry TxSession.ReadCommitted mode (setTimeouts config >> tx)
+  pure $ either (Left . DatabaseError . fromUsageError) id result
+
+resolveTable :: QueueName -> Tx.Transaction Text
+resolveTable queue = Tx.statement (queueNameToText queue, "q") Inspection.formatTableName
+
+-- | Observe rows without leasing, including invisible and malformed rows.
+browseDeadLetters :: DeadLetterAdmin -> QueueName -> DlqPageRequest -> IO (Either DlqError DeadLetterPage)
+browseDeadLetters admin@(DeadLetterAdmin _ config) queue request = case validate of
+  Left err -> pure (Left err)
+  Right registration -> runTransaction admin TxSession.Read $ do
+    table <- resolveTable queue
+    rows <- Tx.statement (fmap (\cursor -> cursor.afterMessageId) request.afterCursor, request.pageLimit + 1) (Inspection.peekStatement table)
+    let page = Vector.take (fromIntegral request.pageLimit) rows
+        cursor =
+          if Vector.length rows > fromIntegral request.pageLimit
+            then (\msg -> DlqCursor queue msg.messageId) <$> (page Vector.!? (Vector.length page - 1))
+            else Nothing
+    pure $ Right (DeadLetterPage (Vector.map (inspectItem registration) page) cursor)
+  where
+    validate = do
+      registration <- findRegistration config queue
+      when (request.pageLimit < 1 || request.pageLimit > 500) $
+        Left (InvalidPageRequest (InvalidPageLimit request.pageLimit))
+      case request.afterCursor of
+        Just cursor | cursor.deadLetterQueue /= queue -> Left (InvalidPageRequest (CursorQueueMismatch queue cursor.deadLetterQueue))
+        _ -> pure ()
+      pure registration
+
+-- | Observe one registered DLQ row without leasing it. Malformed rows are
+-- returned as inspectable items; only actual absence yields EntryNotFound.
+fetchDeadLetter :: DeadLetterAdmin -> QueueName -> MessageId -> IO (Either DlqError DeadLetterItem)
+fetchDeadLetter admin@(DeadLetterAdmin _ config) queue messageId = case findRegistration config queue of
+  Left err -> pure (Left err)
+  Right registration -> runTransaction admin TxSession.Read $ do
+    table <- resolveTable queue
+    result <- Tx.statement messageId (Inspection.lookupStatement table)
+    pure $ maybe (Left (EntryNotFound queue messageId)) (Right . inspectItem registration) result
+
+-- | Serialize on the selected row and atomically delete/send its original body.
+-- Retry the complete operation after uncertain commit. A new ID/count/time and
+-- tail position are assigned; distinct fan-out copies have distinct claims.
+redriveDeadLetter :: DeadLetterAdmin -> QueueName -> MessageId -> IO (Either DlqError RedriveOutcome)
+redriveDeadLetter admin@(DeadLetterAdmin _ config) queue messageId
+  | not config.redriveEnabled = pure (Left RedriveDisabled)
+  | otherwise = case findRegistration config queue of
+      Left err -> pure (Left err)
+      Right registration -> runTransaction admin TxSession.Write $ do
+        table <- resolveTable queue
+        locked <- Tx.statement messageId (lockedLookup table)
+        case locked of
+          Nothing -> pure (Right AlreadyAbsent)
+          Just msg -> case decodeEntry registration msg of
+            Left err -> pure $ Left $ case err.problem of
+              SourceNotAllowed source -> DisallowedSource source
+              _ -> MalformedEnvelope err
+            Right entry -> case entry.resolvedSource of
+              Nothing -> pure (Left UnresolvedSource)
+              Just source -> do
+                claimed <- Tx.statement (Q.MessageQuery queue messageId) Msg.deleteMessage
+                if not claimed
+                  then pure (Right AlreadyAbsent)
+                  else do
+                    let headers = case entry.originalHeaders of
+                          Just saved -> Just saved
+                          Nothing -> msg.headers
+                    newId <- case headers of
+                      Nothing -> Tx.statement (Q.SendMessage source (MessageBody entry.originalPayload) Nothing) Msg.sendMessage
+                      Just value -> Tx.statement (Q.SendMessageWithHeaders source (MessageBody entry.originalPayload) (MessageHeaders value) Nothing) Msg.sendMessageWithHeaders
+                    pure (Right (Redriven newId))
+
+lockedLookup :: Text -> Statement.Statement MessageId (Maybe Message)
+lockedLookup table =
+  Statement.unpreparable
+    ( "select msg_id, read_ct, enqueued_at, last_read_at, vt, message, headers from pgmq."
+        <> Inspection.quoteIdentifier table
+        <> " where msg_id = $1 for update"
+    )
+    (E.param (E.nonNullable messageIdValue))
+    (D.rowMaybe messageDecoder)
diff --git a/src/Shibuya/Adapter/Pgmq/Internal.hs b/src/Shibuya/Adapter/Pgmq/Internal.hs
--- a/src/Shibuya/Adapter/Pgmq/Internal.hs
+++ b/src/Shibuya/Adapter/Pgmq/Internal.hs
@@ -107,12 +107,12 @@
     PollingConfig (..),
   )
 import Shibuya.Adapter.Pgmq.Convert
-  ( mkDlqPayload,
+  ( mkDlqPayloadForQueue,
     pgmqMessageToEnvelope,
   )
 import Shibuya.Core.Ack (AckDecision (..), DeadLetterReason (..), RetryDelay (..))
 import Shibuya.Core.AckHandle (AckHandle (..))
-import Shibuya.Core.Ingested (Ingested (..))
+import Shibuya.Core.Ingested (DeliveryStatus (..), Ingested (..))
 import Shibuya.Core.Lease (Lease (..))
 import Shibuya.Core.Types (TraceHeaders)
 import Shibuya.Telemetry.Effect (Tracing)
@@ -309,7 +309,7 @@
     Left err -> throwError (fromUsageError err)
     Right () -> pure ()
   where
-    dlqBody = mkDlqPayload msg reason dlqConfig.includeMetadata
+    dlqBody = mkDlqPayloadForQueue config.queueName msg reason dlqConfig.includeMetadata
     sourceQuery = MessageQuery config.queueName msg.messageId
     session =
       Transaction.Sessions.transaction
@@ -446,7 +446,8 @@
           Ingested
             { envelope = pgmqMessageToEnvelope msg,
               ack = ackHandle,
-              lease = Just lease
+              lease = Just lease,
+              deliveryStatus = pure DeliveryCurrent
             }
 
 finalizeAutoDeadLetter ::
diff --git a/test/DeadLetterExample.hs b/test/DeadLetterExample.hs
new file mode 100644
--- /dev/null
+++ b/test/DeadLetterExample.hs
@@ -0,0 +1,14 @@
+module DeadLetterExample (configureOrders, inspectOrders) where
+
+import Hasql.Pool qualified as Pool
+import Pgmq.Types (QueueName)
+import Shibuya.Adapter.Pgmq.DeadLetter
+
+configureOrders :: Pool.Pool -> QueueName -> QueueName -> Either DlqConfigError DeadLetterAdmin
+configureOrders administrativePool source dlq =
+  mkDeadLetterAdmin
+    administrativePool
+    (defaultDeadLetterAdminConfig [DlqRegistration dlq [source]])
+
+inspectOrders :: DeadLetterAdmin -> QueueName -> IO (Either DlqError DeadLetterPage)
+inspectOrders admin dlq = browseDeadLetters admin dlq defaultDlqPageRequest
diff --git a/test/Main.hs b/test/Main.hs
--- a/test/Main.hs
+++ b/test/Main.hs
@@ -3,6 +3,7 @@
 import Shibuya.Adapter.Pgmq.ChaosSpec qualified as ChaosSpec
 import Shibuya.Adapter.Pgmq.ConfigSpec qualified as ConfigSpec
 import Shibuya.Adapter.Pgmq.ConvertSpec qualified as ConvertSpec
+import Shibuya.Adapter.Pgmq.DeadLetterSpec qualified as DeadLetterSpec
 import Shibuya.Adapter.Pgmq.IntegrationSpec qualified as IntegrationSpec
 import Shibuya.Adapter.Pgmq.InternalSpec qualified as InternalSpec
 import Shibuya.Adapter.Pgmq.PropertySpec qualified as PropertySpec
@@ -18,3 +19,4 @@
     describe "Shibuya.Adapter.Pgmq Properties" PropertySpec.spec
   describe "Integration Tests" IntegrationSpec.spec
   describe "Chaos Tests" ChaosSpec.spec
+  DeadLetterSpec.spec
diff --git a/test/Shibuya/Adapter/Pgmq/ChaosSpec.hs b/test/Shibuya/Adapter/Pgmq/ChaosSpec.hs
--- a/test/Shibuya/Adapter/Pgmq/ChaosSpec.hs
+++ b/test/Shibuya/Adapter/Pgmq/ChaosSpec.hs
@@ -15,10 +15,12 @@
 import Control.Exception qualified as Exception
 import Control.Monad (forM_, when)
 import Data.Aeson (Value (..), object, (.=))
+import Data.Aeson.Key qualified as Key
 import Data.Aeson.KeyMap qualified as KeyMap
 import Data.IORef (IORef, atomicModifyIORef', newIORef, readIORef)
 import Data.Int (Int32)
 import Data.Text qualified as Text
+import Data.Text.Encoding qualified as Text
 import Data.Vector qualified as Vector
 import Effectful (Eff, IOE, Limit (..), Persistence (..), UnliftStrategy (..), liftIO, runEff, withEffToIO, (:>))
 import Effectful.Dispatch.Dynamic (interpret)
@@ -27,13 +29,17 @@
 import Hasql.Pool qualified as Pool
 import Hasql.Session qualified as Session
 import Hasql.Statement qualified as Statement
+import OpenTelemetry.Attributes (emptyAttributes)
+import OpenTelemetry.Exporter.InMemory (inMemoryListExporter)
+import OpenTelemetry.Trace.Core qualified as OTel
+import OpenTelemetry.Trace.Id.Generator (IdGenerator (..))
 import Pgmq.Effectful (Pgmq, PgmqRuntimeError, runPgmq)
 import Pgmq.Effectful qualified as PgmqEff
 import Pgmq.Effectful.Effect qualified as PgmqEffect
 import Pgmq.Hasql.Encoders qualified as Encoders
 import Pgmq.Hasql.Sessions qualified as Sessions
-import Pgmq.Hasql.Statements.Types (ReadMessage (..), SendMessage (..), SendMessageWithHeaders (..))
-import Pgmq.Types (Message (..), MessageBody (..), MessageHeaders (..), QueueName, parseQueueName)
+import Pgmq.Hasql.Statements.Types (BindTopic (..), ReadMessage (..), SendMessage (..), SendMessageWithHeaders (..))
+import Pgmq.Types (Message (..), MessageBody (..), MessageHeaders (..), QueueName, parseQueueName, parseRoutingKey, parseTopicPattern)
 import Shibuya.Adapter (Adapter (..))
 import Shibuya.Adapter.Pgmq
   ( PgmqAcknowledgementException,
@@ -47,6 +53,7 @@
     directDeadLetter,
     mkPgmqAdapterEnv,
     pgmqAdapter,
+    topicDeadLetter,
   )
 import Shibuya.Adapter.Pgmq.Internal (mkAckHandle, mkIngested)
 import Shibuya.App
@@ -67,7 +74,8 @@
 import Shibuya.Core.Ingested (Ingested (..))
 import Shibuya.Handler (Handler)
 import Shibuya.Internal.Runner.Master (getLifecycleSnapshot)
-import Shibuya.Telemetry.Effect (Tracing, runTracingNoop)
+import Shibuya.Telemetry.Effect (Tracing, runTracing, runTracingNoop)
+import Shibuya.Telemetry.Propagation (extractTraceContext)
 import Streamly.Data.Stream qualified as Stream
 import System.Environment (lookupEnv)
 import System.Timeout (timeout)
@@ -334,6 +342,72 @@
             }
     Vector.length sourceMessages `shouldBe` 0
 
+  forM_ [False, True] $ \topicRoute ->
+    forM_ [False, True] $ \withOriginal ->
+      it ("DLQ consumer trace " <> (if topicRoute then "topic" else "direct") <> (if withOriginal then " with producer" else " without headers")) $ \TestFixture {pool, queueName, dlqName} -> do
+        let producerParent = "00-0af7651916cd43dd8448eb211c80319c-b7ad6b7169203331-01"
+            producerState = "vendor=opaque"
+            original = object ["traceparent" .= (producerParent :: Text.Text), "tracestate" .= (producerState :: Text.Text), "custom" .= ("kept" :: Text.Text)]
+            routingKey = either (error . show) id $ parseRoutingKey "audit.dlq"
+            topicPattern = either (error . show) id $ parseTopicPattern "audit.dlq"
+            config =
+              (defaultConfig queueName)
+                { visibilityTimeout = 5,
+                  batchSize = 1,
+                  polling = StandardPolling {pollInterval = 0.1},
+                  deadLetterConfig = Just $ if topicRoute then topicDeadLetter routingKey True else directDeadLetter dlqName True
+                }
+        when topicRoute $ runPgmqSession pool $ Sessions.bindTopic $ BindTopic topicPattern dlqName
+        runPgmqSession pool $ do
+          if withOriginal
+            then do
+              _ <- Sessions.sendMessageWithHeaders $ SendMessageWithHeaders queueName (MessageBody (String "traced-dlq")) (MessageHeaders original) (Just 0)
+              pure ()
+            else do
+              _ <- Sessions.sendMessage $ SendMessage queueName (MessageBody (String "traced-dlq")) (Just 0)
+              pure ()
+        (spanProcessor, spansRef) <- inMemoryListExporter
+        processedRef <- newIORef (0 :: Int)
+        Exception.bracket
+          (OTel.createTracerProvider [spanProcessor] OTel.emptyTracerProviderOptions {OTel.tracerProviderOptionsIdGenerator = DefaultIdGenerator})
+          (\p -> OTel.shutdownTracerProvider p (Just 5_000_000))
+          $ \provider -> do
+            let tracer = OTel.makeTracer provider (OTel.InstrumentationLibrary "dlq-audit" "" "" emptyAttributes) OTel.tracerOptions
+            runAdapterIO pool $ runTracing tracer $ do
+              adapter <- requireAdapter pool config
+              result <- runApp defaultAppConfig [(ProcessorId "dlq-identity", mkProcessor adapter (deadLetterHandler processedRef))]
+              case result of
+                Left err -> liftIO $ expectationFailure $ show err
+                Right appHandle -> do
+                  liftIO $ waitForProcessed processedRef 1 3_000_000
+                  drained <- stopAppGracefully defaultShutdownConfig {drainTimeout = 5} appHandle
+                  liftIO $ drained `shouldBe` True
+        messages <- runPgmqSession pool $ Sessions.readMessage $ ReadMessage dlqName 30 (Just 10) Nothing
+        Vector.length messages `shouldBe` 1
+        spans <- readIORef spansRef
+        named <- mapM (\sp -> do hot <- readIORef (OTel.spanHot sp); pure (OTel.hotName hot, sp)) spans
+        case (Vector.uncons messages, [sp | ("dlq-identity process", sp) <- named]) of
+          (Just (msg, _), [consumer]) -> case msg.headers of
+            Just (Object hdrs) -> do
+              let stringHeader k = case KeyMap.lookup k hdrs of
+                    Just (String v) -> Just (Text.encodeUtf8 v)
+                    _ -> Nothing
+                  outgoing = [(Text.encodeUtf8 (Key.toText k), v) | k <- ["traceparent", "tracestate"], Just v <- [stringHeader k]]
+              case extractTraceContext outgoing of
+                Nothing -> expectationFailure "expected valid consumer context in DLQ"
+                Just ctx -> do
+                  OTel.traceId ctx `shouldBe` OTel.traceId (OTel.spanContext consumer)
+                  OTel.spanId ctx `shouldBe` OTel.spanId (OTel.spanContext consumer)
+              if withOriginal
+                then do
+                  KeyMap.lookup "x-shibuya-upstream-traceparent" hdrs `shouldBe` Just (String producerParent)
+                  KeyMap.lookup "x-shibuya-upstream-tracestate" hdrs `shouldBe` Just (String producerState)
+                  KeyMap.lookup "custom" hdrs `shouldBe` Just (String "kept")
+                  stringHeader "traceparent" `shouldNotBe` Just (Text.encodeUtf8 producerParent)
+                else KeyMap.lookup "x-shibuya-upstream-traceparent" hdrs `shouldBe` Nothing
+            _ -> expectationFailure "expected DLQ headers"
+          _ -> expectationFailure "expected one DLQ message and one consumer span"
+
   it "preserves trace headers when moving to DLQ" $ \TestFixture {pool, queueName, dlqName} -> do
     -- Send a message with trace headers
     let traceparent = "00-0af7651916cd43dd8448eb211c80319c-b7ad6b7169203331-01" :: Text.Text
@@ -1162,7 +1236,7 @@
       adapter <- requireAdapter pool config
       let handler = slowHandler processedRef 10_000_000
           processor = mkProcessor adapter handler
-      appResult <- runApp (AppConfig {strategy = IgnoreFailures, inboxSize = 2}) [(ProcessorId "noloss-test", processor)]
+      appResult <- runApp (defaultAppConfig {strategy = IgnoreFailures, inboxSize = 2}) [(ProcessorId "noloss-test", processor)]
       case appResult of
         Left err -> liftIO $ expectationFailure $ "Failed to start app: " <> show err
         Right appHandle -> do
@@ -1328,7 +1402,7 @@
     adapter <- requireAdapter pool config
     let handler = slowHandler processedRef 400000 -- 0.4s per message
         processor = mkProcessor adapter handler
-    appResult <- runApp (AppConfig {strategy = IgnoreFailures, inboxSize = 2}) [(ProcessorId "shutdown-measure", processor)]
+    appResult <- runApp (defaultAppConfig {strategy = IgnoreFailures, inboxSize = 2}) [(ProcessorId "shutdown-measure", processor)]
     case appResult of
       Left err -> liftIO $ expectationFailure $ "Failed to start app: " <> show err
       Right appHandle -> do
diff --git a/test/Shibuya/Adapter/Pgmq/ConvertSpec.hs b/test/Shibuya/Adapter/Pgmq/ConvertSpec.hs
--- a/test/Shibuya/Adapter/Pgmq/ConvertSpec.hs
+++ b/test/Shibuya/Adapter/Pgmq/ConvertSpec.hs
@@ -21,6 +21,7 @@
   extractTraceHeadersSpec
   pgmqMessageToEnvelopeSpec
   mkDlqPayloadSpec
+  mkDlqPayloadForQueueSpec
 
 -- | Tests for messageIdToShibuya
 messageIdToShibuyaSpec :: Spec
@@ -455,3 +456,40 @@
   case mkDeadLetterCode "keiro.router.selection.recipient_overflow" of
     Left err -> error (Text.unpack err)
     Right code -> code
+
+mkDlqPayloadForQueueSpec :: Spec
+mkDlqPayloadForQueueSpec = describe "mkDlqPayloadForQueue" $ do
+  let timestamp = UTCTime (fromGregorian 2026 10 25) 0
+      queue = either (error . show) id (Pgmq.parseQueueName "orders")
+      msg =
+        Pgmq.Message
+          (Pgmq.MessageId 42)
+          timestamp
+          timestamp
+          (Just timestamp)
+          3
+          (Pgmq.MessageBody (object ["order" .= (7 :: Int)]))
+          (Just (object ["x-pgmq-group" .= ("a" :: Text.Text)]))
+      reason = ApplicationFailure representativeDeadLetterCode "注文: failed"
+      baseFields =
+        [ "original_queue" .= ("orders" :: Text.Text),
+          "original_message" .= object ["order" .= (7 :: Int)],
+          "dead_letter_reason" .= ("keiro.router.selection.recipient_overflow: 注文: failed" :: Text.Text),
+          "dead_letter_reason_code" .= ("keiro.router.selection.recipient_overflow" :: Text.Text),
+          "dead_letter_reason_detail" .= ("注文: failed" :: Text.Text)
+        ]
+  it "writes the exact queue-aware envelope without optional metadata" $
+    mkDlqPayloadForQueue queue msg reason False `shouldBe` Pgmq.MessageBody (object baseFields)
+  it "writes the exact queue-aware envelope with original metadata" $
+    mkDlqPayloadForQueue queue msg reason True
+      `shouldBe` Pgmq.MessageBody
+        ( object
+            ( baseFields
+                ++ [ "original_message_id" .= (42 :: Int),
+                     "original_enqueued_at" .= timestamp,
+                     "last_read_at" .= timestamp,
+                     "read_count" .= (3 :: Int),
+                     "original_headers" .= object ["x-pgmq-group" .= ("a" :: Text.Text)]
+                   ]
+            )
+        )
diff --git a/test/Shibuya/Adapter/Pgmq/DeadLetterSpec.hs b/test/Shibuya/Adapter/Pgmq/DeadLetterSpec.hs
new file mode 100644
--- /dev/null
+++ b/test/Shibuya/Adapter/Pgmq/DeadLetterSpec.hs
@@ -0,0 +1,549 @@
+module Shibuya.Adapter.Pgmq.DeadLetterSpec (spec) where
+
+import Control.Concurrent (threadDelay)
+import Control.Concurrent.Async (async, cancel, concurrently, waitCatch, withAsync)
+import Control.Concurrent.MVar (newEmptyMVar, putMVar, takeMVar)
+import Control.Exception (bracket, throwIO)
+import Control.Monad (forM_, void)
+import Data.Aeson (Value (..), object, (.=))
+import Data.Aeson.Key qualified
+import Data.Aeson.KeyMap qualified as KM
+import Data.Either (isLeft)
+import Data.Text (Text)
+import Data.Time (secondsToDiffTime)
+import Data.Vector qualified as V
+import Effectful (liftIO, runEff)
+import Effectful.Error.Static (runErrorNoCallStack)
+import Hasql.Connection.Settings qualified as Settings
+import Hasql.Decoders qualified as D
+import Hasql.Encoders qualified as E
+import Hasql.Pool qualified as Pool
+import Hasql.Pool.Config qualified as PoolConfig
+import Hasql.Session qualified as Session
+import Hasql.Statement qualified as Statement
+import Pgmq.Effectful (PgmqRuntimeError, runPgmq)
+import Pgmq.Hasql.Sessions qualified as S
+import Pgmq.Hasql.Statements.Inspection qualified as Inspection
+import Pgmq.Hasql.Statements.Types qualified as Q
+import Pgmq.Types qualified as P
+import Shibuya.Adapter.Pgmq
+import Shibuya.Adapter.Pgmq.DeadLetter
+import Shibuya.App (ProcessorId (..), defaultAppConfig, defaultShutdownConfig, mkProcessor, runApp, stopAppGracefully)
+import Shibuya.Core.Ack (AckDecision (..), DeadLetterReason (..), mkDeadLetterCode)
+import Shibuya.Core.Ingested (Message (..))
+import Shibuya.Core.Types (Envelope (..))
+import Shibuya.Telemetry.Effect (runTracingNoop)
+import System.Timeout (timeout)
+import Test.Hspec hiding (before)
+import TmpPostgres
+
+spec :: Spec
+spec = describe "Dead-letter administration" $ do
+  around database $ do
+    it "preserves an application's exact reason through a real handler, browse and fetch" $ \fixture -> do
+      let pool = fixture.pool
+          source = fixture.queueName
+          dlq = fixture.dlqName
+          payload = object ["order" .= (42 :: Int)]
+          detail = "注文の検証: failed: exact text"
+          code = either (error . show) id (mkDeadLetterCode "orders.validation.failed")
+          config =
+            (defaultConfig source)
+              { batchSize = 1,
+                polling = StandardPolling 0.01,
+                deadLetterConfig = Just (directDeadLetter dlq True)
+              }
+      void $ runPgmqSession pool $ S.sendMessage (Q.SendMessage source (P.MessageBody payload) Nothing)
+      outcome <- runEff $ runErrorNoCallStack @PgmqRuntimeError $ runPgmq pool $ runTracingNoop $ do
+        adapter <- pgmqAdapter (mkPgmqAdapterEnv pool) config >>= either (error . show) pure
+        app <- runApp defaultAppConfig [(ProcessorId "dlq-roundtrip", mkProcessor adapter (\_ -> pure (AckDeadLetter (ApplicationFailure code detail))))] >>= either (error . show) pure
+        liftIO $ awaitDlq pool dlq
+        drained <- stopAppGracefully defaultShutdownConfig app
+        liftIO $ drained `shouldBe` True
+      either throwIO pure outcome
+      admin <- requireAdmin fixture False
+      page <- browseDeadLetters admin dlq defaultDlqPageRequest >>= right
+      V.length page.items `shouldBe` 1
+      let item = V.head page.items
+      entry <- right item.decodedEntry
+      entry.reasonCode `shouldBe` "orders.validation.failed"
+      entry.reasonDetail `shouldBe` String detail
+      entry.originalPayload `shouldBe` payload
+      entry.resolvedSource `shouldBe` Just source
+      entry.sourceResolution `shouldBe` PayloadSource
+      entry.originalReadCount `shouldBe` Just 1
+      entry.dlqReadCount `shouldBe` 0
+      entry.dlqEnqueuedAt `shouldBe` item.rawMessage.enqueuedAt
+      fetchDeadLetter admin dlq item.rawMessage.messageId `shouldReturn` Right item
+      redriveDeadLetter admin dlq item.rawMessage.messageId `shouldReturn` Left RedriveDisabled
+
+    it "pages by exclusive IDs across malformed and invisible rows without changing any row" $ \fixture -> do
+      admin <- requireAdmin fixture False
+      browseDeadLetters admin fixture.dlqName defaultDlqPageRequest `shouldReturn` Right (DeadLetterPage V.empty Nothing)
+      firstId <- seed fixture (envelope (String "first"))
+      corruptId <- seed fixture (String "not an envelope")
+      lastId <- seed fixture (envelope Null)
+      void $ runPgmqSession fixture.pool $ S.readMessage (Q.ReadMessage fixture.dlqName 300 (Just 2) Nothing)
+      before <- snapshot fixture.pool fixture.dlqName
+      firstPage <- browseDeadLetters admin fixture.dlqName (DlqPageRequest 2 Nothing) >>= right
+      V.map (\item -> item.rawMessage.messageId) firstPage.items `shouldBe` V.fromList [firstId, corruptId]
+      firstPage.nextCursor `shouldBe` Just (DlqCursor fixture.dlqName corruptId)
+      (firstPage.items V.! 1).decodedEntry `shouldSatisfy` isLeft
+      lastPage <- browseDeadLetters admin fixture.dlqName (DlqPageRequest 2 firstPage.nextCursor) >>= right
+      V.map (\item -> item.rawMessage.messageId) lastPage.items `shouldBe` V.singleton lastId
+      lastPage.nextCursor `shouldBe` Nothing
+      nullEntry <- right (V.head lastPage.items).decodedEntry
+      nullEntry.originalPayload `shouldBe` Null
+      forM_ before $ \msg -> void $ fetchDeadLetter admin fixture.dlqName msg.messageId >>= right
+      snapshot fixture.pool fixture.dlqName `shouldReturn` before
+      void $ runPgmqSession fixture.pool $ S.deleteMessage (Q.MessageQuery fixture.dlqName firstId)
+      browseDeadLetters admin fixture.dlqName (DlqPageRequest 2 firstPage.nextCursor) `shouldReturn` Right lastPage
+      exact <- browseDeadLetters admin fixture.dlqName (DlqPageRequest 2 Nothing) >>= right
+      V.length exact.items `shouldBe` 2
+      exact.nextCursor `shouldBe` Nothing
+      fetchDeadLetter admin fixture.dlqName firstId `shouldReturn` Left (EntryNotFound fixture.dlqName firstId)
+
+    it "decodes structured-only, dual-written and legacy history while preserving arbitrary JSON detail" $ \fixture -> do
+      admin <- requireAdmin fixture False
+      let detail = object ["nested" .= [Null, object ["ok" .= True]]]
+          structured = object ["original_message" .= Null, "dead_letter_reason_code" .= ("invalid_payload" :: Text), "dead_letter_reason_detail" .= detail]
+          legacy = object ["original_message" .= String "old", "dead_letter_reason" .= ("poison_pill: 原因: retained" :: Text)]
+          bare = object ["original_message" .= Null, "dead_letter_reason" .= ("max_retries_exceeded" :: Text)]
+      ids <- mapM (seed fixture) [structured, putField "dead_letter_reason" (String "ignored: legacy") structured, legacy, bare]
+      entries <- mapM (\messageId -> fetchDeadLetter admin fixture.dlqName messageId >>= right >>= right . (\item -> item.decodedEntry)) ids
+      map (\entry -> entry.reasonCode) entries `shouldBe` ["invalid_payload", "invalid_payload", "poison_pill", "max_retries_exceeded"]
+      map (\entry -> entry.reasonDetail) entries `shouldBe` [detail, detail, String "原因: retained", Null]
+      map (\entry -> entry.reasonRepresentation) entries `shouldBe` [StructuredReason, StructuredReason, LegacyReason, LegacyReason]
+      forM_ entries $ \entry -> do
+        entry.originalReadCount `shouldBe` Nothing
+        entry.originalMessageId `shouldBe` Nothing
+        entry.originalEnqueuedAt `shouldBe` Nothing
+        entry.dlqReadCount `shouldBe` 0
+        entry.sourceResolution `shouldBe` ConfiguredFallback
+        entry.resolvedSource `shouldBe` Just fixture.queueName
+
+    it "keeps malformed optional metadata and partial structured reasons inspectable and refuses redrive" $ \fixture -> do
+      admin <- requireAdmin fixture True
+      let payload = envelope (String "valid body")
+          corrupt =
+            [ ("original_message", deleteField "original_message" payload),
+              ("dead_letter_reason_code", deleteField "dead_letter_reason_code" payload),
+              ("dead_letter_reason_detail", deleteField "dead_letter_reason_detail" payload),
+              ("dead_letter_reason_code", putField "dead_letter_reason_code" Null payload),
+              ("original_queue", putField "original_queue" (Number 42) payload),
+              ("original_queue", putField "original_queue" (String "../bad") payload),
+              ("read_count", putField "read_count" (String "2") payload),
+              ("original_message_id", putField "original_message_id" Null payload),
+              ("original_enqueued_at", putField "original_enqueued_at" (String "not-a-time") payload),
+              ("last_read_at", putField "last_read_at" (Number 7) payload)
+            ]
+      forM_ corrupt $ \(field, body) -> do
+        messageId <- seed fixture body
+        item <- fetchDeadLetter admin fixture.dlqName messageId >>= right
+        case item.decodedEntry of
+          Right _ -> expectationFailure "corrupt envelope decoded successfully"
+          Left err -> do
+            err.field `shouldBe` field
+            redriveDeadLetter admin fixture.dlqName messageId `shouldReturn` Left (MalformedEnvelope err)
+      rows <- snapshot fixture.pool fixture.dlqName
+      page <- browseDeadLetters admin fixture.dlqName defaultDlqPageRequest >>= right
+      V.length page.items `shouldBe` length corrupt
+      snapshot fixture.pool fixture.dlqName `shouldReturn` rows
+
+    it "validates limits, queue-bound cursors and managed queue identity before querying" $ \fixture -> do
+      admin <- requireAdmin fixture False
+      forM_ [0, -1, 501, maxBound] $ \limit ->
+        browseDeadLetters admin fixture.dlqName (DlqPageRequest limit Nothing) `shouldReturn` Left (InvalidPageRequest (InvalidPageLimit limit))
+      browseDeadLetters admin fixture.dlqName (DlqPageRequest 50 (Just (DlqCursor fixture.queueName (P.MessageId 1))))
+        `shouldReturn` Left (InvalidPageRequest (CursorQueueMismatch fixture.dlqName fixture.queueName))
+      browseDeadLetters admin fixture.queueName defaultDlqPageRequest `shouldReturn` Left (UnknownManagedDlq fixture.queueName)
+      fetchDeadLetter admin fixture.queueName (P.MessageId 1) `shouldReturn` Left (UnknownManagedDlq fixture.queueName)
+      redriveDeadLetter admin fixture.queueName (P.MessageId 1) `shouldReturn` Left RedriveDisabled
+
+    it "does not turn missing tables into an empty page or EntryNotFound" $ \fixture -> do
+      admin <- requireAdmin fixture False
+      void $ runPgmqSession fixture.pool $ S.dropQueue fixture.dlqName
+      browseDeadLetters admin fixture.dlqName defaultDlqPageRequest >>= (`shouldSatisfy` isDatabaseError)
+      fetchDeadLetter admin fixture.dlqName (P.MessageId 1) >>= (`shouldSatisfy` isDatabaseError)
+      runPgmqSession fixture.pool $ S.createQueue fixture.dlqName
+
+    it "requires valid direct mappings and positive deadlines at handle construction" $ \fixture -> do
+      let queue = fixture.dlqName
+          source = fixture.queueName
+          registration = DlqRegistration queue [source]
+          config = defaultDeadLetterAdminConfig [registration]
+          rejects cfg expected = case mkDeadLetterAdmin fixture.pool cfg of
+            Left err -> err `shouldBe` expected
+            Right _ -> expectationFailure "invalid config accepted"
+      rejects (config {registrations = [registration, registration]}) (DuplicateManagedQueue queue)
+      rejects (config {registrations = [DlqRegistration queue []]}) (EmptySourceQueues queue)
+      rejects (config {registrations = [DlqRegistration queue [source, source]]}) (DuplicateSourceQueues queue)
+      rejects (config {registrations = [DlqRegistration queue [queue]]}) (SelfMapping queue)
+      rejects (config {statementTimeoutMs = 0}) (InvalidStatementTimeout 0)
+      rejects (config {lockTimeoutMs = -1}) (InvalidLockTimeout (-1))
+      registrationFromAdapterConfig (defaultConfig source) `shouldBe` Left NoDirectDeadLetterTarget
+      registrationFromAdapterConfig ((defaultConfig source) {deadLetterConfig = Just (directDeadLetter queue False)}) `shouldBe` Right registration
+      let key = either (error . show) id (P.parseRoutingKey "fail.orders")
+      registrationFromAdapterConfig ((defaultConfig source) {deadLetterConfig = Just (topicDeadLetter key True)}) `shouldBe` Left NoDirectDeadLetterTarget
+      registrationFromAdapterConfig ((defaultConfig source) {deadLetterConfig = Just (directDeadLetter source True)}) `shouldBe` Left (SelfMapping source)
+
+    it "resolves registered shared DLQs by payload and refuses ambiguous or disallowed sources" $ \fixture -> do
+      let other = namedQueue "another_source"
+          forbidden = namedQueue "unregistered_source"
+      admin <-
+        right $
+          mkDeadLetterAdmin fixture.pool $
+            (defaultDeadLetterAdminConfig [DlqRegistration fixture.dlqName [fixture.queueName, other]]) {redriveEnabled = True}
+      explicit <- seed fixture (putField "original_queue" (String (P.queueNameToText other)) (envelope Null))
+      ambiguous <- seed fixture (envelope Null)
+      disallowed <- seed fixture (putField "original_queue" (String (P.queueNameToText forbidden)) (envelope Null))
+      item <- fetchDeadLetter admin fixture.dlqName explicit >>= right
+      entry <- right item.decodedEntry
+      entry.resolvedSource `shouldBe` Just other
+      entry.sourceResolution `shouldBe` PayloadSource
+      rows <- snapshot fixture.pool fixture.dlqName
+      redriveDeadLetter admin fixture.dlqName ambiguous `shouldReturn` Left UnresolvedSource
+      redriveDeadLetter admin fixture.dlqName disallowed `shouldReturn` Left (DisallowedSource forbidden)
+      snapshot fixture.pool fixture.dlqName `shouldReturn` rows
+
+    it "redrives the original payload once and preserves saved group/trace headers" $ \fixture -> do
+      admin <- requireAdmin fixture True
+      let body = object ["order" .= (42 :: Int)]
+          saved = object ["x-pgmq-group" .= ("customer-a" :: Text), "traceparent" .= ("original" :: Text)]
+          wrapper = putField "original_headers" saved (envelope body)
+      messageId <- seedWithHeaders fixture wrapper (object ["traceparent" .= ("dlq" :: Text)])
+      before <- snapshot fixture.pool fixture.dlqName
+      disabled <- requireAdmin fixture False
+      redriveDeadLetter disabled fixture.dlqName messageId `shouldReturn` Left RedriveDisabled
+      snapshot fixture.pool fixture.dlqName `shouldReturn` before
+      result <- redriveDeadLetter admin fixture.dlqName messageId >>= right
+      newId <- case result of
+        Redriven value -> pure value
+        AlreadyAbsent -> expectationFailure "fresh row absent" >> fail "absent"
+      snapshot fixture.pool fixture.dlqName `shouldReturn` V.empty
+      sourceRows <- snapshot fixture.pool fixture.queueName
+      V.length sourceRows `shouldBe` 1
+      let restored = V.head sourceRows
+      restored.messageId `shouldBe` newId
+      restored.body `shouldBe` P.MessageBody body
+      restored.headers `shouldBe` Just saved
+      restored.readCount `shouldBe` 0
+      redriveDeadLetter admin fixture.dlqName messageId `shouldReturn` Right AlreadyAbsent
+      -- Discarded confirmed response followed by a fresh retry, not network loss.
+      redriveDeadLetter admin fixture.dlqName messageId `shouldReturn` Right AlreadyAbsent
+      snapshot fixture.pool fixture.queueName `shouldReturn` sourceRows
+      consumed <- runPgmqSession fixture.pool $ S.readMessage (Q.ReadMessage fixture.queueName 30 (Just 1) Nothing)
+      V.map (\msg -> msg.body) consumed `shouldBe` V.singleton (P.MessageBody body)
+
+    it "distinguishes missing original headers from explicit null and accepts null payload" $ \fixture -> do
+      admin <- requireAdmin fixture True
+      let trace = object ["traceparent" .= ("dlq-span" :: Text), "x-pgmq-group" .= ("g" :: Text)]
+      explicitNull <- seedWithHeaders fixture (putField "original_headers" Null (envelope Null)) trace
+      missing <- seedWithHeaders fixture (envelope (String "metadata-disabled")) trace
+      forM_ [explicitNull, missing] $ \messageId -> void $ redriveDeadLetter admin fixture.dlqName messageId >>= right
+      rows <- snapshot fixture.pool fixture.queueName
+      V.map (\msg -> msg.body) rows `shouldBe` V.fromList [P.MessageBody Null, P.MessageBody (String "metadata-disabled")]
+      V.map (\msg -> msg.headers) rows `shouldBe` V.fromList [Just Null, Just trace]
+
+    it "restores arbitrary JSON-valued saved headers supported by PGMQ" $ \fixture -> do
+      admin <- requireAdmin fixture True
+      let headers = [String "scalar", Array (V.fromList [Null, Bool True]), Number 7, Bool False]
+      forM_ headers $ \saved -> do
+        messageId <- seed fixture (putField "original_headers" saved (envelope Null))
+        redriveDeadLetter admin fixture.dlqName messageId >>= (`shouldSatisfy` isRedriven)
+      rows <- snapshot fixture.pool fixture.queueName
+      V.map (\msg -> msg.headers) rows `shouldBe` V.fromList (map Just headers)
+
+    it "serializes independent concurrent redrives on one durable row claim" $ \fixture -> do
+      admin <- requireAdmin fixture True
+      messageId <- seed fixture (envelope (String "race"))
+      results <- concurrently (redriveDeadLetter admin fixture.dlqName messageId) (redriveDeadLetter admin fixture.dlqName messageId)
+      case results of
+        (Right (Redriven _), Right AlreadyAbsent) -> pure ()
+        (Right AlreadyAbsent, Right (Redriven _)) -> pure ()
+        _ -> expectationFailure (show results)
+      rows <- snapshot fixture.pool fixture.queueName
+      V.length rows `shouldBe` 1
+      snapshot fixture.pool fixture.dlqName `shouldReturn` V.empty
+
+    it "rolls the deletion back on send failure and can retry after the source is recreated" $ \fixture -> do
+      admin <- requireAdmin fixture True
+      messageId <- seed fixture (envelope (String "rollback"))
+      before <- snapshot fixture.pool fixture.dlqName
+      void $ runPgmqSession fixture.pool $ S.dropQueue fixture.queueName
+      redriveDeadLetter admin fixture.dlqName messageId >>= (`shouldSatisfy` isDatabaseError)
+      snapshot fixture.pool fixture.dlqName `shouldReturn` before
+      runPgmqSession fixture.pool $ S.createQueue fixture.queueName
+      redriveDeadLetter admin fixture.dlqName messageId >>= (`shouldSatisfy` isRedriven)
+      rows <- snapshot fixture.pool fixture.queueName
+      V.map (\msg -> msg.body) rows `shouldBe` V.singleton (P.MessageBody (String "rollback"))
+
+    it "treats topic fan-out copies as two separately redrivable entries" $ \fixture -> do
+      let other = namedQueue "topic_dlq_copy"
+          key = either (error . show) id (P.parseRoutingKey "fail.orders")
+          pattern' = either (error . show) id (P.parseTopicPattern "fail.#")
+      runPgmqSession fixture.pool $ do
+        S.createQueue other
+        S.bindTopic (Q.BindTopic pattern' fixture.dlqName)
+        S.bindTopic (Q.BindTopic pattern' other)
+      produceFailure
+        fixture
+        ( (defaultConfig fixture.queueName)
+            { batchSize = 1,
+              polling = StandardPolling 0.01,
+              deadLetterConfig = Just (topicDeadLetter key True)
+            }
+        )
+        (String "fanout")
+      admin <-
+        right $
+          mkDeadLetterAdmin fixture.pool $
+            (defaultDeadLetterAdminConfig [DlqRegistration q [fixture.queueName] | q <- [fixture.dlqName, other]]) {redriveEnabled = True}
+      copies <-
+        mapM
+          ( \q -> do
+              page <- browseDeadLetters admin q defaultDlqPageRequest >>= right
+              V.length page.items `shouldBe` 1
+              pure (q, (V.head page.items).rawMessage.messageId)
+          )
+          [fixture.dlqName, other]
+      ((q1, id1), (q2, id2)) <- case copies of
+        [first, second] -> pure (first, second)
+        _ -> expectationFailure "expected exactly two topic copies" >> fail "copy count"
+      (r1, r2) <- concurrently (redriveDeadLetter admin q1 id1) (redriveDeadLetter admin q2 id2)
+      r1 `shouldSatisfy` isRedriven
+      r2 `shouldSatisfy` isRedriven
+      rows <- snapshot fixture.pool fixture.queueName
+      V.map (\msg -> msg.body) rows `shouldBe` V.replicate 2 (P.MessageBody (String "fanout"))
+      void $ runPgmqSession fixture.pool $ S.dropQueue other
+
+    it "redrives real metadata-disabled writes through a normal handler with the original payload" $ \fixture -> do
+      let config =
+            (defaultConfig fixture.queueName)
+              { batchSize = 1,
+                polling = StandardPolling 0.01,
+                deadLetterConfig = Just (directDeadLetter fixture.dlqName False)
+              }
+      produceFailure fixture config (String "no metadata")
+      admin <- requireAdmin fixture True
+      page <- browseDeadLetters admin fixture.dlqName defaultDlqPageRequest >>= right
+      let item = V.head page.items
+      entry <- right item.decodedEntry
+      entry.originalReadCount `shouldBe` Nothing
+      entry.originalHeaders `shouldBe` Nothing
+      entry.sourceResolution `shouldBe` PayloadSource
+      redriveDeadLetter admin fixture.dlqName item.rawMessage.messageId >>= (`shouldSatisfy` isRedriven)
+      -- Run an ordinary handler against the source and observe its public input.
+      received <- newEmptyMVar
+      outcome <- runEff $ runErrorNoCallStack @PgmqRuntimeError $ runPgmq fixture.pool $ runTracingNoop $ do
+        adapter <- pgmqAdapter (mkPgmqAdapterEnv fixture.pool) config >>= either (error . show) pure
+        app <- runApp defaultAppConfig [(ProcessorId "redriven-input", mkProcessor adapter (\message -> liftIO (putMVar received message.envelope.payload) >> pure AckOk))] >>= either (error . show) pure
+        actual <- liftIO $ timeout 5000000 (takeMVar received)
+        liftIO $ actual `shouldBe` Just (String "no metadata")
+        void $ stopAppGracefully defaultShutdownConfig app
+      either throwIO pure outcome
+
+    it "includes source identity for automatic dead-letter moves" $ \fixture -> do
+      let config =
+            (defaultConfig fixture.queueName)
+              { maxRetries = 0,
+                batchSize = 1,
+                polling = StandardPolling 0.01,
+                deadLetterConfig = Just (directDeadLetter fixture.dlqName False)
+              }
+      produceFailure fixture config (String "automatic")
+      admin <- requireAdmin fixture False
+      page <- browseDeadLetters admin fixture.dlqName defaultDlqPageRequest >>= right
+      entry <- right (V.head page.items).decodedEntry
+      entry.reasonCode `shouldBe` "max_retries_exceeded"
+      entry.resolvedSource `shouldBe` Just fixture.queueName
+      entry.originalReadCount `shouldBe` Nothing
+
+    it "returns a full 500-row page with one lookahead row and an exact boundary" $ \fixture -> do
+      ids <- mapM (\_ -> seed fixture (envelope Null)) [1 .. 501 :: Int]
+      admin <- requireAdmin fixture False
+      page <- browseDeadLetters admin fixture.dlqName (DlqPageRequest 500 Nothing) >>= right
+      V.length page.items `shouldBe` 500
+      page.nextCursor `shouldBe` Just (DlqCursor fixture.dlqName (ids !! 499))
+      next <- browseDeadLetters admin fixture.dlqName (DlqPageRequest 500 page.nextCursor) >>= right
+      V.map (\item -> item.rawMessage.messageId) next.items `shouldBe` V.singleton (last ids)
+      next.nextCursor `shouldBe` Nothing
+  administrativePoolSpec
+
+database :: (TestFixture -> IO ()) -> IO ()
+database action = withPgmqDb (\pool -> withTestFixture pool action) >>= either (error . show) pure
+
+right :: (Show e) => Either e a -> IO a
+right = either (\err -> expectationFailure (show err) >> fail "expected Right") pure
+
+requireAdmin :: TestFixture -> Bool -> IO DeadLetterAdmin
+requireAdmin fixture enabled =
+  right $
+    mkDeadLetterAdmin fixture.pool $
+      (defaultDeadLetterAdminConfig [DlqRegistration fixture.dlqName [fixture.queueName]]) {redriveEnabled = enabled}
+
+awaitDlq :: Pool.Pool -> P.QueueName -> IO ()
+awaitDlq pool queue = go (500 :: Int)
+  where
+    go 0 = expectationFailure "handler did not commit its DLQ move"
+    go n = do
+      rows <- runPgmqSession pool $ S.peekMessages (Q.PeekMessages (P.queueNameToText queue) Nothing 1)
+      if V.null rows then threadDelay 10000 >> go (n - 1) else pure ()
+
+envelope :: Value -> Value
+envelope body =
+  object
+    [ "original_message" .= body,
+      "dead_letter_reason_code" .= ("orders.validation.failed" :: Text),
+      "dead_letter_reason_detail" .= String "detail"
+    ]
+
+putField :: Data.Aeson.Key.Key -> Value -> Value -> Value
+putField field value (Object fields) = Object (KM.insert field value fields)
+putField _ _ _ = error "expected object fixture"
+
+deleteField :: Data.Aeson.Key.Key -> Value -> Value
+deleteField field (Object fields) = Object (KM.delete field fields)
+deleteField _ _ = error "expected object fixture"
+
+seed :: TestFixture -> Value -> IO P.MessageId
+seed fixture body = runPgmqSession fixture.pool $ S.sendMessage (Q.SendMessage fixture.dlqName (P.MessageBody body) Nothing)
+
+seedWithHeaders :: TestFixture -> Value -> Value -> IO P.MessageId
+seedWithHeaders fixture body headers =
+  runPgmqSession fixture.pool $
+    S.sendMessageWithHeaders (Q.SendMessageWithHeaders fixture.dlqName (P.MessageBody body) (P.MessageHeaders headers) Nothing)
+
+snapshot :: Pool.Pool -> P.QueueName -> IO (V.Vector P.Message)
+snapshot pool q = runPgmqSession pool $ S.peekMessages (Q.PeekMessages (P.queueNameToText q) Nothing 10000)
+
+namedQueue :: Text -> P.QueueName
+namedQueue = either (error . show) id . P.parseQueueName
+
+isDatabaseError :: Either DlqError a -> Bool
+isDatabaseError (Left (DatabaseError _)) = True
+isDatabaseError _ = False
+
+isRedriven :: Either DlqError RedriveOutcome -> Bool
+isRedriven (Right (Redriven _)) = True
+isRedriven _ = False
+
+produceFailure :: TestFixture -> PgmqAdapterConfig -> Value -> IO ()
+produceFailure fixture config body = do
+  void $ runPgmqSession fixture.pool $ S.sendMessage (Q.SendMessage fixture.queueName (P.MessageBody body) Nothing)
+  outcome <- runEff $ runErrorNoCallStack @PgmqRuntimeError $ runPgmq fixture.pool $ runTracingNoop $ do
+    adapter <- pgmqAdapter (mkPgmqAdapterEnv fixture.pool) config >>= either (error . show) pure
+    app <- runApp defaultAppConfig [(ProcessorId "source-identity", mkProcessor adapter (\_ -> pure (AckDeadLetter (InvalidPayload "failed"))))] >>= either (error . show) pure
+    liftIO $ awaitDlq fixture.pool fixture.dlqName
+    void $ stopAppGracefully defaultShutdownConfig app
+  either throwIO pure outcome
+
+administrativePoolSpec :: Spec
+administrativePoolSpec = do
+  it "scopes statement and lock deadlines to each transaction on a borrowed single-connection pool" $
+    withPools $ \fixture adminPool -> do
+      admin <- right $ mkDeadLetterAdmin adminPool (defaultDeadLetterAdminConfig [DlqRegistration fixture.dlqName [fixture.queueName]])
+      before <- poolSettings adminPool
+      messageId <- seed fixture (envelope Null)
+      void $ browseDeadLetters admin fixture.dlqName defaultDlqPageRequest >>= right
+      void $ fetchDeadLetter admin fixture.dlqName messageId >>= right
+      poolSettings adminPool `shouldReturn` before
+      rows <- snapshot fixture.pool fixture.dlqName
+      let config =
+            (defaultDeadLetterAdminConfig [DlqRegistration fixture.dlqName [fixture.queueName]])
+              { redriveEnabled = True,
+                statementTimeoutMs = 1000,
+                lockTimeoutMs = 20
+              }
+      bounded <- right $ mkDeadLetterAdmin adminPool config
+      withLockedQueue fixture.pool fixture.dlqName $ do
+        result <- timeout 1000000 (redriveDeadLetter bounded fixture.dlqName messageId)
+        result `shouldSatisfy` maybe False isDatabaseError
+      poolSettings adminPool `shouldReturn` before
+      snapshot fixture.pool fixture.dlqName `shouldReturn` rows
+      -- A shorter statement deadline also interrupts read-side lock waits.
+      shortStatement <- right $ mkDeadLetterAdmin adminPool (config {statementTimeoutMs = 20, lockTimeoutMs = 1000})
+      withLockedQueue fixture.pool fixture.dlqName $ do
+        result <- timeout 1000000 (browseDeadLetters shortStatement fixture.dlqName defaultDlqPageRequest)
+        result `shouldSatisfy` maybe False isDatabaseError
+      poolSettings adminPool `shouldReturn` before
+      redriveDeadLetter bounded fixture.dlqName messageId >>= (`shouldSatisfy` isRedriven)
+      poolSettings adminPool `shouldReturn` before
+
+  it "lets cancellation escape without a successful absence or a lost DLQ row" $
+    withPools $ \fixture adminPool -> do
+      admin <-
+        right $
+          mkDeadLetterAdmin adminPool $
+            (defaultDeadLetterAdminConfig [DlqRegistration fixture.dlqName [fixture.queueName]]) {redriveEnabled = True}
+      messageId <- seed fixture (envelope (String "cancelled"))
+      before <- snapshot fixture.pool fixture.dlqName
+      withLockedQueue fixture.pool fixture.dlqName $
+        withAsync (redriveDeadLetter admin fixture.dlqName messageId) $ \request -> do
+          awaitBlockedRedrive fixture.pool fixture.dlqName
+          cancel request
+          waitCatch request >>= (`shouldSatisfy` isLeft)
+      snapshot fixture.pool fixture.dlqName `shouldReturn` before
+      redriveDeadLetter admin fixture.dlqName messageId >>= (`shouldSatisfy` isRedriven)
+
+  it "checks disabled mutations and invalid requests even with an unusable database pool" $ do
+    let settings = Settings.connectionString "host=/tmp/plan9-nonexistent-postgres connect_timeout=1"
+        config =
+          PoolConfig.settings
+            [ PoolConfig.size 1,
+              PoolConfig.acquisitionTimeout (secondsToDiffTime 1),
+              PoolConfig.staticConnectionSettings settings
+            ]
+        source = namedQueue "source"
+        dlq = namedQueue "dlq"
+    bracket (Pool.acquire config) Pool.release $ \pool -> do
+      admin <- right $ mkDeadLetterAdmin pool (defaultDeadLetterAdminConfig [DlqRegistration dlq [source]])
+      redriveDeadLetter admin source (P.MessageId 1) `shouldReturn` Left RedriveDisabled
+      browseDeadLetters admin dlq (DlqPageRequest 0 Nothing) `shouldReturn` Left (InvalidPageRequest (InvalidPageLimit 0))
+      fetchDeadLetter admin source (P.MessageId 1) `shouldReturn` Left (UnknownManagedDlq source)
+      fetchDeadLetter admin dlq (P.MessageId 1) >>= (`shouldSatisfy` isDatabaseError)
+
+withPools :: (TestFixture -> Pool.Pool -> IO ()) -> IO ()
+withPools action = withPgmqDbPools 4 1 (\workers admin -> withTestFixture workers (\fixture -> action fixture admin)) >>= either (error . show) pure
+
+poolSettings :: Pool.Pool -> IO (Text, Text)
+poolSettings pool =
+  runPgmqSession pool $
+    Session.statement () $
+      Statement.unpreparable
+        "select current_setting('statement_timeout'), current_setting('lock_timeout')"
+        E.noParams
+        (D.singleRow ((,) <$> D.column (D.nonNullable D.text) <*> D.column (D.nonNullable D.text)))
+
+withLockedQueue :: Pool.Pool -> P.QueueName -> IO a -> IO a
+withLockedQueue pool q action = do
+  ready <- newEmptyMVar
+  release <- newEmptyMVar
+  let sql statement = Session.statement () (Statement.unpreparable statement E.noParams D.noResult)
+      hold = runPgmqSession pool $ do
+        sql "begin"
+        sql ("lock table pgmq." <> Inspection.quoteIdentifier ("q_" <> P.queueNameToText q) <> " in access exclusive mode")
+        liftIO (putMVar ready ())
+        liftIO (takeMVar release)
+        sql "rollback"
+  bracket (async hold) (\worker -> putMVar release () >> void (waitCatch worker)) $ \_ -> do
+    acquired <- timeout 5000000 (takeMVar ready)
+    acquired `shouldBe` Just ()
+    action
+
+-- Wait for the request to be inside its transaction, rather than merely
+-- cancelling a thread which might still be acquiring its first connection.
+awaitBlockedRedrive :: Pool.Pool -> P.QueueName -> IO ()
+awaitBlockedRedrive pool q = go (200 :: Int)
+  where
+    patternText = "%from pgmq." <> Inspection.quoteIdentifier ("q_" <> P.queueNameToText q) <> "%for update"
+    waiting =
+      Session.statement patternText $
+        Statement.unpreparable
+          "select exists(select 1 from pg_stat_activity where wait_event_type = 'Lock' and query like $1)"
+          (E.param (E.nonNullable E.text))
+          (D.singleRow (D.column (D.nonNullable D.bool)))
+    go 0 = expectationFailure "redrive did not reach its row-lock wait"
+    go n = do
+      blocked <- runPgmqSession pool waiting
+      if blocked then pure () else threadDelay 10000 >> go (n - 1)
diff --git a/test/TmpPostgres.hs b/test/TmpPostgres.hs
--- a/test/TmpPostgres.hs
+++ b/test/TmpPostgres.hs
@@ -5,6 +5,7 @@
 module TmpPostgres
   ( -- * Test Execution
     withPgmqDb,
+    withPgmqDbPools,
     withRestartablePgmqDb,
     withTestFixture,
 
@@ -20,6 +21,7 @@
 import Control.Monad ((>=>))
 import Data.IORef (newIORef, readIORef, writeIORef)
 import Data.List.NonEmpty (NonEmpty ((:|)))
+import Data.Monoid qualified as EphemeralMonoid
 import Data.Text (Text)
 import Data.Text qualified as Text
 import Data.Time (secondsToDiffTime)
@@ -35,6 +37,8 @@
 import Pgmq.Hasql.Sessions qualified as Pgmq
 import Pgmq.Migration qualified as Migration
 import Pgmq.Types (QueueName, parseQueueName)
+import System.Directory qualified as EphemeralDirectory
+import System.Posix.User qualified as EphemeralUser
 import System.Random (randomIO)
 
 -- | Test fixture containing pool and queue names for a test
@@ -49,7 +53,7 @@
 -- This creates an ephemeral PostgreSQL instance, installs the pgmq schema,
 -- and then runs the provided action with a connection pool.
 withPgmqDb :: (Pool.Pool -> IO a) -> IO (Either StartError a)
-withPgmqDb action = Pg.with $ \db -> do
+withPgmqDb action = withEphemeralPg $ \db -> do
   let connSettings = Pg.connectionSettings db
 
   -- Install pgmq schema
@@ -61,13 +65,21 @@
     Pool.release
     action
 
+-- | Independent worker and administrative pools against the same disposable DB.
+withPgmqDbPools :: Int -> Int -> (Pool.Pool -> Pool.Pool -> IO a) -> IO (Either StartError a)
+withPgmqDbPools workerSize adminSize action = withEphemeralPg $ \db -> do
+  let settings = Pg.connectionSettings db
+  installPgmqSchema settings
+  bracket (createPoolWithSize workerSize settings) Pool.release $ \workerPool ->
+    bracket (createPoolWithSize adminSize settings) Pool.release (action workerPool)
+
 -- | Run against an ephemeral database that the test may restart in place.
 -- The restart preserves the data directory, releases stale pooled connections,
 -- and lets the next pool use establish a fresh connection to the same socket.
 withRestartablePgmqDb ::
   (Pool.Pool -> IO (Either StartError ()) -> IO a) ->
   IO (Either StartError a)
-withRestartablePgmqDb action = Pg.with $ \initialDb ->
+withRestartablePgmqDb action = withEphemeralPg $ \initialDb ->
   bracket
     (newIORef initialDb)
     (readIORef >=> Pg.stop)
@@ -151,13 +163,26 @@
 
 -- | Create a connection pool from connection settings.
 createPool :: Settings.Settings -> IO Pool.Pool
-createPool connSettings = do
+createPool = createPoolWithSize 10
+
+createPoolWithSize :: Int -> Settings.Settings -> IO Pool.Pool
+createPoolWithSize poolSize connSettings = do
   let poolConfig =
         PoolConfig.settings
-          [ PoolConfig.size 10,
+          [ PoolConfig.size poolSize,
             PoolConfig.acquisitionTimeout (secondsToDiffTime 10),
             PoolConfig.agingTimeout (secondsToDiffTime 3600),
             PoolConfig.idlenessTimeout (secondsToDiffTime 60),
             PoolConfig.staticConnectionSettings connSettings
           ]
   Pool.acquire poolConfig
+
+-- | Stable per-user root lets the next invocation reap abandoned clusters.
+-- See mori://shinzui/ephemeral-pg/docs/guides (temporary-roots-and-stale-cleanup.md; artifact URI pending).
+withEphemeralPg :: (Pg.Database -> IO a) -> IO (Either Pg.StartError a)
+withEphemeralPg action = do
+  uid <- EphemeralUser.getEffectiveUserID
+  let root = "/tmp/ephpg-shibuya-pgmq-adapter-" <> show uid
+  EphemeralDirectory.createDirectoryIfMissing True root
+  let config = Pg.defaultConfig {Pg.temporaryRoot = EphemeralMonoid.Last (Just root)}
+  Pg.withConfig config action