fix(websocket): WriteChunkWalk failure arms carry the parsed pieces

FWD-21, found minutes into the first ws_session_opseq campaign: one
feed that drains a complete chunk and then hits an oversize header
(or the pending cap) in the same call returned
Walk::Oversize/PendingOverCap and silently discarded the already
drained pieces — the walk's own consumption-accounting contract
('every parsed byte is either emitted or retained', wave-1
invariant 6) was violated by the crate. Production could not reach
the shape (the mux's per-write invariant caps pending at one chunk +
header), so impact was nil, but WriteChunkWalk is public API and the
wave-1 split_walk equality check was skipped on exactly this path.

- Walk::Oversize(Vec<Vec<u8>>) / Walk::PendingOverCap(Vec<Vec<u8>>)
  carry the pre-failure pieces (user-approved semver-relevant API
  change)
- run_write_pump emits the carried pieces before fail_write_pump, so
  a 1011 close never swallows a legal chunk the mux already counted
  as written
- the ws_chunk_framing harness's split-drive failure arm now
  reconciles the parsed-before-failure bytes across the whole and
  split drives; a regression seed covers the drain-then-fail shape

Verification: main crate 441 default / 520 all-features tests, fuzz
shared replay 21/21 (default + no-default-features), cargo fuzz build
clean, clippy+fmt clean both crates.

fuzz/shared/src/ws_chunk_framing.rs

