bitwarden_ipc/

ipc_client.rs

use std::sync::{Arc, Mutex};

use bitwarden_threading::cancellation_token::CancellationToken;
use serde::de::DeserializeOwned;
use thiserror::Error;
use tokio::select;

use crate::{
    broadcast::{BroadcastGroup, EndpointSendResult},
    constants::CHANNEL_BUFFER_CAPACITY,
    error::{
        AlreadyRunningError, BroadcastError, ErrorKind, IpcErrorKind, ReceiveError, SendError,
        SubscribeError, TypedReceiveError,
    },
    fragmentation::{
        FragmentationError, Fragments, MAX_PAYLOAD_SIZE, REASSEMBLY_IDLE_TIMEOUT,
        REASSEMBLY_TIMEOUT, Reassembler,
    },
    ipc_client_trait::IpcClient,
    message::{IncomingMessage, OutgoingMessage, PayloadTypeName, TypedIncomingMessage},
    rpc::{
        error::RpcError,
        exec::{
            handler::{ErasedRpcHandler, RpcRequestInfo},
            handler_registry::RpcHandlerRegistry,
        },
        request_message::{RPC_REQUEST_PAYLOAD_TYPE_NAME, RpcRequestPayload},
        response_message::OutgoingRpcResponseMessage,
    },
    serde_utils,
    traits::{BroadcastResolution, CommunicationBackend, CryptoProvider, SessionRepository},
};

/// A subscription to receive messages over IPC.
/// The subcription will start buffering messages after its creation and return them
/// when receive() is called. Messages received before the subscription was created will not be
/// returned.
pub struct IpcClientSubscription {
    pub(crate) receiver: tokio::sync::broadcast::Receiver<IncomingMessage>,
    pub(crate) topic: Option<String>,
}

/// A subscription to receive messages over IPC.
/// The subcription will start buffering messages after its creation and return them
/// when receive() is called. Messages received before the subscription was created will not be
/// returned.
pub struct IpcClientTypedSubscription<Payload: DeserializeOwned + PayloadTypeName>(
    IpcClientSubscription,
    std::marker::PhantomData<Payload>,
);

/// Internal shared state for the IPC client.
struct IpcClientInner<Crypto, Com, Ses>
where
    Crypto: CryptoProvider<Com, Ses>,
    Com: CommunicationBackend,
    Ses: SessionRepository<Crypto::Session>,
{
    crypto: Crypto,
    communication: Com,
    sessions: Ses,

    handlers: RpcHandlerRegistry,
    reassembler: Reassembler,
    incoming: Mutex<Option<tokio::sync::broadcast::WeakSender<IncomingMessage>>>,
    cancellation_token: Mutex<Option<CancellationToken>>,
}

/// An IPC client that handles communication between different components and clients.
/// It uses a crypto provider to encrypt and decrypt messages, a communication backend to send and
/// receive messages, and a session repository to persist sessions.
///
/// This is the concrete implementation of the [`IpcClient`] trait.
pub struct IpcClientImpl<Crypto, Com, Ses>
where
    Crypto: CryptoProvider<Com, Ses>,
    Com: CommunicationBackend,
    Ses: SessionRepository<Crypto::Session>,
{
    inner: Arc<IpcClientInner<Crypto, Com, Ses>>,
}

impl<Crypto, Com, Ses> Clone for IpcClientImpl<Crypto, Com, Ses>
where
    Crypto: CryptoProvider<Com, Ses>,
    Com: CommunicationBackend,
    Ses: SessionRepository<Crypto::Session>,
{
    fn clone(&self) -> Self {
        Self {
            inner: self.inner.clone(),
        }
    }
}

impl<Crypto, Com, Ses> IpcClientImpl<Crypto, Com, Ses>
where
    Crypto: CryptoProvider<Com, Ses>,
    Com: CommunicationBackend,
    Ses: SessionRepository<Crypto::Session>,
{
    /// Create a new IPC client with the provided crypto provider, communication backend, and
    /// session repository.
    pub fn new(crypto: Crypto, communication: Com, sessions: Ses) -> Self {
        Self {
            inner: Arc::new(IpcClientInner {
                crypto,
                communication,
                sessions,

                handlers: RpcHandlerRegistry::new(),
                reassembler: Reassembler::new(REASSEMBLY_TIMEOUT, REASSEMBLY_IDLE_TIMEOUT),
                incoming: Mutex::new(None),
                cancellation_token: Mutex::new(None),
            }),
        }
    }
}

#[async_trait::async_trait]
impl<Crypto, Com, Ses> crate::ipc_client_trait::IpcClient for IpcClientImpl<Crypto, Com, Ses>
where
    Crypto: CryptoProvider<Com, Ses>,
    Com: CommunicationBackend,
    Ses: SessionRepository<Crypto::Session>,
{
    async fn start(
        &self,
        cancellation_token: Option<CancellationToken>,
    ) -> Result<(), AlreadyRunningError> {
        if self.is_running() {
            return Err(AlreadyRunningError);
        }

        let cancellation_token = cancellation_token.unwrap_or_default();
        self.inner
            .cancellation_token
            .lock()
            .expect("Failed to lock cancellation token mutex")
            .replace(cancellation_token.clone());

        let com_receiver = self.inner.communication.subscribe().await;
        let (client_tx, _) = tokio::sync::broadcast::channel(CHANNEL_BUFFER_CAPACITY);

        self.inner
            .incoming
            .lock()
            .expect("Failed to lock incoming mutex")
            .replace(client_tx.downgrade());

        let inner = self.inner.clone();
        let future = async move {
            loop {
                select! {
                    _ = cancellation_token.cancelled() => {
                        tracing::debug!("Cancellation signal received, stopping IPC client");
                        break;
                    }
                    received = inner.crypto.receive(&com_receiver, &inner.communication, &inner.sessions) => {
                        match received {
                            Ok(message) => {
                                let Some(message) = inner.reassembler.process(message) else {
                                    continue;
                                };
                                if message.topic.as_deref() == Some(RPC_REQUEST_PAYLOAD_TYPE_NAME) {
                                    handle_rpc_request(&inner, message);
                                } else if client_tx.send(message).is_err() {
                                    tracing::debug!("No subscriptions for incoming message, dropping it");
                                }
                            }
                            Err(error) if matches!(error.kind(), ErrorKind::Fatal) => {
                                tracing::error!(?error, "Fatal error receiving message, stopping IPC client");
                                break;
                            }
                            Err(error) => {
                                tracing::warn!(?error, "Recoverable error receiving message, continuing");
                            }
                        }
                    }
                }
            }
            tracing::debug!("IPC client shutting down");
            stop_inner(&inner);
        };

        bitwarden_threading::spawn(future);

        Ok(())
    }

    fn is_running(&self) -> bool {
        let has_incoming = self
            .inner
            .incoming
            .lock()
            .expect("Failed to lock incoming mutex")
            .as_ref()
            .is_some_and(|sender| sender.strong_count() > 0);
        let has_cancellation_token = self
            .inner
            .cancellation_token
            .lock()
            .expect("Failed to lock cancellation token mutex")
            .is_some();
        has_incoming && has_cancellation_token
    }

    async fn send(&self, message: OutgoingMessage) -> Result<(), SendError> {
        send_message(&self.inner, message).await.result
    }

    async fn broadcast(
        &self,
        group: BroadcastGroup,
        payload: Vec<u8>,
        topic: Option<String>,
    ) -> Result<Vec<EndpointSendResult>, BroadcastError> {
        let endpoints = match self.inner.communication.resolve(&group).await {
            BroadcastResolution::Endpoints(endpoints) => endpoints,
            BroadcastResolution::Delegate(_) => return Err(BroadcastError::DelegationUnsupported),
            BroadcastResolution::Unsupported => return Err(BroadcastError::Unsupported),
        };

        // Validate the whole resolution before sending anything, so a backend that names an
        // endpoint outside the group cannot cause a partial delivery to it.
        if let Some(endpoint) = endpoints.iter().find(|endpoint| !group.contains(endpoint)) {
            return Err(BroadcastError::OutOfScope(endpoint.clone()));
        }

        // Endpoints are sent to one at a time. The crypto provider may serialize sends anyway (the
        // Noise provider holds one lock per client for the whole send, handshake included), in
        // which case concurrent sends would not deliver any sooner. A fatal error stops the client,
        // so the remaining endpoints are not attempted and are absent from the results.
        let mut results = Vec::with_capacity(endpoints.len());
        for endpoint in endpoints {
            let message = OutgoingMessage {
                payload: payload.clone(),
                destination: endpoint.clone(),
                topic: topic.clone(),
            };
            let outcome = send_message(&self.inner, message).await;
            results.push(EndpointSendResult {
                endpoint,
                result: outcome.result,
            });
            if outcome.fatal {
                break;
            }
        }

        Ok(results)
    }

    async fn subscribe(
        &self,
        topic: Option<String>,
    ) -> Result<IpcClientSubscription, SubscribeError> {
        Ok(IpcClientSubscription {
            receiver: self
                .inner
                .incoming
                .lock()
                .expect("Failed to lock incoming mutex")
                .as_ref()
                .and_then(|sender| sender.upgrade())
                .ok_or(SubscribeError::NotStarted)?
                .subscribe(),
            topic,
        })
    }

    async fn register_rpc_handler_erased(&self, name: &str, handler: Box<dyn ErasedRpcHandler>) {
        self.inner
            .handlers
            .register_erased(name.to_owned(), handler)
            .await;
    }
}

