diff --git a/CHANGELOG.md b/CHANGELOG.md
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -6,6 +6,51 @@
 
 ## [Unreleased]
 
+## 0.17.0.0 — 2026-09-17
+
+### Breaking Changes
+
+- `DlqEntry` now exposes `originalHeaders`; external record construction and
+  exhaustive positional matches must account for the new field. `purgeDlq` now
+  returns `PurgeDlqResult` instead of `()`. Callers that discard the result may
+  still compile but must inspect `PurgeDlqBlocked` rather than assuming deletion.
+- Require the released `pgmq-config`, `pgmq-core`, `pgmq-effectful`,
+  `pgmq-hasql`, and test-only `pgmq-migration` 0.6 family together with
+  `shibuya-pgmq-adapter ^>=0.16.0.0`. The re-exported `QueueMetrics` record now
+  includes nullable `defaultPartitionLength`, so callers that construct it or
+  match it positionally must handle the new field. Partitioned provisioning
+  deliberately passes `premake = Nothing` to retain the PGMQ server default
+  rather than exposing explicit premake control.
+- `Job` gains the required `jobOrdering` field. Explicit tuning must match the
+  declaration; invalid raw tuning and legacy FIFO batches larger than one now
+  throw `JobConsumptionConfigError` before adapter construction or reads.
+
+### New Features
+
+- Add `FifoHeads`, backed by PGMQ's grouped-head read in both worker and
+  one-shot paths. It safely batches independent absolute group heads while a
+  failed, invisible, or delayed head blocks successors in its own group.
+- Add `mkPartitionSpec` and `PartitionSpecConfigError` to reject obvious empty,
+  mixed-unit, non-positive, out-of-range, and too-short numeric partition
+  settings before database access. The raw constructor remains available for
+  server-specific values.
+
+- DLQ redrive preserves the wrapper's original producer headers, including FIFO,
+  trace, and application metadata. Missing and JSON-null legacy headers remain
+  headerless and never fall back to the DLQ row's own headers.
+- `archiveDlqEntries` archives known message ids even while inspection has hidden
+  them. Ordinary `purgeDlq` refuses when metrics report hidden rows and returns
+  the deleted or blocked count; `purgeDlqForce` is the distinctly named
+  unconditional escape hatch.
+
+### Other Changes
+
+- Clarify that partition retention drops whole active and archive partitions and
+  can remove unprocessed work after a sufficiently long outage or backlog.
+- Describe FIFO provisioning as PGMQ's conventional name-detected GIN index.
+  No supplemental grouped-read index is recommended pending reproducible
+  full-workload measurements, and Keiro installs no experimental startup DDL.
+
 ## 0.16.0.0 — 2026-09-07
 
 ### Other Changes
diff --git a/keiro-pgmq.cabal b/keiro-pgmq.cabal
--- a/keiro-pgmq.cabal
+++ b/keiro-pgmq.cabal
@@ -1,7 +1,7 @@
-cabal-version:   3.0
-name:            keiro-pgmq
-version:         0.16.0.0
-synopsis:        PostgreSQL job-queue (PGMQ) integration for Keiro
+cabal-version: 3.0
+name: keiro-pgmq
+version: 0.17.0.0
+synopsis: PostgreSQL job-queue (PGMQ) integration for Keiro
 description:
   A typed background-job queue for Keiro applications on top of PGMQ (the
   PostgreSQL-native message queue) and shibuya (a Broadway-style worker
@@ -9,29 +9,34 @@
   retry/dead-letter policy, then write a plain domain handler of type
   @p -> Eff es JobOutcome@.
 
-license:         BSD-3-Clause
-license-file:    LICENSE
-author:          Nadeem Bitar
-maintainer:      nadeem@gmail.com
-copyright:       2026 Nadeem Bitar
-category:        Control
-homepage:        https://github.com/shinzui/keiro#readme
-bug-reports:     https://github.com/shinzui/keiro/issues
-build-type:      Simple
-tested-with:     GHC >=9.12 && <9.13
+license: BSD-3-Clause
+license-file: LICENSE
+author: Nadeem Bitar
+maintainer: nadeem@gmail.com
+copyright: 2026 Nadeem Bitar
+category: Control
+homepage: https://github.com/shinzui/keiro#readme
+bug-reports: https://github.com/shinzui/keiro/issues
+build-type: Simple
+tested-with: ghc >=9.12 && <9.13
 extra-doc-files: CHANGELOG.md
 
 source-repository head
-  type:     git
+  type: git
   location: https://github.com/shinzui/keiro.git
 
 common warnings
   ghc-options:
-    -Wall -Wcompat -Widentities -Wincomplete-record-updates
-    -Wincomplete-uni-patterns -Wpartial-fields -Wredundant-constraints
+    -Wall
+    -Wcompat
+    -Widentities
+    -Wincomplete-record-updates
+    -Wincomplete-uni-patterns
+    -Wpartial-fields
+    -Wredundant-constraints
 
 common shared
-  default-language:   GHC2024
+  default-language: GHC2024
   default-extensions:
     DeriveAnyClass
     DuplicateRecordFields
@@ -43,7 +48,7 @@
     PackageImports
 
 library
-  import:          warnings, shared
+  import: warnings, shared
   exposed-modules:
     Keiro.PGMQ
     Keiro.PGMQ.Codec
@@ -52,49 +57,53 @@
     Keiro.PGMQ.Metrics
     Keiro.PGMQ.Runtime
 
-  hs-source-dirs:  src
+  hs-source-dirs: src
   build-depends:
-    , aeson                 >=2.2       && <2.3
-    , base                  >=4.21      && <5
-    , effectful-core        >=2.6       && <2.7
-    , hasql                 >=1.10      && <1.11
-    , hasql-pool            >=1.2       && <1.5
-    , hs-opentelemetry-api  >=1.0       && <1.1
-    , keiro-core            ^>=0.16.0.0
-    , pgmq-config           >=0.5       && <0.6
-    , pgmq-core             >=0.5       && <0.6
-    , pgmq-effectful        >=0.5       && <0.6
-    , pgmq-hasql            >=0.5       && <0.6
-    , shibuya-core          ^>=0.9.0.0
-    , shibuya-pgmq-adapter  ^>=0.14.0.0
-    , streamly-core         >=0.3       && <0.4
-    , text                  >=2.1       && <2.2
-    , time                  >=1.12      && <1.15
+    aeson >=2.2 && <2.3,
+    base >=4.21 && <5,
+    effectful-core >=2.6 && <2.7,
+    hasql >=1.10 && <1.11,
+    hasql-pool >=1.2 && <1.5,
+    hs-opentelemetry-api >=1.0 && <1.1,
+    keiro-core ^>=0.17.0.0,
+    pgmq-config >=0.6 && <0.7,
+    pgmq-core >=0.6 && <0.7,
+    pgmq-effectful >=0.6 && <0.7,
+    pgmq-hasql >=0.6 && <0.7,
+    shibuya-core ^>=0.9.0.0,
+    shibuya-pgmq-adapter ^>=0.16.0.0,
+    streamly-core >=0.3 && <0.4,
+    text >=2.1 && <2.2,
+    time >=1.12 && <1.15,
 
 test-suite keiro-pgmq-test
-  import:         warnings, shared
-  type:           exitcode-stdio-1.0
+  import: warnings, shared
+  type: exitcode-stdio-1.0
   hs-source-dirs: test
-  main-is:        Main.hs
-  ghc-options:    -threaded -rtsopts -with-rtsopts=-N
+  main-is: Main.hs
+  ghc-options:
+    -threaded
+    -rtsopts
+    -with-rtsopts=-N
+
   build-depends:
-    , aeson                                >=2.2       && <2.3
-    , base                                 >=4.21      && <5
-    , effectful-core                       >=2.6       && <2.7
-    , hasql                                >=1.10      && <1.11
-    , hasql-pool                           >=1.2       && <1.5
-    , hs-opentelemetry-api                 >=1.0       && <1.1
-    , hs-opentelemetry-exporter-in-memory  >=1.0       && <1.1
-    , hs-opentelemetry-propagator-w3c      >=1.0       && <1.1
-    , hs-opentelemetry-sdk                 >=1.0       && <1.1
-    , hspec                                >=2.11
-    , keiro-core                           ^>=0.16.0.0
-    , keiro-pgmq
-    , keiro-test-support                   ^>=0.16.0.0
-    , pgmq-config                          >=0.5       && <0.6
-    , pgmq-core                            >=0.5       && <0.6
-    , pgmq-effectful                       >=0.5       && <0.6
-    , pgmq-migration                       >=0.5       && <0.6
-    , shibuya-core                         ^>=0.9.0.0
-    , shibuya-pgmq-adapter                 ^>=0.14.0.0
-    , text                                 >=2.1       && <2.2
+    aeson >=2.2 && <2.3,
+    base >=4.21 && <5,
+    effectful-core >=2.6 && <2.7,
+    hasql >=1.10 && <1.11,
+    hasql-pool >=1.2 && <1.5,
+    hs-opentelemetry-api >=1.0 && <1.1,
+    hs-opentelemetry-exporter-in-memory >=1.0 && <1.1,
+    hs-opentelemetry-propagator-w3c >=1.0 && <1.1,
+    hs-opentelemetry-sdk >=1.0 && <1.1,
+    hspec >=2.11,
+    keiro-core ^>=0.17.0.0,
+    keiro-pgmq,
+    keiro-test-support ^>=0.17.0.0,
+    pgmq-config >=0.6 && <0.7,
+    pgmq-core >=0.6 && <0.7,
+    pgmq-effectful >=0.6 && <0.7,
+    pgmq-migration >=0.6 && <0.7,
+    shibuya-core ^>=0.9.0.0,
+    shibuya-pgmq-adapter ^>=0.16.0.0,
+    text >=2.1 && <2.2,
diff --git a/src/Keiro/PGMQ/Dlq.hs b/src/Keiro/PGMQ/Dlq.hs
--- a/src/Keiro/PGMQ/Dlq.hs
+++ b/src/Keiro/PGMQ/Dlq.hs
@@ -12,23 +12,46 @@
 -- treat metadata as optional so operators can still inspect legacy or hand-written
 -- rows.
 --
--- PGMQ does not expire DLQ rows by itself. The retention model is "archive-then-
--- purge": 'archiveDlq' retains dead letters by moving them into the archive table
--- @pgmq.a_<dlq>@ (preserving @enqueued_at@ / @read_ct@ and stamping @archived_at@)
--- for audit, while 'purgeDlq' deletes them permanently. An operator who needs an
--- audit trail runs 'archiveDlq' (retain) and may then 'purgeDlq' (clear the active
--- table); an operator who does not keeps using 'purgeDlq' alone. Either way, alert
--- on the DLQ's depth via 'Keiro.PGMQ.Metrics.jobDlqMetrics'.
+-- 'readDlq', 'redriveDlq', and count-based 'archiveDlq' read visible rows and hide
+-- each row they inspect for 30 seconds. To retain inspected rows without waiting,
+-- keep their 'dlqMessageId' values and pass them to 'archiveDlqEntries'. Verify
+-- that every requested id was returned before considering retention complete:
 --
--- Redrive is at-least-once: a crash after sending the original payload back to the
--- main queue but before deleting the DLQ row leaves the payload in both places.
--- Handlers must therefore be idempotent.
+-- @
+-- entries <- readDlq job 100
+-- let requested = fmap dlqMessageId entries
+-- archived <- archiveDlqEntries job requested
+-- if Set.fromList archived /= Set.fromList requested
+--   then stopAndInvestigate
+--   else purgeDlq job
+-- @
+--
+-- PGMQ does not expire ordinary DLQ rows by itself. Archiving moves active rows
+-- into @pgmq.a_<dlq>@ for audit. Partitioned archive retention remains subject to
+-- its configured partition maintenance; archiving does not promise indefinite
+-- retention. 'purgeDlq' deletes only after a metrics snapshot reports no hidden
+-- rows, returning 'PurgeDlqBlocked' otherwise. The snapshot and deletion are not
+-- atomic. For a full-queue audit, pause producers and other operators, inspect and
+-- archive every row that requires retention, verify the active depth is zero, and
+-- only then consider purge. A bounded read followed by purge can still delete
+-- uninspected visible rows. 'purgeDlqForce' is the explicitly unconditional escape
+-- hatch and permanently deletes hidden rows too. Alert on DLQ depth via
+-- 'Keiro.PGMQ.Metrics.jobDlqMetrics'.
+--
+-- Redrive preserves the wrapper's original producer headers, including FIFO group
+-- and trace metadata, but it sends a new main-queue row with a new id, a fresh read
+-- count, and a new position at the back of its group. Redrive is at-least-once: a
+-- crash after sending the original payload but before deleting the DLQ row leaves
+-- the payload in both places. Handlers must therefore be idempotent.
 module Keiro.PGMQ.Dlq
   ( DlqEntry (..),
     readDlq,
     redriveDlq,
+    PurgeDlqResult (..),
     purgeDlq,
+    purgeDlqForce,
     archiveDlq,
+    archiveDlqEntries,
     archiveDlqEntry,
     archiveDlqEntryById,
   )
@@ -44,13 +67,17 @@
 import "base" Data.Int (Int32, Int64)
 import "effectful-core" Effectful (Eff, IOE, (:>))
 import "pgmq-effectful" Pgmq.Effectful
-  ( Message (..),
+  ( BatchMessageQuery (..),
+    Message (..),
     MessageBody (..),
+    MessageHeaders (..),
     MessageId (..),
     MessageQuery (..),
     Pgmq,
+    QueueMetrics (..),
     ReadMessage (..),
     SendMessage (..),
+    SendMessageWithHeaders (..),
   )
 import "pgmq-effectful" Pgmq.Effectful qualified as Pgmq
 import "text" Data.Text (Text)
@@ -68,6 +95,8 @@
     originalMessageId :: !(Maybe Int64),
     originalEnqueuedAt :: !(Maybe UTCTime),
     readCount :: !(Maybe Int64),
+    -- | The producer headers preserved in the DLQ wrapper, when present.
+    originalHeaders :: !(Maybe Value),
     -- | Full DLQ wrapper for forensics.
     rawBody :: !Value
   }
@@ -78,7 +107,8 @@
     deadLetterReason :: !Text,
     envelopeOriginalMessageId :: !(Maybe Int64),
     envelopeOriginalEnqueuedAt :: !(Maybe UTCTime),
-    envelopeReadCount :: !(Maybe Int64)
+    envelopeReadCount :: !(Maybe Int64),
+    envelopeOriginalHeaders :: !(Maybe Value)
   }
 
 parseDlqEnvelope :: Value -> Either Text DlqEnvelope
