liminal-sdk 0.19.5

//! Client-side WebSocket subscription stream: the receive half of the
//! delivery pump, sibling to the TCP [`SubscriptionStream`].
//!
//! One subscription per dedicated connection (the v1 shape). The socket open
//! is authorized through the client unit exactly like the request/response
//! transport (R2.2), the background reader blocks on socket input with no
//! timer or polling loop, and teardown shuts the socket down so the blocked
//! reader exits on the socket's own typed terminal event (LAW-1: the socket
//! signals; nothing sweeps).
//!
//! [`SubscriptionStream`]: crate::SubscriptionStream

use alloc::format;
use alloc::string::ToString;
use alloc::sync::Arc;
use alloc::vec::Vec;

use core::time::Duration;

use std::net::TcpStream;
use std::thread::JoinHandle;

use liminal::protocol::{Frame, SchemaId, decode};
use liminal_protocol::outcome::ReconnectState;
use spin::Mutex;

use crate::SdkError;
use crate::remote::SETUP_TIMEOUT;
use crate::remote::mailbox::{
    Mailbox, MailboxItem, MailboxLimits, MailboxStats, MailboxWords, ReaderEnd, offer_or_end,
};

/// The WebSocket subscription's words in the errors its mailbox hands the
/// consumer.
const WEBSOCKET_SUBSCRIPTION_WORDS: MailboxWords = MailboxWords {
    receive: "websocket subscription receive failed",
    item: "delivery",
    reader: "subscription reader",
};

use super::binding::{AttemptFateOutcome, OpenRequestDecision, WebSocketAuthorityBinding};
use super::connection_error;
use super::core::{
    DriverOutput, FrameCorrelation, ResponseExpectation, SocketCommand, SocketEvent,
    WebSocketFrameDriver,
};
use super::liminal_ws_message_bound;
use super::std_socket::{SocketRead, WsSocket};

/// Minimum protocol version this client advertises during the handshake.
const CLIENT_MIN_VERSION: liminal::protocol::ProtocolVersion =
    liminal::protocol::ProtocolVersion::new(1, 0);
/// Maximum protocol version this client advertises during the handshake; the
/// server selects the highest version both support, so 1.0 stays offered.
const CLIENT_MAX_VERSION: liminal::protocol::ProtocolVersion =
    liminal::protocol::DELIVER_MESSAGE_ID_VERSION;
/// The single application stream this subscription's deliveries ride on.
const SUBSCRIPTION_STREAM_ID: u32 = 1;
/// In-flight window advertised on subscribe (advisory in v1; TCP parity).
const SUBSCRIBE_MAX_IN_FLIGHT: u32 = 1024;

/// A message the server delivered on this WebSocket subscription.
///
/// Mirrors the TCP [`DeliveredMessage`](crate::DeliveredMessage) surface; the
/// TCP type keeps its fields private, so the WebSocket sibling carries its own
/// identical shape.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct WebSocketDeliveredMessage {
    delivery_seq: u64,
    schema_id: SchemaId,
    payload: Vec<u8>,
    /// The server's id for the message and its boot, when a 1.2 server sent it.
    message: Option<liminal::protocol::DeliveredId>,
}

impl WebSocketDeliveredMessage {
    /// The per-subscription monotonic delivery sequence (starts at 1).
    #[must_use]
    pub const fn delivery_seq(&self) -> u64 {
        self.delivery_seq
    }

    /// The schema id the server selected for this subscription's stream.
    #[must_use]
    pub const fn schema_id(&self) -> SchemaId {
        self.schema_id
    }

    /// The delivered payload bytes.
    #[must_use]
    pub fn payload(&self) -> &[u8] {
        &self.payload
    }

    /// Consumes the message, returning the owned payload bytes.
    #[must_use]
    pub fn into_payload(self) -> Vec<u8> {
        self.payload
    }

    /// The server's id for this message and the boot it was drawn in, when the
    /// server sent it (protocol 1.2, Fable 4c). `None` from an older server,
    /// and for a durable message replayed from the store.
    #[must_use]
    pub const fn message(&self) -> Option<liminal::protocol::DeliveredId> {
        self.message
    }