/// The outcome of [`send_message`].
struct SendOutcome {
    result: Result<(), SendError>,
    /// Whether the send failed with a fatal error, which has stopped the client.
    fatal: bool,
}

/// Send a message through the crypto provider, split into as many fragments as the provider's
/// payload limit requires. Fragments are sent in order and sending stops at the first failed
/// fragment. A fatal error stops the client.
async fn send_message<Crypto, Com, Ses>(
    inner: &IpcClientInner<Crypto, Com, Ses>,
    message: OutgoingMessage,
) -> SendOutcome
where
    Crypto: CryptoProvider<Com, Ses>,
    Com: CommunicationBackend,
    Ses: SessionRepository<Crypto::Session>,
{
    let fragments = match Fragments::split(message, inner.crypto.max_payload_len()) {
        Ok(fragments) => fragments,
        Err(e) => {
            return SendOutcome {
                result: Err(SendError::Other(e.to_string())),
                fatal: false,
            };
        }
    };

    for fragment in fragments {
        let outcome = send_fragment(inner, fragment).await;
        if outcome.result.is_err() {
            return outcome;
        }
    }

    SendOutcome {
        result: Ok(()),
        fatal: false,
    }
}

/// Send a single fragment through the crypto provider, stopping the client on a fatal error.
async fn send_fragment<Crypto, Com, Ses>(
    inner: &IpcClientInner<Crypto, Com, Ses>,
    message: OutgoingMessage,
) -> SendOutcome
where
    Crypto: CryptoProvider<Com, Ses>,
    Com: CommunicationBackend,
    Ses: SessionRepository<Crypto::Session>,
{
    let result = inner
        .crypto
        .send(&inner.communication, &inner.sessions, message)
        .await;

    if let Err(ref error) = result {
        match error.kind() {
            ErrorKind::Fatal => {
                tracing::error!(?error, "Fatal error sending message, stopping IPC client");
                stop_inner(inner);
            }
            // An unreachable destination is an expected condition and not logged
            ErrorKind::Unreachable => {}
            // Every other recoverable send failure is still surfaced.
            ErrorKind::Other => {
                tracing::debug!(
                    ?error,
                    "Recoverable error sending message, IPC client will continue running"
                );
            }
        }
    }

    SendOutcome {
        fatal: matches!(&result, Err(e) if e.kind() == ErrorKind::Fatal),
        result: result.map_err(|e| match e.kind() {
            ErrorKind::Unreachable => SendError::Unreachable,
            _ => SendError::Other(format!("{e:?}")),
        }),
    }
}

fn stop_inner<Crypto, Com, Ses>(inner: &IpcClientInner<Crypto, Com, Ses>)
where
    Crypto: CryptoProvider<Com, Ses>,
    Com: CommunicationBackend,
    Ses: SessionRepository<Crypto::Session>,
{
    let mut cancellation_token = inner
        .cancellation_token
        .lock()
        .expect("Failed to lock cancellation token mutex");

    if let Some(cancellation_token) = cancellation_token.take() {
        cancellation_token.cancel();
    }
}

fn handle_rpc_request<Crypto, Com, Ses>(
    inner: &Arc<IpcClientInner<Crypto, Com, Ses>>,
    incoming_message: IncomingMessage,
) where
    Crypto: CryptoProvider<Com, Ses>,
    Com: CommunicationBackend,
    Ses: SessionRepository<Crypto::Session>,
{
    let inner = inner.clone();
    let future = async move {
        #[derive(Debug, Error)]
        enum HandleError {
            #[error("Failed to deserialize request message: {0}")]
            Deserialize(String),

            #[error("Failed to serialize response message: {0}")]
            Serialize(String),
        }

        async fn handle(
            incoming_message: IncomingMessage,
            handlers: &RpcHandlerRegistry,
        ) -> Result<OutgoingMessage, HandleError> {
            let IncomingMessage {
                payload, source, ..
            } = incoming_message;
            let request = RpcRequestPayload::from_slice(payload).map_err(
                |e: serde_utils::DeserializeError| HandleError::Deserialize(e.to_string()),
            )?;

            let info = RpcRequestInfo {
                source: source.clone(),
            };
            let response = handlers.handle(&request, info).await;

            let response_message = OutgoingRpcResponseMessage {
                request_id: request.request_id(),
                request_type: request.request_type(),
                result: response,
            };

            // Publish the response on the dedicated topic the requester is subscribed to.
            let mut payload = serde_utils::to_vec(&response_message)
                .map_err(|e: serde_utils::SerializeError| HandleError::Serialize(e.to_string()))?;
            if payload.len() > MAX_PAYLOAD_SIZE {
                // Report the error, or the requester waits for a response that never arrives.
                let error = FragmentationError::PayloadTooLarge(payload.len()).to_string();
                let error_message = OutgoingRpcResponseMessage {
                    result: Err(RpcError::ResponseSerialization(error)),
                    ..response_message
                };
                payload = serde_utils::to_vec(&error_message).map_err(
                    |e: serde_utils::SerializeError| HandleError::Serialize(e.to_string()),
                )?;
            }

            Ok(OutgoingMessage {
                payload,
                destination: source.into(),
                topic: Some(request.response_topic().to_owned()),
            })
        }

        match handle(incoming_message, &inner.handlers).await {
            Ok(outgoing_message) => {
                // Send through the client, so the response is fragmented like any other message.
                match (IpcClientImpl { inner }).send(outgoing_message).await {
                    // A requester that went away is an expected condition and not logged.
                    Ok(()) | Err(SendError::Unreachable) => {}
                    Err(error) => tracing::error!(%error, "Failed to send response message"),
                }
            }
            Err(error) => {
                tracing::error!(%error, "Error handling RPC request");
            }
        }
    };

    bitwarden_threading::spawn(future);
}

impl IpcClientSubscription {
    /// Receive a message, optionally filtering by topic.
    /// Setting the cancellation_token to `None` will wait indefinitely.
    pub async fn receive(
        &mut self,
        cancellation_token: Option<CancellationToken>,
    ) -> Result<IncomingMessage, ReceiveError> {
        let cancellation_token = cancellation_token.unwrap_or_default();

        loop {
            select! {
                _ = cancellation_token.cancelled() => {
                    return Err(ReceiveError::Cancelled)
                }
                result = self.receiver.recv() => {
                    let received = result?;
                    if self.topic.is_none() || received.topic == self.topic {
                        return Ok::<IncomingMessage, ReceiveError>(received);
                    }
                }
            }
        }
    }
}