@@ -93,13 +123,15 @@
         envelopeOriginalMessageId <- obj .:? "original_message_id"
         envelopeOriginalEnqueuedAt <- obj .:? "original_enqueued_at"
         envelopeReadCount <- obj .:? "read_count"
+        envelopeOriginalHeaders <- obj .:? "original_headers"
         pure
           DlqEnvelope
             { originalMessage,
               deadLetterReason,
               envelopeOriginalMessageId,
               envelopeOriginalEnqueuedAt,
-              envelopeReadCount
+              envelopeReadCount,
+              envelopeOriginalHeaders
             }
 
     firstText = \case
@@ -134,6 +166,7 @@
               originalMessageId = Nothing,
               originalEnqueuedAt = Nothing,
               readCount = Nothing,
+              originalHeaders = Nothing,
               rawBody = body
             }
         Right envelope ->
@@ -144,12 +177,14 @@
               originalMessageId = envelope.envelopeOriginalMessageId,
               originalEnqueuedAt = envelope.envelopeOriginalEnqueuedAt,
               readCount = envelope.envelopeReadCount,
+              originalHeaders = envelope.envelopeOriginalHeaders,
               rawBody = body
             }
 
--- | Move up to @n@ DLQ rows back to the main queue. Redriven messages start a
--- fresh PGMQ @read_ct@ on the main queue. Malformed DLQ wrappers are left in the
--- DLQ for inspection.
+-- | Move up to @n@ visible DLQ rows back to the main queue. Each read hides the
+-- DLQ row for 30 seconds. Redriven messages preserve wrapper producer headers but
+-- receive a new id, queue position, and fresh PGMQ @read_ct@. Malformed wrappers
+-- are left in the DLQ for inspection.
 redriveDlq :: (Pgmq :> es, IOE :> es) => Job p -> Int -> Eff es Int
 redriveDlq job n
   | n <= 0 = pure 0
@@ -179,13 +214,22 @@
         Left _err ->
           pure count
         Right envelope -> do
-          _ <-
-            Pgmq.sendMessage
-              SendMessage
-                { queueName = job.jobQueue.physicalName,
-                  messageBody = MessageBody envelope.originalMessage,
-                  delay = Nothing
-                }
+          _ <- case envelope.envelopeOriginalHeaders of
+            Just headers ->
+              Pgmq.sendMessageWithHeaders
+                SendMessageWithHeaders
+                  { queueName = job.jobQueue.physicalName,
+                    messageBody = MessageBody envelope.originalMessage,
+                    messageHeaders = MessageHeaders headers,
+                    delay = Nothing
+                  }
+            Nothing ->
+              Pgmq.sendMessage
+                SendMessage
+                  { queueName = job.jobQueue.physicalName,
+                    messageBody = MessageBody envelope.originalMessage,
+                    delay = Nothing
+                  }
           void $
             Pgmq.deleteMessage
               MessageQuery
@@ -194,12 +238,32 @@
                 }
           pure (count + 1)
 
--- | Delete all rows currently in the DLQ.
-purgeDlq :: (Pgmq :> es, IOE :> es) => Job p -> Eff es ()
-purgeDlq job =
-  void (Pgmq.deleteAllMessagesFromQueue job.jobQueue.dlqName)
+-- | Result of a visibility-safe DLQ purge attempt.
+data PurgeDlqResult
+  = -- | The DLQ had no hidden rows at the metrics snapshot and was purged.
+    PurgeDlqPurged !Int64
+  | -- | Purge was refused because this many rows were hidden at the snapshot.
+    PurgeDlqBlocked !Int64
+  deriving stock (Eq, Show)
 
