liminal_bus_protocol/

frame.rs

use super::{
    causal::MessageId,
    envelope::{MessageEnvelope, SchemaId},
    error::ProtocolError,
    version::ProtocolVersion,
};

/// Number of bytes in every serialized frame header.
pub const HEADER_LEN: usize = 10;

/// Frame-flag bit set on a [`Frame::ConversationMessage`] to request a correlated
/// reply.
///
/// A client sets this bit on the request frame of a request-reply round trip. The
/// server, after delivering the message to the conversation participant, drains
/// the participant's reply and sends it back as a `ConversationMessage` carrying
/// the same `conversation_id` (the correlation key) and this same flag bit. A
/// `ConversationMessage` WITHOUT this bit keeps the pre-existing fire-and-forget
/// semantics: the server stays silent on success. The bit travels in the frame
/// header's `flags` byte, which the codec already round-trips, so no wire-format
/// change is required.
pub const CONVERSATION_REPLY_REQUESTED_FLAG: u8 = 0x01;

/// Frame-flag bit set on a [`Frame::Publish`] to declare that the frame body
/// carries a trailing idempotency-key string field (the dedup-on-delivery key).
///
/// A publisher sets this bit when it wants the server to consult its dedup cache
/// keyed by the trailing idempotency key, delivering the message to subscribers
/// AT MOST ONCE across re-publishes of the same key. A `Publish` frame WITHOUT
/// this bit keeps the pre-existing wire layout EXACTLY: no trailing field is
/// written and none is read, so a no-key publish is byte-identical to before.
/// The bit travels in the frame header's `flags` byte, which the codec already
/// round-trips, so no header-format change is required (the 13-L0 precedent).
pub const PUBLISH_IDEMPOTENCY_KEY_FLAG: u8 = 0x02;

/// Frame-flag bit set on a [`Frame::PublishAck`] to report a GENUINE delivery
/// ack: the published message was accepted by at least one live subscriber on
/// this publish.
///
/// This is distinct from the backpressure `Accept`/`Defer`/`Reject` signal: a
/// `PublishAck` always means the publish was processed without error, but only a
/// `PublishAck` carrying this bit means a subscriber actually received the
/// message. An ack WITHOUT this bit means the publish succeeded but reached no
/// subscriber (an empty channel, or a duplicate suppressed by dedup-on-delivery),
/// so a caller that needs a true delivery ack can observe the difference. The bit
/// rides the existing `flags` byte, so no wire-format change is required.
pub const PUBLISH_DELIVERED_FLAG: u8 = 0x01;

/// Frame-flag bit set on a [`Frame::PublishAck`] to declare that the frame body
/// carries a trailing durable-position field after `message_id`.
///
/// The durable position is the store-assigned position of the row the publish
/// answers for, written as a `u64` big-endian (the width and byte order of
/// `message_id`). An ack WITHOUT this bit keeps the older layout EXACTLY: no
/// trailing field is written and none is read, so an ack that carries no
/// position is byte-identical to before (the [`PUBLISH_IDEMPOTENCY_KEY_FLAG`]
/// precedent). The codec sets or clears the bit from the frame's
/// `durable_position`, so a caller never sets it by hand.
pub const PUBLISH_DURABLE_POSITION_FLAG: u8 = 0x02;

/// Protocol version that carries the [`Frame::PublishAck`] durable position.
///
/// A server writes an ack carrying a position only on a connection that
/// negotiated this version or later; a connection at 1.0 always receives the
/// older ack layout.
pub const PUBLISH_ACK_POSITION_VERSION: ProtocolVersion = ProtocolVersion::new(1, 1);

/// Frame-flag bit declaring that a `PublishAck` carries its message's boot.
///
/// The boot is a big-endian `u64` after the durable position when there is
/// one, and after `message_id` otherwise. Without this bit the older layout is
/// unchanged. The codec sets or clears it from the frame's `boot`, so a caller
/// never sets it by hand.
pub const PUBLISH_ACK_BOOT_FLAG: u8 = 0x04;

/// Protocol version that carries the [`Frame::PublishAck`] boot.
///
/// A server writes the boot only on a connection that negotiated this version
/// or later; an older connection always receives the older ack layout.
pub const PUBLISH_ACK_BOOT_VERSION: ProtocolVersion = ProtocolVersion::new(1, 4);

/// Frame-flag bit set on a [`Frame::Deliver`] to declare that the frame body
/// carries a trailing [`DeliveredId`] after the envelope.
///
/// The pair is written as two `u64`s big-endian, the boot then the id. A
/// Deliver WITHOUT this bit keeps the older layout EXACTLY: nothing trailing is
/// written and nothing is read, so a 1.0 or 1.1 peer receives the bytes it
/// always did. The codec sets or clears the bit from the frame's `message`, so a
/// caller never sets it by hand (the [`PUBLISH_DURABLE_POSITION_FLAG`]
/// precedent).
pub const DELIVER_MESSAGE_ID_FLAG: u8 = 0x01;

/// A `Deliver` carries a trailing big-endian durable position.
pub const DELIVER_POSITION_FLAG: u8 = 0x02;

/// Protocol version supporting positioned deliveries and `DeliverAck`.
pub const INBOX_ACK_VERSION: ProtocolVersion = ProtocolVersion::new(1, 5);

/// Protocol version that carries the [`Frame::Deliver`] message id.
///
/// A server writes a Deliver carrying a [`DeliveredId`] only on a connection
/// that negotiated this version or later; a connection at 1.0 or 1.1 always
/// receives the older Deliver layout.
pub const DELIVER_MESSAGE_ID_VERSION: ProtocolVersion = ProtocolVersion::new(1, 2);