impl<Payload> IpcClientTypedSubscription<Payload>
where
    Payload: DeserializeOwned + PayloadTypeName,
{
    pub(crate) fn new(subscription: IpcClientSubscription) -> Self {
        Self(subscription, std::marker::PhantomData)
    }

    /// Receive a message.
    /// Setting the cancellation_token to `None` will wait indefinitely.
    pub async fn receive(
        &mut self,
        cancellation_token: Option<CancellationToken>,
    ) -> Result<TypedIncomingMessage<Payload>, TypedReceiveError> {
        let received = self.0.receive(cancellation_token).await?;
        received
            .try_into()
            .map_err(|e: serde_utils::DeserializeError| TypedReceiveError::Typing(e.to_string()))
    }
}

#[cfg(test)]
mod tests {
    use std::{collections::HashMap, time::Duration};

    use bitwarden_threading::time::sleep;
    use serde::{Deserialize, Serialize};

    use super::*;
    use crate::{
        IpcClientExt,
        endpoint::{Endpoint, HostId, Source},
        ipc_client_trait::IpcClient,
        message::PayloadTypeName,
        rpc::{
            request::RpcRequest,
            request_message::{RPC_REQUEST_PAYLOAD_TYPE_NAME, RpcRequestMessage},
            response_message::IncomingRpcResponseMessage,
        },
        traits::{InMemorySessionRepository, NoEncryptionCryptoProvider, TestCommunicationBackend},
    };

    /// Error type for [`TestCryptoProvider`] that carries an explicit fatal/recoverable
    /// classification so tests can exercise both control-flow paths.
    #[derive(Debug, Clone)]
    struct TestCryptoError {
        // Read only through the derived `Debug` impl (the outer `SendError` wraps it via
        // `{e:?}`), which dead-code analysis does not count as a use.
        #[allow(dead_code)]
        message: String,
        fatal: bool,
    }

    impl IpcErrorKind for TestCryptoError {
        fn kind(&self) -> ErrorKind {
            if self.fatal {
                ErrorKind::Fatal
            } else {
                ErrorKind::Other
            }
        }
    }

    struct TestCryptoProvider {
        /// Simulate a send result. Set to `None` wait indefinitely
        send_result: Option<Result<(), TestCryptoError>>,
        /// Simulate a receive result. Set to `None` wait indefinitely
        receive_result: Option<Result<IncomingMessage, TestCryptoError>>,
    }

    type TestSessionRepository = InMemorySessionRepository<String>;
    impl CryptoProvider<TestCommunicationBackend, TestSessionRepository> for TestCryptoProvider {
        type Session = String;
        type SendError = TestCryptoError;
        type ReceiveError = TestCryptoError;

        fn max_payload_len(&self) -> usize {
            usize::MAX
        }

        async fn receive(
            &self,
            _receiver: &<TestCommunicationBackend as CommunicationBackend>::Receiver,
            _communication: &TestCommunicationBackend,
            _sessions: &TestSessionRepository,
        ) -> Result<IncomingMessage, Self::ReceiveError> {
            match &self.receive_result {
                Some(result) => {
                    // Yield (and throttle) so a recoverable error that makes the processing loop
                    // `continue` doesn't busy-spin and starve the single-threaded test runtime.
                    // Real backends await their underlying transport here, which has the same
                    // effect.
                    sleep(Duration::from_millis(5)).await;
                    result.clone()
                }
                None => {
                    // Simulate waiting for a message but never returning
                    sleep(Duration::from_secs(600)).await;
                    Err(TestCryptoError {
                        message: "Simulated timeout".to_string(),
                        fatal: true,
                    })
                }
            }
        }

        async fn send(
            &self,
            _communication: &TestCommunicationBackend,
            _sessions: &TestSessionRepository,
            _message: OutgoingMessage,
        ) -> Result<(), Self::SendError> {
            match &self.send_result {
                Some(result) => result.clone(),
                None => {
                    // Simulate waiting for a message to be send but never returning
                    sleep(Duration::from_secs(600)).await;
                    Err(TestCryptoError {
                        message: "Simulated timeout".to_string(),
                        fatal: true,
                    })
                }
            }
        }
    }

    #[tokio::test]
    async fn returns_send_error_when_crypto_provider_returns_error() {
        let message = OutgoingMessage {
            payload: vec![],
            destination: Endpoint::BrowserBackground { id: HostId::Own },
            topic: None,
        };
        let crypto_provider = TestCryptoProvider {
            send_result: Some(Err(TestCryptoError {
                message: "Crypto error".to_string(),
                fatal: false,
            })),
            receive_result: Some(Err(TestCryptoError {
                message: "Should not have be called".to_string(),
                fatal: false,
            })),
        };
        let communication_provider = TestCommunicationBackend::new();
        let session_map = TestSessionRepository::new(HashMap::new());
        let client = IpcClientImpl::new(crypto_provider, communication_provider, session_map);
        let _ = client.start(None).await;

        let error = client.send(message).await.unwrap_err();

        assert!(error.to_string().contains("Crypto error"));
    }

    #[tokio::test]
    async fn communication_provider_has_outgoing_message_when_sending_through_ipc_client() {
        let message = OutgoingMessage {
            payload: vec![],
            destination: Endpoint::BrowserBackground { id: HostId::Own },
            topic: None,
        };
        let crypto_provider = NoEncryptionCryptoProvider;
        let communication_provider = TestCommunicationBackend::new();
        let session_map = InMemorySessionRepository::new(HashMap::new());
        let client =
            IpcClientImpl::new(crypto_provider, communication_provider.clone(), session_map);
        let _ = client.start(None).await;

        client.send(message.clone()).await.unwrap();

        let outgoing_messages = communication_provider.outgoing().await;
        assert_eq!(outgoing_messages, vec![message]);
    }

    #[tokio::test]
    async fn returns_received_message_when_received_from_backend() {
        let message = IncomingMessage {
            payload: vec![],
            source: Source::Web {
                tab_id: 9001,
                document_id: "doc-1".to_string(),
                origin: "https://example.com".to_string(),
            },
            destination: Endpoint::BrowserBackground { id: HostId::Own },
            topic: None,
        };
        let crypto_provider = NoEncryptionCryptoProvider;
        let communication_provider = TestCommunicationBackend::new();
        let session_map = InMemorySessionRepository::new(HashMap::new());
        let client =
            IpcClientImpl::new(crypto_provider, communication_provider.clone(), session_map);
        let _ = client.start(None).await;

        let mut subscription = client
            .subscribe(None)
            .await
            .expect("Subscribing should not fail");
        communication_provider.push_incoming(message.clone());
        let received_message = subscription.receive(None).await.unwrap();

        assert_eq!(received_message, message);
    }

    #[tokio::test]
    async fn channel_keeps_no_message_every_subscription_has_received() {
        let message = IncomingMessage {
            payload: vec![1, 2, 3],
            source: Source::DesktopMain,
            destination: Endpoint::BrowserBackground { id: HostId::Own },
            topic: Some("topic".to_owned()),
        };
        let communication_provider = TestCommunicationBackend::new();
        let client = IpcClientImpl::new(
            NoEncryptionCryptoProvider,
            communication_provider.clone(),
            InMemorySessionRepository::new(HashMap::new()),
        );
        let _ = client.start(None).await;
        let queued = || {
            client
                .inner
                .incoming
                .lock()
                .unwrap()
                .as_ref()
                .and_then(|sender| sender.upgrade())
                .expect("Client should be running")
                .len()
        };

        // Without subscriptions, the message is dropped and the client keeps running.
        communication_provider.push_incoming(message.clone());
        sleep(Duration::from_millis(50)).await;
        assert_eq!(queued(), 0);
        assert!(client.is_running());

        let mut subscription = client.subscribe(None).await.unwrap();
        communication_provider.push_incoming(message.clone());
        assert_eq!(subscription.receive(None).await.unwrap(), message);
        assert_eq!(queued(), 0);
    }

