diff --git a/crates/zakura-network/src/zakura/block_sync/admission.rs b/crates/zakura-network/src/zakura/block_sync/admission.rs
index c31e423fc6..811beb6f9f 100644
--- a/crates/zakura-network/src/zakura/block_sync/admission.rs
+++ b/crates/zakura-network/src/zakura/block_sync/admission.rs
@@ -7,16 +7,10 @@ use super::{
     state::next_height,
 };
 
-/// Delivery rate assumed when sizing an above-floor deadline for a peer whose
-/// measured BtlBw is still near zero, so the patience window is bounded rather than
-/// unbounded. A worst-case `MAX_BLOCK_BYTES` body at this rate transfers in ~8 s, so
-/// with the `request_timeout` base the above-floor deadline tops out near 16 s — the
-/// "a block every ~16 s is fine" tolerance the directive sets for speculative work.
-const ABOVE_FLOOR_DEADLINE_MIN_BYTES_PER_SEC: u64 = 256 * 1024;
-/// Delivery rate assumed for floor rescue before a peer has a fresh byte-rate
-/// sample. This keeps the rescue leash short while allowing a full 2 MB body roughly
-/// two seconds of transfer time.
-const FLOOR_DEADLINE_MIN_BYTES_PER_SEC: u64 = 1024 * 1024;
+/// Minimum rate used for deadline estimates. One maximum-size body adds about
+/// eight seconds; earlier unreceived bodies add their bounded transfer estimates.
+/// A session with no accepted progress still reaches its separate liveness limit.
+const DEADLINE_MIN_BYTES_PER_SEC: u64 = 256 * 1024;
 
 /// Estimated resident-memory multiple of a *decoded* block body's serialized size.
 ///
@@ -114,43 +108,36 @@ pub(super) fn request_priority(
     }
 }
 
-/// The per-request network deadline (the one sanctioned timer), set by priority:
+/// A bounded request deadline including estimated transfer time.
 ///
-/// - **Floor**: a short rescue leash plus the expected transfer time. On expiry the
-///   lowest missing height is rescued to a faster carrier (returned to the queue + the
-///   peer retry-avoided), so the contiguous floor never waits on a slow peer — and the
-///   peer is *not* disconnected.
-/// - **Above-floor**: the base `request_timeout` plus the size-expected transfer time
-///   (`estimated_bytes / BtlBw`), so a legitimately slow large-body fetch runs to
-///   completion. These deadlines never gate the floor, so they can afford to be
-///   patient; `btlbw_bytes_per_sec` is the peer's measured rate (`None` cold-start),
-///   floored at [`ABOVE_FLOOR_DEADLINE_MIN_BYTES_PER_SEC`].
+/// A floor request uses the short rescue deadline only with a fresh delivery-rate
+/// measurement. Without one, it gets the normal deadline: prematurely expiring a
+/// cold peer's only probe prevents its first body from establishing progress.
+/// Block-count measurements qualify for rescue but use the byte-rate fallback
+/// when estimating transfer time.
+/// Above-floor requests always use the normal deadline. Transfer bytes include
+/// earlier unreceived responses on the ordered data stream. Expiry returns work
+/// for retry; the separate block-progress deadline still bounds a silent session.
 pub(super) fn request_deadline(
     priority: RequestPriority,
     queued_at: Instant,
     request_timeout: Duration,
     floor_rescue_timeout: Duration,
-    estimated_bytes: u64,
+    expected_transfer_bytes: u64,
     btlbw_bytes_per_sec: Option<u64>,
+    has_delivery_measurement: bool,
 ) -> Instant {
-    match priority {
-        RequestPriority::Floor => {
-            let rate = btlbw_bytes_per_sec
-                .unwrap_or(0)
-                .max(FLOOR_DEADLINE_MIN_BYTES_PER_SEC);
-            let transfer = Duration::from_secs_f64(estimated_bytes as f64 / rate as f64);
-            queued_at + floor_rescue_timeout + transfer
-        }
-        RequestPriority::AboveFloor => {
-            let rate = btlbw_bytes_per_sec
-                .unwrap_or(0)
-                .max(ABOVE_FLOOR_DEADLINE_MIN_BYTES_PER_SEC);
-            // One body per request, so `estimated_bytes / rate` is at most
-            // `MAX_BLOCK_BYTES / rate` (~8 s): finite and non-negative.
-            let transfer = Duration::from_secs_f64(estimated_bytes as f64 / rate as f64);
-            queued_at + request_timeout + transfer
-        }
-    }
+    let base = if priority == RequestPriority::Floor && has_delivery_measurement {
+        floor_rescue_timeout
+    } else {
+        request_timeout
+    };
+    let rate = btlbw_bytes_per_sec
+        .unwrap_or(0)
+        .max(DEADLINE_MIN_BYTES_PER_SEC);
+    // Peer and node admission bound these values well within f64's exact integer range.
+    let transfer = Duration::from_secs_f64(expected_transfer_bytes as f64 / rate as f64);
+    queued_at + base + transfer
 }
 
 /// Heights within one worst-case checkpoint range above the verified tip bypass
@@ -385,7 +372,8 @@ mod tests {
             TIMEOUT,
             RESCUE,
             2_000_000,
-            None,
+            Some(1024 * 1024),
+            true,
         );
         assert_eq!(
             deadline,
@@ -393,6 +381,52 @@ mod tests {
         );
     }
 