    /// The causal parent a reply to this message names, `<boot>:<id>`, or
    /// `None` when the message carried no id: never an invented one.
    #[must_use]
    pub fn reply_parent(&self) -> Option<liminal::protocol::MessageId> {
        self.message
            .map(liminal::protocol::DeliveredId::reply_parent)
    }
}

impl MailboxItem for WebSocketDeliveredMessage {
    fn payload_len(&self) -> usize {
        self.payload.len()
    }
}

/// A connected WebSocket subscription whose background reader surfaces
/// delivered messages.
#[derive(Debug)]
pub struct WebSocketSubscriptionStream {
    /// Shutdown handle for the one socket; used on drop and by the close handle.
    pub(super) shutdown: TcpStream,
    subscription_id: u64,
    /// Delivered messages, bounded in count and bytes, then the one typed end
    /// the reader stopped with. A `SubscribeError` arriving mid-stream is the
    /// ONLY explanation the consumer will ever get for deliveries stopping, so
    /// it is that end, received after every delivery before it (P0 #55). TCP
    /// parity: the sibling carries the identical shape.
    pub(super) mailbox: Arc<Mailbox<WebSocketDeliveredMessage>>,
    binding: Arc<Mutex<WebSocketAuthorityBinding>>,
    reader: Option<JoinHandle<()>>,
}

impl WebSocketSubscriptionStream {
    /// Connects to the `ws://` address, performs the liminal handshake,
    /// subscribes to `channel`, and starts the background reader.
    ///
    /// Deliveries the server coalesces with the `SubscribeAck` are retained
    /// and surfaced first, never dropped.
    ///
    /// # Errors
    ///
    /// Returns [`SdkError::Connection`] when the client unit refuses the
    /// open or the socket cannot be opened, and [`SdkError::Protocol`] when
    /// the handshake or subscribe is rejected.
    pub fn open(
        address: &str,
        channel: &str,
        accepted_schemas: Vec<SchemaId>,
    ) -> Result<Self, SdkError> {
        Self::open_with_auth(address, channel, accepted_schemas, &[])
    }

    /// Connects to the `ws://` address, performs the liminal handshake
    /// carrying `auth_token`, subscribes to `channel`, and starts the
    /// background reader.
    ///
    /// A subscription owns a dedicated connection (the v1 shape), so it
    /// presents its own credential in its own `Connect` frame; the token a
    /// request/response transport was built with lives on that transport's
    /// socket and cannot travel here. Additive to [`open`]: an empty token is
    /// exactly the open-access handshake `open` performs, so an ungated server
    /// sees byte-identical bytes either way. TCP parity with
    /// [`SubscriptionStream::open_with_auth`].
    ///
    /// # Errors
    ///
    /// Returns [`SdkError::Connection`] when the client unit refuses the open,
    /// the socket cannot be opened, or the token is rejected, and
    /// [`SdkError::Protocol`] when the subscribe is rejected.
    ///
    /// [`open`]: Self::open
    /// [`SubscriptionStream::open_with_auth`]: crate::SubscriptionStream::open_with_auth
    pub fn open_with_auth(
        address: &str,
        channel: &str,
        accepted_schemas: Vec<SchemaId>,
        auth_token: &[u8],
    ) -> Result<Self, SdkError> {
        Self::open_with_limits(
            address,
            channel,
            accepted_schemas,
            auth_token,
            MailboxLimits::SUBSCRIPTION,
        )
    }