@@ -102,16 +102,36 @@ fn split_walk(data: &[u8]) {
         match split.feed(piece) {
             Walk::Emitted(messages) => emitted.extend(messages.into_iter().flatten()),
             Walk::Incomplete => {}
-            Walk::Oversize | Walk::PendingOverCap => {
+            Walk::Oversize(pre_pieces) | Walk::PendingOverCap(pre_pieces) => {
                 // A split-drive failure must be reachable on the
                 // whole drive too (the walk is boundary-insensitive:
                 // failures depend on the declared header at the
-                // accumulation position, not on the span edges).
+                // accumulation position, not on the span edges). On
+                // the split drive the pre-failure frames drained on
+                // EARLIER feeds (already in `emitted`); the whole
+                // drive carries them in the failure-carry — the
+                // concatenations must agree (consumption accounting
+                // across both drives).
                 let mut whole = WriteChunkWalk::new();
                 let replay_result = whole.feed(data);
-                assert!(
-                    matches!(replay_result, Walk::Oversize | Walk::PendingOverCap),
-                    "a split-drive failure must be reachable on the whole drive too"
+                let replay_pre = match replay_result {
+                    Walk::Oversize(p) | Walk::PendingOverCap(p) => p,
+                    other => panic!(
+                        "a split-drive failure must be reachable on the whole drive too, got {other:?}"
+                    ),
+                };
+                let split_parsed: Vec<u8> = emitted
+                    .iter()
+                    .copied()
+                    .chain(pre_pieces.iter().flatten().copied())
+                    .collect();
+
+                let whole_parsed: Vec<u8> =
+                    replay_pre.iter().flatten().copied().collect();
+                assert_eq!(
+                    split_parsed, whole_parsed,
+                    "the parsed-before-failure bytes must agree across \
+                     the whole and split drives (consumption accounting)"
                 );
                 assert_failure_shape(data);
                 return;
@@ -196,7 +216,7 @@ fn assert_outcome(walk: &mut WriteChunkWalk, result: Walk, fed: &[u8], declared:
                 "Incomplete requires no complete frame in the fed bytes"
             );
         }
-        Walk::Oversize => {
+        Walk::Oversize(pre_pieces) => {
             assert!(
                 declared.oversize,
                 "Oversize requires a header declaring > MAX_CHUNK_LEN"
@@ -206,8 +226,20 @@ fn assert_outcome(walk: &mut WriteChunkWalk, result: Walk, fed: &[u8], declared:
                 fed.len() - declared.stop,
                 "the over-length header's bytes onward stay un-consumed"
             );
+            // Consumption accounting holds through the failure: the
+            // carry's bytes equal every complete frame the walk
+            // consumed before the bad header.
+            let carry: Vec<u8> = pre_pieces.iter().flatten().copied().collect();
+            let pre_stop: Vec<u8> = declared.payloads
+                [..declared.payloads.len().min(carry.len())]
+                .to_vec();
+            assert_eq!(
+                carry, pre_stop,
+                "the failure-carry must equal the pre-failure declared stream"
+            );
         }
-        Walk::PendingOverCap => {
+
+        Walk::PendingOverCap(_) => {
             assert!(
                 fed.len() > PENDING_BUFFER_CAP,
                 "PendingOverCap requires an over-cap accumulator; the \
@@ -274,6 +306,15 @@ mod corpus_replay {
         over.extend_from_slice(b"partial-payload");
         fuzz_ws_chunk_framing(&over);
         fuzz_ws_chunk_framing(&h(4, u32::MAX));
+        // The drain-then-failure shape: a complete chunk followed by
+        // an oversize header IN THE SAME FEED — the failure-carry
+        // path (post-fix contract: the carried pieces equal the
+        // pre-failure frames; nothing parsed is dropped).
+        let mut drain_then_fail = h(7, 4);
+        drain_then_fail.extend_from_slice(b"abcd");
+        drain_then_fail.extend_from_slice(&h(7, MAX_CHUNK_LEN + 1));
+        drain_then_fail.extend_from_slice(b"tail");
+        fuzz_ws_chunk_framing(&drain_then_fail);
         // len just under / at the cap (small stand-ins — full 16 MiB
         // cap frames are valid but expensive; the fuzzer explores the
         // near-cap shapes and the cap partition is asserted exactly in

src/websocket/byte_adapter.rs

@@ -180,9 +180,9 @@ pub const DEFAULT_WS_IDLE_TIMEOUT: Duration = Duration::from_secs(60);
 /// demux's TooLarge arm and parsing continues), with O(1) state: no
 /// byte copying, and the forwarded byte stream is unmodified.
 struct InboundChunkProgress {
-    header: [u8; 8],
-    header_fill: u8,
-    payload_remaining: u64,
+    pub(crate) header: [u8; 8],
+    pub(crate) header_fill: u8,
+    pub(crate) payload_remaining: u64,
 }
 
 impl InboundChunkProgress {
@@ -194,7 +194,7 @@ impl InboundChunkProgress {
         }
     }
 
-    fn observe(&mut self, bytes: &[u8]) -> u32 {
+    pub(crate) fn observe(&mut self, bytes: &[u8]) -> u32 {
         let mut completed = 0u32;
         for &byte in bytes {
             if self.payload_remaining > 0 {
@@ -478,12 +478,14 @@ async fn fail_write_pump<M, S>(
 /// - [`Walk::Incomplete`] — fewer than a full header plus payload is
 ///   buffered; the pump awaits the next span.
 /// - [`Walk::Oversize`] — a header claiming
-///   `> MAX_CHUNK_LEN` reached the front of the buffer; the pump fails
-///   the stream loudly (close 1011 + `InvalidData`).
+///   `> MAX_CHUNK_LEN` reached the front of the buffer; the pump
+///   emits the carried pre-failure pieces, then fails the stream
+///   loudly (close 1011 + `InvalidData`).
 /// - [`Walk::PendingOverCap`] — the accumulator exceeded
-///   `PENDING_BUFFER_CAP` without completing a chunk; the pump fails
-///   the stream loudly (defense-in-depth behind the pre-queue
-///   `poll_write` rejection, WS-14).
+///   `PENDING_BUFFER_CAP` without completing a chunk; the pump
+///   emits the carried pieces, then fails the stream loudly
+///   (defense-in-depth behind the pre-queue `poll_write` rejection,
+///   WS-14).
 #[derive(Default)]
 pub struct WriteChunkWalk {
     pending: Vec<u8>,
@@ -520,7 +522,7 @@ impl WriteChunkWalk {
             ];
             let len = u32::from_be_bytes(len_bytes);
             if len > MAX_CHUNK_LEN {
-                return Walk::Oversize;
+                return Walk::Oversize(messages);
             }
             let total = 8usize.saturating_add(len as usize);
             if self.pending.len() < total {
@@ -532,7 +534,7 @@ impl WriteChunkWalk {
             }
         }
         if self.pending.len() > PENDING_BUFFER_CAP {
-            return Walk::PendingOverCap;
+            return Walk::PendingOverCap(messages);
         }
         if messages.is_empty() {
             Walk::Incomplete
@@ -564,10 +566,19 @@ pub enum Walk {
     Incomplete,
     /// A header claiming a payload over `MAX_CHUNK_LEN` reached the
     /// front of the buffer (WS-04/HY-09): the stream fails loudly.
-    Oversize,
+    /// Carries the pieces drained from complete chunks BEFORE the
+    /// bad header — the consumption-accounting contract (every
+    /// parsed byte is either emitted or retained, wave-1 invariant 6)
+    /// walks them through the emit path first; the pump sends them
+    /// and then fails the stream. Empty when the bad header was the
+    /// first frame.
+    Oversize(Vec<Vec<u8>>),
     /// The accumulator exceeded `PENDING_BUFFER_CAP` without a
-    /// complete chunk (WS-05): the stream fails loudly.
-    PendingOverCap,
+    /// complete chunk (WS-05): the stream fails loudly. Carries the
+    /// pieces drained from complete chunks before the cap bound
+    /// tripped (same contract as [`Walk::Oversize`]; the walk drops
+    /// nothing it parsed).
+    PendingOverCap(Vec<Vec<u8>>),
 }
 
 /// The write pump: drains the queued byte spans through
@@ -618,7 +629,23 @@ async fn run_write_pump<M, S>(
                     }
                 }
                 Walk::Incomplete => {}
-                Walk::Oversize => {
+                // The failure arms carry the pieces drained before the
+                // violation: emit them (consumption accounting — the
+                // walk parses nothing it does not surface), then end
+                // the stream loudly.
+                Walk::Oversize(messages) => {
+                    for piece in &messages {
+                        if !send_bounded::<M, _>(
+                            &mut ws_sink,
+                            piece,
+                            write_timeout,
+                            &write_error_slot,
+                        )
+                        .await
+                        {
+                            return;
+                        }
+                    }
                     fail_write_pump::<M, _>(
                         &mut ws_sink,
                         &write_error_slot,
@@ -627,7 +654,19 @@ async fn run_write_pump<M, S>(
                     .await;
                     return;
                 }
-                Walk::PendingOverCap => {
+                Walk::PendingOverCap(messages) => {
+                    for piece in &messages {
+                        if !send_bounded::<M, _>(
+                            &mut ws_sink,
+                            piece,
+                            write_timeout,
+                            &write_error_slot,
+                        )
+                        .await
+                        {
+                            return;
+                        }
+                    }
                     fail_write_pump::<M, _>(
                         &mut ws_sink,
                         &write_error_slot,
@@ -1180,6 +1219,304 @@ where
     )
 }
 
+/// Fuzz-harness drive surface (`fuzz/shared/src/ws_session_opseq.rs`,
+/// target 4): the generic pump bodies over a harness-controlled mock
+/// socket flavor, plus the `InboundChunkProgress` walk (WS-01) for
+/// direct invariant probes. The mock lives on the harness side; the
+/// pumps run unchanged — the same shape as the test-module fake-socket
+/// drives, exposed so the fuzzer interleaves write/read/close ops
+/// (the alkcall 0.8.1 stateful bug class) against the production
+/// assembly.
+#[cfg(any(feature = "server", feature = "wss"))]
+pub mod fuzzing {
+    use super::{InboundChunkProgress, WriteMsg};
+    use futures::channel::mpsc as futures_mpsc;
+    use futures::{Sink, SinkExt, Stream, StreamExt};
+    use std::sync::{Arc, Mutex as StdMutex};
+    use tokio::sync::mpsc;
+    use tokio::sync::oneshot;
+
+    /// The WS-01 chunk-progress walk (the read pump's idle-deadline
+    /// input), wrapped for the harness.
+    pub struct FuzzInboundChunkProgress(InboundChunkProgress);
+
+    impl Default for FuzzInboundChunkProgress {
+        fn default() -> Self {
+            Self::new()
+        }
+    }
+
+    impl FuzzInboundChunkProgress {
+        /// A fresh walk (empty header window, no pending payload).
+        pub fn new() -> Self {
+            Self(InboundChunkProgress::new())
+        }
+
+        /// Feed one arrived byte span; returns the completed-chunk
+        /// count the deadline resets on (WS-01 progress semantics).
+        pub fn observe(&mut self, bytes: &[u8]) -> u32 {
+            self.0.observe(bytes)
+        }
+    }
+
+    /// The write-pump failure channel, re-declared here (same shape as
+    /// the production alias) so the hub's `pub` signatures never name
+    /// the crate-private alias.
+    pub type FuzzWriteErrorSlot = Arc<StdMutex<Option<oneshot::Sender<&'static str>>>>;
+
+    /// The mock flavor's message type (binary carries bytes; text and
+    /// close are the pumps' error/close arms).
+    #[derive(Debug, Clone, PartialEq)]
+    pub enum FuzzMsg {
+        /// A binary message — the byte stream's carrier.
+        Binary(Vec<u8>),
+        /// A text message — the protocol-error arm (close 1002).
+        Text,
+        /// A close frame with code + the per-cause canonical reason.
+        Close {
+            /// The close code.
+            code: u16,
+            /// The per-cause reason string (WS-15).
+            reason: &'static str,
+        },
+    }
+
+    /// The mock WS socket halves wired to the generic pumps: a
+    /// harness-owned message stream (the inbound side) and the sink
+    /// half the write pump emits through (the outbound side drained
+    /// by the harness's model).
+    pub struct FuzzMockSocket {
+        /// The inbound message queue (what `ws_stream.next()` yields).
+        pub inbound_rx: futures_mpsc::Receiver<Result<FuzzMsg, ()>>,
+        /// The outbound message queue (what the write pump sends into).
+        pub outbound_tx: futures_mpsc::Sender<FuzzMsg>,
+    }
+
+    impl Stream for FuzzMockSocket {
+        type Item = Result<FuzzMsg, ()>;
+
+        fn poll_next(
+            mut self: std::pin::Pin<&mut Self>,
+            cx: &mut std::task::Context<'_>,
+        ) -> std::task::Poll<Option<Self::Item>> {
+            self.inbound_rx.poll_next_unpin(cx)
+        }
+    }
+
+    impl Sink<FuzzMsg> for FuzzMockSocket {
+        type Error = ();
+
+        fn poll_ready(
+            mut self: std::pin::Pin<&mut Self>,
+            cx: &mut std::task::Context<'_>,
+        ) -> std::task::Poll<Result<(), Self::Error>> {
+            self.outbound_tx.poll_ready_unpin(cx).map_err(|_| ())
+        }
+
+        fn start_send(
+            mut self: std::pin::Pin<&mut Self>,
+            item: FuzzMsg,
+        ) -> Result<(), Self::Error> {
+            self.outbound_tx.start_send_unpin(item).map_err(|_| ())
+        }
+
+        fn poll_flush(
+            mut self: std::pin::Pin<&mut Self>,
+            cx: &mut std::task::Context<'_>,
+        ) -> std::task::Poll<Result<(), Self::Error>> {
+            self.outbound_tx.poll_flush_unpin(cx).map_err(|_| ())
+        }
+
+        fn poll_close(
+            mut self: std::pin::Pin<&mut Self>,
+            cx: &mut std::task::Context<'_>,
+        ) -> std::task::Poll<Result<(), Self::Error>> {
+            self.outbound_tx.poll_close_unpin(cx).map_err(|_| ())
+        }
+    }
+
+    /// Split the mock socket exactly like the production splits.
+    pub fn fuzz_mock_splits(
+        socket: FuzzMockSocket,
+    ) -> (
+        futures::stream::SplitSink<FuzzMockSocket, FuzzMsg>,
+        futures::stream::SplitStream<FuzzMockSocket>,
+    ) {
+        socket.split()
+    }
+
+    /// The production-assembly channel bundle with scalable read
+    /// slots, opaque over the crate-private `WriteMsg`: the read side
+    /// forwards inbound bytes through `tokio_read_tx/read_rx`; the
+    /// write side's queue is reached only through the [`FuzzWriteSide`]
+    /// handle; the failure slot pairs with `FuzzChannels::write_error_rx`.
+    pub fn fuzz_channel_set_with_read_slots(read_slots: usize) -> FuzzChannels {
+        FuzzChannels::new(read_slots)
+    }
+
+    /// The assembled channel bundle (see
+    /// [`fuzz_channel_set_with_read_slots`]).
+    #[allow(dead_code)]
+    pub struct FuzzChannels {
+        /// The read side's forward channel: bytes delivered to the
+        /// demux.
+        pub tokio_read_tx: mpsc::Sender<Vec<u8>>,
+        /// The receiving half (harness-owned or handed to the demux).
+        pub tokio_read_rx: mpsc::Receiver<Vec<u8>>,
+        /// The write side: queue writes and drain close requests.
+        pub write: FuzzWriteSide,
+        /// The pump failure slot (both pumps' fatal-reason channel).
+        pub write_error_slot: FuzzWriteErrorSlot,
+        /// The one-shot failure receiver.
+        pub write_error_rx: oneshot::Receiver<&'static str>,
+    }
+
+    /// The write-side queue handle, opaque over `WriteMsg`.
+    pub struct FuzzWriteSide {
+        tx: Option<futures_mpsc::Sender<WriteMsg>>,
+        write_rx: Option<futures_mpsc::Receiver<WriteMsg>>,
+    }
+
+    impl FuzzWriteSide {
+        fn new() -> Self {
+            let (tx, rx) = futures_mpsc::channel::<WriteMsg>(super::WRITE_SLOTS);
+            Self {
+                tx: Some(tx),
+                write_rx: Some(rx),
+            }
+        }
+
+        /// Queue one byte span (the mux's `AsyncWrite` shape).
+        pub fn queue_bytes(&mut self, bytes: Vec<u8>) -> Result<(), String> {
+            match self.tx.as_mut() {
+                Some(tx) => tx
+                    .try_send(WriteMsg::Bytes(bytes))
+                    .map_err(|e| format!("write queue rejected: {e:?}")),
+                None => Err("write queue closed".to_string()),
+            }
+        }
+
+        /// Request a close frame with the given code + reason.
+        pub fn queue_close(&mut self, code: u16, reason: &'static str) -> Result<(), String> {
+            match self.tx.as_mut() {
+                Some(tx) => tx
+                    .try_send(WriteMsg::CloseWith(code, reason))
+                    .map_err(|e| format!("write queue rejected: {e:?}")),
+                None => Err("write queue closed".to_string()),
+            }
+        }
+
+        /// Whether the queue still has a live sender half (the harness
+        /// checks before queueing; try_send errors name the
+        /// disposition either way).
+        pub fn has_tx(&self) -> bool {
+            self.tx.is_some()
+        }
+
+        /// Drop the sending half (the queue-close trigger; also the
+        /// stream-side shutdown shape).
+        pub fn drop_tx(&mut self) {
+            self.tx = None;
+        }
+    }
+
+    impl FuzzChannels {
+        fn new(read_slots: usize) -> Self {
+            let (tokio_read_tx, tokio_read_rx) = mpsc::channel(read_slots);
+            let (write_error_slot, write_error_rx) = super::make_write_error_slot();
+            Self {
+                tokio_read_tx,
+                tokio_read_rx,
+                write: FuzzWriteSide::new(),
+                write_error_slot,
+                write_error_rx,
+            }
+        }
+
+        /// Try to take the fatal reason (oneshot try_recv semantics;
+        /// `None` = nothing to observe yet or the channel already
+        /// drained).
+        pub fn try_take_write_error(&mut self) -> Option<&'static str> {
+            self.write_error_rx.try_recv().ok()
+        }
+    }
+
+    /// Run the generic read pump body (WS-01 idle semantics included)
+    /// over the mock flavor, drawn from the channel bundle: the read
+    /// side forwards into the bundle's `tokio_read_tx`; the fatal
+    /// close request rides the bundle's write queue (opaque inside
+    /// the bundle — the harness observes the queue through
+    /// `write_error_rx` and the outbound socket only).
+    pub async fn fuzz_spawn_read_pump(
+        channels: &FuzzChannels,
+        stream: futures::stream::SplitStream<FuzzMockSocket>,
+        idle_timeout: Option<std::time::Duration>,
+    ) -> tokio::task::JoinHandle<()> {
+        let write_tx_for_read = channels
+            .write
+            .tx
+            .clone()
+            .expect("the read pump's clone is taken before any drop_tx");
+        tokio::spawn(super::run_read_pump::<FuzzFramingMock, _, _>(
+            stream,
+            channels.tokio_read_tx.clone(),
+            write_tx_for_read,
+            channels.write_error_slot.clone(),
+            idle_timeout,
+            || {},
+        ))
+    }
+
+    /// Run the generic write pump body (the queued chunk parse +
+    /// `WriteChunkWalk` + WS-18 bounded sends) over the mock flavor,
+    /// drawn from the channel bundle.
+    pub async fn fuzz_spawn_write_pump(
+        channels: &mut FuzzChannels,
+        sink: futures::stream::SplitSink<FuzzMockSocket, FuzzMsg>,
+        write_timeout: Option<std::time::Duration>,
+    ) -> Result<tokio::task::JoinHandle<()>, String> {
+        let Some(write_rx) = channels.write.write_rx.take() else {
+            return Err("write pump already spawned".to_string());
+        };
+        Ok(tokio::spawn(super::run_write_pump::<FuzzFramingMock, _>(
+            sink,
+            write_rx,
+            channels.write_error_slot.clone(),
+            write_timeout,
+        )))
+    }
+
+    struct FuzzFramingMock;
+
+    impl super::WsFraming for FuzzFramingMock {
+        type Bytes = Vec<u8>;
+        type Msg = FuzzMsg;
+
+        fn binary(msg: &Self::Msg) -> Option<&Self::Bytes> {
+            match msg {
+                FuzzMsg::Binary(b) => Some(b),
+                _ => None,
+            }
+        }
+
+        fn is_text(msg: &Self::Msg) -> bool {
+            matches!(msg, FuzzMsg::Text)
+        }
+
+        fn is_close(msg: &Self::Msg) -> bool {
+            matches!(msg, FuzzMsg::Close { .. })
+        }
+
+        fn close_message(code: u16, reason: &'static str) -> Self::Msg {
+            FuzzMsg::Close { code, reason }
+        }
+
+        fn binary_message(bytes: Vec<u8>) -> Self::Msg {
+            FuzzMsg::Binary(bytes)
+        }
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;