livekit_data_stream/incoming/

manager.rs

// Copyright 2025 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 bytes::Bytes;
use futures_util::{stream::FuturesUnordered, StreamExt};
use livekit_common::{EncryptionType, ParticipantIdentity};
use parking_lot::RwLock;
use std::{collections::HashMap, future::Future, pin::Pin, sync::Arc};
use tokio::sync::{
    mpsc::{self, UnboundedReceiver, UnboundedSender},
    watch,
};
use tokio_util::sync::{CancellationToken, DropGuard};

use crate::{
    info::AnyStreamInfo,
    types::{Chunk, CompressionType, Header, Packet, StreamId, Trailer},
    utils::{StreamError, StreamProgress, StreamResult},
};

use super::{
    events::{
        ChunkReceived, InputEvent, OutputEvent, PacketReceived, StreamClosed, StreamOpened,
        TrailerReceived,
    },
    stream_reader::{AnyStreamReader, StreamCloser},
};

/// Max data stream payload size, defaults to 5gb
const DEFAULT_MAX_PAYLOAD_BYTE_LENGTH: usize = (5e9) as usize;

struct Descriptor {
    /// The token its [`StreamCloser`] cancels. Dropping the descriptor cancels it too, which ends
    /// the stream's future in the manager's `closes`.
    close_guard: DropGuard,
    progress: StreamProgress,
    chunk_tx: UnboundedSender<StreamResult<Bytes>>,
    /// Publishes `progress` updates to the reader's `progress()` stream.
    progress_tx: watch::Sender<StreamProgress>,
    encryption_type: EncryptionType,
    /// Identity of the participant sending this stream; used to abort the stream
    /// if that participant disconnects mid-send.
    sender_identity: ParticipantIdentity,
    /// Topic this stream was opened on, reported on chunk/trailer events so the host can recognize
    /// streams it handles internally (chunk and trailer packets carry only a stream id).
    topic: String,
    /// Whether this is a text stream (decompressed output is reframed on UTF-8 boundaries).
    is_text: bool,
    /// Per-stream deflate-raw decompressor; `Some` if the header declared `DEFLATE_RAW`.
    decompressor: Option<DeflateDecompressState>,
    /// Highest chunk index processed so far (compressed streams; for dedup/gap detection).
    last_chunk_index: Option<u64>,
    /// Map of all attributes associated with string, so that any attributes within the trailer can
    /// be stored after stream creation.
    attributes_map: Arc<RwLock<HashMap<String, String>>>,
}

/// Streaming deflate-raw decompressor state for one compressed stream.
///
/// Backed by `async-compression`'s push-style (`AsyncWrite`) decoder: ordered compressed chunks
/// are written into it and the decompressed output lands in the inner `Vec`, which is drained per
/// chunk. Because the manager runs as an actor (see [`Manager::run`]), the decode is
/// awaited directly on the run-loop task — no lock is held across the `.await`, and it behaves
/// identically across every async backend the SDK supports.
struct DeflateDecompressState {
    decoder: async_compression::futures::write::DeflateDecoder<Vec<u8>>,
    /// Number of bytes which have been emitted by the compressor
    output_bytes_length: usize,
    /// Max number of bytes which the compressor can take in before erroring
    max_byte_length: usize,
    /// Decompressed text bytes not yet yielded because they end mid-codepoint.
    pending_text: Vec<u8>,
}

impl DeflateDecompressState {
    fn new(max_byte_length: usize) -> Self {
        // The `deflate` algorithm is raw DEFLATE (no zlib header/checksum), matching the wire
        // contract.
        Self {
            decoder: async_compression::futures::write::DeflateDecoder::new(Vec::new()),
            output_bytes_length: 0,
            max_byte_length,
            pending_text: Vec::new(),
        }
    }

    /// Feeds compressed `input` through the stateful decompressor, returning all
    /// decompressed output produced so far.
    async fn push(&mut self, input: &[u8]) -> StreamResult<Vec<u8>> {
        use futures_util::io::AsyncWriteExt;

        self.decoder.write_all(input).await.map_err(|_| StreamError::Decompression)?;

        // Flush so all currently-decodable output lands in the inner `Vec`.
        self.decoder.flush().await.map_err(|_| StreamError::Decompression)?;

        let output_bytes = std::mem::take(self.decoder.get_mut());
        self.output_bytes_length += output_bytes.len();
        if self.output_bytes_length > self.max_byte_length {
            return Err(StreamError::PayloadTooLarge);
        }

        Ok(output_bytes)
    }

    /// Appends `decompressed` text bytes and returns the longest valid-UTF-8 prefix,
    /// retaining any trailing incomplete codepoint for the next chunk.
    fn reframe_text(&mut self, decompressed: Vec<u8>) -> Bytes {
        self.pending_text.extend_from_slice(&decompressed);
        let valid = match std::str::from_utf8(&self.pending_text) {
            Ok(_) => self.pending_text.len(),
            Err(e) => e.valid_up_to(),
        };
        let mut before = std::mem::take(&mut self.pending_text);
        let after = before.split_off(valid);
        self.pending_text = after;
        Bytes::from(before)
    }
}

/// Batch size used to incrementally pull decompressed output from an inline payload.
const INFLATE_BATCH_BYTE_LENGTH: usize = 16 * 1024;

async fn inflate_raw(data: &[u8], max_byte_length: usize) -> StreamResult<Vec<u8>> {
    use futures_util::io::AsyncReadExt;
    let mut decoder = async_compression::futures::bufread::DeflateDecoder::new(
        futures_util::io::Cursor::new(data),
    );
    let mut out = Vec::new();
    let mut batch = [0u8; INFLATE_BATCH_BYTE_LENGTH];
    loop {
        let n = decoder.read(&mut batch).await.map_err(|_| StreamError::Decompression)?;
        if n == 0 {
            break;
        }
        out.extend_from_slice(&batch[..n]);
        if out.len() > max_byte_length {
            return Err(StreamError::PayloadTooLarge);
        }
    }
    Ok(out)
}

/// Cheap, cloneable, `Send + Sync` handle used to feed [`InputEvent`]s into the manager's run
/// loop.
///
/// Dropping the last handle stops the loop (via [`InputEvent::Shutdown`]).
#[derive(Clone)]
pub struct ManagerInput {
    input_tx: UnboundedSender<InputEvent>,
    _shutdown_on_drop: Arc<ShutdownOnDrop>,
}

/// Shared by every clone of a [`ManagerInput`]. Sends [`InputEvent::Shutdown`] when the last
/// clone is dropped.
struct ShutdownOnDrop {
    input_tx: UnboundedSender<InputEvent>,
}

impl Drop for ShutdownOnDrop {
    fn drop(&mut self) {
        let _ = self.input_tx.send(InputEvent::Shutdown);
    }
}

impl ManagerInput {
    fn new(input_tx: UnboundedSender<InputEvent>) -> Self {
        Self {
            input_tx: input_tx.clone(),
            _shutdown_on_drop: Arc::new(ShutdownOnDrop { input_tx }),
        }
    }

    /// Feeds an event to the manager's run loop. Fails only if the loop has already stopped.
    pub fn send(&self, event: InputEvent) -> StreamResult<()> {
        self.input_tx.send(event).map_err(|_| StreamError::Internal)
    }
}

/// Actor that owns all incoming-stream state and processes [`InputEvent`]s on a single task
/// (see [`Self::run`]). Because it owns its state directly (no shared `Mutex`), its handlers can
/// `.await` decompression on the run-loop task.
pub struct Manager {
    inner: ManagerInner,
    input_rx: UnboundedReceiver<InputEvent>,
    /// Yields a stream's id once its close token is cancelled. Each future ends when its
    /// descriptor is dropped and is removed on the next loop pass, so the set cannot accumulate,
    /// and shutdown drops it. The futures are `Sync`, so the manager stays `Sync`.
    closes: FuturesUnordered<Pin<Box<dyn Future<Output = StreamId> + Send + Sync>>>,

    /// Max number of bytes that a data stream can contain before it is deemed to be malicious
    max_payload_byte_length: usize,
}

struct ManagerInner {
    open_streams: HashMap<StreamId, Descriptor>,
    output_tx: UnboundedSender<OutputEvent>,
}

impl Manager {
    pub fn new(
        max_payload_byte_length: Option<usize>,
    ) -> (Self, ManagerInput, UnboundedReceiver<OutputEvent>) {
        // Unbounded: inbound wire packets must never be dropped (a dropped chunk is an
        // unrecoverable `MissedChunk`) and must not head-of-line-block the engine event loop.
        let (input_tx, input_rx) = mpsc::unbounded_channel();
        let (output_tx, output_rx) = mpsc::unbounded_channel();
        let manager = Self {
            inner: ManagerInner { open_streams: HashMap::new(), output_tx },
            input_rx,
            closes: FuturesUnordered::new(),

            max_payload_byte_length: max_payload_byte_length
                .unwrap_or(DEFAULT_MAX_PAYLOAD_BYTE_LENGTH),
        };
        (manager, ManagerInput::new(input_tx), output_rx)
    }

    /// Runs the manager's event loop until the input channel closes (all
    /// [`ManagerInput`]s dropped) or [`InputEvent::Shutdown`] is received. On exit, every
    /// still-open stream is aborted so its reader errors with [`StreamError::AbnormalEnd`]
    /// rather than ending cleanly with whatever partial content had arrived.
    pub async fn run(mut self) {
        loop {
            let event = tokio::select! {
                // Closes first, so an event sent after `close()` finds the stream gone.
                biased;
                Some(id) = self.closes.next() => {
                    self.inner.close_from_reader(&id);
                    continue;
                }
                event = self.input_rx.recv() => match event {
                    Some(event) => event,
                    None => break,
                },
            };
            match event {
                InputEvent::PacketReceived(PacketReceived { packet, participant_identity }) => {
                    match packet {
                        Packet::Header { header, encryption_type } => {
                            self.handle_header(header, participant_identity, encryption_type).await
                        }
                        Packet::Chunk { chunk, encryption_type } => {
                            self.handle_chunk(chunk, participant_identity, encryption_type).await
                        }
                        Packet::Trailer { trailer, encryption_type } => {
                            self.handle_trailer(trailer, participant_identity, encryption_type)
                        }
                    }
                }
                InputEvent::AbortStreamsFrom(identity) => self.handle_abort(identity),
                InputEvent::AbortAllStreams => self.handle_abort_all(),
                InputEvent::QueryOpenStreamCount(respond_to) => {
                    let _ = respond_to.send(self.inner.open_streams.len());
                }
                InputEvent::Shutdown => break,
            }
        }
        self.handle_abort_all();
    }