    /// [`Self::open_with_auth`] with the mailbox bounds supplied.
    fn open_with_limits(
        address: &str,
        channel: &str,
        accepted_schemas: Vec<SchemaId>,
        auth_token: &[u8],
        limits: MailboxLimits,
    ) -> Result<Self, SdkError> {
        let message_bound = liminal_ws_message_bound()?;
        let mut binding = WebSocketAuthorityBinding::new();
        match binding.request_open() {
            OpenRequestDecision::Authorized { .. } => {}
            OpenRequestDecision::Refused(refusal) => {
                return Err(connection_error(&format!(
                    "client authority refused the subscription open: {refusal:?}"
                )));
            }
        }
        match Self::open_link(
            address,
            channel,
            accepted_schemas,
            message_bound,
            auth_token,
        ) {
            Ok((socket, driver, subscription_id, pending)) => {
                match binding.connection_established() {
                    AttemptFateOutcome::Recorded { .. } => {}
                    AttemptFateOutcome::Refused(refusal) => {
                        return Err(SdkError::Protocol {
                            description: format!(
                                "client authority refused the Connected fate for the \
                                 subscription open: {refusal:?}"
                            ),
                        });
                    }
                }
                Self::start(socket, driver, binding, subscription_id, pending, limits)
            }
            Err(error) => match binding.open_failed() {
                AttemptFateOutcome::Recorded { .. } => Err(error),
                AttemptFateOutcome::Refused(refusal) => Err(SdkError::Protocol {
                    description: format!(
                        "subscription open failed ({error}) and the client authority \
                             refused the Failed fate: {refusal:?}"
                    ),
                }),
            },
        }
    }

    /// Blocks up to `timeout` for the next delivered message.
    ///
    /// # Errors
    ///
    /// Returns [`SdkError::Connection`] when no message arrives within
    /// `timeout`. Once the reader has stopped and every held delivery has been
    /// taken, returns the error it stopped with (the server's mid-stream
    /// `SubscribeError`, a decode failure, or [`SdkError::Backpressure`] naming
    /// a delivery too large for the mailbox), or a named stop when the link
    /// ended without one.
    pub fn recv_timeout(&self, timeout: Duration) -> Result<WebSocketDeliveredMessage, SdkError> {
        self.mailbox.take(timeout)
    }

    /// The mailbox counters: deliveries and bytes held now and at peak, how
    /// often the reader stopped reading for a consumer that had not made
    /// room, refused and stray frames, and errors no consumer was left to
    /// receive.
    ///
    /// # Errors
    ///
    /// Returns [`SdkError::Connection`] when the mailbox lock is poisoned.
    pub fn stats(&self) -> Result<MailboxStats, SdkError> {
        self.mailbox.stats()
    }

    /// The server-assigned id for this subscription.
    #[must_use]
    pub const fn subscription_id(&self) -> u64 {
        self.subscription_id
    }

    /// The client unit's reconnect state for this subscription's connection.
    #[must_use]
    pub fn reconnect_state(&self) -> ReconnectState {
        self.binding.lock().reconnect_state()
    }

    /// Performs the socket open, handshake carrying `auth_token`, and subscribe
    /// exchange.
    fn open_link(
        address: &str,
        channel: &str,
        accepted_schemas: Vec<SchemaId>,
        message_bound: usize,
        auth_token: &[u8],
    ) -> Result<
        (
            WsSocket,
            WebSocketFrameDriver,
            u64,
            Vec<WebSocketDeliveredMessage>,
        ),
        SdkError,
    > {
        let mut driver = WebSocketFrameDriver::new();
        let command = driver
            .command_open()
            .map_err(|refusal| SdkError::Protocol {
                description: format!("subscription driver refused its first open: {refusal:?}"),
            })?;
        if command != SocketCommand::Open {
            return Err(SdkError::Protocol {
                description: "subscription driver emitted a non-open first command".to_string(),
            });
        }
        let mut socket = WsSocket::connect(address, message_bound)?;
        // Name the deadline this reader gives a synchronous control-frame reply
        // rather than inheriting the socket layer's unnamed default. One value,
        // shared by all three readers; `start` takes it back off before the
        // background reader ever blocks on this socket.
        socket.set_read_timeout(Some(SETUP_TIMEOUT))?;
        let step = driver.handle_event(SocketEvent::Opened);
        if step.output != DriverOutput::Opened {
            return Err(SdkError::Protocol {
                description: format!("subscription driver refused the opened socket: {step:?}"),
            });
        }

        let mut pending = Vec::new();
        let connect = Frame::Connect {
            flags: 0,
            min_version: CLIENT_MIN_VERSION,
            max_version: CLIENT_MAX_VERSION,
            auth_token: auth_token.to_vec(),
        };
        match setup_exchange(&mut socket, &mut driver, &connect, &mut pending)? {
            Frame::ConnectAck { .. } => {}
            Frame::ConnectError {
                reason_code,
                message,
                ..
            } => {
                return Err(connection_error(&format!(
                    "server rejected subscription connection (reason {reason_code}): {}",
                    message.unwrap_or_else(|| "no detail".to_string())
                )));
            }
            other => {
                return Err(unexpected_setup_frame("ConnectAck", &other));
            }
        }

        let subscribe = Frame::Subscribe {
            flags: 0,
            stream_id: SUBSCRIPTION_STREAM_ID,
            channel: channel.to_string(),
            accepted_schemas,
            max_in_flight: SUBSCRIBE_MAX_IN_FLIGHT,
        };
        let subscription_id =
            match setup_exchange(&mut socket, &mut driver, &subscribe, &mut pending)? {
                Frame::SubscribeAck {
                    subscription_id, ..
                } => subscription_id,
                Frame::SubscribeError {
                    reason_code,
                    message,
                    ..
                } => {
                    return Err(SdkError::Protocol {
                        description: format!(
                            "server rejected subscribe (reason {reason_code}): {}",
                            message.unwrap_or_else(|| "no detail".to_string())
                        ),
                    });
                }
                other => {
                    return Err(unexpected_setup_frame("SubscribeAck", &other));
                }
            };
        Ok((socket, driver, subscription_id, pending))
    }