--- | Archive (retain) up to @n@ DLQ rows: move each out of the active DLQ table
+-- | Delete all rows currently in the DLQ only when none are hidden by a prior
+-- read. The metrics check and deletion are separate operations, so callers that
+-- need a full-queue audit must also quiesce concurrent readers and writers.
+purgeDlq :: (Pgmq :> es, IOE :> es) => Job p -> Eff es PurgeDlqResult
+purgeDlq job = do
+  metrics <- Pgmq.queueMetrics job.jobQueue.dlqName
+  let invisible = metrics.queueLength - metrics.queueVisibleLength
+  if invisible > 0
+    then pure (PurgeDlqBlocked invisible)
+    else PurgeDlqPurged <$> Pgmq.deleteAllMessagesFromQueue job.jobQueue.dlqName
+
+-- | Unconditionally delete every active DLQ row, including hidden rows.
+purgeDlqForce :: (Pgmq :> es, IOE :> es) => Job p -> Eff es Int64
+purgeDlqForce job =
+  Pgmq.deleteAllMessagesFromQueue job.jobQueue.dlqName
+
+-- | Archive (retain) up to @n@ visible DLQ rows, hiding each inspected row for
+-- 30 seconds: move each out of the active DLQ table
 -- @pgmq.q_<dlq>@ into the archive table @pgmq.a_<dlq>@, preserving @enqueued_at@ /
 -- @read_ct@ and stamping @archived_at@. Returns the number archived. This is the
 -- audit-retention counterpart to the delete-only 'purgeDlq'. At-most-once per row
@@ -231,6 +295,17 @@
     archiveOne count message = do
       moved <- archiveDlqEntry job message.messageId
       pure (if moved then count + 1 else count)
+
+-- | Archive the specified DLQ rows, including rows hidden by a prior read.
+-- Returns only ids that were present and moved; returned order is unspecified.
+archiveDlqEntries :: (Pgmq :> es, IOE :> es) => Job p -> [MessageId] -> Eff es [MessageId]
+archiveDlqEntries _job [] = pure []
+archiveDlqEntries job messageIds =
+  Pgmq.batchArchiveMessages
+    BatchMessageQuery
+      { queueName = job.jobQueue.dlqName,
+        messageIds
+      }
 
 -- | Archive one specific DLQ row by message id. 'True' if a row was moved.
 archiveDlqEntry :: (Pgmq :> es, IOE :> es) => Job p -> MessageId -> Eff es Bool
diff --git a/src/Keiro/PGMQ/Job.hs b/src/Keiro/PGMQ/Job.hs
--- a/src/Keiro/PGMQ/Job.hs
+++ b/src/Keiro/PGMQ/Job.hs
@@ -7,7 +7,7 @@
 -- Integration Points and depended on by the two consumer migrations), and it
 -- keeps the producer signatures uniform with the processor/runner ones. We
 -- therefore silence the otherwise-correct redundant-constraint warning here.