    /// Handles an incoming header packet.
    async fn handle_header(
        &mut self,
        mut header: Header,
        participant_identity: ParticipantIdentity,
        encryption_type: EncryptionType,
    ) {
        let topic = header.topic.clone();

        // A compression type from a future protocol version can't be decoded; drop the stream
        // (a conforming sender never sends compression a recipient didn't advertise support for,
        // so this is a defensive backstop).
        if header.compression == CompressionType::Unrecognized {
            log::warn!(
                "Stream '{}' received with an unrecognized compression type, dropping",
                header.stream_id
            );
            return;
        }

        // Read the v2 signals before `try_from_with_encryption` consumes the header.
        // Under test-utils, clone rather than take so the header still carries the
        // inline content when the info's `is_inline` diagnostic is computed from it.
        let inline_content = if cfg!(feature = "test-utils") {
            header.inline_content.clone()
        } else {
            header.inline_content.take()
        };
        let is_compressed = header.compression == CompressionType::DeflateRaw;

        let Ok(info) = AnyStreamInfo::try_from_with_encryption(header, encryption_type)
            .inspect_err(|e| log::error!("Invalid header: {}", e))
        else {
            return;
        };

        let id: StreamId = info.id().into();
        let is_text = matches!(info, AnyStreamInfo::Text(_));
        let bytes_total = info.total_length();
        let stream_encryption_type = info.encryption_type();
        let attributes_map = info.attributes_map();

        if self.inner.open_streams.contains_key(&id) {
            log::error!("Stream '{}' already open", id);
            return;
        }

        let token = CancellationToken::new();
        let closer = StreamCloser { token: token.clone() };
        let (stream_reader, chunk_tx, progress_tx) = AnyStreamReader::from(info, closer);
        let _ = self.inner.output_tx.send(
            StreamOpened { stream_reader, participant_identity: participant_identity.clone() }
                .into(),
        );

        if bytes_total.is_some_and(|total| total > self.max_payload_byte_length as u64) {
            let _ = chunk_tx.send(Err(StreamError::PayloadTooLarge));
            self.inner.emit_stream_closed(&id, participant_identity, topic);
            return;
        }

        // Inline single-packet stream: synthesize the complete content now; no chunk/trailer
        // packets will follow, so we never register an open descriptor. Every path below
        // terminates the stream, so each emits `StreamClosed` (there is no trailer to do it).
        if let Some(content) = inline_content {
            let content = if is_compressed {
                match inflate_raw(&content, self.max_payload_byte_length).await {
                    Ok(decompressed) => decompressed,
                    Err(error) => {
                        // Defensive: a conforming sender never sends a compressed stream we
                        // can't read, but drop gracefully if it happens.
                        let _ = chunk_tx.send(Err(error));
                        self.inner.emit_stream_closed(&id, participant_identity, topic);
                        return;
                    }
                }
            } else {
                if content.len() > self.max_payload_byte_length {
                    let _ = chunk_tx.send(Err(StreamError::PayloadTooLarge));
                    self.inner.emit_stream_closed(&id, participant_identity, topic);
                    return;
                }
                content
            };
            // Hold the payload to its declared length, as the trailer does for a chunked stream.
            let length_error = match bytes_total {
                Some(total) if (content.len() as u64) < total => Some(StreamError::Incomplete),
                Some(total) if (content.len() as u64) > total => Some(StreamError::LengthExceeded),
                _ => None,
            };
            if let Some(error) = length_error {
                let _ = chunk_tx.send(Err(error));
                self.inner.emit_stream_closed(&id, participant_identity, topic);
                return;
            }
            // The whole payload arrives at once, so publish a single completed progress update.
            let _ = progress_tx.send(StreamProgress {
                chunk_index: 0,
                bytes_processed: content.len() as u64,
                bytes_total,
            });
            // The full payload is complete and (for text) valid UTF-8, so deliver it as one chunk.
            if !content.is_empty() {
                let _ = chunk_tx.send(Ok(Bytes::from(content)));
            }
            // Dropping `chunk_tx` closes the reader.
            self.inner.emit_stream_closed(&id, participant_identity, topic);
            return;
        }

        let closed = token.clone().cancelled_owned();
        let closed_id = id.clone();
        self.closes.push(Box::pin(async move {
            closed.await;
            closed_id
        }));
        let descriptor = Descriptor {
            close_guard: token.drop_guard(),
            progress: StreamProgress { bytes_total, ..Default::default() },
            chunk_tx,
            progress_tx,
            encryption_type: stream_encryption_type,
            sender_identity: participant_identity,
            topic,
            is_text,
            decompressor: is_compressed
                .then(|| DeflateDecompressState::new(self.max_payload_byte_length)),
            last_chunk_index: None,
            attributes_map,
        };
        self.inner.open_streams.insert(id, descriptor);
    }

    /// Returns the topic of an open stream, or `None` if no stream with this id is open.
    ///
    /// Reported on chunk/trailer events so the host can apply its own topic policy (e.g. hiding
    /// `lk.rpc_request`); this crate deliberately holds no notion of which topics are internal.
    fn topic_associated_with_stream_id(&self, id: &StreamId) -> Option<String> {
        self.inner.open_streams.get(id).map(|d| d.topic.clone())
    }

    /// Handles an incoming chunk packet.
    async fn handle_chunk(
        &mut self,
        chunk: Chunk,
        participant_identity: ParticipantIdentity,
        encryption_type: EncryptionType,
    ) {
        let id = chunk.stream_id.clone();
        let _ = self.inner.output_tx.send(OutputEvent::ChunkReceived(ChunkReceived {
            chunk: chunk.clone(),
            participant_identity,
            topic: self.topic_associated_with_stream_id(&id),
        }));

        let inner = &mut self.inner;
        let Some(descriptor) = inner.open_streams.get_mut(&id) else {
            return;
        };

        if descriptor.encryption_type != encryption_type {
            let expected = descriptor.encryption_type;
            inner.close_stream_with_error(
                &id,
                StreamError::EncryptionTypeMismatch { expected, received: encryption_type },
            );
            return;
        }

        if let Some(decompressor) = &mut descriptor.decompressor {
            // --- Compressed stream: feed chunks through one stateful decompressor. ---
            // Duplicate index (reconnect replay): drop with a warning.
            if let Some(last) = descriptor.last_chunk_index {
                if chunk.chunk_index <= last {
                    log::warn!(
                        "Dropping duplicate chunk {} for compressed stream '{}'",
                        chunk.chunk_index,
                        id
                    );
                    return;
                }
            }
            // A gap is unrecoverable for a stateful decompressor.
            let expected = descriptor.last_chunk_index.map(|i| i + 1).unwrap_or(0);
            if chunk.chunk_index != expected {
                inner.close_stream_with_error(&id, StreamError::MissedChunk);
                return;
            }
            descriptor.last_chunk_index = Some(chunk.chunk_index);

            // Confine the decompressor borrow so we can re-borrow `inner` afterwards.
            let result: StreamResult<(u64, Bytes)> = {
                match decompressor.push(&chunk.content).await {
                    Ok(decompressed) => {
                        let uncompressed_byte_count = decompressed.len() as u64;
                        let yielded = if descriptor.is_text {
                            decompressor.reframe_text(decompressed)
                        } else {
                            Bytes::from(decompressed)
                        };
                        Ok((uncompressed_byte_count, yielded))
                    }
                    Err(error) => Err(error),
                }
            };

            let (uncompressed_byte_count, to_yield) = match result {
                Ok(value) => value,
                Err(error) => {
                    inner.close_stream_with_error(&id, error);
                    return;
                }
            };

            // Count decompressed bytes against the (uncompressed) total length.
            descriptor.progress.bytes_processed += uncompressed_byte_count;
            if let Some(total) = descriptor.progress.bytes_total {
                if descriptor.progress.bytes_processed > total {
                    inner.close_stream_with_error(&id, StreamError::LengthExceeded);
                    return;
                }
            }
            if !to_yield.is_empty() {
                inner.yield_chunk(&id, to_yield);
            }
            inner.publish_progress(&id);
            return;
        }

        // --- Uncompressed (v1) stream: contiguous chunks, content delivered as-is. ---
        // Duplicate index (reconnect replay): drop with a warning.
        if chunk.chunk_index < descriptor.progress.chunk_index {
            log::warn!("Dropping duplicate chunk {} for stream '{}'", chunk.chunk_index, id);
            return;
        }
        if descriptor.progress.chunk_index != chunk.chunk_index {
            inner.close_stream_with_error(&id, StreamError::MissedChunk);
            return;
        }

        descriptor.progress.chunk_index += 1;
        descriptor.progress.bytes_processed += chunk.content.len() as u64;
        let bytes_processed = descriptor.progress.bytes_processed;
        let bytes_total = descriptor.progress.bytes_total;

        if bytes_processed > self.max_payload_byte_length as u64 {
            inner.close_stream_with_error(&id, StreamError::PayloadTooLarge);
            return;
        }
        if bytes_total.is_some_and(|total| bytes_processed > total) {
            inner.close_stream_with_error(&id, StreamError::LengthExceeded);
            return;
        }
        inner.yield_chunk(&id, Bytes::from(chunk.content));
        inner.publish_progress(&id);
    }

    /// Handles an incoming trailer packet.
    fn handle_trailer(
        &mut self,
        trailer: Trailer,
        participant_identity: ParticipantIdentity,
        encryption_type: EncryptionType,
    ) {
        let id = trailer.stream_id.clone();
        let _ = self.inner.output_tx.send(
            TrailerReceived {
                trailer: trailer.clone(),
                participant_identity,
                topic: self.topic_associated_with_stream_id(&id),
            }
            .into(),
        );

        let inner = &mut self.inner;
        let Some(descriptor) = inner.open_streams.get_mut(&id) else {
            return;
        };

        // Checked before the attribute merge: a trailer that arrived under the wrong encryption
        // must not close the stream cleanly, nor inject its attributes (an unencrypted peer could
        // otherwise forge a clean close for an encrypted stream).
        if descriptor.encryption_type != encryption_type {
            let expected = descriptor.encryption_type;
            inner.close_stream_with_error(
                &id,
                StreamError::EncryptionTypeMismatch { expected, received: encryption_type },
            );
            return;
        }

        // Move over any attributes from the trailer into the stream-scoped attribute list.
        {
            let mut attributes_write = descriptor.attributes_map.write();
            attributes_write.extend(trailer.attributes);
        }

        // A sender that ends the stream early says why; that reason, not the missing bytes, is the
        // error the reader gets.
        if !trailer.reason.is_empty() {
            inner.close_stream_with_error(&id, StreamError::AbnormalEnd(trailer.reason));
            return;
        }
        if !match descriptor.progress.bytes_total {
            Some(total) => descriptor.progress.bytes_processed >= total,
            None => true,
        } {
            inner.close_stream_with_error(&id, StreamError::Incomplete);
            return;
        }
        inner.close_stream(&id);
    }