/// The server's id for a delivered message, with the boot it was drawn in.
///
/// `id` is the number the publisher's `PublishAck` carried and the console's
/// tap id names; it is drawn from a counter that restarts on every boot, so
/// `boot` (the server incarnation's start, in wall-clock milliseconds since
/// the epoch) says which run of the server the number belongs to. A reply that
/// names a request does so by both, and a tap pairs it only within its own boot.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct DeliveredId {
    /// The server incarnation's start, in wall-clock milliseconds since the epoch.
    pub boot: u64,
    /// The server's message id within that boot.
    pub id: u64,
}

impl DeliveredId {
    /// The causal parent a reply to this message names: `<boot>:<id>`, both in
    /// decimal. It is the tap id's number with the boot in place of the device
    /// (Hermes 7 Oct 08:12), so a tap pairs a reply only within its own boot.
    #[must_use]
    pub fn reply_parent(self) -> MessageId {
        MessageId::new(format!("{}:{}", self.boot, self.id))
    }

    /// The message a reply's parent names, when the parent is exactly the form
    /// [`DeliveredId::reply_parent`] writes; any other form names none. Reads
    /// without allocating, because a tap reads it once per reply.
    #[must_use]
    pub fn from_parent(parent: &MessageId) -> Option<Self> {
        let (boot, id) = parent.as_str().split_once(':')?;
        Some(Self {
            boot: canonical_decimal(boot)?,
            id: canonical_decimal(id)?,
        })
    }
}

/// `digits` as a `u64` when it is the decimal [`u64`]'s `Display` writes: one
/// or more ASCII digits, no sign, no leading zero, within range.
fn canonical_decimal(digits: &str) -> Option<u64> {
    let bytes = digits.as_bytes();
    let canonical = !bytes.is_empty()
        && bytes.iter().all(u8::is_ascii_digit)
        && (bytes.len() == 1 || bytes.first() != Some(&b'0'));
    if canonical { digits.parse().ok() } else { None }
}

/// Status byte prefixing a [`Frame::WorkerRegisterAck`] payload that signals the
/// registration was accepted (no further payload follows).
pub(crate) const WORKER_REGISTER_ACK_ACCEPTED: u8 = 0x00;

/// Status byte prefixing a [`Frame::WorkerRegisterAck`] payload that signals the
/// registration was rejected (a length-prefixed reason string follows).
pub(crate) const WORKER_REGISTER_ACK_REJECTED: u8 = 0x01;

/// Status byte prefixing a [`Frame::LinkOpenAck`] payload that signals the link
/// opened: the responder's handshake message follows.
pub(crate) const LINK_OPEN_ACK_OPENED: u8 = 0x00;

/// Status byte prefixing a [`Frame::LinkOpenAck`] payload that signals the open
/// is held under a Lys draft: the grant and the draft id follow.
pub(crate) const LINK_OPEN_ACK_HELD: u8 = 0x01;

/// Status byte prefixing a [`Frame::LinkOpenAck`] payload that signals the open
/// was refused: the refusal follows as a string.
pub(crate) const LINK_OPEN_ACK_REFUSED: u8 = 0x02;

/// Wire byte of [`LinkEnd::Export`].
pub(crate) const LINK_END_EXPORT: u8 = 0x00;

/// Wire byte of [`LinkEnd::Import`].
pub(crate) const LINK_END_IMPORT: u8 = 0x01;

/// Length of the Ed25519 public key a [`LinkOffer`] carries.
pub const LINK_MACHINE_LEN: usize = 32;

/// Protocol version that carries [`Frame::DisconnectReason`] (ACCESS-005 R5,
/// docs/design/protocol/REVOCATION-CLOSE.md).
///
/// A server writes the reason-bearing close only on a connection that
/// negotiated this version or later; a server enforcing grant revocation
/// refuses `Connect` from a peer that cannot negotiate it. A connection below
/// it only ever sees the payload-free [`Frame::Disconnect`], byte for byte.
pub const REVOCATION_CLOSE_VERSION: ProtocolVersion = ProtocolVersion::new(1, 3);

/// The one body version of [`Frame::DisconnectReason`] this implementation
/// writes and reads; any other is refused by name, never skipped.
pub const DISCONNECT_REASON_BODY_VERSION: u8 = 1;

/// The longest `reason_name` a [`Frame::DisconnectReason`] may carry, in
/// bytes. Every enumerated name is shorter; a longer one is refused.
pub const DISCONNECT_REASON_NAME_MAX: usize = 64;

/// The exact length of a [`Frame::DisconnectReason`] body is
/// `1 + 2 + 4 + name.len()`, so this is its upper bound.
pub const DISCONNECT_REASON_BODY_MAX: usize = 1 + 2 + 4 + DISCONNECT_REASON_NAME_MAX;

/// Why the server closed a connection, carried by [`Frame::DisconnectReason`]
/// as a stable code and its stable name (ACCESS-005 R5).
///
/// The registry is closed: a code or name outside it, or a code paired with
/// another code's name, refuses to decode.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum DisconnectReason {
    /// The right that admitted the connection was withdrawn through a Lys
    /// grant revocation.
    GrantRevoked,
    /// The grant log the connection's binding was decided from was replaced;
    /// its rights cannot be compared against the new log.
    RevocationLogReset,
}

impl DisconnectReason {
    /// Every reason, in code order.
    pub const ALL: [Self; 2] = [Self::GrantRevoked, Self::RevocationLogReset];