-{-# OPTIONS_GHC -Wno-redundant-constraints #-}
+{-# OPTIONS_GHC -Wno-redundant-constraints -Wno-partial-fields #-}
 
 -- | Layer 2 of @keiro-pgmq@: the typed-'Job' ergonomics built on top of
 -- 'Keiro.PGMQ.Runtime'. This is the payoff layer that absorbs the boilerplate two
@@ -81,6 +81,7 @@
     JobOrdering (..),
     JobTuning (..),
     JobTuningConfigError (..),
+    JobConsumptionConfigError (..),
     mkJobTuning,
     defaultJobTuning,
     withOrdering,
@@ -104,6 +105,8 @@
     -- * Queue lifecycle
     QueueKind (..),
     PartitionSpec (..),
+    PartitionSpecConfigError (..),
+    mkPartitionSpec,
     QueueProvision (..),
     standardProvision,
     unloggedProvision,
@@ -157,7 +160,7 @@
     mergeTraceHeaders,
   )
 import "pgmq-effectful" Pgmq.Effectful qualified as Pgmq
-import "pgmq-effectful" Pgmq.Effectful.Effect (readGrouped, readGroupedRoundRobin)
+import "pgmq-effectful" Pgmq.Effectful.Effect (readGrouped, readGroupedHead, readGroupedRoundRobin)
 import "pgmq-hasql" Pgmq.Hasql.Statements.Types (ReadGrouped (..))
 import "shibuya-core" Shibuya.App
   ( AppConfig (..),
@@ -225,6 +228,7 @@
   )
 import "text" Data.Text (Text)
 import "text" Data.Text qualified as Text
+import "text" Data.Text.Read qualified as TextRead
 import "time" Data.Time (NominalDiffTime, nominalDiffTimeToSeconds)
 
 -- | What a job handler decides. Never exposes shibuya/PGMQ wire types to the caller.
@@ -297,17 +301,22 @@
 -- selection order (@msg_id@ ascending), with NO per-key delivery-order guarantee
 -- under concurrent workers, retries, or visibility-timeout expiry.
 --
--- 'FifoThroughput' and 'FifoRoundRobin' enable strict per-group ordering via PGMQ
--- message groups (the reserved @x-pgmq-group@ header). Within one group, messages
--- are delivered in strict send order; distinct groups proceed in parallel.
+-- 'FifoHeads' enables strict per-group ordering via PGMQ message groups (the
+-- reserved @x-pgmq-group@ header). It returns at most one absolute head per
+-- group, so a failed or delayed head blocks its successors while distinct groups
+-- stay independently eligible. Current Keiro handlers remain serial.
+--
 -- 'FifoThroughput' fills a batch from the oldest eligible group first (SQS-style,
 -- @read_grouped@); 'FifoRoundRobin' interleaves fairly across groups
--- (@read_grouped_rr@). Delivery is still at-least-once and there is no
--- deduplication, so handlers must be idempotent.
+-- (@read_grouped_rr@). Those legacy strategies may safely use only a batch size
+-- of one; larger batches can lease a successor before its predecessor settles.
+-- Delivery is still at-least-once and there is no deduplication, so handlers
+-- must be idempotent.
 data JobOrdering
   = Unordered
   | FifoThroughput
   | FifoRoundRobin
+  | FifoHeads
   deriving stock (Eq, Show)
 
 -- | How a consumer reads the queue.
@@ -356,6 +365,37 @@
 validPolling (LongPoll maxPollSeconds pollIntervalMs) =
   maxPollSeconds > 0 && pollIntervalMs > 0
 
+-- | Invalid or contradictory consumer configuration. These checks run before
+-- an adapter is constructed or a direct PGMQ read is issued, including when raw
+-- 'JobTuning' and 'Job' constructors are used.
+data JobConsumptionConfigError
+  = InvalidJobTuning !JobTuningConfigError
+  | JobOrderingMismatch
+      { jobOrderingDeclared :: !JobOrdering,
+        tuningOrderingGiven :: !JobOrdering
+      }
+  | UnsafeLegacyFifoBatch
+      { unsafeOrdering :: !JobOrdering,
+        unsafeBatchSize :: !Int32
+      }
+  deriving stock (Eq, Show)
+  deriving anyclass (Exception)
+
+validateJobConsumptionConfig :: Job p -> JobTuning -> Either JobConsumptionConfigError ()
+validateJobConsumptionConfig job tuning = do
+  case mkJobTuning tuning.visibilityTimeout tuning.batchSize tuning.polling of
+    Left err -> Left (InvalidJobTuning err)
+    Right _ -> Right ()
+  if job.jobOrdering == tuning.ordering
+    then Right ()
+    else Left JobOrderingMismatch {jobOrderingDeclared = job.jobOrdering, tuningOrderingGiven = tuning.ordering}
+  case tuning.ordering of
+    FifoThroughput | tuning.batchSize > 1 -> unsafe
+    FifoRoundRobin | tuning.batchSize > 1 -> unsafe
+    _ -> Right ()
+  where
+    unsafe = Left UnsafeLegacyFifoBatch {unsafeOrdering = tuning.ordering, unsafeBatchSize = tuning.batchSize}
+
 toPollingConfig :: JobPolling -> PollingConfig
 toPollingConfig (PollEvery interval) = StandardPolling interval
 toPollingConfig (LongPoll maxPollSeconds pollIntervalMs) = LongPolling maxPollSeconds pollIntervalMs
@@ -365,6 +405,7 @@
 toFifoConfig Unordered = Nothing
 toFifoConfig FifoThroughput = Just (FifoConfig ThroughputOptimized)
 toFifoConfig FifoRoundRobin = Just (FifoConfig RoundRobin)
+toFifoConfig FifoHeads = Just (FifoConfig HeadPerGroup)
 
 retryDelaySeconds :: RetryDelay -> NominalDiffTime
 retryDelaySeconds (RetryDelay seconds) = seconds
@@ -380,14 +421,16 @@
       clamped = max minSec (min maxSec seconds)
    in ceiling clamped
 
--- | A declarative job: a queue, a payload codec, and a retry policy, named for
--- telemetry. Construct one and pair it with a handler of type
+-- | A declarative job: a queue, a payload codec, an ordering contract, and a
+-- retry policy, named for telemetry. Construct one and pair it with a handler of type
 -- @p -> Eff es 'JobOutcome'@.
 data Job p = Job
   { -- | Used as the shibuya 'ProcessorId' and telemetry label.
     jobName :: !Text,
     jobQueue :: !QueueRef,
     jobCodec :: !(JobCodec p),
+    -- | Required consumption ordering. Explicit tuning must match this value.
+    jobOrdering :: !JobOrdering,
     jobPolicy :: !RetryPolicy
   }
 
@@ -529,9 +572,9 @@
 -- | Enqueue a payload into the FIFO group named by @groupKey@. The group key is
 -- written under the reserved @x-pgmq-group@ JSONB header, which PGMQ's grouped
 -- reads and the shibuya adapter use to order deliveries per group. Consume with an
--- ordered 'JobTuning' (see 'withOrdering') to honor the order; within one group,
--- messages are handled in strict send order while distinct groups proceed in
--- parallel.
+-- ordered 'JobTuning' (see 'withOrdering') to honor the order. With 'FifoHeads',
+-- messages are handled in strict send order within one group while distinct
+-- groups can be claimed together; current Keiro handlers remain serial.
 enqueueToGroup ::
   (Pgmq :> es, IOE :> es) => Job p -> Text -> p -> Eff es MessageId
 enqueueToGroup job groupKey p =
@@ -554,25 +597,92 @@
 --     table is truncated to empty on a database crash. For transient, regenerable
 --     work.
 --   * 'PartitionedKind' — storage split across child tables by time or message-id
---     range, managed by the PostgreSQL extension @pg_partman@. Requires a
---     @pg_partman@-enabled server (see 'partitionedProvision').
+--     range, managed by the PostgreSQL extension @pg_partman@. This mode is
+--     experimental and requires a @pg_partman@-enabled server. Retention drops
+--     whole old partitions, including unprocessed active rows (see
+--     'partitionedProvision').
 data QueueKind
   = StandardKind
   | UnloggedKind
   | PartitionedKind !PartitionSpec
   deriving stock (Eq, Show)
 
--- | Partition interval + retention interval for a partitioned queue. Both are
--- PostgreSQL/@pg_partman@ duration or integer strings — e.g. @"daily"@ or
--- @"10000"@ for the interval, @"7 days"@ or @"100000"@ for the retention.
+-- | Partition interval and retention policy for a partitioned queue. PGMQ passes
+-- both values to PostgreSQL/@pg_partman@: time values are opaque strings such as
+-- @"daily"@ and @"7 days"@, while positive PostgreSQL @INTEGER@ text selects
+-- message-id-range partitioning.
+--
+-- Retention is not per-message expiry. Maintenance permanently drops whole old
+-- partitions from both the active queue and its archive without checking whether
+-- an active message was processed. Size retention above the worst expected
+-- consumer outage and backlog age, with an operator-chosen safety margin.
+--
+-- The raw constructor is intentionally available for server-specific values but
+-- is unvalidated. Prefer 'mkPartitionSpec' for the bounded checks Keiro can make
+-- without duplicating PostgreSQL's interval parser.
 data PartitionSpec = PartitionSpec
   { partitionInterval :: !Text,
     retentionInterval :: !Text
   }
   deriving stock (Eq, Show)
 
+-- | Why 'mkPartitionSpec' rejected a partition configuration.
+data PartitionSpecConfigError
+  = EmptyPartitionInterval
+  | EmptyRetentionInterval
+  | NonPositivePartitionInterval !Integer
+  | NonPositiveRetentionInterval !Integer
+  | PartitionIntervalOutsidePostgresInteger !Integer
+  | MixedPartitionUnits !Text !Text
+  | RetentionBelowPartitionInterval !Integer !Integer
+  deriving stock (Eq, Show)
+
+-- | Construct a partition specification after applying Keiro's bounded
+-- preflight checks. Inputs are trimmed. Fully numeric values must use matching
+-- message-id units; the partition span must fit PostgreSQL's positive signed
+-- 32-bit @INTEGER@ classifier, and numeric retention must cover at least one
+-- configured span. Nonnumeric pairs remain opaque for PostgreSQL/@pg_partman@ to
+-- validate.
+--
+-- These checks reject obvious mistakes; they do not prove a safe retention
+-- horizon. Operators must still account for worst-case outage and backlog age.
+mkPartitionSpec :: Text -> Text -> Either PartitionSpecConfigError PartitionSpec
+mkPartitionSpec rawPartition rawRetention
+  | Text.null partition = Left EmptyPartitionInterval
+  | Text.null retention = Left EmptyRetentionInterval
+  | otherwise =
+      case (completeInteger partition, completeInteger retention) of
+        (Just partitionValue, Just retentionValue)
+          | partitionValue <= 0 -> Left (NonPositivePartitionInterval partitionValue)
+          | partitionValue > toInteger (maxBound :: Int32) ->
+              Left (PartitionIntervalOutsidePostgresInteger partitionValue)
+          | retentionValue <= 0 -> Left (NonPositiveRetentionInterval retentionValue)
+          | retentionValue < partitionValue ->
+              Left (RetentionBelowPartitionInterval partitionValue retentionValue)
+          | otherwise -> Right spec
+        (Just partitionValue, Nothing)
+          | partitionValue <= 0 -> Left (NonPositivePartitionInterval partitionValue)
+          | partitionValue > toInteger (maxBound :: Int32) ->
+              Left (PartitionIntervalOutsidePostgresInteger partitionValue)
+          | otherwise -> Left (MixedPartitionUnits partition retention)
+        (Nothing, Just retentionValue)
+          | retentionValue <= 0 -> Left (NonPositiveRetentionInterval retentionValue)
+          | otherwise -> Left (MixedPartitionUnits partition retention)
+        (Nothing, Nothing) -> Right spec
+  where
+    partition = Text.strip rawPartition
+    retention = Text.strip rawRetention
+    spec = PartitionSpec {partitionInterval = partition, retentionInterval = retention}
+
+    completeInteger value =
+      case TextRead.signed TextRead.decimal value of
+        Right (parsed, remainder) | Text.null remainder -> Just parsed
+        _ -> Nothing
+
 -- | The provisioning choice for a job's /main/ queue: which storage shape, and
--- whether to create the FIFO GIN index. The DLQ (when the policy enables one) is
+-- whether to create PGMQ's conventional FIFO GIN index on @headers@. Presence is
+-- reported by the upstream index name; it does not prove that PostgreSQL uses the
+-- index for grouped-read expressions. The DLQ (when the policy enables one) is
 -- always a plain standard queue with no FIFO index.
 data QueueProvision = QueueProvision
   { provisionKind :: !QueueKind,
@@ -588,12 +698,17 @@
 unloggedProvision :: QueueProvision
 unloggedProvision = QueueProvision {provisionKind = UnloggedKind, provisionFifoIndex = False}
 
--- | A partitioned main queue (no FIFO index) with the given interval/retention.
+-- | An experimental partitioned main queue (no FIFO index) with the given
+-- interval and retention. Maintenance drops whole old active and archive
+-- partitions, so unprocessed work can be lost when retention is shorter than an
+-- outage or backlog. Prefer a value built by 'mkPartitionSpec'.
 partitionedProvision :: PartitionSpec -> QueueProvision
 partitionedProvision spec =
   QueueProvision {provisionKind = PartitionedKind spec, provisionFifoIndex = False}
 
--- | Turn on FIFO-index creation for a provisioning choice.
+-- | Request PGMQ's conventional FIFO GIN index for a provisioning choice. This
+-- preserves upstream provisioning and presence-reporting behavior; it is not a
+-- claim that the index accelerates every grouped-read query.
 withFifoIndexProvision :: QueueProvision -> QueueProvision
 withFifoIndexProvision provision = provision {provisionFifoIndex = True}
 
@@ -615,7 +730,8 @@
             job.jobQueue.physicalName
             Config.PartitionConfig
               { Config.partitionInterval = spec.partitionInterval,
-                Config.retentionInterval = spec.retentionInterval
+                Config.retentionInterval = spec.retentionInterval,
+                Config.premake = Nothing
               }
     mainConfig
       | provision.provisionFifoIndex = Config.withFifoIndex mainBase
@@ -639,13 +755,18 @@
 ensureJobQueue :: (Pgmq :> es) => Job p -> Eff es ()
 ensureJobQueue = ensureJobQueueWith standardProvision
 
--- | Create the FIFO GIN index on the job's /main/ queue's @headers@ column —
--- the index PGMQ's grouped/ordered reads (@read_grouped@/@read_grouped_rr@) match
--- against. Idempotent: the index step is always re-applied and the underlying SQL
--- is @CREATE INDEX IF NOT EXISTS@, so a second call is a harmless no-op. Routing
+-- | Request PGMQ's conventional FIFO GIN index on the job's /main/ queue's
+-- @headers@ column. PGMQ reports presence by the conventional index name; neither
+-- presence nor successful creation proves that PostgreSQL uses it for the group
+-- expression in @read_grouped@, @read_grouped_rr@, or @read_grouped_head@. No
+-- supplemental expression index is currently recommended without reproducible
+-- workload measurements.
+--
+-- Idempotent: the index step is always re-applied and the underlying SQL is
+-- @CREATE INDEX IF NOT EXISTS@, so a second call is a harmless no-op. Routing
 -- through @pgmq-config@'s reconciler (which lists existing queues first) means
 -- calling this on an already-provisioned queue does not recreate the queue. This
--- is the artifact the FIFO ordered-delivery plan
+-- is the conventional provisioning artifact the FIFO ordered-delivery plan
 -- (@docs/plans/77-add-fifo-ordered-delivery-via-message-groups-to-keiro-pgmq.md@)
 -- consumes for ordered jobs.
 ensureFifoIndex :: (Pgmq :> es) => Job p -> Eff es ()
@@ -654,9 +775,10 @@
     [Config.withFifoIndex (Config.standardQueue job.jobQueue.physicalName)]
 
 -- | Provision an ordered job's queue: create the main queue (and the DLQ when
--- the policy uses one) plus the FIFO GIN index that grouped reads need. Composes
--- 'ensureJobQueue' and 'ensureFifoIndex'; both are idempotent, so this is safe to
--- call at every startup.
+-- the policy uses one) plus PGMQ's conventional FIFO GIN index. This preserves
+-- upstream's ordered-queue setup but does not claim grouped-read acceleration.
+-- Composes 'ensureJobQueue' and 'ensureFifoIndex'; both are idempotent, so this is
+-- safe to call at every startup.
 ensureOrderedJobQueue :: (Pgmq :> es) => Job p -> Eff es ()
 ensureOrderedJobQueue job = do
   ensureJobQueue job