    #[tokio::test]
    async fn skips_non_matching_topics_and_returns_first_matching_message() {
        let non_matching_message = IncomingMessage {
            payload: vec![],
            source: Source::Web {
                tab_id: 9001,
                document_id: "doc-1".to_string(),
                origin: "https://example.com".to_string(),
            },
            destination: Endpoint::BrowserBackground { id: HostId::Own },
            topic: Some("non_matching_topic".to_owned()),
        };
        let matching_message = IncomingMessage {
            payload: vec![109],
            source: Source::Web {
                tab_id: 9001,
                document_id: "doc-1".to_string(),
                origin: "https://example.com".to_string(),
            },
            destination: Endpoint::BrowserBackground { id: HostId::Own },
            topic: Some("matching_topic".to_owned()),
        };

        let crypto_provider = NoEncryptionCryptoProvider;
        let communication_provider = TestCommunicationBackend::new();
        let session_map = InMemorySessionRepository::new(HashMap::new());
        let client =
            IpcClientImpl::new(crypto_provider, communication_provider.clone(), session_map);
        let _ = client.start(None).await;
        let mut subscription = client
            .subscribe(Some("matching_topic".to_owned()))
            .await
            .expect("Subscribing should not fail");
        communication_provider.push_incoming(non_matching_message.clone());
        communication_provider.push_incoming(non_matching_message.clone());
        communication_provider.push_incoming(matching_message.clone());

        let received_message: IncomingMessage = subscription.receive(None).await.unwrap();

        assert_eq!(received_message, matching_message);
    }

    #[tokio::test]
    async fn skips_unrelated_messages_and_returns_typed_message() {
        #[derive(Debug, Clone, PartialEq, Deserialize, Serialize)]
        struct TestPayload {
            some_data: String,
        }

        impl PayloadTypeName for TestPayload {
            const PAYLOAD_TYPE_NAME: &str = "TestPayload";
        }

        let unrelated = IncomingMessage {
            payload: vec![],
            source: Source::Web {
                tab_id: 9001,
                document_id: "doc-1".to_string(),
                origin: "https://example.com".to_string(),
            },
            destination: Endpoint::BrowserBackground { id: HostId::Own },
            topic: None,
        };
        let typed_message = crate::message::TypedIncomingMessage {
            payload: TestPayload {
                some_data: "Hello, world!".to_string(),
            },
            source: Source::Web {
                tab_id: 9001,
                document_id: "doc-1".to_string(),
                origin: "https://example.com".to_string(),
            },
            destination: Endpoint::BrowserBackground { id: HostId::Own },
        };

        let crypto_provider = NoEncryptionCryptoProvider;
        let communication_provider = TestCommunicationBackend::new();
        let session_map = InMemorySessionRepository::new(HashMap::new());
        let client =
            IpcClientImpl::new(crypto_provider, communication_provider.clone(), session_map);
        let _ = client.start(None).await;
        let mut subscription = client
            .subscribe_typed::<TestPayload>()
            .await
            .expect("Subscribing should not fail");
        communication_provider.push_incoming(unrelated.clone());
        communication_provider.push_incoming(unrelated.clone());
        communication_provider.push_incoming(
            typed_message
                .clone()
                .try_into()
                .expect("Serialization should not fail"),
        );

        let received_message = subscription.receive(None).await.unwrap();

        assert_eq!(received_message, typed_message);
    }

    #[tokio::test]
    async fn returns_error_if_related_message_was_not_deserializable() {
        #[derive(Debug, Clone, PartialEq, Deserialize, Serialize)]
        struct TestPayload {
            some_data: String,
        }

        impl PayloadTypeName for TestPayload {
            const PAYLOAD_TYPE_NAME: &str = "TestPayload";
        }

        let non_deserializable_message = IncomingMessage {
            payload: vec![],
            source: Source::Web {
                tab_id: 9001,
                document_id: "doc-1".to_string(),
                origin: "https://example.com".to_string(),
            },
            destination: Endpoint::BrowserBackground { id: HostId::Own },
            topic: Some("TestPayload".to_owned()),
        };

        let crypto_provider = NoEncryptionCryptoProvider;
        let communication_provider = TestCommunicationBackend::new();
        let session_map = InMemorySessionRepository::new(HashMap::new());
        let client =
            IpcClientImpl::new(crypto_provider, communication_provider.clone(), session_map);
        let _ = client.start(None).await;
        let mut subscription = client
            .subscribe_typed::<TestPayload>()
            .await
            .expect("Subscribing should not fail");
        communication_provider.push_incoming(non_deserializable_message.clone());

        let result = subscription.receive(None).await;
        assert!(matches!(result, Err(TypedReceiveError::Typing(_))));
    }

    #[tokio::test]
    async fn ipc_client_stops_if_crypto_returns_fatal_send_error() {
        let message = OutgoingMessage {
            payload: vec![],
            destination: Endpoint::BrowserBackground { id: HostId::Own },
            topic: None,
        };
        let crypto_provider = TestCryptoProvider {
            send_result: Some(Err(TestCryptoError {
                message: "Crypto error".to_string(),
                fatal: true,
            })),
            receive_result: None,
        };
        let communication_provider = TestCommunicationBackend::new();
        let session_map = TestSessionRepository::new(HashMap::new());
        let client = IpcClientImpl::new(crypto_provider, communication_provider, session_map);
        let _ = client.start(None).await;

        let error = client.send(message).await.unwrap_err();
        let is_running = client.is_running();

        assert!(error.to_string().contains("Crypto error"));
        assert!(!is_running);
    }

    #[tokio::test]
    async fn ipc_client_keeps_running_if_crypto_returns_recoverable_send_error() {
        let message = OutgoingMessage {
            payload: vec![],
            destination: Endpoint::BrowserBackground { id: HostId::Own },
            topic: None,
        };
        let crypto_provider = TestCryptoProvider {
            // A recoverable send error (e.g. a handshake timeout because the peer is down) must
            // not tear down the shared client.
            send_result: Some(Err(TestCryptoError {
                message: "Crypto error".to_string(),
                fatal: false,
            })),
            // Block forever on receive so the loop stays alive while we inspect it.
            receive_result: None,
        };
        let communication_provider = TestCommunicationBackend::new();
        let session_map = TestSessionRepository::new(HashMap::new());
        let client = IpcClientImpl::new(crypto_provider, communication_provider, session_map);
        let _ = client.start(None).await;

        let error = client.send(message).await.unwrap_err();
        let is_running = client.is_running();

        // The error is still surfaced to the caller...
        assert!(error.to_string().contains("Crypto error"));
        // ...but the client keeps running so future sends/requests can succeed.
        assert!(is_running);
    }

    #[tokio::test]
    async fn ipc_client_stops_if_crypto_returns_fatal_receive_error() {
        let crypto_provider = TestCryptoProvider {
            send_result: None,
            receive_result: Some(Err(TestCryptoError {
                message: "Crypto error".to_string(),
                fatal: true,
            })),
        };
        let communication_provider = TestCommunicationBackend::new();
        let session_map = TestSessionRepository::new(HashMap::new());
        let client = IpcClientImpl::new(crypto_provider, communication_provider, session_map);
        let cancellation_token = CancellationToken::new();
        let _ = client.start(Some(cancellation_token.clone())).await;

        // Give the client some time to process the error
        tokio::time::sleep(Duration::from_millis(100)).await;
        let is_running = client.is_running();

        assert!(!is_running);
        assert!(cancellation_token.is_cancelled());
    }

    #[tokio::test]
    async fn ipc_client_keeps_running_if_crypto_returns_recoverable_receive_error() {
        let crypto_provider = TestCryptoProvider {
            send_result: None,
            // A recoverable receive error must not stop the processing loop.
            receive_result: Some(Err(TestCryptoError {
                message: "Crypto error".to_string(),
                fatal: false,
            })),
        };
        let communication_provider = TestCommunicationBackend::new();
        let session_map = TestSessionRepository::new(HashMap::new());
        let client = IpcClientImpl::new(crypto_provider, communication_provider, session_map);
        let cancellation_token = CancellationToken::new();
        let _ = client.start(Some(cancellation_token.clone())).await;

        // Give the client time to hit the recoverable receive error (repeatedly).
        tokio::time::sleep(Duration::from_millis(100)).await;
        let is_running = client.is_running();

        assert!(is_running);
        assert!(!cancellation_token.is_cancelled());
    }