    /// The reason's stable wire code.
    #[must_use]
    pub const fn code(self) -> u16 {
        match self {
            Self::GrantRevoked => 0x0001,
            Self::RevocationLogReset => 0x0002,
        }
    }

    /// The reason's stable name: canonical lower-case ASCII.
    #[must_use]
    pub const fn name(self) -> &'static str {
        match self {
            Self::GrantRevoked => "grant_revoked",
            Self::RevocationLogReset => "revocation_log_reset",
        }
    }

    /// The reason a code and name name together, when both are in the
    /// registry and agree; `None` otherwise.
    #[must_use]
    pub fn from_wire(code: u16, name: &str) -> Option<Self> {
        Self::ALL
            .into_iter()
            .find(|reason| reason.code() == code && reason.name() == name)
    }
}

/// Protocol frame categories and their stable wire discriminants.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum FrameType {
    /// Connection request.
    Connect,
    /// Successful connection response.
    ConnectAck,
    /// Failed connection response.
    ConnectError,
    /// Connection close notification.
    Disconnect,
    /// Channel subscription request.
    Subscribe,
    /// Successful subscription response.
    SubscribeAck,
    /// Failed subscription response.
    SubscribeError,
    /// Channel unsubscription request.
    Unsubscribe,
    /// Channel publish request.
    Publish,
    /// Successful publish response.
    PublishAck,
    /// Failed publish response.
    PublishError,
    /// Conversation lifecycle open.
    ConversationOpen,
    /// Conversation message delivery.
    ConversationMessage,
    /// Conversation lifecycle close.
    ConversationClose,
    /// Conversation processing error.
    ConversationError,
    /// In-band backpressure acceptance.
    Accept,
    /// In-band backpressure deferral.
    Defer,
    /// In-band backpressure rejection.
    Reject,
    /// Connection keepalive ping.
    Ping,
    /// Connection keepalive pong.
    Pong,
    /// Server-initiated push of an opaque payload to a connected client.
    Push,
    /// Client-initiated correlated reply to a server push.
    PushReply,
    /// Worker self-registration announcing identity and routing dimensions.
    WorkerRegister,
    /// Server acknowledgement of a worker registration (accepted or rejected).
    WorkerRegisterAck,
    /// Server-to-client delivery of a subscribed channel message.
    Deliver,
    /// Client acknowledgment of a held durable delivery.
    DeliverAck,
    /// A machine opening one end of a link on an application stream.
    LinkOpen,
    /// The server's answer to a [`FrameType::LinkOpen`].
    LinkOpenAck,
    /// One sealed message crossing an open link.
    LinkMessage,
    /// Server-to-client close carrying a stable reason (ACCESS-005 R5),
    /// written only on a connection at [`REVOCATION_CLOSE_VERSION`] or later.
    DisconnectReason,
    /// Forward-compatible frame type not known to this implementation.
    Unknown(u8),
}

impl FrameType {
    /// Return true when this frame type must appear on stream 0.
    #[must_use]
    pub const fn is_control(self) -> bool {
        matches!(
            self,
            Self::Connect
                | Self::ConnectAck
                | Self::ConnectError
                | Self::Disconnect
                | Self::DisconnectReason
                | Self::Ping
                | Self::Pong
                | Self::WorkerRegister
                | Self::WorkerRegisterAck
        )
    }
}

impl From<u8> for FrameType {
    fn from(value: u8) -> Self {
        match value {
            0x01 => Self::Connect,
            0x02 => Self::ConnectAck,
            0x03 => Self::ConnectError,
            0x04 => Self::Disconnect,
            0x05 => Self::Subscribe,
            0x06 => Self::SubscribeAck,
            0x07 => Self::SubscribeError,
            0x08 => Self::Unsubscribe,
            0x09 => Self::Publish,
            0x0A => Self::PublishAck,
            0x0B => Self::PublishError,
            0x0C => Self::ConversationOpen,
            0x0D => Self::ConversationMessage,
            0x0E => Self::ConversationClose,
            0x0F => Self::ConversationError,
            0x10 => Self::Accept,
            0x11 => Self::Defer,
            0x12 => Self::Reject,
            0x13 => Self::Ping,
            0x14 => Self::Pong,
            0x15 => Self::Push,
            0x16 => Self::PushReply,
            0x17 => Self::WorkerRegister,
            0x18 => Self::WorkerRegisterAck,
            0x19 => Self::Deliver,
            // 0x1A is the participant protocol's outer carrier
            // (`liminal_protocol::wire::PARTICIPANT_FRAME_TYPE`) and must stay
            // `Unknown` here, so the link frames sit after it.
            0x1D => Self::LinkOpen,
            0x1E => Self::LinkOpenAck,
            0x1F => Self::LinkMessage,
            0x20 => Self::DisconnectReason,
            0x23 => Self::DeliverAck,
            unknown => Self::Unknown(unknown),
        }
    }
}