    /// Starts the background reader over the established link.
    fn start(
        socket: WsSocket,
        driver: WebSocketFrameDriver,
        binding: WebSocketAuthorityBinding,
        subscription_id: u64,
        pending: Vec<WebSocketDeliveredMessage>,
        limits: MailboxLimits,
    ) -> Result<Self, SdkError> {
        // The control exchange is over, so its deadline comes off: the reader
        // blocks on socket input with no read window. Teardown shuts the socket
        // down, which surfaces as a typed terminal event. This is the shape the
        // two TCP readers were generalized toward — it must not regress into a
        // cadence, whatever period that cadence would carry.
        socket.set_read_timeout(None)?;
        let shutdown = socket.try_clone_stream()?;
        let binding = Arc::new(Mutex::new(binding));
        let reader_binding = Arc::clone(&binding);
        let mailbox = Arc::new(Mailbox::new(limits, WEBSOCKET_SUBSCRIPTION_WORDS));
        let reader_mailbox = Arc::clone(&mailbox);
        let reader = std::thread::Builder::new()
            .name("liminal-ws-subscription-reader".to_string())
            .spawn(move || {
                let end = run_reader(socket, driver, &reader_binding, &reader_mailbox, pending);
                reader_mailbox.close(end);
            })
            .map_err(|source| SdkError::Protocol {
                description: format!(
                    "failed to start websocket subscription reader thread: {source}"
                ),
            })?;
        Ok(Self {
            shutdown,
            subscription_id,
            mailbox,
            binding,
            reader: Some(reader),
        })
    }
}

impl Drop for WebSocketSubscriptionStream {
    fn drop(&mut self) {
        // Best-effort teardown: shutting the shared socket down surfaces a
        // typed terminal to the blocked reader, which records it and exits.
        // The server frees the subscription on the connection's terminal fate
        // (R1.3), so no unsubscribe write is required from this half.
        self.shutdown.shutdown(std::net::Shutdown::Both).ok();
        // Nothing receives from here on: release a reader waiting on a full
        // mailbox so the join cannot wait on a consumer that has gone.
        self.mailbox.release();
        if let Some(reader) = self.reader.take() {
            reader.join().ok();
        }
    }
}