    #[tokio::test]
    async fn ipc_client_is_not_running_if_cancellation_token_is_cancelled() {
        let crypto_provider = TestCryptoProvider {
            send_result: None,
            receive_result: None,
        };
        let communication_provider = TestCommunicationBackend::new();
        let session_map = TestSessionRepository::new(HashMap::new());
        let client = IpcClientImpl::new(crypto_provider, communication_provider, session_map);
        let cancellation_token = CancellationToken::new();
        let _ = client.start(Some(cancellation_token.clone())).await;

        // Give the client some time to process
        tokio::time::sleep(Duration::from_millis(100)).await;

        // Cancel the token and give the client some time to process the cancellation
        cancellation_token.cancel();
        tokio::time::sleep(Duration::from_millis(100)).await;
        let is_running = client.is_running();

        assert!(!is_running);
    }

    #[tokio::test]
    async fn ipc_client_is_running_if_no_errors_are_encountered() {
        let crypto_provider = TestCryptoProvider {
            send_result: None,
            receive_result: None,
        };
        let communication_provider = TestCommunicationBackend::new();
        let session_map = TestSessionRepository::new(HashMap::new());
        let client = IpcClientImpl::new(crypto_provider, communication_provider, session_map);
        let cancellation_token = CancellationToken::new();
        let _ = client.start(Some(cancellation_token.clone())).await;

        // Give the client some time to process
        tokio::time::sleep(Duration::from_millis(100)).await;
        let is_running = client.is_running();

        assert!(is_running);
        assert!(!cancellation_token.is_cancelled());
    }

    #[tokio::test]
    async fn ipc_client_is_not_running_if_not_started() {
        let crypto_provider = TestCryptoProvider {
            send_result: None,
            receive_result: None,
        };
        let communication_provider = TestCommunicationBackend::new();
        let session_map = TestSessionRepository::new(HashMap::new());
        let client = IpcClientImpl::new(crypto_provider, communication_provider, session_map);

        // Give the client some time to process
        tokio::time::sleep(Duration::from_millis(100)).await;
        let is_running = client.is_running();

        assert!(!is_running);
    }

    #[tokio::test]
    async fn ipc_client_start_returns_error_if_already_running() {
        let crypto_provider = TestCryptoProvider {
            send_result: None,
            receive_result: None,
        };
        let communication_provider = TestCommunicationBackend::new();
        let session_map = TestSessionRepository::new(HashMap::new());
        let client = IpcClientImpl::new(crypto_provider, communication_provider, session_map);
        let cancellation_token = CancellationToken::new();
        let first_result = client.start(Some(cancellation_token.clone())).await;
        assert_eq!(first_result, Ok(()));

        // Give the client some time to process
        tokio::time::sleep(Duration::from_millis(100)).await;
        assert!(client.is_running());

        let second_result = client.start(Some(cancellation_token.clone())).await;
        assert_eq!(second_result, Err(AlreadyRunningError));
    }

    mod request {
        use super::*;
        use crate::{RpcHandler, RpcRequestInfo};

        #[derive(Debug, Clone, PartialEq, Deserialize, Serialize)]
        struct TestRequest {
            a: i32,
            b: i32,
        }

        #[derive(Debug, Clone, PartialEq, Deserialize, Serialize)]
        struct TestResponse {
            result: i32,
        }

        impl RpcRequest for TestRequest {
            type Response = TestResponse;

            const NAME: &str = "TestRequest";
        }

        struct TestHandler;

        impl RpcHandler for TestHandler {
            type Request = TestRequest;

            async fn handle(&self, request: Self::Request, _info: RpcRequestInfo) -> TestResponse {
                TestResponse {
                    result: request.a + request.b,
                }
            }
        }

        /// A request whose handler responds with the source it was given, so tests can observe
        /// what the IPC client passes to handlers.
        #[derive(Debug, Clone, PartialEq, Deserialize, Serialize)]
        struct EchoSourceRequest;

        impl RpcRequest for EchoSourceRequest {
            type Response = Source;

            const NAME: &str = "EchoSourceRequest";
        }

        struct EchoSourceHandler;

        impl RpcHandler for EchoSourceHandler {
            type Request = EchoSourceRequest;

            async fn handle(&self, _request: Self::Request, info: RpcRequestInfo) -> Source {
                info.source
            }
        }

        /// A second request type whose response is a unit enum, which serde serializes to a bare
        /// JSON string. Deserializing it as [`TestResponse`] fails, which is what makes it a
        /// faithful stand-in for the real `GetBiometricsStatus` / `UnlockBiometrics` pair.
        #[derive(Debug, Clone, PartialEq, Deserialize, Serialize)]
        struct OtherRequest;

        #[derive(Debug, Clone, PartialEq, Deserialize, Serialize)]
        enum OtherResponse {
            Available,
        }

        impl RpcRequest for OtherRequest {
            type Response = OtherResponse;

            const NAME: &str = "OtherRequest";
        }

        #[tokio::test]
        async fn request_sends_message_and_returns_response() {
            let crypto_provider = NoEncryptionCryptoProvider;
            let communication_provider = TestCommunicationBackend::new();
            let session_map = InMemorySessionRepository::default();
            let client =
                IpcClientImpl::new(crypto_provider, communication_provider.clone(), session_map);
            let _ = client.start(None).await;
            let request = TestRequest { a: 1, b: 2 };
            let response = TestResponse { result: 3 };

            // Send the request
            let request_clone = request.clone();
            let client_clone = client.clone();
            let result_handle = tokio::spawn(async move {
                client_clone
                    .request::<TestRequest>(
                        request_clone,
                        Endpoint::BrowserBackground { id: HostId::Own },
                        None,
                    )
                    .await
            });
            tokio::time::sleep(Duration::from_millis(100)).await;

            // Read and verify the outgoing message
            let outgoing_messages = communication_provider.outgoing().await;
            let outgoing_request: RpcRequestMessage<TestRequest> =
                serde_utils::from_slice(&outgoing_messages[0].payload)
                    .expect("Deserialization should not fail");
            assert_eq!(outgoing_request.request_type, "TestRequest");
            assert_eq!(outgoing_request.request, request);

            // Simulate receiving a response
            let simulated_response = IncomingRpcResponseMessage {
                result: Ok(response),
                request_id: outgoing_request.request_id.clone(),
                request_type: outgoing_request.request_type.clone(),
            };
            let simulated_response = IncomingMessage {
                payload: serde_utils::to_vec(&simulated_response)
                    .expect("Serialization should not fail"),
                source: Source::BrowserBackground { id: HostId::Own },
                destination: Endpoint::Web {
                    tab_id: 9001,
                    document_id: "doc-1".to_string(),
                },
                // The requester subscribes to the dedicated topic it minted, so the response must
                // be published there.
                topic: Some(outgoing_request.response_topic.clone()),
            };
            communication_provider.push_incoming(simulated_response);

            // Wait for the response
            let result = result_handle.await.unwrap();
            assert_eq!(result.unwrap().result, 3);
        }