impl From<FrameType> for u8 {
    fn from(value: FrameType) -> Self {
        match value {
            FrameType::Connect => 0x01,
            FrameType::ConnectAck => 0x02,
            FrameType::ConnectError => 0x03,
            FrameType::Disconnect => 0x04,
            FrameType::Subscribe => 0x05,
            FrameType::SubscribeAck => 0x06,
            FrameType::SubscribeError => 0x07,
            FrameType::Unsubscribe => 0x08,
            FrameType::Publish => 0x09,
            FrameType::PublishAck => 0x0A,
            FrameType::PublishError => 0x0B,
            FrameType::ConversationOpen => 0x0C,
            FrameType::ConversationMessage => 0x0D,
            FrameType::ConversationClose => 0x0E,
            FrameType::ConversationError => 0x0F,
            FrameType::Accept => 0x10,
            FrameType::Defer => 0x11,
            FrameType::Reject => 0x12,
            FrameType::Ping => 0x13,
            FrameType::Pong => 0x14,
            FrameType::Push => 0x15,
            FrameType::PushReply => 0x16,
            FrameType::WorkerRegister => 0x17,
            FrameType::WorkerRegisterAck => 0x18,
            FrameType::Deliver => 0x19,
            FrameType::DeliverAck => 0x23,
            FrameType::LinkOpen => 0x1D,
            FrameType::LinkOpenAck => 0x1E,
            FrameType::LinkMessage => 0x1F,
            FrameType::DisconnectReason => 0x20,
            FrameType::Unknown(type_id) => type_id,
        }
    }
}

/// Fixed-size frame prefix: type, flags, stream identifier, and payload length.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct FrameHeader {
    /// Frame type read from or written to byte 0.
    pub frame_type: FrameType,
    /// Frame flags read from or written to byte 1.
    pub flags: u8,
    /// Stream identifier read from or written to bytes 2..6.
    pub stream_id: u32,
    /// Payload length read from or written to bytes 6..10.
    pub payload_length: u32,
}

impl FrameHeader {
    /// Serialized header length in bytes.
    pub const WIRE_LEN: usize = HEADER_LEN;
}

/// Typed activity surface committed by a worker build.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct WorkerActivityDescriptor {
    /// Activity type.
    pub name: String,
    /// Canonical JSON Schema accepted by the worker.
    pub input_schema_json: String,
    /// Canonical JSON Schema produced by the worker.
    pub output_schema_json: String,
}

/// Self-describing worker registration carried by [`Frame::WorkerRegister`].
///
/// A worker announces the routing dimensions it serves plus a stable identity so
/// the server can associate the worker with its connection and the application can
/// route work to it. `node` is optional locality (the routing model treats node as
/// an optional dimension); the codec encodes it with a presence byte rather than
/// flattening `None` to an empty string.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct WorkerRegistration {
    /// Further subjects this worker declares it serves, each decided as `serve`
    /// at registration and checked against a site's declared subjects in its trial.
    /// These declarations confer no prefix grant over `task_queue`.
    pub namespaces: Vec<String>,
    /// The exact subject routed to this worker, decided as `serve` before the
    /// namespaces at registration and bound unchanged for each routed reply.
    pub task_queue: String,
    /// Optional node locality; `None` when the worker declares no node affinity.
    pub node: Option<String>,
    /// Activity types this worker can execute.
    pub activity_types: Vec<String>,
    /// Stable worker identity.
    pub identity: String,
    /// Typed action descriptors; empty identifies a pre-contract worker.
    pub activities: Vec<WorkerActivityDescriptor>,
}

/// Outcome of a worker registration, carried by [`Frame::WorkerRegisterAck`].
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum WorkerRegisterOutcome {
    /// The server accepted the registration.
    Accepted,
    /// The server rejected the registration, with a human-readable reason.
    Rejected {
        /// Human-readable rejection reason surfaced to the worker.
        reason: String,
    },
}

/// The end of a link the sender of a [`Frame::LinkOpen`] takes (ACCESS-002
/// R1): the exporting machine sends messages across, the importing one
/// receives them. The server takes the other end.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum LinkEnd {
    /// The sender exports: its messages cross to the server.
    Export,
    /// The sender imports: the server's messages cross to it.
    Import,
}

/// The link a machine offers to open, carried by [`Frame::LinkOpen`].
///
/// `machine` is the sender's Ed25519 public key as Lys recorded it; the
/// server refuses an offer whose key is not the one recorded for the
/// connection's verified holder. `handshake` is the first Noise `KK` message,
/// keyed from that identity's X25519 derivation.
#[derive(Clone, PartialEq, Eq)]
pub struct LinkOffer {
    /// The link's id under the network.
    pub link_id: String,
    /// The end the sender takes.
    pub end: LinkEnd,
    /// The sender's Ed25519 public key as Lys recorded it.
    pub machine: [u8; LINK_MACHINE_LEN],
    /// The sender's first handshake message.
    pub handshake: Vec<u8>,
}

impl core::fmt::Debug for LinkOffer {
    fn fmt(&self, formatter: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
        formatter
            .debug_struct("LinkOffer")
            .field("link_id", &self.link_id)
            .field("end", &self.end)
            .field("machine", &"32 bytes")
            .field("handshake", &self.handshake.len())
            .finish()
    }
}

/// Outcome of a link open, carried by [`Frame::LinkOpenAck`].
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum LinkOpenOutcome {
    /// Lys allowed the open outright and the handshake completed; the
    /// responder's handshake message follows, after which the stream is a
    /// keyed link.
    Opened {
        /// The server's second Noise `KK` message.
        handshake: Vec<u8>,
    },
    /// Lys holds the open under a draft; nothing is keyed until it is
    /// approved and offered again.
    Held {
        /// The grant requiring approval.
        grant: String,
        /// The draft Lys opened.
        draft: String,
    },
    /// The open was refused, by name.
    Refused {
        /// The refusal: the four-field access refusal as JSON, or the
        /// handshake's or binding's own words.
        reason: String,
    },
}