/// Sends one setup request and blocks for its correlated control reply,
/// retaining (never dropping) any deliveries that arrive first.
fn setup_exchange(
    socket: &mut WsSocket,
    driver: &mut WebSocketFrameDriver,
    request: &Frame,
    pending: &mut Vec<WebSocketDeliveredMessage>,
) -> Result<Frame, SdkError> {
    let bytes = super::encode_frame(request)?;
    let command = driver
        .command_send(bytes, ResponseExpectation::Correlated)
        .map_err(|refusal| SdkError::Protocol {
            description: format!("subscription driver refused the setup send: {refusal:?}"),
        })?;
    let SocketCommand::SendBinary(payload) = command else {
        return Err(SdkError::Protocol {
            description: "subscription driver emitted a non-send command for a send".to_string(),
        });
    };
    if let Err(failure) = socket.send_binary(payload) {
        let step = driver.handle_event(SocketEvent::Failed(failure));
        if step.command == Some(SocketCommand::Close) {
            socket.execute_close();
        }
        return Err(connection_error(&format!(
            "failed to send subscription setup frame: {}",
            socket
                .last_failure_detail()
                .unwrap_or("websocket send failed")
        )));
    }
    loop {
        let event = match socket.read_event() {
            SocketRead::TimedOut => {
                return Err(connection_error(
                    "subscription connection timed out waiting for a control-frame reply",
                ));
            }
            SocketRead::Event(event) => event,
        };
        let step = driver.handle_event(event);
        if step.command == Some(SocketCommand::Close) {
            socket.execute_close();
        }
        match step.output {
            DriverOutput::Frame { bytes, correlation } => {
                let frame = decode_message(&bytes)?;
                match correlation {
                    FrameCorrelation::UnsolicitedDelivery => {
                        if let Some(message) = delivered_message(frame) {
                            pending.push(message);
                        }
                    }
                    FrameCorrelation::CorrelatedResponse | FrameCorrelation::UnsolicitedFrame => {
                        return Ok(frame);
                    }
                }
            }
            DriverOutput::Terminal(terminal) => {
                return Err(connection_error(&format!(
                    "subscription connection terminated during setup: {terminal:?}"
                )));
            }
            DriverOutput::Opened
            | DriverOutput::PostTerminalIgnored(_)
            | DriverOutput::Refused(_) => {
                return Err(SdkError::Protocol {
                    description: format!(
                        "subscription driver produced an unexpected setup output: {:?}",
                        step.output
                    ),
                });
            }
        }
    }
}

/// Background loop: feeds every socket fact through the driver and surfaces
/// each delivery in `mailbox`. Ends on the link's typed terminal fate
/// (recorded into the client unit), returning the first error it met (a
/// decode failure, the server's `SubscribeError`, or a delivery that could
/// never fit the mailbox), or a plain stop when the link ended without one.
/// A full mailbox holds the loop inside its offer, so the socket is not read
/// until the consumer makes room.
fn run_reader(
    mut socket: WsSocket,
    mut driver: WebSocketFrameDriver,
    binding: &Mutex<WebSocketAuthorityBinding>,
    mailbox: &Mailbox<WebSocketDeliveredMessage>,
    pending: Vec<WebSocketDeliveredMessage>,
) -> ReaderEnd {
    // The first error the reader meets is the one the consumer receives; the
    // loop keeps running after it only to let the link's terminal fate reach
    // the client unit.
    let mut first_error: Option<ReaderEnd> = None;
    for message in pending {
        if let Some(end) = offer_or_end(mailbox, message) {
            close_link(&mut socket, &mut driver);
            first_error = Some(end);
            break;
        }
    }
    loop {
        let event = match socket.read_event() {
            // No read window is armed on the reader socket; a timeout here
            // means the OS returned early, and re-entering the blocking read
            // is the only correct continuation (not a poll: no interval).
            SocketRead::TimedOut => continue,
            SocketRead::Event(event) => event,
        };
        let step = driver.handle_event(event);
        if step.command == Some(SocketCommand::Close) {
            socket.execute_close();
        }
        match step.output {
            DriverOutput::Frame { bytes, correlation } => {
                if first_error.is_some() {
                    // The link is closing on an error already recorded; what
                    // still arrives is read only to reach the terminal.
                    continue;
                }
                first_error = handle_frame(&bytes, correlation, mailbox);
                if first_error.is_some() {
                    close_link(&mut socket, &mut driver);
                }
            }
            DriverOutput::Terminal(terminal) => {
                // The one typed fate of this link enters the client unit.
                let _outcome = binding.lock().established_terminal(&terminal);
                return first_error.unwrap_or(ReaderEnd::Stopped);
            }
            DriverOutput::PostTerminalIgnored(_) => {
                return first_error.unwrap_or(ReaderEnd::Stopped);
            }
            DriverOutput::Opened | DriverOutput::Refused(_) => {}
        }
    }
}

