asupersync 0.3.4

//! Immutable, reference-counted byte slice.

use super::buf::Buf;
use std::ops::{Deref, RangeBounds};
use std::sync::Arc;

/// Immutable byte slice with cheap cloning.
///
/// Cloning a `Bytes` is O(1) - it just increments a reference count.
/// Slicing is also O(1) - no data is copied, just the view is adjusted.
///
/// # Implementation
///
/// This implementation uses `Arc<Vec<u8>>` for shared ownership rather than
/// raw pointers, ensuring memory safety without unsafe code.
///
/// # Examples
///
/// ```
/// use asupersync::bytes::Bytes;
///
/// // Create from static data (no allocation)
/// let b = Bytes::from_static(b"hello world");
/// assert_eq!(&b[..], b"hello world");
///
/// // Clone is cheap (reference counting)
/// let b2 = b.clone();
/// assert_eq!(&b2[..], b"hello world");
///
/// // Slicing is O(1)
/// let hello = b.slice(0..5);
/// assert_eq!(&hello[..], b"hello");
/// ```
#[derive(Clone, Default)]
pub struct Bytes {
    /// The backing storage.
    data: BytesInner,
    /// Start offset within the backing storage.
    start: usize,
    /// Length of this view.
    len: usize,
}

#[derive(Clone, Default)]
enum BytesInner {
    /// Static data (no allocation, 'static lifetime).
    Static(&'static [u8]),
    /// Heap-allocated, reference-counted data.
    Shared(Arc<Vec<u8>>),
    /// Empty bytes (no allocation).
    #[default]
    Empty,
}

impl Bytes {
    /// Create an empty `Bytes`.
    ///
    /// No allocation occurs.
    #[inline]
    #[must_use]
    pub const fn new() -> Self {
        Self {
            data: BytesInner::Empty,
            start: 0,
            len: 0,
        }
    }

    /// Create `Bytes` from a static byte slice.
    ///
    /// No allocation occurs - the bytes point directly to static memory.
    ///
    /// # Examples
    ///
    /// ```
    /// use asupersync::bytes::Bytes;
    ///
    /// let b = Bytes::from_static(b"hello");
    /// assert_eq!(&b[..], b"hello");
    /// ```
    #[inline]
    #[must_use]
    pub const fn from_static(bytes: &'static [u8]) -> Self {
        Self {
            data: BytesInner::Static(bytes),
            start: 0,
            len: bytes.len(),
        }
    }

    /// Copy data from a slice into a new `Bytes`.
    ///
    /// This allocates and copies the data.
    ///
    /// # Examples
    ///
    /// ```
    /// use asupersync::bytes::Bytes;
    ///
    /// let data = vec![1, 2, 3, 4, 5];
    /// let b = Bytes::copy_from_slice(&data);
    /// assert_eq!(&b[..], &[1, 2, 3, 4, 5]);
    /// ```
    #[inline]
    #[must_use]
    pub fn copy_from_slice(data: &[u8]) -> Self {
        if data.is_empty() {
            return Self::new();
        }
        let vec = data.to_vec();
        let len = vec.len();
        Self {
            data: BytesInner::Shared(Arc::new(vec)),
            start: 0,
            len,
        }
    }

    /// Returns the number of bytes.
    #[inline]
    #[must_use]
    pub const fn len(&self) -> usize {
        self.len
    }

    /// Returns true if empty.
    #[inline]
    #[must_use]
    pub const fn is_empty(&self) -> bool {
        self.len == 0
    }