@@ -731,6 +853,7 @@
   (JobContext es -> p -> Eff es JobOutcome) ->
   Eff es (ProcessorId, QueueProcessor es)
 jobProcessorWithContext tuning job handle = do
+  either (liftIO . throwIO) pure (validateJobConsumptionConfig job tuning)
   env <- ask
   adapter <-
     pgmqAdapter env (adapterConfigFor tuning job)
@@ -752,7 +875,7 @@
   (p -> Eff es JobOutcome) ->
   Eff es (ProcessorId, QueueProcessor es)
 jobProcessor job handle =
-  jobProcessorWithContext defaultJobTuning job (\_context p -> handle p)
+  jobProcessorWithContext (withOrdering job.jobOrdering defaultJobTuning) job (\_context p -> handle p)
 
 -- | Continuous, multi-processor run (the @rei@ cadence): run a supervised app
 -- over several processors built with 'jobProcessor'. Returns the app handle; the
@@ -865,9 +988,9 @@
   Job p ->
   (JobContext es -> p -> Eff es JobOutcome) ->
   Eff es Int
-runJobOnceWithContext tuning n job handle
-  | n <= 0 = pure 0
-  | otherwise = drain 0
+runJobOnceWithContext tuning n job handle = do
+  either (liftIO . throwIO) pure (validateJobConsumptionConfig job tuning)
+  if n <= 0 then pure 0 else drain 0
   where
     drain handled
       | handled >= n = pure handled
@@ -896,6 +1019,13 @@
                     visibilityTimeout = tuning.visibilityTimeout,
                     qty = qty
                   }
+            FifoHeads ->
+              readGroupedHead
+                ReadGrouped
+                  { queueName = job.jobQueue.physicalName,
+                    visibilityTimeout = tuning.visibilityTimeout,
+                    qty = qty
+                  }
           if null messages
             then pure handled
             else do
@@ -1042,7 +1172,7 @@
 runJobOnce n job handle =
   void $
     runJobOnceWithContext
-      defaultJobTuning
+      (withOrdering job.jobOrdering defaultJobTuning)
       n
       job
       (\_context p -> handle p)
diff --git a/src/Keiro/PGMQ/Metrics.hs b/src/Keiro/PGMQ/Metrics.hs
--- a/src/Keiro/PGMQ/Metrics.hs
+++ b/src/Keiro/PGMQ/Metrics.hs
@@ -7,8 +7,9 @@
 -- queue and its DEAD-LETTER queue without the caller deriving any physical name.
 --
 -- Use 'jobDlqMetrics' (its 'queueLength') for the depth alerting that
--- 'Keiro.PGMQ.Dlq' recommends; pair it with 'Keiro.PGMQ.Dlq.archiveDlq' /
--- 'Keiro.PGMQ.Dlq.purgeDlq' for retention.
+-- 'Keiro.PGMQ.Dlq' recommends; after inspection, pair it with
+-- 'Keiro.PGMQ.Dlq.archiveDlqEntries' and inspect the typed
+-- 'Keiro.PGMQ.Dlq.PurgeDlqResult' returned by 'Keiro.PGMQ.Dlq.purgeDlq'.
 module Keiro.PGMQ.Metrics
   ( QueueMetrics (..),
     jobQueueMetrics,
diff --git a/test/Main.hs b/test/Main.hs
--- a/test/Main.hs
+++ b/test/Main.hs
@@ -4,6 +4,7 @@
 {-# LANGUAGE DerivingStrategies #-}
 {-# LANGUAGE LambdaCase #-}
 {-# LANGUAGE OverloadedRecordDot #-}
+{-# LANGUAGE TypeApplications #-}
 
 -- | End-to-end integration test for @keiro-pgmq@.
 --
@@ -17,7 +18,7 @@
 module Main (main) where
 
 import Control.Concurrent (threadDelay)
-import Control.Exception (bracket, throwIO)
+import Control.Exception (bracket, throwIO, try)
 import Data.Aeson (FromJSON, ToJSON, Value (..), object, parseJSON, toJSON, (.=))
 import Data.Aeson.Key qualified as Key
 import Data.Aeson.KeyMap qualified as KeyMap
@@ -57,7 +58,7 @@
 import OpenTelemetry.Trace.Id.Generator.Default (defaultIdGenerator)
 import OpenTelemetry.Util (appendOnlyBoundedCollectionValues)
 import Pgmq.Config.Types qualified as Config
-import Pgmq.Effectful (Message (..), MessageBody (..), Pgmq, QueueMetrics (..), ReadMessage (..), SendMessage (..))
+import Pgmq.Effectful (Message (..), MessageBody (..), Pgmq, ReadMessage (..), SendMessage (..))
 import Pgmq.Effectful qualified as Pgmq
 import Pgmq.Migration qualified as Migration
 import Pgmq.Types (QueueName, parseQueueName, queueNameToText)
@@ -130,6 +131,7 @@
     { jobName = name,
       jobQueue = queueRef name,
       jobCodec = aesonJobCodec,
+      jobOrdering = Unordered,
       jobPolicy = defaultRetryPolicy
     }
 
@@ -376,6 +378,33 @@
     mkJobTuning 30 1 (PollEvery 1)
       `shouldBe` Right defaultJobTuning
 
+  it "validates partition specifications without parsing time intervals" $ \_connStr -> do
+    mkPartitionSpec " daily " " 7 days "
+      `shouldBe` Right (PartitionSpec "daily" "7 days")
+    mkPartitionSpec " 10000 " " 100000 "
+      `shouldBe` Right (PartitionSpec "10000" "100000")
+    mkPartitionSpec "" "7 days"
+      `shouldBe` Left EmptyPartitionInterval
+    mkPartitionSpec "daily" "   "
+      `shouldBe` Left EmptyRetentionInterval
+    mkPartitionSpec "0" "100"
+      `shouldBe` Left (NonPositivePartitionInterval 0)
+    mkPartitionSpec "-1" "100"
+      `shouldBe` Left (NonPositivePartitionInterval (-1))
+    mkPartitionSpec "100" "0"
+      `shouldBe` Left (NonPositiveRetentionInterval 0)
+    mkPartitionSpec "100" "-1"
+      `shouldBe` Left (NonPositiveRetentionInterval (-1))
+    mkPartitionSpec "2147483648" "2147483648"
+      `shouldBe` Left (PartitionIntervalOutsidePostgresInteger 2147483648)
+    mkPartitionSpec "100" "7 days"
+      `shouldBe` Left (MixedPartitionUnits "100" "7 days")
+    mkPartitionSpec "daily" "100"
+      `shouldBe` Left (MixedPartitionUnits "daily" "100")
+    mkPartitionSpec "100" "99"
+      `shouldBe` Left (RetentionBelowPartitionInterval 100 99)
+    PartitionSpec "" "" `shouldBe` PartitionSpec "" ""
+
   it "derives distinct physical names for long logical queue names" $ \_connStr -> do
     let commonPrefix = Text.replicate 43 "a"
         first = queueRef (commonPrefix <> "x")
@@ -688,10 +717,11 @@
   it "readDlq decodes the original dead-lettered payload" $ \connStr -> do
     let job = mkJob "keiro_pgmq_test.dlq_read"
         payload = Ping "poison" 3
+        headers = object ["tenant" .= ("acme" :: Text)]
     entries <-
       runDb connStr $ do
         ensureJobQueue job
-        _ <- enqueue job payload
+        _ <- enqueueWithHeaders job (MessageHeaders headers) payload
         runJobOnce 1 job (\_ -> pure (Dead "bad"))
         readDlq job 1
     case entries of
@@ -700,6 +730,7 @@
         entry.originalPayload `shouldBe` Right payload
         entry.originalMessageId `shouldSatisfy` (/= Nothing)
         entry.readCount `shouldBe` Just 1
+        entry.originalHeaders `shouldBe` Just headers
       _ -> expectationFailure ("expected one DLQ entry, got " <> show (length entries))
 
   it "redriveDlq moves dead-lettered payloads back to the main queue" $ \connStr -> do
@@ -720,16 +751,127 @@
     finalMainLen <- runDb connStr (queueLen job.jobQueue.physicalName)
     finalMainLen `shouldBe` 0
 
-  it "purgeDlq empties the DLQ" $ \connStr -> do
+  it "redriveDlq preserves the wrapper's original headers" $ \connStr -> do
+    let job = mkJob "keiro_pgmq_test.dlq_redrive_headers"
+        originalHeaders =
+          object
+            [ "x-pgmq-group" .= ("orders" :: Text),
+              "traceparent" .= ("00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01" :: Text),
+              "tenant" .= ("acme" :: Text)
+            ]
+    messages <-
+      runDb connStr $ do
+        ensureJobQueue job
+        _ <- enqueueWithHeaders job (MessageHeaders originalHeaders) (Ping "redrive-headers" 1)
+        runJobOnce 1 job (\_ -> pure (Dead "bad"))
+        redriven <- redriveDlq job 1
+        liftIO (redriven `shouldBe` 1)
+        readMessages job.jobQueue.physicalName 1
+    case messages of
+      [message] -> message.headers `shouldBe` Just originalHeaders
+      _ -> expectationFailure ("expected one redriven message, got " <> show (length messages))
+
+  it "redriveDlq uses wrapper headers instead of the DLQ row's headers" $ \connStr -> do
+    let job = mkJob "keiro_pgmq_test.dlq_redrive_wrapper_headers"
+        originalHeaders = object ["tenant" .= ("original" :: Text)]
+        dlqRowHeaders = object ["tenant" .= ("dlq-row" :: Text)]
+        wrapper =
+          object
+            [ "original_message" .= Ping "wrapper-headers" 1,
+              "dead_letter_reason" .= ("poison_pill: bad" :: Text),
+              "original_headers" .= originalHeaders
+            ]
+    messages <-
+      runDb connStr $ do
+        ensureJobQueue job
+        _ <-
+          Pgmq.sendMessageWithHeaders
+            Pgmq.SendMessageWithHeaders
+              { queueName = job.jobQueue.dlqName,
+                messageBody = MessageBody wrapper,
+                messageHeaders = MessageHeaders dlqRowHeaders,
+                delay = Nothing
+              }
+        _ <- redriveDlq job 1
+        readMessages job.jobQueue.physicalName 1
+    case messages of
+      [message] -> message.headers `shouldBe` Just originalHeaders
+      _ -> expectationFailure ("expected one redriven message, got " <> show (length messages))
+
+  it "redriveDlq keeps missing and null wrapper headers headerless" $ \connStr -> do
+    let rowHeaders = MessageHeaders (object ["tenant" .= ("dlq-row" :: Text)])
+        redriveLegacy queue wrapper = do
+          let job = mkJob queue
+          runDb connStr $ do
+            ensureJobQueue job
+            _ <-
+              Pgmq.sendMessageWithHeaders
+                Pgmq.SendMessageWithHeaders
+                  { queueName = job.jobQueue.dlqName,
+                    messageBody = MessageBody wrapper,
+                    messageHeaders = rowHeaders,
+                    delay = Nothing
+                  }
+            _ <- redriveDlq job 1
+            readMessages job.jobQueue.physicalName 1
+        missingHeaders =
+          object
+            [ "original_message" .= Ping "missing-headers" 1,
+              "dead_letter_reason" .= ("poison_pill: bad" :: Text)
+            ]
+        nullHeaders =
+          object
+            [ "original_message" .= Ping "null-headers" 1,
+              "dead_letter_reason" .= ("poison_pill: bad" :: Text),
+              "original_headers" .= Null
+            ]
+    missingMessages <- redriveLegacy "keiro_pgmq_test.dlq_redrive_missing_headers" missingHeaders
+    nullMessages <- redriveLegacy "keiro_pgmq_test.dlq_redrive_null_headers" nullHeaders
+    map (.headers) missingMessages `shouldBe` [Nothing]
+    map (.headers) nullMessages `shouldBe` [Nothing]
+
+  it "purgeDlq empties a fully visible DLQ and reports the deleted count" $ \connStr -> do
     let job = mkJob "keiro_pgmq_test.dlq_purge"
-    runDb connStr $ do
+    result <- runDb connStr $ do
       ensureJobQueue job
       _ <- enqueue job (Ping "purge" 1)
       runJobOnce 1 job (\_ -> pure (Dead "bad"))
       purgeDlq job
+    result `shouldBe` PurgeDlqPurged 1
     dlqLen <- runDb connStr (queueLen job.jobQueue.dlqName)
     dlqLen `shouldBe` 0
 
+  it "purgeDlq refuses when inspection has hidden a row and deletes nothing" $ \connStr -> do
+    let job = mkJob "keiro_pgmq_test.dlq_purge_blocked"
+    (result, remaining) <-
+      runDb connStr $ do
+        ensureJobQueue job
+        _ <- enqueue job (Ping "hidden" 1)
+        _ <- enqueue job (Ping "visible" 2)
+        runJobOnce 2 job (\_ -> pure (Dead "bad"))
+        inspected <- readDlq job 1
+        liftIO (length inspected `shouldBe` 1)
+        result <- purgeDlq job
+        remaining <- queueLen job.jobQueue.dlqName
+        pure (result, remaining)
+    result `shouldBe` PurgeDlqBlocked 1
+    remaining `shouldBe` 2
+
+  it "purgeDlqForce deletes hidden rows" $ \connStr -> do
+    let job = mkJob "keiro_pgmq_test.dlq_purge_force"
+    (purged, remaining) <-
+      runDb connStr $ do
+        ensureJobQueue job
+        _ <- enqueue job (Ping "hidden" 1)
+        runJobOnce 1 job (\_ -> pure (Dead "bad"))
+        inspected <- readDlq job 1
+        liftIO (length inspected `shouldBe` 1)
+        purged <- purgeDlqForce job
+        remaining <- queueLen job.jobQueue.dlqName
+        pure (purged, remaining)
+    purged `shouldBe` 1
+    remaining `shouldBe` 0
+
   it "readDlq preserves malformed DLQ wrappers as malformed entries" $ \connStr -> do
     let job = mkJob "keiro_pgmq_test.dlq_malformed"
     entries <-
@@ -749,6 +891,7 @@
         entry.originalPayload `shouldSatisfy` \case
           Left (JobPayloadMalformed _) -> True
           _ -> False
+        entry.originalHeaders `shouldBe` Nothing
       _ -> expectationFailure ("expected one malformed DLQ entry, got " <> show (length entries))
 
   it "undecodable payload routes to the DLQ" $ \connStr -> do
@@ -1033,7 +1176,7 @@
   it "a FIFO delivery carries shibuya.partition on its process span" $ \connStr -> do
     (provider, spansRef) <- setupCapturingProvider
     let tracer = OTel.makeTracer provider "keiro-pgmq-test" OTel.tracerOptions
-        job = mkJob "keiro_pgmq_test.span_partition"
+        job = (mkJob "keiro_pgmq_test.span_partition") {jobOrdering = FifoThroughput}
     drained <-
       runDbTraced connStr tracer $ do
         ensureOrderedJobQueue job
@@ -1066,13 +1209,14 @@
   -- EP-2 M2: partitioned config shape (pure) + pending live test.
   it "ensureJobQueueWith partitioned builds a partitioned QueueConfig" $ \_connStr -> do
     let job = mkJob "keiro_pgmq_test.partitioned"
-        spec = PartitionSpec {partitionInterval = "daily", retentionInterval = "7 days"}
-    case queueProvisionConfigs (partitionedProvision spec) job of
+        partitionSpec = either (error . show) id (mkPartitionSpec "daily" "7 days")
+    case queueProvisionConfigs (partitionedProvision partitionSpec) job of
       (mainCfg : _) ->
         case mainCfg.queueType of
           Config.PartitionedQueue pc -> do
             pc.partitionInterval `shouldBe` "daily"
             pc.retentionInterval `shouldBe` "7 days"
+            pc.premake `shouldBe` Nothing
             mainCfg.queueName `shouldBe` job.jobQueue.physicalName
           other ->
             expectationFailure
@@ -1110,7 +1254,7 @@
       _ -> expectationFailure ("expected one message, got " <> show (length msgs))
 
   it "ensureOrderedJobQueue is idempotent and the queue accepts grouped work" $ \connStr -> do
-    let job = mkJob "keiro_pgmq_test.ordered_setup"
+    let job = (mkJob "keiro_pgmq_test.ordered_setup") {jobOrdering = FifoThroughput}
     len <-
       runDb connStr $ do
         ensureOrderedJobQueue job
@@ -1122,10 +1266,263 @@
         queueLen job.jobQueue.physicalName
     len `shouldBe` 0
 
+  it "rejects invalid, mismatched, and unsafe raw consumption tuning before a read" $ \connStr -> do
+    let unorderedJob = mkJob "keiro_pgmq_test.invalid_consumption"
+        headsJob = unorderedJob {jobOrdering = FifoHeads}
+        throughputJob = unorderedJob {jobOrdering = FifoThroughput}
+        invalidTuning =
+          JobTuning
+            { visibilityTimeout = 30,
+              batchSize = 0,
+              polling = PollEvery 1,
+              ordering = Unordered
+            }
+        mismatchedTuning = defaultJobTuning
+        unsafeTuning =
+          withOrdering FifoThroughput $
+            either (error . show) id $
+              mkJobTuning 30 8 (PollEvery 1)
+        run tuning job =
+          runDb connStr $
+            runJobOnceWithContext tuning 0 job \_ctx _payload -> pure Done
+    invalid <- try @JobConsumptionConfigError (run invalidTuning unorderedJob)
+    mismatch <- try @JobConsumptionConfigError (run mismatchedTuning headsJob)
+    unsafeBatch <- try @JobConsumptionConfigError (run unsafeTuning throughputJob)
+    invalid `shouldBe` Left (InvalidJobTuning (NonPositiveBatchSize 0))
+    mismatch
+      `shouldBe` Left
+        JobOrderingMismatch
+          { jobOrderingDeclared = FifoHeads,
+            tuningOrderingGiven = Unordered
+          }
+    unsafeBatch
+      `shouldBe` Left
+        UnsafeLegacyFifoBatch
+          { unsafeOrdering = FifoThroughput,
+            unsafeBatchSize = 8
+          }
+
+  it "accepts large batches for unordered and grouped-head jobs" $ \connStr -> do
+    let unorderedJob = mkJob "keiro_pgmq_test.valid_unordered_batch"
+        headsJob = (mkJob "keiro_pgmq_test.valid_heads_batch") {jobOrdering = FifoHeads}
+        unorderedTuning = either (error . show) id $ mkJobTuning 30 50 (PollEvery 1)
+        headsTuning = withOrdering FifoHeads unorderedTuning
+        run tuning job =
+          runDb connStr $
+            runJobOnceWithContext tuning 0 job \_ctx _payload -> pure Done
+    run unorderedTuning unorderedJob `shouldReturn` 0
+    run headsTuning headsJob `shouldReturn` 0
+    headsTuning.batchSize `shouldBe` 50
+
+  it "rejects mismatched ordering before emitting a receive span" $ \connStr -> do
+    (provider, spansRef) <- setupCapturingProvider
+    let tracer = OTel.makeTracer provider "keiro-pgmq-test" OTel.tracerOptions
+        job = (mkJob "keiro_pgmq_test.mismatch_no_receive") {jobOrdering = FifoHeads}
+        receiveName = "receive " <> queueNameToText job.jobQueue.physicalName
+    result <-
+      try @JobConsumptionConfigError $
+        runDbTraced connStr tracer $
+          runJobOnceWithContext defaultJobTuning 1 job \_ctx _payload -> pure Done
+    case result of
+      Left JobOrderingMismatch {} -> pure ()
+      Left err -> expectationFailure ("unexpected mismatch error: " <> show err)
+      Right _ -> expectationFailure "expected mismatched ordering to fail"
+    spans <- capturedSpans provider spansRef
+    map csName spans `shouldNotSatisfy` elem receiveName
+
+  it "worker construction rejects invalid raw tuning and mismatched ordering before adapter use" $ \connStr -> do
+    let job = (mkJob "keiro_pgmq_test.worker_mismatch") {jobOrdering = FifoHeads}
+        invalidTuning =
+          JobTuning
+            { visibilityTimeout = 0,
+              batchSize = 1,
+              polling = PollEvery 1,
+              ordering = FifoHeads
+            }
+        build tuning =
+          runDb connStr $
+            jobProcessorWithContext tuning job \_ctx _payload -> pure Done
+    invalid <- try @JobConsumptionConfigError (build invalidTuning)
+    mismatch <-
+      try @JobConsumptionConfigError $
+        build defaultJobTuning
+    case invalid of
+      Left (InvalidJobTuning (NonPositiveVisibilityTimeout 0)) -> pure ()
+      Left err -> expectationFailure ("unexpected invalid-tuning error: " <> show err)
+      Right _ -> expectationFailure "expected worker construction to reject raw invalid tuning"
+    case mismatch of
+      Left JobOrderingMismatch {jobOrderingDeclared = FifoHeads, tuningOrderingGiven = Unordered} -> pure ()
+      Left err -> expectationFailure ("unexpected validation error: " <> show err)
+      Right _ -> expectationFailure "expected worker construction to reject mismatched ordering"
+
+  it "FifoHeads blocks a failed group head while other groups continue" $ \connStr -> do
+    firstA <- newIORef True
+    observed <- newIORef ([] :: [Text])
+    let job = (mkJob "keiro_pgmq_test.fifo_heads_failure") {jobOrdering = FifoHeads}
+        tuning =
+          withOrdering FifoHeads $
+            either (error . show) id $
+              mkJobTuning 1 50 (PollEvery 0.1)
+        handler _ctx payload = do
+          liftIO $ modifyIORef' observed (<> [payload.message])
+          shouldFail <- liftIO $ do
+            firstPending <- readIORef firstA
+            if payload.message == "a1" && firstPending
+              then writeIORef firstA False >> pure True
+              else pure False
+          if shouldFail
+            then liftIO (throwIO (userError "fail a1 once"))
+            else pure Done
+    (firstDrain, independentDrain) <-
+      runDb connStr $ do
+        ensureOrderedJobQueue job
+        _ <- enqueueToGroup job "a" (Ping "a1" 1)
+        _ <- enqueueToGroup job "a" (Ping "a2" 2)
+        _ <- enqueueToGroup job "b" (Ping "b1" 1)
+        _ <- enqueueToGroup job "b" (Ping "b2" 2)
+        firstDrain <- runJobOnceWithContext tuning 2 job handler
+        independentDrain <- runJobOnceWithContext tuning 2 job handler
+        pure (firstDrain, independentDrain)
+    firstDrain `shouldBe` 2
+    independentDrain `shouldBe` 0
+    readIORef observed `shouldReturn` ["a1", "b1", "b2"]
+    threadDelay 1_100_000
+    recovered <- runDb connStr $ runJobOnceWithContext tuning 2 job handler
+    recovered `shouldBe` 2
+    readIORef observed `shouldReturn` ["a1", "b1", "b2", "a1", "a2"]
+
+  it "FifoHeads blocks a group successor when its head returns Retry" $ \connStr -> do
+    observed <- newIORef ([] :: [Text])
+    let job = (mkJob "keiro_pgmq_test.fifo_heads_retry") {jobOrdering = FifoHeads}
+        tuning =
+          withOrdering FifoHeads $
+            either (error . show) id $
+              mkJobTuning 30 50 (PollEvery 0.1)
+        handler _ctx payload = do
+          liftIO $ modifyIORef' observed (<> [payload.message])
+          pure $
+            if payload.message == "a1"
+              then Retry (RetryDelay 30)
+              else Done
+    (firstDrain, blockedDrain) <-
+      runDb connStr $ do
+        ensureOrderedJobQueue job
+        _ <- enqueueToGroup job "a" (Ping "a1" 1)
+        _ <- enqueueToGroup job "a" (Ping "a2" 2)
+        _ <- enqueueToGroup job "b" (Ping "b1" 1)
+        firstDrain <- runJobOnceWithContext tuning 2 job handler
+        blockedDrain <- runJobOnceWithContext tuning 1 job handler
+        pure (firstDrain, blockedDrain)
+    firstDrain `shouldBe` 2
+    blockedDrain `shouldBe` 0
+    readIORef observed `shouldReturn` ["a1", "b1"]
+
+  it "FifoHeads drains sixteen independent heads with one receive" $ \connStr -> do
+    (provider, spansRef) <- setupCapturingProvider
+    let tracer = OTel.makeTracer provider "keiro-pgmq-test" OTel.tracerOptions
+        job = (mkJob "keiro_pgmq_test.fifo_heads_batch") {jobOrdering = FifoHeads}
+        tuning =
+          withOrdering FifoHeads $
+            either (error . show) id $
+              mkJobTuning 30 16 (PollEvery 0.1)
+        messages = [("g" <> Text.pack (show i), Ping ("m" <> Text.pack (show i)) i) | i <- [1 .. 16]]
+        receiveName = "receive " <> queueNameToText job.jobQueue.physicalName
+    drained <-
+      runDbTraced connStr tracer $ do
+        ensureOrderedJobQueue job
+        traverse_ (\(groupName, payload) -> enqueueToGroup job groupName payload) messages
+        runJobOnceWithContext tuning 16 job \_ctx _payload -> pure Done
+    drained `shouldBe` 16
+    spans <- capturedSpans provider spansRef
+    length (filter ((== receiveName) . csName) spans) `shouldBe` 1
+
+  it "FifoHeads blocks a successor behind a delayed head while other groups run" $ \connStr -> do
+    observed <- newIORef ([] :: [Text])
+    let job = (mkJob "keiro_pgmq_test.fifo_heads_delay") {jobOrdering = FifoHeads}
+        tuning =
+          withOrdering FifoHeads $
+            either (error . show) id $
+              mkJobTuning 30 50 (PollEvery 1)
+        handler _ctx payload = do
+          liftIO $ modifyIORef' observed (<> [payload.message])
+          pure Done
+    independentDrain <-
+      runDb connStr $ do
+        ensureOrderedJobQueue job
+        _ <- enqueueToGroup job "a" (Ping "a1" 1)
+        _ <- runJobOnceWithContext tuning 1 job handler
+        _ <- enqueueToGroupWithDelay job 2 "a" (Ping "a2" 2)
+        _ <- enqueueToGroup job "a" (Ping "a3" 3)
+        _ <- enqueueToGroup job "b" (Ping "b1" 1)
+        runJobOnceWithContext tuning 3 job handler
+    independentDrain `shouldBe` 1
+    readIORef observed `shouldReturn` ["a1", "b1"]
+    threadDelay 2_100_000
+    recovered <- runDb connStr $ runJobOnceWithContext tuning 2 job handler
+    recovered `shouldBe` 2
+    readIORef observed `shouldReturn` ["a1", "b1", "a2", "a3"]
+
+  it "FifoHeads advances a group after its head is dead-lettered" $ \connStr -> do
+    observed <- newIORef ([] :: [Text])
+    let job = (mkJob "keiro_pgmq_test.fifo_heads_dead") {jobOrdering = FifoHeads}
+        tuning = withOrdering FifoHeads defaultJobTuning
+        record outcome _ctx payload = do
+          liftIO $ modifyIORef' observed (<> [payload.message])
+          pure outcome
+    (deadCount, successorCount, mainDepth, dlqDepth) <-
+      runDb connStr $ do
+        ensureOrderedJobQueue job
+        _ <- enqueueToGroup job "a" (Ping "a1" 1)
+        _ <- enqueueToGroup job "a" (Ping "a2" 2)
+        deadCount <- runJobOnceWithContext tuning 1 job (record (Dead "poison"))
+        successorCount <- runJobOnceWithContext tuning 1 job (record Done)
+        mainDepth <- queueLen job.jobQueue.physicalName
+        dlqDepth <- queueLen job.jobQueue.dlqName
+        pure (deadCount, successorCount, mainDepth, dlqDepth)
+    deadCount `shouldBe` 1
+    successorCount `shouldBe` 1
+    mainDepth `shouldBe` 0
+    dlqDepth `shouldBe` 1
+    readIORef observed `shouldReturn` ["a1", "a2"]
+
+  it "default one-shot tuning inherits the job's grouped-head ordering" $ \connStr -> do
+    let job = (mkJob "keiro_pgmq_test.default_heads") {jobOrdering = FifoHeads}
+    len <-
+      runDb connStr $ do
+        ensureOrderedJobQueue job
+        _ <- enqueueToGroup job "a" (Ping "a1" 1)
+        runJobOnce 1 job (\_payload -> pure Done)
+        queueLen job.jobQueue.physicalName
+    len `shouldBe` 0
+
+  it "default worker tuning inherits the job's grouped-head ordering" $ \connStr -> do
+    let job = (mkJob "keiro_pgmq_test.default_heads_worker") {jobOrdering = FifoHeads}
+    result <-
+      try @JobConsumptionConfigError $
+        runDb connStr $
+          jobProcessor job (\_payload -> pure Done)
+    case result of
+      Left err -> expectationFailure ("default worker tuning was rejected: " <> show err)
+      Right _ -> pure ()
+
+  it "legacy FIFO strategies remain available with batch size one" $ \connStr -> do
+    traverse_
+      ( \(suffix, declaredOrdering) -> do
+          let job = (mkJob ("keiro_pgmq_test.legacy_fifo_" <> suffix)) {jobOrdering = declaredOrdering}
+              tuning = withOrdering declaredOrdering defaultJobTuning
+          drained <-
+            runDb connStr $ do
+              ensureOrderedJobQueue job
+              _ <- enqueueToGroup job "a" (Ping "a1" 1)
+              runJobOnceWithContext tuning 1 job \_ctx _payload -> pure Done
+          drained `shouldBe` 1
+      )
+      [("throughput", FifoThroughput), ("round_robin", FifoRoundRobin)]
+
   -- EP-3 M4: end-to-end ordering proof.
   it "FifoThroughput drain preserves strict within-group order and fully drains" $ \connStr -> do
     observed <- newIORef ([] :: [Text])
-    let job = mkJob "keiro_pgmq_test.fifo_order"
+    let job = (mkJob "keiro_pgmq_test.fifo_order") {jobOrdering = FifoThroughput}
     drained <-
       runDb connStr $ do
         ensureOrderedJobQueue job
@@ -1144,13 +1541,13 @@
     filter (Text.isPrefixOf "a") log' `shouldBe` ["a1", "a2", "a3"]
     filter (Text.isPrefixOf "b") log' `shouldBe` ["b1", "b2"]
 
-  it "FifoThroughput worker path preserves within-group order" $ \connStr -> do
+  it "FifoHeads worker path preserves within-group order with a large batch" $ \connStr -> do
     observed <- newIORef ([] :: [Text])
-    let job = mkJob "keiro_pgmq_test.fifo_worker"
+    let job = (mkJob "keiro_pgmq_test.fifo_worker") {jobOrdering = FifoHeads}
         tuning =
-          withOrdering FifoThroughput $
+          withOrdering FifoHeads $
             either (error . show) id $
-              mkJobTuning 30 1 (PollEvery 0.1)
+              mkJobTuning 30 50 (PollEvery 0.1)
     processed <-
       runDb connStr $ do
         ensureOrderedJobQueue job
@@ -1177,6 +1574,49 @@
     len <- runDb connStr (queueLen job.jobQueue.physicalName)
     len `shouldBe` 0
 
+  it "FifoHeads worker retries a failed head before its successor while other groups progress" $ \connStr -> do
+    firstA <- newIORef True
+    observed <- newIORef ([] :: [Text])
+    let job = (mkJob "keiro_pgmq_test.fifo_worker_retry") {jobOrdering = FifoHeads}
+        tuning =
+          withOrdering FifoHeads $
+            either (error . show) id $
+              mkJobTuning 30 50 (PollEvery 0.1)
+        handler _ctx payload = do
+          liftIO $ modifyIORef' observed (<> [payload.message])
+          retryA <- liftIO $ do
+            retryPending <- readIORef firstA
+            if payload.message == "a1" && retryPending
+              then writeIORef firstA False >> pure True
+              else pure False
+          pure $
+            if retryA
+              then Retry (RetryDelay 0)
+              else Done
+    processed <-
+      runDb connStr $ do
+        ensureOrderedJobQueue job
+        _ <- enqueueToGroup job "a" (Ping "a1" 1)
+        _ <- enqueueToGroup job "a" (Ping "a2" 2)
+        _ <- enqueueToGroup job "b" (Ping "b1" 1)
+        result <- runJobWorkers IgnoreFailures 16 [jobProcessorWithContext tuning job handler]
+        case result of
+          Left err -> liftIO $ fail ("runJobWorkers failed: " <> show err)
+          Right app -> do
+            ok <- liftIO $ waitUntil do
+              messages <- readIORef observed
+              pure $
+                filter (Text.isPrefixOf "a") messages == ["a1", "a1", "a2"]
+                  && filter (Text.isPrefixOf "b") messages == ["b1"]
+            stopAppQuickly app
+            pure ok
+    processed `shouldBe` True
+    log' <- readIORef observed
+    filter (Text.isPrefixOf "a") log' `shouldBe` ["a1", "a1", "a2"]
+    filter (Text.isPrefixOf "b") log' `shouldBe` ["b1"]
+    len <- runDb connStr (queueLen job.jobQueue.physicalName)
+    len `shouldBe` 0
+
   -- EP-4 M1: typed metrics surface (main + DLQ).
   it "jobQueueMetrics reports main-queue depth after enqueue" $ \connStr -> do
     let job = mkJob "keiro_pgmq_test.metrics_depth"
@@ -1191,6 +1631,7 @@
         pure (mainMetrics, dlqMetrics)
     mainMetrics.queueLength `shouldBe` 3
     mainMetrics.queueVisibleLength `shouldBe` 3
+    mainMetrics.defaultPartitionLength `shouldBe` Nothing
     dlqMetrics.queueLength `shouldBe` 0
 
   it "jobDlqMetrics reports DLQ depth after a Dead outcome" $ \connStr -> do
@@ -1222,18 +1663,68 @@
     retained <- archiveCount connStr (queueNameToText job.jobQueue.dlqName)
     retained `shouldBe` 1
 
+  it "archiveDlqEntries archives inspected rows immediately and is idempotent" $ \connStr -> do
+    let job = mkJob "keiro_pgmq_test.dlq_archive_entries"
+        unknownId = Pgmq.MessageId 999999
+    (entryIds, countArchived, moved, repeated, unknown, remaining) <-
+      runDb connStr $ do
+        ensureJobQueue job
+        _ <- enqueue job (Ping "first" 1)
+        _ <- enqueue job (Ping "second" 2)
+        runJobOnce 2 job (\_ -> pure (Dead "bad"))
+        entries <- readDlq job 2
+        let entryIds = map (.dlqMessageId) entries
+        countArchived <- archiveDlq job 10
+        empty <- archiveDlqEntries job []
+        liftIO (empty `shouldBe` [])
+        moved <- archiveDlqEntries job entryIds
+        repeated <- archiveDlqEntries job entryIds
+        unknown <- archiveDlqEntries job [unknownId]
+        remaining <- queueLen job.jobQueue.dlqName
+        pure (entryIds, countArchived, moved, repeated, unknown, remaining)
+    length entryIds `shouldBe` 2
+    countArchived `shouldBe` 0
+    moved `shouldMatchList` entryIds
+    repeated `shouldBe` []
+    unknown `shouldBe` []
+    remaining `shouldBe` 0
+    retained <- archiveCount connStr (queueNameToText job.jobQueue.dlqName)
+    retained `shouldBe` 2
+
+  it "inspect archive-by-ids and guarded purge completes without waiting" $ \connStr -> do
+    let job = mkJob "keiro_pgmq_test.dlq_no_wait_runbook"
+    (inspectedCount, inspectedIds, archivedIds, purgeResult, remaining) <-
+      runDb connStr $ do
+        ensureJobQueue job
+        _ <- enqueue job (Ping "first" 1)
+        _ <- enqueue job (Ping "second" 2)
+        runJobOnce 2 job (\_ -> pure (Dead "bad"))
+        entries <- readDlq job 2
+        let inspectedIds = map (.dlqMessageId) entries
+        archivedIds <- archiveDlqEntries job inspectedIds
+        purgeResult <- purgeDlq job
+        remaining <- queueLen job.jobQueue.dlqName
+        pure (length entries, inspectedIds, archivedIds, purgeResult, remaining)
+    inspectedCount `shouldBe` 2
+    archivedIds `shouldMatchList` inspectedIds
+    purgeResult `shouldBe` PurgeDlqPurged 0
+    remaining `shouldBe` 0
+    retained <- archiveCount connStr (queueNameToText job.jobQueue.dlqName)
+    retained `shouldBe` 2
+
   -- EP-4 M3: end-to-end retention lifecycle.
   it "archived DLQ rows survive a purge" $ \connStr -> do
     let job = mkJob "keiro_pgmq_test.dlq_archive_purge"
-    archived <-
+    (archived, purged) <-
       runDb connStr $ do
         ensureJobQueue job
         _ <- enqueue job (Ping "poison" 1)
         runJobOnce 1 job (\_ -> pure (Dead "bad"))
         archived <- archiveDlq job 10
-        purgeDlq job
-        pure archived
+        purged <- purgeDlq job
+        pure (archived, purged)
     archived `shouldBe` 1
+    purged `shouldBe` PurgeDlqPurged 0
     dlqLen <- runDb connStr (queueLen job.jobQueue.dlqName)
     dlqLen `shouldBe` 0
     retained <- archiveCount connStr (queueNameToText job.jobQueue.dlqName)