/// Handles one validated frame: a delivery is offered to the mailbox, a server
/// `Disconnect` or `SubscribeError` or a decode failure becomes the reader's
/// end, and a stray frame is counted. Returns the end when there is one.
fn handle_frame(
    bytes: &[u8],
    correlation: FrameCorrelation,
    mailbox: &Mailbox<WebSocketDeliveredMessage>,
) -> Option<ReaderEnd> {
    let frame = match decode_message(bytes) {
        Ok(frame) => frame,
        // The consumer receives the decode error itself, not a bare end.
        Err(error) => return Some(ReaderEnd::Failed(error)),
    };
    match (correlation, frame) {
        (
            FrameCorrelation::UnsolicitedDelivery,
            Frame::Deliver {
                delivery_seq,
                envelope,
                message,
                ..
            },
        ) => offer_or_end(
            mailbox,
            WebSocketDeliveredMessage {
                delivery_seq,
                schema_id: envelope.schema_id,
                payload: envelope.payload,
                message,
            },
        ),
        // A server Disconnect ends the subscription cleanly; commanding close
        // lets the echoed close event mint the one typed terminal.
        (_, Frame::Disconnect { .. }) => Some(ReaderEnd::Stopped),
        // A `SubscribeError` arriving AFTER setup is the server ending this
        // subscription -- the overflow shed sends exactly this and then
        // releases the subscription at the channel actor, so no further
        // delivery can ever arrive. It is the consumer's end, and the link is
        // closed (P0 #55). This is the one exception to the stray-frame rule
        // below, and the distinction is deliveries: ignoring a stray frame
        // protects the deliveries still to come, and here there are none. A
        // consumer already gone is counted by the mailbox's close.
        (
            _,
            Frame::SubscribeError {
                reason_code,
                message,
                ..
            },
        ) => Some(ReaderEnd::Failed(subscription_ended(reason_code, message))),
        // Any other frame on a subscription connection, including a
        // non-`Deliver` frame under the delivery discriminant, is unexpected;
        // it is set aside and counted (TCP parity) so a stray frame cannot drop
        // the deliveries behind it.
        _ => mailbox.record_ignored_frame().err().map(ReaderEnd::Failed),
    }
}

/// Commands a close (when still legal) and executes it on the socket.
fn close_link(socket: &mut WsSocket, driver: &mut WebSocketFrameDriver) {
    if driver.command_close().is_ok() {
        socket.execute_close();
    }
}

/// Decodes one driver-validated message into its canonical frame.
fn decode_message(bytes: &[u8]) -> Result<Frame, SdkError> {
    match decode(bytes) {
        Ok((frame, consumed)) if consumed == bytes.len() => Ok(frame),
        Ok((_, consumed)) => Err(SdkError::Protocol {
            description: format!(
                "subscription decode consumed {consumed} of {} message bytes",
                bytes.len()
            ),
        }),
        Err(error) => Err(SdkError::Protocol {
            description: format!("subscription wire codec error: {error}"),
        }),
    }
}

/// Maps a `Deliver` frame to its delivered message; other frames map to none.
fn delivered_message(frame: Frame) -> Option<WebSocketDeliveredMessage> {
    match frame {
        Frame::Deliver {
            delivery_seq,
            envelope,
            message,
            ..
        } => Some(WebSocketDeliveredMessage {
            delivery_seq,
            schema_id: envelope.schema_id,
            payload: envelope.payload,
            message,
        }),
        _ => None,
    }
}

/// Builds the typed terminal for a `SubscribeError` the server sent mid-stream.
///
/// The server's own detail is carried VERBATIM, and the wording matches the TCP
/// sibling exactly: a consumer that switched transports must not have to learn a
/// second vocabulary for the same event.
fn subscription_ended(reason_code: u16, message: Option<alloc::string::String>) -> SdkError {
    SdkError::Protocol {
        description: format!(
            "server ended the subscription (reason {reason_code}): {}",
            message.unwrap_or_else(|| "no detail".to_string())
        ),
    }
}