/// A typed protocol frame body plus the header metadata required to encode it.
#[derive(Clone, PartialEq, Eq)]
pub enum Frame {
    /// Connection request carrying a supported version range and opaque auth token.
    Connect {
        flags: u8,
        min_version: ProtocolVersion,
        max_version: ProtocolVersion,
        auth_token: Vec<u8>,
    },
    /// Connection success carrying the negotiated protocol version and server capabilities.
    ConnectAck {
        flags: u8,
        selected_version: ProtocolVersion,
        capabilities: u32,
    },
    /// Connection failure carrying a numeric reason and optional message.
    ConnectError {
        flags: u8,
        reason_code: u16,
        message: Option<String>,
    },
    /// Connection close notification with no payload.
    Disconnect { flags: u8 },
    /// Server close carrying a stable reason (ACCESS-005 R5): the body is a
    /// version byte ([`DISCONNECT_REASON_BODY_VERSION`]), the reason's code
    /// and its name. Written only on a connection at
    /// [`REVOCATION_CLOSE_VERSION`] or later; the payload-free
    /// [`Frame::Disconnect`] is unchanged.
    DisconnectReason { flags: u8, reason: DisconnectReason },
    /// Channel subscription request carrying a channel and accepted schema hashes.
    Subscribe {
        flags: u8,
        stream_id: u32,
        channel: String,
        accepted_schemas: Vec<SchemaId>,
        max_in_flight: u32,
    },
    /// Channel subscription success carrying a subscription id and selected schema.
    SubscribeAck {
        flags: u8,
        stream_id: u32,
        subscription_id: u64,
        selected_schema: SchemaId,
    },
    /// Channel subscription failure carrying a numeric reason and optional message.
    SubscribeError {
        flags: u8,
        stream_id: u32,
        reason_code: u16,
        message: Option<String>,
    },
    /// Channel unsubscription request carrying the subscription id.
    Unsubscribe {
        flags: u8,
        stream_id: u32,
        subscription_id: u64,
    },
    /// Publish request carrying a channel and typed message envelope.
    ///
    /// `idempotency_key` is `Some` only when the [`PUBLISH_IDEMPOTENCY_KEY_FLAG`]
    /// flag bit is set; it is the dedup-on-delivery key the server feeds to its
    /// dedup cache. When `None` (and the flag clear) the frame is byte-identical
    /// on the wire to a pre-13-L1 publish.
    Publish {
        flags: u8,
        stream_id: u32,
        channel: String,
        envelope: MessageEnvelope,
        idempotency_key: Option<String>,
    },
    /// Publish success carrying the accepted message id.
    ///
    /// `durable_position` is `Some` only when the ack answers for a stored row
    /// and the connection negotiated [`PUBLISH_ACK_POSITION_VERSION`]; the codec
    /// then sets [`PUBLISH_DURABLE_POSITION_FLAG`] and writes the position after
    /// `message_id`. The `boot` follows it at [`PUBLISH_ACK_BOOT_VERSION`]
    /// or later. Without either optional field the older layout is unchanged.
    PublishAck {
        flags: u8,
        stream_id: u32,
        message_id: u64,
        durable_position: Option<u64>,
        boot: Option<u64>,
    },
    /// Publish failure carrying a numeric reason and optional message.
    PublishError {
        flags: u8,
        stream_id: u32,
        reason_code: u16,
        message: Option<String>,
    },
    /// Conversation open carrying a conversation id and subject.
    ConversationOpen {
        flags: u8,
        stream_id: u32,
        conversation_id: u64,
        subject: String,
    },
    /// Conversation message carrying a conversation id and typed message envelope.
    ConversationMessage {
        flags: u8,
        stream_id: u32,
        conversation_id: u64,
        envelope: MessageEnvelope,
    },
    /// Conversation close carrying a conversation id and optional reason.
    ConversationClose {
        flags: u8,
        stream_id: u32,
        conversation_id: u64,
        reason_code: Option<u16>,
        message: Option<String>,
    },
    /// Conversation failure carrying a conversation id, numeric reason, and optional message.
    ConversationError {
        flags: u8,
        stream_id: u32,
        conversation_id: u64,
        reason_code: u16,
        message: Option<String>,
    },
    /// Backpressure acceptance for a delivered message.
    Accept {
        flags: u8,
        stream_id: u32,
        referenced_message_id: MessageId,
    },
    /// Backpressure deferral for a buffered message.
    Defer {
        flags: u8,
        stream_id: u32,
        referenced_message_id: MessageId,
        reason: Option<String>,
    },
    /// Backpressure rejection for a shed message.
    Reject {
        flags: u8,
        stream_id: u32,
        referenced_message_id: MessageId,
        reason: Option<String>,
    },
    /// Connection keepalive ping.
    Ping { flags: u8 },
    /// Connection keepalive pong.
    Pong { flags: u8 },
    /// Server-initiated push carrying a correlation id and an opaque payload.
    ///
    /// A server writes this frame to a connected client over the client's existing
    /// connection (server-to-client, the inverse of every other request frame). The
    /// `correlation_id` is the key the server uses to match the client's later
    /// [`Frame::PushReply`] back to this push; the `payload` is opaque application
    /// bytes the server hands the client. This is an application-stream frame, so
    /// `stream_id` is non-zero like a publish or conversation message.
    Push {
        flags: u8,
        stream_id: u32,
        correlation_id: u64,
        payload: Vec<u8>,
    },
    /// Client-initiated correlated reply to a [`Frame::Push`].
    ///
    /// After handling a pushed frame the client writes this back on the same
    /// connection, echoing the push's `correlation_id` so the server can match the
    /// reply to the originating push. The `payload` is the client's opaque answer.
    /// With [`super::PUSH_REPLY_REFUSED_FLAG`] it is a canonical
    /// [`super::WorkerRefusal`]; all remaining flag bits are reserved.
    PushReply {
        flags: u8,
        stream_id: u32,
        correlation_id: u64,
        payload: Vec<u8>,
    },
    /// Worker self-registration over an established connection.
    ///
    /// A worker sends this control frame (stream 0) after the connection
    /// handshake to announce its identity and routing dimensions. The server
    /// associates the registration with the connection's process id and surfaces
    /// it to the application via the connection-notifier hook, then answers with a
    /// [`Frame::WorkerRegisterAck`]. `node` is optional locality and is encoded
    /// with a presence byte, never flattened to an empty string.
    WorkerRegister {
        flags: u8,
        registration: WorkerRegistration,
    },
    /// Server acknowledgement of a [`Frame::WorkerRegister`].
    ///
    /// Carries the registration outcome: [`WorkerRegisterOutcome::Accepted`] when
    /// the server (and any configured notifier) accepted the worker, or
    /// [`WorkerRegisterOutcome::Rejected`] carrying a human-readable reason when it
    /// did not. The acknowledgement is synchronous so a worker never believes it is
    /// registered when the application rejected it.
    WorkerRegisterAck {
        flags: u8,
        outcome: WorkerRegisterOutcome,
    },
    /// Server-to-client delivery of a message on a subscription the client opened.
    ///
    /// The server writes this frame on the subscription's own `stream_id` (the
    /// non-zero application stream negotiated by the client's `Subscribe`). It is
    /// the only server-originated delivery frame, and it is only ever sent on a
    /// subscription the client itself opened, so the forward-compatibility rules
    /// for unknown frames never surprise an old client with one. `delivery_seq` is
    /// a per-subscription monotonic counter starting at 1. Durable acknowledgments
    /// name the store `position`, independent of this counter. The `envelope` reuses the same
    /// [`MessageEnvelope`] serialization as publish and conversation frames.
    ///
    /// `message` is `Some` only when the server knows the message's id and boot
    /// and the connection negotiated [`DELIVER_MESSAGE_ID_VERSION`]; the codec
    /// then sets [`DELIVER_MESSAGE_ID_FLAG`] and writes the pair after the
    /// envelope. At [`INBOX_ACK_VERSION`], a consumer's `position` follows the
    /// pair (or envelope when no pair is present), behind [`DELIVER_POSITION_FLAG`].
    /// Without either optional field the older layout is unchanged.
    Deliver {
        flags: u8,
        stream_id: u32,
        delivery_seq: u64,
        envelope: MessageEnvelope,
        message: Option<DeliveredId>,
        /// Store position held until acknowledged at protocol 1.5.
        position: Option<u64>,
    },
    /// Client acknowledgment on the subscription's application stream.
    DeliverAck {
        flags: u8,
        stream_id: u32,
        position: u64,
    },
    /// A machine opening one end of a link on this application stream
    /// (ACCESS-002 R1). The server decides the open with Lys, live, for the
    /// grant the offered end names on the link, proves the offered machine
    /// key is the verified holder's recorded key, and answers the handshake.
    LinkOpen {
        flags: u8,
        stream_id: u32,
        offer: LinkOffer,
    },
    /// The server's answer to a [`Frame::LinkOpen`] on the same stream.
    LinkOpenAck {
        flags: u8,
        stream_id: u32,
        outcome: LinkOpenOutcome,
    },
    /// One message crossing an open link on its stream: the subject it is
    /// published on, and the message envelope sealed by the link's transport.
    /// The receiving end decides the crossing from both machines' passes
    /// before any effect.
    LinkMessage {
        flags: u8,
        stream_id: u32,
        subject: String,
        sealed: Vec<u8>,
    },
    /// Forward-compatible frame preserved after length-delimited skipping.
    Unknown {
        type_id: u8,
        flags: u8,
        stream_id: u32,
        payload: Vec<u8>,
    },
}