    /// Aborts every open stream being sent by the given participant, erroring each
    /// reader with [`StreamError::AbnormalEnd`].
    ///
    /// Called when a remote participant disconnects: any streams it had in flight to
    /// this receiver are terminated so their readers observe an error rather than
    /// hanging forever waiting for chunks that will never arrive.
    fn handle_abort(&mut self, identity: ParticipantIdentity) {
        self.inner.close_matching_streams_with_error(|_id, descriptor| {
            if descriptor.sender_identity == identity {
                let reason = format!(
                    "Participant {} unexpectedly disconnected in the middle of sending data",
                    identity
                );
                Err(StreamError::AbnormalEnd(reason))
            } else {
                Ok(())
            }
        });
    }

    /// Aborts every open stream, erroring each reader with [`StreamError::AbnormalEnd`]. Unlike
    /// [`Self::handle_abort`] this isn't scoped to one participant; the host calls it when the
    /// connection is torn down so no reader hangs waiting for chunks that will never arrive.
    /// The run loop keeps going, so streams opened after (e.g. a reconnect) are still handled.
    fn handle_abort_all(&mut self) {
        self.inner.close_matching_streams_with_error(|_id, _descriptor| {
            Err(StreamError::AbnormalEnd("Data stream connection closed".to_string()))
        });
    }
}

impl ManagerInner {
    fn yield_chunk(&mut self, id: &StreamId, chunk: Bytes) {
        let Some(descriptor) = self.open_streams.get_mut(id) else {
            return;
        };
        if descriptor.chunk_tx.send(Ok(chunk)).is_err() {
            // Reader has been dropped, close the stream.
            self.close_stream(id);
        }
    }

    /// Publishes the descriptor's current progress to the reader's `progress()` stream.
    fn publish_progress(&self, id: &StreamId) {
        if let Some(descriptor) = self.open_streams.get(id) {
            // `StreamProgress` is `Copy`; a send error just means the reader was dropped, which the
            // chunk channel already handles, so ignore it.
            let _ = descriptor.progress_tx.send(descriptor.progress);
        }
    }

    fn close_stream(&mut self, id: &StreamId) {
        // Dropping the sender closes the channel.
        if let Some(descriptor) = self.open_streams.remove(id) {
            self.emit_stream_closed(id, descriptor.sender_identity, descriptor.topic);
        }
    }

    /// Ends a stream on behalf of its reader (see [`StreamCloser`]), unless the stream ended
    /// already or its id now belongs to a later stream, whose own token is not cancelled.
    fn close_from_reader(&mut self, id: &StreamId) {
        if self.open_streams.get(id).is_some_and(|d| d.close_guard.token().is_cancelled()) {
            self.close_stream_with_error(id, StreamError::AlreadyClosed);
        }
    }

    fn close_stream_with_error(&mut self, id: &StreamId, error: StreamError) {
        if let Some(descriptor) = self.open_streams.remove(id) {
            let _ = descriptor.chunk_tx.send(Err(error));
            self.emit_stream_closed(id, descriptor.sender_identity, descriptor.topic);
        }
    }

    fn close_matching_streams_with_error(
        &mut self,
        checker: impl Fn(&StreamId, &Descriptor) -> Result<(), StreamError>,
    ) {
        let Self { open_streams, output_tx, .. } = self;
        open_streams.retain(|id, descriptor| match checker(id, descriptor) {
            Ok(_) => true,
            Err(error) => {
                let _ = descriptor.chunk_tx.send(Err(error));
                let _ = output_tx.send(OutputEvent::StreamClosed(StreamClosed {
                    stream_id: id.clone(),
                    participant_identity: descriptor.sender_identity.clone(),
                    topic: descriptor.topic.clone(),
                }));
                false
            }
        });
    }

