livekit_data_stream/outgoing/

manager.rs

// Copyright 2026 LiveKit, Inc.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

use bmrng::unbounded::{UnboundedRequestReceiver, UnboundedRequestSender};
use chrono::Utc;
use livekit_common::{
    ClientCapability, ParticipantIdentity, RemoteParticipantRegistry,
    CLIENT_PROTOCOL_DATA_STREAM_V2,
};
use livekit_protocol as proto;
use std::{collections::HashMap, path::Path, sync::Arc};
use tokio::sync::Mutex;

use crate::{
    info::{ByteStreamInfo, TextStreamInfo},
    types::{ByteHeader, CompressionType, ContentHeader, Header, StreamId, TextHeader},
    utf8_chunk::Utf8AwareChunkExt,
    utils::{SendError, StreamError, StreamProgress, StreamResult},
};

use super::{
    constants,
    raw_stream::{RawStream, RawStreamOpenOptions},
    stream_writer::{ByteStreamWriter, TextStreamWriter},
    SendByteOptions, SendTextOptions, StreamByteOptions, StreamProgressCallback, StreamTextOptions,
};

/// Generates a random stream identifier (UUID v4).
fn create_random_uuid() -> String {
    uuid::Uuid::new_v4().to_string()
}

#[derive(Clone)]
pub struct Manager {
    /// Request channel for sending packet batches. Each request is an ordered batch the
    /// transport acknowledges as a whole: one-shot sends (`send_text`/`send_bytes`) emit their
    /// entire stream as a single request, while every other call site sends one packet at a time.
    packet_tx: UnboundedRequestSender<Vec<proto::DataPacket>, Result<(), SendError>>,
}

impl Manager {
    pub fn new() -> (Self, UnboundedRequestReceiver<Vec<proto::DataPacket>, Result<(), SendError>>)
    {
        let (packet_tx, packet_rx) = bmrng::unbounded_channel();
        let manager = Self { packet_tx };
        (manager, packet_rx)
    }

    /// Opens an incremental text stream, declaring `options.total_length` when supplied.
    pub async fn stream_text(&self, options: StreamTextOptions) -> StreamResult<TextStreamWriter> {
        // Incremental streams are never inlined or compressed (the content is unknown up front).
        let stream_id: StreamId = options.id.clone().unwrap_or_else(create_random_uuid).into();
        let dests = options.destination_identities.clone();
        let (header, text_header) = build_text_header(
            &options.topic,
            &options.attributes,
            options.text_header(),
            stream_id,
            options.total_length,
            None,
            CompressionType::None,
        );
        enforce_header_size(&header, &dests)?;

        let open_options = RawStreamOpenOptions {
            header: header.clone(),
            destination_identities: dests,
            sender_identity: options.sender_identity.clone(),
            packet_tx: self.packet_tx.clone(),
        };
        let writer = TextStreamWriter::new(
            Arc::new(TextStreamInfo::from_headers(header, text_header)),
            Arc::new(Mutex::new(RawStream::open(open_options).await?)),
        );
        Ok(writer)
    }

    /// Opens an incremental byte stream with the declared total length, when supplied.
    pub async fn stream_bytes(&self, options: StreamByteOptions) -> StreamResult<ByteStreamWriter> {
        let stream_id: StreamId = options.id.clone().unwrap_or_else(create_random_uuid).into();
        let name = options.name.clone().unwrap_or_default();
        let dests = options.destination_identities.clone();
        let (header, byte_header) = build_byte_header(
            &options.topic,
            &options.attributes,
            options.mime_type.as_deref(),
            stream_id,
            name,
            options.total_length,
            None,
            CompressionType::None,
        );
        enforce_header_size(&header, &dests)?;

        let open_options = RawStreamOpenOptions {
            header: header.clone(),
            destination_identities: dests,
            sender_identity: options.sender_identity.clone(),
            packet_tx: self.packet_tx.clone(),
        };
        let writer = ByteStreamWriter::new(
            Arc::new(ByteStreamInfo::from_headers(header, byte_header)),
            Arc::new(Mutex::new(RawStream::open(open_options).await?)),
        );
        Ok(writer)
    }

    /// Sends a complete text payload. Progress is reported once after transport acknowledgement.
    pub async fn send_text(
        &self,
        text: &str,
        options: SendTextOptions,
        remote_participant_registry: &dyn RemoteParticipantRegistry,
    ) -> StreamResult<TextStreamInfo> {
        let stream_id: StreamId = options.id.clone().unwrap_or_else(create_random_uuid).into();
        let total_length = text.len() as u64;

        let eligibility =
            evaluate_eligibility(remote_participant_registry, &options.destination_identities);
        let can_compress = options.compress.unwrap_or(true) && eligibility.compression;

        let text_bytes = text.as_bytes();
        let mut maybe_compressed = MaybeCollectedAsyncReader::from_async_reader(
            async_compression::futures::bufread::DeflateEncoder::new(
                futures_util::io::Cursor::new(text_bytes.to_vec()),
            ),
        );

        // Compress once up front when eligible (the deflate work happens at most once, cached in
        // the returned `Vec`), then decide whether it's worth using.
        let use_compression = can_compress
            && maybe_compressed.as_bytes().await.is_ok_and(|c| c.len() < text_bytes.len());

        // 1. Inline single-packet attempt (no attachments; all recipients are >= v2).
        let (mut header, text_header) = if use_compression {
            build_text_header(
                &options.topic,
                &options.attributes,
                options.text_header(),
                stream_id.clone(),
                Some(total_length),
                Some(maybe_compressed.as_bytes().await?.to_owned()),
                CompressionType::DeflateRaw,
            )
        } else {
            build_text_header(
                &options.topic,
                &options.attributes,
                options.text_header(),
                stream_id.clone(),
                Some(total_length),
                Some(text_bytes.to_vec()),
                CompressionType::None,
            )
        };

        let proto_header = header.clone().into();
        if eligibility.inline
            && options.attached_stream_ids.is_empty()
            && header_packet_fits(&proto_header, &options.destination_identities)
        {
            let mut packet =
                RawStream::create_header_packet(proto_header, options.destination_identities);
            packet.participant_identity =
                options.sender_identity.map(|id| id.into()).unwrap_or_default();
            RawStream::send_packet(&self.packet_tx, packet).await?;
            report_complete(&options.on_progress, total_length);
            return Ok(TextStreamInfo::from_headers(header, text_header));
        }

        // 2/3. Chunked, compressed when eligible else uncompressed. The entire stream — header,
        // chunks, trailer — goes out as one transport request.
        header.inline_content = None;
        enforce_header_size(&header, &options.destination_identities)?;

        let info = TextStreamInfo::from_headers(header.clone(), text_header);
        if use_compression {
            let compressed_bytes = maybe_compressed.as_bytes().await?;
            self.send_one_shot_stream(
                header,
                compressed_bytes.chunks(constants::STREAM_CHUNK_SIZE_BYTES),
                options.destination_identities,
                options.sender_identity,
            )
            .await?;
        } else {
            self.send_one_shot_stream(
                header,
                text_bytes.utf8_aware_chunks(constants::STREAM_CHUNK_SIZE_BYTES),
                options.destination_identities,
                options.sender_identity,
            )
            .await?;
        }
        report_complete(&options.on_progress, total_length);
        Ok(info)
    }

    /// Send bytes to participants in the room.
    ///
    /// This method sends an in-memory blob of bytes to participants in the room
    /// as a byte stream. It opens a stream using the provided options, writes the
    /// entire buffer, and closes the stream before returning. Progress is reported once after
    /// transport acknowledgement.
    ///
    /// The `total_length` in the header is set from the provided data.
    /// The header defaults `name` to `"unknown"`
    /// and `mime_type` to `"application/octet-stream"`.
    pub async fn send_bytes(
        &self,
        data: impl AsRef<[u8]>,
        options: SendByteOptions,
        remote_participant_registry: &dyn RemoteParticipantRegistry,
    ) -> StreamResult<ByteStreamInfo> {
        let bytes = data.as_ref();
        let stream_id: StreamId = options.id.clone().unwrap_or_else(create_random_uuid).into();
        let name = options.name.clone().unwrap_or_else(|| constants::BYTE_DEFAULT_NAME.to_owned());
        let total_length = bytes.len() as u64;

        let eligibility =
            evaluate_eligibility(remote_participant_registry, &options.destination_identities);
        let can_compress = options.compress.unwrap_or(true) && eligibility.compression;

        let mut maybe_compressed = MaybeCollectedAsyncReader::from_async_reader(
            async_compression::futures::bufread::DeflateEncoder::new(
                futures_util::io::Cursor::new(bytes.to_vec()),
            ),
        );

        // Compress once up front when eligible (the deflate work happens at most once, cached in
        // the returned `Vec`), then decide whether it's worth using.
        let use_compression =
            can_compress && maybe_compressed.as_bytes().await.is_ok_and(|c| c.len() < bytes.len());

        // 1. Inline single-packet attempt (if all recipients are >= v2).
        let (mut header, byte_header) = if use_compression {
            build_byte_header(
                &options.topic,
                &options.attributes,
                options.mime_type.as_deref(),
                stream_id.clone(),
                name.clone(),
                Some(total_length), // NOTE: this is purposely always uncompressed length
                Some(maybe_compressed.as_bytes().await?.to_owned()),
                CompressionType::DeflateRaw,
            )
        } else {
            build_byte_header(
                &options.topic,
                &options.attributes,
                options.mime_type.as_deref(),
                stream_id.clone(),
                name.clone(),
                Some(total_length), // NOTE: this is purposely always uncompressed length
                Some(bytes.to_vec()),
                CompressionType::None,
            )
        };

        let proto_header = header.clone().into();
        if eligibility.inline && header_packet_fits(&proto_header, &options.destination_identities)
        {
            let mut packet =
                RawStream::create_header_packet(proto_header, options.destination_identities);
            packet.participant_identity =
                options.sender_identity.map(|id| id.into()).unwrap_or_default();
            RawStream::send_packet(&self.packet_tx, packet).await?;
            report_complete(&options.on_progress, total_length);
            return Ok(ByteStreamInfo::from_headers(header, byte_header));
        }

        // 2/3. Chunked, compressed when eligible else uncompressed. The entire stream — header,
        // chunks, trailer — goes out as one transport request.
        header.inline_content = None;
        enforce_header_size(&header, &options.destination_identities)?;

        let info = ByteStreamInfo::from_headers(header.clone(), byte_header);
        let content = if use_compression { maybe_compressed.as_bytes().await? } else { bytes };
        self.send_one_shot_stream(
            header,
            content.chunks(constants::STREAM_CHUNK_SIZE_BYTES),
            options.destination_identities,
            options.sender_identity,
        )
        .await?;
        report_complete(&options.on_progress, total_length);
        Ok(info)
    }

    /// Streams a file from disk to participants as a byte stream.
    ///
    /// Never uses the inline single-packet path (deciding inline-eligibility would require
    /// buffering and compressing the whole file up front). Compresses when every recipient
    /// supports it. The whole file is never buffered in memory at once.
    ///
    /// `options.on_progress` reports progress as described by [`SendByteOptions::on_progress`].
    pub async fn send_file(
        &self,
        path: impl AsRef<Path>,
        options: SendByteOptions,
        remote_participant_registry: &dyn RemoteParticipantRegistry,
    ) -> StreamResult<ByteStreamInfo> {
        let path = path.as_ref();
        let file_size = tokio::fs::metadata(path)
            .await
            .map(|metadata| metadata.len())
            .map_err(StreamError::from)?;
        let name = options.name.clone().unwrap_or_else(|| {
            path.file_name().and_then(|n| n.to_str()).unwrap_or_default().to_owned()
        });
        let stream_id: StreamId = options.id.clone().unwrap_or_else(create_random_uuid).into();
        let dests = options.destination_identities.clone();

        let eligibility = evaluate_eligibility(remote_participant_registry, &dests);
        let should_compress = options.compress.unwrap_or(true) && eligibility.compression;
        let compression =
            if should_compress { CompressionType::DeflateRaw } else { CompressionType::None };

        let (header, byte_header) = build_byte_header(
            &options.topic,
            &options.attributes,
            options.mime_type.as_deref(),
            stream_id,
            name,
            Some(file_size),
            None,
            compression,
        );
        enforce_header_size(&header, &dests)?;

        let open_options = RawStreamOpenOptions {
            header: header.clone(),
            destination_identities: dests,
            sender_identity: options.sender_identity.clone(),
            packet_tx: self.packet_tx.clone(),
        };
        let info = ByteStreamInfo::from_headers(header, byte_header);
        let mut stream = RawStream::open(open_options).await?;
        stream
            .write_file(path, should_compress, &mut |p| {
                if let Some(callback) = &options.on_progress {
                    callback.call(p);
                }
            })
            .await?;
        stream.close(None, None).await?;
        Ok(info)
    }

    /// Sends a complete one-shot stream — header, pre-split content chunks, trailer — as a
    /// single transport request acknowledged as a whole, rather than one request per packet.
    ///
    /// Only for sends whose full content is already in memory (`send_text`/`send_bytes`);
    /// incremental writers and `send_file` stream packet-by-packet instead.
    async fn send_one_shot_stream<'a>(
        &self,
        header: Header,
        chunks: impl IntoIterator<Item = &'a [u8]>,
        destination_identities: Vec<ParticipantIdentity>,
        sender_identity: Option<ParticipantIdentity>,
    ) -> StreamResult<()> {
        let stream_id = header.stream_id.to_string();
        let destinations: Vec<String> =
            destination_identities.iter().map(|id| id.0.clone()).collect();
        let mut packets =
            vec![RawStream::create_header_packet(header.into(), destination_identities)];
        packets.extend(chunks.into_iter().enumerate().map(|(index, chunk)| {
            RawStream::create_chunk_packet(&stream_id, index as u64, chunk, &destinations)
        }));
        packets.push(RawStream::create_trailer_packet(&stream_id, None, None, &destinations));
        if let Some(sender_identity) = sender_identity {
            let identity: String = sender_identity.into();
            for packet in &mut packets {
                packet.participant_identity = identity.clone();
            }
        }
        RawStream::send_packets(&self.packet_tx, packets).await
    }
}

/// Inline / compression eligibility evaluated over a send's recipients.
struct SendEligibility {
    /// Every recipient advertises `clientProtocol >= 2`.
    inline: bool,
    /// Inline-eligible AND every recipient advertises `CAP_COMPRESSION_DEFLATE_RAW`.
    compression: bool,
}

/// Evaluates inline/compression eligibility over a send's recipients.
///
/// Recipients are the named `destinations`, or every remote participant for a broadcast
/// (empty `destinations`). An empty recipient set (empty room) is eligible for everything.
fn evaluate_eligibility(
    registry: &dyn RemoteParticipantRegistry,
    destinations: &[ParticipantIdentity],
) -> SendEligibility {
    let recipients: Vec<ParticipantIdentity> =
        if destinations.is_empty() { registry.remote_identities() } else { destinations.to_vec() };
    let inline = recipients
        .iter()
        .all(|id| registry.remote_client_protocol(id) >= CLIENT_PROTOCOL_DATA_STREAM_V2);
    let compression = inline
        && recipients.iter().all(|id| {
            registry.remote_capabilities(id).contains(&ClientCapability::CompressionDeflateRaw)
        });

    SendEligibility { inline, compression }
}

/// Wraps an [`AsyncRead`] whose bytes are produced lazily (e.g. a deflate encoder), caching them
/// on first [`Self::collect`] so the underlying work runs at most once. Later `collect` calls
/// return the cached bytes without re-reading.
///
/// (The streaming, still-a-reader-after-collect variant needed by `send_file` will be added when
/// that path is migrated; today's callers only need `collect`.)
enum MaybeCollectedAsyncReader<Reader: futures_util::io::AsyncRead + Unpin> {
    Reader(Reader),
    Collected(Vec<u8>),
}

impl<Reader: futures_util::io::AsyncRead + Unpin> MaybeCollectedAsyncReader<Reader> {
    fn from_async_reader(reader: Reader) -> Self {
        Self::Reader(reader)
    }

    async fn as_bytes(&mut self) -> Result<&[u8], std::io::Error> {
        use futures_util::io::AsyncReadExt;
        match self {
            Self::Collected(_) => { /* no-op, handled below */ }
            Self::Reader(reader) => {
                let mut buf = Vec::new();
                reader.read_to_end(&mut buf).await?;
                *self = Self::Collected(buf);
            }
        }
        let Self::Collected(bytes) = self else { unreachable!("just set to Collected") };
        Ok(bytes)
    }
}

/// Whether the serialized header `DataPacket` fits within the MTU budget.
fn header_packet_fits(
    header: &proto::data_stream::Header,
    destinations: &[ParticipantIdentity],
) -> bool {
    use prost::Message;
    let packet = RawStream::create_header_packet(header.clone(), destinations.to_vec());
    packet.encoded_len() <= constants::STREAM_CHUNK_SIZE_BYTES
}

/// Enforces the header-packet MTU budget on the chunked path (the inline path falls back
/// gracefully instead of erroring).
fn enforce_header_size(header: &Header, destinations: &[ParticipantIdentity]) -> StreamResult<()> {
    let proto_header: proto::data_stream::Header = header.clone().into();
    if header_packet_fits(&proto_header, destinations) {
        Ok(())
    } else {
        Err(StreamError::HeaderTooLarge)
    }
}

fn report_complete(callback: &Option<StreamProgressCallback>, length: u64) {
    if let Some(callback) = callback {
        callback.call(StreamProgress {
            chunk_index: 0,
            bytes_processed: length,
            bytes_total: Some(length),
        });
    }
}

fn build_text_header(
    topic: &str,
    attributes: &HashMap<String, String>,
    text_header: TextHeader,
    stream_id: StreamId,
    total_length: Option<u64>,
    inline_content: Option<Vec<u8>>,
    compression: CompressionType,
) -> (Header, TextHeader) {
    let header = Header {
        stream_id,
        timestamp: Utc::now().timestamp_millis(),
        topic: topic.to_owned(),
        mime_type: constants::TEXT_MIME_TYPE.to_owned(),
        total_length,
        attributes: attributes.clone(),
        content_header: Some(ContentHeader::TextHeader(text_header.clone())),
        inline_content,
        compression,
    };
    (header, text_header)
}

#[allow(clippy::too_many_arguments)]
fn build_byte_header(
    topic: &str,
    attributes: &HashMap<String, String>,
    mime_type: Option<&str>,
    stream_id: StreamId,
    name: String,
    total_length: Option<u64>,
    inline_content: Option<Vec<u8>>,
    compression: CompressionType,
) -> (Header, ByteHeader) {
    let byte_header = ByteHeader { name };
    let header = Header {
        stream_id,
        timestamp: Utc::now().timestamp_millis(),
        topic: topic.to_owned(),
        mime_type: mime_type.unwrap_or(constants::BYTE_MIME_TYPE).to_owned(),
        total_length,
        attributes: attributes.clone(),
        content_header: Some(byte_header.clone().into()),
        inline_content,
        compression,
    };
    (header, byte_header)
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::{backend::somewhat_compressible, outgoing::StreamWriter};
    use livekit_common::{CLIENT_PROTOCOL_DATA_STREAM_RPC, CLIENT_PROTOCOL_DEFAULT};
    use std::{collections::HashMap, sync::Mutex as StdMutex};

    // --- Fake recipient registry ---------------------------------------------------------

    struct FakeRegistry {
        remotes: HashMap<String, (i32, Vec<ClientCapability>)>,
    }

    impl FakeRegistry {
        fn new() -> Self {
            Self { remotes: HashMap::new() }
        }

        fn add(mut self, id: &str, client_protocol: i32, caps: &[ClientCapability]) -> Self {
            self.remotes.insert(id.to_string(), (client_protocol, caps.to_vec()));
            self
        }
    }

    impl RemoteParticipantRegistry for FakeRegistry {
        fn remote_client_protocol(&self, identity: &ParticipantIdentity) -> i32 {
            self.remotes.get(&identity.0).map(|(p, _)| *p).unwrap_or(0)
        }
        fn remote_capabilities(&self, identity: &ParticipantIdentity) -> Vec<ClientCapability> {
            self.remotes.get(&identity.0).map(|(_, c)| c.clone()).unwrap_or_default()
        }
        fn remote_identities(&self) -> Vec<ParticipantIdentity> {
            self.remotes.keys().map(|k| ParticipantIdentity(k.clone())).collect()
        }
    }

    fn pre_v2_room() -> FakeRegistry {
        FakeRegistry::new()
            .add("alice", CLIENT_PROTOCOL_DEFAULT, &[])
            .add("bob", CLIENT_PROTOCOL_DEFAULT, &[])
            .add("jim", CLIENT_PROTOCOL_DATA_STREAM_RPC, &[])
    }

    fn all_v2_room() -> FakeRegistry {
        FakeRegistry::new()
            .add(
                "alice",
                CLIENT_PROTOCOL_DATA_STREAM_V2,
                &[ClientCapability::CompressionDeflateRaw],
            )
            .add("bob", CLIENT_PROTOCOL_DATA_STREAM_V2, &[ClientCapability::CompressionDeflateRaw])
            .add("noCompression", CLIENT_PROTOCOL_DATA_STREAM_V2, &[])
    }

    fn mixed_room() -> FakeRegistry {
        FakeRegistry::new()
            .add("alice", CLIENT_PROTOCOL_DEFAULT, &[])
            .add("bob", CLIENT_PROTOCOL_DATA_STREAM_V2, &[ClientCapability::CompressionDeflateRaw])
            .add("jim", CLIENT_PROTOCOL_DATA_STREAM_V2, &[ClientCapability::CompressionDeflateRaw])
            .add("mallory", CLIENT_PROTOCOL_DEFAULT, &[])
            .add("noCompression", CLIENT_PROTOCOL_DATA_STREAM_V2, &[])
    }

    /// A room where every participant is v2 AND advertises the compression capability.
    fn all_v2_capable_room() -> FakeRegistry {
        FakeRegistry::new()
            .add(
                "alice",
                CLIENT_PROTOCOL_DATA_STREAM_V2,
                &[ClientCapability::CompressionDeflateRaw],
            )
            .add("bob", CLIENT_PROTOCOL_DATA_STREAM_V2, &[ClientCapability::CompressionDeflateRaw])
    }

    // --- Capture harness -----------------------------------------------------------------

    type Sent = Arc<StdMutex<Vec<proto::DataPacket>>>;
    type SentBatches = Arc<StdMutex<Vec<Vec<proto::DataPacket>>>>;

    fn setup() -> (Manager, Sent) {
        let (manager, mut packet_rx) = Manager::new();
        let sent: Sent = Arc::new(StdMutex::new(Vec::new()));
        let sink = sent.clone();
        tokio::spawn(async move {
            while let Ok((packets, responder)) = packet_rx.recv().await {
                sink.lock().unwrap().extend(packets);
                let _ = responder.respond(Ok(()));
            }
        });
        (manager, sent)
    }

    /// Like [`setup`], but records the batch boundaries of each transport request.
    fn setup_batched() -> (Manager, SentBatches) {
        let (manager, mut packet_rx) = Manager::new();
        let sent: SentBatches = Arc::new(StdMutex::new(Vec::new()));
        let sink = sent.clone();
        tokio::spawn(async move {
            while let Ok((packets, responder)) = packet_rx.recv().await {
                sink.lock().unwrap().push(packets);
                let _ = responder.respond(Ok(()));
            }
        });
        (manager, sent)
    }

    fn ids(list: &[&str]) -> Vec<ParticipantIdentity> {
        list.iter().map(|s| ParticipantIdentity(s.to_string())).collect()
    }

    fn text_opts(topic: &str, dests: &[&str]) -> SendTextOptions {
        SendTextOptions::new_with_topic(topic).with_destination_identities(ids(dests))
    }

    fn byte_opts(topic: &str, dests: &[&str]) -> SendByteOptions {
        SendByteOptions::new_with_topic(topic).with_destination_identities(ids(dests))
    }

    fn stream_text_opts(topic: &str, dests: &[&str]) -> StreamTextOptions {
        StreamTextOptions::new_with_topic(topic).with_destination_identities(ids(dests))
    }

    fn stream_byte_opts(topic: &str, dests: &[&str]) -> StreamByteOptions {
        StreamByteOptions::new_with_topic(topic).with_destination_identities(ids(dests))
    }

    #[tokio::test]
    async fn one_shot_progress_follows_acknowledgement() {
        for registry in [pre_v2_room(), all_v2_room()] {
            for text in ["", "hello 🌍"] {
                for succeeds in [false, true] {
                    for is_text in [false, true] {
                        let (manager, mut packets) = Manager::new();
                        let reports = Arc::new(StdMutex::new(Vec::new()));
                        let callback_reports = reports.clone();
                        let callback = StreamProgressCallback::new(move |p| {
                            callback_reports.lock().unwrap().push(p);
                        });
                        let send = async {
                            if is_text {
                                let mut options = text_opts("chat", &[]);
                                options.on_progress = Some(callback);
                                manager.send_text(text, options, &registry).await.map(|_| ())
                            } else {
                                let mut options = byte_opts("blob", &[]);
                                options.on_progress = Some(callback);
                                manager
                                    .send_bytes(text.as_bytes(), options, &registry)
                                    .await
                                    .map(|_| ())
                            }
                        };
                        let acknowledge = async {
                            let (_, responder) = packets.recv().await.unwrap();
                            assert!(reports.lock().unwrap().is_empty());
                            responder
                                .respond(if succeeds { Ok(()) } else { Err(SendError) })
                                .unwrap();
                        };
                        let (result, ()) = tokio::join!(send, acknowledge);
                        assert_eq!(result.is_ok(), succeeds);
                        let reports = reports.lock().unwrap();
                        if succeeds {
                            assert_eq!(reports.len(), 1);
                            assert_eq!(reports[0].bytes_processed(), text.len() as u64);
                            assert_eq!(reports[0].bytes_total(), Some(text.len() as u64));
                        } else {
                            assert!(reports.is_empty());
                        }
                    }
                }
            }
        }
    }

    fn header(p: &proto::DataPacket) -> &proto::data_stream::Header {
        match p.value.as_ref().unwrap() {
            proto::data_packet::Value::StreamHeader(h) => h,
            _ => panic!("expected stream header"),
        }
    }

    fn chunk(p: &proto::DataPacket) -> &proto::data_stream::Chunk {
        match p.value.as_ref().unwrap() {
            proto::data_packet::Value::StreamChunk(c) => c,
            _ => panic!("expected stream chunk"),
        }
    }

    fn is_text_header(h: &proto::data_stream::Header) -> bool {
        matches!(h.content_header, Some(proto::data_stream::header::ContentHeader::TextHeader(_)))
    }

    fn is_byte_header(h: &proto::data_stream::Header) -> bool {
        matches!(h.content_header, Some(proto::data_stream::header::ContentHeader::ByteHeader(_)))
    }

    fn assert_trailer(p: &proto::DataPacket) {
        match p.value.as_ref().unwrap() {
            proto::data_packet::Value::StreamTrailer(t) => assert_eq!(t.reason, ""),
            _ => panic!("expected stream trailer"),
        }
    }

    /// Uniform random bytes — genuinely incompressible (unlike `pseudo_random_text`'s ascii).
    fn random_bytes(len: usize) -> Vec<u8> {
        use rand::{rngs::StdRng, Rng, SeedableRng};
        let mut rng = StdRng::seed_from_u64(0xdead_beef);
        (0..len).map(|_| rng.random::<u8>()).collect()
    }

    fn text_content_header(h: &proto::data_stream::Header) -> &proto::data_stream::TextHeader {
        match h.content_header.as_ref().unwrap() {
            proto::data_stream::header::ContentHeader::TextHeader(t) => t,
            _ => panic!("expected text header"),
        }
    }

    mod room_with_pre_data_streams_v2_participants {
        use super::*;

        #[tokio::test]
        async fn pre_v2_short_text_is_legacy_multipacket() {
            let (m, sent) = setup();
            m.send_text("hello world", text_opts("chat", &[]), &pre_v2_room()).await.unwrap();
            let p = sent.lock().unwrap().clone();
            assert_eq!(p.len(), 3);
            let h = header(&p[0]);
            assert!(is_text_header(h));
            assert_eq!(h.topic, "chat");
            assert_eq!(h.compression(), proto::data_stream::CompressionType::None);
            assert!(h.inline_content.is_none());
            let c = chunk(&p[1]);
            assert_eq!(c.chunk_index, 0);
            assert_eq!(c.content, b"hello world");
            assert_trailer(&p[2]);
        }

        #[tokio::test]
        async fn pre_v2_long_text_splits_at_mtu() {
            let (m, sent) = setup();
            let text = "A".repeat(40_000);
            m.send_text(&text, text_opts("chat", &[]), &pre_v2_room()).await.unwrap();
            let p = sent.lock().unwrap().clone();
            assert_eq!(p.len(), 5); // header + 3 chunks + trailer
            assert_eq!(header(&p[0]).compression(), proto::data_stream::CompressionType::None);
            assert_eq!(chunk(&p[1]).content.len(), 15_000);
            assert_eq!(chunk(&p[2]).content.len(), 15_000);
            assert_eq!(chunk(&p[3]).content.len(), 10_000);
            assert_eq!(chunk(&p[1]).chunk_index, 0);
            assert_eq!(chunk(&p[3]).chunk_index, 2);
            assert_trailer(&p[4]);
        }

        #[tokio::test]
        async fn pre_v2_bytes_is_legacy_multipacket() {
            let (m, sent) = setup();
            m.send_bytes([0u8, 1, 2, 3], byte_opts("blob", &[]), &pre_v2_room()).await.unwrap();
            let p = sent.lock().unwrap().clone();
            assert_eq!(p.len(), 3);
            let h = header(&p[0]);
            assert!(is_byte_header(h));
            assert_eq!(h.compression(), proto::data_stream::CompressionType::None);
            assert!(h.inline_content.is_none());
            assert_eq!(chunk(&p[1]).content, vec![0, 1, 2, 3]);
            assert_trailer(&p[2]);
        }

        #[tokio::test]
        async fn pre_v2_empty_text_sends_header_and_trailer() {
            let (m, sent) = setup();
            m.send_text("", text_opts("chat", &[]), &pre_v2_room()).await.unwrap();
            // A well-formed (if contentless) stream: header declaring zero length, then the trailer.
            let p = sent.lock().unwrap().clone();
            assert_eq!(p.len(), 2);
            assert_eq!(header(&p[0]).total_length, Some(0));
            assert_trailer(&p[1]);
        }
    }

    mod room_with_all_data_streams_v2_participants {
        use super::*;

        mod send_text {
            use super::*;

            #[tokio::test]
            async fn v2_short_compressible_text_inlines_compressed() {
                let (m, sent) = setup();
                let text = "hello hello compressible world";
                m.send_text(text, text_opts("chat", &["alice", "bob"]), &all_v2_room())
                    .await
                    .unwrap();
                let p = sent.lock().unwrap().clone();
                assert_eq!(p.len(), 1);
                let h = header(&p[0]);
                assert!(is_text_header(h));
                assert_eq!(h.compression(), proto::data_stream::CompressionType::DeflateRaw);
                let inline = h.inline_content.as_ref().unwrap();
                assert_ne!(inline.as_slice(), text.as_bytes()); // compressed, not raw
            }

            #[tokio::test]
            async fn v2_short_incompressible_text_inlines_raw() {
                let (m, sent) = setup();
                m.send_text("short", text_opts("chat", &["alice", "bob"]), &all_v2_room())
                    .await
                    .unwrap();
                let p = sent.lock().unwrap().clone();
                assert_eq!(p.len(), 1);
                let h = header(&p[0]);
                assert_eq!(h.compression(), proto::data_stream::CompressionType::None);
                assert_eq!(h.inline_content.as_ref().unwrap().as_slice(), b"short");
            }

            #[tokio::test]
            async fn v2_no_compression_cap_inlines_raw() {
                let (m, sent) = setup();
                let text = "hello hello compressible world";
                m.send_text(text, text_opts("chat", &["noCompression"]), &all_v2_room())
                    .await
                    .unwrap();
                let p = sent.lock().unwrap().clone();
                assert_eq!(p.len(), 1); // inline (gated on protocol) still happens
                let h = header(&p[0]);
                assert_eq!(h.compression(), proto::data_stream::CompressionType::None); // compression gated off by missing cap
                assert_eq!(h.inline_content.as_ref().unwrap().as_slice(), text.as_bytes());
            }

            #[tokio::test]
            async fn v2_large_highly_compressible_text_still_inlines() {
                let (m, sent) = setup();
                let text = "hello world".repeat(20_000);
                m.send_text(&text, text_opts("chat", &["alice", "bob"]), &all_v2_room())
                    .await
                    .unwrap();
                let p = sent.lock().unwrap().clone();
                assert_eq!(p.len(), 1);
                let h = header(&p[0]);
                assert_eq!(h.compression(), proto::data_stream::CompressionType::DeflateRaw);
                assert!(h.inline_content.as_ref().unwrap().len() < text.len());
            }

            #[tokio::test]
            async fn v2_somewhat_compressible_text_is_compressed_multipacket() {
                let (m, sent) = setup();
                let text = somewhat_compressible(50_000);
                m.send_text(&text, text_opts("chat", &["alice", "bob"]), &all_v2_room())
                    .await
                    .unwrap();
                let p = sent.lock().unwrap().clone();
                let h = header(&p[0]);
                assert_eq!(h.compression(), proto::data_stream::CompressionType::DeflateRaw);
                assert!(h.inline_content.is_none());
                let chunks: Vec<_> = p[1..p.len() - 1].iter().map(chunk).collect();
                // Multi-packet, but fewer chunks than an uncompressed send would need (ceil(len/15000)).
                let uncompressed_chunks = text.len().div_ceil(constants::STREAM_CHUNK_SIZE_BYTES);
                assert!(chunks.len() >= 2);
                assert!(chunks.len() < uncompressed_chunks);
                assert_eq!(chunks[0].content.len(), constants::STREAM_CHUNK_SIZE_BYTES); // first chunk is full MTU
                let total: usize = chunks.iter().map(|c| c.content.len()).sum();
                assert!(total < text.len()); // compressed
                assert_trailer(p.last().unwrap());
            }

            #[tokio::test]
            async fn v2_compress_false_short_inlines_raw() {
                let (m, sent) = setup();
                let text = "hello hello compressible world";
                let opts = text_opts("chat", &["alice", "bob"]).with_compress(false);
                m.send_text(text, opts, &all_v2_room()).await.unwrap();
                let p = sent.lock().unwrap().clone();
                assert_eq!(p.len(), 1);
                let h = header(&p[0]);
                assert_eq!(h.compression(), proto::data_stream::CompressionType::None);
                assert_eq!(h.inline_content.as_ref().unwrap().as_slice(), text.as_bytes());
            }

            #[tokio::test]
            async fn v2_compress_false_large_is_uncompressed_multipacket() {
                let (m, sent) = setup();
                let text = "B".repeat(50_000);
                let opts = text_opts("chat", &["alice", "bob"]).with_compress(false);
                m.send_text(&text, opts, &all_v2_room()).await.unwrap();
                let p = sent.lock().unwrap().clone();
                assert_eq!(p.len(), 6); // header + 4 chunks + trailer
                assert_eq!(header(&p[0]).compression(), proto::data_stream::CompressionType::None);
                assert_eq!(chunk(&p[1]).content.len(), 15_000);
            }

            #[tokio::test]
            async fn v2_send_text_with_attachments_never_inlines() {
                let (m, sent) = setup();
                let opts = text_opts("chat", &["alice", "bob"]).with_attached_stream_id("att1");
                m.send_text("hello hello compressible world", opts, &all_v2_room()).await.unwrap();
                let p = sent.lock().unwrap().clone();
                // Attachments disable the single-packet path even when every recipient is v2.
                assert_eq!(p.len(), 3); // header + 1 chunk + trailer
                let h = header(&p[0]);
                assert!(h.inline_content.is_none());
                assert_eq!(text_content_header(h).attached_stream_ids, vec!["att1".to_string()]);
                assert_trailer(&p[2]);
            }

            #[tokio::test]
            async fn v2_large_text_to_uncapable_recipient_is_uncompressed_multipacket() {
                let (m, sent) = setup();
                let text = "A".repeat(40_000);
                m.send_text(&text, text_opts("chat", &["noCompression"]), &all_v2_room())
                    .await
                    .unwrap();
                // Inline is attempted (recipient is v2) but the raw payload overflows the MTU, and
                // compression is gated off by the missing capability — so this falls back to an
                // uncompressed multi-packet stream split at STREAM_CHUNK_SIZE_BYTES.
                let p = sent.lock().unwrap().clone();
                assert_eq!(p.len(), 5);
                let h = header(&p[0]);
                assert_eq!(h.compression(), proto::data_stream::CompressionType::None);
                assert!(h.inline_content.is_none());
                assert_eq!(chunk(&p[1]).content.len(), 15_000);
                assert_eq!(chunk(&p[2]).content.len(), 15_000);
                assert_eq!(chunk(&p[3]).content.len(), 10_000);
                assert_trailer(&p[4]);
            }

            #[tokio::test]
            async fn v2_empty_text_sends_single_inline_packet() {
                let (m, sent) = setup();
                m.send_text("", text_opts("chat", &["alice", "bob"]), &all_v2_room())
                    .await
                    .unwrap();
                let p = sent.lock().unwrap().clone();
                assert_eq!(p.len(), 1);
                let h = header(&p[0]);
                // Deflate framing can't shrink an empty payload, so the raw (empty) bytes are kept.
                assert_eq!(h.compression(), proto::data_stream::CompressionType::None);
                assert_eq!(h.inline_content.as_deref(), Some(&[][..]));
                assert_eq!(h.total_length, Some(0));
            }
        }

        mod send_bytes {
            use super::*;

            #[tokio::test]
            async fn v2_send_bytes_short_incompressible_inlines_raw() {
                let (m, sent) = setup();
                m.send_bytes([0u8, 1, 2, 3], byte_opts("blob", &["alice", "bob"]), &all_v2_room())
                    .await
                    .unwrap();
                let p = sent.lock().unwrap().clone();
                assert_eq!(p.len(), 1);
                let h = header(&p[0]);
                // Tiny payload doesn't shrink under DEFLATE framing, so it's sent raw.
                assert_eq!(h.compression(), proto::data_stream::CompressionType::None);
                assert_eq!(h.inline_content.as_ref().unwrap().as_slice(), &[0u8, 1, 2, 3]);
            }

            #[tokio::test]
            async fn v2_send_bytes_no_compression_cap_inlines_raw() {
                let (m, sent) = setup();
                let payload = "hello hello compressible world".as_bytes();
                m.send_bytes(payload, byte_opts("blob", &["noCompression"]), &all_v2_room())
                    .await
                    .unwrap();
                let p = sent.lock().unwrap().clone();
                assert_eq!(p.len(), 1); // inline is gated on clientProtocol alone
                let h = header(&p[0]);
                assert_eq!(h.compression(), proto::data_stream::CompressionType::None); // compression gated off by the missing cap
                assert_eq!(h.inline_content.as_ref().unwrap().as_slice(), payload);
            }

            #[tokio::test]
            async fn v2_send_bytes_large_highly_compressible_inlines() {
                let (m, sent) = setup();
                let payload = vec![0x01u8; 50_000];
                m.send_bytes(&payload, byte_opts("blob", &["alice", "bob"]), &all_v2_room())
                    .await
                    .unwrap();
                let p = sent.lock().unwrap().clone();
                assert_eq!(p.len(), 1); // compresses well under the MTU, so it still goes inline
                let h = header(&p[0]);
                assert_eq!(h.compression(), proto::data_stream::CompressionType::DeflateRaw);
                assert!(h.inline_content.as_ref().unwrap().len() < payload.len());
                assert_eq!(h.total_length, Some(50_000)); // always the PRE-compression length
            }

            #[tokio::test]
            async fn v2_send_bytes_somewhat_compressible_is_compressed_multipacket() {
                let (m, sent) = setup();
                let payload = somewhat_compressible(50_000).into_bytes();
                m.send_bytes(&payload, byte_opts("blob", &["alice", "bob"]), &all_v2_room())
                    .await
                    .unwrap();
                let p = sent.lock().unwrap().clone();
                let h = header(&p[0]);
                assert!(is_byte_header(h));
                assert_eq!(h.compression(), proto::data_stream::CompressionType::DeflateRaw);
                assert!(h.inline_content.is_none());
                let chunks: Vec<_> = p[1..p.len() - 1].iter().map(chunk).collect();
                assert!(chunks.len() >= 2);
                assert!(chunks.len() < payload.len().div_ceil(constants::STREAM_CHUNK_SIZE_BYTES));
                assert_eq!(chunks[0].content.len(), constants::STREAM_CHUNK_SIZE_BYTES);
                assert_trailer(p.last().unwrap());
            }

            #[tokio::test]
            async fn v2_send_bytes_compress_false_large_is_uncompressed_multipacket() {
                let (m, sent) = setup();
                let payload = vec![0x07u8; 40_000];
                let opts = byte_opts("blob", &["alice", "bob"]).with_compress(false);
                m.send_bytes(&payload, opts, &all_v2_room()).await.unwrap();
                let p = sent.lock().unwrap().clone();
                assert_eq!(p.len(), 5); // header + 15k/15k/10k chunks + trailer
                let h = header(&p[0]);
                assert_eq!(h.compression(), proto::data_stream::CompressionType::None);
                assert!(h.inline_content.is_none());
                assert_eq!(chunk(&p[1]).content.len(), 15_000);
                assert_eq!(chunk(&p[2]).content.len(), 15_000);
                assert_eq!(chunk(&p[3]).content.len(), 10_000);
                assert!(chunk(&p[3]).content.iter().all(|b| *b == 0x07));
                assert_trailer(&p[4]);
            }

            #[tokio::test]
            async fn v2_send_bytes_short_compressible_inlines_compressed() {
                let (m, sent) = setup();
                let payload = "hello hello compressible world".as_bytes().to_vec();
                let mut opts = byte_opts("blob", &["alice", "bob"]);
                opts.attributes.insert("foo".to_string(), "bar".to_string());
                let info = m.send_bytes(&payload, opts, &all_v2_room()).await.unwrap();
                let p = sent.lock().unwrap().clone();
                assert_eq!(p.len(), 1);
                let h = header(&p[0]);
                assert!(is_byte_header(h));
                assert_eq!(h.compression(), proto::data_stream::CompressionType::DeflateRaw);
                assert_ne!(h.inline_content.as_ref().unwrap().as_slice(), payload.as_slice());
                assert_eq!(info.name, "unknown");
                assert_eq!(info.mime_type, "application/octet-stream");
                assert_eq!(info.total_length, Some(payload.len() as u64));
                assert_eq!(info.attributes().get("foo"), Some(&"bar".to_string()));
            }
        }

        mod send_file {
            use super::*;

            async fn write_temp_file(bytes: &[u8]) -> std::path::PathBuf {
                let path =
                    std::env::temp_dir().join(format!("lk_ds_test_{}.bin", create_random_uuid()));
                tokio::fs::write(&path, bytes).await.unwrap();
                path
            }

            #[tokio::test]
            async fn send_file_never_inlines_and_compresses_when_eligible() {
                let (m, sent) = setup();
                let path = write_temp_file(&vec![0x01u8; 10_000]).await;
                m.send_file(&path, byte_opts("file", &["alice", "bob"]), &all_v2_room())
                    .await
                    .unwrap();
                let _ = tokio::fs::remove_file(&path).await;
                let p = sent.lock().unwrap().clone();
                assert_eq!(p.len(), 3); // header + 1 chunk + trailer, NOT inline
                let h = header(&p[0]);
                assert!(is_byte_header(h));
                assert_eq!(h.compression(), proto::data_stream::CompressionType::DeflateRaw);
                assert!(h.inline_content.is_none());
                assert!(chunk(&p[1]).content.len() < 10_000); // compressed
                assert_trailer(&p[2]);
            }

            #[tokio::test]
            async fn send_file_uncompressed_splits_at_mtu() {
                let (m, sent) = setup();
                let path = write_temp_file(&vec![0x07u8; 20_000]).await;
                m.send_file(&path, byte_opts("file", &[]).with_compress(false), &all_v2_room())
                    .await
                    .unwrap();
                let _ = tokio::fs::remove_file(&path).await;
                let p = sent.lock().unwrap().clone();
                assert_eq!(p.len(), 4); // header + 15000 + 5000 + trailer
                assert_eq!(header(&p[0]).compression(), proto::data_stream::CompressionType::None);
                assert_eq!(chunk(&p[1]).content.len(), 15_000);
                assert_eq!(chunk(&p[2]).content.len(), 5_000);
                assert_eq!(chunk(&p[2]).chunk_index, 1);
                assert_trailer(&p[3]);
            }

            #[tokio::test]
            async fn send_file_incompressible_compressed_expands() {
                let (m, sent) = setup();
                let data = random_bytes(50_000);
                let path = write_temp_file(&data).await;
                m.send_file(&path, byte_opts("file", &["alice", "bob"]), &all_v2_room())
                    .await
                    .unwrap();
                let _ = tokio::fs::remove_file(&path).await;

                let p = sent.lock().unwrap().clone();
                assert_eq!(p.len(), 6); // header + 4 chunks + trailer
                let h = header(&p[0]);
                assert_eq!(h.compression(), proto::data_stream::CompressionType::DeflateRaw);
                assert_eq!(chunk(&p[1]).content.len(), 15_000);
                // Deflate adds slight overhead on incompressible data — the accepted trade-off for
                // streaming the file instead of buffering it to make an inline/compression decision.
                let total: usize =
                    p[1..p.len() - 1].iter().map(|packet| chunk(packet).content.len()).sum();
                assert!(total > data.len());
                assert_trailer(p.last().unwrap());
            }

            #[tokio::test]
            async fn send_file_to_no_compression_recipient_is_uncompressed() {
                let (m, sent) = setup();
                let path = write_temp_file(&vec![0x07u8; 10_000]).await;
                m.send_file(&path, byte_opts("file", &["noCompression"]), &all_v2_room())
                    .await
                    .unwrap();
                let _ = tokio::fs::remove_file(&path).await;

                // v2 recipient without the compression capability in a v2 room: uncompressed
                // multi-packet (and never inline).
                let p = sent.lock().unwrap().clone();
                assert_eq!(p.len(), 3);
                let h = header(&p[0]);
                assert!(is_byte_header(h));
                assert_eq!(h.compression(), proto::data_stream::CompressionType::None);
                assert!(h.inline_content.is_none());
                let c = chunk(&p[1]);
                assert_eq!(c.content.len(), 10_000);
                assert!(c.content.iter().all(|b| *b == 0x07));
                assert_trailer(&p[2]);
            }

            /// Sends `bytes` as a file to a compression-capable recipient and returns every
            /// progress report as `(bytes_processed, bytes_total)`.
            async fn send_file_progress(bytes: &[u8], compress: bool) -> Vec<(u64, Option<u64>)> {
                let (m, _sent) = setup();
                let path = write_temp_file(bytes).await;
                let reports = Arc::new(StdMutex::new(Vec::new()));
                let callback_reports = reports.clone();
                m.send_file(
                    &path,
                    byte_opts("file", &["alice"]).with_compress(compress).with_on_progress(
                        move |p| {
                            callback_reports
                                .lock()
                                .unwrap()
                                .push((p.bytes_processed(), p.bytes_total()));
                        },
                    ),
                    &all_v2_room(),
                )
                .await
                .unwrap();
                let _ = tokio::fs::remove_file(&path).await;
                let reports = std::mem::take(&mut *reports.lock().unwrap());
                reports
            }

            #[tokio::test]
            async fn send_file_reports_progress_after_each_chunk() {
                let reports = send_file_progress(&vec![0x07u8; 20_000], false).await;
                assert_eq!(reports, vec![(15_000, Some(20_000)), (20_000, Some(20_000))]);
            }

            /// What the transport and the progress callback saw, in order.
            #[derive(Debug, PartialEq)]
            enum Seen {
                Chunk,
                Progress(u64),
            }

            /// Sends `bytes` as a compressed file over a transport that fails request
            /// `fail_request` (request 1 is the header). Returns what was seen, in order, and the
            /// result of the send.
            async fn compressed_send_file(
                bytes: &[u8],
                fail_request: Option<usize>,
            ) -> (Vec<Seen>, StreamResult<ByteStreamInfo>) {
                let (m, mut packet_rx) = Manager::new();
                let seen = Arc::new(StdMutex::new(Vec::new()));
                let transport_seen = seen.clone();
                tokio::spawn(async move {
                    let mut request = 0;
                    while let Ok((packets, responder)) = packet_rx.recv().await {
                        request += 1;
                        if Some(request) == fail_request {
                            let _ = responder.respond(Err(SendError));
                            continue;
                        }
                        let is_chunk = |p: &proto::DataPacket| {
                            matches!(p.value, Some(proto::data_packet::Value::StreamChunk(_)))
                        };
                        if packets.iter().any(is_chunk) {
                            transport_seen.lock().unwrap().push(Seen::Chunk);
                        }
                        let _ = responder.respond(Ok(()));
                    }
                });
                let path = write_temp_file(bytes).await;
                let progress_seen = seen.clone();
                let result = m
                    .send_file(
                        &path,
                        byte_opts("file", &["alice"]).with_on_progress(move |p| {
                            progress_seen.lock().unwrap().push(Seen::Progress(p.bytes_processed()));
                        }),
                        &all_v2_room(),
                    )
                    .await;
                let _ = tokio::fs::remove_file(&path).await;
                let seen = std::mem::take(&mut *seen.lock().unwrap());
                (seen, result)
            }

            #[tokio::test]
            async fn compressed_send_file_reports_rising_progress_and_the_total_last() {
                // Incompressible, so the compressed stream spans several chunks.
                let (seen, result) = compressed_send_file(&random_bytes(50_000), None).await;
                result.unwrap();
                let reports: Vec<u64> = seen
                    .iter()
                    .filter_map(|s| match s {
                        Seen::Progress(bytes) => Some(*bytes),
                        Seen::Chunk => None,
                    })
                    .collect();
                assert!(reports.len() >= 2, "{seen:?}");
                assert!(reports.windows(2).all(|pair| pair[0] < pair[1]), "{seen:?}");
                // The total is reported once, after the last chunk was sent.
                assert_eq!(seen.last(), Some(&Seen::Progress(50_000)), "{seen:?}");
                assert_eq!(seen[seen.len() - 2], Seen::Chunk, "{seen:?}");
            }

            #[tokio::test]
            async fn compressed_send_file_without_its_last_chunk_never_reports_the_total() {
                let data = random_bytes(50_000);
                let (seen, _) = compressed_send_file(&data, None).await;
                let chunks = seen.iter().filter(|s| **s == Seen::Chunk).count();

                // Request 1 is the header, so request `1 + chunks` carries the last chunk.
                let (seen, result) = compressed_send_file(&data, Some(1 + chunks)).await;
                assert!(matches!(result, Err(StreamError::SendFailed)));
                assert!(!seen.contains(&Seen::Progress(50_000)), "{seen:?}");
            }

            #[tokio::test]
            async fn send_file_reports_progress_for_empty_file() {
                assert_eq!(send_file_progress(&[], false).await, vec![(0, Some(0))]);
                assert_eq!(send_file_progress(&[], true).await, vec![(0, Some(0))]);
            }

            #[tokio::test]
            async fn send_file_empty_file() {
                let (m, sent) = setup();
                let path = write_temp_file(&[]).await;
                m.send_file(&path, byte_opts("file", &["alice", "bob"]), &all_v2_room())
                    .await
                    .unwrap();
                let _ = tokio::fs::remove_file(&path).await;

                // An empty file still produces a well-formed compressed stream: a header declaring zero
                // length, the deflate stream's final block, and a trailer.
                let p = sent.lock().unwrap().clone();
                let h = header(&p[0]);
                assert!(is_byte_header(h));
                assert_eq!(h.total_length, Some(0));
                assert_eq!(h.compression(), proto::data_stream::CompressionType::DeflateRaw);
                assert_trailer(p.last().unwrap());
            }
        }

        #[tokio::test]
        async fn v2_broadcast_all_capable_inlines_compressed() {
            let (m, sent) = setup();
            // No destinations: eligibility is evaluated over every remote participant.
            m.send_text(
                "hello hello compressible world",
                text_opts("chat", &[]),
                &all_v2_capable_room(),
            )
            .await
            .unwrap();
            let p = sent.lock().unwrap().clone();
            assert_eq!(p.len(), 1);
            assert_eq!(
                header(&p[0]).compression(),
                proto::data_stream::CompressionType::DeflateRaw
            );
        }

        #[tokio::test]
        async fn v2_broadcast_with_uncapable_member_inlines_raw() {
            let (m, sent) = setup();
            let text = "hello hello compressible world";
            // all_v2_room includes "noCompression": inline still applies, compression does not.
            m.send_text(text, text_opts("chat", &[]), &all_v2_room()).await.unwrap();
            let p = sent.lock().unwrap().clone();
            assert_eq!(p.len(), 1);
            let h = header(&p[0]);
            assert_eq!(h.compression(), proto::data_stream::CompressionType::None);
            assert_eq!(h.inline_content.as_ref().unwrap().as_slice(), text.as_bytes());
        }

        #[tokio::test]
        async fn empty_room_broadcast_is_v2_eligible() {
            let (m, sent) = setup();
            // Nobody to receive it, so nothing gates the v2 fast path.
            m.send_text(
                "hello hello compressible world",
                text_opts("chat", &[]),
                &FakeRegistry::new(),
            )
            .await
            .unwrap();
            let p = sent.lock().unwrap().clone();
            assert_eq!(p.len(), 1);
            assert_eq!(
                header(&p[0]).compression(),
                proto::data_stream::CompressionType::DeflateRaw
            );
        }
    }

    mod room_with_mixed_participants {
        use super::*;

        #[tokio::test]
        async fn mixed_broadcast_falls_back_to_legacy() {
            let (m, sent) = setup();
            m.send_text("hello world", text_opts("chat", &[]), &mixed_room()).await.unwrap();
            let p = sent.lock().unwrap().clone();
            assert_eq!(p.len(), 3);
            assert_eq!(header(&p[0]).compression(), proto::data_stream::CompressionType::None);
            assert!(header(&p[0]).inline_content.is_none());
            assert_eq!(chunk(&p[1]).content, b"hello world");
        }

        #[tokio::test]
        async fn mixed_targeted_v2_subset_inlines_compressed() {
            let (m, sent) = setup();
            let text = "hello hello compressible world";
            m.send_text(text, text_opts("chat", &["bob", "jim"]), &mixed_room()).await.unwrap();
            let p = sent.lock().unwrap().clone();
            assert_eq!(p.len(), 1);
            let h = header(&p[0]);
            assert_eq!(h.compression(), proto::data_stream::CompressionType::DeflateRaw);
            assert_ne!(h.inline_content.as_ref().unwrap().as_slice(), text.as_bytes());
        }

        #[tokio::test]
        async fn mixed_targeted_subset_missing_cap_inlines_uncompressed() {
            let (m, sent) = setup();
            let text = "hello hello compressible world";
            m.send_text(text, text_opts("chat", &["bob", "jim", "noCompression"]), &mixed_room())
                .await
                .unwrap();
            let p = sent.lock().unwrap().clone();
            assert_eq!(p.len(), 1);
            let h = header(&p[0]);
            assert_eq!(h.compression(), proto::data_stream::CompressionType::None);
            assert_eq!(h.inline_content.as_ref().unwrap().as_slice(), text.as_bytes());
        }
    }

    // --- Sender identity ------------------------------------------------------------------

    #[tokio::test]
    async fn stream_text_with_sender_identity_stamps_every_packet() {
        let (m, sent) = setup();
        let opts = stream_text_opts("chat", &[]).with_sender_identity("impostor");
        let writer = m.stream_text(opts).await.unwrap();
        writer.write("hello").await.unwrap();
        writer.close().await.unwrap();
        let p = sent.lock().unwrap().clone();
        assert_eq!(p.len(), 3);
        assert!(p.iter().all(|pkt| pkt.participant_identity == "impostor"));
    }

    #[tokio::test]
    async fn send_text_inline_with_sender_identity_stamps_packet() {
        let (m, sent) = setup();
        let opts = text_opts("chat", &["alice", "bob"]).with_sender_identity("impostor");
        m.send_text("hello hello compressible world", opts, &all_v2_room()).await.unwrap();
        let p = sent.lock().unwrap().clone();
        assert_eq!(p.len(), 1);
        assert_eq!(p[0].participant_identity, "impostor");
    }

    #[tokio::test]
    async fn packets_carry_no_identity_when_sender_identity_unset() {
        let (m, sent) = setup();
        let writer = m.stream_text(stream_text_opts("chat", &[])).await.unwrap();
        writer.close().await.unwrap();
        let p = sent.lock().unwrap().clone();
        assert!(p.iter().all(|pkt| pkt.participant_identity.is_empty()));
    }

    // --- Destination identities -----------------------------------------------------------

    // The server routes each packet on its own `destination_identities`, so a chunk or
    // trailer without them would be broadcast to the whole room.

    #[tokio::test]
    async fn one_shot_send_text_targets_every_packet() {
        let (m, sent) = setup();
        let text = "A".repeat(20_000);
        m.send_text(&text, text_opts("chat", &["alice", "bob"]), &pre_v2_room()).await.unwrap();
        let p = sent.lock().unwrap().clone();
        assert_eq!(p.len(), 4); // header + 2 chunks + trailer
        assert!(p.iter().all(|pkt| pkt.destination_identities == ["alice", "bob"]));
    }

    #[tokio::test]
    async fn stream_text_targets_every_packet() {
        let (m, sent) = setup();
        let writer = m.stream_text(stream_text_opts("chat", &["alice"])).await.unwrap();
        writer.write("hello").await.unwrap();
        writer.close().await.unwrap();
        let p = sent.lock().unwrap().clone();
        assert_eq!(p.len(), 3);
        assert!(p.iter().all(|pkt| pkt.destination_identities == ["alice"]));
    }

    #[tokio::test]
    async fn send_file_targets_every_packet() {
        let (m, sent) = setup();
        let path = std::env::temp_dir().join(format!("lk_ds_dest_{}.bin", create_random_uuid()));
        tokio::fs::write(&path, vec![0x07u8; 20_000]).await.unwrap();
        let opts = byte_opts("file", &["bob"]).with_compress(false);
        m.send_file(&path, opts, &all_v2_room()).await.unwrap();
        let _ = tokio::fs::remove_file(&path).await;
        let p = sent.lock().unwrap().clone();
        assert_eq!(p.len(), 4); // header + 2 chunks + trailer
        assert!(p.iter().all(|pkt| pkt.destination_identities == ["bob"]));
    }

    #[tokio::test]
    async fn compressed_send_file_progress_targets_every_packet() {
        let (m, sent) = setup();
        let path = std::env::temp_dir().join(format!("lk_ds_dest_{}.bin", create_random_uuid()));
        // Incompressible, so chunks go out both while the file is read and in the tail.
        tokio::fs::write(&path, random_bytes(50_000)).await.unwrap();
        let reports = Arc::new(StdMutex::new(0));
        let callback_reports = reports.clone();
        m.send_file(
            &path,
            byte_opts("file", &["alice"]).with_on_progress(move |_| {
                *callback_reports.lock().unwrap() += 1;
            }),
            &all_v2_room(),
        )
        .await
        .unwrap();
        let _ = tokio::fs::remove_file(&path).await;
        let p = sent.lock().unwrap().clone();
        assert_eq!(header(&p[0]).compression(), proto::data_stream::CompressionType::DeflateRaw);
        assert!(p.len() >= 5, "header, at least 3 chunks, trailer: {}", p.len());
        assert_trailer(p.last().unwrap());
        assert!(p.iter().all(|pkt| pkt.destination_identities == ["alice"]));
        assert!(*reports.lock().unwrap() >= 2);
    }

    // --- Incremental writers never compress or inline ------------------------------------

    #[tokio::test]
    async fn stream_text_never_compresses_or_inlines() {
        let (m, sent) = setup();
        let writer = m.stream_text(stream_text_opts("chat", &["noCompression"])).await.unwrap();
        assert_eq!(sent.lock().unwrap().len(), 1);
        let h0 = sent.lock().unwrap()[0].clone();
        assert!(is_text_header(header(&h0)));
        assert_eq!(header(&h0).compression(), proto::data_stream::CompressionType::None);
        assert!(header(&h0).inline_content.is_none());

        writer.write("hello world").await.unwrap();
        assert_eq!(sent.lock().unwrap().len(), 2);
        assert_eq!(chunk(&sent.lock().unwrap()[1]).content, b"hello world");

        writer.close().await.unwrap();
        let p = sent.lock().unwrap().clone();
        assert_eq!(p.len(), 3);
        assert_trailer(&p[2]);
    }

    #[tokio::test]
    async fn stream_text_declares_total_length() {
        let (m, sent) = setup();
        let writer =
            m.stream_text(stream_text_opts("chat", &[]).with_total_length(5)).await.unwrap();
        assert_eq!(writer.info().total_length, Some(5));
        assert_eq!(header(&sent.lock().unwrap()[0]).total_length, Some(5));
    }

    #[tokio::test]
    async fn stream_bytes_never_compresses_or_inlines() {
        let (m, sent) = setup();
        let writer = m.stream_bytes(stream_byte_opts("blob", &["noCompression"])).await.unwrap();
        assert_eq!(sent.lock().unwrap().len(), 1);
        assert_eq!(
            header(&sent.lock().unwrap()[0]).compression(),
            proto::data_stream::CompressionType::None
        );

        writer.write(&[0u8, 1, 2, 3]).await.unwrap();
        assert_eq!(chunk(&sent.lock().unwrap()[1]).content, vec![0, 1, 2, 3]);

        writer.close().await.unwrap();
        let p = sent.lock().unwrap().clone();
        assert_eq!(p.len(), 3);
        assert_trailer(&p[2]);
    }

    mod header_size_limit {
        use super::*;

        #[tokio::test]
        async fn oversized_attributes_on_chunked_path_errors() {
            let (m, _sent) = setup();
            let mut opts = text_opts("chat", &[]); // pre-v2 below => chunked path
            opts.attributes.insert("big".to_string(), "x".repeat(20_000));
            let result = m.send_text("hello", opts, &pre_v2_room()).await;
            assert!(matches!(result, Err(StreamError::HeaderTooLarge)));
        }

        /// Builds a minimal text header carrying the given attributes.
        fn header_with_attributes(attributes: HashMap<String, String>) -> Header {
            Header {
                stream_id: "s1".into(),
                timestamp: 0,
                topic: "chat".to_string(),
                mime_type: constants::TEXT_MIME_TYPE.to_owned(),
                total_length: None,
                attributes,
                content_header: Some(ContentHeader::TextHeader(TextHeader::default())),
                inline_content: None,
                compression: CompressionType::None,
            }
        }

        #[test]
        fn enforce_header_size_accepts_small_header() {
            // A minimal header serializes well under the MTU budget, so it is accepted.
            let header = header_with_attributes(HashMap::new());
            assert!(enforce_header_size(&header, &[]).is_ok());
        }

        #[test]
        fn enforce_header_size_rejects_large_header() {
            // A giant attribute value pushes the serialized header past the MTU budget.
            let mut attributes = HashMap::new();
            attributes.insert("big".to_string(), "x".repeat(20_000));
            let header = header_with_attributes(attributes);
            assert!(matches!(enforce_header_size(&header, &[]), Err(StreamError::HeaderTooLarge)));
        }
    }

    // Regression test for CLT-2773: dropping a `RawStream` on a thread that has
    // no Tokio runtime in TLS (e.g. the .NET GC finalizer thread in the Unity
    // SDK) used to panic because `Drop` called `tokio::spawn` unconditionally.
    #[test]
    fn drop_raw_stream_on_non_tokio_thread_does_not_panic() {
        let rt = tokio::runtime::Runtime::new().unwrap();

        let raw_stream = rt.block_on(async {
            let (packet_tx, mut packet_rx) =
                bmrng::unbounded_channel::<Vec<proto::DataPacket>, Result<(), SendError>>();

            tokio::spawn(async move {
                while let Ok((_packet, responder)) = packet_rx.recv().await {
                    let _ = responder.respond(Ok(()));
                }
            });

            let header = Header {
                stream_id: "gc-test-stream".into(),
                timestamp: 0,
                topic: "gc-test-topic".to_string(),
                mime_type: constants::TEXT_MIME_TYPE.to_owned(),
                total_length: None,
                attributes: HashMap::new(),
                content_header: None,
                // Data streams v2 fields
                inline_content: None,
                compression: CompressionType::None,
            };

            RawStream::open(RawStreamOpenOptions {
                header,
                destination_identities: vec![],
                sender_identity: None,
                packet_tx,
            })
            .await
            .expect("RawStream should open")
        });

        let drop_thread = std::thread::spawn(move || drop(raw_stream));

        drop_thread.join().expect("Dropping RawStream on a non-Tokio thread must not panic");
    }

    // --- Batching ---------------------------------------------------------------------------

    mod packet_batching {
        use super::*;

        #[tokio::test]
        async fn one_shot_send_text_is_a_single_transport_request() {
            let (m, sent) = setup_batched();
            // 40 KB uncompressed to a pre-v2 room: header + 3 chunks (15k/15k/10k) + trailer,
            // delivered as ONE transport request rather than one per packet.
            let text = "A".repeat(40_000);
            m.send_text(&text, text_opts("chat", &[]), &pre_v2_room()).await.unwrap();
            let batches = sent.lock().unwrap().clone();
            assert_eq!(batches.iter().map(Vec::len).collect::<Vec<_>>(), vec![5]);
            let batch = &batches[0];
            assert!(matches!(batch[0].value, Some(proto::data_packet::Value::StreamHeader(_))));
            for (i, packet) in batch[1..4].iter().enumerate() {
                assert_eq!(chunk(packet).chunk_index, i as u64);
            }
            assert_trailer(&batch[4]);
        }

        #[tokio::test]
        async fn one_shot_send_bytes_is_a_single_transport_request() {
            let (m, sent) = setup_batched();
            let payload = vec![0x07u8; 40_000];
            let opts = byte_opts("blob", &["alice", "bob"]).with_compress(false);
            m.send_bytes(&payload, opts, &all_v2_room()).await.unwrap();
            let batches = sent.lock().unwrap().clone();
            assert_eq!(batches.iter().map(Vec::len).collect::<Vec<_>>(), vec![5]);
        }

        #[tokio::test]
        async fn one_shot_send_with_sender_identity_stamps_every_packet() {
            let (m, sent) = setup_batched();
            let opts = text_opts("chat", &[]).with_sender_identity("impostor");
            m.send_text(&"A".repeat(20_000), opts, &pre_v2_room()).await.unwrap();
            let batches = sent.lock().unwrap().clone();
            assert_eq!(batches.iter().map(Vec::len).collect::<Vec<_>>(), vec![4]);
            assert!(batches[0].iter().all(|pkt| pkt.participant_identity == "impostor"));
        }

        #[tokio::test]
        async fn incremental_writer_sends_per_write() {
            let (m, sent) = setup_batched();
            let writer = m.stream_text(stream_text_opts("chat", &[])).await.unwrap();
            writer.write("hello").await.unwrap();
            writer.write("world").await.unwrap();
            writer.close().await.unwrap();
            // Incremental writes are flushed as they happen — never coalesced across writes.
            let batches = sent.lock().unwrap().clone();
            assert_eq!(batches.iter().map(Vec::len).collect::<Vec<_>>(), vec![1, 1, 1, 1]);
        }

        #[tokio::test]
        async fn send_file_streams_one_packet_per_request() {
            // send_file deliberately never buffers the whole file, so it keeps per-packet
            // requests instead of the one-shot batch.
            let (m, sent) = setup_batched();
            let path =
                std::env::temp_dir().join(format!("lk_ds_batch_{}.bin", create_random_uuid()));
            tokio::fs::write(&path, vec![0x07u8; 20_000]).await.unwrap();
            m.send_file(&path, byte_opts("file", &[]).with_compress(false), &all_v2_room())
                .await
                .unwrap();
            let _ = tokio::fs::remove_file(&path).await;
            let batches = sent.lock().unwrap().clone();
            // Header + 15k chunk + 5k chunk + trailer, each its own request.
            assert_eq!(batches.iter().map(Vec::len).collect::<Vec<_>>(), vec![1, 1, 1, 1]);
        }
    }

    // --- Additional spec-conformance cases ------------------------------------------------

    mod stream_text_bytes {
        use super::*;

        #[tokio::test]
        async fn stream_bytes_multi_write_splits_at_mtu() {
            let (m, sent) = setup();
            let writer = m.stream_bytes(stream_byte_opts("blob", &[])).await.unwrap();
            writer.write(&vec![0x01u8; 20_000]).await.unwrap();
            writer.write(&vec![0x01u8; 20_000]).await.unwrap();
            writer.close().await.unwrap();

            // Each write splits into a 15k + 5k chunk; indices are contiguous ACROSS writes.
            let p = sent.lock().unwrap().clone();
            assert_eq!(p.len(), 6); // header + 4 chunks + trailer
            for (i, expected_len) in [15_000usize, 5_000, 15_000, 5_000].iter().enumerate() {
                let c = chunk(&p[i + 1]);
                assert_eq!(c.chunk_index, i as u64);
                assert_eq!(c.content.len(), *expected_len);
                assert!(c.content.iter().all(|b| *b == 0x01));
            }
            assert_trailer(&p[5]);
        }

        #[tokio::test]
        async fn close_with_options_sends_trailer_attributes() {
            let (m, sent) = setup();
            let writer = m.stream_text(stream_text_opts("chat", &[])).await.unwrap();
            writer.write("hello").await.unwrap();
            let attributes = HashMap::from([("result".to_string(), "ok".to_string())]);
            writer.close_with_options(None, Some(attributes.clone())).await.unwrap();

            let p = sent.lock().unwrap().clone();
            let trailer = match p.last().unwrap().value.as_ref().unwrap() {
                proto::data_packet::Value::StreamTrailer(t) => t,
                _ => panic!("expected stream trailer"),
            };
            assert_eq!(trailer.reason, "");
            assert_eq!(trailer.attributes, attributes);
        }

        #[tokio::test]
        async fn close_with_options_sends_reason_and_attributes() {
            let (m, sent) = setup();
            let writer = m.stream_text(stream_text_opts("chat", &[])).await.unwrap();
            let attributes = HashMap::from([("cause".to_string(), "cancelled".to_string())]);
            writer.close_with_options(Some("aborted"), Some(attributes.clone())).await.unwrap();

            let p = sent.lock().unwrap().clone();
            let trailer = match p.last().unwrap().value.as_ref().unwrap() {
                proto::data_packet::Value::StreamTrailer(t) => t,
                _ => panic!("expected stream trailer"),
            };
            assert_eq!(trailer.reason, "aborted");
            assert_eq!(trailer.attributes, attributes);
        }

        #[tokio::test]
        async fn stream_text_oversized_attributes_errors() {
            let (m, sent) = setup();
            let mut opts = stream_text_opts("chat", &[]);
            opts.attributes.insert("big".to_string(), "x".repeat(20_000));
            let result = m.stream_text(opts).await;
            assert!(matches!(result, Err(StreamError::HeaderTooLarge)));
            // The error must be raised before anything hits the wire.
            assert!(sent.lock().unwrap().is_empty());
        }

        #[tokio::test]
        async fn stream_text_splits_on_utf8_boundaries_at_mtu() {
            let (m, sent) = setup();
            // 14 999 single-byte chars put the 4-byte emoji straddling the 15 000-byte MTU boundary.
            let text = format!("{}😀{}", "a".repeat(14_999), "b".repeat(10));
            let writer = m.stream_text(stream_text_opts("chat", &[])).await.unwrap();
            writer.write(&text).await.unwrap();
            writer.close().await.unwrap();

            let p = sent.lock().unwrap().clone();
            assert_eq!(p.len(), 4); // header + 2 chunks + trailer
            let first = chunk(&p[1]);
            let second = chunk(&p[2]);
            // The split backs off below the MTU rather than bisecting the emoji, so each chunk
            // decodes independently.
            assert_eq!(first.content.len(), 14_999);
            let first_str =
                std::str::from_utf8(&first.content).expect("chunk 0 must be valid UTF-8");
            let second_str =
                std::str::from_utf8(&second.content).expect("chunk 1 must be valid UTF-8");
            assert_eq!(format!("{first_str}{second_str}"), text);
        }
    }
}