impl Frame {
    /// Construct a ping frame, enforcing the stream-0 control-frame invariant.
    ///
    /// # Errors
    ///
    /// Returns [`ProtocolError::InvalidStream`] when `stream_id` is not zero.
    pub fn new_ping(stream_id: u32) -> Result<Self, ProtocolError> {
        validate_stream(FrameType::Ping, stream_id)?;
        Ok(Self::Ping { flags: 0 })
    }

    /// Construct a publish frame, enforcing the non-zero application-stream invariant.
    ///
    /// # Errors
    ///
    /// Returns [`ProtocolError::InvalidStream`] when `stream_id` is zero.
    pub fn new_publish(
        stream_id: u32,
        channel: impl Into<String>,
        envelope: MessageEnvelope,
    ) -> Result<Self, ProtocolError> {
        validate_stream(FrameType::Publish, stream_id)?;
        Ok(Self::Publish {
            flags: 0,
            stream_id,
            channel: channel.into(),
            envelope,
            idempotency_key: None,
        })
    }

    /// Construct a publish frame carrying an idempotency key for dedup-on-delivery.
    ///
    /// The returned frame has [`PUBLISH_IDEMPOTENCY_KEY_FLAG`] set and serializes
    /// the trailing key field, so the server consults its dedup cache for this key.
    ///
    /// # Errors
    ///
    /// Returns [`ProtocolError::InvalidStream`] when `stream_id` is zero.
    pub fn new_publish_with_idempotency_key(
        stream_id: u32,
        channel: impl Into<String>,
        envelope: MessageEnvelope,
        idempotency_key: impl Into<String>,
    ) -> Result<Self, ProtocolError> {
        validate_stream(FrameType::Publish, stream_id)?;
        Ok(Self::Publish {
            flags: PUBLISH_IDEMPOTENCY_KEY_FLAG,
            stream_id,
            channel: channel.into(),
            envelope,
            idempotency_key: Some(idempotency_key.into()),
        })
    }

    /// Construct a server-to-client push frame, enforcing the non-zero
    /// application-stream invariant.
    ///
    /// # Errors
    ///
    /// Returns [`ProtocolError::InvalidStream`] when `stream_id` is zero.
    pub fn new_push(
        stream_id: u32,
        correlation_id: u64,
        payload: Vec<u8>,
    ) -> Result<Self, ProtocolError> {
        validate_stream(FrameType::Push, stream_id)?;
        Ok(Self::Push {
            flags: 0,
            stream_id,
            correlation_id,
            payload,
        })
    }

    /// Construct a server-to-client delivery frame on a subscription's stream,
    /// enforcing the non-zero application-stream invariant.
    ///
    /// # Errors
    ///
    /// Returns [`ProtocolError::InvalidStream`] when `stream_id` is zero.
    pub fn new_deliver(
        stream_id: u32,
        delivery_seq: u64,
        envelope: MessageEnvelope,
    ) -> Result<Self, ProtocolError> {
        validate_stream(FrameType::Deliver, stream_id)?;
        Ok(Self::Deliver {
            flags: 0,
            stream_id,
            delivery_seq,
            envelope,
            message: None,
            position: None,
        })
    }

    /// Construct a client-to-server push reply frame, echoing the correlation id of
    /// the originating push, and enforcing the non-zero application-stream invariant.
    ///
    /// # Errors
    ///
    /// Returns [`ProtocolError::InvalidStream`] when `stream_id` is zero.
    pub fn new_push_reply(
        stream_id: u32,
        correlation_id: u64,
        payload: Vec<u8>,
    ) -> Result<Self, ProtocolError> {
        validate_stream(FrameType::PushReply, stream_id)?;
        Ok(Self::PushReply {
            flags: 0,
            stream_id,
            correlation_id,
            payload,
        })
    }

    /// Constructs a named refusal on the existing correlated reply frame.
    ///
    /// # Errors
    /// An invalid application stream or canonical-record encoding failure.
    pub fn new_push_refusal(
        stream_id: u32,
        correlation_id: u64,
        refusal: &super::WorkerRefusal,
    ) -> Result<Self, ProtocolError> {
        validate_stream(FrameType::PushReply, stream_id)?;
        Ok(Self::PushReply {
            flags: super::PUSH_REPLY_REFUSED_FLAG,
            stream_id,
            correlation_id,
            payload: refusal.canonical_bytes()?,
        })
    }

    /// Return the frame type represented by this frame body.
    #[must_use]
    pub const fn frame_type(&self) -> FrameType {
        match self {
            Self::Connect { .. } => FrameType::Connect,
            Self::ConnectAck { .. } => FrameType::ConnectAck,
            Self::ConnectError { .. } => FrameType::ConnectError,
            Self::Disconnect { .. } => FrameType::Disconnect,
            Self::DisconnectReason { .. } => FrameType::DisconnectReason,
            Self::Subscribe { .. } => FrameType::Subscribe,
            Self::SubscribeAck { .. } => FrameType::SubscribeAck,
            Self::SubscribeError { .. } => FrameType::SubscribeError,
            Self::Unsubscribe { .. } => FrameType::Unsubscribe,
            Self::Publish { .. } => FrameType::Publish,
            Self::PublishAck { .. } => FrameType::PublishAck,
            Self::PublishError { .. } => FrameType::PublishError,
            Self::ConversationOpen { .. } => FrameType::ConversationOpen,
            Self::ConversationMessage { .. } => FrameType::ConversationMessage,
            Self::ConversationClose { .. } => FrameType::ConversationClose,
            Self::ConversationError { .. } => FrameType::ConversationError,
            Self::Accept { .. } => FrameType::Accept,
            Self::Defer { .. } => FrameType::Defer,
            Self::Reject { .. } => FrameType::Reject,
            Self::Ping { .. } => FrameType::Ping,
            Self::Pong { .. } => FrameType::Pong,
            Self::Push { .. } => FrameType::Push,
            Self::PushReply { .. } => FrameType::PushReply,
            Self::WorkerRegister { .. } => FrameType::WorkerRegister,
            Self::WorkerRegisterAck { .. } => FrameType::WorkerRegisterAck,
            Self::Deliver { .. } => FrameType::Deliver,
            Self::DeliverAck { .. } => FrameType::DeliverAck,
            Self::LinkOpen { .. } => FrameType::LinkOpen,
            Self::LinkOpenAck { .. } => FrameType::LinkOpenAck,
            Self::LinkMessage { .. } => FrameType::LinkMessage,
            Self::Unknown { type_id, .. } => FrameType::Unknown(*type_id),
        }
    }

