diff --git a/crates/qconnect-transport-ws/Cargo.toml b/crates/qconnect-transport-ws/Cargo.toml
new file mode 100644
index 000000000..0cafbe414
--- /dev/null
+++ b/crates/qconnect-transport-ws/Cargo.toml
@@ -0,0 +1,16 @@
+[package]
+name = "qconnect-transport-ws"
+version.workspace = true
+edition.workspace = true
+license.workspace = true
+description = "WebSocket transport abstractions for Qobuz Connect"
+
+[dependencies]
+qconnect-protocol = { path = "../qconnect-protocol" }
+async-trait = { workspace = true }
+tokio = { workspace = true }
+serde = { workspace = true }
+thiserror = { workspace = true }
+futures-util = { workspace = true }
+tokio-tungstenite = { version = "0.24", features = ["rustls-tls-webpki-roots"] }
+prost = "0.13"
diff --git a/crates/qconnect-transport-ws/src/config.rs b/crates/qconnect-transport-ws/src/config.rs
new file mode 100644
index 000000000..80ccf25c4
--- /dev/null
+++ b/crates/qconnect-transport-ws/src/config.rs
@@ -0,0 +1,28 @@
+#[derive(Debug, Clone)]
+pub struct WsTransportConfig {
+    pub endpoint_url: String,
+    pub jwt_qws: Option<String>,
+    pub reconnect_backoff_ms: u64,
+    pub reconnect_backoff_max_ms: u64,
+    pub connect_timeout_ms: u64,
+    pub keepalive_interval_ms: u64,
+    pub auto_subscribe: bool,
+    pub subscribe_channels: Vec<Vec<u8>>,
+    pub qcloud_proto: u32,
+}
+
+impl Default for WsTransportConfig {
+    fn default() -> Self {
+        Self {
+            endpoint_url: String::new(),
+            jwt_qws: None,
+            reconnect_backoff_ms: 2_000,
+            reconnect_backoff_max_ms: 30_000,
+            connect_timeout_ms: 10_000,
+            keepalive_interval_ms: 30_000,
+            auto_subscribe: true,
+            subscribe_channels: Vec::new(),
+            qcloud_proto: 1,
+        }
+    }
+}
diff --git a/crates/qconnect-transport-ws/src/error.rs b/crates/qconnect-transport-ws/src/error.rs
new file mode 100644
index 000000000..3cf035315
--- /dev/null
+++ b/crates/qconnect-transport-ws/src/error.rs
@@ -0,0 +1,23 @@
+use thiserror::Error;
+
+#[derive(Debug, Error)]
+pub enum WsTransportError {
+    #[error("transport is already connected")]
+    AlreadyConnected,
+    #[error("transport is already running")]
+    AlreadyRunning,
+    #[error("transport is not connected")]
+    NotConnected,
+    #[error("transport event channel is closed")]
+    EventChannelClosed,
+    #[error("transport task channel is closed")]
+    TransportChannelClosed,
+    #[error("transport task join error: {0}")]
+    Join(String),
+    #[error("websocket protocol error: {0}")]
+    Protocol(String),
+    #[error("serialization error: {0}")]
+    Serialization(String),
+    #[error("internal transport error: {0}")]
+    Internal(String),
+}
diff --git a/crates/qconnect-transport-ws/src/lib.rs b/crates/qconnect-transport-ws/src/lib.rs
new file mode 100644
index 000000000..4faa33131
--- /dev/null
+++ b/crates/qconnect-transport-ws/src/lib.rs
@@ -0,0 +1,11 @@
+//! qconnect-transport-ws
+//!
+//! WS transport trait + in-memory transport used during POC development.
+
+mod config;
+mod error;
+mod transport;
+
+pub use config::WsTransportConfig;
+pub use error::WsTransportError;
+pub use transport::{InMemoryWsTransport, NativeWsTransport, TransportEvent, WsTransport};
diff --git a/crates/qconnect-transport-ws/src/transport.rs b/crates/qconnect-transport-ws/src/transport.rs
new file mode 100644
index 000000000..d2d7b48ed
--- /dev/null
+++ b/crates/qconnect-transport-ws/src/transport.rs
@@ -0,0 +1,822 @@
+use std::{
+    sync::Arc,
+    time::{Duration, SystemTime, UNIX_EPOCH},
+};
+
+use async_trait::async_trait;
+use futures_util::{SinkExt, StreamExt};
+use prost::{
+    encoding::{decode_varint, encode_varint},
+    Message,
+};
+use qconnect_protocol::{
+    decode_inbound_json, decode_queue_server_events, encode_outbound_payload_bytes,
+    InboundEnvelope, OutboundEnvelope, QueueServerEvent,
+};
+use serde::{Deserialize, Serialize};
+use tokio::{
+    sync::{broadcast, mpsc, watch, Mutex},
+    task::JoinHandle,
+};
+use tokio_tungstenite::{
+    connect_async_with_config,
+    tungstenite::{protocol::WebSocketConfig, Message as WsMessage},
+};
+
+use crate::{WsTransportConfig, WsTransportError};
+
+const MSG_TYPE_AUTHENTICATE: u8 = 1;
+const MSG_TYPE_SUBSCRIBE: u8 = 2;
+const MSG_TYPE_PAYLOAD: u8 = 6;
+const MSG_TYPE_ERROR: u8 = 9;
+const MSG_TYPE_DISCONNECT: u8 = 10;
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub enum TransportEvent {
+    Connected,
+    Disconnected,
+    Authenticated,
+    Subscribed,
+    ReconnectScheduled {
+        attempt: u32,
+        backoff_ms: u64,
+        reason: String,
+    },
+    KeepalivePingSent,
+    KeepalivePongReceived,
+    TransportError {
+        stage: String,
+        message: String,
+    },
+    InboundFrameDecoded {
+        cloud_message_type: u8,
+        payload_size: usize,
+    },
+    InboundPayloadBytes {
+        cloud_message_type: u8,
+        payload: Vec<u8>,
+    },
+    OutboundSent {
+        message_type: String,
+        action_uuid: String,
+    },
+    InboundQueueServerEvent(QueueServerEvent),
+    InboundReceived(InboundEnvelope),
+}
+
+#[async_trait]
+pub trait WsTransport: Send + Sync {
+    async fn connect(&self, config: WsTransportConfig) -> Result<(), WsTransportError>;
+    async fn disconnect(&self) -> Result<(), WsTransportError>;
+    async fn send(&self, envelope: OutboundEnvelope) -> Result<(), WsTransportError>;
+    fn subscribe(&self) -> broadcast::Receiver<TransportEvent>;
+}
+
+#[derive(Debug, Default)]
+struct InMemoryState {
+    connected: bool,
+    last_config: Option<WsTransportConfig>,
+    sent_messages: Vec<OutboundEnvelope>,
+}
+
+#[derive(Clone)]
+pub struct InMemoryWsTransport {
+    state: Arc<Mutex<InMemoryState>>,
+    events_tx: broadcast::Sender<TransportEvent>,
+}
+
+impl InMemoryWsTransport {
+    pub fn new() -> Self {
+        let (events_tx, _) = broadcast::channel(512);
+        Self {
+            state: Arc::new(Mutex::new(InMemoryState::default())),
+            events_tx,
+        }
+    }
+
+    pub async fn inject_inbound(&self, envelope: InboundEnvelope) -> Result<(), WsTransportError> {
+        self.events_tx
+            .send(TransportEvent::InboundReceived(envelope))
+            .map_err(|_| WsTransportError::EventChannelClosed)?;
+        Ok(())
+    }
+
+    pub async fn sent_messages(&self) -> Vec<OutboundEnvelope> {
+        self.state.lock().await.sent_messages.clone()
+    }
+
+    pub async fn is_connected(&self) -> bool {
+        self.state.lock().await.connected
+    }
+}
+
+impl Default for InMemoryWsTransport {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+#[async_trait]
+impl WsTransport for InMemoryWsTransport {
+    async fn connect(&self, config: WsTransportConfig) -> Result<(), WsTransportError> {
+        let mut state = self.state.lock().await;
+        if state.connected {
+            return Err(WsTransportError::AlreadyConnected);
+        }
+
+        state.connected = true;
+        state.last_config = Some(config);
+        drop(state);
+
+        self.events_tx
+            .send(TransportEvent::Connected)
+            .map_err(|_| WsTransportError::EventChannelClosed)?;
+        Ok(())
+    }
+
+    async fn disconnect(&self) -> Result<(), WsTransportError> {
+        let mut state = self.state.lock().await;
+        if !state.connected {
+            return Err(WsTransportError::NotConnected);
+        }
+
+        state.connected = false;
+        drop(state);
+
+        self.events_tx
+            .send(TransportEvent::Disconnected)
+            .map_err(|_| WsTransportError::EventChannelClosed)?;
+        Ok(())
+    }
+
+    async fn send(&self, envelope: OutboundEnvelope) -> Result<(), WsTransportError> {
+        let mut state = self.state.lock().await;
+        if !state.connected {
+            return Err(WsTransportError::NotConnected);
+        }
+
+        state.sent_messages.push(envelope.clone());
+        drop(state);
+
+        self.events_tx
+            .send(TransportEvent::OutboundSent {
+                message_type: envelope.message_type,
+                action_uuid: envelope.action_uuid,
+            })
+            .map_err(|_| WsTransportError::EventChannelClosed)?;
+        Ok(())
+    }
+
+    fn subscribe(&self) -> broadcast::Receiver<TransportEvent> {
+        self.events_tx.subscribe()
+    }
+}
+
+#[derive(Debug, Default)]
+struct NativeState {
+    connected: bool,
+    running: bool,
+    last_config: Option<WsTransportConfig>,
+}
+
+struct NativeRuntime {
+    outbound_tx: mpsc::Sender<OutboundEnvelope>,
+    shutdown_tx: watch::Sender<bool>,
+    handle: JoinHandle<()>,
+}
+
+#[derive(Clone)]
+pub struct NativeWsTransport {
+    state: Arc<Mutex<NativeState>>,
+    runtime: Arc<Mutex<Option<NativeRuntime>>>,
+    events_tx: broadcast::Sender<TransportEvent>,
+}
+
+impl NativeWsTransport {
+    pub fn new() -> Self {
+        let (events_tx, _) = broadcast::channel(2048);
+        Self {
+            state: Arc::new(Mutex::new(NativeState::default())),
+            runtime: Arc::new(Mutex::new(None)),
+            events_tx,
+        }
+    }
+
+    pub async fn is_connected(&self) -> bool {
+        self.state.lock().await.connected
+    }
+}
+
+impl Default for NativeWsTransport {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+#[async_trait]
+impl WsTransport for NativeWsTransport {
+    async fn connect(&self, config: WsTransportConfig) -> Result<(), WsTransportError> {
+        let mut runtime_guard = self.runtime.lock().await;
+        if runtime_guard.is_some() {
+            return Err(WsTransportError::AlreadyRunning);
+        }
+
+        {
+            let mut state = self.state.lock().await;
+            state.running = true;
+            state.last_config = Some(config.clone());
+        }
+
+        let (outbound_tx, outbound_rx) = mpsc::channel::<OutboundEnvelope>(512);
+        let (shutdown_tx, shutdown_rx) = watch::channel(false);
+
+        let events_tx = self.events_tx.clone();
+        let state = Arc::clone(&self.state);
+        let handle = tokio::spawn(async move {
+            run_native_transport_loop(config, outbound_rx, shutdown_rx, events_tx, state).await;
+        });
+
+        *runtime_guard = Some(NativeRuntime {
+            outbound_tx,
+            shutdown_tx,
+            handle,
+        });
+        Ok(())
+    }
+
+    async fn disconnect(&self) -> Result<(), WsTransportError> {
+        let runtime = self
+            .runtime
+            .lock()
+            .await
+            .take()
+            .ok_or(WsTransportError::NotConnected)?;
+
+        let _ = runtime.shutdown_tx.send(true);
+
+        runtime
+            .handle
+            .await
+            .map_err(|err| WsTransportError::Join(err.to_string()))?;
+
+        let mut state = self.state.lock().await;
+        state.connected = false;
+        state.running = false;
+        Ok(())
+    }
+
+    async fn send(&self, envelope: OutboundEnvelope) -> Result<(), WsTransportError> {
+        let outbound_tx = self
+            .runtime
+            .lock()
+            .await
+            .as_ref()
+            .map(|runtime| runtime.outbound_tx.clone())
+            .ok_or(WsTransportError::NotConnected)?;
+
+        outbound_tx
+            .send(envelope)
+            .await
+            .map_err(|_| WsTransportError::TransportChannelClosed)
+    }
+
+    fn subscribe(&self) -> broadcast::Receiver<TransportEvent> {
+        self.events_tx.subscribe()
+    }
+}
+
+async fn run_native_transport_loop(
+    config: WsTransportConfig,
+    mut outbound_rx: mpsc::Receiver<OutboundEnvelope>,
+    mut shutdown_rx: watch::Receiver<bool>,
+    events_tx: broadcast::Sender<TransportEvent>,
+    state: Arc<Mutex<NativeState>>,
+) {
+    let base_backoff = config.reconnect_backoff_ms.max(200);
+    let max_backoff = config.reconnect_backoff_max_ms.max(base_backoff);
+    let mut backoff = base_backoff;
+    let mut reconnect_attempt: u32 = 0;
+    let mut msg_id: u32 = 0;
+
+    loop {
+        if *shutdown_rx.borrow() {
+            break;
+        }
+
+        let connect_result = tokio::time::timeout(
+            Duration::from_millis(config.connect_timeout_ms.max(1000)),
+            connect_async_with_config(
+                &config.endpoint_url,
+                Some(WebSocketConfig::default()),
+                false,
+            ),
+        )
+        .await;
+
+        let (mut ws, _) = match connect_result {
+            Ok(Ok((ws, response))) => (ws, response),
+            Ok(Err(err)) => {
+                if handle_reconnect_delay(
+                    &events_tx,
+                    &mut shutdown_rx,
+                    &mut reconnect_attempt,
+                    &mut backoff,
+                    max_backoff,
+                    format!("connect_error:{err}"),
+                )
+                .await
+                {
+                    break;
+                }
+                continue;
+            }
+            Err(_) => {
+                if handle_reconnect_delay(
+                    &events_tx,
+                    &mut shutdown_rx,
+                    &mut reconnect_attempt,
+                    &mut backoff,
+                    max_backoff,
+                    "connect_timeout".to_string(),
+                )
+                .await
+                {
+                    break;
+                }
+                continue;
+            }
+        };
+
+        reconnect_attempt = 0;
+        backoff = base_backoff;
+
+        {
+            let mut guard = state.lock().await;
+            guard.connected = true;
+        }
+        emit(&events_tx, TransportEvent::Connected);
+
+        if let Some(jwt_qws) = config.jwt_qws.as_ref() {
+            if let Err(err) = send_authenticate(&mut ws, &mut msg_id, jwt_qws).await {
+                emit(
+                    &events_tx,
+                    TransportEvent::TransportError {
+                        stage: "authenticate".to_string(),
+                        message: err.to_string(),
+                    },
+                );
+                let _ = ws.close(None).await;
+                {
+                    let mut guard = state.lock().await;
+                    guard.connected = false;
+                }
+                emit(&events_tx, TransportEvent::Disconnected);
+                if handle_reconnect_delay(
+                    &events_tx,
+                    &mut shutdown_rx,
+                    &mut reconnect_attempt,
+                    &mut backoff,
+                    max_backoff,
+                    "authenticate_failed".to_string(),
+                )
+                .await
+                {
+                    break;
+                }
+                continue;
+            }
+            emit(&events_tx, TransportEvent::Authenticated);
+        }
+
+        if config.auto_subscribe {
+            if let Err(err) = send_subscribe(
+                &mut ws,
+                &mut msg_id,
+                config.qcloud_proto,
+                &config.subscribe_channels,
+            )
+            .await
+            {
+                emit(
+                    &events_tx,
+                    TransportEvent::TransportError {
+                        stage: "subscribe".to_string(),
+                        message: err.to_string(),
+                    },
+                );
+                let _ = ws.close(None).await;
+                {
+                    let mut guard = state.lock().await;
+                    guard.connected = false;
+                }
+                emit(&events_tx, TransportEvent::Disconnected);
+                if handle_reconnect_delay(
+                    &events_tx,
+                    &mut shutdown_rx,
+                    &mut reconnect_attempt,
+                    &mut backoff,
+                    max_backoff,
+                    "subscribe_failed".to_string(),
+                )
+                .await
+                {
+                    break;
+                }
+                continue;
+            }
+            emit(&events_tx, TransportEvent::Subscribed);
+        }
+
+        let mut keepalive = tokio::time::interval(Duration::from_millis(
+            config.keepalive_interval_ms.max(1_000),
+        ));
+        keepalive.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
+
+        let disconnect_reason = loop {
+            tokio::select! {
+                _ = shutdown_rx.changed() => {
+                    if *shutdown_rx.borrow() {
+                        let _ = ws.close(None).await;
+                        break "shutdown".to_string();
+                    }
+                }
+                maybe_envelope = outbound_rx.recv() => {
+                    match maybe_envelope {
+                        Some(envelope) => {
+                            if let Err(err) = send_outbound_payload(
+                                &mut ws,
+                                &mut msg_id,
+                                config.qcloud_proto,
+                                &envelope,
+                            ).await {
+                                break format!("send_error:{err}");
+                            }
+                            emit(
+                                &events_tx,
+                                TransportEvent::OutboundSent {
+                                    message_type: envelope.message_type,
+                                    action_uuid: envelope.action_uuid,
+                                },
+                            );
+                        }
+                        None => break "outbound_channel_closed".to_string(),
+                    }
+                }
+                _ = keepalive.tick() => {
+                    if let Err(err) = ws.send(WsMessage::Ping(Vec::new().into())).await {
+                        break format!("keepalive_ping_error:{err}");
+                    }
+                    emit(&events_tx, TransportEvent::KeepalivePingSent);
+                }
+                incoming = ws.next() => {
+                    match incoming {
+                        Some(Ok(WsMessage::Binary(data))) => {
+                            if let Err(err) = handle_incoming_binary(&events_tx, &data) {
+                                emit(
+                                    &events_tx,
+                                    TransportEvent::TransportError {
+                                        stage: "decode_inbound_binary".to_string(),
+                                        message: err.to_string(),
+                                    },
+                                );
+                            }
+                        }
+                        Some(Ok(WsMessage::Pong(_))) => {
+                            emit(&events_tx, TransportEvent::KeepalivePongReceived);
+                        }
+                        Some(Ok(WsMessage::Ping(payload))) => {
+                            let _ = ws.send(WsMessage::Pong(payload)).await;
+                        }
+                        Some(Ok(WsMessage::Close(_))) => {
+                            break "remote_close".to_string();
+                        }
+                        Some(Ok(_)) => {}
+                        Some(Err(err)) => {
+                            break format!("ws_read_error:{err}");
+                        }
+                        None => {
+                            break "ws_stream_end".to_string();
+                        }
+                    }
+                }
+            }
+        };
+
+        {
+            let mut guard = state.lock().await;
+            guard.connected = false;
+        }
+        emit(&events_tx, TransportEvent::Disconnected);
+
+        if *shutdown_rx.borrow() {
+            break;
+        }
+
+        if handle_reconnect_delay(
+            &events_tx,
+            &mut shutdown_rx,
+            &mut reconnect_attempt,
+            &mut backoff,
+            max_backoff,
+            disconnect_reason,
+        )
+        .await
+        {
+            break;
+        }
+    }
+
+    let mut guard = state.lock().await;
+    guard.connected = false;
+    guard.running = false;
+}
+
+async fn handle_reconnect_delay(
+    events_tx: &broadcast::Sender<TransportEvent>,
+    shutdown_rx: &mut watch::Receiver<bool>,
+    reconnect_attempt: &mut u32,
+    backoff_ms: &mut u64,
+    max_backoff: u64,
+    reason: String,
+) -> bool {
+    *reconnect_attempt = reconnect_attempt.saturating_add(1);
+
+    emit(
+        events_tx,
+        TransportEvent::ReconnectScheduled {
+            attempt: *reconnect_attempt,
+            backoff_ms: *backoff_ms,
+            reason,
+        },
+    );
+
+    tokio::select! {
+        _ = tokio::time::sleep(Duration::from_millis(*backoff_ms)) => {}
+        _ = shutdown_rx.changed() => {
+            if *shutdown_rx.borrow() {
+                return true;
+            }
+        }
+    }
+
+    *backoff_ms = (*backoff_ms).saturating_mul(2).min(max_backoff);
+    false
+}
+
+fn handle_incoming_binary(
+    events_tx: &broadcast::Sender<TransportEvent>,
+    data: &[u8],
+) -> Result<(), WsTransportError> {
+    let (cloud_message_type, payload) = decode_qcloud_frame(data)?;
+
+    emit(
+        events_tx,
+        TransportEvent::InboundFrameDecoded {
+            cloud_message_type,
+            payload_size: payload.len(),
+        },
+    );
+
+    match cloud_message_type {
+        MSG_TYPE_PAYLOAD => {
+            let cloud_payload = CloudPayload::decode(payload).map_err(|err| {
+                WsTransportError::Protocol(format!("decode payload envelope: {err}"))
+            })?;
+
+            let inner_payload = cloud_payload.payload.unwrap_or_default();
+            emit(
+                events_tx,
+                TransportEvent::InboundPayloadBytes {
+                    cloud_message_type,
+                    payload: inner_payload.clone(),
+                },
+            );
+
+            if let Ok(events) = decode_queue_server_events(&inner_payload) {
+                for event in events {
+                    emit(events_tx, TransportEvent::InboundQueueServerEvent(event));
+                }
+            }
+
+            if let Ok(inbound) = decode_inbound_json(&inner_payload) {
+                emit(events_tx, TransportEvent::InboundReceived(inbound));
+            }
+        }
+        MSG_TYPE_ERROR => {
+            emit(
+                events_tx,
+                TransportEvent::TransportError {
+                    stage: "qcloud_error_frame".to_string(),
+                    message: format!("bytes={}", payload.len()),
+                },
+            );
+        }
+        MSG_TYPE_DISCONNECT => {
+            emit(
+                events_tx,
+                TransportEvent::TransportError {
+                    stage: "qcloud_disconnect_frame".to_string(),
+                    message: "remote_disconnect_signal".to_string(),
+                },
+            );
+        }
+        _ => {}
+    }
+
+    Ok(())
+}
+
+async fn send_authenticate<S>(
+    ws: &mut tokio_tungstenite::WebSocketStream<S>,
+    msg_id: &mut u32,
+    jwt_qws: &str,
+) -> Result<(), WsTransportError>
+where
+    S: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin,
+{
+    let auth = Authenticate {
+        msg_id: Some(next_msg_id(msg_id)),
+        msg_date: Some(now_ms()),
+        jwt: Some(jwt_qws.to_string()),
+    };
+    let frame = encode_qcloud_frame(MSG_TYPE_AUTHENTICATE, &auth.encode_to_vec());
+    ws.send(WsMessage::Binary(frame.into()))
+        .await
+        .map_err(|err| WsTransportError::Protocol(format!("send authenticate: {err}")))
+}
+
+async fn send_subscribe<S>(
+    ws: &mut tokio_tungstenite::WebSocketStream<S>,
+    msg_id: &mut u32,
+    qcloud_proto: u32,
+    channels: &[Vec<u8>],
+) -> Result<(), WsTransportError>
+where
+    S: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin,
+{
+    let subscribe = Subscribe {
+        msg_id: Some(next_msg_id(msg_id)),
+        msg_date: Some(now_ms()),
+        proto: Some(qcloud_proto),
+        channels: channels.to_vec(),
+    };
+
+    let frame = encode_qcloud_frame(MSG_TYPE_SUBSCRIBE, &subscribe.encode_to_vec());
+    ws.send(WsMessage::Binary(frame.into()))
+        .await
+        .map_err(|err| WsTransportError::Protocol(format!("send subscribe: {err}")))
+}
+
+async fn send_outbound_payload<S>(
+    ws: &mut tokio_tungstenite::WebSocketStream<S>,
+    msg_id: &mut u32,
+    qcloud_proto: u32,
+    envelope: &OutboundEnvelope,
+) -> Result<(), WsTransportError>
+where
+    S: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin,
+{
+    let payload_bytes = encode_outbound_payload_bytes(envelope)
+        .map_err(|err| WsTransportError::Serialization(err.to_string()))?;
+
+    let payload = CloudPayload {
+        msg_id: Some(next_msg_id(msg_id)),
+        msg_date: Some(now_ms()),
+        proto: Some(qcloud_proto),
+        src: None,
+        dests: Vec::new(),
+        payload: Some(payload_bytes),
+    };
+
+    let frame = encode_qcloud_frame(MSG_TYPE_PAYLOAD, &payload.encode_to_vec());
+    ws.send(WsMessage::Binary(frame.into()))
+        .await
+        .map_err(|err| WsTransportError::Protocol(format!("send payload: {err}")))
+}
+
+fn encode_qcloud_frame(msg_type: u8, payload: &[u8]) -> Vec<u8> {
+    let mut frame = Vec::with_capacity(1 + 10 + payload.len());
+    frame.push(msg_type);
+    encode_varint(payload.len() as u64, &mut frame);
+    frame.extend_from_slice(payload);
+    frame
+}
+
+fn decode_qcloud_frame(data: &[u8]) -> Result<(u8, &[u8]), WsTransportError> {
+    if data.is_empty() {
+        return Err(WsTransportError::Protocol("empty qcloud frame".to_string()));
+    }
+
+    let msg_type = data[0];
+    let mut cursor = &data[1..];
+    let payload_len = decode_varint(&mut cursor)
+        .map_err(|err| WsTransportError::Protocol(format!("decode varint length: {err}")))?
+        as usize;
+
+    let consumed_varint = data.len().saturating_sub(1 + cursor.len());
+    let payload_start = 1 + consumed_varint;
+
+    if data.len() < payload_start + payload_len {
+        return Err(WsTransportError::Protocol(format!(
+            "truncated qcloud frame: expected={}, got={}",
+            payload_start + payload_len,
+            data.len()
+        )));
+    }
+
+    Ok((msg_type, &data[payload_start..payload_start + payload_len]))
+}
+
+fn emit(events_tx: &broadcast::Sender<TransportEvent>, event: TransportEvent) {
+    let _ = events_tx.send(event);
+}
+
+fn next_msg_id(msg_id: &mut u32) -> u32 {
+    *msg_id = msg_id.saturating_add(1);
+    *msg_id
+}
+
+fn now_ms() -> u64 {
+    SystemTime::now()
+        .duration_since(UNIX_EPOCH)
+        .unwrap_or_default()
+        .as_millis() as u64
+}
+
+#[derive(Clone, PartialEq, ::prost::Message)]
+struct Authenticate {
+    #[prost(uint32, optional, tag = "1")]
+    pub msg_id: Option<u32>,
+    #[prost(uint64, optional, tag = "2")]
+    pub msg_date: Option<u64>,
+    #[prost(string, optional, tag = "3")]
+    pub jwt: Option<String>,
+}
+
+#[derive(Clone, PartialEq, ::prost::Message)]
+struct Subscribe {
+    #[prost(uint32, optional, tag = "1")]
+    pub msg_id: Option<u32>,
+    #[prost(uint64, optional, tag = "2")]
+    pub msg_date: Option<u64>,
+    #[prost(uint32, optional, tag = "3")]
+    pub proto: Option<u32>,
+    #[prost(bytes = "vec", repeated, tag = "4")]
+    pub channels: Vec<Vec<u8>>,
+}
+
+#[derive(Clone, PartialEq, ::prost::Message)]
+struct CloudPayload {
+    #[prost(uint32, optional, tag = "1")]
+    pub msg_id: Option<u32>,
+    #[prost(uint64, optional, tag = "2")]
+    pub msg_date: Option<u64>,
+    #[prost(uint32, optional, tag = "3")]
+    pub proto: Option<u32>,
+    #[prost(bytes = "vec", optional, tag = "4")]
+    pub src: Option<Vec<u8>>,
+    #[prost(bytes = "vec", repeated, tag = "5")]
+    pub dests: Vec<Vec<u8>>,
+    #[prost(bytes = "vec", optional, tag = "7")]
+    pub payload: Option<Vec<u8>>,
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn qcloud_frame_roundtrip() {
+        let payload = vec![0x01, 0x02, 0x03, 0x04];
+        let encoded = encode_qcloud_frame(MSG_TYPE_PAYLOAD, &payload);
+        let (msg_type, decoded_payload) = decode_qcloud_frame(&encoded).expect("decode frame");
+        assert_eq!(msg_type, MSG_TYPE_PAYLOAD);
+        assert_eq!(decoded_payload, payload.as_slice());
+    }
+
+    #[test]
+    fn qcloud_frame_rejects_truncated_payload() {
+        let payload = vec![0x0a, 0x0b, 0x0c];
+        let mut encoded = encode_qcloud_frame(MSG_TYPE_PAYLOAD, &payload);
+        encoded.truncate(encoded.len() - 1);
+        let err = decode_qcloud_frame(&encoded).expect_err("expected truncation error");
+        assert!(matches!(err, WsTransportError::Protocol(_)));
+    }
+
+    #[test]
+    fn cloud_payload_proto_roundtrip() {
+        let payload = CloudPayload {
+            msg_id: Some(12),
+            msg_date: Some(34),
+            proto: Some(1),
+            src: None,
+            dests: vec![vec![1, 2, 3]],
+            payload: Some(vec![9, 8, 7]),
+        };
+        let encoded = payload.encode_to_vec();
+        let decoded = CloudPayload::decode(encoded.as_slice()).expect("decode cloud payload");
+        assert_eq!(decoded.msg_id, Some(12));
+        assert_eq!(decoded.proto, Some(1));
+        assert_eq!(decoded.payload, Some(vec![9, 8, 7]));
+    }
+}