        /// Two requests in flight at once must each receive their own response. Each request mints
        /// its own response topic, so `TestRequest` and `OtherRequest` never see each other's
        /// responses even though their response types differ (`OtherResponse::Available` is a bare
        /// JSON string that would not deserialize into `TestResponse`).
        #[tokio::test]
        async fn concurrent_request_of_a_different_type_does_not_disturb_a_pending_request() {
            let crypto_provider = NoEncryptionCryptoProvider;
            let communication_provider = TestCommunicationBackend::new();
            let session_map = InMemorySessionRepository::default();
            let client =
                IpcClientImpl::new(crypto_provider, communication_provider.clone(), session_map);
            let _ = client.start(None).await;

            // Put both requests in flight at the same time.
            let test_handle = {
                let client = client.clone();
                tokio::spawn(async move {
                    client
                        .request::<TestRequest>(
                            TestRequest { a: 1, b: 2 },
                            Endpoint::BrowserBackground { id: HostId::Own },
                            None,
                        )
                        .await
                })
            };
            let other_handle = {
                let client = client.clone();
                tokio::spawn(async move {
                    client
                        .request::<OtherRequest>(
                            OtherRequest,
                            Endpoint::BrowserBackground { id: HostId::Own },
                            None,
                        )
                        .await
                })
            };
            tokio::time::sleep(Duration::from_millis(100)).await;

            // Recover both requests from the wire, keyed by type, so we can answer each on the
            // dedicated response topic it minted.
            let outgoing = communication_provider.outgoing().await;
            let requests: HashMap<String, RpcRequestMessage<serde_json::Value>> = outgoing
                .iter()
                .map(|message| {
                    let partial: RpcRequestMessage<serde_json::Value> =
                        serde_utils::from_slice(&message.payload)
                            .expect("Deserialization should not fail");
                    (partial.request_type.clone(), partial)
                })
                .collect();

            let respond = |topic: String, payload: Vec<u8>| {
                communication_provider.push_incoming(IncomingMessage {
                    payload,
                    source: Source::BrowserBackground { id: HostId::Own },
                    destination: Endpoint::Web {
                        tab_id: 9001,
                        document_id: "doc-1".to_string(),
                    },
                    topic: Some(topic),
                });
            };

            // Answer `OtherRequest` first, while `TestRequest` is still waiting. Each response goes
            // to its own dedicated topic, so the two can never be confused for one another.
            respond(
                requests["OtherRequest"].response_topic.clone(),
                serde_utils::to_vec(&IncomingRpcResponseMessage {
                    result: Ok(OtherResponse::Available),
                    request_id: requests["OtherRequest"].request_id.clone(),
                    request_type: "OtherRequest".to_string(),
                })
                .expect("Serialization should not fail"),
            );

            respond(
                requests["TestRequest"].response_topic.clone(),
                serde_utils::to_vec(&IncomingRpcResponseMessage {
                    result: Ok(TestResponse { result: 3 }),
                    request_id: requests["TestRequest"].request_id.clone(),
                    request_type: "TestRequest".to_string(),
                })
                .expect("Serialization should not fail"),
            );

            assert_eq!(
                other_handle
                    .await
                    .unwrap()
                    .expect("OtherRequest should succeed"),
                OtherResponse::Available
            );
            assert_eq!(
                test_handle
                    .await
                    .unwrap()
                    .expect("TestRequest should succeed"),
                TestResponse { result: 3 }
            );
        }

        /// A response that reaches this request's dedicated topic but fails to deserialize into the
        /// expected response type is a genuine error for *this* request: because the topic is
        /// dedicated, there is no other request it could belong to, so the failure is surfaced
        /// immediately.
        #[tokio::test]
        async fn malformed_response_surfaces_error_instead_of_hanging() {
            let crypto_provider = NoEncryptionCryptoProvider;
            let communication_provider = TestCommunicationBackend::new();
            let session_map = InMemorySessionRepository::default();
            let client =
                IpcClientImpl::new(crypto_provider, communication_provider.clone(), session_map);
            let _ = client.start(None).await;

            let result_handle = {
                let client = client.clone();
                tokio::spawn(async move {
                    client
                        .request::<TestRequest>(
                            TestRequest { a: 1, b: 2 },
                            Endpoint::BrowserBackground { id: HostId::Own },
                            None,
                        )
                        .await
                })
            };
            tokio::time::sleep(Duration::from_millis(100)).await;

            // Recover the dedicated response topic and answer it with a body that cannot be
            // deserialized into `TestResponse` (`OtherResponse::Available` serializes to a bare
            // string).
            let outgoing = communication_provider.outgoing().await;
            let request: RpcRequestMessage<TestRequest> =
                serde_utils::from_slice(&outgoing[0].payload)
                    .expect("Deserialization should not fail");
            let malformed = IncomingRpcResponseMessage {
                result: Ok(OtherResponse::Available),
                request_id: request.request_id.clone(),
                request_type: request.request_type.clone(),
            };
            communication_provider.push_incoming(IncomingMessage {
                payload: serde_utils::to_vec(&malformed).expect("Serialization should not fail"),
                source: Source::BrowserBackground { id: HostId::Own },
                destination: Endpoint::Web {
                    tab_id: 9001,
                    document_id: "doc-1".to_string(),
                },
                topic: Some(request.response_topic.clone()),
            });

            let result = tokio::time::timeout(Duration::from_secs(5), result_handle)
                .await
                .expect("request must not hang on a malformed response")
                .unwrap();
            assert!(matches!(
                result,
                Err(crate::error::RequestError::Rpc(
                    crate::rpc::error::RpcError::ResponseDeserialization(_)
                ))
            ));
        }

        /// Builds a `TestRequest` response on the request's dedicated topic, as sent by `source`.
        fn test_response_from(
            request: &RpcRequestMessage<TestRequest>,
            result: i32,
            source: Source,
        ) -> IncomingMessage {
            let response = IncomingRpcResponseMessage {
                result: Ok(TestResponse { result }),
                request_id: request.request_id.clone(),
                request_type: request.request_type.clone(),
            };
            IncomingMessage {
                payload: serde_utils::to_vec(&response).expect("Serialization should not fail"),
                source,
                destination: Endpoint::DesktopRenderer,
                topic: Some(request.response_topic.clone()),
            }
        }

        /// A message on the response topic from a peer other than the destination must not be
        /// accepted as the response, and must not stop the request from receiving the real one.
        #[tokio::test]
        async fn response_from_other_source_is_ignored_and_destination_response_is_returned() {
            let communication_provider = TestCommunicationBackend::new();
            let client = IpcClientImpl::new(
                NoEncryptionCryptoProvider,
                communication_provider.clone(),
                InMemorySessionRepository::default(),
            );
            let _ = client.start(None).await;

            let result_handle = {
                let client = client.clone();
                tokio::spawn(async move {
                    client
                        .request::<TestRequest>(
                            TestRequest { a: 1, b: 2 },
                            Endpoint::DesktopMain,
                            None,
                        )
                        .await
                })
            };
            tokio::time::sleep(Duration::from_millis(100)).await;

            let outgoing = communication_provider.outgoing().await;
            let request: RpcRequestMessage<TestRequest> =
                serde_utils::from_slice(&outgoing[0].payload)
                    .expect("Deserialization should not fail");

            communication_provider.push_incoming(test_response_from(
                &request,
                999,
                Source::BrowserBackground { id: HostId::Id(42) },
            ));
            communication_provider.push_incoming(test_response_from(
                &request,
                3,
                Source::DesktopMain,
            ));

            let result = tokio::time::timeout(Duration::from_secs(5), result_handle)
                .await
                .expect("request must complete once the destination responds")
                .unwrap();
            assert_eq!(result, Ok(TestResponse { result: 3 }));
        }

        /// When only a peer other than the destination answers, the request keeps waiting and ends
        /// through its cancellation token, never returning the other peer's result.
        #[tokio::test]
        async fn response_only_from_other_source_waits_until_cancelled() {
            let communication_provider = TestCommunicationBackend::new();
            let client = IpcClientImpl::new(
                NoEncryptionCryptoProvider,
                communication_provider.clone(),
                InMemorySessionRepository::default(),
            );
            let _ = client.start(None).await;

            let cancellation_token = CancellationToken::new();
            let result_handle = {
                let client = client.clone();
                let cancellation_token = cancellation_token.clone();
                tokio::spawn(async move {
                    client
                        .request::<TestRequest>(
                            TestRequest { a: 1, b: 2 },
                            Endpoint::DesktopMain,
                            Some(cancellation_token),
                        )
                        .await
                })
            };
            tokio::time::sleep(Duration::from_millis(100)).await;

            let outgoing = communication_provider.outgoing().await;
            let request: RpcRequestMessage<TestRequest> =
                serde_utils::from_slice(&outgoing[0].payload)
                    .expect("Deserialization should not fail");

            communication_provider.push_incoming(test_response_from(
                &request,
                999,
                Source::DesktopRenderer,
            ));
            tokio::time::sleep(Duration::from_millis(100)).await;
            assert!(
                !result_handle.is_finished(),
                "request must not complete on a response from another source"
            );

            cancellation_token.cancel();
            let result = tokio::time::timeout(Duration::from_secs(5), result_handle)
                .await
                .expect("request must end once cancelled")
                .unwrap();
            assert!(matches!(
                result,
                Err(crate::error::RequestError::Receive(
                    TypedReceiveError::Cancelled
                ))
            ));
        }