/// Builds a protocol error describing an unexpected setup response frame.
fn unexpected_setup_frame(expected: &str, actual: &Frame) -> SdkError {
    SdkError::Protocol {
        description: format!(
            "expected {expected} during subscription setup, received {:?}",
            actual.frame_type()
        ),
    }
}

#[cfg(test)]
#[path = "subscription_decode_tests.rs"]
mod decode_tests;

#[cfg(test)]
mod tests {
    use super::SETUP_TIMEOUT;
    use core::time::Duration;

    /// TOMBSTONE (SDK-010 R5) — the WebSocket subscription reader is the shape
    /// the two TCP readers were generalized toward, and it must stay that shape:
    /// no reader poll family, no stop flag, no cadence in steady state. Its
    /// disarm (`socket.set_read_timeout(None)` before the reader spawns) is
    /// exercised behaviorally by the landed `sdk_ws_e2e` and
    /// `ws_transport_parity` suites, which open a real subscription over a real
    /// server; this guard holds the source itself.
    #[test]
    fn websocket_subscription_source_has_no_retired_reader_poll_family() {
        const SOURCE: &str = include_str!("subscription.rs");
        let production = SOURCE.split("#[cfg(test)]").next().unwrap_or(SOURCE);
        for forbidden in [
            "READER_POLL_TIMEOUT",
            "AtomicBool",
            "stop.load",
            "stop.store",
            "re-check the stop flag",
            "poll the stop flag",
        ] {
            assert!(
                !production.contains(forbidden),
                "retired websocket-subscription-reader poll-family source \
                 `{forbidden}` reappeared"
            );
        }
    }

    /// The one named deadline (SDK-010 R3): 5 s, shared by all three readers and
    /// generalized from the estate's already-ratified value rather than
    /// re-chosen. Pinned here so a per-reader fork of the value fails loudly.
    #[test]
    fn the_named_setup_deadline_is_the_ratified_five_seconds() {
        assert_eq!(SETUP_TIMEOUT, Duration::from_secs(5));
    }

    /// The boot and id a 1.2 server gives a delivery (Fable 4c).
    const DELIVERED: liminal::protocol::DeliveredId = liminal::protocol::DeliveredId {
        boot: 1_791_328_512_345,
        id: 42,
    };

    fn deliver(message: Option<liminal::protocol::DeliveredId>) -> liminal::protocol::Frame {
        liminal::protocol::Frame::Deliver {
            flags: 0,
            stream_id: super::SUBSCRIPTION_STREAM_ID,
            delivery_seq: 1,
            envelope: liminal::protocol::MessageEnvelope::new(
                liminal::protocol::SchemaId::new([7; liminal::protocol::SchemaId::WIRE_LEN]),
                liminal::protocol::CausalContext::independent(),
                alloc::vec![1],
            ),
            message,
        }
    }

    /// 4c, as on TCP: the WebSocket subscription offers 1.2.
    #[test]
    fn the_subscription_offers_the_delivered_id_version() {
        assert_eq!(
            super::CLIENT_MAX_VERSION,
            liminal::protocol::DELIVER_MESSAGE_ID_VERSION
        );
    }

    /// 4c, as on TCP: a delivery keeps the id its Deliver carried and names
    /// its reply's parent by it, and one without an id names no parent.
    #[test]
    fn a_delivered_message_keeps_its_id_and_names_its_reply_parent() {
        let with_id = super::delivered_message(deliver(Some(DELIVERED)));
        assert_eq!(
            with_id
                .as_ref()
                .and_then(super::WebSocketDeliveredMessage::message),
            Some(DELIVERED)
        );
        assert_eq!(
            with_id.and_then(|message| message.reply_parent()),
            Some(liminal::protocol::MessageId::from("1791328512345:42"))
        );
        let without = super::delivered_message(deliver(None));
        assert_eq!(
            without
                .as_ref()
                .and_then(super::WebSocketDeliveredMessage::message),
            None
        );
        assert_eq!(without.and_then(|message| message.reply_parent()), None);
    }
}