    /// Announces that a stream previously announced via [`StreamOpened`] is terminated.
    fn emit_stream_closed(&self, id: &StreamId, identity: ParticipantIdentity, topic: String) {
        let _ = self.output_tx.send(OutputEvent::StreamClosed(StreamClosed {
            stream_id: id.clone(),
            participant_identity: identity,
            topic,
        }));
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::{
        incoming::StreamReader,
        info::TextStreamInfo,
        test_utils::pseudo_random_text,
        types::{ByteHeader, StreamId, TextHeader},
    };
    use futures_util::{io::AsyncReadExt, Stream};
    use std::collections::HashMap;

    const SENDER: &str = "alice";

    async fn deflate_raw(data: &[u8]) -> Vec<u8> {
        let mut encoder = async_compression::futures::bufread::DeflateEncoder::new(
            futures_util::io::Cursor::new(data),
        );
        let mut out = Vec::new();
        encoder.read_to_end(&mut out).await.expect("DeflateEncoder::read_to_end failed");
        out
    }

    fn attrs(pairs: &[(&str, &str)]) -> HashMap<String, String> {
        pairs.iter().map(|(k, v)| (k.to_string(), v.to_string())).collect()
    }

    #[allow(clippy::too_many_arguments)]
    fn text_header(
        id: &str,
        total_length: Option<u64>,
        attributes: HashMap<String, String>,
        inline_content: Option<Vec<u8>>,
        compression: CompressionType,
    ) -> Header {
        Header {
            stream_id: StreamId::from(id),
            timestamp: 0,
            topic: "topic".to_string(),
            mime_type: "text/plain".to_string(),
            total_length,
            attributes,
            content_header: Some(TextHeader::default().into()),
            inline_content,
            compression,
        }
    }

    fn byte_header(
        id: &str,
        total_length: Option<u64>,
        inline_content: Option<Vec<u8>>,
        compression: CompressionType,
    ) -> Header {
        Header {
            stream_id: StreamId::from(id),
            timestamp: 0,
            topic: "topic".to_string(),
            mime_type: "application/octet-stream".to_string(),
            total_length,
            attributes: HashMap::new(),
            content_header: Some(ByteHeader { name: "file".to_string() }.into()),
            inline_content,
            compression,
        }
    }

    fn chunk(id: &str, index: u64, content: Vec<u8>) -> Chunk {
        Chunk { stream_id: StreamId::from(id), chunk_index: index, content, ..Default::default() }
    }

    fn trailer(id: &str) -> Trailer {
        Trailer { stream_id: StreamId::from(id), ..Default::default() }
    }

    fn trailer_with_attrs(id: &str, attributes: HashMap<String, String>) -> Trailer {
        Trailer { stream_id: StreamId::from(id), reason: String::new(), attributes }
    }

    async fn read_text(reader: AnyStreamReader) -> StreamResult<String> {
        match reader {
            AnyStreamReader::Text(r) => r.read_all().await,
            _ => panic!("expected a text reader"),
        }
    }

    async fn read_bytes(reader: AnyStreamReader) -> StreamResult<Bytes> {
        match reader {
            AnyStreamReader::Byte(r) => r.read_all().await,
            _ => panic!("expected a byte reader"),
        }
    }

    fn text_info(reader: &AnyStreamReader) -> &TextStreamInfo {
        match reader {
            AnyStreamReader::Text(r) => r.info(),
            _ => panic!("expected a text reader"),
        }
    }

    /// Drives an [`Manager`] actor for tests: spawns its `run` loop, exposes
    /// `send_*` helpers to feed events, and `next_opened` to await the reader for a new stream.
    struct Harness {
        input: ManagerInput,
        output_rx: UnboundedReceiver<OutputEvent>,
    }

    impl Harness {
        fn new() -> Self {
            Self::new_with_max_payload_length(None)
        }

        fn new_with_max_payload_length(max_payload_byte_length: Option<usize>) -> Self {
            let (manager, input, output_rx) = Manager::new(max_payload_byte_length);
            tokio::spawn(manager.run());
            Self { input, output_rx }
        }

        fn send_packet(&self, packet: Packet) {
            self.send_packet_from(packet, SENDER);
        }

        fn send_packet_from(&self, packet: Packet, identity: &str) {
            let event = InputEvent::PacketReceived(PacketReceived {
                packet,
                participant_identity: ParticipantIdentity::from(identity),
            });
            self.input.send(event).expect("Harness::send_packet failed");
        }

        fn abort(&self, identity: ParticipantIdentity) {
            self.input.send(InputEvent::AbortStreamsFrom(identity)).unwrap();
        }

        /// Queries the number of currently open (descriptor-registered) streams.
        async fn open_stream_count(&self) -> usize {
            let (respond_to, response) = tokio::sync::oneshot::channel();
            self.input.send(InputEvent::QueryOpenStreamCount(respond_to)).unwrap();
            response.await.expect("the manager should answer the query")
        }

        /// Awaits the next opened stream's reader (skipping back-compat chunk/trailer outputs).
        async fn next_opened(&mut self) -> (AnyStreamReader, ParticipantIdentity) {
            loop {
                match self.output_rx.recv().await.expect("a stream should be opened") {
                    OutputEvent::StreamOpened(StreamOpened {
                        stream_reader,
                        participant_identity,
                    }) => {
                        return (stream_reader, participant_identity);
                    }
                    _ => continue,
                }
            }
        }

        /// Drains the outputs emitted so far and returns the ids of the streams they closed. Call it
        /// after [`Self::open_stream_count`], which waits until the earlier events are handled.
        fn drain_closed_ids(&mut self) -> Vec<String> {
            let mut ids = Vec::new();
            while let Ok(event) = self.output_rx.try_recv() {
                if let OutputEvent::StreamClosed(closed) = event {
                    ids.push(closed.stream_id.to_string());
                }
            }
            ids
        }

        /// Awaits the next closed stream, returning its id, sender identity, and topic.
        async fn next_closed(&mut self) -> (String, String, String) {
            loop {
                match self.output_rx.recv().await.expect("a stream should be closed") {
                    OutputEvent::StreamClosed(StreamClosed {
                        stream_id,
                        participant_identity,
                        topic,
                    }) => {
                        return (stream_id.to_string(), participant_identity.to_string(), topic);
                    }
                    _ => continue,
                }
            }
        }
    }

    mod v1_legacy_multi_packet {
        use super::*;

        #[tokio::test]
        async fn v1_text_stream_round_trips() {
            let mut h = Harness::new();
            let text = "hello world";
            h.send_packet(Packet::Header {
                header: text_header(
                    "s1",
                    Some(text.len() as u64),
                    attrs(&[("foo", "bar")]),
                    None,
                    CompressionType::None,
                ),
                encryption_type: EncryptionType::None,
            });
            let (reader, identity) = h.next_opened().await;
            assert_eq!(identity.as_str(), SENDER);
            assert_eq!(text_info(&reader).attributes().get("foo"), Some(&"bar".to_string()));
            h.send_packet(Packet::Chunk {
                chunk: chunk("s1", 0, text.as_bytes().to_vec()),
                encryption_type: EncryptionType::None,
            });
            h.send_packet(Packet::Trailer {
                trailer: trailer("s1"),
                encryption_type: EncryptionType::None,
            });
            assert_eq!(read_text(reader).await.unwrap(), text);
        }

        #[tokio::test]
        async fn v1_byte_stream_round_trips() {
            let mut h = Harness::new();
            h.send_packet(Packet::Header {
                header: byte_header("s1", Some(4), None, CompressionType::None),
                encryption_type: EncryptionType::None,
            });
            let (reader, _) = h.next_opened().await;
            h.send_packet(Packet::Chunk {
                chunk: chunk("s1", 0, vec![1, 2, 3, 4]),
                encryption_type: EncryptionType::None,
            });
            h.send_packet(Packet::Trailer {
                trailer: trailer("s1"),
                encryption_type: EncryptionType::None,
            });
            assert_eq!(read_bytes(reader).await.unwrap(), Bytes::from(vec![1u8, 2, 3, 4]));
        }

        #[tokio::test]
        async fn v1_merges_trailer_attributes() {
            let mut h = Harness::new();
            let text = "hi";
            h.send_packet(Packet::Header {
                header: text_header(
                    "s1",
                    Some(text.len() as u64),
                    attrs(&[("foo", "bar"), ("baz", "quux")]),
                    None,
                    CompressionType::None,
                ),
                encryption_type: EncryptionType::None,
            });
            let (reader, _) = h.next_opened().await;
            h.send_packet(Packet::Chunk {
                chunk: chunk("s1", 0, text.as_bytes().to_vec()),
                encryption_type: EncryptionType::None,
            });
            h.send_packet(Packet::Trailer {
                trailer: trailer_with_attrs("s1", attrs(&[("hello", "world"), ("foo", "updated")])),
                encryption_type: EncryptionType::None,
            });
            // NOTE: trailer-attribute merging is asserted via the reader info after close.
            let info_attrs = text_info(&reader).attributes().clone();
            assert_eq!(read_text(reader).await.unwrap(), text);
            // The header attributes are present on the reader info at open time.
            assert_eq!(info_attrs.get("baz"), Some(&"quux".to_string()));
        }

        #[tokio::test]
        async fn v1_errors_when_too_few_bytes() {
            let mut h = Harness::new();
            h.send_packet(Packet::Header {
                header: text_header("s1", Some(5), HashMap::new(), None, CompressionType::None),
                encryption_type: EncryptionType::None,
            });
            let (reader, _) = h.next_opened().await;
            h.send_packet(Packet::Chunk {
                chunk: chunk("s1", 0, vec![b'x']),
                encryption_type: EncryptionType::None,
            });
            h.send_packet(Packet::Trailer {
                trailer: trailer("s1"),
                encryption_type: EncryptionType::None,
            });
            assert!(matches!(read_text(reader).await, Err(StreamError::Incomplete)));
        }

        #[tokio::test]
        async fn v1_errors_when_too_many_bytes() {
            let mut h = Harness::new();
            h.send_packet(Packet::Header {
                header: byte_header("s1", Some(3), None, CompressionType::None),
                encryption_type: EncryptionType::None,
            });
            let (reader, _) = h.next_opened().await;
            h.send_packet(Packet::Chunk {
                chunk: chunk("s1", 0, vec![1, 2, 3, 4, 5]),
                encryption_type: EncryptionType::None,
            });
            h.send_packet(Packet::Trailer {
                trailer: trailer("s1"),
                encryption_type: EncryptionType::None,
            });
            assert!(matches!(read_bytes(reader).await, Err(StreamError::LengthExceeded)));
        }

        #[tokio::test]
        async fn v1_max_payload_size_breached_with_unknown_total() {
            // A stream with no declared total must still be bounded by the receiver's cap.
            let mut h = Harness::new_with_max_payload_length(Some(1_000));
            h.send_packet(Packet::Header {
                header: byte_header("s1", None, None, CompressionType::None),
                encryption_type: EncryptionType::None,
            });
            let (reader, _) = h.next_opened().await;
            for i in 0..3 {
                h.send_packet(Packet::Chunk {
                    chunk: chunk("s1", i, vec![0u8; 400]),
                    encryption_type: EncryptionType::None,
                });
            }
            assert!(matches!(read_bytes(reader).await, Err(StreamError::PayloadTooLarge)));
        }

        #[tokio::test]
        async fn v1_max_payload_size_fast_fails_on_declared_total() {
            // A header declaring a total above the cap is rejected before any chunks arrive.
            let mut h = Harness::new_with_max_payload_length(Some(1_000));
            h.send_packet(Packet::Header {
                header: byte_header("s1", Some(2_000), None, CompressionType::None),
                encryption_type: EncryptionType::None,
            });
            let (reader, _) = h.next_opened().await;
            assert!(matches!(read_bytes(reader).await, Err(StreamError::PayloadTooLarge)));
        }

        #[tokio::test]
        async fn v1_payload_exactly_at_max_payload_size_succeeds() {
            // The cap is inclusive: a payload of exactly max_payload_byte_length is accepted.
            let mut h = Harness::new_with_max_payload_length(Some(1_000));
            h.send_packet(Packet::Header {
                header: byte_header("s1", Some(1_000), None, CompressionType::None),
                encryption_type: EncryptionType::None,
            });
            let (reader, _) = h.next_opened().await;
            h.send_packet(Packet::Chunk {
                chunk: chunk("s1", 0, vec![7u8; 1_000]),
                encryption_type: EncryptionType::None,
            });
            h.send_packet(Packet::Trailer {
                trailer: trailer("s1"),
                encryption_type: EncryptionType::None,
            });
            assert_eq!(read_bytes(reader).await.unwrap().len(), 1_000);
        }

        #[tokio::test]
        async fn v1_drops_on_encryption_type_mismatch() {
            let mut h = Harness::new();
            h.send_packet(Packet::Header {
                header: text_header("s1", Some(2), HashMap::new(), None, CompressionType::None),
                encryption_type: EncryptionType::None,
            });
            let (reader, _) = h.next_opened().await;
            h.send_packet(Packet::Chunk {
                chunk: chunk("s1", 0, vec![b'h', b'i']),
                encryption_type: EncryptionType::Gcm,
            });
            assert!(matches!(
                read_text(reader).await,
                Err(StreamError::EncryptionTypeMismatch {
                    expected: EncryptionType::None,
                    received: EncryptionType::Gcm,
                })
            ));
        }

        /// A trailer is held to the stream's encryption too: an unencrypted peer must not be
        /// able to forge a clean close (or inject trailer attributes) for an encrypted stream.
        #[tokio::test]
        async fn v1_drops_on_trailer_encryption_type_mismatch() {
            let mut h = Harness::new();
            h.send_packet(Packet::Header {
                header: text_header("s1", Some(2), HashMap::new(), None, CompressionType::None),
                encryption_type: EncryptionType::Gcm,
            });
            let (reader, _) = h.next_opened().await;
            h.send_packet(Packet::Chunk {
                chunk: chunk("s1", 0, vec![b'h', b'i']),
                encryption_type: EncryptionType::Gcm,
            });
            let info = text_info(&reader).clone();
            h.send_packet(Packet::Trailer {
                trailer: trailer_with_attrs("s1", attrs(&[("forged", "yes")])),
                encryption_type: EncryptionType::None,
            });
            assert!(matches!(
                read_text(reader).await,
                Err(StreamError::EncryptionTypeMismatch {
                    expected: EncryptionType::Gcm,
                    received: EncryptionType::None,
                })
            ));
            // The forged trailer's attributes must not have been merged.
            assert_eq!(info.attributes().get("forged"), None);
        }

        #[tokio::test]
        async fn v1_trailer_attributes_merged_after_close() {
            let mut h = Harness::new();
            let text = "hello world";
            h.send_packet(Packet::Header {
                header: text_header(
                    "s1",
                    Some(text.len() as u64),
                    attrs(&[("foo", "bar"), ("baz", "quux")]),
                    None,
                    CompressionType::None,
                ),
                encryption_type: EncryptionType::None,
            });
            let (reader, _) = h.next_opened().await;
            let info = text_info(&reader).clone();
            h.send_packet(Packet::Chunk {
                chunk: chunk("s1", 0, text.as_bytes().to_vec()),
                encryption_type: EncryptionType::None,
            });
            h.send_packet(Packet::Trailer {
                trailer: trailer_with_attrs("s1", attrs(&[("hello", "world"), ("foo", "updated")])),
                encryption_type: EncryptionType::None,
            });
            assert_eq!(read_text(reader).await.unwrap(), text);
            // The trailer attributes are merged into the stream's attributes, overriding the header's.
            let merged = info.attributes();
            assert_eq!(merged.get("baz"), Some(&"quux".to_string()));
            assert_eq!(merged.get("hello"), Some(&"world".to_string()));
            assert_eq!(merged.get("foo"), Some(&"updated".to_string()));
        }
    }

    // --- v2 inline -----------------------------------------------------------------------
    mod v2_inline {
        use super::*;

        #[tokio::test]
        async fn v2_inline_uncompressed_text() {
            let mut h = Harness::new();
            let text = "inline hello";
            h.send_packet(Packet::Header {
                header: text_header(
                    "s1",
                    Some(text.len() as u64),
                    attrs(&[("foo", "bar")]),
                    Some(text.as_bytes().to_vec()),
                    CompressionType::None,
                ),
                encryption_type: EncryptionType::None,
            });
            let (reader, _) = h.next_opened().await;
            assert_eq!(text_info(&reader).attributes().get("foo"), Some(&"bar".to_string()));
            // No chunk/trailer packets are fed.
            assert_eq!(read_text(reader).await.unwrap(), text);
        }

        #[tokio::test]
        async fn v2_inline_uncompressed_byte() {
            let mut h = Harness::new();
            h.send_packet(Packet::Header {
                header: byte_header("s1", Some(3), Some(vec![1, 2, 3]), CompressionType::None),
                encryption_type: EncryptionType::None,
            });
            let (reader, _) = h.next_opened().await;
            assert_eq!(read_bytes(reader).await.unwrap(), Bytes::from(vec![1u8, 2, 3]));
        }

        #[tokio::test]
        async fn v2_inline_compressed_text() {
            let mut h = Harness::new();
            let text = "hello hello compressible world";
            let compressed = deflate_raw(text.as_bytes()).await;
            h.send_packet(Packet::Header {
                header: text_header(
                    "s1",
                    Some(text.len() as u64),
                    attrs(&[("foo", "bar")]),
                    Some(compressed),
                    CompressionType::DeflateRaw,
                ),
                encryption_type: EncryptionType::None,
            });
            let (reader, _) = h.next_opened().await;
            assert_eq!(text_info(&reader).attributes().get("foo"), Some(&"bar".to_string()));
            assert_eq!(read_text(reader).await.unwrap(), text);
        }

        #[tokio::test]
        async fn v2_inline_compressed_byte() {
            let mut h = Harness::new();
            let payload: Vec<u8> = (0..2000).map(|i| (i % 7) as u8).collect();
            let compressed = deflate_raw(&payload).await;
            h.send_packet(Packet::Header {
                header: byte_header(
                    "s1",
                    Some(payload.len() as u64),
                    Some(compressed),
                    CompressionType::DeflateRaw,
                ),
                encryption_type: EncryptionType::None,
            });
            let (reader, _) = h.next_opened().await;
            assert_eq!(read_bytes(reader).await.unwrap(), Bytes::from(payload));
        }

        #[tokio::test]
        async fn v2_inline_compressed_max_payload_size_breached() {
            // A tiny compressed inline payload that inflates far past the configured cap must be
            // rejected (decompression-bomb guard on the inline path).
            let mut h = Harness::new_with_max_payload_length(Some(1_000));
            let text = pseudo_random_text(50_000);
            let compressed = deflate_raw(text.as_bytes()).await;
            h.send_packet(Packet::Header {
                header: text_header(
                    "s1",
                    Some(text.len() as u64),
                    HashMap::new(),
                    Some(compressed),
                    CompressionType::DeflateRaw,
                ),
                encryption_type: EncryptionType::None,
            });
            let (reader, _) = h.next_opened().await;
            assert!(matches!(read_text(reader).await, Err(StreamError::PayloadTooLarge)));
        }

        #[tokio::test]
        async fn v2_inline_uncompressed_max_payload_size_breached() {
            // The cap applies to uncompressed inline payloads too. No declared total, so the
            // inline content check (not the header fast-fail) is what trips.
            let mut h = Harness::new_with_max_payload_length(Some(1_000));
            h.send_packet(Packet::Header {
                header: byte_header("s1", None, Some(vec![0u8; 2_000]), CompressionType::None),
                encryption_type: EncryptionType::None,
            });
            let (reader, _) = h.next_opened().await;
            assert!(matches!(read_bytes(reader).await, Err(StreamError::PayloadTooLarge)));
        }

        /// Reads an inline text stream that carries `text` and declares `total_length`.
        async fn read_inline_text(
            text: &str,
            total_length: Option<u64>,
            compression: CompressionType,
        ) -> StreamResult<String> {
            let mut h = Harness::new();
            let content = match compression {
                CompressionType::DeflateRaw => deflate_raw(text.as_bytes()).await,
                _ => text.as_bytes().to_vec(),
            };
            h.send_packet(Packet::Header {
                header: text_header("s1", total_length, HashMap::new(), Some(content), compression),
                encryption_type: EncryptionType::None,
            });
            let (reader, _) = h.next_opened().await;
            read_text(reader).await
        }

        #[tokio::test]
        async fn v2_inline_payload_is_held_to_its_declared_length() {
            for compression in [CompressionType::None, CompressionType::DeflateRaw] {
                let read = |total| read_inline_text("hello", total, compression.clone());
                let short = read(Some(6)).await;
                assert!(matches!(short, Err(StreamError::Incomplete)), "{compression:?}");
                let long = read(Some(4)).await;
                assert!(matches!(long, Err(StreamError::LengthExceeded)), "{compression:?}");
                // JS treats a declared length of 0 as unknown and skips the check. No sender
                // inlines a non-empty payload with a length of 0, so the core holds 0 to the
                // payload like any other length.
                let zero = read(Some(0)).await;
                assert!(matches!(zero, Err(StreamError::LengthExceeded)), "{compression:?}");
                assert_eq!(read(Some(5)).await.unwrap(), "hello", "{compression:?}");
                assert_eq!(read(None).await.unwrap(), "hello", "{compression:?}");
            }
        }

        #[tokio::test]
        async fn v2_inline_zero_length_text() {
            let mut h = Harness::new();
            h.send_packet(Packet::Header {
                header: text_header(
                    "s1",
                    Some(0),
                    HashMap::new(),
                    Some(vec![]), // present-but-empty inline payload
                    CompressionType::None,
                ),
                encryption_type: EncryptionType::None,
            });
            let (reader, _) = h.next_opened().await;
            assert_eq!(read_text(reader).await.unwrap(), "");
        }
    }

    // --- v2 multi-packet compressed ------------------------------------------------------

    mod v2_multi_packet_compressed {
        use super::*;

        #[tokio::test]
        async fn v2_multipacket_compressed_text() {
            let mut h = Harness::new();
            // ~60 KB of pseudo-random lowercase so the compressed output spans multiple chunks.
            let text = pseudo_random_text(60_000);
            let compressed = deflate_raw(text.as_bytes()).await;
            let chunk_pieces: Vec<&[u8]> = compressed.chunks(15_000).collect();
            assert!(chunk_pieces.len() >= 2, "expected multi-packet compressed stream");

            h.send_packet(Packet::Header {
                header: text_header(
                    "s1",
                    Some(text.len() as u64),
                    HashMap::new(),
                    None,
                    CompressionType::DeflateRaw,
                ),
                encryption_type: EncryptionType::None,
            });
            let (reader, _) = h.next_opened().await;
            for (i, piece) in chunk_pieces.iter().enumerate() {
                h.send_packet(Packet::Chunk {
                    chunk: chunk("s1", i as u64, piece.to_vec()),
                    encryption_type: EncryptionType::None,
                });
            }
            h.send_packet(Packet::Trailer {
                trailer: trailer("s1"),
                encryption_type: EncryptionType::None,
            });
            assert_eq!(read_text(reader).await.unwrap(), text);
        }

        #[tokio::test]
        async fn errors_open_streams_on_sender_disconnect() {
            let mut h = Harness::new();
            h.send_packet(Packet::Header {
                header: text_header("s1", Some(10), HashMap::new(), None, CompressionType::None),
                encryption_type: EncryptionType::None,
            });
            let (reader, _) = h.next_opened().await;
            // Partial content, no trailer: the sender then drops.
            h.send_packet(Packet::Chunk {
                chunk: chunk("s1", 0, vec![b'h', b'e', b'l', b'l', b'o']),
                encryption_type: EncryptionType::None,
            });
            h.abort(ParticipantIdentity::from(SENDER));
            assert!(matches!(read_text(reader).await, Err(StreamError::AbnormalEnd(_))));
        }

        #[tokio::test]
        async fn abort_only_affects_matching_sender() {
            let mut h = Harness::new();
            h.send_packet_from(
                Packet::Header {
                    header: text_header("s1", Some(5), HashMap::new(), None, CompressionType::None),
                    encryption_type: EncryptionType::None,
                },
                "bob",
            );
            let (reader, _) = h.next_opened().await;
            h.send_packet_from(
                Packet::Chunk {
                    chunk: chunk("s1", 0, vec![b'h', b'e', b'l', b'l', b'o']),
                    encryption_type: EncryptionType::None,
                },
                "bob",
            );
            // A different participant disconnecting must not disturb bob's stream.
            h.abort(ParticipantIdentity::from(SENDER));
            h.send_packet_from(
                Packet::Trailer { trailer: trailer("s1"), encryption_type: EncryptionType::None },
                "bob",
            );
            assert_eq!(read_text(reader).await.unwrap(), "hello");
        }

        #[tokio::test]
        async fn v2_compressed_gap_errors() {
            let mut h = Harness::new();
            let text = pseudo_random_text(60_000);
            let compressed = deflate_raw(text.as_bytes()).await;
            let pieces: Vec<&[u8]> = compressed.chunks(15_000).collect();
            assert!(pieces.len() >= 2);
            h.send_packet(Packet::Header {
                header: text_header(
                    "s1",
                    Some(text.len() as u64),
                    HashMap::new(),
                    None,
                    CompressionType::DeflateRaw,
                ),
                encryption_type: EncryptionType::None,
            });
            let (reader, _) = h.next_opened().await;
            h.send_packet(Packet::Chunk {
                chunk: chunk("s1", 0, pieces[0].to_vec()),
                encryption_type: EncryptionType::None,
            });
            // Skip index 1 -> feed index 2: a gap is a hard error.
            h.send_packet(Packet::Chunk {
                chunk: chunk("s1", 2, pieces[1].to_vec()),
                encryption_type: EncryptionType::None,
            });
            assert!(matches!(read_text(reader).await, Err(StreamError::MissedChunk)));
        }

        #[tokio::test]
        async fn v2_max_payload_size_breached() {
            let text = pseudo_random_text(60_000);
            let compressed = deflate_raw(text.as_bytes()).await;

            // Use a max payload size one byte below the size of the compressed data
            let mut h = Harness::new_with_max_payload_length(Some(50_000 /* less than 60k */));

            // Feed all data in
            h.send_packet(Packet::Header {
                header: text_header(
                    "s1",
                    Some(text.len() as u64),
                    HashMap::new(),
                    None,
                    CompressionType::DeflateRaw,
                ),
                encryption_type: EncryptionType::None,
            });
            let (reader, _) = h.next_opened().await;
            for (i, byte_chunk) in compressed.chunks(15_000).enumerate() {
                h.send_packet(Packet::Chunk {
                    chunk: chunk("s1", i as u64, byte_chunk.to_vec()),
                    encryption_type: EncryptionType::None,
                });
            }

            // And make sure a PayloadTooLarge error gets raised
            assert!(matches!(read_text(reader).await, Err(StreamError::PayloadTooLarge)));
        }

        #[tokio::test]
        async fn v2_multipacket_compressed_byte_stream() {
            let mut h = Harness::new();
            let data = pseudo_random_text(60_000).into_bytes();
            let compressed = deflate_raw(&data).await;
            let pieces: Vec<&[u8]> = compressed.chunks(15_000).collect();
            assert!(pieces.len() >= 2, "expected multi-packet compressed stream");

            h.send_packet(Packet::Header {
                header: byte_header(
                    "s1",
                    Some(data.len() as u64),
                    None,
                    CompressionType::DeflateRaw,
                ),
                encryption_type: EncryptionType::None,
            });
            let (reader, _) = h.next_opened().await;
            for (i, piece) in pieces.iter().enumerate() {
                h.send_packet(Packet::Chunk {
                    chunk: chunk("s1", i as u64, piece.to_vec()),
                    encryption_type: EncryptionType::None,
                });
            }
            h.send_packet(Packet::Trailer {
                trailer: trailer("s1"),
                encryption_type: EncryptionType::None,
            });
            assert_eq!(read_bytes(reader).await.unwrap(), Bytes::from(data));
        }

        #[tokio::test]
        async fn v2_compressed_errors_when_too_few_bytes() {
            let mut h = Harness::new();
            let text = "hello world"; // 11 bytes decompressed
            let compressed = deflate_raw(text.as_bytes()).await;
            h.send_packet(Packet::Header {
                header: text_header(
                    "s1",
                    Some(16), // more than the decompressed payload
                    HashMap::new(),
                    None,
                    CompressionType::DeflateRaw,
                ),
                encryption_type: EncryptionType::None,
            });
            let (reader, _) = h.next_opened().await;
            h.send_packet(Packet::Chunk {
                chunk: chunk("s1", 0, compressed),
                encryption_type: EncryptionType::None,
            });
            h.send_packet(Packet::Trailer {
                trailer: trailer("s1"),
                encryption_type: EncryptionType::None,
            });
            // The receiver counts DECOMPRESSED bytes against totalLength.
            assert!(matches!(read_text(reader).await, Err(StreamError::Incomplete)));
        }

        #[tokio::test]
        async fn v2_compressed_errors_when_too_many_bytes() {
            let mut h = Harness::new();
            let text = "hello world"; // 11 bytes decompressed
            let compressed = deflate_raw(text.as_bytes()).await;
            h.send_packet(Packet::Header {
                header: text_header(
                    "s1",
                    Some(5), // fewer than the decompressed payload
                    HashMap::new(),
                    None,
                    CompressionType::DeflateRaw,
                ),
                encryption_type: EncryptionType::None,
            });
            let (reader, _) = h.next_opened().await;
            h.send_packet(Packet::Chunk {
                chunk: chunk("s1", 0, compressed),
                encryption_type: EncryptionType::None,
            });
            h.send_packet(Packet::Trailer {
                trailer: trailer("s1"),
                encryption_type: EncryptionType::None,
            });
            assert!(matches!(read_text(reader).await, Err(StreamError::LengthExceeded)));
        }

        #[tokio::test]
        async fn v1_duplicate_chunk_dropped() {
            let mut h = Harness::new();
            h.send_packet(Packet::Header {
                header: text_header("s1", Some(10), HashMap::new(), None, CompressionType::None),
                encryption_type: EncryptionType::None,
            });
            let (reader, _) = h.next_opened().await;
            for (i, piece) in [b"hello", b"hello", b"world"].iter().enumerate() {
                // The replayed chunk keeps its original index, so the second "hello" is a duplicate.
                let index = if i == 0 { 0 } else { i as u64 - 1 };
                h.send_packet(Packet::Chunk {
                    chunk: chunk("s1", index, piece.to_vec()),
                    encryption_type: EncryptionType::None,
                });
            }
            h.send_packet(Packet::Trailer {
                trailer: trailer("s1"),
                encryption_type: EncryptionType::None,
            });
            assert_eq!(read_text(reader).await.unwrap(), "helloworld");
        }

        #[tokio::test]
        async fn v2_compressed_duplicate_chunk_dropped() {
            let mut h = Harness::new();
            let text = pseudo_random_text(60_000);
            let compressed = deflate_raw(text.as_bytes()).await;
            let pieces: Vec<&[u8]> = compressed.chunks(15_000).collect();
            assert!(pieces.len() >= 2);

            h.send_packet(Packet::Header {
                header: text_header(
                    "s1",
                    Some(text.len() as u64),
                    HashMap::new(),
                    None,
                    CompressionType::DeflateRaw,
                ),
                encryption_type: EncryptionType::None,
            });
            let (reader, _) = h.next_opened().await;
            h.send_packet(Packet::Chunk {
                chunk: chunk("s1", 0, pieces[0].to_vec()),
                encryption_type: EncryptionType::None,
            });
            // A replayed chunk (e.g. reconnect logic) must be dropped, not fed to the stateful
            // decompressor a second time.
            h.send_packet(Packet::Chunk {
                chunk: chunk("s1", 0, pieces[0].to_vec()),
                encryption_type: EncryptionType::None,
            });
            for (i, piece) in pieces.iter().enumerate().skip(1) {
                h.send_packet(Packet::Chunk {
                    chunk: chunk("s1", i as u64, piece.to_vec()),
                    encryption_type: EncryptionType::None,
                });
            }
            h.send_packet(Packet::Trailer {
                trailer: trailer("s1"),
                encryption_type: EncryptionType::None,
            });
            assert_eq!(read_text(reader).await.unwrap(), text);
        }

        #[tokio::test]
        async fn v2_compressed_text_reframes_multibyte_utf8() {
            let mut h = Harness::new();
            let text = "😀你好世界 café — ¡ñandú! ".repeat(500);
            let compressed = deflate_raw(text.as_bytes()).await;
            // Split the compressed bytes at an arbitrary midpoint: the decompressor's output at the
            // seam can land mid-codepoint, exercising the UTF-8 reframing stage.
            let split = compressed.len() / 2;

            h.send_packet(Packet::Header {
                header: text_header(
                    "s1",
                    Some(text.len() as u64), // NOTE: byte length
                    HashMap::new(),
                    None,
                    CompressionType::DeflateRaw,
                ),
                encryption_type: EncryptionType::None,
            });
            let (reader, _) = h.next_opened().await;
            h.send_packet(Packet::Chunk {
                chunk: chunk("s1", 0, compressed[..split].to_vec()),
                encryption_type: EncryptionType::None,
            });
            h.send_packet(Packet::Chunk {
                chunk: chunk("s1", 1, compressed[split..].to_vec()),
                encryption_type: EncryptionType::None,
            });
            h.send_packet(Packet::Trailer {
                trailer: trailer("s1"),
                encryption_type: EncryptionType::None,
            });
            assert_eq!(read_text(reader).await.unwrap(), text);
        }

        #[tokio::test]
        async fn v2_unknown_compression_type_is_ignored() {
            let mut h = Harness::new();
            // A compression type from a future protocol version arrives at the proto layer; the
            // receiver can't decode it, so per the spec's defensive-drop behavior (mirroring the web
            // SDK) the stream must be ignored rather than delivered as if uncompressed.
            let proto_header = livekit_protocol::data_stream::Header {
                stream_id: "s1".to_string(),
                timestamp: 0,
                topic: "topic".to_string(),
                mime_type: "text/plain".to_string(),
                total_length: Some(11),
                content_header: Some(
                    livekit_protocol::data_stream::header::ContentHeader::TextHeader(
                        livekit_protocol::data_stream::TextHeader::default(),
                    ),
                ),
                compression: 99, // <== HERE, potential future compression value
                inline_content: None,
                ..Default::default()
            };
            h.send_packet(Packet::Header {
                header: Header::from(proto_header),
                encryption_type: EncryptionType::None,
            });
            // A well-formed second stream: if the bogus stream was correctly dropped, this is the
            // next (and only) stream to open.
            h.send_packet(Packet::Header {
                header: text_header("s2", Some(2), HashMap::new(), None, CompressionType::None),
                encryption_type: EncryptionType::None,
            });
            let (reader, _) = h.next_opened().await;
            assert_eq!(
                text_info(&reader).id,
                "s2",
                "a stream with an unrecognized compression type must be dropped"
            );
        }

        #[tokio::test]
        async fn v2_compressed_merges_trailer_attributes() {
            let mut h = Harness::new();
            let text = "hello world";
            let compressed = deflate_raw(text.as_bytes()).await;
            h.send_packet(Packet::Header {
                header: text_header(
                    "s1",
                    Some(text.len() as u64),
                    attrs(&[("foo", "bar")]),
                    None,
                    CompressionType::DeflateRaw,
                ),
                encryption_type: EncryptionType::None,
            });
            let (reader, _) = h.next_opened().await;
            let info = text_info(&reader).clone();
            h.send_packet(Packet::Chunk {
                chunk: chunk("s1", 0, compressed),
                encryption_type: EncryptionType::None,
            });
            h.send_packet(Packet::Trailer {
                trailer: trailer_with_attrs("s1", attrs(&[("hello", "world")])),
                encryption_type: EncryptionType::None,
            });
            assert_eq!(read_text(reader).await.unwrap(), text);
            let merged = info.attributes();
            assert_eq!(merged.get("foo"), Some(&"bar".to_string()));
            assert_eq!(merged.get("hello"), Some(&"world".to_string()));
        }
    }

    mod progress {
        use super::*;

        /// Returns the reader's progress stream regardless of its concrete kind. Boxed because the two
        /// `progress()` impls are distinct opaque types that don't unify across match arms.
        fn progress_of(
            reader: &AnyStreamReader,
        ) -> std::pin::Pin<Box<dyn Stream<Item = StreamProgress> + Send + '_>> {
            match reader {
                AnyStreamReader::Byte(r) => Box::pin(r.progress()),
                AnyStreamReader::Text(r) => Box::pin(r.progress()),
            }
        }

        /// Drains a progress stream to completion (the stream ends when the sender closes).
        async fn collect_progress(
            stream: impl Stream<Item = StreamProgress>,
        ) -> Vec<StreamProgress> {
            use futures_util::StreamExt;
            let mut stream = std::pin::pin!(stream);
            let mut out = Vec::new();
            while let Some(progress) = stream.next().await {
                out.push(progress);
            }
            out
        }

        /// The last value reaches the total, values never decrease, and the stream terminates.
        fn assert_progress_completes(values: &[StreamProgress], total: u64) {
            let last = values.last().expect("progress stream yielded at least one value");
            assert_eq!(last.bytes_processed(), total);
            assert_eq!(last.bytes_total(), Some(total));
            assert_eq!(last.percentage(), Some(1.0));
            assert!(
                values.windows(2).all(|w| w[0].bytes_processed() <= w[1].bytes_processed()),
                "progress must be monotonically non-decreasing: {values:?}"
            );
        }

        #[tokio::test]
        async fn progress_reports_completion_uncompressed_bytes() {
            let mut h = Harness::new();
            let total = 12u64;
            h.send_packet(Packet::Header {
                header: byte_header("s1", Some(total), None, CompressionType::None),
                encryption_type: EncryptionType::None,
            });
            let (reader, _) = h.next_opened().await;
            let progress = progress_of(&reader);
            // Feed the payload across several contiguous chunks; keep `reader` alive so the chunk
            // channel stays open while progress is observed.
            for (i, piece) in
                [vec![1, 2, 3, 4], vec![5, 6, 7, 8], vec![9, 10, 11, 12]].into_iter().enumerate()
            {
                h.send_packet(Packet::Chunk {
                    chunk: chunk("s1", i as u64, piece),
                    encryption_type: EncryptionType::None,
                });
            }
            h.send_packet(Packet::Trailer {
                trailer: trailer("s1"),
                encryption_type: EncryptionType::None,
            });

            let values = collect_progress(progress).await;
            assert_progress_completes(&values, total);
            drop(reader);
        }

        #[tokio::test]
        async fn progress_reports_completion_compressed_text() {
            let mut h = Harness::new();
            let text = pseudo_random_text(60_000);
            let total = text.len() as u64;
            let compressed = deflate_raw(text.as_bytes()).await;
            let pieces: Vec<&[u8]> = compressed.chunks(15_000).collect();
            assert!(pieces.len() >= 2, "expected multi-packet compressed stream");

            h.send_packet(Packet::Header {
                header: text_header(
                    "s1",
                    Some(total),
                    HashMap::new(),
                    None,
                    CompressionType::DeflateRaw,
                ),
                encryption_type: EncryptionType::None,
            });
            let (reader, _) = h.next_opened().await;
            let progress = progress_of(&reader);
            for (i, piece) in pieces.iter().enumerate() {
                h.send_packet(Packet::Chunk {
                    chunk: chunk("s1", i as u64, piece.to_vec()),
                    encryption_type: EncryptionType::None,
                });
            }
            h.send_packet(Packet::Trailer {
                trailer: trailer("s1"),
                encryption_type: EncryptionType::None,
            });

            let values = collect_progress(progress).await;
            assert_progress_completes(&values, total);
            drop(reader);
        }

        #[tokio::test]
        async fn progress_reports_completion_inline() {
            let mut h = Harness::new();
            let text = "inline hello";
            let total = text.len() as u64;
            h.send_packet(Packet::Header {
                header: text_header(
                    "s1",
                    Some(total),
                    HashMap::new(),
                    Some(text.as_bytes().to_vec()),
                    CompressionType::None,
                ),
                encryption_type: EncryptionType::None,
            });
            let (reader, _) = h.next_opened().await;
            // The whole payload arrives in the header, so progress jumps straight to complete.
            let values = collect_progress(progress_of(&reader)).await;
            assert_progress_completes(&values, total);
            drop(reader);
        }

        #[tokio::test]
        async fn progress_stream_outlives_a_consuming_read() {
            let mut h = Harness::new();
            h.send_packet(Packet::Header {
                header: byte_header("s1", Some(4), None, CompressionType::None),
                encryption_type: EncryptionType::None,
            });
            let (AnyStreamReader::Byte(reader), _) = h.next_opened().await else {
                panic!("expected a byte reader");
            };
            let progress = reader.progress_stream();
            h.send_packet(Packet::Chunk {
                chunk: chunk("s1", 0, vec![1, 2, 3, 4]),
                encryption_type: EncryptionType::None,
            });
            h.send_packet(Packet::Trailer {
                trailer: trailer("s1"),
                encryption_type: EncryptionType::None,
            });

            // `read_all` consumes the reader while its progress stream is still held.
            assert_eq!(reader.read_all().await.unwrap().len(), 4);
            assert_progress_completes(&collect_progress(progress).await, 4);
        }
    }