        /// A web source carries an `origin` that its endpoint does not, so the response is matched
        /// against the destination by endpoint identity (tab and document) alone.
        #[tokio::test]
        async fn response_from_web_destination_is_accepted_regardless_of_origin() {
            let communication_provider = TestCommunicationBackend::new();
            let client = IpcClientImpl::new(
                NoEncryptionCryptoProvider,
                communication_provider.clone(),
                InMemorySessionRepository::default(),
            );
            let _ = client.start(None).await;

            let result_handle = {
                let client = client.clone();
                tokio::spawn(async move {
                    client
                        .request::<TestRequest>(
                            TestRequest { a: 1, b: 2 },
                            Endpoint::Web {
                                tab_id: 9001,
                                document_id: "doc-1".to_string(),
                            },
                            None,
                        )
                        .await
                })
            };
            tokio::time::sleep(Duration::from_millis(100)).await;

            let outgoing = communication_provider.outgoing().await;
            let request: RpcRequestMessage<TestRequest> =
                serde_utils::from_slice(&outgoing[0].payload)
                    .expect("Deserialization should not fail");

            // Same tab, different document: a page the request was not sent to.
            communication_provider.push_incoming(test_response_from(
                &request,
                999,
                Source::Web {
                    tab_id: 9001,
                    document_id: "doc-2".to_string(),
                    origin: "https://vault.bitwarden.com".to_string(),
                },
            ));
            communication_provider.push_incoming(test_response_from(
                &request,
                3,
                Source::Web {
                    tab_id: 9001,
                    document_id: "doc-1".to_string(),
                    origin: "https://vault.bitwarden.com".to_string(),
                },
            ));

            let result = tokio::time::timeout(Duration::from_secs(5), result_handle)
                .await
                .expect("request must complete once the destination responds")
                .unwrap();
            assert_eq!(result, Ok(TestResponse { result: 3 }));
        }

        #[tokio::test]
        async fn incoming_rpc_message_handles_request_and_returns_response() {
            let crypto_provider = NoEncryptionCryptoProvider;
            let communication_provider = TestCommunicationBackend::new();
            let session_map = InMemorySessionRepository::default();
            let client =
                IpcClientImpl::new(crypto_provider, communication_provider.clone(), session_map);
            let _ = client.start(None).await;
            let request_id = uuid::Uuid::new_v4().to_string();
            let request = TestRequest { a: 1, b: 2 };
            let response = TestResponse { result: 3 };

            // Register the handler
            client.register_rpc_handler(TestHandler).await;

            // Simulate receiving a request
            let response_topic = format!("RpcResponseMessage:{request_id}");
            let simulated_request = RpcRequestMessage {
                request,
                request_id: request_id.clone(),
                request_type: "TestRequest".to_string(),
                response_topic: response_topic.clone(),
            };
            let simulated_request_message = IncomingMessage {
                payload: serde_utils::to_vec(&simulated_request)
                    .expect("Serialization should not fail"),
                source: Source::Web {
                    tab_id: 9001,
                    document_id: "doc-1".to_string(),
                    origin: "https://example.com".to_string(),
                },
                destination: Endpoint::BrowserBackground { id: HostId::Own },
                topic: Some(RPC_REQUEST_PAYLOAD_TYPE_NAME.to_owned()),
            };
            communication_provider.push_incoming(simulated_request_message);

            // Give the client some time to process the request
            tokio::time::sleep(Duration::from_millis(100)).await;

            // Read and verify the outgoing message
            let outgoing_messages = communication_provider.outgoing().await;
            let outgoing_response: IncomingRpcResponseMessage<TestResponse> =
                serde_utils::from_slice(&outgoing_messages[0].payload)
                    .expect("Deserialization should not fail");

            // The response must be published on the dedicated topic carried by the request.
            assert_eq!(outgoing_messages[0].topic, Some(response_topic));
            assert_eq!(outgoing_response.request_type, "TestRequest");
            assert_eq!(outgoing_response.result, Ok(response));
        }

        #[tokio::test]
        async fn ipc_client_stops_if_rpc_response_send_returns_fatal_error() {
            let request = RpcRequestMessage {
                request: TestRequest { a: 1, b: 2 },
                request_id: "request".to_string(),
                request_type: "TestRequest".to_string(),
                response_topic: "RpcResponseMessage:request".to_string(),
            };
            let crypto_provider = TestCryptoProvider {
                send_result: Some(Err(TestCryptoError {
                    message: "Crypto error".to_string(),
                    fatal: true,
                })),
                receive_result: Some(Ok(IncomingMessage {
                    payload: serde_utils::to_vec(&request).expect("Serialization should not fail"),
                    source: Source::Web {
                        tab_id: 9001,
                        document_id: "doc-1".to_string(),
                        origin: "https://example.com".to_string(),
                    },
                    destination: Endpoint::BrowserBackground { id: HostId::Own },
                    topic: Some(RPC_REQUEST_PAYLOAD_TYPE_NAME.to_owned()),
                })),
            };
            let client = IpcClientImpl::new(
                crypto_provider,
                TestCommunicationBackend::new(),
                TestSessionRepository::new(HashMap::new()),
            );
            client.register_rpc_handler(TestHandler).await;
            let _ = client.start(None).await;

            // Give the client some time to receive the request and fail to send the response
            tokio::time::sleep(Duration::from_millis(100)).await;

            assert!(!client.is_running());
        }

        #[tokio::test]
        async fn incoming_rpc_message_passes_message_source_to_handler() {
            let crypto_provider = NoEncryptionCryptoProvider;
            let communication_provider = TestCommunicationBackend::new();
            let session_map = InMemorySessionRepository::default();
            let client =
                IpcClientImpl::new(crypto_provider, communication_provider.clone(), session_map);
            let _ = client.start(None).await;
            client.register_rpc_handler(EchoSourceHandler).await;

            let source = Source::Web {
                tab_id: 9001,
                document_id: "doc-1".to_string(),
                origin: "https://example.com".to_string(),
            };
            let request_id = uuid::Uuid::new_v4().to_string();
            let simulated_request = RpcRequestMessage {
                request: EchoSourceRequest,
                request_id: request_id.clone(),
                request_type: EchoSourceRequest::NAME.to_string(),
                response_topic: format!("RpcResponseMessage:{request_id}"),
            };
            communication_provider.push_incoming(IncomingMessage {
                payload: serde_utils::to_vec(&simulated_request)
                    .expect("Serialization should not fail"),
                source: source.clone(),
                destination: Endpoint::BrowserBackground { id: HostId::Own },
                topic: Some(RPC_REQUEST_PAYLOAD_TYPE_NAME.to_owned()),
            });

            // Give the client some time to process the request
            tokio::time::sleep(Duration::from_millis(100)).await;

            let outgoing_messages = communication_provider.outgoing().await;
            let outgoing_response: IncomingRpcResponseMessage<Source> =
                serde_utils::from_slice(&outgoing_messages[0].payload)
                    .expect("Deserialization should not fail");

            assert_eq!(outgoing_response.result, Ok(source));
        }
    }

    mod fragmentation {
        use super::*;
        use crate::{
            RequestError, RpcHandler, RpcRequestInfo,
            crypto_provider::noise::crypto_provider::NoiseCryptoProvider,
            fragmentation::{FRAGMENT_TOPIC, MAX_PAYLOAD_SIZE, test_support::*},
            traits::TestTwoWayCommunicationBackend,
        };

        /// Larger than several Noise messages, and not a multiple of the fragment size.
        const LARGE_LEN: usize = 4 * 65535 + 123;

        fn noise_client(
            backend: TestTwoWayCommunicationBackend,
        ) -> IpcClientImpl<
            NoiseCryptoProvider,
            TestTwoWayCommunicationBackend,
            InMemorySessionRepository<crate::NoiseCryptoProviderState>,
        > {
            IpcClientImpl::new(
                NoiseCryptoProvider::new(),
                backend,
                InMemorySessionRepository::new(HashMap::new()),
            )
        }

        #[tokio::test]
        async fn large_payload_round_trips_through_noise_once_on_its_topic() {
            let (backend_1, backend_2) = TestTwoWayCommunicationBackend::new();
            let client_1 = noise_client(backend_1);
            let client_2 = noise_client(backend_2);
            client_1.start(None).await.unwrap();
            client_2.start(None).await.unwrap();
            let mut all_messages = client_2.subscribe(None).await.unwrap();
            let mut on_topic = client_2.subscribe(Some("bulk".to_owned())).await.unwrap();
            let mut on_fragment_topic = client_2
                .subscribe(Some(format!("{FRAGMENT_TOPIC}/bulk")))
                .await
                .unwrap();

            let payload = large_payload(LARGE_LEN, 0);
            let message = OutgoingMessage {
                payload: payload.clone(),
                destination: Endpoint::DesktopMain,
                topic: Some("bulk".to_owned()),
            };
            client_1.send(message).await.unwrap();

            let received = all_messages.receive(None).await.unwrap();
            assert_eq!(received.topic.as_deref(), Some("bulk"));
            assert_eq!(received.payload, payload);

            // Neither the fragments nor a second copy reach subscribers.
            let next = bitwarden_threading::time::timeout(
                Duration::from_millis(200),
                all_messages.receive(None),
            )
            .await;
            assert!(next.is_err(), "unexpected extra message: {next:?}");

            // Topic filters see the reassembled message, never its fragments.
            assert_eq!(on_topic.receive(None).await.unwrap().payload, payload);
            let fragment = bitwarden_threading::time::timeout(
                Duration::from_millis(10),
                on_fragment_topic.receive(None),
            )
            .await;
            assert!(
                fragment.is_err(),
                "fragment reached a subscriber: {fragment:?}"
            );
        }

        #[derive(Debug, Clone, Serialize, Deserialize)]
        struct EchoRequest {
            data: String,
        }

        impl RpcRequest for EchoRequest {
            type Response = String;

            const NAME: &str = "EchoRequest";
        }

        struct EchoHandler;

        impl RpcHandler for EchoHandler {
            type Request = EchoRequest;

            async fn handle(&self, request: EchoRequest, _info: RpcRequestInfo) -> String {
                request.data.repeat(2)
            }
        }

        #[tokio::test]
        async fn rpc_with_response_larger_than_one_noise_message_succeeds() {
            let (backend_1, backend_2) = TestTwoWayCommunicationBackend::new();
            let client_1 = noise_client(backend_1);
            let client_2 = noise_client(backend_2);
            client_1.start(None).await.unwrap();
            client_2.start(None).await.unwrap();
            client_2.register_rpc_handler(EchoHandler).await;

            // The request itself also exceeds one Noise message.
            let data = "a".repeat(LARGE_LEN / 2);
            let response = client_1
                .request(
                    EchoRequest { data: data.clone() },
                    Endpoint::DesktopMain,
                    None,
                )
                .await
                .unwrap();

            assert_eq!(response, data.repeat(2));
        }

        #[tokio::test]
        async fn rpc_response_over_the_maximum_returns_an_error_to_the_requester() {
            let (backend_1, backend_2) = TestTwoWayCommunicationBackend::new();
            let client_1 = noise_client(backend_1);
            let client_2 = noise_client(backend_2);
            client_1.start(None).await.unwrap();
            client_2.start(None).await.unwrap();
            client_2.register_rpc_handler(EchoHandler).await;

            // The request fits, but the doubled response does not.
            let data = "a".repeat(MAX_PAYLOAD_SIZE / 2 + 1);
            let result = bitwarden_threading::time::timeout(
                Duration::from_secs(10),
                client_1.request(EchoRequest { data }, Endpoint::DesktopMain, None),
            )
            .await
            .expect("Requester should receive an error response");

            assert!(
                matches!(
                    result,
                    Err(RequestError::Rpc(RpcError::ResponseSerialization(_)))
                ),
                "{result:?}"
            );
        }

        #[tokio::test]
        async fn client_keeps_receiving_after_invalid_fragments() {
            let backend = TestCommunicationBackend::new();
            let client = IpcClientImpl::new(
                NoEncryptionCryptoProvider,
                backend.clone(),
                InMemorySessionRepository::default(),
            );
            client.start(None).await.unwrap();
            let mut subscription = client.subscribe(None).await.unwrap();
            let source = web_source(1);

            backend.push_incoming(incoming(
                outgoing(vec![1, 2, 3], FRAGMENT_TOPIC),
                source.clone(),
            ));
            backend.push_incoming(crafted(source.clone(), 1, 5, 1024));
            backend.push_incoming(crafted(source.clone(), 2, 0, MAX_PAYLOAD_SIZE + 1));
            backend.push_incoming(crafted(source.clone(), 3, 0, 1024));
            backend.push_incoming(crafted(source.clone(), 3, 0, 1024));

            let payload = large_payload(FRAGMENT_LEN * 2 + 1, 0);
            for fragment in split_all(outgoing(payload.clone(), "bulk"), MAX_LEN) {
                backend.push_incoming(incoming(fragment, source.clone()));
            }

            let received = subscription.receive(None).await.unwrap();
            assert_eq!(received.topic.as_deref(), Some("bulk"));
            assert_eq!(received.payload, payload);
            assert!(client.is_running());
        }

        #[tokio::test]
        async fn send_rejects_reserved_fragment_topic() {
            let client = IpcClientImpl::new(
                NoEncryptionCryptoProvider,
                TestCommunicationBackend::new(),
                InMemorySessionRepository::default(),
            );
            let result = client.send(outgoing(vec![0; 16], FRAGMENT_TOPIC)).await;
            assert!(matches!(result, Err(SendError::Other(_))));
        }

        /// Records sent messages of up to [`MAX_LEN`] bytes and fails the send at index `fail_at`.
        struct FailingCryptoProvider {
            fail_at: usize,
            fatal: bool,
            sent: Arc<Mutex<Vec<OutgoingMessage>>>,
        }

        impl CryptoProvider<TestCommunicationBackend, TestSessionRepository> for FailingCryptoProvider {
            type Session = String;
            type SendError = TestCryptoError;
            type ReceiveError = TestCryptoError;

            fn max_payload_len(&self) -> usize {
                MAX_LEN
            }

            async fn send(
                &self,
                _communication: &TestCommunicationBackend,
                _sessions: &TestSessionRepository,
                message: OutgoingMessage,
            ) -> Result<(), TestCryptoError> {
                let mut sent = self.sent.lock().unwrap();
                if sent.len() == self.fail_at {
                    return Err(TestCryptoError {
                        message: "Send failed".to_owned(),
                        fatal: self.fatal,
                    });
                }
                sent.push(message);
                Ok(())
            }

            async fn receive(
                &self,
                _receiver: &<TestCommunicationBackend as CommunicationBackend>::Receiver,
                _communication: &TestCommunicationBackend,
                _sessions: &TestSessionRepository,
            ) -> Result<IncomingMessage, TestCryptoError> {
                std::future::pending().await
            }
        }

        #[tokio::test]
        async fn failed_fragment_send_stops_the_message_and_fatal_errors_stop_the_client() {
            for fatal in [false, true] {
                let sent = Arc::new(Mutex::new(Vec::new()));
                let client = IpcClientImpl::new(
                    FailingCryptoProvider {
                        fail_at: 1,
                        fatal,
                        sent: sent.clone(),
                    },
                    TestCommunicationBackend::new(),
                    TestSessionRepository::default(),
                );
                client.start(None).await.unwrap();

                let message = outgoing(large_payload(FRAGMENT_LEN * 3, 0), "bulk");
                assert!(client.send(message).await.is_err());

                // Fragments after the failed one are not sent.
                assert_eq!(sent.lock().unwrap().len(), 1);
                assert_eq!(client.is_running(), !fatal);
            }
        }
    }
}