    /// Returns a slice of self for the given range.
    ///
    /// # Panics
    ///
    /// Panics if the range is out of bounds.
    ///
    /// # Examples
    ///
    /// ```
    /// use asupersync::bytes::Bytes;
    ///
    /// let b = Bytes::from_static(b"hello world");
    /// let hello = b.slice(0..5);
    /// assert_eq!(&hello[..], b"hello");
    /// ```
    #[inline]
    #[must_use]
    pub fn slice(&self, range: impl RangeBounds<usize>) -> Self {
        use std::ops::Bound;

        let start = match range.start_bound() {
            Bound::Included(&n) => n,
            Bound::Excluded(&n) => n.checked_add(1).expect("range start overflow"),
            Bound::Unbounded => 0,
        };

        let end = match range.end_bound() {
            Bound::Included(&n) => n.checked_add(1).expect("range end overflow"),
            Bound::Excluded(&n) => n,
            Bound::Unbounded => self.len,
        };

        assert!(
            start <= end && end <= self.len,
            "slice bounds out of range: start={start}, end={end}, len={}",
            self.len
        );

        // br-asupersync-zfhz06: under a high-volume zero-copy pipeline,
        // repeated `slice()`/`split_to()` calls can advance `self.start`
        // close to `usize::MAX`. The bounds check above only validates
        // `end <= self.len`, not `self.start + start`. Use `checked_add`
        // and panic on overflow rather than silently wrap and read from
        // the wrong offset of the shared underlying buffer.
        Self {
            data: self.data.clone(),
            start: self
                .start
                .checked_add(start)
                .expect("Bytes::slice offset overflow"),
            len: end - start,
        }
    }

    /// Split off the bytes from `at` to the end.
    ///
    /// Self becomes `[0, at)`, returns `[at, len)`.
    ///
    /// # Panics
    ///
    /// Panics if `at > len`.
    ///
    /// # Examples
    ///
    /// ```
    /// use asupersync::bytes::Bytes;
    ///
    /// let mut b = Bytes::from_static(b"hello world");
    /// let world = b.split_off(6);
    /// assert_eq!(&b[..], b"hello ");
    /// assert_eq!(&world[..], b"world");
    /// ```
    #[inline]
    #[must_use]
    pub fn split_off(&mut self, at: usize) -> Self {
        assert!(
            at <= self.len,
            "split_off out of bounds: at={at}, len={}",
            self.len
        );

        // br-asupersync-zfhz06: see slice() for the overflow rationale.
        let other = Self {
            data: self.data.clone(),
            start: self
                .start
                .checked_add(at)
                .expect("Bytes::split_off offset overflow"),
            len: self.len - at,
        };

        self.len = at;
        other
    }

    /// Split off bytes from the beginning.
    ///
    /// Self becomes `[at, len)`, returns `[0, at)`.
    ///
    /// # Panics
    ///
    /// Panics if `at > len`.
    ///
    /// # Examples
    ///
    /// ```
    /// use asupersync::bytes::Bytes;
    ///
    /// let mut b = Bytes::from_static(b"hello world");
    /// let hello = b.split_to(6);
    /// assert_eq!(&hello[..], b"hello ");
    /// assert_eq!(&b[..], b"world");
    /// ```
    #[inline]
    #[must_use]
    pub fn split_to(&mut self, at: usize) -> Self {
        assert!(
            at <= self.len,
            "split_to out of bounds: at={at}, len={}",
            self.len
        );

        let other = Self {
            data: self.data.clone(),
            start: self.start,
            len: at,
        };

        // br-asupersync-zfhz06: see slice() for the overflow rationale.
        // Repeated split_to() advances `self.start`; after enough small
        // advances on a 32-bit target the unchecked `+=` would wrap.
        self.start = self
            .start
            .checked_add(at)
            .expect("Bytes::split_to offset overflow");
        self.len -= at;
        other
    }

    /// Truncate the buffer to `len` bytes.
    ///
    /// If `len` is greater than the current length, this has no effect.
    #[inline]
    pub fn truncate(&mut self, len: usize) {
        if len < self.len {
            self.len = len;
        }
    }

    /// Clear the buffer, making it empty.
    #[inline]
    pub fn clear(&mut self) {
        self.len = 0;
    }

    /// Get the underlying byte slice.
    ///
    /// br-asupersync-zfhz06: `self.start + self.len` is computed via
    /// `checked_add` to fail loudly rather than read from the wrong
    /// offset of the shared underlying buffer if internal accounting
    /// somehow drifted past `usize::MAX`. The constructor and every
    /// mutator (`slice`, `split_to`, `split_off`) now also use
    /// `checked_add`, so this is belt-and-braces — the invariant
    /// is enforced at every point where it can be broken.
    #[inline]
    fn as_slice(&self) -> &[u8] {
        let end = self
            .start
            .checked_add(self.len)
            .expect("Bytes::as_slice start + len overflow");
        match &self.data {
            BytesInner::Empty => &[],
            BytesInner::Static(s) => &s[self.start..end],
            BytesInner::Shared(arc) => &arc[self.start..end],
        }
    }
}

impl Deref for Bytes {
    type Target = [u8];

    #[inline]
    fn deref(&self) -> &[u8] {
        self.as_slice()
    }
}

impl AsRef<[u8]> for Bytes {
    #[inline]
    fn as_ref(&self) -> &[u8] {
        self.as_slice()
    }
}

impl From<Vec<u8>> for Bytes {
    #[inline]
    fn from(vec: Vec<u8>) -> Self {
        if vec.is_empty() {
            return Self::new();
        }
        let len = vec.len();
        Self {
            data: BytesInner::Shared(Arc::new(vec)),
            start: 0,
            len,
        }
    }
}

impl From<&'static [u8]> for Bytes {
    #[inline]
    fn from(slice: &'static [u8]) -> Self {
        Self::from_static(slice)
    }
}

impl From<&'static str> for Bytes {
    #[inline]
    fn from(s: &'static str) -> Self {
        Self::from_static(s.as_bytes())
    }
}

impl From<String> for Bytes {
    #[inline]
    fn from(s: String) -> Self {
        Self::from(s.into_bytes())
    }
}

impl std::fmt::Debug for Bytes {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("Bytes")
            .field("len", &self.len)
            .field("start", &self.start)
            .field("data", &self.as_slice())
            .finish()
    }
}

impl PartialEq for Bytes {
    #[inline]
    fn eq(&self, other: &Self) -> bool {
        self.as_slice() == other.as_slice()
    }
}

impl Eq for Bytes {}

impl PartialEq<[u8]> for Bytes {
    #[inline]
    fn eq(&self, other: &[u8]) -> bool {
        self.as_slice() == other
    }
}

impl PartialEq<Bytes> for [u8] {
    #[inline]
    fn eq(&self, other: &Bytes) -> bool {
        self == other.as_slice()
    }
}

impl PartialEq<Vec<u8>> for Bytes {
    #[inline]
    fn eq(&self, other: &Vec<u8>) -> bool {
        self.as_slice() == other.as_slice()
    }
}

impl std::hash::Hash for Bytes {
    #[inline]
    fn hash<H: std::hash::Hasher>(&self, state: &mut H) {
        self.as_slice().hash(state);
    }
}

// === Buf trait implementation ===

/// A cursor for reading from Bytes.
///
/// This wrapper tracks the read position, allowing Bytes to implement Buf.
#[derive(Clone, Debug)]
pub struct BytesCursor {
    inner: Bytes,
    pos: usize,
}

impl BytesCursor {
    /// Create a new cursor at position 0.
    #[inline]
    #[must_use]
    pub fn new(bytes: Bytes) -> Self {
        Self {
            inner: bytes,
            pos: 0,
        }
    }

    /// Get a reference to the underlying Bytes.
    #[inline]
    #[must_use]
    pub fn get_ref(&self) -> &Bytes {
        &self.inner
    }

    /// Consume the cursor, returning the underlying Bytes.
    #[inline]
    #[must_use]
    pub fn into_inner(self) -> Bytes {
        self.inner
    }

    /// Get the current position.
    #[inline]
    #[must_use]
    pub fn position(&self) -> usize {
        self.pos
    }

    /// Set the position.
    #[inline]
    pub fn set_position(&mut self, pos: usize) {
        self.pos = pos;
    }
}

impl Buf for BytesCursor {
    #[inline]
    fn remaining(&self) -> usize {
        self.inner.len().saturating_sub(self.pos)
    }

    #[inline]
    fn chunk(&self) -> &[u8] {
        let slice = self.inner.as_slice();
        if self.pos >= slice.len() {
            &[]
        } else {
            &slice[self.pos..]
        }
    }

    #[inline]
    fn advance(&mut self, cnt: usize) {
        assert!(
            cnt <= self.remaining(),
            "advance out of bounds: cnt={cnt}, remaining={}",
            self.remaining()
        );
        self.pos += cnt;
    }
}

impl Bytes {
    /// Create a cursor for reading from this Bytes.
    ///
    /// The cursor implements `Buf` and tracks the read position.
    ///
    /// # Examples
    ///
    /// ```
    /// use asupersync::bytes::{Bytes, Buf};
    ///
    /// let b = Bytes::from_static(b"\x00\x01\x02\x03");
    /// let mut cursor = b.reader();
    /// assert_eq!(cursor.get_u8(), 0);
    /// assert_eq!(cursor.get_u8(), 1);
    /// assert_eq!(cursor.remaining(), 2);
    /// ```
    #[inline]
    #[must_use]
    pub fn reader(self) -> BytesCursor {
        BytesCursor::new(self)
    }
}

#[cfg(test)]
mod tests {
    #![allow(
        clippy::pedantic,
        clippy::nursery,
        clippy::expect_fun_call,
        clippy::map_unwrap_or,
        clippy::cast_possible_wrap,
        clippy::future_not_send
    )]
    use super::*;
    use proptest::prelude::*;

    fn init_test(name: &str) {
        crate::test_utils::init_test_logging();
        crate::test_phase!(name);
    }

    #[test]
    fn test_bytes_new() {
        init_test("test_bytes_new");
        let b = Bytes::new();
        let empty = b.is_empty();
        crate::assert_with_log!(empty, "empty", true, empty);
        let len = b.len();
        crate::assert_with_log!(len == 0, "len", 0, len);
        crate::test_complete!("test_bytes_new");
    }

    #[test]
    fn test_bytes_from_static() {
        init_test("test_bytes_from_static");
        let b = Bytes::from_static(b"hello world");
        let len = b.len();
        crate::assert_with_log!(len == 11, "len", 11, len);
        let ok = &b[..] == b"hello world";
        crate::assert_with_log!(ok, "contents", b"hello world", &b[..]);
        crate::test_complete!("test_bytes_from_static");
    }

    #[test]
    fn test_bytes_copy_from_slice() {
        init_test("test_bytes_copy_from_slice");
        let data = vec![1u8, 2, 3, 4, 5];
        let b = Bytes::copy_from_slice(&data);
        let len = b.len();
        crate::assert_with_log!(len == 5, "len", 5, len);
        let ok = b[..] == data[..];
        crate::assert_with_log!(ok, "contents", &data[..], &b[..]);
        crate::test_complete!("test_bytes_copy_from_slice");
    }

    #[test]
    fn test_bytes_clone_is_cheap() {
        init_test("test_bytes_clone_is_cheap");
        let b1 = Bytes::copy_from_slice(&vec![0u8; 1_000_000]);
        let start = std::time::Instant::now();
        for _ in 0..1000 {
            let _b2 = b1.clone();
        }
        let elapsed = start.elapsed();
        // Should be very fast (reference counting)
        let fast = elapsed.as_millis() < 50;
        crate::assert_with_log!(fast, "clone fast", true, fast);
        crate::test_complete!("test_bytes_clone_is_cheap");
    }

    #[test]
    fn test_bytes_slice() {
        init_test("test_bytes_slice");
        let b = Bytes::from_static(b"hello world");

        let hello = b.slice(0..5);
        let ok = &hello[..] == b"hello";
        crate::assert_with_log!(ok, "hello", b"hello", &hello[..]);

        let world = b.slice(6..);
        let ok = &world[..] == b"world";
        crate::assert_with_log!(ok, "world", b"world", &world[..]);

        let middle = b.slice(3..8);
        let ok = &middle[..] == b"lo wo";
        crate::assert_with_log!(ok, "middle", b"lo wo", &middle[..]);
        crate::test_complete!("test_bytes_slice");
    }

    #[test]
    fn test_bytes_split_off() {
        init_test("test_bytes_split_off");
        let mut b = Bytes::from_static(b"hello world");
        let world = b.split_off(6);

        let ok = &b[..] == b"hello ";
        crate::assert_with_log!(ok, "left", b"hello ", &b[..]);
        let ok = &world[..] == b"world";
        crate::assert_with_log!(ok, "world", b"world", &world[..]);
        crate::test_complete!("test_bytes_split_off");
    }

    #[test]
    fn test_bytes_split_to() {
        init_test("test_bytes_split_to");
        let mut b = Bytes::from_static(b"hello world");
        let hello = b.split_to(6);

        let ok = &hello[..] == b"hello ";
        crate::assert_with_log!(ok, "left", b"hello ", &hello[..]);
        let ok = &b[..] == b"world";
        crate::assert_with_log!(ok, "world", b"world", &b[..]);
        crate::test_complete!("test_bytes_split_to");
    }

    #[test]
    fn test_bytes_truncate() {
        init_test("test_bytes_truncate");
        let mut b = Bytes::from_static(b"hello world");
        b.truncate(5);
        let ok = &b[..] == b"hello";
        crate::assert_with_log!(ok, "truncate", b"hello", &b[..]);

        // Truncate to larger has no effect
        b.truncate(100);
        let ok = &b[..] == b"hello";
        crate::assert_with_log!(ok, "truncate unchanged", b"hello", &b[..]);
        crate::test_complete!("test_bytes_truncate");
    }

    #[test]
    fn test_bytes_clear() {
        init_test("test_bytes_clear");
        let mut b = Bytes::from_static(b"hello world");
        b.clear();
        let empty = b.is_empty();
        crate::assert_with_log!(empty, "empty", true, empty);
        crate::test_complete!("test_bytes_clear");
    }

    #[test]
    fn test_bytes_from_vec() {
        init_test("test_bytes_from_vec");
        let v = vec![1u8, 2, 3];
        let b: Bytes = v.into();
        let ok = b[..] == [1, 2, 3];
        crate::assert_with_log!(ok, "from vec", &[1, 2, 3], &b[..]);
        crate::test_complete!("test_bytes_from_vec");
    }

    #[test]
    fn test_bytes_from_string() {
        init_test("test_bytes_from_string");
        let s = String::from("hello");
        let b: Bytes = s.into();
        let ok = &b[..] == b"hello";
        crate::assert_with_log!(ok, "from string", b"hello", &b[..]);
        crate::test_complete!("test_bytes_from_string");
    }

    #[test]
    #[should_panic(expected = "slice bounds out of range")]
    fn test_bytes_slice_panic() {
        init_test("test_bytes_slice_panic");
        let b = Bytes::from_static(b"hello");
        let _bad = b.slice(0..100);
    }

    #[test]
    #[should_panic(expected = "out of bounds")]
    fn test_bytes_split_off_panic() {
        init_test("test_bytes_split_off_panic");
        let mut b = Bytes::from_static(b"hello");
        let _bad = b.split_off(100);
    }

    // =========================================================================
    // Wave 32: Data-type trait coverage
    // =========================================================================

    #[test]
    fn bytes_debug() {
        let b = Bytes::from_static(b"hi");
        let dbg = format!("{b:?}");
        assert!(dbg.contains("Bytes"));
        assert!(dbg.contains("len"));
    }

    #[test]
    fn bytes_default() {
        let b = Bytes::default();
        assert!(b.is_empty());
        assert_eq!(b.len(), 0);
    }

    #[test]
    fn bytes_hash() {
        use std::collections::HashSet;
        let a = Bytes::from_static(b"hello");
        let b = Bytes::copy_from_slice(b"hello");
        let c = Bytes::from_static(b"world");
        let mut set = HashSet::new();
        set.insert(a);
        set.insert(b);
        set.insert(c);
        assert_eq!(set.len(), 2);
    }

    #[test]
    fn bytes_partial_eq_vec() {
        let b = Bytes::from_static(b"hello");
        let v = b"hello".to_vec();
        assert!(b == v);
    }

    #[test]
    fn bytes_as_ref() {
        let b = Bytes::from_static(b"hello");
        let r: &[u8] = b.as_ref();
        assert_eq!(r, b"hello");
    }

    #[test]
    fn bytes_from_static_slice() {
        let b: Bytes = (b"test" as &'static [u8]).into();
        assert_eq!(&b[..], b"test");
    }

    #[test]
    fn bytes_from_str() {
        let b: Bytes = "hello".into();
        assert_eq!(&b[..], b"hello");
    }

    #[test]
    fn bytes_cursor_debug_clone() {
        let b = Bytes::from_static(b"hello");
        let cursor = BytesCursor::new(b);
        let dbg = format!("{cursor:?}");
        assert!(dbg.contains("BytesCursor"));
        let cloned = cursor;
        assert_eq!(cloned.position(), 0);
    }

    #[test]
    fn bytes_cursor_position() {
        let b = Bytes::from_static(b"hello");
        let mut cursor = BytesCursor::new(b);
        assert_eq!(cursor.position(), 0);
        cursor.set_position(3);
        assert_eq!(cursor.position(), 3);
    }

    #[test]
    fn bytes_cursor_get_ref_into_inner() {
        let b = Bytes::from_static(b"hello");
        let cursor = BytesCursor::new(b.clone());
        assert_eq!(cursor.get_ref(), &b);
        let inner = cursor.into_inner();
        assert_eq!(&inner[..], b"hello");
    }

    #[test]
    fn bytes_cursor_buf_trait() {
        let b = Bytes::from_static(b"hello");
        let mut cursor = b.reader();
        assert_eq!(cursor.remaining(), 5);
        assert_eq!(cursor.chunk(), b"hello");
        cursor.advance(3);
        assert_eq!(cursor.remaining(), 2);
        assert_eq!(cursor.chunk(), b"lo");
    }

    #[test]
    fn test_bytes_equality() {
        init_test("test_bytes_equality");
        let b1 = Bytes::from_static(b"hello");
        let b2 = Bytes::copy_from_slice(b"hello");
        let ok = b1 == b2;
        crate::assert_with_log!(ok, "b1 == b2", b2, b1);
        let ok = b1 == b"hello"[..];
        crate::assert_with_log!(ok, "b1 == slice", b"hello".as_slice(), b1);
        let ok = b"hello"[..] == b1;
        crate::assert_with_log!(ok, "slice == b1", b1, b"hello".as_slice());
        crate::test_complete!("test_bytes_equality");
    }

    fn shared_arc_ptr(bytes: &Bytes) -> *const Vec<u8> {
        match &bytes.data {
            BytesInner::Shared(arc) => Arc::as_ptr(arc),
            _ => panic!("expected shared bytes backing"),
        }
    }

    #[test]
    fn bytes_conformance_clone_preserves_shared_backing_and_full_view() {
        let original = Bytes::copy_from_slice(b"conformance");
        let clone = original.clone();

        assert!(std::ptr::eq(
            shared_arc_ptr(&original),
            shared_arc_ptr(&clone)
        ));
        assert_eq!(original.start, 0);
        assert_eq!(clone.start, 0);
        assert_eq!(original.len, b"conformance".len());
        assert_eq!(clone.len, b"conformance".len());
        assert_eq!(&original[..], b"conformance");
        assert_eq!(&clone[..], b"conformance");
    }

    #[test]
    fn bytes_conformance_slice_preserves_backing_with_adjusted_offsets() {
        let original = Bytes::copy_from_slice(b"grpc-frame");
        let slice = original.slice(5..10);

        assert!(std::ptr::eq(
            shared_arc_ptr(&original),
            shared_arc_ptr(&slice)
        ));
        assert_eq!(original.start, 0);
        assert_eq!(slice.start, 5);
        assert_eq!(original.len, b"grpc-frame".len());
        assert_eq!(slice.len, 5);
        assert_eq!(&slice[..], b"frame");
        assert_eq!(&original[..], b"grpc-frame");
    }

    #[test]
    fn bytes_conformance_split_to_keeps_prefix_and_suffix_on_same_backing() {
        let mut working = Bytes::copy_from_slice(b"wire-format");
        let witness = working.clone();
        let prefix = working.split_to(5);

        assert!(std::ptr::eq(
            shared_arc_ptr(&witness),
            shared_arc_ptr(&prefix)
        ));
        assert!(std::ptr::eq(
            shared_arc_ptr(&witness),
            shared_arc_ptr(&working)
        ));
        assert_eq!(prefix.start, 0);
        assert_eq!(prefix.len, 5);
        assert_eq!(working.start, 5);
        assert_eq!(working.len, b"wire-format".len() - 5);
        assert_eq!(&prefix[..], b"wire-");
        assert_eq!(&working[..], b"format");
        assert_eq!(&witness[..], b"wire-format");
    }

    #[test]
    fn bytes_conformance_split_off_keeps_tail_view_on_same_backing() {
        let mut working = Bytes::copy_from_slice(b"task-region");
        let witness = working.clone();
        let tail = working.split_off(5);

        assert!(std::ptr::eq(
            shared_arc_ptr(&witness),
            shared_arc_ptr(&tail)
        ));
        assert!(std::ptr::eq(
            shared_arc_ptr(&witness),
            shared_arc_ptr(&working)
        ));
        assert_eq!(working.start, 0);
        assert_eq!(working.len, 5);
        assert_eq!(tail.start, 5);
        assert_eq!(tail.len, b"task-region".len() - 5);
        assert_eq!(&working[..], b"task-");
        assert_eq!(&tail[..], b"region");
        assert_eq!(&witness[..], b"task-region");
    }

    proptest! {
        #[test]
        fn metamorphic_slice_matches_split_extraction(
            data in prop::collection::vec(any::<u8>(), 0..128),
            start in 0usize..128,
            end in 0usize..128,
        ) {
            let bytes = Bytes::copy_from_slice(&data);
            let len = bytes.len();
            let start = start.min(len);
            let end = end.min(len);
            let (start, end) = if start <= end {
                (start, end)
            } else {
                (end, start)
            };

            let direct = bytes.slice(start..end);

            let mut split_view = bytes.clone();
            let _prefix = split_view.split_to(start);
            let middle = split_view.split_to(end - start);

            prop_assert_eq!(
                &direct[..],
                &middle[..],
                "direct slicing and split-based extraction must expose identical contiguous bytes",
            );
            prop_assert_eq!(
                &direct[..],
                &data[start..end],
                "both extraction paths must match the original contiguous source slice",
            );
        }

        #[test]
        fn metamorphic_split_recombine_preserves_payload(
            data in prop::collection::vec(any::<u8>(), 0..192),
            first_split in 0usize..192,
            middle_len in 0usize..192,
        ) {
            let bytes = Bytes::copy_from_slice(&data);
            let witness = bytes.clone();
            let len = bytes.len();
            let first_split = first_split.min(len);
            let middle_len = middle_len.min(len - first_split);

            let mut middle = bytes.clone();
            let prefix = middle.split_to(first_split);
            let suffix = middle.split_off(middle_len);

            prop_assert_eq!(prefix.len(), first_split);
            prop_assert_eq!(middle.len(), middle_len);
            prop_assert_eq!(suffix.len(), len - first_split - middle_len);

            prop_assert_eq!(&prefix[..], &data[..first_split]);
            prop_assert_eq!(
                &middle[..],
                &data[first_split..first_split + middle_len],
            );
            prop_assert_eq!(&suffix[..], &data[first_split + middle_len..]);
            prop_assert_eq!(&witness[..], data.as_slice());

            let mut recombined = Vec::with_capacity(len);
            recombined.extend_from_slice(&prefix);
            recombined.extend_from_slice(&middle);
            recombined.extend_from_slice(&suffix);

            prop_assert_eq!(
                recombined.as_slice(),
                data.as_slice(),
                "split_to/split_off views must recombine into the original byte order",
            );
        }
    }

    #[test]
    fn bytes_conformance_clone_shares_data() {
        let data = b"hello world";
        let original = Bytes::copy_from_slice(data);
        let clone1 = original.clone();
        let clone2 = original.clone();

        assert_eq!(&original[..], data);
        assert_eq!(&clone1[..], data);
        assert_eq!(&clone2[..], data);

        // Verify pointer equality of the underlying slice to confirm sharing
        assert_eq!(
            shared_arc_ptr(&original),
            shared_arc_ptr(&clone1),
            "clones must share the same backing allocation"
        );
    }

    #[test]
    fn bytes_conformance_truncate_and_clear() {
        let mut b = Bytes::copy_from_slice(b"hello world");

        b.truncate(5);
        assert_eq!(&b[..], b"hello");

        b.truncate(10); // no-op since len is already 5
        assert_eq!(&b[..], b"hello");

        b.clear();
        assert_eq!(&b[..], b"");
        assert!(b.is_empty());
    }

    #[test]
    fn bytes_conformance_nested_slicing() {
        let mut b = Bytes::copy_from_slice(b"the quick brown fox jumps over the lazy dog");

        // "quick brown fox jumps over the lazy dog"
        let mut part1 = b.split_off(4);
        assert_eq!(&b[..], b"the ");

        // "quick brown fox"
        let part2 = part1.split_to(15);
        assert_eq!(&part2[..], b"quick brown fox");

        // "brown"
        let brown = part2.slice(6..11);
        assert_eq!(&brown[..], b"brown");

        // "fox jumps over"
        let mut fox = part1.slice(0..15);
        assert_eq!(&fox[..], b" jumps over the");

        // "over"
        let over = fox.split_off(7);
        assert_eq!(&fox[..], b" jumps ");
        assert_eq!(&over[..], b"over the");

        // Verify all point to the original allocation
        assert_eq!(shared_arc_ptr(&b), shared_arc_ptr(&part2));
        assert_eq!(shared_arc_ptr(&b), shared_arc_ptr(&brown));
    }

    #[test]
    #[should_panic(expected = "slice bounds out of range")]
    fn bytes_conformance_slice_out_of_bounds() {
        let b = Bytes::copy_from_slice(b"hello");
        let _ = b.slice(0..10);
    }

    #[test]
    #[should_panic(expected = "split_to out of bounds")]
    fn bytes_conformance_split_to_out_of_bounds() {
        let mut b = Bytes::copy_from_slice(b"hello");
        let _ = b.split_to(10);
    }

    #[test]
    #[should_panic(expected = "split_off out of bounds")]
    fn bytes_conformance_split_off_out_of_bounds() {
        let mut b = Bytes::copy_from_slice(b"hello");
        let _ = b.split_off(10);
    }
}