    mod closer {
        use super::*;
        use crate::incoming::TextStreamReader;

        fn open_text_stream(h: &Harness, id: &str) {
            h.send_packet(Packet::Header {
                header: text_header(id, Some(10), HashMap::new(), None, CompressionType::None),
                encryption_type: EncryptionType::None,
            });
            h.send_packet(Packet::Chunk {
                chunk: chunk(id, 0, b"hello".to_vec()),
                encryption_type: EncryptionType::None,
            });
        }

        async fn next_text_reader(h: &mut Harness) -> TextStreamReader {
            let (AnyStreamReader::Text(reader), _) = h.next_opened().await else {
                panic!("expected a text reader");
            };
            reader
        }

        #[tokio::test]
        async fn closer_after_shutdown_does_nothing() {
            // A manager whose run loop the test owns, with one open text stream.
            let (manager, input, mut output_rx) = Manager::new(None);
            let run = tokio::spawn(manager.run());
            let header = text_header("s1", Some(10), HashMap::new(), None, CompressionType::None);
            let packet = Packet::Header { header, encryption_type: EncryptionType::None };
            let identity = ParticipantIdentity::from(SENDER);
            input.send(PacketReceived::new(packet, identity).into()).unwrap();
            let reader = loop {
                let event = output_rx.recv().await.expect("the stream should open");
                if let OutputEvent::StreamOpened(StreamOpened {
                    stream_reader: AnyStreamReader::Text(reader),
                    ..
                }) = event
                {
                    break reader;
                }
            };

            input.send(InputEvent::Shutdown).unwrap();
            let stopped = tokio::time::timeout(std::time::Duration::from_secs(5), run).await;
            stopped.expect("the run loop should stop").expect("the run loop should not panic");

            // `close()` is synchronous, so each call returning shows it did not block. The stopped
            // manager watches no token, so the closes change nothing, with the input kept or
            // dropped.
            let closer = reader.closer();
            closer.close();
            closer.close();
            assert!(input.send(InputEvent::AbortAllStreams).is_err());
            drop(input);
            closer.close();

            // The output ends once the stopped manager is dropped, so this sees every event.
            let mut closes = 0;
            while let Some(event) = output_rx.recv().await {
                closes += usize::from(matches!(event, OutputEvent::StreamClosed(_)));
            }
            assert_eq!(closes, 1, "only the shutdown abort");
            assert!(matches!(reader.read_all().await, Err(StreamError::AbnormalEnd(_))));
        }