+    #[test]
+    fn unmeasured_floor_probe_gets_the_normal_bounded_deadline() {
+        let now = Instant::now();
+        for size in [1, 2_000_000] {
+            let deadline = request_deadline(
+                RequestPriority::Floor,
+                now,
+                TIMEOUT,
+                RESCUE,
+                size,
+                None,
+                false,
+            );
+            assert_eq!(
+                deadline,
+                request_deadline(
+                    RequestPriority::AboveFloor,
+                    now,
+                    TIMEOUT,
+                    RESCUE,
+                    size,
+                    None,
+                    false,
+                )
+            );
+            assert!(deadline > now + TIMEOUT);
+            assert!(deadline < now + Duration::from_secs(16));
+        }
+    }
+
+    #[test]
+    fn floor_rescue_allows_the_measured_large_body_transfer_time() {
+        let now = Instant::now();
+        let deadline = request_deadline(
+            RequestPriority::Floor,
+            now,
+            TIMEOUT,
+            RESCUE,
+            2 * 1024 * 1024,
+            Some(512 * 1024),
+            true,
+        );
+        assert_eq!(deadline, now + RESCUE + Duration::from_secs(4));
+        assert!(deadline < now + TIMEOUT);
+    }
+
     #[test]
     fn above_floor_deadline_grows_with_body_size() {
         let now = Instant::now();
@@ -405,6 +439,7 @@ mod tests {
             RESCUE,
             256 * 1024,
             None,
+            false,
         );
         let large = request_deadline(
             RequestPriority::AboveFloor,
@@ -413,6 +448,7 @@ mod tests {
             RESCUE,
             2 * 1024 * 1024,
             None,
+            false,
         );
         assert_eq!(small, now + TIMEOUT + Duration::from_secs(1));
         assert_eq!(large, now + TIMEOUT + Duration::from_secs(8));
@@ -431,6 +467,7 @@ mod tests {
             RESCUE,
             2 * 1024 * 1024,
             Some(64 * 1024 * 1024),
+            true,
         );
         assert!(fast > now + TIMEOUT);
         assert!(fast < now + TIMEOUT + Duration::from_millis(100));
diff --git a/crates/zakura-network/src/zakura/block_sync/bbr.rs b/crates/zakura-network/src/zakura/block_sync/bbr.rs
index 7a3bdbfa46..e8f71482a1 100644
--- a/crates/zakura-network/src/zakura/block_sync/bbr.rs
+++ b/crates/zakura-network/src/zakura/block_sync/bbr.rs
@@ -1030,6 +1030,8 @@ mod bbr_tests {
         let now = Instant::now();
         for _ in 0..n {
             window.outstanding.push(OutstandingBlockRange {
+                write_status: super::super::work_queue::RequestWriteStatus::written_for_tests(),
+                charged_for_liveness: true,
                 request: BlockRangeRequest {
                     owner: super::super::test_work_owner(),
                     start_height: block::Height(0),
@@ -1135,6 +1137,8 @@ mod bbr_tests {
             // A `u32` index; the test count is tiny so the cast is safe.
             let height = block::Height(1 + i as u32);
             window.outstanding.push(OutstandingBlockRange {
+                write_status: super::super::work_queue::RequestWriteStatus::written_for_tests(),
+                charged_for_liveness: true,
                 request: BlockRangeRequest {
                     owner: super::super::test_work_owner(),
                     start_height: height,
diff --git a/crates/zakura-network/src/zakura/block_sync/peer_routine.rs b/crates/zakura-network/src/zakura/block_sync/peer_routine.rs
index 2cd6b14c79..379ab8243a 100644
--- a/crates/zakura-network/src/zakura/block_sync/peer_routine.rs
+++ b/crates/zakura-network/src/zakura/block_sync/peer_routine.rs
@@ -41,14 +41,12 @@ use super::{
         DownloadWindow, LivenessOutcome, OutstandingBlockRange, ReceivedBlockTracker,
         ThroughputMeter,
     },
-    work_queue::{WorkItem, WorkQueue, WorkReturnOutcome},
+    work_queue::{RequestWrite, WorkItem, WorkQueue, WorkReturnOutcome},
     BlockSyncMessage, BlockSyncMisbehavior, BlockSyncPeerSession, BlockSyncStatus,
     ZakuraBlockSyncConfig, ZakuraPeerId, ZakuraTrace, MSG_BS_BLOCK,
 };
 use crate::zakura::transport::OrderedStreamFailure;
-use crate::zakura::{
-    trace::BlockBodySource, Admit, FramedRecv, OrderedSendError, SinkReject, ZakuraConnId,
-};
+use crate::zakura::{trace::BlockBodySource, Admit, FramedRecv, SinkReject, ZakuraConnId};
 use std::{sync::Arc, time::Duration, time::Instant};
 use tokio::time;
 use zakura_chain::{block, serialization::ZcashSerialize};
@@ -465,6 +463,7 @@ impl PeerRoutine {
             let retry_filter_deadline = if self.session.outbound_capacity() > 0 {
                 self.try_fill().await
             } else {
+                self.gc_skipped_outstanding();
                 None
             };
             let outbound_queue_has_capacity = self.session.outbound_capacity() > 0;
@@ -534,6 +533,7 @@ impl PeerRoutine {
         guard: &mut crate::zakura::SessionGuard,
         frame: crate::zakura::Frame,
     ) -> Result<(), SinkReject> {
+        self.gc_skipped_outstanding();
         match guard.admit(&frame) {
             Admit::Pass => {}
             Admit::Throttle => {
@@ -625,6 +625,24 @@ impl PeerRoutine {
         Ok(())
     }
 
+    /// Return only unreceived work still owned by this routine. A body already
+    /// handed to the sequencer keeps its ownership and cannot be requeued here.
+    fn return_unreceived_requests(&mut self, reason: &'static str) {
+        let outstanding_ranges = std::mem::take(&mut self.window.outstanding);
+        for outstanding in outstanding_ranges {
+            let unreceived: Vec<_> = unreceived_heights(&outstanding).collect();
+            let outcome = self
+                .work
+                .release_reserved_and_return_items_detailed_for_owner(
+                    outstanding.request.owner,
+                    unreceived.iter().copied(),
+                );
+            self.budget.release(outcome.released_bytes);
+            self.trace_work_returned(reason, &outstanding, unreceived.len(), outcome);
+        }
+        self.registry.clear_outstanding(&self.peer, self.generation);
+    }
+
     async fn reserve_body_decode_permit(
         &self,
     ) -> Result<mpsc::OwnedPermit<SequencedBody>, SinkReject> {
@@ -713,18 +731,7 @@ impl PeerRoutine {
         // dropped successor heights. Return our unreceived outstanding to
         // `work.pending` (a no-op for heights already dropped from `in_flight` by
         // `reset_above`) and release their reservations exactly once.
-        let outstanding = std::mem::take(&mut self.window.outstanding);
-        for outstanding in outstanding {
-            let unreceived: Vec<_> = unreceived_heights(&outstanding).collect();
-            let outcome = self
-                .work
-                .release_reserved_and_return_items_detailed_for_owner(
-                    outstanding.request.owner,
-                    unreceived.iter().copied(),
-                );
-            self.budget.release(outcome.released_bytes);
-            self.trace_work_returned("view_reset", &outstanding, unreceived.len(), outcome);
-        }
+        self.return_unreceived_requests("view_reset");
         self.retry_avoid.clear();
         // Clear our (now-empty) registry outstanding and refresh slot diagnostics.
         self.publish_outstanding();
@@ -732,7 +739,7 @@ impl PeerRoutine {
         // no-progress probe streak must not stay charged: reset it (and clear the idle
         // liveness deadline) so an unproven peer whose only probe was in flight at the
         // reset can probe again instead of wedging at its cap.
-        self.window.note_view_reset();
+        self.window.note_locally_returned_requests();
         // Ping the producer immediately: `reset_above` emptied `pending`, and the
         // reactor's post-reset query may have run while our (now cleared) outstanding
         // still inflated the low-water gate. Without this ping a routine that then
@@ -790,6 +797,7 @@ impl PeerRoutine {
     /// There is no floor gate: downloads are governed by the byte budget and
     /// per-peer slots, never floor-distance / near-tip lag.
     async fn try_fill(&mut self) -> Option<Instant> {
+        self.gc_skipped_outstanding();
         // The BBR cwnd is clamped to the peer's advertised hard cap inside
         // `available_slots`, so there is no separate window to reconcile on a
         // `Status` change.
@@ -812,6 +820,7 @@ impl PeerRoutine {
         // `&'static str` reason via `break`; a pass that issues nothing (`fill_sent == 0`)
         // is a candidate bubble.
         let mut fill_sent = 0u32;
+        let request_sender = self.session.request_sender();
         let fill_stop: FillStop = loop {
             // Floor bypass scaled by reliability: a healthy saturated carrier keeps the
             // full bypass so the floor keeps moving; a failing/sealed peer earns *no*
@@ -838,6 +847,27 @@ impl PeerRoutine {
             if floor_slots == 0 {
                 break FillStop::CwndSaturated;
             }
+            // Reserve transport capacity before taking work or charging bytes.
+            let slot = match request_sender.try_reserve_guarded() {
+                Ok(slot) => slot,
+                Err(crate::zakura::transport::GuardedReserveError::Full) => {
+                    break FillStop::OutboundFull
+                }
+                Err(error) => {
+                    tracing::debug!(
+                        peer = ?self.peer,
+                        generation = self.generation,
+                        ?error,
+                        "could not reserve guarded block request transport capacity"
+                    );
+                    self.session.cancel_token().cancel();
+                    break FillStop::SendError;
+                }
+            };
+            let Some(request_id) = self.next_request_id else {
+                break FillStop::Internal;
+            };
+            self.next_request_id = request_id.get().checked_add(1).and_then(NonZeroU64::new);
             let in_bypass = normal_slots == 0;
             let (servable_low, servable_high) = (self.servable_low, self.servable_high);
 
@@ -893,11 +923,13 @@ impl PeerRoutine {
                                 .window
                                 .cwnd_byte_headroom_at(floor_bonus, now)
                                 .unwrap_or(u64::MAX);
-                            items = self.work.take_in_range_budgeted(
+                            items = self.work.take_for_request(
                                 servable_low,
                                 grant.take_high,
                                 max_count,
                                 grant.max_request_bytes.min(floor_cwnd_cap).max(1),
+                                self.generation,
+                                request_id,
                             );
                         }
                         AdmissionOutcome::LookaheadAtCap => break FillStop::LookaheadCap,
@@ -938,11 +970,13 @@ impl PeerRoutine {
                             .window
                             .cwnd_byte_headroom_at(0, now)
                             .unwrap_or(u64::MAX);
-                        items = self.work.take_in_range_budgeted(
+                        items = self.work.take_for_request(
                             servable_low,
                             grant.take_high,
                             max_count,
                             grant.max_request_bytes.min(above_cwnd_cap),
+                            self.generation,
+                            request_id,
                         );
                     }
                     // A floor-priority start while the floor arm deferred to a
@@ -963,7 +997,7 @@ impl PeerRoutine {
             // with heights this routine recently *failed* (RangeUnavailable /
             // timeout / send-failure), quietly put those back so another peer can
             // contest them first, and only keep the suffix this routine is allowed
-            // to re-take. `return_items_quiet` does NOT notify (the other peers were
+            // to re-take. `return_unpublished` does NOT notify (the other peers were
             // already woken by the original failure return), so this cannot
             // self-wake into a take/return spin. If the whole chunk is still
             // avoided, break — the routine wakes to retry when the avoid window
@@ -981,22 +1015,20 @@ impl PeerRoutine {
                 let Some(keep) =
                     first_allowed_run(&items, |(height, item)| is_allowed(height, item))
                 else {
-                    let avoided: Vec<_> = items.iter().map(|(h, _)| *h).collect();
-                    self.work.return_items_quiet(avoided);
+                    self.work.return_unpublished(&items);
                     retry_filter_deadline = Some(self.retry_filter_wake_deadline(now));
                     break FillStop::RetryAvoid;
                 };
                 let keep_len = keep.len();
                 let mut returned_avoided = false;
                 if keep.start > 0 {
-                    let avoided: Vec<_> = items.drain(..keep.start).map(|(h, _)| h).collect();
-                    self.work.return_items_quiet(avoided);
+                    let avoided: Vec<_> = items.drain(..keep.start).collect();
+                    self.work.return_unpublished(&avoided);
                     returned_avoided = true;
                 }
                 if keep_len < items.len() {
                     let avoided = items.split_off(keep_len);
-                    self.work
-                        .return_items_quiet(avoided.into_iter().map(|(height, _)| height));
+                    self.work.return_unpublished(&avoided);
                     returned_avoided = true;
                 }
                 if returned_avoided {
@@ -1038,39 +1070,17 @@ impl PeerRoutine {
                 self.return_taken_items(&items);
                 break FillStop::Budget;
             }
-            let Some(request_id) = self.next_request_id else {
-                self.budget.release(reserved_bytes);
-                self.return_taken_items(&items);
-                break FillStop::Internal;
-            };
-            self.next_request_id = request_id.get().checked_add(1).and_then(NonZeroU64::new);
             let owner = scope.bind(self.generation, request_id);
-            let marked = self
-                .work
-                .mark_reserved_for_owner(owner, items.iter().map(|(height, _)| *height));
-            if marked != reserved_bytes {
-                self.budget.release(reserved_bytes);
-                let _ = self
-                    .work
-                    .release_reserved_and_return_items_detailed_for_owner(
-                        owner,
-                        items.iter().map(|(height, _)| *height),
-                    );
-                break FillStop::Internal;
-            }
-
+            let claim = RequestWrite::new(
+                owner,
+                items.clone(),
+                self.work.clone(),
+                self.budget.clone(),
+                self.session.cancel_token(),
+            );
             let count = match u32::try_from(kept_count) {
                 Ok(count) => count,
-                Err(_) => {
-                    let released = self
-                        .work
-                        .release_reserved_and_return_items_detailed_for_owner(
-                            owner,
-                            items.iter().map(|(height, _)| *height),
-                        );
-                    self.budget.release(released.released_bytes);
-                    break FillStop::Internal;
-                }
+                Err(_) => break FillStop::Internal,
             };
             let request = BlockRangeRequest {
                 owner,
@@ -1096,62 +1106,69 @@ impl PeerRoutine {
                 start_height: request.start_height,
                 count: request.count,
             };
-            if let Err(error) = self
-                .session
-                .try_send_get_blocks(request.start_height, request.count)
-            {
-                tracing::debug!(
-                    peer = ?self.peer,
-                    start_height = ?request.start_height,
-                    count = request.count,
-                    ?error,
-                    "failed to queue Zakura block-sync GetBlocks"
-                );
-                self.trace_queue_send_failed(&msg, &error);
-                // Return every still-reserved height to the queue. A competing
-                // peer's late body may have claimed a taken height and released its
-                // request reservation during the reserve await; leave that height
-                // in flight rather than re-queueing or releasing it twice.
-                let released = self
-                    .work
-                    .release_reserved_and_return_items_detailed_for_owner(
-                        request.owner,
-                        items.iter().map(|(height, _)| *height),
-                    );
-                self.budget.release(released.released_bytes);
-                if matches!(error, OrderedSendError::Full) {
-                    break FillStop::OutboundFull;
+            let frame = match msg.encode_frame() {
+                Ok(frame) => frame,
+                Err(_) => {
+                    self.session.cancel_token().cancel();
+                    break FillStop::SendError;
                 }
-                self.session.cancel_token().cancel();
-                break FillStop::SendError;
-            }
+            };
 
+            // A block-count delivery sample proves progress even though it cannot
+            // supply a byte rate for estimating transfer time.
+            let byte_rate = self.window.bbr_btlbw_bytes_per_sec(queued_at);
+            let has_delivery_measurement =
+                byte_rate.is_some() || self.window.bbr_btlbw_milliblocks(queued_at).is_some();
             let deadline = request_deadline(
                 request_priority,
                 queued_at,
                 self.config.request_timeout,
                 self.config.effective_floor_rescue_timeout(),
-                reserved_bytes,
+                // Responses share an ordered stream. Include earlier unreceived
+                // work so this request cannot expire while those bodies arrive.
+                self.window
+                    .outstanding_reserved_bytes()
+                    .saturating_add(reserved_bytes),
                 // Filter BtlBw by the request's send time so a stale-high rate from a
                 // now-slow peer cannot tighten the deadline below what it can meet.
-                self.window.bbr_btlbw_bytes_per_sec(queued_at),
+                byte_rate,
+                has_delivery_measurement,
             );
+            let request_start_height = request.start_height;
+            let request_count = request.count;
+            let request_estimated_bytes = request.estimated_bytes;
+            let mut delivered = false;
+            if !claim.publish(|| {
+                self.window.outstanding.push(OutstandingBlockRange {
+                    request,
+                    write_status: claim.status(),
+                    charged_for_liveness: false,
+                    queued_at,
+                    deadline,
+                    delivery_snapshot: self.window.delivery_snapshot(queued_at),
+                    delivered_bytes: 0,
+                    received: ReceivedBlockTracker::default(),
+                });
+                delivered = slot.send_request(frame, claim.clone());
+            }) {
+                break FillStop::Internal;
+            }
+            if !delivered {
+                tracing::debug!(
+                    peer = ?self.peer,
+                    generation = self.generation,
+                    start_height = ?request_start_height,
+                    count = request_count,
+                    "block request transport closed during publication"
+                );
+                claim.delivery_failed();
+                break FillStop::SendError;
+            }
             metrics::counter!("sync.block.request.sent").increment(1);
             if in_bypass {
                 // A floor request borrowed a bypass slot while the cwnd was saturated.
                 metrics::counter!("sync.block.request.floor_bypass").increment(1);
             }
-            let request_start_height = request.start_height;
-            let request_count = request.count;
-            let request_estimated_bytes = request.estimated_bytes;
-            self.window.outstanding.push(OutstandingBlockRange {
-                request,
-                queued_at,
-                deadline,
-                delivery_snapshot: self.window.delivery_snapshot(queued_at),
-                delivered_bytes: 0,
-                received: ReceivedBlockTracker::default(),
-            });
             self.window
                 .arm_liveness(queued_at, self.config.effective_liveness_timeout());
             self.publish_outstanding();
@@ -1258,8 +1275,7 @@ impl PeerRoutine {
     /// not re-wake its own want-work arm into a take/return spin, and any other
     /// peer waiting on budget capacity is woken by the matching `budget.release`.
     fn return_taken_items(&self, items: &[(block::Height, WorkItem)]) {
-        self.work
-            .return_items_quiet(items.iter().map(|(height, _)| *height));
+        self.work.return_unpublished(items);
     }
 
     /// Record heights this routine just returned on a failure so it will not
@@ -1283,6 +1299,23 @@ impl PeerRoutine {
     }
 
     fn expire_due_timeouts(&mut self, now: Instant) -> bool {
+        self.gc_skipped_outstanding();
+        // Arbitrate due queued frames against writer startup before charging
+        // failures. Keep started requests outstanding while refunding skipped
+        // probes so their liveness deadline cannot be cleared as idle.
+        let mut skipped = Vec::new();
+        let mut index = 0;
+        while index < self.window.outstanding.len() {
+            let outstanding = &self.window.outstanding[index];
+            if outstanding.deadline <= now
+                && (outstanding.write_status.expire_unwritten()
+                    || outstanding.write_status.was_skipped())
+            {
+                skipped.push(self.window.retire_locally(index));
+            } else {
+                index += 1;
+            }
+        }
         let mut timed_out = Vec::new();
         let mut index = 0;
         while index < self.window.outstanding.len() {
@@ -1292,11 +1325,13 @@ impl PeerRoutine {
                 index += 1;
             }
         }
-        if timed_out.is_empty() {
+        if skipped.is_empty() && timed_out.is_empty() {
             return false;
         }
-        self.window.record_timeout(timed_out.len());
-        for outstanding in &timed_out {
+        if !timed_out.is_empty() {
+            self.window.record_timeout(timed_out.len());
+        }
+        for outstanding in skipped.iter().chain(&timed_out) {
             // Return only the unreceived heights — received ones are buffered (in
             // `in_flight` until committed); re-queuing them would re-fetch a body
             // we already hold (the WorkQueue single-owner invariant forbids it).
@@ -1390,6 +1425,7 @@ impl PeerRoutine {
         // Local reset may have withdrawn the requests while decoding waited
         // for capacity. It must remain neutral when the failed stream retires.
         self.on_view_changed();
+        self.gc_skipped_outstanding();
         if self.window.outstanding.is_empty() && self.window.block_liveness_deadline.is_none() {
             return Ok(());
         }
@@ -1419,7 +1455,7 @@ impl PeerRoutine {
         let mut index = 0;
         while index < self.window.outstanding.len() {
             if self.window.outstanding[index].request.end_height() <= floor {
-                let outstanding = self.window.outstanding.remove(index);
+                let outstanding = self.window.retire_locally(index);
                 // Release only estimates whose per-height ledger is still
                 // `Reserved`. A competing delivery changes that ledger to
                 // `Released` at receipt, so floor GC must not release it again.
@@ -1441,6 +1477,14 @@ impl PeerRoutine {
         }
     }
 
+    /// A skipped frame ends the peer obligation independently of any received
+    /// body whose owner must remain available to the sequencer.
+    fn gc_skipped_outstanding(&mut self) {
+        if self.window.discard_skipped_requests() {
+            self.publish_outstanding();
+        }
+    }
+
     /// Free request slots after the central queue retires their exact owners.
     /// Queue retirement already released their reservations.
     /// The cleanup path drops only routine-local and registry bookkeeping.
@@ -1459,7 +1503,7 @@ impl PeerRoutine {
             if still_owned {
                 index += 1;
             } else {
-                self.window.outstanding.remove(index);
+                self.window.retire_locally(index);
                 removed = true;
             }
         }
@@ -2060,7 +2104,7 @@ impl PeerRoutine {
         if index >= self.window.outstanding.len() {
             return;
         }
-        self.window.outstanding.remove(index);
+        self.window.retire_locally(index);
         self.publish_outstanding();
         self.window.disarm_liveness_after_progress_if_idle();
     }
@@ -2142,24 +2186,34 @@ impl PeerRoutine {
     /// `work.in_flight` instead — the producer's `!in_flight_contains` clause
     /// already keeps them out of `pending`.
     fn publish_outstanding(&self) {
-        let mut map: BTreeMap<block::Height, super::peer_registry::OutstandingMeta> =
-            BTreeMap::new();
+        let mut unreceived = Vec::new();
         for outstanding in &self.window.outstanding {
             for expected in &outstanding.request.expected_blocks {
                 if !outstanding.has_received(expected.height) {
-                    map.insert(
-                        expected.height,
-                        super::peer_registry::OutstandingMeta {
-                            owner: outstanding.request.owner,
-                            hash: expected.hash,
-                            estimated_bytes: expected.estimated_bytes,
-                            queued_at: outstanding.queued_at,
-                            deadline: outstanding.deadline,
-                        },
-                    );
+                    unreceived.push((expected.height, (outstanding, expected)));
                 }
             }
         }
+        // Filter before combining overlapping heights so a stale request cannot
+        // hide the current owner's metadata.
+        self.work.retain_owned(&mut unreceived, |(outstanding, _)| {
+            outstanding.request.owner
+        });
+        let map = unreceived
+            .into_iter()
+            .map(|(height, (outstanding, expected))| {
+                (
+                    height,
+                    super::peer_registry::OutstandingMeta {
+                        owner: outstanding.request.owner,
+                        hash: expected.hash,
+                        estimated_bytes: expected.estimated_bytes,
+                        queued_at: outstanding.queued_at,
+                        deadline: outstanding.deadline,
+                    },
+                )
+            })
+            .collect::<BTreeMap<_, _>>();
         if map.is_empty() {
             self.registry.clear_outstanding(&self.peer, self.generation);
         } else {
@@ -2264,25 +2318,7 @@ impl Drop for PeerRoutine {
     /// The reactor owns entry insert (on connect) and remove (on disconnect/
     /// admission-reject); see `handle_peer_disconnected`.
     fn drop(&mut self) {
-        let outstanding_ranges = std::mem::take(&mut self.window.outstanding);
-        for outstanding in outstanding_ranges {
-            let unreceived: Vec<_> = outstanding
-                .request
-                .expected_blocks
-                .iter()
-                .filter(|expected| !outstanding.has_received(expected.height))
-                .map(|expected| expected.height)
-                .collect();
-            let outcome = self
-                .work
-                .release_reserved_and_return_items_detailed_for_owner(
-                    outstanding.request.owner,
-                    unreceived.iter().copied(),
-                );
-            self.budget.release(outcome.released_bytes);
-            self.trace_work_returned("peer_routine_drop", &outstanding, unreceived.len(), outcome);
-        }
-        self.registry.clear_outstanding(&self.peer, self.generation);
+        self.return_unreceived_requests("peer_routine_drop");
     }
 }
 
@@ -2302,7 +2338,7 @@ mod tests {
     use super::super::sequencer_task::initial_view;
     use super::super::state::{ByteBudget, ThroughputMeter};
     use super::super::work_queue::WorkQueue;
-    use super::super::{BlockSyncFrontiers, BlockSyncPeerSession, ZakuraBlockSyncConfig};
+    use super::super::{BlockSyncFrontiers, BlockSyncPeerSession, CwndUnit, ZakuraBlockSyncConfig};
     use super::PeerRoutine;
     use crate::zakura::framed_channel;
     use crate::zakura::trace::ZakuraTrace;
@@ -2468,7 +2504,19 @@ mod tests {
     /// and is sent without a sequencer round trip.
     #[tokio::test]
     async fn floor_overdraft_is_bounded_and_immediate() {
-        let config = ZakuraBlockSyncConfig::default();
+        for unit in [CwndUnit::Bytes, CwndUnit::Blocks] {
+            for measurement_age in [None, Some(Duration::ZERO), Some(Duration::from_secs(11))] {
+                check_floor_overdraft_and_deadline(unit, measurement_age).await;
+            }
+        }
+    }
+
+    async fn check_floor_overdraft_and_deadline(unit: CwndUnit, measurement_age: Option<Duration>) {
+        let config = ZakuraBlockSyncConfig {
+            bbr_cwnd_unit: unit,
+            bbr_delivery_rate_window: Duration::from_secs(10),
+            ..ZakuraBlockSyncConfig::default()
+        };
 
         // A byte budget reserved down to exactly zero free: the case that used to wedge.
         let mut budget = ByteBudget::new(8_192);
@@ -2530,8 +2578,32 @@ mod tests {
         routine.servable_low = block::Height(1);
         routine.servable_high = block::Height(10);
 
+        if let Some(age) = measurement_age {
+            let delivered_at = Instant::now() - age;
+            let elapsed = Duration::from_secs(1);
+            let snapshot = routine.window.delivery_snapshot(delivered_at - elapsed);
+            routine
+                .window
+                .record_delivery(delivered_at, elapsed, 1, 256 * 1024, snapshot);
+        }
         let _ = routine.try_fill().await;
 
+        let outstanding = &routine.window.outstanding[0];
+        let base = if measurement_age == Some(Duration::ZERO) {
+            routine.config.effective_floor_rescue_timeout()
+        } else {
+            routine.config.request_timeout
+        };
+        let transfer = Duration::from_secs_f64(
+            f64::from(u32::try_from(outstanding.request.estimated_bytes).unwrap())
+                / (256.0 * 1024.0),
+        );
+        assert_eq!(
+            outstanding.deadline,
+            outstanding.queued_at + base + transfer,
+            "unit={unit:?}, measurement_age={measurement_age:?}"
+        );
+
         // The floor request went out synchronously (no funding round trip)…
         let frame = timeout(Duration::from_secs(5), out_recv.recv())
             .await
@@ -2558,8 +2630,465 @@ mod tests {
         );
     }
 
-    /// Routine teardown must not release or requeue a height already received
-    /// through first-completion-wins.
+    #[derive(Clone, Copy, Debug, Eq, PartialEq)]
+    enum DeadlineWriteState {
+        Queued,
+        Started,
+        Written,
+    }
+
+    #[tokio::test]
+    async fn deadline_expiry_only_penalizes_requests_that_started() {
+        use DeadlineWriteState::*;
+
+        for writes in [
+            vec![Queued],
+            vec![Started],
+            vec![Written],
+            vec![Started, Queued],
+            vec![Written, Queued],
+            vec![Written, Started, Queued],
+        ] {
+            for progress_after_first in [false, true] {
+                check_deadline_write_states(&writes, progress_after_first).await;
+            }
+        }
+    }
+
+    async fn check_deadline_write_states(
+        writes: &[DeadlineWriteState],
+        progress_after_first: bool,
+    ) {
+        let config = ZakuraBlockSyncConfig {
+            initial_block_probe_requests: u32::try_from(writes.len()).unwrap(),
+            ..ZakuraBlockSyncConfig::default()
+        };
+        let liveness_timeout = config.effective_liveness_timeout();
+        let mut expected_window = super::DownloadWindow::new(&config);
+        let budget = ByteBudget::new(1_000_000);
+        let work = Arc::new(WorkQueue::new(block::Height(0)));
+        work.set_estimate_floor_for_tests(1);
+        let cancel = CancellationToken::new();
+        let (out_send, mut out_recv) = crate::zakura::transport::worker_framed_channel(16);
+        let (_in_send, in_recv) = framed_channel(16);
+        let peer = ZakuraPeerId::new(vec![9u8; 32]).unwrap();
+        let session = BlockSyncPeerSession::for_test(peer.clone(), out_send, cancel.clone());
+        let registry = Arc::new(PeerRegistry::new());
+        let generation = registry
+            .admit_session(
+                &peer,
+                crate::zakura::ServicePeerDirection::Outbound,
+                &config,
+                0,
+                Instant::now(),
+            )
+            .generation();
+        let (sequencer_input_tx, _sequencer_input_rx) = mpsc::channel(16);
+        let (routine_to_reactor_tx, _routine_to_reactor_rx) = mpsc::channel(16);
+        let (_view_tx, view_rx) = watch::channel(initial_view(BlockSyncFrontiers {
+            finalized_height: block::Height(0),
+            verified_block_tip: block::Height(0),
+            verified_block_hash: block::Hash([0; 32]),
+        }));
+        let mut routine = PeerRoutine::new(
+            peer.clone(),
+            0,
+            session,
+            in_recv,
+            config,
+            true,
+            generation,
+            budget.clone(),
+            work.clone(),
+            registry.clone(),
+            Arc::new(Mutex::new(ThroughputMeter::new(Instant::now()))),
+            sequencer_input_tx,
+            Arc::new(AtomicU64::new(0)),
+            Arc::new(AtomicU64::new(0)),
+            routine_to_reactor_tx,
+            view_rx,
+            cancel.clone(),
+            ZakuraTrace::noop(),
+        );
+        routine.handle_status(super::BlockSyncStatus {
+            servable_low: block::Height(1),
+            servable_high: block::Height(10),
+            max_blocks_per_response: 1,
+            ..super::BlockSyncStatus::default()
+        });
+
+        let mut started_writes = Vec::new();
+        let mut peer_timeouts = Vec::new();
+        for (index, state) in writes.iter().enumerate() {
+            let height = u8::try_from(index + 2).unwrap();
+            work.extend(
+                super::super::test_work_scope(),
+                [(
+                    block::Height(u32::from(height)),
+                    block::Hash([height; 32]),
+                    BlockSizeEstimate::Confirmed(1_000),
+                )],
+            );
+            routine.try_fill().await;
+            assert_eq!(routine.window.outstanding.len(), index + 1);
+            match state {
+                DeadlineWriteState::Queued => {}
+                DeadlineWriteState::Started => {
+                    let frame = out_recv.recv().await.unwrap();
+                    let (started_tx, started_rx) = tokio::sync::oneshot::channel();
+                    let (finish_tx, finish_rx) = tokio::sync::oneshot::channel();
+                    let writer = tokio::spawn(frame.write_with(move |_| async move {
+                        started_tx.send(()).unwrap();
+                        finish_rx.await.unwrap();
+                        Ok::<_, ()>(())
+                    }));
+                    timeout(Duration::from_secs(1), started_rx)
+                        .await
+                        .unwrap()
+                        .unwrap();
+                    started_writes.push((finish_tx, writer));
+                    peer_timeouts.push(block::Height(u32::from(height)));
+                }
+                DeadlineWriteState::Written => {
+                    out_recv
+                        .recv()
+                        .await
+                        .unwrap()
+                        .write_with(|_| async { Ok::<_, ()>(()) })
+                        .await
+                        .unwrap();
+                    peer_timeouts.push(block::Height(u32::from(height)));
+                }
+            }
+            if progress_after_first && index == 0 {
+                routine
+                    .window
+                    .note_block_progress(Instant::now(), liveness_timeout);
+            }
+        }
+        let first_owner = routine.window.outstanding[0].request.owner;
+        let liveness_deadline = routine.window.block_liveness_deadline.unwrap();
+        let deadline = routine
+            .window
+            .outstanding
+            .iter()
+            .map(|request| request.deadline)
+            .max()
+            .unwrap();
+        assert!(routine.expire_due_timeouts(deadline));
+        if !peer_timeouts.is_empty() {
+            expected_window.record_timeout(peer_timeouts.len());
+        }
+        assert_eq!(
+            routine.window.bbr_reliability_permille(),
+            expected_window.bbr_reliability_permille(),
+            "only started writes count as peer failures: {writes:?}, progress={progress_after_first}"
+        );
+        assert_eq!(
+            routine.window.bbr_effective_cwnd(),
+            expected_window.bbr_effective_cwnd(),
+            "queued-only expiry must not dip the congestion window"
+        );
+        assert_eq!(
+            routine.window.requests_without_block_progress,
+            u32::try_from(peer_timeouts.len())
+                .unwrap()
+                .saturating_sub(u32::from(
+                    progress_after_first && writes[0] != DeadlineWriteState::Queued
+                )),
+        );
+        assert_eq!(
+            routine.retry_avoid.keys().copied().collect::<Vec<_>>(),
+            peer_timeouts,
+        );
+        assert!(routine.window.outstanding.is_empty());
+        assert_eq!(budget.reserved(), 0);
+        assert_eq!(work.reserved_bytes(), 0);
+        assert_eq!(work.pending_len(), writes.len());
+        for index in 0..writes.len() {
+            let height = block::Height(u32::try_from(index + 2).unwrap());
+            assert!(work.pending_contains(height));
+            assert!(!registry.peer_has_outstanding_height(&peer, height));
+        }
+        assert_eq!(
+            routine.window.check_liveness(liveness_deadline),
+            if peer_timeouts.is_empty() {
+                super::LivenessOutcome::Ok
+            } else {
+                super::LivenessOutcome::Park
+            },
+        );
+        assert!(!cancel.is_cancelled());
+
+        if peer_timeouts.is_empty() {
+            // Reissue before the old queued frame is drained. Its later drop
+            // must preserve the replacement's ownership and byte charge.
+            routine.try_fill().await;
+            assert_eq!(routine.window.outstanding.len(), 1);
+            assert_ne!(routine.window.outstanding[0].request.owner, first_owner);
+        }
+        for (finish, writer) in started_writes {
+            finish.send(()).unwrap();
+            timeout(Duration::from_secs(1), writer)
+                .await
+                .unwrap()
+                .unwrap()
+                .unwrap();
+        }
+        for state in writes {
+            if *state == DeadlineWriteState::Queued {
+                out_recv
+                    .recv()
+                    .await
+                    .unwrap()
+                    .write_with(|_| async {
+                        panic!("a request expired before writer startup must not reach the wire");
+                        #[allow(unreachable_code)]
+                        Ok::<_, ()>(())
+                    })
+                    .await
+                    .unwrap();
+            }
+        }
+        assert_eq!(
+            budget.reserved(),
+            if peer_timeouts.is_empty() { 1_000 } else { 0 }
+        );
+        assert_eq!(budget.reserved(), work.reserved_bytes());
+        assert!(!cancel.is_cancelled());
+    }
+
+    #[derive(Clone, Copy)]
+    enum ReceiptCleanup {
+        BeforeWriter,
+        Writer,
+        CommittedGc,
+        ObsoleteGc,
+    }
+
+    #[tokio::test]
+    async fn skipped_write_retires_only_its_own_peer_obligation() {
+        for order in [
+            ReceiptCleanup::Writer,
+            ReceiptCleanup::CommittedGc,
+            ReceiptCleanup::ObsoleteGc,
+        ] {
+            for (other_request, progress_before_other) in
+                [(false, false), (true, false), (true, true)]
+            {
+                check_skipped_write_obligation(order, other_request, progress_before_other).await;
+            }
+        }
+    }
+
+    #[tokio::test]
+    async fn competing_receipt_retires_unwritable_request_before_writer_reaches_it() {
+        check_skipped_write_obligation(ReceiptCleanup::BeforeWriter, false, false).await;
+    }
+
+    async fn check_skipped_write_obligation(
+        order: ReceiptCleanup,
+        other_request: bool,
+        progress_before_other: bool,
+    ) {
+        let config = ZakuraBlockSyncConfig {
+            request_timeout: if matches!(order, ReceiptCleanup::BeforeWriter) {
+                Duration::from_secs(1)
+            } else {
+                ZakuraBlockSyncConfig::default().request_timeout
+            },
+            initial_block_probe_requests: if other_request { 2 } else { 1 },
+            ..ZakuraBlockSyncConfig::default()
+        };
+        let liveness_timeout = config.effective_liveness_timeout();
+        let mut budget = ByteBudget::new(1_000_000);
+        let work = Arc::new(WorkQueue::new(block::Height(0)));
+        work.set_estimate_floor_for_tests(1);
+        let add_work = |height: u8| {
+            work.extend(
+                super::super::test_work_scope(),
+                [(
+                    block::Height(u32::from(height)),
+                    block::Hash([height; 32]),
+                    BlockSizeEstimate::Confirmed(
+                        if matches!(order, ReceiptCleanup::BeforeWriter) {
+                            1_000_000
+                        } else {
+                            1_000
+                        },
+                    ),
+                )],
+            );
+        };
+        // Height 1 stays missing, so the competing body at height 2 cannot move
+        // the committed floor and conceal a stale outstanding request.
+        add_work(2);
+        let cancel = CancellationToken::new();
+        let (out_send, mut out_recv) = crate::zakura::transport::worker_framed_channel(16);
+        let (_in_send, in_recv) = framed_channel(16);
+        let peer = ZakuraPeerId::new(vec![9u8; 32]).unwrap();
+        let session = BlockSyncPeerSession::for_test(peer.clone(), out_send, cancel.clone());
+        let registry = Arc::new(PeerRegistry::new());
+        let generation = registry
+            .admit_session(
+                &peer,
+                crate::zakura::ServicePeerDirection::Outbound,
+                &config,
+                0,
+                Instant::now(),
+            )
+            .generation();
+        let (sequencer_input_tx, _sequencer_input_rx) = mpsc::channel(16);
+        let (routine_to_reactor_tx, _routine_to_reactor_rx) = mpsc::channel(16);
+        let (view_tx, view_rx) = watch::channel(initial_view(BlockSyncFrontiers {
+            finalized_height: block::Height(0),
+            verified_block_tip: block::Height(0),
+            verified_block_hash: block::Hash([0; 32]),
+        }));
+        let mut routine = PeerRoutine::new(
+            peer.clone(),
+            0,
+            session,
+            in_recv,
+            config,
+            true,
+            generation,
+            budget.clone(),
+            work.clone(),
+            registry.clone(),
+            Arc::new(Mutex::new(ThroughputMeter::new(Instant::now()))),
+            sequencer_input_tx,
+            Arc::new(AtomicU64::new(0)),
+            Arc::new(AtomicU64::new(0)),
+            routine_to_reactor_tx,
+            view_rx,
+            cancel.clone(),
+            ZakuraTrace::noop(),
+        );
+        routine.handle_status(super::BlockSyncStatus {
+            servable_low: block::Height(1),
+            servable_high: block::Height(10),
+            max_blocks_per_response: 1,
+            ..super::BlockSyncStatus::default()
+        });
+        routine.try_fill().await;
+        assert_eq!(routine.window.outstanding.len(), 1);
+        let owner = routine.window.outstanding[0].request.owner;
+        let deadline = routine.window.outstanding[0].deadline;
+        let queued = out_recv.recv().await.unwrap();
+        assert!(registry.peer_has_outstanding_height(&peer, block::Height(2)));
+
+        if progress_before_other {
+            routine
+                .window
+                .note_block_progress(Instant::now(), liveness_timeout);
+        }
+        if other_request {
+            add_work(3);
+            routine.try_fill().await;
+            assert_eq!(routine.window.outstanding.len(), 2);
+        }
+        let skipped = work.subscribe_available().notified();
+        tokio::pin!(skipped);
+        skipped.as_mut().enable();
+        // Another peer wins height 2 before this writer starts. Its body retains
+        // this exact owner until the sequencer consumes it.
+        budget.release(
+            work.release_active_reserved_height_for_owner(owner, block::Height(2))
+                .unwrap(),
+        );
+        match order {
+            ReceiptCleanup::BeforeWriter => {
+                let liveness_deadline = routine.window.block_liveness_deadline.unwrap();
+                assert!(
+                    liveness_deadline < deadline,
+                    "transfer allowance outlasts liveness"
+                );
+                assert_eq!(work.owner_for_height(block::Height(2)), Some(owner));
+                assert_eq!(budget.reserved(), 0);
+                routine.gc_obsolete_outstanding();
+                let result = routine.handle_deadlines(liveness_deadline).await;
+                assert!(
+                    result.is_ok(),
+                    "a request made unwritable by receipt must not park its peer: {result:?}"
+                );
+                assert!(routine.window.outstanding.is_empty());
+                return;
+            }
+            ReceiptCleanup::Writer => {}
+            ReceiptCleanup::CommittedGc => {
+                budget.release(work.advance_floor(block::Height(2)));
+                view_tx.send_modify(|view| view.download_floor = block::Height(2));
+                routine.gc_committed_outstanding();
+            }
+            ReceiptCleanup::ObsoleteGc => {
+                budget.release(work.advance_floor(block::Height(2)));
+                routine.gc_obsolete_outstanding();
+            }
+        }
+        queued
+            .write_with(|_| async {
+                panic!("a superseded request must not reach the wire");
+                #[allow(unreachable_code)]
+                Ok::<_, ()>(())
+            })
+            .await
+            .unwrap();
+        if matches!(order, ReceiptCleanup::Writer) {
+            timeout(Duration::from_secs(1), skipped)
+                .await
+                .expect("receipt retirement must wake its routine");
+        }
+        if other_request {
+            out_recv
+                .recv()
+                .await
+                .unwrap()
+                .write_with(|_| async { Ok::<_, ()>(()) })
+                .await
+                .unwrap();
+        }
+
+        // Deadline handling must reconcile the skipped write before scoring a
+        // timeout, even if it was the event that woke the routine.
+        routine.handle_deadlines(deadline).await.unwrap();
+        assert!(!registry.peer_has_outstanding_height(&peer, block::Height(2)));
+        assert_eq!(
+            work.owner_for_height(block::Height(2)),
+            matches!(order, ReceiptCleanup::Writer).then_some(owner),
+        );
+        assert!(!work.pending_contains(block::Height(2)));
+        assert!(routine.retry_avoid.is_empty());
+        assert!(!cancel.is_cancelled());
+        if other_request {
+            assert_eq!(routine.window.outstanding.len(), 1);
+            assert_eq!(routine.window.requests_without_block_progress, 1);
+            assert!(registry.peer_has_outstanding_height(&peer, block::Height(3)));
+            let liveness_deadline = routine.window.block_liveness_deadline.unwrap();
+            assert_eq!(
+                routine.window.check_liveness(liveness_deadline),
+                super::LivenessOutcome::Park
+            );
+            assert_eq!(budget.reserved(), 1_000);
+        } else {
+            assert!(routine.window.outstanding.is_empty());
+            assert_eq!(routine.window.requests_without_block_progress, 0);
+            assert!(routine.window.block_liveness_deadline.is_none());
+            routine
+                .handle_deadlines(deadline + liveness_timeout)
+                .await
+                .unwrap();
+            assert_eq!(budget.reserved(), 0);
+            add_work(3);
+            routine.try_fill().await;
+            assert_eq!(
+                routine.window.outstanding.len(),
+                1,
+                "the cold peer can probe again"
+            );
+        }
+    }
+
     #[tokio::test]
     async fn routine_drop_leaves_a_body_won_by_another_peer_to_the_sequencer() {
         let config = ZakuraBlockSyncConfig::default();
diff --git a/crates/zakura-network/src/zakura/block_sync/peer_routine/trace.rs b/crates/zakura-network/src/zakura/block_sync/peer_routine/trace.rs
index 117b7e98e1..38deaf3ceb 100644
--- a/crates/zakura-network/src/zakura/block_sync/peer_routine/trace.rs
+++ b/crates/zakura-network/src/zakura/block_sync/peer_routine/trace.rs
@@ -1,6 +1,6 @@
 use super::super::trace::{
     block_sync_message_label, elapsed_us, height as trace_height, peer as trace_peer,
-    saturating_usize, BlockTraceEvent, BlockTraceFields, BoolOrU64, QueueSendFailedEvent,
+    saturating_usize, BlockTraceEvent, BlockTraceFields, BoolOrU64,
 };
 use super::*;
 use crate::zakura::trace::block_sync_trace as bs_trace;
@@ -67,6 +67,7 @@ impl PeerRoutine {
         let unreceived_count = u64::try_from(unreceived_count).unwrap_or(u64::MAX);
         if outcome.missing_count == 0
             && outcome.released_count == 0
+            && outcome.committed_count == 0
             && outcome.returned_count == unreceived_count
         {
             return;
@@ -109,18 +110,6 @@ impl PeerRoutine {
         });
     }
 
-    pub(super) fn trace_queue_send_failed(&self, msg: &BlockSyncMessage, error: &OrderedSendError) {
-        self.trace.emit_event(|| {
-            QueueSendFailedEvent::peer_routine(
-                &self.peer,
-                msg,
-                error,
-                self.session.outbound_capacity(),
-                self.session.outbound_max_capacity(),
-            )
-        });
-    }
-
     pub(super) fn trace_get_blocks_sent(
         &self,
         start_height: block::Height,
@@ -288,6 +277,7 @@ impl PeerRoutine {
 fn insert_work_return_outcome(row: &mut BlockTraceFields, outcome: WorkReturnOutcome) {
     row.released_bytes = Some(outcome.released_bytes);
     row.returned_count = Some(outcome.returned_count);
+    row.committed_count = Some(outcome.committed_count);
     row.already_pending_count = Some(outcome.already_pending_count);
     row.released_count = Some(outcome.released_count);
     row.missing_count = Some(outcome.missing_count);
diff --git a/crates/zakura-network/src/zakura/block_sync/reactor.rs b/crates/zakura-network/src/zakura/block_sync/reactor.rs
index 1e468b03d2..28e0a56f32 100644
--- a/crates/zakura-network/src/zakura/block_sync/reactor.rs
+++ b/crates/zakura-network/src/zakura/block_sync/reactor.rs
@@ -575,13 +575,6 @@ impl BlockSyncReactor {
             ) {
                 continue;
             }
-            if servable_peers > 2 {
-                self.registry.avoid_floor_height_until(
-                    &claim.peer,
-                    claim.height,
-                    now + self.startup.config.effective_floor_peer_avoid_cooldown(),
-                );
-            }
             let released = self
                 .state
                 .work_queue
@@ -590,6 +583,16 @@ impl BlockSyncReactor {
                     [claim.height],
                 );
             self.state.budget.release(released.released_bytes);
+            // Settlement arbitrates against writer startup. Skipped frames and
+            // already-settled work must not count against the peer.
+            if servable_peers > 2 && released.returned_count > 0 && !released.request_was_unwritten
+            {
+                self.registry.avoid_floor_height_until(
+                    &claim.peer,
+                    claim.height,
+                    now + self.startup.config.effective_floor_peer_avoid_cooldown(),
+                );
+            }
             self.trace_floor_watchdog_cancelled(&claim, released);
             metrics::counter!("sync.block.floor_watchdog.cancelled").increment(1);
             tracing::debug!(
diff --git a/crates/zakura-network/src/zakura/block_sync/reactor/trace.rs b/crates/zakura-network/src/zakura/block_sync/reactor/trace.rs
index 5f04272e90..6ee9304b92 100644
--- a/crates/zakura-network/src/zakura/block_sync/reactor/trace.rs
+++ b/crates/zakura-network/src/zakura/block_sync/reactor/trace.rs
@@ -22,6 +22,7 @@ impl BlockSyncReactor {
             row.estimated_bytes = Some(claim.meta.estimated_bytes);
             row.released_bytes = Some(released.released_bytes);
             row.returned_count = Some(released.returned_count);
+            row.committed_count = Some(released.committed_count);
             row.already_pending_count = Some(released.already_pending_count);
             row.released_count = Some(released.released_count);
             row.missing_count = Some(released.missing_count);
diff --git a/crates/zakura-network/src/zakura/block_sync/service.rs b/crates/zakura-network/src/zakura/block_sync/service.rs
index 143d945c6e..8000707635 100644
--- a/crates/zakura-network/src/zakura/block_sync/service.rs
+++ b/crates/zakura-network/src/zakura/block_sync/service.rs
@@ -85,6 +85,10 @@ impl BlockSyncPeerSession {
         self.cancel_token.clone()
     }
 
+    pub(super) fn request_sender(&self) -> FramedSend {
+        self.send.clone()
+    }
+
     /// Current free slots in this peer's bounded outbound stream queue.
     pub fn outbound_capacity(&self) -> usize {
         self.send.capacity()
diff --git a/crates/zakura-network/src/zakura/block_sync/state.rs b/crates/zakura-network/src/zakura/block_sync/state.rs
index ffdc12bc66..5e0839c071 100644
--- a/crates/zakura-network/src/zakura/block_sync/state.rs
+++ b/crates/zakura-network/src/zakura/block_sync/state.rs
@@ -2,7 +2,7 @@ use super::{
     bbr::{rounded_usize, BbrState},
     config::*,
     request::*,
-    work_queue::WorkQueue,
+    work_queue::{RequestWriteStatus, WorkQueue},
     *,
 };
 use crate::zakura::{ServicePeerDirection, ServicePeerSnapshot, ZakuraBlockSyncCandidateState};
@@ -650,7 +650,7 @@ impl DownloadWindow {
     /// Bytes reserved across this peer's in-flight requests (the per-request size
     /// estimates of heights not yet received). Recomputed on demand — the byte unit is
     /// experimental; a hot path would maintain a running counter instead.
-    fn outstanding_reserved_bytes(&self) -> u64 {
+    pub(super) fn outstanding_reserved_bytes(&self) -> u64 {
         self.outstanding.iter().fold(0u64, |acc, range| {
             acc.saturating_add(range.reserved_bytes())
         })
@@ -703,6 +703,9 @@ impl DownloadWindow {
     }
 
     pub(super) fn arm_liveness(&mut self, now: Instant, timeout: Duration) {
+        if let Some(outstanding) = self.outstanding.last_mut() {
+            outstanding.charged_for_liveness = true;
+        }
         self.last_request_at = Some(now);
         self.requests_without_block_progress =
             self.requests_without_block_progress.saturating_add(1);
@@ -714,6 +717,9 @@ impl DownloadWindow {
     pub(super) fn note_block_progress(&mut self, now: Instant, timeout: Duration) {
         self.last_block_at = Some(now);
         self.requests_without_block_progress = 0;
+        for outstanding in &mut self.outstanding {
+            outstanding.charged_for_liveness = false;
+        }
         self.block_liveness_deadline = if self.outstanding.is_empty() {
             None
         } else {
@@ -738,21 +744,47 @@ impl DownloadWindow {
         }
     }
 
-    /// Reset per-view no-progress accounting after a destructive view reset. The reset
-    /// returned this peer's outstanding to the queue on *our* initiative (a reorg/rollback,
-    /// not the peer's fault), so the in-flight probe streak must not stay charged against
-    /// it: clearing `requests_without_block_progress` lets an unproven peer probe again
-    /// instead of wedging at its one-probe cap forever (the reset also cleared its liveness
-    /// deadline, so nothing would disconnect it). Proof state (`last_block_at`) is preserved.
-    pub(super) fn note_view_reset(&mut self) {
+    /// Clear the probe streak after we return requests on our own initiative,
+    /// such as a view reset. Keep proof of earlier
+    /// progress, but let even an unproven peer receive work again when we resume.
+    pub(super) fn note_locally_returned_requests(&mut self) {
         self.requests_without_block_progress = 0;
         self.clear_liveness_if_idle();
     }
 
-    /// Push the block-liveness deadline out by `timeout` when a would-be park is
-    /// attributable to *local* outbound backpressure, not the peer: while our outbound queue
-    /// is full the routine stops draining inbound, so a useful body may be sitting unread.
-    /// Avoids punishing the peer for our own write-side congestion.
+    /// Retire attempts that never reached the wire, refunding only their own
+    /// probe charges. Keep unanswered requests from this peer accountable.
+    pub(super) fn discard_skipped_requests(&mut self) -> bool {
+        let mut removed = false;
+        let mut index = 0;
+        while index < self.outstanding.len() {
+            if self.outstanding[index].write_status.was_skipped() {
+                self.retire_locally(index);
+                removed = true;
+            } else {
+                index += 1;
+            }
+        }
+        removed
+    }
+
+    /// End a locally retired obligation before discarding its write status. If
+    /// transport has not started, it must skip the frame and refund this probe.
+    pub(super) fn retire_locally(&mut self, index: usize) -> OutstandingBlockRange {
+        let outstanding = self.outstanding.remove(index);
+        outstanding.write_status.expire_unwritten();
+        if outstanding.write_status.was_skipped() && outstanding.charged_for_liveness {
+            self.requests_without_block_progress =
+                self.requests_without_block_progress.saturating_sub(1);
+        }
+        if self.requests_without_block_progress == 0 {
+            self.clear_liveness_if_idle();
+        }
+        outstanding
+    }
+
+    /// Give a briefly congested writer time to deliver our queued request before
+    /// parking the peer for not answering it. The caller bounds this grace.
     pub(super) fn extend_liveness_deadline(&mut self, now: Instant, timeout: Duration) {
         self.block_liveness_deadline = Some(now + timeout);
     }
@@ -860,6 +892,9 @@ impl PeerBlockState {
 #[derive(Clone, Debug)]
 pub(super) struct OutstandingBlockRange {
     pub(super) request: BlockRangeRequest,
+    pub(super) write_status: RequestWriteStatus,
+    /// Whether this attempt still contributes to the no-progress probe streak.
+    pub(super) charged_for_liveness: bool,
     pub(super) queued_at: Instant,
     pub(super) deadline: Instant,
     pub(super) delivery_snapshot: DeliverySnapshot,
@@ -975,12 +1010,12 @@ const _: () = assert!(MAX_BS_BLOCKS_PER_REQUEST <= RECEIVED_TRACKER_OFFSET_CAPAC
 #[derive(Clone, Debug, Default)]
 pub(super) struct ReceivedBlockTracker {
     bits: u128,
-    count: usize,
 }
 
 impl ReceivedBlockTracker {
     pub(super) fn len(&self) -> usize {
-        self.count
+        // At most 128 set bits fit in usize on every supported target.
+        self.bits.count_ones() as usize
     }
 
     fn contains_offset(&self, offset: u32) -> bool {
@@ -995,7 +1030,6 @@ impl ReceivedBlockTracker {
             return false;
         }
         self.bits |= bit;
-        self.count = self.count.saturating_add(1);
         true
     }
 
diff --git a/crates/zakura-network/src/zakura/block_sync/tests.rs b/crates/zakura-network/src/zakura/block_sync/tests.rs
index c113f8f403..a7ff1fd7bf 100644
--- a/crates/zakura-network/src/zakura/block_sync/tests.rs
+++ b/crates/zakura-network/src/zakura/block_sync/tests.rs
@@ -1118,6 +1118,8 @@ fn window_request(height: u32) -> OutstandingBlockRange {
     let byte = u8::try_from(height).expect("test heights fit in u8");
     let now = Instant::now();
     OutstandingBlockRange {
+        write_status: work_queue::RequestWriteStatus::written_for_tests(),
+        charged_for_liveness: true,
         request: BlockRangeRequest {
             owner: test_work_owner(),
             start_height: block::Height(height),
@@ -1142,6 +1144,8 @@ fn window_request_range(start: u32, count: u32) -> OutstandingBlockRange {
     let byte = u8::try_from(start).expect("test heights fit in u8");
     let now = Instant::now();
     OutstandingBlockRange {
+        write_status: work_queue::RequestWriteStatus::written_for_tests(),
+        charged_for_liveness: true,
         request: BlockRangeRequest {
             owner: test_work_owner(),
             start_height: block::Height(start),
@@ -1371,7 +1375,7 @@ fn view_reset_reclears_probe_streak_so_unproven_peer_can_reprobe() {
     // A destructive reset returns the peer's outstanding to the queue on our
     // initiative, then runs the reset hook.
     window.outstanding.clear();
-    window.note_view_reset();
+    window.note_locally_returned_requests();
 
     // The peer can probe again (streak below the cap) and is not left as a zombie
     // (liveness cleared, so `check_liveness` is `Ok`, and proof state is untouched).
@@ -1409,7 +1413,7 @@ fn view_reset_preserves_proof_but_reclears_streak() {
     assert_eq!(window.no_progress_request_cap(), 8);
 
     window.outstanding.clear();
-    window.note_view_reset();
+    window.note_locally_returned_requests();
 
     assert_eq!(window.requests_without_block_progress, 0);
     assert!(
@@ -3091,6 +3095,134 @@ fn floor_watchdog_skips_received_heights() {
     assert_eq!(queue.advance_floor(block::Height(1)), 0);
 }
 
+#[tokio::test]
+async fn floor_watchdog_only_avoids_requests_that_started_writing() {
+    use super::work_queue::RequestWrite;
+    use crate::zakura::transport::FrameWriteClaim;
+
+    #[derive(Clone, Copy, Debug)]
+    enum WriteState {
+        Queued,
+        Started,
+        Written,
+        Dropped,
+    }
+
+    for write_state in [
+        WriteState::Queued,
+        WriteState::Started,
+        WriteState::Written,
+        WriteState::Dropped,
+    ] {
+        let config = immediate_body_download_config();
+        let (_tip_tx, tip_rx) = watch::channel((block::Height(0), block::Hash([0; 32])));
+        let startup = BlockSyncStartup::new(
+            BlockSyncFrontiers {
+                finalized_height: block::Height(0),
+                verified_block_tip: block::Height(0),
+                verified_block_hash: block::Hash([0; 32]),
+            },
+            (block::Height(0), block::Hash([0; 32])),
+            tip_rx,
+            config.clone(),
+        );
+        let (handle, _actions, reactor_task) = spawn_block_sync_reactor(startup);
+        let wiring = handle.routine_wiring.as_ref().unwrap();
+        let now = Instant::now();
+        let mut generation = 0;
+        // Register three servable peers without routines, so only the running
+        // reactor can expire the claim and apply floor avoidance.
+        for id in 1..=3 {
+            let current = wiring
+                .registry
+                .admit_session(&peer(id), ServicePeerDirection::Outbound, &config, 0, now)
+                .generation();
+            wiring.registry.upsert_status(&peer(id), current, status());
+            if id == 1 {
+                generation = current;
+            }
+        }
+        assert_eq!(wiring.registry.floor_gap_servable(block::Height(1)).0, 3);
+        wiring.work.set_estimate_floor_for_tests(1);
+        wiring.work.extend(
+            test_work_scope(),
+            [(
+                block::Height(1),
+                block::Hash([1; 32]),
+                BlockSizeEstimate::Confirmed(100),
+            )],
+        );
+        let items = wiring.work.take_for_request(
+            block::Height(1),
+            block::Height(1),
+            1,
+            100,
+            generation,
+            std::num::NonZeroU64::new(1).unwrap(),
+        );
+        assert_eq!(items.len(), 1);
+        let owner = items[0].1.owner.unwrap();
+        assert!(wiring.budget.clone().try_reserve(100));
+        let write = RequestWrite::new(
+            owner,
+            items,
+            wiring.work.clone(),
+            wiring.budget.clone(),
+            CancellationToken::new(),
+        );
+        assert!(write.publish(|| {
+            wiring.registry.set_outstanding(
+                &peer(1),
+                generation,
+                BTreeMap::from([(
+                    block::Height(1),
+                    OutstandingMeta {
+                        owner,
+                        hash: block::Hash([1; 32]),
+                        estimated_bytes: 100,
+                        queued_at: now,
+                        deadline: now,
+                    },
+                )]),
+            );
+        }));
+        let started = matches!(write_state, WriteState::Started | WriteState::Written);
+        if started {
+            assert!(write.try_start());
+        }
+        if matches!(write_state, WriteState::Written) {
+            write.written();
+        }
+        let write_status = write.status();
+        let write = if matches!(write_state, WriteState::Dropped) {
+            drop(write);
+            None
+        } else {
+            Some(write)
+        };
+
+        await_until(
+            "watchdog settles the expired floor claim",
+            Duration::from_secs(2),
+            || wiring.registry.total_unreceived() == 0 && wiring.budget.reserved() == 0,
+        )
+        .await
+        .unwrap();
+        assert_eq!(write_status.was_skipped(), !started, "{write_state:?}");
+        assert_eq!(wiring.work.reserved_bytes(), 0);
+        assert!(wiring.work.pending_contains(block::Height(1)));
+        assert_eq!(
+            wiring
+                .registry
+                .is_floor_height_avoided(&peer(1), block::Height(1), Instant::now()),
+            started,
+            "floor avoidance must reflect the settled write: {write_state:?}",
+        );
+        reactor_task.abort();
+        drop(write);
+    }
+}
+
 #[test]
 fn late_body_does_not_resurrect_charge() {
     let queue = work_queue_with(0, [needed(1, BlockSizeEstimate::Advertised(100))]);
@@ -4067,8 +4199,15 @@ async fn block_liveness_parks_silent_peer_and_traces_reason() {
 
 #[tokio::test]
 async fn late_unowned_body_is_rejected_and_the_session_is_parked() {
-    // A body cannot count as progress after its request ownership expires.
-    // Verify both the missing submission and the local park.
+    check_cold_probe_deadline(true).await;
+}
+
+#[tokio::test]
+async fn cold_probe_can_finish_after_the_short_floor_rescue_deadline() {
+    check_cold_probe_deadline(false).await;
+}
+
+async fn check_cold_probe_deadline(expired: bool) {
     let mut config = immediate_body_download_config();
     // Short request/floor-rescue leash so the probe times out fast; the liveness
     // deadline (request_timeout * 4 = 1.2s) is what a false disconnect would trip.
@@ -4138,12 +4277,10 @@ async fn late_unowned_body_is_rejected_and_the_session_is_parked() {
     assert_eq!(start_height, block::Height(1));
     assert_eq!(count, 1);
 
-    // Let that probe time out on the floor-rescue leash: height 1 returns to the
-    // queue and, being unproven, the peer is now gated at its one-probe cap.
-    tokio::time::sleep(Duration::from_millis(200)).await;
+    // An unmeasured peer gets the normal deadline for its only probe. Deliver
+    // after the short rescue deadline, or after the normal deadline has expired.
+    tokio::time::sleep(Duration::from_millis(if expired { 500 } else { 200 })).await;
 
-    // The body arrives after retirement of its request owner.
-    // Do not submit it to the verifier or count it as timely progress.
     inbound_tx
         .send(
             BlockSyncMessage::Block(blocks[0].clone())
@@ -4153,20 +4290,27 @@ async fn late_unowned_body_is_rejected_and_the_session_is_parked() {
         .await
         .expect("late block frame queues");
 
-    assert!(
-        tokio::time::timeout(Duration::from_millis(200), async {
-            loop {
-                if matches!(
-                    next_action(&mut actions).await,
-                    BlockSyncAction::SubmitBlock { .. }
-                ) {
-                    break;
-                }
+    let submitted = tokio::time::timeout(Duration::from_millis(200), async {
+        loop {
+            if matches!(
+                next_action(&mut actions).await,
+                BlockSyncAction::SubmitBlock { .. }
+            ) {
+                break;
             }
-        })
-        .await
-        .is_err(),
-        "a completion whose request owner retired must not reach the verifier",
+        }
+    })
+    .await;
+    if !expired {
+        submitted.expect("the still-owned cold probe must reach the verifier");
+        assert_eq!(handle.peer_snapshot().outbound_peers, 1);
+        assert!(!connection_cancel.is_cancelled());
+        reactor_task.abort();
+        return;
+    }
+    assert!(
+        submitted.is_err(),
+        "a retired owner cannot reach the verifier"
     );
     await_until(
         "late unowned body does not prevent the session park",
@@ -5788,6 +5932,8 @@ fn outstanding_three_block_range(budget: &mut ByteBudget) -> OutstandingBlockRan
     assert!(budget.try_reserve(request.estimated_bytes));
     let now = Instant::now();
     OutstandingBlockRange {
+        write_status: work_queue::RequestWriteStatus::written_for_tests(),
+        charged_for_liveness: true,
         request,
         queued_at: now,
         deadline: now,
@@ -6184,6 +6330,8 @@ fn underestimated_body_is_buffered_and_releases_only_its_estimate() {
     assert!(budget.try_reserve(request.estimated_bytes));
     let now = Instant::now();
     let mut outstanding = OutstandingBlockRange {
+        write_status: work_queue::RequestWriteStatus::written_for_tests(),
+        charged_for_liveness: true,
         request,
         queued_at: now,
         deadline: now,
diff --git a/crates/zakura-network/src/zakura/block_sync/trace.rs b/crates/zakura-network/src/zakura/block_sync/trace.rs
index 15ea861268..6967f9ebc7 100644
--- a/crates/zakura-network/src/zakura/block_sync/trace.rs
+++ b/crates/zakura-network/src/zakura/block_sync/trace.rs
@@ -160,6 +160,8 @@ pub(super) struct BlockTraceFields {
     #[serde(skip_serializing_if = "Option::is_none")]
     pub returned_count: Option<u64>,
     #[serde(skip_serializing_if = "Option::is_none")]
+    pub committed_count: Option<u64>,
+    #[serde(skip_serializing_if = "Option::is_none")]
     pub already_pending_count: Option<u64>,
     #[serde(skip_serializing_if = "Option::is_none")]
     pub released_count: Option<u64>,
@@ -357,36 +359,6 @@ impl QueueSendFailedEvent {
         )
     }
 
-    /// Peer routines historically projected range fields only for GetBlocks.
-    pub(super) fn peer_routine(
-        peer: &ZakuraPeerId,
-        message: &BlockSyncMessage,
-        error: &OrderedSendError,
-        queue_capacity: usize,
-        queue_max_capacity: usize,
-    ) -> Self {
-        let message_fields = match message {
-            BlockSyncMessage::GetBlocks {
-                start_height,
-                count,
-            } => MessageFields {
-                range_start: Some(height(*start_height)),
-                range_count: Some(u64::from(*count)),
-                ..MessageFields::default()
-            },
-            _ => MessageFields::default(),
-        };
-        Self::build(
-            peer,
-            message,
-            error,
-            None,
-            queue_capacity,
-            queue_max_capacity,
-            message_fields,
-        )
-    }
-
     fn build(
         peer: &ZakuraPeerId,
         message: &BlockSyncMessage,
diff --git a/crates/zakura-network/src/zakura/block_sync/work_queue.rs b/crates/zakura-network/src/zakura/block_sync/work_queue.rs
index 181172c258..a4850b864e 100644
--- a/crates/zakura-network/src/zakura/block_sync/work_queue.rs
+++ b/crates/zakura-network/src/zakura/block_sync/work_queue.rs
@@ -21,13 +21,20 @@
 //! reservation); it exists only to carry the `SizeMismatch` tolerance check
 //! through to the reactor's receive path and request budget.
 
-use std::sync::Mutex as StdMutex;
+use std::{
+    num::NonZeroU64,
+    sync::{Arc, Mutex as StdMutex},
+};
 
 use tokio::sync::Notify;
 use zakura_chain::block;
 
 use super::{request::BlockSizeEstimate, state::BlockBudgetLedger};
 
+mod request_write;
+use request_write::RequestWriteRegistration;
+pub(super) use request_write::{RequestWrite, RequestWriteStatus};
+
 /// Lower clamp on a body-size estimate.
 pub(super) const DEFAULT_BS_SIZE_FLOOR_BYTES: u64 = 1024;
 
@@ -36,8 +43,10 @@ pub(super) const DEFAULT_BS_SIZE_FLOOR_BYTES: u64 = 1024;
 pub(super) struct WorkItem {
     /// Exact durable coordinates that authorized this body download.
     pub(super) scope: zakura_header_chain::BodyWorkAuthority,
-    /// Exact active range request, set only while reserved in flight.
+    /// Exact attempt, retained from provisional take through response receipt.
     pub(super) owner: Option<zakura_header_chain::BodyWorkOwner>,
+    /// Taken by an attempt that has not yet published its reservation and frame.
+    provisional: bool,
     /// Expected hash of the block at this height (drives the response match).
     pub(super) hash: block::Hash,
     /// The block's size estimate. Used for request budget reservation and the
@@ -50,10 +59,14 @@ pub(super) struct WorkItem {
 /// Diagnostics for an attempted `in_flight -> pending` retry transition.
 #[derive(Copy, Clone, Debug, Default, Eq, PartialEq)]
 pub(super) struct WorkReturnOutcome {
-    /// Reserved bytes released while moving items back to `pending`.
+    /// This owner's frame was skipped before writing, atomically with settlement.
+    pub(super) request_was_unwritten: bool,
+    /// Reserved bytes released while returning or discarding items.
     pub(super) released_bytes: u64,
     /// Reserved items successfully moved back to `pending`.
     pub(super) returned_count: u64,
+    /// Owned heights discarded because they are already committed.
+    pub(super) committed_count: u64,
     /// Requested heights that were already back in `pending`.
     pub(super) already_pending_count: u64,
     /// Received items still present in `in_flight` with a `Released` ledger.
@@ -78,12 +91,27 @@ struct WorkQueueInner {
     /// item, maintained incrementally at each ledger transition so
     /// [`WorkQueue::reserved_bytes`]
     reserved_bytes: u64,
+    /// Weak owners allow reset notification without retaining session resources.
+    /// Their status survives until writing or terminal cleanup finishes.
+    request_writes:
+        std::collections::HashMap<zakura_header_chain::BodyWorkOwner, RequestWriteRegistration>,
 }
 
 impl WorkQueueInner {
     fn estimate_bytes(&self, estimate: BlockSizeEstimate) -> u64 {
         estimate_bytes_with(estimate, self.floor_estimate_bytes)
     }
+
+    fn owner_for_height(
+        &self,
+        height: block::Height,
+    ) -> Option<zakura_header_chain::BodyWorkOwner> {
+        self.pending
+            .get(&height)
+            .or_else(|| self.in_flight.get(&height))
+            .filter(|item| !item.provisional)
+            .and_then(|item| item.owner)
+    }
 }
 
 /// Compute a clamped body-size estimate from a [`BlockSizeEstimate`] hint.
@@ -115,6 +143,7 @@ impl WorkQueue {
                 current_authority: None,
                 floor_estimate_bytes: DEFAULT_BS_SIZE_FLOOR_BYTES,
                 reserved_bytes: 0,
+                request_writes: std::collections::HashMap::new(),
             }),
             available: Notify::new(),
         }
@@ -158,6 +187,7 @@ impl WorkQueue {
                     WorkItem {
                         scope,
                         owner: None,
+                        provisional: false,
                         hash,
                         estimated_bytes,
                         budget: BlockBudgetLedger::Released,
@@ -241,12 +271,45 @@ impl WorkQueue {
     /// The estimate cap bounds the request's summed byte reservation. To
     /// guarantee progress, the first eligible item is always taken when
     /// `max_count > 0`, even if its estimate alone exceeds the cap.
+    #[cfg(test)]
     pub(super) fn take_in_range_budgeted(
         &self,
         low: block::Height,
         high: block::Height,
         max_count: usize,
         max_estimated_bytes: u64,
+    ) -> Vec<(block::Height, WorkItem)> {
+        self.take_budgeted(low, high, max_count, max_estimated_bytes, None)
+    }
+
+    /// Give the provisional take an exact owner before releasing the queue lock.
+    /// A reset followed by a new take cannot be undone by this attempt's cleanup.
+    #[allow(clippy::too_many_arguments)]
+    pub(super) fn take_for_request(
+        &self,
+        low: block::Height,
+        high: block::Height,
+        max_count: usize,
+        max_estimated_bytes: u64,
+        session_id: u64,
+        request_id: NonZeroU64,
+    ) -> Vec<(block::Height, WorkItem)> {
+        self.take_budgeted(
+            low,
+            high,
+            max_count,
+            max_estimated_bytes,
+            Some((session_id, request_id)),
+        )
+    }
+
+    fn take_budgeted(
+        &self,
+        low: block::Height,
+        high: block::Height,
+        max_count: usize,
+        max_estimated_bytes: u64,
+        attempt: Option<(u64, NonZeroU64)>,
     ) -> Vec<(block::Height, WorkItem)> {
         // An empty count or inverted range is a caller bug, not a real "nothing to
         // take": every caller computes `low <= high` and a positive count before
@@ -297,6 +360,10 @@ impl WorkQueue {
                 item.scope = authority;
             }
         }
+        for (_, item) in &mut taken {
+            item.owner = attempt.map(|(session, request)| item.scope.bind(session, request));
+            item.provisional = attempt.is_some();
+        }
         for (height, item) in &taken {
             inner.pending.remove(height);
             inner.in_flight.insert(*height, *item);
@@ -333,12 +400,18 @@ impl WorkQueue {
     /// freshly-registered `available` future and busy-loop the want-work arm
     /// (a self-wake spin); other peers were already woken by the original failure
     /// `return_items`, so suppressing the notify only affects the caller.
-    pub(super) fn return_items_quiet(&self, heights: impl IntoIterator<Item = block::Height>) {
+    pub(super) fn return_unpublished(&self, items: &[(block::Height, WorkItem)]) {
         let mut inner = self.lock();
-        for height in heights {
-            if let Some(mut item) = inner.in_flight.remove(&height) {
+        for (height, taken) in items {
+            if !inner.in_flight.get(height).is_some_and(|item| {
+                item.owner == taken.owner && item.owner.is_some() && item.provisional
+            }) {
+                continue;
+            }
+            if let Some(mut item) = inner.in_flight.remove(height) {
                 item.owner = None;
-                inner.pending.insert(height, item);
+                item.provisional = false;
+                inner.pending.insert(*height, item);
             }
         }
     }
@@ -352,6 +425,7 @@ impl WorkQueue {
         self.mark_reserved_matching(None, heights)
     }
 
+    #[cfg(test)]
     pub(super) fn mark_reserved_for_owner(
         &self,
         owner: zakura_header_chain::BodyWorkOwner,
@@ -360,6 +434,7 @@ impl WorkQueue {
         self.mark_reserved_matching(Some(owner), heights)
     }
 
+    #[cfg(test)]
     fn mark_reserved_matching(
         &self,
         owner: Option<zakura_header_chain::BodyWorkOwner>,
@@ -393,6 +468,8 @@ impl WorkQueue {
         self.release_active_reserved_height_matching(None, height)
     }
 
+    /// End the receipt reservation and retire any queued request it invalidates.
+    /// Returns all released bytes, including the request's other unsent heights.
     pub(super) fn release_active_reserved_height_for_owner(
         &self,
         owner: zakura_header_chain::BodyWorkOwner,
@@ -406,18 +483,23 @@ impl WorkQueue {
         owner: Option<zakura_header_chain::BodyWorkOwner>,
         height: block::Height,
     ) -> Option<u64> {
-        let mut inner = self.lock();
-        let released = {
-            let item = inner.in_flight.get_mut(&height)?;
-            if owner.is_some_and(|owner| item.owner != Some(owner)) {
-                return None;
-            }
-            if !item.budget.is_reserved() {
-                return None;
-            }
-            item.budget.release_reserved()
+        let (released, claim) = {
+            let mut inner = self.lock();
+            let (released, owner) = {
+                let item = inner.in_flight.get_mut(&height)?;
+                if owner.is_some_and(|owner| item.owner != Some(owner)) {
+                    return None;
+                }
+                if !item.budget.is_reserved() {
+                    return None;
+                }
+                (item.budget.release_reserved(), item.owner)
+            };
+            inner.reserved_bytes = inner.reserved_bytes.saturating_sub(released);
+            let (outcome, claim) = self.return_items_locked(&mut inner, owner, []);
+            (released.saturating_add(outcome.released_bytes), claim)
         };
-        inner.reserved_bytes = inner.reserved_bytes.saturating_sub(released);
+        drop(claim);
         Some(released)
     }
 
@@ -427,6 +509,8 @@ impl WorkQueue {
         self.claim_received_matching(None, height)
     }
 
+    /// Claim a received height and return its reservation plus any unsent bytes
+    /// released by retiring the queued request.
     pub(super) fn claim_received_for_owner(
         &self,
         owner: zakura_header_chain::BodyWorkOwner,
@@ -440,26 +524,31 @@ impl WorkQueue {
         owner: Option<zakura_header_chain::BodyWorkOwner>,
         height: block::Height,
     ) -> u64 {
-        let mut inner = self.lock();
-        if let Some(item) = inner.in_flight.get_mut(&height) {
-            if owner.is_some_and(|owner| item.owner != Some(owner)) {
-                return 0;
-            }
-            let released = item.budget.release_reserved();
-            inner.reserved_bytes = inner.reserved_bytes.saturating_sub(released);
-            return released;
-        }
-        if let Some(mut item) = inner.pending.remove(&height) {
-            if owner.is_some_and(|owner| item.owner != Some(owner)) {
-                inner.pending.insert(height, item);
+        let (released, claim) = {
+            let mut inner = self.lock();
+            let (released, owner) = if let Some(item) = inner.in_flight.get_mut(&height) {
+                if owner.is_some_and(|owner| item.owner != Some(owner)) {
+                    return 0;
+                }
+                (item.budget.release_reserved(), item.owner)
+            } else if let Some(mut item) = inner.pending.remove(&height) {
+                if owner.is_some_and(|owner| item.owner != Some(owner)) {
+                    inner.pending.insert(height, item);
+                    return 0;
+                }
+                let released = item.budget.release_reserved();
+                let owner = item.owner;
+                inner.in_flight.insert(height, item);
+                (released, owner)
+            } else {
                 return 0;
-            }
-            let released = item.budget.release_reserved();
-            inner.in_flight.insert(height, item);
+            };
             inner.reserved_bytes = inner.reserved_bytes.saturating_sub(released);
-            return released;
-        }
-        0
+            let (outcome, claim) = self.return_items_locked(&mut inner, owner, []);
+            (released.saturating_add(outcome.released_bytes), claim)
+        };
+        drop(claim);
+        released
     }
 
     /// Release active request reservations, leaving received heights in place.
@@ -549,6 +638,8 @@ impl WorkQueue {
         self.release_reserved_and_return_items_detailed_matching(None, heights)
     }
 
+    /// Return this owner's unreceived heights. Expiring a queued frame also
+    /// returns every other unsent height owned by that request under the same lock.
     pub(super) fn release_reserved_and_return_items_detailed_for_owner(
         &self,
         owner: zakura_header_chain::BodyWorkOwner,
@@ -562,59 +653,97 @@ impl WorkQueue {
         owner: Option<zakura_header_chain::BodyWorkOwner>,
         heights: impl IntoIterator<Item = block::Height>,
     ) -> WorkReturnOutcome {
-        let mut moved = false;
-        let mut outcome = WorkReturnOutcome::default();
-        {
+        let (outcome, claim) = {
             let mut inner = self.lock();
-            for height in heights {
-                outcome.min_height = Some(
-                    outcome
-                        .min_height
-                        .map_or(height, |current| current.min(height)),
-                );
-                outcome.max_height = Some(
-                    outcome
-                        .max_height
-                        .map_or(height, |current| current.max(height)),
-                );
-                let Some(item) = inner.in_flight.get(&height) else {
-                    if inner.pending.contains_key(&height) {
-                        outcome.already_pending_count =
-                            outcome.already_pending_count.saturating_add(1);
-                    } else {
-                        outcome.missing_count = outcome.missing_count.saturating_add(1);
-                    }
-                    continue;
-                };
-                if owner.is_some_and(|owner| item.owner != Some(owner)) {
-                    outcome.missing_count = outcome.missing_count.saturating_add(1);
-                    continue;
+            self.return_items_locked(&mut inner, owner, heights)
+        };
+        drop(claim);
+        outcome
+    }
+
+    /// Settle requested heights and any queued request they retire. The caller
+    /// must retain the returned claim until after releasing the queue lock.
+    fn return_items_locked(
+        &self,
+        inner: &mut WorkQueueInner,
+        owner: Option<zakura_header_chain::BodyWorkOwner>,
+        heights: impl IntoIterator<Item = block::Height>,
+    ) -> (WorkReturnOutcome, Option<Arc<RequestWrite>>) {
+        let mut outcome = WorkReturnOutcome::default();
+        let mut heights: std::collections::BTreeSet<_> = heights.into_iter().collect();
+        let registration = owner.and_then(|owner| inner.request_writes.get(&owner));
+        let claim = registration.and_then(|registration| registration.claim.upgrade());
+        if let Some(registration) = registration {
+            // The writer claims under this same lock. Expiry skips an
+            // unwritten frame even while its last owner is being dropped.
+            if registration.status.expire_unwritten() {
+                self.available.notify_waiters();
+            }
+            outcome.request_was_unwritten = registration.status.was_skipped();
+            if outcome.request_was_unwritten {
+                if let Some(claim) = &claim {
+                    // Skipping an unsent frame retires its whole request.
+                    heights.extend(claim.heights());
                 }
-                match item.budget {
-                    BlockBudgetLedger::Released => {
-                        outcome.released_count = outcome.released_count.saturating_add(1);
-                        continue;
-                    }
-                    BlockBudgetLedger::Reserved(_) => {}
+            }
+        }
+        for height in heights {
+            outcome.min_height = Some(
+                outcome
+                    .min_height
+                    .map_or(height, |current| current.min(height)),
+            );
+            outcome.max_height = Some(
+                outcome
+                    .max_height
+                    .map_or(height, |current| current.max(height)),
+            );
+            let Some(item) = inner.in_flight.get(&height) else {
+                if inner.pending.contains_key(&height) {
+                    outcome.already_pending_count = outcome.already_pending_count.saturating_add(1);
+                } else {
+                    outcome.missing_count = outcome.missing_count.saturating_add(1);
                 }
+                continue;
+            };
+            if owner.is_some_and(|owner| item.owner != Some(owner)) {
+                outcome.missing_count = outcome.missing_count.saturating_add(1);
+                continue;
+            }
+            if height <= inner.floor {
                 let mut item = inner
                     .in_flight
                     .remove(&height)
-                    .expect("reserved item exists because it was just checked");
+                    .expect("owned item exists because it was just checked");
                 outcome.released_bytes = outcome
                     .released_bytes
                     .saturating_add(item.budget.release_reserved());
-                item.owner = None;
-                outcome.returned_count = outcome.returned_count.saturating_add(1);
-                inner.pending.insert(height, item);
-                moved = true;
+                outcome.committed_count = outcome.committed_count.saturating_add(1);
+                continue;
+            }
+            match item.budget {
+                BlockBudgetLedger::Released => {
+                    outcome.released_count = outcome.released_count.saturating_add(1);
+                    continue;
+                }
+                BlockBudgetLedger::Reserved(_) => {}
             }
-            inner.reserved_bytes = inner.reserved_bytes.saturating_sub(outcome.released_bytes);
+            let mut item = inner
+                .in_flight
+                .remove(&height)
+                .expect("reserved item exists because it was just checked");
+            outcome.released_bytes = outcome
+                .released_bytes
+                .saturating_add(item.budget.release_reserved());
+            item.owner = None;
+            outcome.returned_count = outcome.returned_count.saturating_add(1);
+            inner.pending.insert(height, item);
         }
-        if moved {
+        inner.reserved_bytes = inner.reserved_bytes.saturating_sub(outcome.released_bytes);
+        if outcome.returned_count > 0 {
             self.available.notify_waiters();
         }
-        outcome
+        (outcome, claim)
     }
 
     /// Garbage-collect committed heights: raise the floor to `max(self.floor,
@@ -664,6 +793,17 @@ impl WorkQueue {
     /// in [`advance_floor`](Self::advance_floor).
     pub(super) fn reset_above(&self, floor: block::Height) -> u64 {
         let mut inner = self.lock();
+        // Retain upgraded owners until after unlocking: a concurrently dropped
+        // queue entry can make this the last reference, whose Drop settles work.
+        let claims: Vec<_> = inner
+            .request_writes
+            .values()
+            .filter_map(|registration| registration.claim.upgrade())
+            .collect();
+        for claim in claims.iter().filter(|claim| claim.has_height_above(floor)) {
+            claim.reset();
+            inner.request_writes.remove(&claim.owner());
+        }
         inner.floor = floor;
         // Pop only the `> floor` suffix from each map (O(removed · log n)); see the
         // note in `advance_floor` on why a full-map `retain` is too expensive here.
@@ -689,6 +829,8 @@ impl WorkQueue {
             released = released.saturating_add(item.budget.release_reserved());
         }
         inner.reserved_bytes = inner.reserved_bytes.saturating_sub(released);
+        drop(inner);
+        drop(claims);
         released
     }
 
@@ -874,12 +1016,18 @@ impl WorkQueue {
         &self,
         height: block::Height,
     ) -> Option<zakura_header_chain::BodyWorkOwner> {
+        self.lock().owner_for_height(height)
+    }
+
+    /// Filter height metadata against current owners under one queue lock.
+    /// `Copy` prevents removed values from dropping request resources under the lock.
+    pub(super) fn retain_owned<T: Copy>(
+        &self,
+        entries: &mut Vec<(block::Height, T)>,
+        owner_of: impl Fn(&T) -> zakura_header_chain::BodyWorkOwner,
+    ) {
         let inner = self.lock();
-        inner
-            .pending
-            .get(&height)
-            .or_else(|| inner.in_flight.get(&height))
-            .and_then(|item| item.owner)
+        entries.retain(|(height, entry)| inner.owner_for_height(*height) == Some(owner_of(entry)));
     }
 
     pub(super) fn pending_contains(&self, height: block::Height) -> bool {
diff --git a/crates/zakura-network/src/zakura/block_sync/work_queue/request_write.rs b/crates/zakura-network/src/zakura/block_sync/work_queue/request_write.rs
new file mode 100644
index 0000000000..6516112d62
--- /dev/null
+++ b/crates/zakura-network/src/zakura/block_sync/work_queue/request_write.rs
@@ -0,0 +1,224 @@
+//! Ownership from provisional download reservation through the request write.
+
+use std::sync::{
+    atomic::{AtomicU8, Ordering},
+    Arc, Weak,
+};
+
+use tokio_util::sync::CancellationToken;
+use zakura_header_chain::BodyWorkOwner;
+
+use super::{block, BlockBudgetLedger, WorkItem, WorkQueue};
+use crate::zakura::transport::{ByteBudget, FrameWriteClaim};
+
+const UNPUBLISHED: u8 = 0;
+const QUEUED: u8 = 1;
+const STARTED: u8 = 2;
+const WRITTEN: u8 = 3;
+const EXPIRED: u8 = 4;
+
+#[cfg(test)]
+mod tests;
+
+/// Observes whether transport skipped an attempt without retaining its work or
+/// byte reservation. The routine can keep this after the writer drops its claim.
+#[derive(Clone, Debug)]
+pub(in crate::zakura::block_sync) struct RequestWriteStatus(Arc<AtomicU8>);
+
+impl RequestWriteStatus {
+    pub(in crate::zakura::block_sync) fn was_skipped(&self) -> bool {
+        self.0.load(Ordering::Acquire) == EXPIRED
+    }
+
+    /// Retire a queued attempt atomically against writer startup. Returns true
+    /// only for the transition; a started frame remains the transport's owner.
+    pub(in crate::zakura::block_sync) fn expire_unwritten(&self) -> bool {
+        self.0
+            .compare_exchange(QUEUED, EXPIRED, Ordering::AcqRel, Ordering::Acquire)
+            .is_ok()
+    }
+}
+
+/// Keep the write disposition readable while the last owner is being dropped.
+#[derive(Debug)]
+pub(super) struct RequestWriteRegistration {
+    pub(super) claim: Weak<RequestWrite>,
+    pub(super) status: RequestWriteStatus,
+}
+
+/// One exact attempt owns its provisional budget until publication. Afterwards
+/// the work ledger arbitrates every release, including reset and response arrival.
+#[derive(Debug)]
+pub(crate) struct RequestWrite {
+    owner: BodyWorkOwner,
+    items: Vec<(block::Height, WorkItem)>,
+    estimated_bytes: u64,
+    work: Arc<WorkQueue>,
+    budget: ByteBudget,
+    cancel: CancellationToken,
+    state: Arc<AtomicU8>,
+}
+
+impl RequestWrite {
+    pub(in crate::zakura::block_sync) fn new(
+        owner: BodyWorkOwner,
+        items: Vec<(block::Height, WorkItem)>,
+        work: Arc<WorkQueue>,
+        budget: ByteBudget,
+        cancel: CancellationToken,
+    ) -> Arc<Self> {
+        let estimated_bytes = items.iter().map(|(_, item)| item.estimated_bytes).sum();
+        Arc::new(Self {
+            owner,
+            items,
+            estimated_bytes,
+            work,
+            budget,
+            cancel,
+            state: Arc::new(AtomicU8::new(UNPUBLISHED)),
+        })
+    }
+
+    pub(in crate::zakura::block_sync) fn status(&self) -> RequestWriteStatus {
+        RequestWriteStatus(self.state.clone())
+    }
+
+    /// Record outstanding state and enqueue into already-reserved capacity under
+    /// the same lock used by reset and writer claim. The caller retains this Arc
+    /// until return, even if a closed queue immediately discards its copy.
+    pub(in crate::zakura::block_sync) fn publish(self: &Arc<Self>, publish: impl FnOnce()) -> bool {
+        let mut inner = self.work.lock();
+        if self.cancel.is_cancelled()
+            || self.items.is_empty()
+            || !self.items.iter().all(|(height, taken)| {
+                inner.in_flight.get(height).is_some_and(|item| {
+                    item.owner == Some(self.owner) && item.hash == taken.hash && item.provisional
+                })
+            })
+        {
+            return false;
+        }
+        assert_eq!(
+            self.state.load(Ordering::Acquire),
+            UNPUBLISHED,
+            "a request is published once"
+        );
+        for (height, _) in &self.items {
+            let item = inner
+                .in_flight
+                .get_mut(height)
+                .expect("every provisional item was checked under this lock");
+            item.budget = BlockBudgetLedger::reserved(item.estimated_bytes);
+            item.provisional = false;
+        }
+        inner.reserved_bytes = inner.reserved_bytes.saturating_add(self.estimated_bytes);
+        inner.request_writes.insert(
+            self.owner,
+            RequestWriteRegistration {
+                claim: Arc::downgrade(self),
+                status: self.status(),
+            },
+        );
+        self.state.store(QUEUED, Ordering::Release);
+        publish();
+        true
+    }
+
+    pub(super) fn owner(&self) -> BodyWorkOwner {
+        self.owner
+    }
+
+    pub(super) fn heights(&self) -> impl Iterator<Item = block::Height> + '_ {
+        self.items.iter().map(|(height, _)| *height)
+    }
+
+    /// A reserved queue slot lost its receiver during publication. Settle the
+    /// ledger now even if that closed channel retains a copy of this claim.
+    pub(in crate::zakura::block_sync) fn delivery_failed(&self) {
+        self.cancel.cancel();
+        self.expire_unwritten();
+        let released = self
+            .work
+            .release_reserved_and_return_items_detailed_for_owner(
+                self.owner,
+                self.items.iter().map(|(height, _)| *height),
+            );
+        self.budget.clone().release(released.released_bytes);
+    }
+
+    pub(super) fn has_height_above(&self, floor: block::Height) -> bool {
+        #[cfg(test)]
+        tests::before_reset_height_check();
+        self.items.last().is_some_and(|(height, _)| *height > floor)
+    }
+
+    pub(super) fn expire_unwritten(&self) {
+        if self.status().expire_unwritten() {
+            // A competing body can settle every byte without returning work.
+            // Wake the routine even when claim cleanup has no budget to release.
+            self.work.available.notify_waiters();
+        }
+    }
+
+    pub(super) fn reset(&self) {
+        self.expire_unwritten();
+        if self.state.load(Ordering::Acquire) == STARTED {
+            // The prefix already belongs to this ordered stream. Cancelling the
+            // session resets the pair; no later request may follow that prefix.
+            self.cancel.cancel();
+        }
+    }
+}
+
+impl FrameWriteClaim for RequestWrite {
+    fn try_start(&self) -> bool {
+        let inner = self.work.lock();
+        let current = !self.cancel.is_cancelled()
+            && self.items.iter().all(|(height, _)| {
+                inner
+                    .in_flight
+                    .get(height)
+                    .is_some_and(|item| item.owner == Some(self.owner) && item.budget.is_reserved())
+            });
+        if !current {
+            self.expire_unwritten();
+            return false;
+        }
+        self.state
+            .compare_exchange(QUEUED, STARTED, Ordering::AcqRel, Ordering::Acquire)
+            .is_ok()
+    }
+
+    fn written(&self) {
+        self.state.store(WRITTEN, Ordering::Release);
+        self.work.lock().request_writes.remove(&self.owner);
+    }
+}
+
+impl Drop for RequestWrite {
+    fn drop(&mut self) {
+        match self.state.load(Ordering::Acquire) {
+            UNPUBLISHED => {
+                self.work.return_unpublished(&self.items);
+                self.budget.release(self.estimated_bytes);
+            }
+            WRITTEN => {}
+            state => {
+                #[cfg(test)]
+                tests::before_drop_cleanup();
+                self.expire_unwritten();
+                if state == STARTED {
+                    self.cancel.cancel();
+                }
+                let released = self
+                    .work
+                    .release_reserved_and_return_items_detailed_for_owner(
+                        self.owner,
+                        self.items.iter().map(|(height, _)| *height),
+                    );
+                self.budget.release(released.released_bytes);
+                self.work.lock().request_writes.remove(&self.owner);
+            }
+        }
+    }
+}
diff --git a/crates/zakura-network/src/zakura/block_sync/work_queue/request_write/tests.rs b/crates/zakura-network/src/zakura/block_sync/work_queue/request_write/tests.rs
new file mode 100644
index 0000000000..365b8c74bb
--- /dev/null
+++ b/crates/zakura-network/src/zakura/block_sync/work_queue/request_write/tests.rs
@@ -0,0 +1,712 @@
+use super::*;
+use crate::zakura::{
+    block_sync::{test_work_scope, BlockSizeEstimate},
+    transport::worker_framed_channel,
+    Frame, FramedSend,
+};
+use std::{num::NonZeroU64, time::Duration};
+
+impl RequestWriteStatus {
+    pub(in crate::zakura::block_sync) fn written_for_tests() -> Self {
+        Self(Arc::new(AtomicU8::new(WRITTEN)))
+    }
+}
+
+thread_local! {
+    static BEFORE_RESET_HEIGHT_CHECK: std::cell::RefCell<Option<Box<dyn FnOnce()>>> =
+        std::cell::RefCell::new(None);
+    static BEFORE_DROP_CLEANUP: std::cell::RefCell<Option<Box<dyn FnOnce()>>> =
+        std::cell::RefCell::new(None);
+}
+
+pub(super) fn before_reset_height_check() {
+    if let Some(hook) = BEFORE_RESET_HEIGHT_CHECK.with_borrow_mut(Option::take) {
+        hook();
+    }
+}
+
+pub(super) fn before_drop_cleanup() {
+    if let Some(hook) = BEFORE_DROP_CLEANUP.with_borrow_mut(Option::take) {
+        hook();
+    }
+}
+
+#[test]
+fn settlement_observes_write_status_while_last_owner_is_dropping() {
+    for started in [false, true] {
+        let mut f = Fixture::new();
+        let claim = f.take(1);
+        assert!(claim.publish(|| {}));
+        if started {
+            assert!(claim.try_start());
+        }
+        let weak = Arc::downgrade(&claim);
+        let owner = claim.owner();
+        let work = f.work.clone();
+        let mut budget = f.budget.clone();
+        // Arbitrate expiry after the last strong reference is gone, before Drop
+        // acquires the work lock to return the request's remaining heights.
+        BEFORE_DROP_CLEANUP.with_borrow_mut(|hook| {
+            *hook = Some(Box::new(move || {
+                assert!(weak.upgrade().is_none());
+                let outcome = work.release_reserved_and_return_items_detailed_for_owner(
+                    owner,
+                    [block::Height(1)],
+                );
+                assert_eq!(outcome.request_was_unwritten, !started);
+                assert_eq!(outcome.returned_count, 1);
+                assert_eq!(outcome.released_bytes, 100);
+                budget.release(outcome.released_bytes);
+            }));
+        });
+        drop(claim);
+        assert_eq!(f.budget.reserved(), 0);
+        assert_eq!(f.work.reserved_bytes(), 0);
+        assert_eq!(f.work.pending_len(), 2);
+        assert!(f.work.lock().request_writes.is_empty());
+    }
+}
+
+#[test]
+fn reset_drops_unaffected_last_claim_after_unlocking() {
+    let (done_tx, done_rx) = std::sync::mpsc::channel();
+    let reset = std::thread::spawn(move || {
+        let mut f = Fixture::new();
+        let claim = f.take(1);
+        assert!(claim.publish(|| {}));
+        // Release the writer's reference after reset upgrades the weak reference,
+        // but before it rejects this claim at or below the new floor.
+        BEFORE_RESET_HEIGHT_CHECK.with_borrow_mut(|hook| *hook = Some(Box::new(|| drop(claim))));
+        assert_eq!(f.work.reset_above(block::Height(2)), 0);
+        assert_eq!(f.budget.reserved(), 0);
+        assert_eq!(f.work.reserved_bytes(), 0);
+        assert_eq!(f.work.pending_len(), 0);
+        assert_eq!(f.work.in_flight_len(), 0);
+        assert!(!f.cancel.is_cancelled());
+        done_tx.send(()).unwrap();
+    });
+    done_rx
+        .recv_timeout(Duration::from_secs(2))
+        .expect("dropping the last unaffected claim must not deadlock reset");
+    reset.join().unwrap();
+}
+
+struct Fixture {
+    work: Arc<WorkQueue>,
+    budget: ByteBudget,
+    cancel: CancellationToken,
+}
+
+impl Fixture {
+    fn new() -> Self {
+        let fixture = Self {
+            work: Arc::new(WorkQueue::new(block::Height(0))),
+            budget: ByteBudget::new(1000),
+            cancel: CancellationToken::new(),
+        };
+        fixture.work.set_estimate_floor_for_tests(1);
+        fixture.refill();
+        fixture
+    }
+
+    fn refill(&self) {
+        self.work.extend(
+            test_work_scope(),
+            [
+                (
+                    block::Height(1),
+                    block::Hash([1; 32]),
+                    BlockSizeEstimate::Confirmed(100),
+                ),
+                (
+                    block::Height(2),
+                    block::Hash([2; 32]),
+                    BlockSizeEstimate::Confirmed(100),
+                ),
+            ],
+        );
+    }
+
+    fn take(&mut self, id: u64) -> Arc<RequestWrite> {
+        let items = self.work.take_for_request(
+            block::Height(1),
+            block::Height(2),
+            2,
+            200,
+            7,
+            NonZeroU64::new(id).unwrap(),
+        );
+        assert_eq!(items.len(), 2);
+        assert!(self.budget.try_reserve(200));
+        RequestWrite::new(
+            items[0].1.owner.unwrap(),
+            items,
+            self.work.clone(),
+            self.budget.clone(),
+            self.cancel.clone(),
+        )
+    }
+
+    fn expire(&mut self, owner: BodyWorkOwner) {
+        let outcome = self
+            .work
+            .release_reserved_and_return_items_detailed_for_owner(
+                owner,
+                [block::Height(1), block::Height(2)],
+            );
+        self.budget.release(outcome.released_bytes);
+    }
+
+    fn reset(&mut self) {
+        self.budget.release(self.work.reset_above(block::Height(0)));
+    }
+}
+
+#[test]
+fn batched_ownership_filter_follows_publication_receipt_and_replacement() {
+    let mut f = Fixture::new();
+    let claim = f.take(1);
+    let owner = claim.owner();
+    let candidates = |owner| {
+        vec![
+            (block::Height(1), (owner, 11)),
+            (block::Height(2), (owner, 22)),
+            (block::Height(3), (owner, 33)),
+        ]
+    };
+    let work = f.work.clone();
+    let retained = |mut entries: Vec<(_, (BodyWorkOwner, u8))>| {
+        work.retain_owned(&mut entries, |(owner, _)| *owner);
+        entries
+    };
+
+    assert!(retained(candidates(owner)).is_empty(), "unpublished work");
+    assert!(claim.publish(|| {}));
+    assert_eq!(
+        retained(candidates(owner)),
+        vec![
+            (block::Height(1), (owner, 11)),
+            (block::Height(2), (owner, 22)),
+        ],
+    );
+
+    f.budget.release(
+        f.work
+            .release_active_reserved_height_for_owner(owner, block::Height(1))
+            .unwrap(),
+    );
+    assert_eq!(
+        retained(candidates(owner)),
+        vec![(block::Height(1), (owner, 11))],
+        "receipt retains its owner while the unsent height returns to pending",
+    );
+
+    f.reset();
+    f.refill();
+    assert!(retained(candidates(owner)).is_empty(), "reset work");
+    let replacement = f.take(2);
+    assert!(retained(candidates(replacement.owner())).is_empty());
+    assert!(replacement.publish(|| {}));
+    let mut mixed = candidates(owner);
+    mixed.insert(0, (block::Height(2), (replacement.owner(), 22)));
+    assert_eq!(
+        retained(mixed),
+        vec![(block::Height(2), (replacement.owner(), 22))],
+        "a later stale entry for the same height must not hide its current owner",
+    );
+}
+
+fn publish(claim: &Arc<RequestWrite>, sender: &FramedSend) {
+    let slot = sender.try_reserve_guarded().unwrap();
+    assert!(claim.publish(|| {
+        assert!(slot.send_request(
+            Frame {
+                message_type: 2,
+                flags: 0,
+                payload: vec![0; 9]
+            },
+            claim.clone()
+        ));
+    }));
+}
+
+#[tokio::test]
+async fn expiry_before_writer_claim_skips_the_entire_frame() {
+    let mut f = Fixture::new();
+    let (sender, mut receiver) = worker_framed_channel(1);
+    let claim = f.take(1);
+    let status = claim.status();
+    publish(&claim, &sender);
+    f.expire(claim.owner());
+    drop(claim);
+    receiver
+        .recv()
+        .await
+        .unwrap()
+        .write_with(|_| async {
+            panic!("an expired unwritten request must never reach QUIC");
+            #[allow(unreachable_code)]
+            Ok::<_, ()>(())
+        })
+        .await
+        .unwrap();
+    assert!(status.was_skipped());
+    assert_eq!(f.budget.reserved(), 0);
+    assert_eq!(f.work.pending_len(), 2);
+    assert!(!f.cancel.is_cancelled());
+}
+
+#[tokio::test]
+async fn started_request_finishes_after_expiry_without_releasing_its_replacement() {
+    let mut f = Fixture::new();
+    let (sender, mut receiver) = worker_framed_channel(1);
+    let claim = f.take(1);
+    let status = claim.status();
+    let owner = claim.owner();
+    publish(&claim, &sender);
+    drop(claim);
+    let queued = receiver.recv().await.unwrap();
+    let (started_tx, started_rx) = tokio::sync::oneshot::channel();
+    let (finish_tx, finish_rx) = tokio::sync::oneshot::channel();
+    let writer = tokio::spawn(queued.write_with(|_| async {
+        started_tx.send(()).unwrap();
+        finish_rx.await.unwrap();
+        Ok::<_, ()>(())
+    }));
+    tokio::time::timeout(Duration::from_secs(1), started_rx)
+        .await
+        .unwrap()
+        .unwrap();
+    f.expire(owner);
+    assert!(
+        !status.was_skipped(),
+        "an already-started request is still written"
+    );
+    assert_eq!(f.budget.reserved(), 0);
+    let replacement = f.take(2);
+    publish(&replacement, &sender);
+    assert!(
+        !f.cancel.is_cancelled(),
+        "expiry alone lets the started frame finish"
+    );
+    finish_tx.send(()).unwrap();
+    tokio::time::timeout(Duration::from_secs(1), writer)
+        .await
+        .unwrap()
+        .unwrap()
+        .unwrap();
+    assert_eq!(f.budget.reserved(), 200);
+    assert_eq!(
+        f.work.owner_for_height(block::Height(1)),
+        Some(replacement.owner())
+    );
+    drop(receiver);
+    drop(replacement);
+    assert_eq!(f.budget.reserved(), 0);
+}
+
+#[tokio::test]
+async fn aborting_a_partial_write_cancels_the_session_and_returns_work() {
+    let mut f = Fixture::new();
+    let (sender, mut receiver) = worker_framed_channel(1);
+    let claim = f.take(1);
+    publish(&claim, &sender);
+    drop(claim);
+    let queued = receiver.recv().await.unwrap();
+    let (started_tx, started_rx) = tokio::sync::oneshot::channel();
+    let writer = tokio::spawn(queued.write_with(|_| async {
+        started_tx.send(()).unwrap();
+        std::future::pending::<Result<(), ()>>().await
+    }));
+    tokio::time::timeout(Duration::from_secs(1), started_rx)
+        .await
+        .unwrap()
+        .unwrap();
+    writer.abort();
+    assert!(writer.await.unwrap_err().is_cancelled());
+    assert!(f.cancel.is_cancelled());
+    assert_eq!(f.budget.reserved(), 0);
+    assert_eq!(f.work.pending_len(), 2);
+}
+
+#[test]
+fn reset_before_publication_cannot_return_a_replacement_take() {
+    let mut f = Fixture::new();
+    let old = f.take(1);
+    f.reset();
+    f.refill();
+    let replacement = f.take(2);
+    assert!(!old.publish(|| panic!("reset invalidated this take")));
+    drop(old);
+    assert_eq!(f.budget.reserved(), 200);
+    assert_eq!(f.work.in_flight_len(), 2);
+    let (sender, receiver) = worker_framed_channel(1);
+    publish(&replacement, &sender);
+    assert_eq!(
+        f.work.owner_for_height(block::Height(1)),
+        Some(replacement.owner())
+    );
+    drop(receiver);
+    drop(replacement);
+    assert_eq!(f.budget.reserved(), 0);
+}
+
+#[test]
+fn reset_cancels_a_started_request_but_only_skips_a_queued_request() {
+    for started in [false, true] {
+        let mut f = Fixture::new();
+        let (sender, receiver) = worker_framed_channel(1);
+        let claim = f.take(1);
+        publish(&claim, &sender);
+        if started {
+            assert!(claim.try_start());
+        }
+        f.reset();
+        assert_eq!(f.cancel.is_cancelled(), started);
+        assert!(!claim.try_start());
+        drop(receiver);
+        drop(claim);
+        assert_eq!(f.budget.reserved(), 0);
+    }
+}
+
+#[test]
+fn queue_failure_after_receipt_preserves_the_received_height() {
+    let mut f = Fixture::new();
+    let (sender, receiver) = worker_framed_channel(1);
+    let claim = f.take(1);
+    publish(&claim, &sender);
+    f.budget.release(
+        f.work
+            .release_active_reserved_height_for_owner(claim.owner(), block::Height(1))
+            .unwrap(),
+    );
+    drop(receiver);
+    drop(claim);
+    assert_eq!(f.budget.reserved(), 0);
+    assert_eq!(f.work.in_flight_len(), 1);
+    assert!(!f.work.pending_contains(block::Height(1)));
+    assert!(f.work.pending_contains(block::Height(2)));
+}
+
+#[test]
+fn receiver_closing_after_slot_reservation_settles_publication_once() {
+    let mut f = Fixture::new();
+    let (sender, receiver) = worker_framed_channel(1);
+    let slot = sender.try_reserve_guarded().unwrap();
+    let claim = f.take(1);
+    drop(receiver);
+    let mut delivered = true;
+    assert!(claim.publish(|| {
+        delivered = slot.send_request(
+            Frame {
+                message_type: 2,
+                flags: 0,
+                payload: vec![],
+            },
+            claim.clone(),
+        );
+    }));
+    assert!(!delivered);
+    claim.delivery_failed();
+    drop(claim);
+    assert_eq!(f.budget.reserved(), 0);
+    assert_eq!(f.work.reserved_bytes(), 0);
+    assert_eq!(f.work.pending_len(), 2);
+}
+
+#[test]
+fn reset_cannot_interleave_outstanding_publication_and_enqueue() {
+    let mut f = Fixture::new();
+    let claim = f.take(1);
+    let (entered_tx, entered_rx) = std::sync::mpsc::channel();
+    let (finish_tx, finish_rx) = std::sync::mpsc::channel();
+    let (reset_tx, reset_rx) = std::sync::mpsc::channel();
+    let (attempt_tx, attempt_rx) = std::sync::mpsc::channel();
+    let work = f.work.clone();
+    std::thread::scope(|scope| {
+        let claim = claim.clone();
+        scope.spawn(move || {
+            let (sender, receiver) = worker_framed_channel(1);
+            let slot = sender.try_reserve_guarded().unwrap();
+            assert!(claim.publish(|| {
+                entered_tx.send(()).unwrap();
+                finish_rx.recv_timeout(Duration::from_secs(2)).unwrap();
+                assert!(slot.send_request(
+                    Frame {
+                        message_type: 2,
+                        flags: 0,
+                        payload: vec![]
+                    },
+                    claim.clone()
+                ));
+            }));
+            drop(receiver);
+        });
+        entered_rx.recv_timeout(Duration::from_secs(2)).unwrap();
+        scope.spawn(move || {
+            attempt_tx.send(()).unwrap();
+            reset_tx.send(work.reset_above(block::Height(0))).unwrap();
+        });
+        attempt_rx.recv_timeout(Duration::from_secs(2)).unwrap();
+        assert!(matches!(
+            reset_rx.recv_timeout(Duration::from_millis(20)),
+            Err(std::sync::mpsc::RecvTimeoutError::Timeout)
+        ));
+        finish_tx.send(()).unwrap();
+        f.budget
+            .release(reset_rx.recv_timeout(Duration::from_secs(2)).unwrap());
+    });
+    drop(claim);
+    assert_eq!(f.budget.reserved(), 0);
+    assert_eq!(f.work.reserved_bytes(), 0);
+}
+
+#[tokio::test]
+async fn partial_queued_expiry_returns_the_whole_request_before_queue_drain() {
+    for heights in [
+        vec![block::Height(1)],
+        vec![block::Height(2)],
+        vec![block::Height(1), block::Height(2)],
+    ] {
+        let mut f = Fixture::new();
+        let (sender, mut receiver) = worker_framed_channel(2);
+        let claim = f.take(1);
+        let owner = claim.owner();
+        let status = claim.status();
+        publish(&claim, &sender);
+        drop(claim);
+        let outcome = f
+            .work
+            .release_reserved_and_return_items_detailed_for_owner(owner, heights);
+        f.budget.release(outcome.released_bytes);
+        assert!(status.was_skipped());
+        assert_eq!(outcome.returned_count, 2);
+        assert_eq!(outcome.min_height, Some(block::Height(1)));
+        assert_eq!(outcome.max_height, Some(block::Height(2)));
+        assert_eq!(outcome.released_bytes, 200);
+        assert_eq!(f.work.pending_len(), 2);
+        assert_eq!(f.work.in_flight_len(), 0);
+        assert_eq!(f.budget.reserved(), 0);
+        assert_eq!(f.work.reserved_bytes(), 0);
+
+        let replacement = f.take(2);
+        publish(&replacement, &sender);
+        let stale = f
+            .work
+            .release_reserved_and_return_items_detailed_for_owner(owner, [block::Height(1)]);
+        assert_eq!(stale.released_bytes, 0);
+        assert_eq!(stale.missing_count, 2);
+        receiver
+            .recv()
+            .await
+            .unwrap()
+            .write_with(|_| async {
+                panic!("the expired request cannot reach the transport");
+                #[allow(unreachable_code)]
+                Ok::<(), ()>(())
+            })
+            .await
+            .unwrap();
+        for height in [block::Height(1), block::Height(2)] {
+            assert_eq!(f.work.owner_for_height(height), Some(replacement.owner()));
+        }
+        assert_eq!(f.budget.reserved(), 200);
+        assert_eq!(f.work.reserved_bytes(), 200);
+        assert!(!f.cancel.is_cancelled());
+        drop(receiver);
+        drop(replacement);
+        assert_eq!(f.budget.reserved(), 0);
+    }
+}
+
+#[test]
+fn partial_started_expiry_keeps_the_other_height_reserved() {
+    let mut f = Fixture::new();
+    let (sender, receiver) = worker_framed_channel(1);
+    let claim = f.take(1);
+    publish(&claim, &sender);
+    assert!(claim.try_start());
+    let outcome = f
+        .work
+        .release_reserved_and_return_items_detailed_for_owner(claim.owner(), [block::Height(1)]);
+    f.budget.release(outcome.released_bytes);
+    assert!(!claim.status().was_skipped());
+    assert_eq!(outcome.returned_count, 1);
+    assert_eq!(outcome.released_bytes, 100);
+    assert_eq!(f.work.pending_len(), 1);
+    assert_eq!(
+        f.work.owner_for_height(block::Height(2)),
+        Some(claim.owner())
+    );
+    assert_eq!(f.budget.reserved(), 100);
+    assert_eq!(f.work.reserved_bytes(), 100);
+    assert!(!f.cancel.is_cancelled());
+    claim.written();
+    f.expire(claim.owner());
+    drop(receiver);
+    drop(claim);
+    assert!(!f.cancel.is_cancelled());
+    assert_eq!(f.budget.reserved(), 0);
+}
+
+#[test]
+fn receipt_retires_only_unwritten_requests_and_preserves_the_body() {
+    for started in [false, true] {
+        for claim_received in [false, true] {
+            let mut f = Fixture::new();
+            let (sender, receiver) = worker_framed_channel(1);
+            let claim = f.take(1);
+            publish(&claim, &sender);
+            if started {
+                assert!(claim.try_start());
+            }
+            let released = if claim_received {
+                f.work
+                    .claim_received_for_owner(claim.owner(), block::Height(1))
+            } else {
+                f.work
+                    .release_active_reserved_height_for_owner(claim.owner(), block::Height(1))
+                    .unwrap()
+            };
+            f.budget.release(released);
+            assert_eq!(released, if started { 100 } else { 200 });
+            assert_eq!(claim.status().was_skipped(), !started);
+            assert_eq!(f.budget.reserved(), if started { 100 } else { 0 });
+            assert_eq!(f.work.reserved_bytes(), f.budget.reserved());
+            assert_eq!(
+                f.work.owner_for_height(block::Height(1)),
+                Some(claim.owner())
+            );
+            assert!(!f.work.pending_contains(block::Height(1)));
+            assert_eq!(f.work.pending_contains(block::Height(2)), !started);
+
+            let outcome = f.work.release_reserved_and_return_items_detailed_for_owner(
+                claim.owner(),
+                [block::Height(2)],
+            );
+            f.budget.release(outcome.released_bytes);
+            assert_eq!(outcome.returned_count, u64::from(started));
+            assert_eq!(outcome.released_bytes, if started { 100 } else { 0 });
+            if started {
+                claim.written();
+            }
+            drop(receiver);
+            drop(claim);
+            assert!(!f.cancel.is_cancelled());
+            assert_eq!(f.budget.reserved(), 0);
+            assert_eq!(f.work.in_flight_len(), 1);
+        }
+    }
+}
+
+#[test]
+fn queued_expiry_racing_reset_settles_each_reservation_once() {
+    for _ in 0..8 {
+        let mut f = Fixture::new();
+        let (sender, receiver) = worker_framed_channel(1);
+        let claim = f.take(1);
+        publish(&claim, &sender);
+        let owner = claim.owner();
+        let barrier = std::sync::Barrier::new(2);
+        let work = f.work.clone();
+        std::thread::scope(|scope| {
+            let barrier = &barrier;
+            let mut budget = f.budget.clone();
+            scope.spawn(move || {
+                barrier.wait();
+                budget.release(work.reset_above(block::Height(0)));
+            });
+            barrier.wait();
+            let outcome = f
+                .work
+                .release_reserved_and_return_items_detailed_for_owner(owner, [block::Height(1)]);
+            f.budget.release(outcome.released_bytes);
+        });
+        assert_eq!(f.budget.reserved(), 0);
+        assert_eq!(f.work.reserved_bytes(), 0);
+        assert_eq!(f.work.pending_len(), 0);
+        assert_eq!(f.work.in_flight_len(), 0);
+        drop(receiver);
+        drop(claim);
+        assert_eq!(f.budget.reserved(), 0);
+        assert!(!f.cancel.is_cancelled());
+    }
+}
+
+#[test]
+fn expiry_after_forward_reset_discards_committed_heights() {
+    for received in [false, true] {
+        let mut f = Fixture::new();
+        let (sender, receiver) = worker_framed_channel(1);
+        let claim = f.take(1);
+        publish(&claim, &sender);
+        if received {
+            // Receipt retires queued requests immediately. A written request can
+            // retain both a received prefix and an unreceived suffix until reset.
+            assert!(claim.try_start());
+            claim.written();
+            f.budget.release(
+                f.work
+                    .release_active_reserved_height_for_owner(claim.owner(), block::Height(1))
+                    .unwrap(),
+            );
+        }
+        // Reset retains heights at or below its new floor. Cleanup must discard
+        // that prefix instead of offering already-committed work to another peer.
+        f.budget.release(f.work.reset_above(block::Height(2)));
+        let outcome = f.work.release_reserved_and_return_items_detailed_for_owner(
+            claim.owner(),
+            [block::Height(1), block::Height(2)],
+        );
+        assert_eq!(outcome.committed_count, 2);
+        assert_eq!(outcome.returned_count, 0);
+        assert_eq!(outcome.released_bytes, if received { 100 } else { 200 });
+        f.budget.release(outcome.released_bytes);
+        assert_eq!(f.work.pending_len(), 0);
+        assert_eq!(f.work.in_flight_len(), 0);
+        assert_eq!(f.work.reserved_bytes(), 0);
+        assert_eq!(f.budget.reserved(), 0);
+        drop(receiver);
+        drop(claim);
+        assert_eq!(f.work.pending_len(), 0);
+        assert_eq!(f.budget.reserved(), 0);
+        assert!(!f.cancel.is_cancelled());
+    }
+}
+
+#[tokio::test]
+async fn request_write_cleanup_preserves_the_recorded_transport_failure() {
+    use crate::zakura::transport::{OrderedStreamFailure, OrderedStreamFailureCause};
+
+    for failure in [
+        OrderedStreamFailure::RemoteClose,
+        OrderedStreamFailure::WriteTimeout,
+    ] {
+        let mut f = Fixture::new();
+        let (sender, mut receiver) = worker_framed_channel(1);
+        let claim = f.take(1);
+        publish(&claim, &sender);
+        drop(claim);
+        let cause = OrderedStreamFailureCause::default();
+        let result = receiver
+            .recv()
+            .await
+            .unwrap()
+            .write_with(|_| async {
+                assert!(!f.cancel.is_cancelled());
+                assert_eq!(f.budget.reserved(), 200);
+                // The transport records the error while the real request owner is
+                // still alive. Its destructor must retain that cause through cleanup.
+                cause.record(failure);
+                Err::<(), _>("write failed")
+            })
+            .await;
+        assert!(result.is_err());
+        assert!(f.cancel.is_cancelled());
+        assert_eq!(cause.get(), Some(failure));
+        assert_eq!(f.work.pending_len(), 2);
+        assert_eq!(f.work.reserved_bytes(), 0);
+        assert_eq!(f.budget.reserved(), 0);
+    }
+}
diff --git a/crates/zakura-network/src/zakura/testkit/blocksync_fuzz/tests.rs b/crates/zakura-network/src/zakura/testkit/blocksync_fuzz/tests.rs
index d10961f481..42bd468c6e 100644
--- a/crates/zakura-network/src/zakura/testkit/blocksync_fuzz/tests.rs
+++ b/crates/zakura-network/src/zakura/testkit/blocksync_fuzz/tests.rs
@@ -588,10 +588,8 @@ async fn fuzz_peer_slows_radically_is_kept() {
         (0, 0),
         "a peer that only slowed down (still delivering) must not be rejected or parked",
     );
-    // It kept delivering, so it was never sealed off like a dropper: its reliability
-    // recovers to a healthy settled band (late bodies credit back transition timeouts),
-    // well clear of the sealed (~0) range even though a lone slow peer serving its own
-    // contiguous floor carries some steady re-request churn.
+    // Transfer-aware deadlines may avoid every timeout. Otherwise reliability
+    // must recover as the slow peer continues delivering.
     assert!(
         report.final_reliability_permille >= 300,
         "a slow-but-delivering peer's reliability must stay well clear of the sealed range \
@@ -600,7 +598,8 @@ async fn fuzz_peer_slows_radically_is_kept() {
         report.min_reliability_permille,
     );
     assert!(
-        report.final_reliability_permille > report.min_reliability_permille,
+        report.final_reliability_permille == 1000
+            || report.final_reliability_permille > report.min_reliability_permille,
         "reliability must recover from its transition trough (settled {} vs trough {})",
         report.final_reliability_permille,
         report.min_reliability_permille,
diff --git a/docs/changelog/unreleased/944.md b/docs/changelog/unreleased/944.md
new file mode 100644
index 0000000000..4e78e3df95
--- /dev/null
+++ b/docs/changelog/unreleased/944.md
@@ -0,0 +1,15 @@
+## Fixed
+
+- Settle outgoing GetBlocks request ownership exactly once across expiry,
+  cancellation, and reset, including requests still queued for transport.
+  Return every unsent height immediately when any part of a queued request
+  expires, preserving received bodies and replacement requests. Discard
+  already-committed heights during request cleanup after a forward reset.
+  Retire requests skipped or expired before writing without charging the peer
+  for an unanswered probe or timeout, including floor watchdog avoidance.
+  Retire queued requests immediately when a competing body makes them unwritable,
+  preserving the received body and returning the remaining unsent work.
+  Give initial probes and queued responses time to arrive from slow peers,
+  while preserving the shorter floor rescue deadline for measured peers in
+  both byte-count and block-count modes
+  ([#944](https://github.com/zakura-core/zakura/pull/944)).
diff --git a/docs/specs/blocksync/congestion_control.md b/docs/specs/blocksync/congestion_control.md
index dfb7aa766e..9378b79e1f 100644
--- a/docs/specs/blocksync/congestion_control.md
+++ b/docs/specs/blocksync/congestion_control.md
@@ -99,15 +99,36 @@ the previous base round-trip, so one tick's burst can't inflate the BDR max.
 **A slow peer holds the contiguous floor.** The lowest missing height gates commit; one
 slow carrier must not pin it.
 
-- A floor request MUST carry a short leash (`floor_rescue_timeout`, 2 s); on expiry the
-  height MUST return to the queue and the peer be retry-avoided — rescued, not
-  disconnected (record-only).
+- A floor request with a fresh delivery-rate sample MUST use `floor_rescue_timeout`
+  (2 s) plus estimated transfer time. Without a sample, it MUST use `request_timeout`
+  (8 s) plus transfer time so the cold peer's only probe can complete. On expiry the
+  height MUST return to the queue. If its request write started, the peer MUST
+  be retry-avoided — rescued, not disconnected (record-only).
 - The floor MAY borrow up to `floor_bypass_slots` (2) bodies beyond a saturated window
   (within the request-count cap, reserving real budget). The borrow MUST scale by the
   peer's reliability, so a sealed peer earns **no** bypass; if every servable carrier is
   sealed, the floor waits for a fresh one.
-- Above-floor speculation SHOULD use a size-aware deadline (`request_timeout + bytes ÷
-  BDR`) and MUST NOT gate the floor.
+- Above-floor requests use `request_timeout` plus estimated transfer time.
+- Both lanes MUST include this response and earlier unreceived responses in the byte
+  estimate: responses share an ordered stream. Use the measured byte rate with a
+  256 KiB/s lower bound; use that lower bound when the rate is unmeasured.
+
+If a request expires before its write starts, it MUST be skipped and its work
+and reservation returned without a timeout penalty, retry avoidance, or an
+unanswered-probe charge. Requests whose writes have started remain accountable
+for timeouts even if the write has not finished.
+This applies to both routine expiry and the central floor watchdog. The watchdog
+MUST decide avoidance from the write disposition captured during settlement,
+atomically against writer startup.
+If a competing body makes a queued request unwritable, receipt MUST retire that
+request immediately, return its remaining unsent work, and remove its peer
+obligation without waiting for the writer. The received body keeps its owner
+until the sequencer consumes it.
+
+For example, suppose B takes four seconds to send each 2 MiB block. If A queues
+block 101 behind block 100, block 101 needs eight seconds of transfer allowance,
+plus its base timeout. Allowing time only for block 101 can expire it while B is
+still delivering the earlier response.
 
 **Unbounded memory under attacker-controlled bodies or stalls.**
 
@@ -128,9 +149,10 @@ slow carrier must not pin it.
   single always-taken item that guarantees floor progress.
 - The reorder look-ahead and the serving-request heap MUST be bounded.
 
-**An unbounded wait wedges a peer.** Every outbound request MUST have a network deadline —
-the only sanctioned timer. When BDR is near zero the above-floor deadline assumes a
-minimum delivery rate, so it stays finite (~16 s worst case).
+**An unbounded wait wedges a peer.** Every outbound request MUST have a finite network
+deadline. Admission bounds the outstanding byte estimate, and the minimum rate bounds
+its transfer allowance. The separate block-progress deadline still retires a silent
+session even when queued responses have later individual deadlines.
 
 **A peer accepts requests but never delivers bodies (probe-first).** Admission in front
 of the window MUST enforce a no-progress policy: