diff --git a/CHANGELOG.md b/CHANGELOG.md
index 288461586..aa70b13e5 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -22,6 +22,7 @@ Notable changes between releases. Detailed migration notes for storage transitio
 
 ### Fixed
 
+- Queue and claim ring prunes no longer starve under sustained claim load (#721). Prune requested `ACCESS EXCLUSIVE` on the sealed slot's partitions for 10 ms, so with claims, completions and heartbeats always overlapping it never got the lock. That request now waits the prune lock timeout, lets waiting readers through, and retries with twice the wait up to 320 ms (`QueueStorage::prune_lock_wait_ceiling`) before reporting `blocked`. Queue prune also ran its safety checks under that lock, so when it did get it every reader of those partitions waited for the checks, which take seconds on a large slot. It now takes `SHARE ROW EXCLUSIVE` first, which blocks writes to the sealed slot and no reads, runs the checks, the failed-row carry and the rollup scan under it, and requests `ACCESS EXCLUSIVE` only for the `TRUNCATE`; after the upgrade it re-checks active leases and pending ready rows, which is what catches a lease committed while it waited. Queue prune requests the segment partition first because claims lock it first; a statement that holds a later partition while it waits for an earlier one still blocks that attempt until its timeout. Claim prune still runs its open-claim check under `ACCESS EXCLUSIVE`, because heartbeat, retry, fail and snooze read `lease_claims` and then update it in one transaction and would deadlock with a lock upgrade. A deadlock that PostgreSQL resolves by cancelling a prune is now counted as `blocked` rather than `error`. Readers of a sealed slot can now wait up to 320 ms behind a contended prune instead of 10 ms. The startup warning compares the 320 ms ceiling, not the base timeout, with `deadlock_timeout`, and eight times the ceiling with `statement_timeout`, because the lock timeout applies to each of the eight partitions one lock statement takes.
 - The embedded admin UI now works when its axum router is nested under a path such as `/awa` (#502). Assets, API requests, navigation and direct deep links automatically use the mount path, including outer nested routers, without rebuilding the frontend or changing existing router calls.
 - `admin::list_jobs` and `admin::job_kind_overviews` no longer omit a failed job, or leave it out of its kind's count, while queue prune moves its row out of the slot being pruned. Both read several queries from one snapshot and release partition locks between them, so the queries after a prune committed saw neither the pruned slot's rows nor the moved ones. These reads and the point reads such as `admin::get_job` now compare the relfilenode of every ring partition before their snapshot and after it ends, and read again in a new snapshot when prune or the terminal-count rollup truncated one in between, where 0.6.16 read a single job again only when it found nothing. After three reads that each overlapped a truncation, the fourth locks the ring partitions before taking its snapshot and holds them until it commits, so a read never returns a result a truncation could have changed; prune skips its pass if it cannot get its locks within its lock timeout. The comparison adds two catalog lookups to each read and takes no partition locks, so prune is blocked no more often than before.
 - Claiming from a ready slot that holds tombstones no longer scans the slot's whole tombstone partition on every claim (#713, reported and fixed by @Wallace99). Each claim checks its candidate rows against `ready_tombstones`, and for a small batch PostgreSQL could plan that check as a hashed subplan over the partition instead of one index probe per row: a 32-row claim read 10,000 tombstone rows in the regression test. The check now carries `OFFSET 0`, which keeps it correlated. Tombstones are left by cancelled, retried and moved jobs, so the cost grew with churn in the slot. Applied by the `correlated_claim_tombstones` schema patch; no configuration or protocol change.
diff --git a/awa-model/src/queue_storage.rs b/awa-model/src/queue_storage.rs
index 9b9cf77f4..2695177a7 100644
--- a/awa-model/src/queue_storage.rs
+++ b/awa-model/src/queue_storage.rs
@@ -619,6 +619,15 @@ fn is_lock_contention_error(err: &sqlx::Error) -> bool {
     )
 }
 
+/// A lock timeout, or a deadlock in which the prune was the victim.
+fn is_prune_contention_error(err: &sqlx::Error) -> bool {
+    matches!(
+        err,
+        sqlx::Error::Database(db_err)
+            if matches!(db_err.code().as_deref(), Some("55P03" | "40P01"))
+    )
+}
+
 async fn set_lock_timeout_tx(
     tx: &mut sqlx::Transaction<'_, Postgres>,
     timeout: Duration,
@@ -633,15 +642,75 @@ async fn set_lock_timeout_tx(
     Ok(())
 }
 
+/// The longest single wait for prune's `ACCESS EXCLUSIVE` upgrade when the
+/// prune lock timeout is shorter than this.
+const PRUNE_LOCK_UPGRADE_CEILING: Duration = Duration::from_millis(320);
+
+/// Partitions of a queue ring slot that queue prune locks in one statement.
+const QUEUE_PRUNE_LOCKED_PARTITIONS: u32 = 8;
+
+/// The longest single wait of [`acquire_prune_exclusive_lock_tx`] for `base`.
+fn prune_lock_upgrade_ceiling(base: Duration) -> Duration {
+    base.max(PRUNE_LOCK_UPGRADE_CEILING)
+}
+
+/// Take `ACCESS EXCLUSIVE` on `tables` for a prune's truncate.
+///
+/// A pending `ACCESS EXCLUSIVE` request stalls every new reader of the
+/// tables, so each attempt waits a bounded time and then lets the readers
+/// through before the next, longer attempt. The first attempt waits `base`
+/// and each later one twice as long, up to [`prune_lock_upgrade_ceiling`].
+/// A deadlock reported for the request counts as contention: PostgreSQL
+/// raises it at once when a transaction that already reads one of the tables
+/// is waiting to write a table the prune holds in `SHARE ROW EXCLUSIVE`. The
+/// last failure is returned, and `lock_timeout` is back at `base` on success.
+async fn acquire_prune_exclusive_lock_tx(
+    tx: &mut sqlx::Transaction<'_, Postgres>,
+    tables: &str,
+    base: Duration,
+) -> Result<(), AwaError> {
+    let ceiling = prune_lock_upgrade_ceiling(base);
+    let mut timeout = base;
+    loop {
+        sqlx::query("SAVEPOINT awa_prune_lock_upgrade")
+            .execute(tx.as_mut())
+            .await
+            .map_err(map_sqlx_error)?;
+        set_lock_timeout_tx(tx, timeout).await?;
+        let locked = sqlx::query(&format!("LOCK TABLE {tables} IN ACCESS EXCLUSIVE MODE"))
+            .execute(tx.as_mut())
+            .await;
+        match locked {
+            Ok(_) => {
+                sqlx::query("RELEASE SAVEPOINT awa_prune_lock_upgrade")
+                    .execute(tx.as_mut())
+                    .await
+                    .map_err(map_sqlx_error)?;
+                return set_lock_timeout_tx(tx, base).await;
+            }
+            Err(err) if is_prune_contention_error(&err) && timeout < ceiling => {
+                sqlx::query("ROLLBACK TO SAVEPOINT awa_prune_lock_upgrade")
+                    .execute(tx.as_mut())
+                    .await
+                    .map_err(map_sqlx_error)?;
+                tokio::time::sleep(base).await;
+                timeout = timeout.saturating_mul(2).min(ceiling);
+            }
+            Err(err) => return Err(map_sqlx_error(err)),
+        }
+    }
+}
+
 /// Commit or roll back a maintenance transaction that has set the prune
 /// `lock_timeout`.
 ///
 /// The timeout applies to every later statement in the transaction, not only
 /// the explicit `LOCK TABLE`: an insert, update or read of another relation can
 /// also queue behind a conflicting lock (autovacuum's heap truncation takes
-/// `ACCESS EXCLUSIVE`, for example). A lock timeout anywhere therefore rolls the
-/// whole transaction back and reports `blocked`, so the caller backs off and
-/// retries instead of surfacing an error.
+/// `ACCESS EXCLUSIVE`, for example). A lock timeout anywhere, or a deadlock
+/// that PostgreSQL resolves by cancelling the prune, therefore rolls the whole
+/// transaction back and reports `blocked`, so the caller backs off and retries
+/// instead of surfacing an error.
 async fn settle_prune_tx<T>(
     tx: sqlx::Transaction<'_, Postgres>,
     result: Result<T, AwaError>,
@@ -655,7 +724,7 @@ async fn settle_prune_tx<T>(
         Err(err) => {
             let _ = tx.rollback().await;
             match err {
-                AwaError::Database(ref db_err) if is_lock_contention_error(db_err) => Ok(blocked),
+                AwaError::Database(ref db_err) if is_prune_contention_error(db_err) => Ok(blocked),
                 err => Err(err),
             }
         }
@@ -3167,6 +3236,29 @@ impl QueueStorage {
         self.prune_lock_timeout
     }
 
+    /// The longest single lock wait of a prune.
+    ///
+    /// Queue and claim prune first wait [`Self::prune_lock_timeout`] for the
+    /// `ACCESS EXCLUSIVE` lock they truncate under, then retry with twice the
+    /// wait each time up to this value before reporting
+    /// [`PruneOutcome::Blocked`]. It is 320 ms, or the prune lock timeout
+    /// when that is longer.
+    pub fn prune_lock_wait_ceiling(&self) -> Duration {
+        prune_lock_upgrade_ceiling(self.prune_lock_timeout)
+    }
+
+    /// The longest a single prune lock statement can wait.
+    ///
+    /// `lock_timeout` applies to each relation a `LOCK TABLE` statement
+    /// acquires, and queue prune locks eight partitions in one statement, so
+    /// the statement can wait eight times [`Self::prune_lock_wait_ceiling`].
+    /// A `statement_timeout` at or below this can cancel the prune before
+    /// its lock timeout fires.
+    pub fn prune_lock_statement_wait_ceiling(&self) -> Duration {
+        self.prune_lock_wait_ceiling()
+            .saturating_mul(QUEUE_PRUNE_LOCKED_PARTITIONS)
+    }
+
     pub fn from_existing_schema(schema: impl Into<String>) -> Result<Self, AwaError> {
         Self::new(QueueStorageConfig {
             schema: schema.into(),
@@ -18177,16 +18269,34 @@ impl QueueStorage {
         let locked: Result<PruneOutcome, AwaError> = async {
             set_lock_timeout_tx(&mut tx, self.prune_lock_timeout).await?;
 
+            // Locks are requested in this order, and a statement that holds
+            // a later child while it waits behind the request for an earlier
+            // one deadlocks with prune until the lock timeout. Claims take
+            // the segment child first, and retry, fail and snooze take the
+            // done child before the ready child.
+            let slot_children = format!(
+                "{segment_child}, {done_child}, {ready_child}, {claim_attempt_child}, {tomb_child}, {receipt_batch_child}, {receipt_tomb_child}, {delta_child}"
+            );
+            // SHARE ROW EXCLUSIVE waits for every transaction that has
+            // written to this slot and blocks new writers, while claims,
+            // completions and admin reads keep reading the sealed children.
+            // The gates and the scans below can therefore take as long as
+            // the slot needs without stalling traffic; ACCESS EXCLUSIVE is
+            // only requested once the truncate is the next statement.
             sqlx::query(&format!(
-                "LOCK TABLE {ready_child}, {claim_attempt_child}, {done_child}, {tomb_child}, {segment_child}, {receipt_batch_child}, {receipt_tomb_child}, {delta_child} IN ACCESS EXCLUSIVE MODE"
+                "LOCK TABLE {slot_children} IN SHARE ROW EXCLUSIVE MODE"
             ))
             .execute(tx.as_mut())
             .await
             .map_err(map_sqlx_error)?;
+            sqlx::query("SAVEPOINT awa_prune_slot_frozen")
+                .execute(tx.as_mut())
+                .await
+                .map_err(map_sqlx_error)?;
 
-            // The pre-lock gates are only an MVCC snapshot. If an in-flight
-            // producer or claimer committed while the exclusive partition
-            // lock waited, these post-lock gates see it before TRUNCATE.
+            // The pre-lock gates are only an MVCC snapshot. Writers that
+            // touched the slot have committed by the time the lock is
+            // granted, so these gates see them.
             let active_leases =
                 queue_prune_has_active_leases_tx(&mut tx, schema, slot, generation).await?;
             if active_leases {
@@ -18412,9 +18522,37 @@ impl QueueStorage {
             .await
             .map_err(map_sqlx_error)?;
 
-            sqlx::query(&format!(
-                "TRUNCATE TABLE {ready_child}, {claim_attempt_child}, {done_child}, {tomb_child}, {segment_child}, {receipt_batch_child}, {receipt_tomb_child}, {delta_child}"
-            ))
+            acquire_prune_exclusive_lock_tx(&mut tx, &slot_children, self.prune_lock_timeout).await?;
+
+            // A receipt claim inserts its attempt batch into this slot, so
+            // it cannot have started after the slot was frozen, and closures
+            // only accumulate: the claim-reference proof above still holds.
+            // A claim that writes a lease directly touches no child of the
+            // slot, and one that read its ready row before the freeze can
+            // commit until the upgrade is granted. These two gates see it.
+            let active_leases =
+                queue_prune_has_active_leases_tx(&mut tx, schema, slot, generation).await?;
+            let pending = !active_leases
+                && queue_prune_has_pending_ready_tx(&mut tx, schema, &ready_child, generation)
+                    .await?;
+            if active_leases || pending {
+                // Drops the carried failed rows along with the slot locks.
+                sqlx::query("ROLLBACK TO SAVEPOINT awa_prune_slot_frozen")
+                    .execute(tx.as_mut())
+                    .await
+                    .map_err(map_sqlx_error)?;
+                return Ok(PruneOutcome::SkippedActive {
+                    slot,
+                    reason: if active_leases {
+                        SkipReason::QueueActiveLeases
+                    } else {
+                        SkipReason::QueuePendingReady
+                    },
+                    count: busy_indicator(true),
+                });
+            }
+
+            sqlx::query(&format!("TRUNCATE TABLE {slot_children}"))
             .execute(tx.as_mut())
             .await
             .map_err(map_sqlx_error)?;
@@ -18743,13 +18881,16 @@ impl QueueStorage {
             return Ok(PruneOutcome::Noop);
         };
 
-        // Lock the slot row so concurrent rotate/prune observe the same
-        // state machine transition.
+        // The ring state row locked above already serializes rotate and
+        // prune. The slot row is only read here: rescue takes its table
+        // locks on the slot's partitions and then locks this row for its
+        // cursor, so holding the row while waiting for ACCESS EXCLUSIVE on
+        // those partitions would deadlock with every rescue of the slot.
+        // Prune locks the row once it holds the partitions.
         let slot_locked: Option<i32> = sqlx::query_scalar(&format!(
             r#"
             SELECT slot FROM {schema}.claim_ring_slots
             WHERE slot = $1
-            FOR UPDATE
             "#
         ))
         .bind(slot)
@@ -18819,17 +18960,26 @@ impl QueueStorage {
         let locked: Result<PruneOutcome, AwaError> = async {
             set_lock_timeout_tx(&mut tx, self.prune_lock_timeout).await?;
 
+            let slot_children = format!(
+                "{claim_child}, {claim_batch_child}, {closure_child}, {closure_batch_child}"
+            );
+            // Heartbeat, retry, fail and snooze read `lease_claims` and then
+            // update it in the same transaction. A weaker lock held across
+            // the proof below would let them read and then block their
+            // update, and the request for ACCESS EXCLUSIVE would deadlock
+            // with every one of them. Claim prune therefore takes
+            // ACCESS EXCLUSIVE before the proof.
+            acquire_prune_exclusive_lock_tx(&mut tx, &slot_children, self.prune_lock_timeout)
+                .await?;
+
             sqlx::query(&format!(
-                "LOCK TABLE {claim_child}, {claim_batch_child}, {closure_child}, {closure_batch_child} IN ACCESS EXCLUSIVE MODE"
+                "SELECT slot FROM {schema}.claim_ring_slots WHERE slot = $1 FOR UPDATE"
             ))
+            .bind(slot)
             .execute(tx.as_mut())
             .await
             .map_err(map_sqlx_error)?;
 
-            // After taking ACCESS EXCLUSIVE, recheck that the slot is still
-            // not the current one. The ring-state lock should already make
-            // this stable, but keeping the explicit gate documents the
-            // truncate precondition.
             let current_slot: i32 = sqlx::query_scalar(&format!(
                 r#"
                 SELECT current_slot FROM {schema}.claim_ring_state WHERE singleton = TRUE
@@ -18864,11 +19014,9 @@ impl QueueStorage {
                 });
             }
 
-            sqlx::query(&format!(
-                "TRUNCATE TABLE {claim_child}, {claim_batch_child}, {closure_child}, {closure_batch_child}"
-            ))
-            .execute(tx.as_mut())
-            .await
+            sqlx::query(&format!("TRUNCATE TABLE {slot_children}"))
+                .execute(tx.as_mut())
+                .await
                 .map_err(map_sqlx_error)?;
             self.ensure_claim_batch_job_index_tx(&mut tx, slot).await?;
             sqlx::query(&format!(
diff --git a/awa-worker/src/client.rs b/awa-worker/src/client.rs
index 2d620f561..634404055 100644
--- a/awa-worker/src/client.rs
+++ b/awa-worker/src/client.rs
@@ -1588,8 +1588,15 @@ impl Client {
         let Some(runtime) = self.storage.queue_storage() else {
             return;
         };
-        let timeout = runtime.store.prune_lock_timeout();
-        let risks = match crate::storage::check_prune_lock_timeout(&self.pool, timeout).await {
+        let timeout = runtime.store.prune_lock_wait_ceiling();
+        let statement_wait = runtime.store.prune_lock_statement_wait_ceiling();
+        let risks = match crate::storage::check_prune_lock_timeout(
+            &self.pool,
+            timeout,
+            statement_wait,
+        )
+        .await
+        {
             Ok(risks) => risks,
             Err(err) => {
                 debug!(error = %err, "Could not read lock settings to check the prune lock timeout");
@@ -1599,17 +1606,20 @@ impl Client {
         for risk in risks {
             match risk {
                 crate::storage::PruneLockTimeoutRisk::DeadlockTimeout { limit_ms } => warn!(
-                    prune_lock_timeout_ms = timeout.as_millis() as u64,
+                    prune_lock_wait_ceiling_ms = timeout.as_millis() as u64,
                     deadlock_timeout_ms = limit_ms,
-                    "Prune lock timeout is at or above deadlock_timeout; a waiting prune can \
+                    "Prune's longest lock wait is at or above deadlock_timeout; a waiting prune can \
                      make PostgreSQL run deadlock detection and cancel a concurrent claim. \
-                     Keep the timeout well below deadlock_timeout"
+                     Keep the prune lock timeout and its 320 ms retry ceiling well below \
+                     deadlock_timeout"
                 ),
                 crate::storage::PruneLockTimeoutRisk::StatementTimeout { limit_ms } => warn!(
-                    prune_lock_timeout_ms = timeout.as_millis() as u64,
+                    prune_lock_statement_wait_ms = statement_wait.as_millis() as u64,
                     statement_timeout_ms = limit_ms,
-                    "Prune lock timeout is at or above statement_timeout; prune is cancelled \
-                     by the statement timeout and fails instead of reporting blocked"
+                    "A prune lock statement can wait as long as statement_timeout, because the \
+                     lock timeout applies to each of the eight partitions it locks; prune is \
+                     then cancelled by the statement timeout and fails instead of reporting \
+                     blocked"
                 ),
             }
         }
@@ -3153,18 +3163,26 @@ mod tests {
             .await
             .expect("connect with session limits");
 
-        assert!(check_prune_lock_timeout(&pool, Duration::from_millis(100))
-            .await
-            .unwrap()
-            .is_empty());
+        assert!(check_prune_lock_timeout(
+            &pool,
+            Duration::from_millis(100),
+            Duration::from_millis(100)
+        )
+        .await
+        .unwrap()
+        .is_empty());
         assert_eq!(
-            check_prune_lock_timeout(&pool, Duration::from_millis(300))
-                .await
-                .unwrap(),
+            check_prune_lock_timeout(
+                &pool,
+                Duration::from_millis(300),
+                Duration::from_millis(300)
+            )
+            .await
+            .unwrap(),
             vec![PruneLockTimeoutRisk::DeadlockTimeout { limit_ms: 200 }]
         );
         assert_eq!(
-            check_prune_lock_timeout(&pool, Duration::from_secs(5))
+            check_prune_lock_timeout(&pool, Duration::from_secs(5), Duration::from_secs(5))
                 .await
                 .unwrap(),
             vec![
diff --git a/awa-worker/src/storage.rs b/awa-worker/src/storage.rs
index 828c0a73d..6f06c06a7 100644
--- a/awa-worker/src/storage.rs
+++ b/awa-worker/src/storage.rs
@@ -79,10 +79,13 @@ pub(crate) enum PruneLockTimeoutRisk {
     StatementTimeout { limit_ms: u64 },
 }
 
-/// Compare a prune lock timeout with the session's `deadlock_timeout` and
-/// `statement_timeout` (milliseconds; a zero `statement_timeout` is disabled).
+/// Compare the longest single prune lock wait with the session's
+/// `deadlock_timeout`, and the longest wait of one prune lock statement with
+/// its `statement_timeout` (milliseconds; a zero `statement_timeout` is
+/// disabled).
 pub(crate) fn prune_lock_timeout_risks(
     timeout: Duration,
+    statement_wait: Duration,
     deadlock_timeout_ms: Option<u64>,
     statement_timeout_ms: Option<u64>,
 ) -> Vec<PruneLockTimeoutRisk> {
@@ -94,7 +97,7 @@ pub(crate) fn prune_lock_timeout_risks(
         }
     }
     if let Some(limit_ms) = statement_timeout_ms.filter(|limit| *limit > 0) {
-        if timeout_ms >= u128::from(limit_ms) {
+        if statement_wait.as_millis() >= u128::from(limit_ms) {
             risks.push(PruneLockTimeoutRisk::StatementTimeout { limit_ms });
         }
     }
@@ -106,6 +109,7 @@ pub(crate) fn prune_lock_timeout_risks(
 pub(crate) async fn check_prune_lock_timeout(
     pool: &sqlx::PgPool,
     timeout: Duration,
+    statement_wait: Duration,
 ) -> Result<Vec<PruneLockTimeoutRisk>, sqlx::Error> {
     let settings: Vec<(String, i64)> = sqlx::query_as(
         "SELECT name, setting::bigint FROM pg_settings \
@@ -121,6 +125,7 @@ pub(crate) async fn check_prune_lock_timeout(
     };
     Ok(prune_lock_timeout_risks(
         timeout,
+        statement_wait,
         setting("deadlock_timeout"),
         setting("statement_timeout"),
     ))
@@ -134,18 +139,38 @@ mod tests {
     fn prune_lock_timeout_risks_flag_timeouts_at_or_above_server_limits() {
         let below = Duration::from_millis(999);
         let at = Duration::from_millis(1000);
-        assert!(prune_lock_timeout_risks(below, Some(1000), Some(0)).is_empty());
+        assert!(prune_lock_timeout_risks(below, below, Some(1000), Some(0)).is_empty());
         assert_eq!(
-            prune_lock_timeout_risks(at, Some(1000), Some(0)),
+            prune_lock_timeout_risks(at, at, Some(1000), Some(0)),
             vec![PruneLockTimeoutRisk::DeadlockTimeout { limit_ms: 1000 }]
         );
         assert_eq!(
-            prune_lock_timeout_risks(Duration::from_secs(30), Some(1000), Some(5000)),
+            prune_lock_timeout_risks(
+                Duration::from_secs(30),
+                Duration::from_secs(30),
+                Some(1000),
+                Some(5000)
+            ),
             vec![
                 PruneLockTimeoutRisk::DeadlockTimeout { limit_ms: 1000 },
                 PruneLockTimeoutRisk::StatementTimeout { limit_ms: 5000 },
             ]
         );
-        assert!(prune_lock_timeout_risks(Duration::from_secs(30), None, None).is_empty());
+        assert!(prune_lock_timeout_risks(
+            Duration::from_secs(30),
+            Duration::from_secs(30),
+            None,
+            None
+        )
+        .is_empty());
+        assert_eq!(
+            prune_lock_timeout_risks(
+                Duration::from_millis(320),
+                Duration::from_millis(2560),
+                Some(1000),
+                Some(2000)
+            ),
+            vec![PruneLockTimeoutRisk::StatementTimeout { limit_ms: 2000 }]
+        );
     }
 }
diff --git a/awa/tests/queue_storage_runtime_test.rs b/awa/tests/queue_storage_runtime_test.rs
index 501665d23..ea033c9f4 100644
--- a/awa/tests/queue_storage_runtime_test.rs
+++ b/awa/tests/queue_storage_runtime_test.rs
@@ -11736,31 +11736,31 @@ async fn fail_jobs_in_sealed_slot(
 /// reaches the pruned slot after the prune commits sees the slot empty, and
 /// cannot see the failed row prune re-homed into the live slot either.
 ///
-/// Holds prune after it has locked slot 0 and before it carries the failed
-/// row into slot 1, starts `read` so it takes its snapshot and queues behind
-/// prune's slot 0 lock in the statement containing `read_fragment`, then lets
-/// prune commit.
+/// Holds prune after it has carried the failed row into slot 1 and truncated
+/// slot 0, at the counter cleanup that follows, starts `read` so it takes its
+/// snapshot and queues behind prune's slot 0 lock in the statement containing
+/// `read_fragment`, then lets prune commit.
 async fn read_across_failed_row_carry<T>(
     pool: &sqlx::PgPool,
     store: &QueueStorage,
     read_fragment: &str,
     read: impl std::future::Future<Output = T>,
 ) -> T {
-    let mut holder = pool.begin().await.expect("begin carry holder");
+    let mut holder = pool.begin().await.expect("begin prune holder");
     sqlx::query(&format!(
-        "LOCK TABLE {}.done_entries_1 IN SHARE MODE",
+        "LOCK TABLE {}.queue_terminal_live_counts IN SHARE MODE",
         store.schema()
     ))
     .execute(holder.as_mut())
     .await
-    .expect("hold the live done slot");
+    .expect("hold the live terminal counters");
 
     let prune = store.prune_oldest(pool, Duration::from_secs(3600));
     let read = async {
-        wait_for_lock_waiter(pool, "WITH carried AS").await;
+        wait_for_lock_waiter(pool, "queue_terminal_live_counts WHERE ready_slot").await;
         let release = async {
             wait_for_lock_waiter(pool, read_fragment).await;
-            holder.rollback().await.expect("release carry holder");
+            holder.rollback().await.expect("release prune holder");
         };
         let (value, ()) = tokio::join!(read, release);
         value
@@ -22833,3 +22833,647 @@ async fn cursor_advance_requires_claim_evidence(receipts: bool) {
         "the job whose claim was lost stays claimable"
     );
 }
+
+/// Children of one sealed slot of each ring, as prune locks them.
+fn sealed_slot_children(schema: &str, queue_slot: usize, claim_slot: usize) -> Vec<String> {
+    let mut children: Vec<String> = [
+        "ready_entries",
+        "ready_claim_attempt_batches",
+        "done_entries",
+        "ready_tombstones",
+        "ready_segments",
+        "receipt_completion_batches",
+        "receipt_completion_tombstones",
+        "queue_terminal_count_deltas",
+    ]
+    .iter()
+    .map(|table| format!("{schema}.{table}_{queue_slot}"))
+    .collect();
+    children.extend(
+        [
+            "lease_claims",
+            "lease_claim_batches",
+            "lease_claim_closures",
+            "lease_claim_closure_batches",
+        ]
+        .iter()
+        .map(|table| format!("{schema}.{table}_{claim_slot}")),
+    );
+    children
+}
+
+/// Runs `operation` while another session holds `ACCESS EXCLUSIVE` on
+/// `locked`, and records its name if it had not finished after two seconds.
+/// The lock is then released and the same operation is left to finish, so
+/// later steps have its result.
+async fn run_against_locked_children<T, F, Fut>(
+    pool: &sqlx::PgPool,
+    locked: &[String],
+    blocked: &mut Vec<&'static str>,
+    name: &'static str,
+    operation: F,
+) -> T
+where
+    F: FnOnce() -> Fut,
+    Fut: std::future::Future<Output = T>,
+{
+    if locked.is_empty() {
+        return operation().await;
+    }
+    let mut holder = pool.begin().await.expect("begin lock holder");
+    sqlx::query(&format!(
+        "LOCK TABLE {} IN {} MODE",
+        locked.join(", "),
+        std::env::var("AWA_AUDIT_LOCK_MODE").unwrap_or_else(|_| "ACCESS EXCLUSIVE".into())
+    ))
+    .execute(holder.as_mut())
+    .await
+    .expect("lock the sealed slot's children");
+    let operation = operation();
+    tokio::pin!(operation);
+    match tokio::time::timeout(Duration::from_secs(2), &mut operation).await {
+        Ok(value) => {
+            holder.rollback().await.expect("release the sealed slot");
+            value
+        }
+        Err(_) => {
+            blocked.push(name);
+            if std::env::var("AWA_AUDIT_SEALED_LOCKS").is_ok() {
+                let waiting: Vec<(String, String, String)> = sqlx::query_as(
+                    "SELECT l.mode, l.relation::regclass::text, left(regexp_replace(a.query, '\\s+', ' ', 'g'), 150)
+                     FROM pg_locks l JOIN pg_stat_activity a ON a.pid = l.pid
+                     WHERE NOT l.granted AND l.locktype = 'relation'",
+                )
+                .fetch_all(holder.as_mut())
+                .await
+                .expect("read waiting locks");
+                for (mode, relation, query) in waiting {
+                    eprintln!("WAIT {name}: {mode} on {relation} :: {query}");
+                }
+            }
+            holder.rollback().await.expect("release the sealed slot");
+            operation.await
+        }
+    }
+}
+
+/// Readers of the ring parents hold `ACCESS SHARE` on every child, sealed
+/// slots included. When several of them overlap there is never a moment with
+/// no holder, so a prune that only ever waits its base lock timeout cannot
+/// win (#721). Each reader here holds the lock for 40 ms and four of them are
+/// staggered 10 ms apart. They take the children in the order a claim and a
+/// heartbeat do, so one that arrives while prune waits holds nothing prune
+/// still needs.
+#[tokio::test(flavor = "multi_thread", worker_threads = 8)]
+async fn test_prune_wins_against_overlapping_short_readers() {
+    let (_db_guard, pool) = setup_pool(12).await;
+    let schema = "awa_qs_prune_overlapping_readers";
+    let store = create_store_with_config(
+        &pool,
+        QueueStorageConfig {
+            schema: schema.to_string(),
+            queue_slot_count: 4,
+            lease_slot_count: 2,
+            claim_slot_count: 4,
+            ..Default::default()
+        },
+    )
+    .await;
+    let queue = "qs_prune_overlapping_readers";
+
+    store.enqueue_batch(&pool, queue, 2, 4).await.unwrap();
+    let finished = store
+        .claim_runtime_batch(&pool, queue, 4, Duration::from_secs(30))
+        .await
+        .unwrap();
+    assert_eq!(finished.len(), 4);
+    store
+        .complete_runtime_batch(&pool, &finished)
+        .await
+        .unwrap();
+    assert!(matches!(
+        store.rotate(&pool).await.unwrap(),
+        RotateOutcome::Rotated { slot: 1, .. }
+    ));
+    assert!(matches!(
+        store.rotate_claims(&pool).await.unwrap(),
+        RotateOutcome::Rotated { slot: 1, .. }
+    ));
+
+    let stop = Arc::new(std::sync::atomic::AtomicBool::new(false));
+    let mut readers = Vec::new();
+    for reader in 0..4u64 {
+        let pool = pool.clone();
+        let stop = stop.clone();
+        let read = format!(
+            "SELECT (SELECT count(*) FROM {schema}.ready_segments)
+                  + (SELECT count(*) FROM {schema}.ready_entries)
+                  + (SELECT count(*) FROM {schema}.ready_tombstones)
+                  + (SELECT count(*) FROM {schema}.lease_claims)
+                  + (SELECT count(*) FROM {schema}.lease_claim_closures)
+             FROM pg_sleep(0.04)"
+        );
+        readers.push(tokio::spawn(async move {
+            tokio::time::sleep(Duration::from_millis(10 * reader)).await;
+            let mut slowest = Duration::ZERO;
+            while !stop.load(std::sync::atomic::Ordering::Relaxed) {
+                let started = Instant::now();
+                let _: i64 = sqlx::query_scalar(&read)
+                    .fetch_one(&pool)
+                    .await
+                    .expect("reader");
+                slowest = slowest.max(started.elapsed());
+            }
+            slowest
+        }));
+    }
+    tokio::time::sleep(Duration::from_millis(200)).await;
+
+    let mut queue_outcomes = Vec::new();
+    for _ in 0..5 {
+        let outcome = store.prune_oldest(&pool, Duration::ZERO).await.unwrap();
+        let pruned = matches!(outcome, PruneOutcome::Pruned { slot: 0, .. });
+        queue_outcomes.push(outcome);
+        if pruned {
+            break;
+        }
+    }
+    let mut claim_outcomes = Vec::new();
+    for _ in 0..5 {
+        let outcome = store.prune_oldest_claims(&pool).await.unwrap();
+        let pruned = matches!(outcome, PruneOutcome::Pruned { slot: 0, .. });
+        claim_outcomes.push(outcome);
+        if pruned {
+            break;
+        }
+    }
+
+    stop.store(true, std::sync::atomic::Ordering::Relaxed);
+    let mut slowest = Duration::ZERO;
+    for reader in readers {
+        slowest = slowest.max(reader.await.expect("reader task"));
+    }
+
+    assert!(
+        matches!(
+            queue_outcomes.last(),
+            Some(PruneOutcome::Pruned { slot: 0, .. })
+        ),
+        "queue prune never won: {queue_outcomes:?}"
+    );
+    assert!(
+        matches!(
+            claim_outcomes.last(),
+            Some(PruneOutcome::Pruned { slot: 0, .. })
+        ),
+        "claim prune never won: {claim_outcomes:?}"
+    );
+    // A reader waits at most one upgrade attempt, which is capped at 32
+    // times the 10 ms lock timeout, on top of its own 40 ms.
+    assert!(
+        slowest < Duration::from_millis(1_000),
+        "a reader stalled for {slowest:?} behind prune"
+    );
+}
+
+/// A claim that writes a lease directly touches no child of the queue slot
+/// it claims from, so freezing the slot's writers does not wait for it. If
+/// it read its ready row before the freeze and commits while prune waits to
+/// upgrade its lock, only the gates after the upgrade can see the lease.
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn test_queue_prune_sees_a_lease_committed_while_it_waits_to_truncate() {
+    let (_db_guard, pool) = setup_pool(6).await;
+    let schema = "awa_qs_prune_late_lease";
+    let store = create_store_with_config(
+        &pool,
+        QueueStorageConfig {
+            schema: schema.to_string(),
+            queue_slot_count: 4,
+            lease_slot_count: 2,
+            claim_slot_count: 4,
+            ..Default::default()
+        },
+    )
+    .await;
+    let store = Arc::new(store);
+    let queue = "qs_prune_late_lease";
+
+    store.enqueue_batch(&pool, queue, 2, 2).await.unwrap();
+    let finished = store
+        .claim_runtime_batch(&pool, queue, 2, Duration::from_secs(30))
+        .await
+        .unwrap();
+    store
+        .complete_runtime_batch(&pool, &finished)
+        .await
+        .unwrap();
+    assert!(matches!(
+        store.rotate(&pool).await.unwrap(),
+        RotateOutcome::Rotated { slot: 1, .. }
+    ));
+    let generation: i64 = sqlx::query_scalar(&format!(
+        "SELECT generation FROM {schema}.queue_ring_slots WHERE slot = 0"
+    ))
+    .fetch_one(&pool)
+    .await
+    .unwrap();
+
+    let mut claimer = pool.begin().await.unwrap();
+    let _: i64 = sqlx::query_scalar(&format!("SELECT count(*) FROM {schema}.ready_entries_0"))
+        .fetch_one(claimer.as_mut())
+        .await
+        .unwrap();
+    sqlx::query(&format!(
+        "INSERT INTO {schema}.leases
+             (lease_slot, lease_generation, ready_slot, ready_generation,
+              job_id, queue, priority, lane_seq)
+         VALUES (0, 0, 0, $1, 9000001, $2, 2, 9000001)"
+    ))
+    .bind(generation)
+    .bind(queue)
+    .execute(claimer.as_mut())
+    .await
+    .unwrap();
+
+    let prune = {
+        let store = store.clone();
+        let pool = pool.clone();
+        tokio::spawn(async move { store.prune_oldest(&pool, Duration::ZERO).await })
+    };
+    tokio::time::sleep(Duration::from_millis(60)).await;
+    claimer.commit().await.unwrap();
+
+    let outcome = prune.await.expect("prune task").unwrap();
+    assert!(
+        matches!(
+            outcome,
+            PruneOutcome::SkippedActive {
+                slot: 0,
+                reason: SkipReason::QueueActiveLeases,
+                ..
+            }
+        ),
+        "prune truncated or missed a slot with a live lease: {outcome:?}"
+    );
+    let ready_rows: i64 =
+        sqlx::query_scalar(&format!("SELECT count(*) FROM {schema}.ready_entries_0"))
+            .fetch_one(&pool)
+            .await
+            .unwrap();
+    assert_eq!(ready_rows, 2, "the skipped prune left the slot intact");
+}
+
+/// Heartbeat, retry, fail and snooze read `lease_claims` and then update it
+/// in one transaction, which locks every claim partition for reading and then
+/// for writing. Claim prune must neither fail nor cancel such a transaction
+/// when the write arrives while it is pruning a sealed slot.
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn test_claim_prune_does_not_deadlock_with_a_read_then_write_transaction() {
+    let (_db_guard, pool) = setup_pool(6).await;
+    let schema = "awa_qs_claim_prune_read_write";
+    let store = create_store_with_config(
+        &pool,
+        QueueStorageConfig {
+            schema: schema.to_string(),
+            queue_slot_count: 4,
+            lease_slot_count: 2,
+            claim_slot_count: 4,
+            ..Default::default()
+        },
+    )
+    .await;
+    let store = Arc::new(store);
+    let queue = "qs_claim_prune_read_write";
+
+    store.enqueue_batch(&pool, queue, 2, 2).await.unwrap();
+    let finished = store
+        .claim_runtime_batch(&pool, queue, 2, Duration::from_secs(30))
+        .await
+        .unwrap();
+    store
+        .complete_runtime_batch(&pool, &finished)
+        .await
+        .unwrap();
+    assert!(matches!(
+        store.rotate_claims(&pool).await.unwrap(),
+        RotateOutcome::Rotated { slot: 1, .. }
+    ));
+
+    let mut writer = pool.begin().await.unwrap();
+    sqlx::query(&format!(
+        "LOCK TABLE {schema}.lease_claims_0 IN ROW SHARE MODE"
+    ))
+    .execute(writer.as_mut())
+    .await
+    .unwrap();
+
+    let prune = {
+        let store = store.clone();
+        let pool = pool.clone();
+        tokio::spawn(async move { store.prune_oldest_claims(&pool).await })
+    };
+    tokio::time::sleep(Duration::from_millis(30)).await;
+    sqlx::query(&format!(
+        "LOCK TABLE {schema}.lease_claims_0 IN ROW EXCLUSIVE MODE"
+    ))
+    .execute(writer.as_mut())
+    .await
+    .expect("the writer must not be cancelled or time out behind prune");
+    tokio::time::sleep(Duration::from_millis(30)).await;
+    writer.commit().await.unwrap();
+
+    let outcome = prune
+        .await
+        .expect("prune task")
+        .expect("claim prune must not fail on a deadlock");
+    assert!(
+        matches!(outcome, PruneOutcome::Pruned { slot: 0, .. }),
+        "claim prune should win once the writer commits, got {outcome:?}"
+    );
+}
+
+/// Rescue takes its table locks on a claim slot's partitions and then locks
+/// the slot's `claim_ring_slots` row for its cursor. Claim prune must not
+/// hold that row while it waits for the partitions, or the two deadlock.
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn test_claim_prune_does_not_hold_the_slot_row_while_it_waits_for_partitions() {
+    let (_db_guard, pool) = setup_pool(6).await;
+    let schema = "awa_qs_claim_prune_slot_row";
+    let store = create_store_with_config(
+        &pool,
+        QueueStorageConfig {
+            schema: schema.to_string(),
+            queue_slot_count: 4,
+            lease_slot_count: 2,
+            claim_slot_count: 4,
+            ..Default::default()
+        },
+    )
+    .await;
+    let store = Arc::new(store);
+    let queue = "qs_claim_prune_slot_row";
+
+    store.enqueue_batch(&pool, queue, 2, 2).await.unwrap();
+    let finished = store
+        .claim_runtime_batch(&pool, queue, 2, Duration::from_secs(30))
+        .await
+        .unwrap();
+    store
+        .complete_runtime_batch(&pool, &finished)
+        .await
+        .unwrap();
+    assert!(matches!(
+        store.rotate_claims(&pool).await.unwrap(),
+        RotateOutcome::Rotated { slot: 1, .. }
+    ));
+
+    let mut rescue = pool.begin().await.unwrap();
+    sqlx::query(&format!(
+        "LOCK TABLE {schema}.lease_claims_0 IN ACCESS SHARE MODE"
+    ))
+    .execute(rescue.as_mut())
+    .await
+    .unwrap();
+
+    let prune = {
+        let store = store.clone();
+        let pool = pool.clone();
+        tokio::spawn(async move { store.prune_oldest_claims(&pool).await })
+    };
+    tokio::time::sleep(Duration::from_millis(30)).await;
+    let started = Instant::now();
+    sqlx::query(&format!(
+        "SELECT slot FROM {schema}.claim_ring_slots WHERE slot = 0 FOR UPDATE"
+    ))
+    .execute(rescue.as_mut())
+    .await
+    .expect("lock the slot row");
+    let row_wait = started.elapsed();
+    tokio::time::sleep(Duration::from_millis(30)).await;
+    rescue.commit().await.unwrap();
+
+    let outcome = prune.await.expect("prune task").expect("claim prune");
+    assert!(
+        row_wait < Duration::from_millis(200),
+        "the slot row lock waited {row_wait:?} behind claim prune"
+    );
+    assert!(
+        matches!(outcome, PruneOutcome::Pruned { slot: 0, .. }),
+        "claim prune should win once rescue commits, got {outcome:?}"
+    );
+}
+
+/// Prune takes `ACCESS EXCLUSIVE` on the children of a sealed slot. Live
+/// traffic works in the current slot, so it must not need a lock on any
+/// sealed child (#721).
+///
+/// Set `AWA_AUDIT_SEALED_LOCKS` to lock one child at a time and print which
+/// operations wait on it.
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+#[ignore = "fails until the hot paths address slot children directly (#721)"]
+async fn test_hot_paths_do_not_lock_sealed_slot_children() {
+    let (_db_guard, pool) = setup_pool(10).await;
+    let schema = "awa_qs_sealed_slot_locks";
+    let store = create_store_with_config(
+        &pool,
+        QueueStorageConfig {
+            schema: schema.to_string(),
+            queue_slot_count: 4,
+            lease_slot_count: 2,
+            claim_slot_count: 4,
+            ..Default::default()
+        },
+    )
+    .await;
+    let queue = "qs_sealed_slot_locks";
+
+    // Work that starts and finishes in slot 0 of the queue and claim rings.
+    store.enqueue_batch(&pool, queue, 2, 2).await.unwrap();
+    let finished = store
+        .claim_runtime_batch(&pool, queue, 2, Duration::from_secs(30))
+        .await
+        .unwrap();
+    assert_eq!(finished.len(), 2);
+    store
+        .complete_runtime_batch(&pool, &finished)
+        .await
+        .unwrap();
+    assert!(matches!(
+        store.rotate(&pool).await.unwrap(),
+        RotateOutcome::Rotated { slot: 1, .. }
+    ));
+    assert!(matches!(
+        store.rotate_claims(&pool).await.unwrap(),
+        RotateOutcome::Rotated { slot: 1, .. }
+    ));
+
+    let all_locked = sealed_slot_children(schema, 0, 0);
+    let audit = std::env::var("AWA_AUDIT_SEALED_LOCKS").is_ok();
+    // Plans are cached per session and change shape after five executions,
+    // so the traffic runs on one connection and is warmed up unlocked first.
+    let mut lock_sets: Vec<Vec<String>> = vec![Vec::new(); 8];
+    let audit_mode = std::env::var("AWA_AUDIT_SEALED_LOCKS").unwrap_or_default();
+    if audit && audit_mode != "all" && audit_mode != "1" {
+        // A comma-separated list of name fragments locks the matching
+        // children together, which shows the one an operation takes first.
+        lock_sets.push(
+            all_locked
+                .iter()
+                .filter(|child| audit_mode.split(',').any(|part| child.contains(part)))
+                .cloned()
+                .collect(),
+        );
+    } else if audit && audit_mode != "all" {
+        lock_sets.extend(all_locked.iter().map(|child| vec![child.clone()]));
+    } else {
+        lock_sets.push(all_locked.clone());
+    }
+    let holder_pool = pool.clone();
+    let pool = PgPoolOptions::new()
+        .max_connections(1)
+        .connect_with((*pool.connect_options()).clone())
+        .await
+        .expect("single-connection traffic pool");
+    let mut all_blocked = Vec::new();
+    for locked in &lock_sets {
+        let locked = locked.clone();
+        let mut blocked = Vec::new();
+
+        run_against_locked_children(&holder_pool, &locked, &mut blocked, "enqueue", || async {
+            store.enqueue_batch(&pool, queue, 2, 8).await.unwrap()
+        })
+        .await;
+
+        let deadline_claims = run_against_locked_children(
+            &holder_pool,
+            &locked,
+            &mut blocked,
+            "claim (deadline)",
+            || async {
+                store
+                    .claim_runtime_batch(&pool, queue, 4, Duration::from_secs(30))
+                    .await
+                    .unwrap()
+            },
+        )
+        .await;
+        assert_eq!(deadline_claims.len(), 4);
+        let pairs: Vec<(i64, i64)> = deadline_claims
+            .iter()
+            .map(|job| (job.job.id, job.job.run_lease))
+            .collect();
+        run_against_locked_children(&holder_pool, &locked, &mut blocked, "heartbeat", || async {
+            store.heartbeat_batch(&pool, &pairs).await.unwrap()
+        })
+        .await;
+        run_against_locked_children(
+            &holder_pool,
+            &locked,
+            &mut blocked,
+            "complete (deadline)",
+            || async {
+                store
+                    .complete_runtime_batch(&pool, &deadline_claims[..1])
+                    .await
+                    .unwrap()
+            },
+        )
+        .await;
+        run_against_locked_children(&holder_pool, &locked, &mut blocked, "retry", || async {
+            store
+                .retry_after(
+                    &pool,
+                    deadline_claims[1].job.id,
+                    deadline_claims[1].job.run_lease,
+                    Duration::from_secs(60),
+                    None,
+                )
+                .await
+                .unwrap()
+        })
+        .await;
+        run_against_locked_children(&holder_pool, &locked, &mut blocked, "fail", || async {
+            store
+                .fail_terminal(
+                    &pool,
+                    deadline_claims[2].job.id,
+                    deadline_claims[2].job.run_lease,
+                    "boom",
+                    None,
+                )
+                .await
+                .unwrap()
+        })
+        .await;
+        run_against_locked_children(&holder_pool, &locked, &mut blocked, "snooze", || async {
+            store
+                .snooze(
+                    &pool,
+                    deadline_claims[3].job.id,
+                    deadline_claims[3].job.run_lease,
+                    Duration::from_secs(60),
+                    None,
+                )
+                .await
+                .unwrap()
+        })
+        .await;
+
+        let compact_claims = run_against_locked_children(
+            &holder_pool,
+            &locked,
+            &mut blocked,
+            "claim (compact)",
+            || async {
+                store
+                    .claim_runtime_batch(&pool, queue, 4, Duration::ZERO)
+                    .await
+                    .unwrap()
+            },
+        )
+        .await;
+        assert_eq!(compact_claims.len(), 4);
+        let refs: Vec<_> = compact_claims
+            .iter()
+            .map(|job| {
+                awa::model::queue_storage::ClaimHeartbeatRef::from_claim(
+                    job.job.id,
+                    job.job.run_lease,
+                    &job.claim,
+                )
+                .expect("compact claim identity")
+            })
+            .collect();
+        run_against_locked_children(
+            &holder_pool,
+            &locked,
+            &mut blocked,
+            "heartbeat (compact)",
+            || async { store.heartbeat_claims_batch(&pool, &refs).await.unwrap() },
+        )
+        .await;
+        run_against_locked_children(
+            &holder_pool,
+            &locked,
+            &mut blocked,
+            "complete (compact)",
+            || async {
+                store
+                    .complete_runtime_batch(&pool, &compact_claims)
+                    .await
+                    .unwrap()
+            },
+        )
+        .await;
+
+        if audit {
+            eprintln!("AUDIT {} => {blocked:?}", locked.join(","));
+        }
+        all_blocked.extend(blocked);
+    }
+
+    assert!(
+        all_blocked.is_empty(),
+        "hot paths waited on a sealed slot's children: {all_blocked:?}"
+    );
+}
diff --git a/correctness/run-tlc-suite.sh b/correctness/run-tlc-suite.sh
index 106e78b49..475a9d29e 100755
--- a/correctness/run-tlc-suite.sh
+++ b/correctness/run-tlc-suite.sh
@@ -36,10 +36,14 @@ AwaStorageLockOrder.tla        ;; AwaStorageLockOrder.cfg                 ;; pas
 AwaStorageLockOrder.tla        ;; AwaStorageLockOrderDeadlockDemo.cfg     ;; NoDeadlock is violated ;; deadlock detector witness
 AwaStorageLockOrder.tla        ;; AwaStorageLockOrderOldStripedClaimDeadlock.cfg ;; NoDeadlock is violated ;; old striped-claim order witness
 AwaStorageLockOrder.tla        ;; AwaStorageLockOrderOldLaneProvisioningDeadlock.cfg ;; NoDeadlock is violated ;; old QueueStorage lane-provisioning order witness
+AwaStorageLockOrder.tla        ;; AwaStorageLockOrderOldClaimPruneSlotRowDeadlock.cfg ;; NoDeadlock is violated ;; old claim-prune slot-row order against rescue witness
 AwaDeadTupleContract.tla       ;; AwaDeadTupleContract.cfg                ;; pass ;; architectural reclaim contract
 AwaStorageTransition.tla       ;; -                                       ;; pass ;; transition control plane + #456 reschedule drain
 AwaQuiescedTransition.tla      ;; -                                       ;; pass ;; quiesced routing flip serializes snapshot evidence
 AwaQuiescedTransition.tla      ;; AwaQuiescedTransitionUnlocked.cfg       ;; QuiescedAtFlip is violated ;; unlocked liveness check race
+AwaQueuePruneTwoPhase.tla      ;; -                                       ;; pass ;; two-phase queue prune against a lease claim
+AwaQueuePruneTwoPhase.tla      ;; AwaQueuePruneTwoPhaseReceiptBarrier.cfg ;; pass ;; receipt claims are fenced by the writer freeze alone
+AwaQueuePruneTwoPhase.tla      ;; AwaQueuePruneTwoPhaseNoRecheck.cfg      ;; TruncatedSlotIsFinished is violated ;; lease claim committing during the upgrade witness
 AwaStorageTransition.tla       ;; AwaStorageTransitionCurrentGate.cfg     ;; MixedHasQueueExecutor is violated ;; pre-v014 capability-gate witness
 AwaStorageTransition.tla       ;; AwaStorageTransitionRescheduleStaysCanonical.cfg ;; MixedTransitionCanReduceCanonicalBacklog is violated ;; pre-#456 canonical re-schedule wedge witness
 AwaMigratedRescheduleUnique.tla ;; -                                      ;; pass ;; migrated re-schedule duplicate cancellation
diff --git a/correctness/storage/AwaQueuePruneTwoPhase.cfg b/correctness/storage/AwaQueuePruneTwoPhase.cfg
new file mode 100644
index 000000000..aa1be7662
--- /dev/null
+++ b/correctness/storage/AwaQueuePruneTwoPhase.cfg
@@ -0,0 +1,5 @@
+CONSTANT ReceiptClaims = FALSE
+CONSTANT RecheckAfterUpgrade = TRUE
+SPECIFICATION Spec
+INVARIANT TruncatedSlotIsFinished
+CHECK_DEADLOCK FALSE
diff --git a/correctness/storage/AwaQueuePruneTwoPhase.tla b/correctness/storage/AwaQueuePruneTwoPhase.tla
new file mode 100644
index 000000000..8d9ae7bca
--- /dev/null
+++ b/correctness/storage/AwaQueuePruneTwoPhase.tla
@@ -0,0 +1,94 @@
+---- MODULE AwaQueuePruneTwoPhase ----
+EXTENDS Naturals, TLC
+CONSTANTS ReceiptClaims, RecheckAfterUpgrade
+VARIABLES prune, claim, cursorPast
+vars == <<prune, claim, cursorPast>>
+
+\* One sealed queue slot holding one ready row, one claimer and the prune of
+\* that slot. Prune holds SHARE ROW EXCLUSIVE on the slot's children from
+\* "frozen" to "proved" and ACCESS EXCLUSIVE in "exclusive".
+\*
+\* claim: "idle"      the row is unclaimed
+\*        "read"      a claim selected the row and holds ACCESS SHARE on the
+\*                    ready child until it commits
+\*        "written"   the claim inserted its lease or receipt, uncommitted
+\*        "committed" the claim is visible to other transactions
+\*        "done"      the job completed and its claim is closed
+\*
+\* A receipt claim inserts an attempt batch into the slot, which needs
+\* ROW EXCLUSIVE on a child. A lease claim writes no child of the slot.
+PruneFreezesWriters == prune \in {"frozen", "leases", "pending", "proved", "exclusive"}
+ClaimReadsSlot == claim \in {"read", "written"}
+ClaimWritesSlot == ReceiptClaims /\ claim = "written"
+
+Init == /\ prune = "idle"
+        /\ claim = "idle"
+        /\ cursorPast = FALSE
+
+ClaimRead == /\ claim = "idle"
+             /\ prune \notin {"exclusive", "truncated"}
+             /\ claim' = "read"
+             /\ UNCHANGED <<prune, cursorPast>>
+ClaimWrite == /\ claim = "read"
+              /\ ReceiptClaims => ~PruneFreezesWriters
+              /\ claim' = "written"
+              /\ UNCHANGED <<prune, cursorPast>>
+ClaimCommit == /\ claim = "written"
+               /\ claim' = "committed"
+               /\ UNCHANGED <<prune, cursorPast>>
+\* The claim cursor moves only after the claim has committed.
+AdvanceCursor == /\ claim \in {"committed", "done"}
+                 /\ ~cursorPast
+                 /\ cursorPast' = TRUE
+                 /\ UNCHANGED <<prune, claim>>
+\* Completion writes the slot's terminal children.
+Complete == /\ claim = "committed"
+            /\ ~PruneFreezesWriters
+            /\ prune # "truncated"
+            /\ claim' = "done"
+            /\ UNCHANGED <<prune, cursorPast>>
+
+\* SHARE ROW EXCLUSIVE waits for transactions that wrote the slot.
+PruneFreeze == /\ prune = "idle"
+               /\ ~ClaimWritesSlot
+               /\ prune' = "frozen"
+               /\ UNCHANGED <<claim, cursorPast>>
+\* Only a committed lease claim has a visible lease row.
+LeaseVisible == ~ReceiptClaims /\ claim = "committed"
+LeaseGate == /\ prune = "frozen"
+             /\ prune' = IF LeaseVisible THEN "skipped" ELSE "leases"
+             /\ UNCHANGED <<claim, cursorPast>>
+PendingGate == /\ prune = "leases"
+               /\ prune' = IF cursorPast THEN "pending" ELSE "skipped"
+               /\ UNCHANGED <<claim, cursorPast>>
+\* The count proof: every committed receipt claim of the slot is closed.
+ClaimRefProof == /\ prune = "pending"
+                 /\ prune' = IF ReceiptClaims /\ claim = "committed"
+                             THEN "skipped" ELSE "proved"
+                 /\ UNCHANGED <<claim, cursorPast>>
+\* ACCESS EXCLUSIVE waits for every reader of the slot.
+Upgrade == /\ prune = "proved"
+           /\ ~ClaimReadsSlot
+           /\ prune' = "exclusive"
+           /\ UNCHANGED <<claim, cursorPast>>
+Truncate == /\ prune = "exclusive"
+            /\ prune' = IF RecheckAfterUpgrade /\ (LeaseVisible \/ ~cursorPast)
+                        THEN "skipped" ELSE "truncated"
+            /\ UNCHANGED <<claim, cursorPast>>
+\* A lock timeout rolls the prune back at any point before the truncate.
+Abort == /\ PruneFreezesWriters
+         /\ prune' = "skipped"
+         /\ UNCHANGED <<claim, cursorPast>>
+Retry == /\ prune = "skipped"
+         /\ prune' = "idle"
+         /\ UNCHANGED <<claim, cursorPast>>
+
+Next == \/ ClaimRead \/ ClaimWrite \/ ClaimCommit \/ AdvanceCursor \/ Complete
+        \/ PruneFreeze \/ LeaseGate \/ PendingGate \/ ClaimRefProof
+        \/ Upgrade \/ Truncate \/ Abort \/ Retry
+Spec == Init /\ [][Next]_vars
+
+\* Truncating the slot never drops an unclaimed row or the row of a claim
+\* that is still open.
+TruncatedSlotIsFinished == prune = "truncated" => claim = "done"
+====
diff --git a/correctness/storage/AwaQueuePruneTwoPhaseNoRecheck.cfg b/correctness/storage/AwaQueuePruneTwoPhaseNoRecheck.cfg
new file mode 100644
index 000000000..32edc7efc
--- /dev/null
+++ b/correctness/storage/AwaQueuePruneTwoPhaseNoRecheck.cfg
@@ -0,0 +1,5 @@
+CONSTANT ReceiptClaims = FALSE
+CONSTANT RecheckAfterUpgrade = FALSE
+SPECIFICATION Spec
+INVARIANT TruncatedSlotIsFinished
+CHECK_DEADLOCK FALSE
diff --git a/correctness/storage/AwaQueuePruneTwoPhaseReceiptBarrier.cfg b/correctness/storage/AwaQueuePruneTwoPhaseReceiptBarrier.cfg
new file mode 100644
index 000000000..84191e4cc
--- /dev/null
+++ b/correctness/storage/AwaQueuePruneTwoPhaseReceiptBarrier.cfg
@@ -0,0 +1,5 @@
+CONSTANT ReceiptClaims = TRUE
+CONSTANT RecheckAfterUpgrade = FALSE
+SPECIFICATION Spec
+INVARIANT TruncatedSlotIsFinished
+CHECK_DEADLOCK FALSE
diff --git a/correctness/storage/AwaStorageLockOrder.tla b/correctness/storage/AwaStorageLockOrder.tla
index 727d505fd..17213ad9e 100644
--- a/correctness/storage/AwaStorageLockOrder.tla
+++ b/correctness/storage/AwaStorageLockOrder.tla
@@ -351,13 +351,16 @@ CloseReceiptPlan(claimSlot) ==
 \*   UPDATE claim_ring_slots[slot] rescue cursor
 \* The leases anti-join takes AccessShare on the leases
 \* parent so rescue can race against prune_oldest_leases.
+\* PostgreSQL takes every relation lock of a statement when it plans it,
+\* before the executor takes any row lock. The cursor row's FOR UPDATE is
+\* written first in the statement but is therefore acquired last.
 RescueReceiptsPlan(claimSlot) ==
-    << Step(ClaimRingSlotResource(claimSlot), ModeExclusive),
-       Step(ClaimChildResource(claimSlot), ModeShared),
+    << Step(ClaimChildResource(claimSlot), ModeShared),
        Step(ClaimBatchChildResource(claimSlot), ModeShared),
        Step(LeasesParentResource, ModeShared),
        Step(ClosureChildResource(claimSlot), ModeShared),
-       Step(ClosureBatchChildResource(claimSlot), ModeShared) >>
+       Step(ClosureBatchChildResource(claimSlot), ModeShared),
+       Step(ClaimRingSlotResource(claimSlot), ModeExclusive) >>
 
 \* rescue_expired_receipt_deadlines_tx
 \*   SELECT ... FROM claim_ring_slots[slot] FOR UPDATE
@@ -370,11 +373,11 @@ RescueReceiptsPlan(claimSlot) ==
 \* Same resource order as stale receipt rescue; only the index order and
 \* advancement predicate differ.
 RescueReceiptDeadlinesPlan(claimSlot) ==
-    << Step(ClaimRingSlotResource(claimSlot), ModeExclusive),
-       Step(ClaimChildResource(claimSlot), ModeShared),
+    << Step(ClaimChildResource(claimSlot), ModeShared),
        Step(LeasesParentResource, ModeShared),
        Step(ClosureChildResource(claimSlot), ModeShared),
-       Step(ClosureBatchChildResource(claimSlot), ModeShared) >>
+       Step(ClosureBatchChildResource(claimSlot), ModeShared),
+       Step(ClaimRingSlotResource(claimSlot), ModeExclusive) >>
 
 \* ensure_running_leases_from_receipts_tx
 \*   transaction-scoped per-(job_id, run_lease) advisory lock first
@@ -492,11 +495,11 @@ PruneReadyPlan(slot) ==
        Step(ClosureProofResource, ModeShared),
        Step(ClosureBatchProofResource, ModeShared),
        Step(ReadyChildResource(slot), ModeShared),
+       Step(ReadySegmentChildResource(slot), ModeExclusive),
+       Step(DoneChildResource(slot), ModeExclusive),
        Step(ReadyChildResource(slot), ModeExclusive),
        Step(ReadyClaimAttemptBatchChildResource(slot), ModeExclusive),
-       Step(DoneChildResource(slot), ModeExclusive),
        Step(ReadyTombstoneChildResource(slot), ModeExclusive),
-       Step(ReadySegmentChildResource(slot), ModeExclusive),
        Step(ReceiptBatchChildResource(slot), ModeExclusive),
        Step(ReceiptTombstoneChildResource(slot), ModeExclusive),
        Step(TerminalDeltaChildResource(slot), ModeExclusive),
@@ -536,7 +539,11 @@ RotateClaimsPlan(nextSlot) ==
 \*     models the successful truncate path after the proof has passed
 \*   TRUNCATE claim_child[slot] + claim_batch_child[slot] +
 \*     closure_child[slot] + closure_batch_child[slot]
-PruneClaimsPlan(slot) ==
+\* The slot row is locked only once the partitions are held. Rescue takes
+\* the partitions and then the slot row, so the 0.6.16 order below, slot row
+\* first, deadlocked with every rescue of the slot for as long as prune
+\* waited for the partitions.
+OldPruneClaimsPlan(slot) ==
     << Step(ClaimRingStateResource, ModeExclusive),
        Step(ClaimRingSlotResource(slot), ModeExclusive),
        Step(ClaimChildResource(slot), ModeShared),
@@ -548,6 +555,18 @@ PruneClaimsPlan(slot) ==
        Step(ClosureChildResource(slot), ModeExclusive),
        Step(ClosureBatchChildResource(slot), ModeExclusive) >>
 
+PruneClaimsPlan(slot) ==
+    << Step(ClaimRingStateResource, ModeExclusive),
+       Step(ClaimChildResource(slot), ModeShared),
+       Step(ClaimBatchChildResource(slot), ModeShared),
+       Step(ClosureChildResource(slot), ModeShared),
+       Step(ClosureBatchChildResource(slot), ModeShared),
+       Step(ClaimChildResource(slot), ModeExclusive),
+       Step(ClaimBatchChildResource(slot), ModeExclusive),
+       Step(ClosureChildResource(slot), ModeExclusive),
+       Step(ClosureBatchChildResource(slot), ModeExclusive),
+       Step(ClaimRingSlotResource(slot), ModeExclusive) >>
+
 VARIABLES
     heldLocks,     \* [resource -> set of <<tx, mode>>]
     txState,       \* [tx -> "idle" | "running" | "committed"]
@@ -956,6 +975,33 @@ NextOldLaneProvisioningDeadlockDemo ==
 SpecOldLaneProvisioningDeadlockDemo ==
     ConfigOK /\ Init /\ [][NextOldLaneProvisioningDeadlockDemo]_vars
 
+\* ---- Historical claim-prune slot-row deadlock harness ----
+\*
+\* 0.6.16 claim prune locked the slot's `claim_ring_slots` row before the
+\* slot's partitions. Rescue holds the partitions and waits for that row
+\* while prune holds the row and waits for the partitions. TLC is expected
+\* to flag NoDeadlock on this spec.
+
+StartOldPruneClaims(t, slot) ==
+    /\ t \in TxIds
+    /\ txState[t] = "idle"
+    /\ slot \in ClaimSlots
+    /\ txState' = [txState EXCEPT ![t] = "running"]
+    /\ txPlan' = [txPlan EXCEPT ![t] = OldPruneClaimsPlan(slot)]
+    /\ txNextStep' = [txNextStep EXCEPT ![t] = 1]
+    /\ UNCHANGED heldLocks
+
+NextOldClaimPruneSlotRowDeadlockDemo ==
+    \/ \E t \in TxIds, cs \in ClaimSlots : StartRescueReceipts(t, cs)
+    \/ \E t \in TxIds, cs \in ClaimSlots : StartOldPruneClaims(t, cs)
+    \/ \E t \in TxIds : AcquireNext(t)
+    \/ \E t \in TxIds : Commit(t)
+    \/ \E t \in TxIds : Recycle(t)
+    \/ Stutter
+
+SpecOldClaimPruneSlotRowDeadlockDemo ==
+    ConfigOK /\ Init /\ [][NextOldClaimPruneSlotRowDeadlockDemo]_vars
+
 \* ---- Invariants ----
 
 TypeOK ==
diff --git a/correctness/storage/AwaStorageLockOrderOldClaimPruneSlotRowDeadlock.cfg b/correctness/storage/AwaStorageLockOrderOldClaimPruneSlotRowDeadlock.cfg
new file mode 100644
index 000000000..00bd3063d
--- /dev/null
+++ b/correctness/storage/AwaStorageLockOrderOldClaimPruneSlotRowDeadlock.cfg
@@ -0,0 +1,20 @@
+\* Historical claim-prune harness. 0.6.16 claim prune locked the slot's
+\* claim_ring_slots row before the slot's partitions, while rescue takes the
+\* partitions first. TLC is expected to flag NoDeadlock.
+CONSTANTS
+  TxIds = {"t1", "t2"}
+  Queues = {"q1#0", "q1#1"}
+  StripeA = "q1#0"
+  StripeB = "q1#1"
+  Priorities = {1}
+  LeaseSlots = {1}
+  ReadySlots = {1}
+  ClaimSlots = {1}
+
+SPECIFICATION SpecOldClaimPruneSlotRowDeadlockDemo
+
+INVARIANTS
+  TypeOK
+  LockCompatibility
+  HeldOnlyByRunningTxs
+  NoDeadlock
diff --git a/correctness/storage/MAPPING.md b/correctness/storage/MAPPING.md
index 93ad5ea8d..6818e1e78 100644
--- a/correctness/storage/MAPPING.md
+++ b/correctness/storage/MAPPING.md
@@ -65,7 +65,7 @@ The TLA+ lifecycle model does not represent the completed-history rollup cache,
 | `PurgeDlq(j)` | `purge_dlq_job` / `purge_dlq` in `awa-model/src/dlq.rs:382, 423` | `DELETE FROM dlq_entries WHERE ...` |
 | `RotateReadySegments` | maintenance `rotate_ready` (`awa-worker/src/maintenance.rs`) | `UPDATE queue_ring_state SET current_slot = next` + partition attach/detach |
 | `RotateLeaseSegments` | `QueueStorage::rotate_leases` | `UPDATE lease_ring_state` with child-partition busy check |
-| `PruneReadySegment(seg)` | maintenance `QueueStorage::prune_oldest` for the ready/terminal queue family | `FOR UPDATE` on `queue_ring_state` and `queue_ring_slots[slot]`, prove the sealed-slot skip gates first (active leases, receipt claims that still lack durable closure evidence, and retained ready rows still at or ahead of their lane's claim cursor), then `LOCK TABLE ... ACCESS EXCLUSIVE` with a short transaction-local `lock_timeout`, recheck the skip gates under the exclusive locks, and `TRUNCATE`; this guard prevents queue prune from removing terminal history before claim closure evidence exists or removing ready/tombstone cursor evidence before the lane cursor has passed it. Ready-backed terminal rows, compact completion batches, compact tombstones, ready tombstones, ready-segment metadata, and pending `done_entries` terminal-count deltas are reclaimed with their retained ready bodies; Rust updates `{schema}.queue_terminal_rollups` after successful terminal prune accounting |
+| `PruneReadySegment(seg)` | maintenance `QueueStorage::prune_oldest` for the ready/terminal queue family | `FOR UPDATE` on `queue_ring_state` and `queue_ring_slots[slot]`, prove the sealed-slot skip gates first (active leases, receipt claims that still lack durable closure evidence, and retained ready rows still at or ahead of their lane's claim cursor), then `LOCK TABLE ... SHARE ROW EXCLUSIVE` with a short transaction-local `lock_timeout`, recheck the skip gates under that lock, upgrade to `ACCESS EXCLUSIVE` with a bounded escalating wait, recheck active leases and pending ready rows (`AwaQueuePruneTwoPhase.tla`), and `TRUNCATE`; this guard prevents queue prune from removing terminal history before claim closure evidence exists or removing ready/tombstone cursor evidence before the lane cursor has passed it. Ready-backed terminal rows, compact completion batches, compact tombstones, ready tombstones, ready-segment metadata, and pending `done_entries` terminal-count deltas are reclaimed with their retained ready bodies; Rust updates `{schema}.queue_terminal_rollups` after successful terminal prune accounting |
 | `PruneLeaseSegment` | `QueueStorage::prune_oldest_leases` | `FOR UPDATE` on `lease_ring_state` and `lease_ring_slots[slot]`, then `LOCK TABLE ... ACCESS EXCLUSIVE` with a short transaction-local `lock_timeout`, recheck active rows, then `TRUNCATE` the selected `leases_N` child |
 | `RotateClaimSegments` | maintenance `QueueStorage::rotate_claims`, wired via `Maintenance::rotate_queue_storage_claims` at the `claim_rotate_interval` tick | `FOR UPDATE` on `claim_ring_state`, busy-check the three child partitions, then `UPDATE claim_ring_state SET current_slot = next, generation = next_gen` with compare-and-swap on `(current_slot, generation)` |
 | `PruneClaimSegment(seg)` | `QueueStorage::prune_oldest_claims` | `FOR UPDATE` on `claim_ring_state`, `FOR UPDATE` on `claim_ring_slots[slot]`, proof that every row-local claim or compact claim-batch item in the sealed slot has durable closure evidence, then `LOCK TABLE` `lease_claims_N`, `lease_claim_batches_N`, `lease_claim_closures_N`, and `lease_claim_closure_batches_N` `IN ACCESS EXCLUSIVE MODE` with a short transaction-local `lock_timeout`, recheck not-current, recheck the open-claim proof, reset that slot's stale-rescue cursor, then `TRUNCATE` the four children. The proof compares exact child-row counts (`lease_claims` rows plus `lease_claim_batches.claimed_count` totals versus explicit closures plus compact `closed_count` totals). If counts do not prove closure, prune returns `SkippedActive` and tries again later instead of running an unbounded per-claim anti-join over retained compact batches. |
@@ -213,12 +213,14 @@ The spec's PruneReadySegment transition also captures the analogous concern in `
 
 1. `FOR UPDATE` on `queue_ring_state` to serialise against concurrent rotates
 2. `FOR UPDATE` on the target `queue_ring_slots` row
-3. `LOCK TABLE ... IN ACCESS EXCLUSIVE MODE` on the ready and done partition children under a short transaction-local `lock_timeout` — this conflicts with the AccessShare lock that the claim CTE takes when reading `{schema}.ready_entries_%s`, but prune bails if it cannot own the segment promptly
-4. Only AFTER the lock is held does the count-active-leases check run inside the same transaction — so any lease inserted by a concurrent claim will be visible to the check
+3. `LOCK TABLE ... IN SHARE ROW EXCLUSIVE MODE` on the slot's children under a short transaction-local `lock_timeout`. It waits for every transaction that has written to the slot, blocks new writers, and conflicts with no reader
+4. The skip gates (active leases, pending ready rows, the claim-reference count proof), the failed-row carry and the rollup scan run under that lock. A receipt claim inserts its attempt batch into the slot, so none can begin after step 3 and the count proof stays true
+5. `LOCK TABLE ... IN ACCESS EXCLUSIVE MODE` on the same children, retried from a savepoint with a doubling `lock_timeout` up to a fixed ceiling
+6. The active-lease and pending-ready gates run again. A claim that writes a lease directly touches no child of the slot; if it read its ready row before step 3 it can commit until step 5 is granted, and only this recheck sees it
 
-All automatic prune paths use a bounded lock timeout for child-table `ACCESS EXCLUSIVE` locks so they abort gracefully under sustained contention while still being able to break starvation under continuous short readers.
+`AwaQueuePruneTwoPhase.tla` models steps 3 to 6 against one claimer. `AwaQueuePruneTwoPhaseNoRecheck.cfg` drops step 6 and finds the lease claim that commits during the upgrade; `AwaQueuePruneTwoPhaseReceiptBarrier.cfg` shows receipt claims are safe without it.
 
-So the "check-then-act" framing is inaccurate: the Rust code is "lock-then-check-then-act", with the lock being the load-bearing part.
+So the "check-then-act" framing is inaccurate: the Rust code is "lock-then-check-then-act", with the writer lock and the recheck after the upgrade being the load-bearing parts.
 
 ### Role of the race spec going forward
 
diff --git a/correctness/storage/README.md b/correctness/storage/README.md
index 4ed7be41c..896c727f4 100644
--- a/correctness/storage/README.md
+++ b/correctness/storage/README.md
@@ -151,6 +151,7 @@ Configs:
 - [`AwaStorageLockOrderDeadlockDemo.cfg`](./AwaStorageLockOrderDeadlockDemo.cfg): sanity harness using a deliberately cycle-creating pair of plans — **NoDeadlock tripped in 5 steps** (confirms the checker works).
 - [`AwaStorageLockOrderOldStripedClaimDeadlock.cfg`](./AwaStorageLockOrderOldStripedClaimDeadlock.cfg): the historical claim shape that walked several physical stripes against the enqueue order — **NoDeadlock tripped**.
 - [`AwaStorageLockOrderOldLaneProvisioningDeadlock.cfg`](./AwaStorageLockOrderOldLaneProvisioningDeadlock.cfg): the historical `QueueStorage` order for a new lane, provisioning lock before the `queue_lanes` row, against a SQL producer of the same lane — **NoDeadlock tripped in 5 steps** (#692).
+- [`AwaStorageLockOrderOldClaimPruneSlotRowDeadlock.cfg`](./AwaStorageLockOrderOldClaimPruneSlotRowDeadlock.cfg): the 0.6.16 claim prune order, the slot's `claim_ring_slots` row before the slot's partitions, against stale receipt rescue, which takes the partitions and then the row — **NoDeadlock tripped** (#721).
 
 Run:
 
@@ -161,7 +162,9 @@ Run:
 ./correctness/run-tlc.sh storage/AwaStorageLockOrder.tla storage/AwaStorageLockOrderOldLaneProvisioningDeadlock.cfg
 ```
 
-Coverage note: the plans model the lock steps that actually appear in the Rust SQL (`FOR UPDATE` / `FOR SHARE` / `LOCK TABLE ACCESS EXCLUSIVE` / the implicit AccessShare of SELECT on partition children). They do NOT model implicit table-level locks beyond what is named, or Postgres's lock-timeout / deadlock-detector abort choice. The spec treats a waits-for cycle as a safety violation, which is conservative — Postgres would abort one transaction and let the other proceed. For our purposes "this sequence of lock requests could produce a cycle" is the thing we want to catch, regardless of how the runtime resolves it.
+Plan order follows PostgreSQL, not the SQL text: a statement takes every relation lock when it is planned and its row locks afterwards, so a `FOR UPDATE` written in a leading CTE is still acquired after the locks on every table the statement reads. The rescue plans listed the cursor row first until #721, which hid their cycle with claim prune.
+
+Coverage note: the plans model the lock steps that actually appear in the Rust SQL (`FOR UPDATE` / `FOR SHARE` / `LOCK TABLE ACCESS EXCLUSIVE` / the implicit AccessShare of SELECT on partition children). They do NOT model implicit table-level locks beyond what is named (a claim, completion or heartbeat that reads a partitioned parent through a cached generic plan locks every slot's partition, and the plans list only the slot the statement targets), the difference between `ROW EXCLUSIVE`, `SHARE ROW EXCLUSIVE` and `ACCESS SHARE`, or Postgres's lock-timeout / deadlock-detector abort choice. The spec treats a waits-for cycle as a safety violation, which is conservative — Postgres would abort one transaction and let the other proceed. For our purposes "this sequence of lock requests could produce a cycle" is the thing we want to catch, regardless of how the runtime resolves it.
 
 ## Race-exposure companion spec
 
@@ -278,3 +281,5 @@ See [`../README.md`](../README.md) for the full Known Divergences list. Specific
 - **Unique-claim conflicts on `retry_from_dlq`.** The Rust contract says retry-from-dlq returns `UniqueConflict` and leaves the DLQ row intact when a replacement owns the unique slot. The model doesn't have unique keys, so TLC simply explores `RetryFromDlq(j)` whenever the precondition holds. Preserving-the-DLQ-row-on-conflict is enforced in SQL (`awa-model/src/queue_storage.rs::sync_unique_claim`).
 
 `AwaAsyncClaimCursor.tla` refines async claim acknowledgment, WAL durability, standby promotion and post-commit sequence replay. Its safe config checks no skipped jobs; `AwaAsyncClaimCursorUnguarded.cfg` preserves the lost-claim cursor-skip counterexample.
+
+`AwaQueuePruneTwoPhase.tla` models queue prune's two lock phases against one claimer: `SHARE ROW EXCLUSIVE` for the skip gates, then `ACCESS EXCLUSIVE` and a recheck before the truncate. Its default config checks a claim that writes a lease directly; `AwaQueuePruneTwoPhaseReceiptBarrier.cfg` checks receipt claims without the recheck; `AwaQueuePruneTwoPhaseNoRecheck.cfg` preserves the counterexample of a lease claim committing during the upgrade.
diff --git a/docs/configuration.md b/docs/configuration.md
index dba17cf96..ac30a1bec 100644
--- a/docs/configuration.md
+++ b/docs/configuration.md
@@ -626,7 +626,7 @@ await client.start(
 | `queue_rotate_interval` | `1000ms` | How often ready/terminal segments rotate |
 | `lease_rotate_interval` | `1000ms` | How often lease segments rotate |
 | `claim_rotate_interval` | matches `queue_rotate_interval` | How often the ADR-023 claim-ring rotates. Set with `ClientBuilder::claim_rotate_interval` (Rust) or `queue_storage_claim_rotate_interval_ms` (Python). Test harnesses sometimes set this to a long interval to pin claim-ring layout for deterministic count assertions. |
-| `prune_lock_timeout` | `10ms` | The PostgreSQL `lock_timeout` the maintenance leader applies when it takes the exclusive locks to reclaim a queue, lease or claim ring partition (and when it folds terminal-count deltas). The timeout applies separately to each lock acquisition, not to the whole prune. Queue prune locks up to eight partitions, so its partition-lock waits alone can approach eight times this value. A lock timeout rolls the prune back, is counted as `blocked`, and is retried on a later maintenance tick. Set with `ClientBuilder::prune_lock_timeout` (Rust) or `queue_storage_prune_lock_timeout_ms` (Python). Must be greater than zero and at most `2147483647` ms; keep it well below `deadlock_timeout` and `statement_timeout`. The short default keeps prune's pending `ACCESS EXCLUSIVE` request from stalling claims and enqueues queued behind it. See below for when to raise it. |
+| `prune_lock_timeout` | `10ms` | The PostgreSQL `lock_timeout` the maintenance leader applies when it takes the exclusive locks to reclaim a queue, lease or claim ring partition (and when it folds terminal-count deltas). The timeout applies separately to each lock acquisition, not to the whole prune. Queue prune locks up to eight partitions, so its partition-lock waits alone can approach eight times this value. Queue prune first takes `SHARE ROW EXCLUSIVE`, which blocks writes to the sealed slot but no reads, runs its checks under it, and requests `ACCESS EXCLUSIVE` only for the `TRUNCATE`; claim prune requests `ACCESS EXCLUSIVE` before its check. For both, that request waits this timeout, then retries with twice the wait each time up to 320 ms (or this timeout when it is longer), releasing the readers queued behind it between attempts. A lock timeout on the last attempt, or a deadlock resolved by cancelling the prune, rolls the prune back, is counted as `blocked`, and is retried on a later maintenance tick. Set with `ClientBuilder::prune_lock_timeout` (Rust) or `queue_storage_prune_lock_timeout_ms` (Python). Must be greater than zero and at most `2147483647` ms; keep it, and the 320 ms retry ceiling, well below `deadlock_timeout` and `statement_timeout`. The short default keeps prune's pending `ACCESS EXCLUSIVE` request from stalling claims and enqueues queued behind it. See below for when to raise it. |
 
 The benchmark harness in [postgresql-job-queue-benchmarking](https://github.com/hardbyte/postgresql-job-queue-benchmarking) reads `QUEUE_SLOT_COUNT`, `LEASE_SLOT_COUNT`, `CLAIM_SLOT_COUNT`, `QUEUE_STRIPE_COUNT`, and `LEASE_CLAIM_RECEIPTS` from the environment. Those env vars are benchmark configuration, not general worker-runtime configuration.
 
@@ -636,7 +636,7 @@ Use the defaults unless you have a reason not to:
 - Increase `lease_slot_count` if lease churn is high enough that dead tuples in the lease ring stop collapsing promptly.
 - Increase `claim_slot_count` if the rotation cadence (`claim_rotate_interval`) plus the slot count combine to a partition retention window shorter than your longest in-flight zero-deadline short job; running out of empty slots forces `rotate_claims` to return `SkippedBusy` and the receipt-plane churn falls back onto a smaller working set of partitions.
 - Increase `queue_stripe_count` only for measured workloads where many small replicas contend on the same logical queue. This is queue-storage configuration, not a per-queue `QueueConfig` field: it gives each logical queue the same number of physical stripes. Striping spreads a hot logical queue over `queue#N` physical coordination paths, but it weakens perfect global ordering and can regress calmer shapes if overused. For single hot-queue workloads, benchmark `2` stripes first; higher values should be treated as workload-specific tuning rather than a general default.
-- Increase `prune_lock_timeout` when `increase(awa_maintenance_prune_attempts_total{awa_ring_outcome="blocked"}[15m])` stays above zero and the rings stop recycling, typically because frequent short reads of the ring partitions (admin or dashboard queries over the queue-storage tables) hold a conflicting lock at the instant prune tries. A stalled ring grows its current slot, which makes those queries slower and the contention worse. Raise it in small steps (for example `50ms`, then `100ms`, then `250ms`) and keep it well below PostgreSQL's `deadlock_timeout` (default `1s`) and below any `statement_timeout` on the pool's connections. A single lock wait that reaches `deadlock_timeout` makes PostgreSQL run its deadlock detector, and if claims or enqueues are part of a lock cycle with it, the detector can cancel them rather than the prune; a short `lock_timeout` abandons the prune first, so the prune is always the one that gives way. A `statement_timeout` at or below the lock timeout cancels the prune statement before the lock timeout fires, so it fails with an error instead of reporting `blocked`. Startup logs a warning when the configured timeout is at or above either setting. While prune waits, new accesses to the same partition also queue behind its lock request, so a longer timeout trades claim latency for reclamation. Shortening the readers is the more durable fix when they run longer than a few hundred milliseconds. Watch `awa_maintenance_prune_attempts_total` by `awa_ring` and `awa_ring_outcome`: `pruned` should advance steadily, `blocked` means the lock timeout expired, `skipped_active` (with `awa_ring_reason`) means live rows kept the slot, `backed_off` means prune was skipped for up to 32 ticks after consecutive `skipped_active` or `blocked` results, so a persistently blocked ring is retried less often over time, and `error` is a database failure.
+- Increase `prune_lock_timeout` when `increase(awa_maintenance_prune_attempts_total{awa_ring_outcome="blocked"}[15m])` stays above zero and the rings stop recycling, typically because frequent short reads of the ring partitions (admin or dashboard queries over the queue-storage tables) hold a conflicting lock at the instant prune tries. A stalled ring grows its current slot, which makes those queries slower and the contention worse. Raise it in small steps (for example `50ms`, then `100ms`, then `250ms`) and keep it well below PostgreSQL's `deadlock_timeout` (default `1s`) and below any `statement_timeout` on the pool's connections. A single lock wait that reaches `deadlock_timeout` makes PostgreSQL run its deadlock detector, and if claims or enqueues are part of a lock cycle with it, the detector can cancel them rather than the prune; a short `lock_timeout` abandons the prune first, so the prune is always the one that gives way. A `statement_timeout` at or below the lock timeout cancels the prune statement before the lock timeout fires, so it fails with an error instead of reporting `blocked`. Startup logs a warning when the longest prune lock wait (the configured timeout or the 320 ms retry ceiling, whichever is longer) is at or above `deadlock_timeout`, or when eight times that wait, the most one eight-partition lock statement can wait, is at or above `statement_timeout`. While prune waits, new accesses to the same partition also queue behind its lock request, so a longer timeout trades claim latency for reclamation. Shortening the readers is the more durable fix when they run longer than a few hundred milliseconds. Watch `awa_maintenance_prune_attempts_total` by `awa_ring` and `awa_ring_outcome`: `pruned` should advance steadily, `blocked` means the lock timeout expired, `skipped_active` (with `awa_ring_reason`) means live rows kept the slot, `backed_off` means prune was skipped for up to 32 ticks after consecutive `skipped_active` or `blocked` results, so a persistently blocked ring is retried less often over time, and `error` is a database failure.
 - Consider increasing rotation intervals before raising `prune_lock_timeout`: longer intervals can reduce how often prune requests exclusive locks, while a longer timeout lets each request wait longer. Confirm the effect with workload measurements.
 - Decrease rotation intervals to tighten dead-tuple bounds at the cost of more frequent rotate/prune work.