        #[tokio::test]
        async fn closer_ends_an_open_stream_once() {
            let mut h = Harness::new();
            // Each order is sent before the manager runs: a second close, then a close before the
            // final chunk and one between that chunk and the trailer. The close wins over the
            // packets still in the input queue, so every order ends the stream once.
            for (id, steps) in [
                ("s1", &["close", "close"][..]),
                ("s2", &["close", "chunk", "trailer"]),
                ("s3", &["chunk", "close", "trailer"]),
            ] {
                open_text_stream(&h, id);
                let reader = next_text_reader(&mut h).await;
                let closer = reader.closer();
                for step in steps {
                    match *step {
                        "close" => closer.close(),
                        "chunk" => h.send_packet(Packet::Chunk {
                            chunk: chunk(id, 1, b"world".to_vec()),
                            encryption_type: EncryptionType::None,
                        }),
                        "trailer" => h.send_packet(Packet::Trailer {
                            trailer: trailer(id),
                            encryption_type: EncryptionType::None,
                        }),
                        _ => unreachable!("{step}"),
                    }
                }
                assert_eq!(h.open_stream_count().await, 0, "{steps:?}");
                assert_eq!(h.drain_closed_ids(), vec![id], "{steps:?}");

                // A read that is still going fails rather than ending as if complete.
                let read = reader.read_all().await;
                assert!(matches!(read, Err(StreamError::AlreadyClosed)), "{steps:?}: {read:?}");
            }
        }