    /// Return the frame flags stored in the fixed header.
    #[must_use]
    pub const fn flags(&self) -> u8 {
        match self {
            Self::Connect { flags, .. }
            | Self::ConnectAck { flags, .. }
            | Self::ConnectError { flags, .. }
            | Self::Disconnect { flags, .. }
            | Self::DisconnectReason { flags, .. }
            | Self::Subscribe { flags, .. }
            | Self::SubscribeAck { flags, .. }
            | Self::SubscribeError { flags, .. }
            | Self::Unsubscribe { flags, .. }
            | Self::Publish { flags, .. }
            | Self::PublishAck { flags, .. }
            | Self::PublishError { flags, .. }
            | Self::ConversationOpen { flags, .. }
            | Self::ConversationMessage { flags, .. }
            | Self::ConversationClose { flags, .. }
            | Self::ConversationError { flags, .. }
            | Self::Accept { flags, .. }
            | Self::Defer { flags, .. }
            | Self::Reject { flags, .. }
            | Self::Ping { flags }
            | Self::Pong { flags }
            | Self::Push { flags, .. }
            | Self::PushReply { flags, .. }
            | Self::WorkerRegister { flags, .. }
            | Self::WorkerRegisterAck { flags, .. }
            | Self::DeliverAck { flags, .. }
            | Self::Deliver { flags, .. }
            | Self::LinkOpen { flags, .. }
            | Self::LinkOpenAck { flags, .. }
            | Self::LinkMessage { flags, .. }
            | Self::Unknown { flags, .. } => *flags,
        }
    }

    /// Return the flags byte the codec writes into the fixed header.
    ///
    /// This is [`Frame::flags`] except on a [`Frame::PublishAck`], whose
    /// [`PUBLISH_DURABLE_POSITION_FLAG`] bit is set exactly when the ack carries
    /// a durable position and whose [`PUBLISH_ACK_BOOT_FLAG`] tracks `boot`,
    /// and a [`Frame::Deliver`], whose
    /// [`DELIVER_MESSAGE_ID_FLAG`] bit is set exactly when it carries a
    /// [`DeliveredId`], so the bit and the trailing field never disagree.
    #[must_use]
    pub const fn wire_flags(&self) -> u8 {
        match self {
            Self::PublishAck {
                flags,
                durable_position,
                boot,
                ..
            } => {
                let position_flag = if durable_position.is_some() {
                    PUBLISH_DURABLE_POSITION_FLAG
                } else {
                    0
                };
                let boot_flag = if boot.is_some() {
                    PUBLISH_ACK_BOOT_FLAG
                } else {
                    0
                };
                (*flags & !(PUBLISH_DURABLE_POSITION_FLAG | PUBLISH_ACK_BOOT_FLAG))
                    | position_flag
                    | boot_flag
            }
            Self::Deliver {
                flags,
                message,
                position,
                ..
            } => {
                let identity = if message.is_some() {
                    DELIVER_MESSAGE_ID_FLAG
                } else {
                    0
                };
                let positioned = if position.is_some() {
                    DELIVER_POSITION_FLAG
                } else {
                    0
                };
                (*flags & !(DELIVER_MESSAGE_ID_FLAG | DELIVER_POSITION_FLAG))
                    | identity
                    | positioned
            }
            _ => self.flags(),
        }
    }

    /// Return the stream id stored in the fixed header.
    #[must_use]
    pub const fn stream_id(&self) -> u32 {
        match self {
            Self::Connect { .. }
            | Self::ConnectAck { .. }
            | Self::ConnectError { .. }
            | Self::Disconnect { .. }
            | Self::DisconnectReason { .. }
            | Self::Ping { .. }
            | Self::Pong { .. }
            | Self::WorkerRegister { .. }
            | Self::WorkerRegisterAck { .. } => 0,
            Self::Subscribe { stream_id, .. }
            | Self::SubscribeAck { stream_id, .. }
            | Self::SubscribeError { stream_id, .. }
            | Self::Unsubscribe { stream_id, .. }
            | Self::Publish { stream_id, .. }
            | Self::PublishAck { stream_id, .. }
            | Self::PublishError { stream_id, .. }
            | Self::ConversationOpen { stream_id, .. }
            | Self::ConversationMessage { stream_id, .. }
            | Self::ConversationClose { stream_id, .. }
            | Self::ConversationError { stream_id, .. }
            | Self::Accept { stream_id, .. }
            | Self::Defer { stream_id, .. }
            | Self::Reject { stream_id, .. }
            | Self::Push { stream_id, .. }
            | Self::PushReply { stream_id, .. }
            | Self::DeliverAck { stream_id, .. }
            | Self::Deliver { stream_id, .. }
            | Self::LinkOpen { stream_id, .. }
            | Self::LinkOpenAck { stream_id, .. }
            | Self::LinkMessage { stream_id, .. }
            | Self::Unknown { stream_id, .. } => *stream_id,
        }
    }

    /// Validate the stream invariant for this frame.
    pub(crate) fn validate(&self) -> Result<(), ProtocolError> {
        validate_stream(self.frame_type(), self.stream_id())?;

        if let Self::PushReply { flags, .. } = self {
            if flags & !super::PUSH_REPLY_REFUSED_FLAG != 0 {
                return Err(ProtocolError::codec("push_reply_reserved_flags"));
            }
        }

        if let Self::Subscribe { max_in_flight, .. } = self {
            if *max_in_flight == 0 {
                return Err(ProtocolError::codec(
                    "max_in_flight must be greater than zero",
                ));
            }
        }

        Ok(())
    }
}

/// Validate stream placement for a known frame type.
///
/// # Errors
///
/// Returns [`ProtocolError::InvalidStream`] when `stream_id` is invalid for
/// `frame_type`.
pub fn validate_stream(frame_type: FrameType, stream_id: u32) -> Result<(), ProtocolError> {
    if matches!(frame_type, FrameType::Unknown(_)) {
        return Ok(());
    }

    let valid = if frame_type.is_control() {
        stream_id == 0
    } else {
        stream_id >= 1
    };

    if valid {
        Ok(())
    } else {
        Err(ProtocolError::invalid_stream(frame_type, stream_id))
    }
}

#[cfg(test)]
mod tests;

#[cfg(test)]
mod channel_fixture_tests;