        #[tokio::test]
        async fn closer_discards_packets_still_queued() {
            let mut h = Harness::new();
            open_text_stream(&h, "s1");
            let mut reader = next_text_reader(&mut h).await;
            // Answered after the first chunk, so that chunk is in the reader's channel.
            assert_eq!(h.open_stream_count().await, 1);

            h.send_packet(Packet::Chunk {
                chunk: chunk("s1", 1, b"world".to_vec()),
                encryption_type: EncryptionType::None,
            });
            reader.closer().close();
            assert_eq!(h.open_stream_count().await, 0);

            assert_eq!(reader.next().await.unwrap().unwrap(), "hello");
            assert!(matches!(reader.next().await, Some(Err(StreamError::AlreadyClosed))));
            assert!(reader.next().await.is_none());
        }

        #[tokio::test]
        async fn closer_spares_a_later_stream_with_the_same_id() {
            let mut h = Harness::new();
            open_text_stream(&h, "s1");
            let first = next_text_reader(&mut h).await;
            let first_closer = first.closer();
            h.send_packet(Packet::Chunk {
                chunk: chunk("s1", 1, b"world".to_vec()),
                encryption_type: EncryptionType::None,
            });
            h.send_packet(Packet::Trailer {
                trailer: trailer("s1"),
                encryption_type: EncryptionType::None,
            });
            assert_eq!(first.read_all().await.unwrap(), "helloworld");
            assert_eq!(h.open_stream_count().await, 0);
            assert_eq!(h.drain_closed_ids(), vec!["s1"]);

            // A new stream reuses the id, and then the first stream's closer is used late.
            open_text_stream(&h, "s1");
            let second = next_text_reader(&mut h).await;
            first_closer.close();
            assert_eq!(h.open_stream_count().await, 1);
            assert!(h.drain_closed_ids().is_empty());

            h.send_packet(Packet::Chunk {
                chunk: chunk("s1", 1, b"world".to_vec()),
                encryption_type: EncryptionType::None,
            });
            h.send_packet(Packet::Trailer {
                trailer: trailer("s1"),
                encryption_type: EncryptionType::None,
            });
            assert_eq!(second.read_all().await.unwrap(), "helloworld");
        }
    }

    /// The open-stream count reflects descriptor-registered streams, answered in order with the
    /// events enqueued before the query — tests use it to wait for a header/abort to land.
    mod open_stream_count {
        use super::*;

        #[tokio::test]
        async fn counts_streams_across_their_lifecycle() {
            let mut h = Harness::new();
            assert_eq!(h.open_stream_count().await, 0);

            h.send_packet(Packet::Header {
                header: text_header("s1", Some(5), HashMap::new(), None, CompressionType::None),
                encryption_type: EncryptionType::None,
            });
            assert_eq!(h.open_stream_count().await, 1);

            h.send_packet(Packet::Header {
                header: text_header("s2", Some(5), HashMap::new(), None, CompressionType::None),
                encryption_type: EncryptionType::None,
            });
            assert_eq!(h.open_stream_count().await, 2);

            // Keep the readers alive so the streams aren't closed by reader drop.
            let (reader1, _) = h.next_opened().await;
            let (reader2, _) = h.next_opened().await;

            h.send_packet(Packet::Chunk {
                chunk: chunk("s1", 0, b"hello".to_vec()),
                encryption_type: EncryptionType::None,
            });
            h.send_packet(Packet::Trailer {
                trailer: trailer("s1"),
                encryption_type: EncryptionType::None,
            });
            assert_eq!(h.open_stream_count().await, 1);

            h.abort(ParticipantIdentity::from(SENDER));
            assert_eq!(h.open_stream_count().await, 0);
            drop((reader1, reader2));
        }

        #[tokio::test]
        async fn inline_streams_are_never_counted() {
            let mut h = Harness::new();
            h.send_packet(Packet::Header {
                header: text_header(
                    "s1",
                    Some(5),
                    HashMap::new(),
                    Some(b"hello".to_vec()),
                    CompressionType::None,
                ),
                encryption_type: EncryptionType::None,
            });
            let (reader, _) = h.next_opened().await;
            // The inline stream completes during header handling; no descriptor is registered.
            assert_eq!(h.open_stream_count().await, 0);
            assert_eq!(read_text(reader).await.unwrap(), "hello");
        }
    }

    /// Every opened stream terminates with exactly one `StreamClosed`, whatever the terminal
    /// path — hosts rely on it to sequence handler invocations for ordered topics.
    /// Stopping the run loop must error in-flight readers; dropping their chunk senders alone
    /// would end them cleanly, making truncated content look complete.
    mod shutdown {
        use super::*;

        #[tokio::test]
        async fn shutdown_errors_open_streams() {
            let mut h = Harness::new();
            h.send_packet(Packet::Header {
                header: text_header("s1", Some(10), HashMap::new(), None, CompressionType::None),
                encryption_type: EncryptionType::None,
            });
            let (text_reader, _) = h.next_opened().await;
            h.send_packet(Packet::Chunk {
                chunk: chunk("s1", 0, b"hello".to_vec()),
                encryption_type: EncryptionType::None,
            });

            let compressed = deflate_raw(b"hello world").await;
            h.send_packet(Packet::Header {
                header: byte_header("s2", Some(11), None, CompressionType::DeflateRaw),
                encryption_type: EncryptionType::None,
            });
            let (byte_reader, _) = h.next_opened().await;
            h.send_packet(Packet::Chunk {
                chunk: chunk("s2", 0, compressed[..compressed.len() / 2].to_vec()),
                encryption_type: EncryptionType::None,
            });

            h.input.send(InputEvent::Shutdown).unwrap();
            assert!(matches!(read_text(text_reader).await, Err(StreamError::AbnormalEnd(_))));
            assert!(matches!(read_bytes(byte_reader).await, Err(StreamError::AbnormalEnd(_))));
        }

        #[tokio::test]
        async fn dropping_all_inputs_errors_open_streams() {
            let Harness { input, mut output_rx } = Harness::new();
            input
                .send(InputEvent::PacketReceived(PacketReceived {
                    packet: Packet::Header {
                        header: text_header(
                            "s1",
                            Some(10),
                            HashMap::new(),
                            None,
                            CompressionType::None,
                        ),
                        encryption_type: EncryptionType::None,
                    },
                    participant_identity: ParticipantIdentity::from(SENDER),
                }))
                .unwrap();
            let reader = loop {
                if let OutputEvent::StreamOpened(StreamOpened { stream_reader, .. }) =
                    output_rx.recv().await.expect("a stream should be opened")
                {
                    break stream_reader;
                }
            };

            drop(input);
            assert!(matches!(read_text(reader).await, Err(StreamError::AbnormalEnd(_))));
        }
    }

    mod stream_closed {
        use super::*;

        #[tokio::test]
        async fn trailer_close_emits_stream_closed() {
            let mut h = Harness::new();
            let text = "hello world";
            h.send_packet(Packet::Header {
                header: text_header(
                    "s1",
                    Some(text.len() as u64),
                    HashMap::new(),
                    None,
                    CompressionType::None,
                ),
                encryption_type: EncryptionType::None,
            });
            let (reader, _) = h.next_opened().await;
            h.send_packet(Packet::Chunk {
                chunk: chunk("s1", 0, text.as_bytes().to_vec()),
                encryption_type: EncryptionType::None,
            });
            h.send_packet(Packet::Trailer {
                trailer: trailer("s1"),
                encryption_type: EncryptionType::None,
            });
            assert_eq!(h.next_closed().await, ("s1".into(), SENDER.into(), "topic".into()));
            assert_eq!(read_text(reader).await.unwrap(), text);
        }

        #[tokio::test]
        async fn inline_stream_emits_stream_closed() {
            // Inline single-packet streams never receive a trailer, so the closed signal must be
            // synthesized when the inline payload completes.
            let mut h = Harness::new();
            h.send_packet(Packet::Header {
                header: text_header(
                    "s1",
                    Some(5),
                    HashMap::new(),
                    Some(b"hello".to_vec()),
                    CompressionType::None,
                ),
                encryption_type: EncryptionType::None,
            });
            let (reader, _) = h.next_opened().await;
            assert_eq!(h.next_closed().await, ("s1".into(), SENDER.into(), "topic".into()));
            assert_eq!(read_text(reader).await.unwrap(), "hello");
        }

        #[tokio::test]
        async fn error_close_emits_stream_closed() {
            let mut h = Harness::new();
            h.send_packet(Packet::Header {
                header: text_header("s1", Some(10), HashMap::new(), None, CompressionType::None),
                encryption_type: EncryptionType::None,
            });
            let (reader, _) = h.next_opened().await;
            // A chunk-index gap closes the stream with `MissedChunk`.
            h.send_packet(Packet::Chunk {
                chunk: chunk("s1", 5, b"hello".to_vec()),
                encryption_type: EncryptionType::None,
            });
            assert_eq!(h.next_closed().await, ("s1".into(), SENDER.into(), "topic".into()));
            assert!(matches!(read_text(reader).await, Err(StreamError::MissedChunk)));
        }

        #[tokio::test]
        async fn abort_emits_stream_closed() {
            let mut h = Harness::new();
            h.send_packet(Packet::Header {
                header: text_header("s1", Some(10), HashMap::new(), None, CompressionType::None),
                encryption_type: EncryptionType::None,
            });
            let (reader, _) = h.next_opened().await;
            h.abort(ParticipantIdentity::from(SENDER));
            assert_eq!(h.next_closed().await, ("s1".into(), SENDER.into(), "topic".into()));
            assert!(matches!(read_text(reader).await, Err(StreamError::AbnormalEnd(_))));
        }

        #[tokio::test]
        async fn trailer_for_unopened_stream_emits_no_stream_closed() {
            let mut h = Harness::new();
            h.send_packet(Packet::Trailer {
                trailer: trailer("never-opened"),
                encryption_type: EncryptionType::None,
            });
            // A second, well-formed inline stream: if the orphan trailer had produced a closed
            // event, it would be observed before this stream's.
            h.send_packet(Packet::Header {
                header: text_header(
                    "s2",
                    Some(2),
                    HashMap::new(),
                    Some(b"hi".to_vec()),
                    CompressionType::None,
                ),
                encryption_type: EncryptionType::None,
            });
            let (closed_id, _, _) = h.next_closed().await;
            assert_eq!(closed_id, "s2");
        }
    }

    #[tokio::test]
    async fn empty_chunks_are_ignored() {
        let mut h = Harness::new();
        let text = "hello world";
        h.send_packet(Packet::Header {
            header: text_header(
                "s1",
                Some(text.len() as u64),
                HashMap::new(),
                None,
                CompressionType::None,
            ),
            encryption_type: EncryptionType::None,
        });
        let (reader, _) = h.next_opened().await;
        // An empty chunk must not count against totalLength or corrupt the stream.
        h.send_packet(Packet::Chunk {
            chunk: chunk("s1", 0, vec![]),
            encryption_type: EncryptionType::None,
        });
        h.send_packet(Packet::Chunk {
            chunk: chunk("s1", 1, text.as_bytes().to_vec()),
            encryption_type: EncryptionType::None,
        });
        h.send_packet(Packet::Trailer {
            trailer: trailer("s1"),
            encryption_type: EncryptionType::None,
        });
        assert_eq!(read_text(reader).await.unwrap(), text);
    }

    #[tokio::test]
    async fn trailer_with_reason_errors_abnormal_end() {
        let mut h = Harness::new();
        // The stream is short of its declared length, and the trailer's reason must win over that.
        h.send_packet(Packet::Header {
            header: text_header("s1", Some(10), HashMap::new(), None, CompressionType::None),
            encryption_type: EncryptionType::None,
        });
        let (reader, _) = h.next_opened().await;
        h.send_packet(Packet::Chunk {
            chunk: chunk("s1", 0, b"hello".to_vec()),
            encryption_type: EncryptionType::None,
        });
        h.send_packet(Packet::Trailer {
            trailer: Trailer {
                stream_id: StreamId::from("s1"),
                reason: "cancelled".to_string(),
                attributes: HashMap::new(),
            },
            encryption_type: EncryptionType::None,
        });
        assert!(
            matches!(read_text(reader).await, Err(StreamError::AbnormalEnd(r)) if r == "cancelled")
        );
    }

    #[tokio::test]
    async fn text_stream_with_attachments_round_trips() {
        let mut h = Harness::new();
        let text = "hello world";

        // Text stream whose header references an attachment stream id, body inline.
        let text_hdr =
            TextHeader { attached_stream_ids: vec![StreamId::from("att1")], ..Default::default() };
        h.send_packet(Packet::Header {
            header: Header {
                stream_id: StreamId::from("s1"),
                timestamp: 0,
                topic: "topic".to_string(),
                mime_type: "text/plain".to_string(),
                total_length: Some(text.len() as u64),
                attributes: HashMap::new(),
                content_header: Some(text_hdr.into()),
                inline_content: Some(text.as_bytes().to_vec()),
                compression: CompressionType::None,
            },
            encryption_type: EncryptionType::None,
        });
        let (text_reader, _) = h.next_opened().await;
        assert_eq!(text_info(&text_reader).attached_stream_ids, vec!["att1".to_string()]);
        assert_eq!(read_text(text_reader).await.unwrap(), text);

        // The attachment arrives as its own byte stream under the referenced id.
        h.send_packet(Packet::Header {
            header: byte_header("att1", Some(3), None, CompressionType::None),
            encryption_type: EncryptionType::None,
        });
        let (byte_reader, _) = h.next_opened().await;
        h.send_packet(Packet::Chunk {
            chunk: chunk("att1", 0, vec![1, 2, 3]),
            encryption_type: EncryptionType::None,
        });
        h.send_packet(Packet::Trailer {
            trailer: trailer("att1"),
            encryption_type: EncryptionType::None,
        });
        assert_eq!(read_bytes(byte_reader).await.unwrap(), Bytes::from(vec![1u8, 2, 3]));
    }

    /// Chunk and trailer packets carry only a stream id, so the manager reports the topic of the
    /// stream they belong to. Hosts rely on this to filter events for topics they handle
    /// internally (e.g. RPC), so it is the only signal available to them for these two events.
    mod reported_topic {
        use super::*;

        /// Awaits the next chunk/trailer output, returning the topic it reported.
        async fn next_raw_topic(h: &mut Harness) -> Option<String> {
            loop {
                match h.output_rx.recv().await.expect("an output event should be emitted") {
                    OutputEvent::ChunkReceived(ChunkReceived { topic, .. })
                    | OutputEvent::TrailerReceived(TrailerReceived { topic, .. }) => {
                        return topic;
                    }
                    _ => continue,
                }
            }
        }

        #[tokio::test]
        async fn chunk_and_trailer_report_the_topic_of_their_stream() {
            let mut h = Harness::new();
            h.send_packet(Packet::Header {
                header: Header {
                    topic: "lk.rpc_request".to_string(),
                    ..text_header("s1", Some(2), HashMap::new(), None, CompressionType::None)
                },
                encryption_type: EncryptionType::None,
            });
            let (reader, _) = h.next_opened().await;

            h.send_packet(Packet::Chunk {
                chunk: chunk("s1", 0, b"hi".to_vec()),
                encryption_type: EncryptionType::None,
            });
            assert_eq!(next_raw_topic(&mut h).await.as_deref(), Some("lk.rpc_request"));

            h.send_packet(Packet::Trailer {
                trailer: trailer("s1"),
                encryption_type: EncryptionType::None,
            });
            assert_eq!(next_raw_topic(&mut h).await.as_deref(), Some("lk.rpc_request"));

            // The stream itself still opens and reads normally; reporting the topic does not
            // suppress anything in this crate.
            assert_eq!(read_text(reader).await.unwrap(), "hi");
        }

        #[tokio::test]
        async fn chunk_for_an_unopened_stream_reports_no_topic() {
            let mut h = Harness::new();
            h.send_packet(Packet::Chunk {
                chunk: chunk("never-opened", 0, b"hi".to_vec()),
                encryption_type: EncryptionType::None,
            });
            assert_eq!(next_raw_topic(&mut h).await, None);
        }
    }
}