whatsapp_rust/send/

mod.rs

//! Outgoing message pipeline.
//!
//! Cost model, clock reads: one per send operation, sampled as `SendInstant`
//! and carried to every stamp. A new timestamp here should take that instant
//! rather than read again, or the path silently accumulates reads the way it
//! had accumulated four.

use crate::client::Client;
use crate::types::message::EditAttribute;
use anyhow::anyhow;
use log::debug;
use wacore::libsignal::protocol::SignalProtocolError;
use wacore::send::StanzaType;
use wacore::types::jid::JidExt;
use wacore::types::message::AddressingMode;
#[cfg(test)]
use wacore_binary::DeviceKey;
use wacore_binary::Node;
use wacore_binary::builder::NodeBuilder;
use wacore_binary::{Jid, JidExt as _, Server};
use waproto::whatsapp as wa;

use crate::client::ClientError;
use crate::features::GroupError;
use crate::request::IqError;
use thiserror::Error;

mod actions;
mod tctoken_lifecycle;

/// Error returned by the message send path ([`Client::send_message`],
/// [`Client::send_text`], [`Client::forward_message`], reactions, edits,
/// revokes, pins, polls, events, comments, status) and the bot
/// [`crate::bot::MessageContext`] helpers.
///
/// Wraps the shared [`ClientError`] (transport/connection/IQ) and surfaces the
/// actionable send-time failure modes explicitly. `Internal` is the last-resort
/// catch-all for crypto/encoding paths that still thread `anyhow` internally.
#[derive(Debug, Error)]
#[non_exhaustive]
pub enum SendError {
    /// Connection/transport/IQ failure (embeds the shared base error).
    // No `#[from]`: the manual `From<ClientError>` impl flattens a bare `?` so
    // `NotLoggedIn`/`Iq` stay matchable instead of nesting under `Client(..)`.
    #[error("{0}")]
    Client(#[source] ClientError),
    /// The client has no PN/LID identity yet (not paired / mid LID migration).
    #[error("client is not logged in")]
    NotLoggedIn,
    /// An IQ issued as part of the send (e.g. a group-info query) failed.
    #[error("IQ request failed: {0}")]
    Iq(#[from] IqError),
    /// The recipient JID or send arguments are invalid for this operation
    /// (e.g. a newsletter JID on the E2E path, an empty status recipient list).
    #[error("invalid send request: {0}")]
    InvalidRequest(String),
    /// Catch-all for internal send failures (Signal encrypt, protobuf, group
    /// resolution) that have no dedicated variant yet. `Display` forwards to
    /// the inner error while `source()` still exposes it for downcast.
    #[error("{0}")]
    Internal(#[from] anyhow::Error),
}

impl SendError {
    /// Map an `anyhow::Error` bubbled up from a helper that still threads
    /// `anyhow` (e.g. `send_message_impl`, `require_pn`) into a typed
    /// `SendError`, recovering the concrete [`ClientError`]. Without this the
    /// blanket `#[from] anyhow::Error` would funnel a logged-out
    /// `ClientError::NotLoggedIn` into the un-matchable `Internal` catch-all.
    pub(crate) fn from_anyhow(err: anyhow::Error) -> Self {
        // A validation deeper in the pipeline may already be a typed `SendError`
        // (e.g. send_message_impl's newsletter/status guards); recover it so it
        // stays matchable instead of collapsing into `Internal`.
        let err = match err.downcast::<SendError>() {
            Ok(send) => return send,
            Err(other) => other,
        };
        // A group-metadata IQ in the send path (e.g. query_info) bubbles up as
        // `GroupError`; flatten it before the `ClientError` check so an IQ
        // failure surfaces as `SendError::Iq`, not the `Internal` catch-all.
        let err = match err.downcast::<GroupError>() {
            Ok(group) => return group.into(),
            Err(other) => other,
        };
        match err.downcast::<ClientError>() {
            Ok(client) => client.into(),
            Err(other) => match other.downcast::<IqError>() {
                Ok(iq) => SendError::Iq(iq),
                Err(other) => SendError::Internal(other),
            },
        }
    }
}

impl From<ClientError> for SendError {
    fn from(err: ClientError) -> Self {
        match err {
            ClientError::NotLoggedIn => SendError::NotLoggedIn,
            ClientError::Iq(iq) => SendError::Iq(iq),
            client => SendError::Client(client),
        }
    }
}

impl From<GroupError> for SendError {
    fn from(err: GroupError) -> Self {
        match err {
            GroupError::Iq(iq) => SendError::Iq(iq),
            GroupError::InvalidRequest(msg) => SendError::InvalidRequest(msg),
            GroupError::Internal(e) => SendError::from_anyhow(e),
            // No dedicated variant for MEX mutations or description conflicts;
            // preserve the full typed error as the `Internal` source so its
            // Display/source chain survives.
            group @ (GroupError::Mex(_) | GroupError::DescriptionConflict) => {
                SendError::Internal(group.into())
            }
        }
    }
}

/// Returns a `GroupInfo` whose participant list is guaranteed to contain our own
/// sending JID, without deep-cloning the shared (cached) metadata in the common
/// case where the server's participant list already includes us.
fn ensure_self_in_group(
    info: std::sync::Arc<wacore::client::context::GroupInfo>,
    own_sending_jid: &Jid,
) -> std::sync::Arc<wacore::client::context::GroupInfo> {
    if info
        .participants
        .iter()
        .any(|participant| participant.is_same_user_as(own_sending_jid))
    {
        info
    } else {
        let mut owned = (*info).clone();
        owned.participants.push(own_sending_jid.to_non_ad());
        std::sync::Arc::new(owned)
    }
}

/// SKDM update data — only populated for group sends, deferred until after
/// send_node(). This matches WhatsApp Web which only calls markHasSenderKey()
/// after server ACK.
struct SkdmUpdate {
    to_str: String,
    devices: Vec<Jid>,
    stale_users: Vec<String>,
}

/// One send branch's result: the wire stanza plus the state the shared
/// epilogue of `send_message_impl` consumes. Each branch runs as its own
/// boxed future, so a DM send never pays for the group branch's frame.
struct SendBranchOutput {
    node: Node,
    /// Generated `MessageContextInfo.message_secret`, persisted after send_node.
    msg_secret: Option<[u8; 32]>,
    /// Group sends: the identity (LID or PN) the secret must be keyed under,
    /// matching what `<meta target_sender_jid>` echoes back.
    group_sender_identity: Option<Jid>,
    skdm_update: Option<SkdmUpdate>,
    /// Single-flight for cold group sends: held from SKDM target resolution
    /// through `update_sender_key_devices` in the epilogue so a concurrent
    /// cold send re-resolves against the winner's warm marking.
    distribution_guard: Option<async_lock::MutexGuardArc<()>>,
    issue_tc_token_after_send: bool,
    dm_phash: Option<wacore_binary::CompactString>,
}

struct GroupBranchRequest<'a> {
    to: Jid,
    message: &'a wa::Message,
    request_id: &'a str,
    force_key_distribution: bool,
    edit: Option<EditAttribute>,
    extra_stanza_nodes: &'a [Node],
    group_metadata_freshness: crate::cache::Freshness,
    device_freshness: crate::cache::Freshness,
    borrowed_message_id: bool,
}

struct DmBranchRequest<'a> {
    to: Jid,
    message: &'a wa::Message,
    request_id: &'a str,
    sent_at: SendInstant,
    edit: Option<EditAttribute>,
    extra_stanza_nodes: Vec<Node>,
    is_status_addon: bool,
    device_freshness: crate::cache::Freshness,
    borrowed_message_id: bool,
}

enum GroupDeviceSnapshot {
    Owned(wacore::send::ResolvedGroupDevices),
    Shared(std::sync::Arc<wacore::send::ResolvedGroupDevices>),
}

impl AsRef<wacore::send::ResolvedGroupDevices> for GroupDeviceSnapshot {
    fn as_ref(&self) -> &wacore::send::ResolvedGroupDevices {
        match self {
            Self::Owned(devices) => devices,
            Self::Shared(devices) => devices,
        }
    }
}

/// Keep each branch future out of the shared send frame. In tracing builds the
/// dedicated span lets allocation profilers distinguish this deliberate box
/// from work performed while polling the selected branch.
#[inline]
#[cfg_attr(
    feature = "tracing",
    tracing::instrument(
        name = "wa.send.branch_box",
        level = "debug",
        skip_all,
        fields(future_bytes = size_of::<F>())
    )
)]
fn box_send_branch<F>(future: F) -> std::pin::Pin<Box<F>>
where
    F: Future,
{
    Box::pin(future)
}

/// True when every SKDM target belongs to our own account (PN or LID user).
/// Own devices are never memoized warm (WA Web's `!isMeDevice` guard on
/// `markHasSenderKey`), so an own-only `needs` set is the permanent
/// warm-send steady state — not a cold-group signal.
fn skdm_needs_only_own_devices(needs: &[Jid], own_pn: Option<&Jid>, own_lid: Option<&Jid>) -> bool {
    !needs.is_empty()
        && needs.iter().all(|j| {
            own_pn.is_some_and(|p| j.is_same_user_as(p))
                || own_lid.is_some_and(|l| j.is_same_user_as(l))
        })
}

const RESERVED_EXTRA_STANZA_CHILDREN: &[&str] =
    &["enc", "participants", "device-identity", "plaintext"];

fn validate_extra_stanza_nodes(nodes: &[Node]) -> Result<(), SendError> {
    if let Some(node) = nodes.iter().find(|node| {
        RESERVED_EXTRA_STANZA_CHILDREN
            .iter()
            .any(|reserved| node.tag == *reserved)
    }) {
        return Err(SendError::InvalidRequest(format!(
            "extra stanza child <{}> is reserved by the send pipeline",
            node.tag
        )));
    }
    Ok(())
}

impl SendBranchOutput {
    fn stanza_only(node: Node) -> Self {
        Self {
            node,
            msg_secret: None,
            group_sender_identity: None,
            skdm_update: None,
            distribution_guard: None,
            issue_tc_token_after_send: false,
            dm_phash: None,
        }
    }
}

/// Options for [`Client::send_message_with_options`].
///
/// Start from [`SendOptions::default`] and chain the `with_*` setters; the
/// struct is `#[non_exhaustive]` so new knobs can be added without breaking
/// consumers.
///
/// ```
/// # use whatsapp_rust::send::SendOptions;
/// let options = SendOptions::default().with_message_id("3EB0ABCDEF");
/// ```
#[derive(Debug, Clone, Default)]
#[non_exhaustive]
pub struct SendOptions {
    /// Override the auto-generated message ID.
    /// Useful for resending a failed message with the same ID or idempotency.
    pub message_id: Option<String>,
    /// Extra XML child nodes on the message stanza.
    pub extra_stanza_nodes: Vec<Node>,
    /// Ephemeral duration in seconds. Sets `contextInfo.expiration` on the
    /// message (WA Web `EProtoGenerator.js:183` parity).
    /// Common values: 86400 (24h), 604800 (7d), 7776000 (90d).
    pub ephemeral_expiration: Option<u32>,
    /// Force the `<message type="...">` attribute instead of deriving it from
    /// content. Escape hatch for a type the classifier can't infer.
    pub stanza_type_override: Option<StanzaType>,
    /// Freshness policy for group metadata used by this send.
    pub group_metadata_freshness: crate::cache::Freshness,
    /// Freshness policy for recipient device lists used by this send.
    pub device_freshness: crate::cache::Freshness,
}

impl SendOptions {
    /// See [`SendOptions::message_id`].
    #[must_use]
    pub fn with_message_id(mut self, message_id: impl Into<String>) -> Self {
        self.message_id = Some(message_id.into());
        self
    }

    /// See [`SendOptions::extra_stanza_nodes`].
    #[must_use]
    pub fn with_extra_stanza_nodes(mut self, nodes: Vec<Node>) -> Self {
        self.extra_stanza_nodes = nodes;
        self
    }

    /// See [`SendOptions::ephemeral_expiration`].
    #[must_use]
    pub fn with_ephemeral_expiration(mut self, seconds: u32) -> Self {
        self.ephemeral_expiration = Some(seconds);
        self
    }

    /// See [`SendOptions::stanza_type_override`].
    #[must_use]
    pub fn with_stanza_type_override(mut self, stanza_type: StanzaType) -> Self {
        self.stanza_type_override = Some(stanza_type);
        self
    }

    /// See [`SendOptions::group_metadata_freshness`].
    #[must_use]
    pub fn with_group_metadata_freshness(mut self, freshness: crate::cache::Freshness) -> Self {
        self.group_metadata_freshness = freshness;
        self
    }

    /// See [`SendOptions::device_freshness`].
    #[must_use]
    pub fn with_device_freshness(mut self, freshness: crate::cache::Freshness) -> Self {
        self.device_freshness = freshness;
        self
    }
}

/// Options for [`Client::edit_message_with_options`].
///
/// Start from [`EditOptions::default`] and chain the `with_*` setters; the
/// struct is `#[non_exhaustive]` so new knobs can be added without breaking
/// consumers.
#[derive(Debug, Clone, Default)]
#[non_exhaustive]
pub struct EditOptions {
    /// Override the outer stanza id (default: a fresh id, like
    /// [`Client::edit_message`]). Pinning it to an **existing** message's id is
    /// a best-effort, side-effect-aware operation:
    /// - No id-keyed local state is bound to the borrowed id — the edit does not
    ///   persist an outbound message secret or a retry-cache entry under it, so
    ///   the original message's secret and retry content are left intact.
    /// - Whether the wire-level collision is honored is server- and
    ///   client-dependent (the server may dedupe against the outer id), so treat
    ///   the visible outcome as non-guaranteed.
    pub stanza_id: Option<String>,
}

impl EditOptions {
    /// See [`EditOptions::stanza_id`].
    #[must_use]
    pub fn with_stanza_id(mut self, stanza_id: impl Into<String>) -> Self {
        self.stanza_id = Some(stanza_id.into());
        self
    }
}

/// The wall-clock second one send operation is stamped with.
///
/// Sampled once where the operation starts and carried down, so the message id,
/// the biz node, the privacy-token decision and the outbound message secret
/// describe one instant instead of four reads that can straddle a second
/// boundary, on a path where a clock read is not always cheap.
#[derive(Debug, Clone, Copy)]
pub(crate) struct SendInstant(i64);

impl SendInstant {
    pub(crate) fn now() -> Self {
        Self(wacore::time::now_secs())
    }

    pub(crate) fn unix_secs(self) -> i64 {
        self.0
    }

    /// Saturated at 0 for the encodings that carry unsigned seconds.
    pub(crate) fn unix_secs_u64(self) -> u64 {
        self.0.max(0) as u64
    }
}

#[derive(Default)]
pub(crate) struct SendPipelineOptions<'a> {
    /// Instant this operation is stamped with, when the caller already sampled
    /// one. `None` makes [`Client::send_message_impl`] sample its own.
    pub(crate) sent_at: Option<SendInstant>,
    /// Borrowed on purpose: the caller that already owns an id (because it
    /// returns it, or stamped state with it) lends it for the whole send
    /// instead of handing over a copy.
    pub(crate) request_id: Option<&'a str>,
    pub(crate) peer: bool,
    pub(crate) force_key_distribution: bool,
    pub(crate) edit: Option<EditAttribute>,
    pub(crate) extra_stanza_nodes: Vec<Node>,
    pub(crate) stanza_type: Option<StanzaType>,
    pub(crate) group_metadata_freshness: crate::cache::Freshness,
    pub(crate) device_freshness: crate::cache::Freshness,
    /// The outer stanza id is borrowed from another message (caller-forced
    /// `request_id`), so id-keyed state must NOT be bound to it: skip
    /// `add_recent_message` (retry cache) and `persist_outbound_msg_secret`.
    /// Without this, the borrowed id clobbers the original message's retry
    /// content and outbound secret.
    pub(crate) borrowed_message_id: bool,
}

/// Result of a successfully sent message.
#[derive(Debug, Clone, PartialEq, Eq)]
#[non_exhaustive]
pub struct SendResult {
    pub message_id: String,
    pub to: Jid,
}

impl SendResult {
    /// `participant` is `None` -- only valid for the sender's own messages.
    pub fn message_key(&self) -> wa::MessageKey {
        wa::MessageKey {
            remote_jid: Some(self.to.to_string()),
            from_me: Some(true),
            id: Some(self.message_id.clone()),
            participant: None,
        }
    }
}

/// Duration for pinned messages. Default is 7 days (matches WA Web).
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
#[non_exhaustive]
pub enum PinDuration {
    Hours24,
    #[default]
    Days7,
    Days30,
}

impl PinDuration {
    fn as_secs(self) -> u32 {
        match self {
            Self::Hours24 => 86_400,
            Self::Days7 => 604_800,
            Self::Days30 => 2_592_000,
        }
    }
}

/// Specifies who is revoking (deleting) the message.
#[derive(Debug, Clone, PartialEq, Eq, Default)]
#[non_exhaustive]
pub enum RevokeType {
    /// The message sender deleting their own message.
    #[default]
    Sender,
    /// A group admin deleting another user's message.
    /// `original_sender` is the JID of the user who sent the message being deleted.
    Admin { original_sender: Jid },
}

/// Derive stanza-level edit attribute and meta node from message content.
///
/// The `edit` attribute and the `<meta>` child are independent in WA Web: the
/// edit attribute comes from `editAttribute(msg, subtype)` and the meta node
/// from `genMetaNode(...)`. A message can carry both (e.g. a poll vote sets
/// `polltype=vote` meta; an event edit sets both `event_type=edit` meta and
/// `edit="1"` attribute).
pub(crate) fn infer_stanza_metadata(msg: &wa::Message) -> (Option<EditAttribute>, Option<Node>) {
    use wacore::proto_helpers::MessageExt;
    let edit = EditAttribute::infer_from_message(msg);

    // genMetaNode builds a single <meta> carrying every applicable attr together,
    // so accumulate onto one node instead of emitting at most one attr.
    let mut meta = NodeBuilder::new("meta");
    let mut has_attr = false;

    if msg.poll_creation_message.is_set()
        || msg.poll_creation_message_v2.is_set()
        || msg.poll_creation_message_v3.is_set()
    {
        meta = meta.attr("polltype", "creation");
        has_attr = true;
    } else if let Some(poll_update) = msg.poll_update_message.as_option()
        && poll_update.vote.is_set()
    {
        meta = meta.attr("polltype", "vote");
        has_attr = true;
        // TODO: polltype="result_snapshot" for poll_result_snapshot_message (gated behind AB flag)
    } else if msg.event_message.is_set() {
        meta = meta.attr("event_type", "creation");
        has_attr = true;
    } else if msg.enc_event_response_message.is_set() {
        meta = meta.attr("event_type", "response");
        has_attr = true;
    } else if let Some(sec) = msg.secret_encrypted_message.as_option()
        && sec.secret_enc_type
            == Some(wa::message::secret_encrypted_message::SecretEncType::EventEdit)
    {
        meta = meta.attr("event_type", "edit");
        has_attr = true;
    } else if let Some(ml) = msg
        .protocol_message
        .as_option()
        .and_then(|pm| pm.member_label.as_option())
    {
        // genMetaNode (MsgMetaNode `d`/`p`): a member_label protocol message carries
        // appdata="member_tag" and tag_reason="user_delete" when the label is cleared
        // (empty/absent), "user_update" otherwise.
        let tag_reason = if ml.label.as_deref().unwrap_or("").is_empty() {
            "user_delete"
        } else {
            "user_update"
        };
        meta = meta
            .attr("appdata", "member_tag")
            .attr("tag_reason", tag_reason);
        has_attr = true;
    }

    // genMetaNode: `view_once="true"` whenever the media is view-once (wrapper or
    // inline flag). Detection covers both via MessageExt::is_view_once.
    if msg.is_view_once() {
        meta = meta.attr("view_once", "true");
        has_attr = true;
    }

    (edit, has_attr.then(|| meta.build()))
}

fn validate_status_message_id(
    message: &wa::Message,
    outer_id: Option<&str>,
) -> Result<(), SendError> {
    let Some(outer_id) = outer_id else {
        return Ok(());
    };
    if outer_id.is_empty() {
        return Err(SendError::InvalidRequest(
            "status message ID must not be empty".into(),
        ));
    }
    if wacore::send::status_revoke_target_id(message) == Some(outer_id) {
        return Err(SendError::InvalidRequest(
            "status revoke stanza ID must differ from the revoked message ID".into(),
        ));
    }
    Ok(())
}

/// Offset subtracted from the current unix timestamp to produce the
/// `privacy_mode_ts` attr value on a `<biz>` stanza. Empirically confirmed
/// against live WhatsApp servers.
const BIZ_PRIVACY_MODE_TS_OFFSET: u64 = 77_980_457;

enum BizCategory<'a> {
    /// `<biz actual_actors host_storage privacy_mode_ts native_flow_name=X/>` — no children.
    /// The one shape WA Web also emits attrs-only: `createFanoutMsgStanza`
    /// builds exactly these four attrs (and never a child) when the peer
    /// contact carries a `privacyMode`.
    PaymentSimple(&'a str),
    /// Nested form with `name="mixed"`.
    Mixed,
}

/// Pick the `<biz>` shape for a native-flow message from its first button.
///
/// Only the payment family keeps a literal flow name. Every other name routes
/// through `mixed`, because live probes (issue #1132: fifteen sends from a
/// Business account to a consumer handset) found that every one of the named
/// nested shapes is refused — `cta_url`, `call_permission_request` and
/// `payment_info` with a 473, `open_webview` and `galaxy_message` with a 405 —
/// while `mixed` delivered on all ten of its attempts. Leading an otherwise
/// byte-identical message with a `quick_reply` re-classified it to `mixed` and
/// it went through, so the name is the only variable.
///
/// The name is very likely not the whole story, and the rest is worth a live
/// probe before anyone reinstates the named form. The WA Web bundle
/// (`WAWebSendMsgFanout.createFanoutMsgStanza`) builds a `<biz>` in exactly one
/// of three mutually exclusive shapes, and our nested form matches none of
/// them: it merges the privacy attrs with the nested child, stamps a `v="9"` on
/// `<native_flow>` that WA Web never emits (all three of its builders pass
/// `name` alone), and adds a `<quality_control>` child that appears in the
/// bundle only in the INCOMING parser. `mixed` may simply be the one name the
/// server does not validate strictly enough to notice.
fn classify_button(button_name: &str) -> BizCategory<'_> {
    match button_name {
        // Untouched by the collapse: #1132 probed only `payment_info` of the
        // six, and a merchant-provisioned account on real payment rails may
        // legitimately answer differently than the test account did.
        "payment_info" => BizCategory::PaymentSimple("payment_info"),
        "review_and_pay" => BizCategory::PaymentSimple("order_details"),
        "review_order" | "order_status" => BizCategory::PaymentSimple("order_status"),
        "payment_status" => BizCategory::PaymentSimple("payment_status"),
        "payment_method" => BizCategory::PaymentSimple("payment_method"),
        "payment_reminder" => BizCategory::PaymentSimple("payment_reminder"),

        _ => BizCategory::Mixed,
    }
}

/// Does this interactive message carry something a client can render on its
/// own, independent of native-flow buttons?
///
/// WA Web's rule verbatim (the `f` term of `getNativeFlowNameFromMsg`): a body,
/// a header title, a footer, or a header image. `hasMediaAttachment` and the
/// non-image header media are deliberately not part of it.
fn has_renderable_envelope(im: &wa::message::InteractiveMessage) -> bool {
    let non_empty = |text: Option<&str>| text.is_some_and(|t| !t.is_empty());

    if non_empty(im.body.as_option().and_then(|b| b.text.as_deref()))
        || non_empty(im.footer.as_option().and_then(|f| f.text.as_deref()))
    {
        return true;
    }
    let Some(header) = im.header.as_option() else {
        return false;
    };
    non_empty(header.title.as_deref())
        || matches!(
            header.media,
            Some(wa::message::interactive_message::header::Media::ImageMessage(_))
        )
}

/// Classify an interactive payload into the `<biz>` shape it should carry,
/// mirroring WA Web's `getNativeFlowNameFromMsg`: the first native-flow
/// button's name decides when there is one, otherwise a payload that renders
/// on its own is announced as `mixed`.
///
/// That second arm is what makes a carousel work (issue #1133): its buttons
/// live on the cards, not at the top level, so the button rule never fires and
/// the message used to leave without a `<biz>` at all — accepted, acked, and
/// then invisible on the handset. A `shopStorefrontMessage` is excluded, as it
/// is in WA Web.
fn classify_interactive(im: &wa::message::InteractiveMessage) -> Option<BizCategory<'_>> {
    use wa::message::interactive_message::InteractiveMessage as Payload;
    match im.interactive_message.as_ref()? {
        Payload::NativeFlowMessage(nf) if !nf.buttons.is_empty() => {
            nf.buttons.first()?.name.as_deref().map(classify_button)
        }
        Payload::ShopStorefrontMessage(_) => None,
        _ => has_renderable_envelope(im).then_some(BizCategory::Mixed),
    }
}

/// Derive the `<biz>` stanza child for interactive messages.
///
/// Returns `None` when the message has no interactive payload, or one that
/// announces nothing (a storefront, or a payload with neither buttons nor any
/// renderable envelope). Otherwise returns the assembled `<biz>` node. The
/// caller is responsible for prepending `<bot biz_bot="1"/>` for DM-bound
/// sends (see `build_extra_stanza_nodes`).
///
/// `now_unix_secs` is the current wall-clock time in unix seconds. Taking it
/// as a parameter keeps the function pure and lets tests pin the resulting
/// `privacy_mode_ts` deterministically without touching the global time
/// provider.
fn infer_biz_node(msg: &wa::Message, now_unix_secs: u64) -> Option<Node> {
    let category = classify_interactive(extract_interactive_message(msg)?)?;
    let privacy_mode_ts = now_unix_secs
        .saturating_sub(BIZ_PRIVACY_MODE_TS_OFFSET)
        .to_string();

    Some(match category {
        BizCategory::PaymentSimple(flow_name) => NodeBuilder::new("biz")
            .attr("actual_actors", "2")
            .attr("host_storage", "2")
            .attr("privacy_mode_ts", &privacy_mode_ts)
            .attr("native_flow_name", flow_name)
            .build(),
        BizCategory::Mixed => build_nested_biz(&privacy_mode_ts, "mixed"),
    })
}

fn build_nested_biz(privacy_mode_ts: &str, flow_name: &str) -> Node {
    NodeBuilder::new("biz")
        .attr("actual_actors", "2")
        .attr("host_storage", "2")
        .attr("privacy_mode_ts", privacy_mode_ts)
        .children([
            NodeBuilder::new("interactive")
                .attr("type", "native_flow")
                .attr("v", "1")
                .children([NodeBuilder::new("native_flow")
                    .attr("v", "9")
                    .attr("name", flow_name)
                    .build()])
                .build(),
            NodeBuilder::new("quality_control")
                .attr("source_type", "third_party")
                .build(),
        ])
        .build()
}

fn extract_interactive_message(msg: &wa::Message) -> Option<&wa::message::InteractiveMessage> {
    // Only checks documentWithCaptionMessage wrapper (for media headers) and direct field.
    // Does not use unwrap_message() since we need the InteractiveMessage specifically.
    if let Some(doc) = msg.document_with_caption_message.as_option()
        && let Some(inner) = doc.message.as_option()
        && let Some(im) = inner.interactive_message.as_option()
    {
        return Some(im);
    }
    msg.interactive_message.as_option()
}

/// Assemble the `extra_stanza_nodes` vector for a non-newsletter send.
///
/// Order: `inferred_meta`, optional `<bot biz_bot="1"/>` (DM only), `<biz>`,
/// then any user-provided extra nodes. Pure so the caller stays trivial and
/// the assembly logic is unit-testable.
fn build_extra_stanza_nodes(
    to: &Jid,
    inferred_meta: Option<Node>,
    biz: Option<Node>,
    user_nodes: Vec<Node>,
) -> Vec<Node> {
    if inferred_meta.is_none() && biz.is_none() {
        return user_nodes;
    }
    let bot_emitted = biz.is_some() && !to.is_group();
    let extra = inferred_meta.is_some() as usize + biz.is_some() as usize + bot_emitted as usize;
    let mut nodes = Vec::with_capacity(user_nodes.len() + extra);
    nodes.extend(inferred_meta);
    if let Some(node) = biz {
        if bot_emitted {
            nodes.push(NodeBuilder::new("bot").attr("biz_bot", "1").build());
        }
        nodes.push(node);
    }
    nodes.extend(user_nodes);
    nodes
}

fn build_revoke_message(
    remote_jid: &Jid,
    from_me: bool,
    message_id: String,
    participant: Option<String>,
) -> wa::Message {
    wa::Message {
        protocol_message: buffa::MessageField::some(wa::message::ProtocolMessage {
            key: buffa::MessageField::some(wa::MessageKey {
                remote_jid: Some(remote_jid.to_string()),
                from_me: Some(from_me),
                id: Some(message_id),
                participant,
            }),
            r#type: Some(wa::message::protocol_message::Type::Revoke),
            ..Default::default()
        }),
        ..Default::default()
    }
}

/// A newsletter (channel) admin op on an existing message: edit (with the
/// replacement body) or revoke. Keeping content tied to the variant makes the
/// invalid edit-without-body / revoke-with-body states unrepresentable.
pub(crate) enum NewsletterEdit<'a> {
    Edit(&'a wa::Message),
    Revoke,
}

/// Build a newsletter (channel) plaintext edit/revoke stanza. The target is keyed
/// by `message_id` (the original message's stanza id string, the wire `id`), NOT
/// by `server_id`: WA Web (mergeNewsletterClientIDMixin -> `id`) and whatsmeow
/// (sendNewsletter, req.ID = protocolMessage.key.id) both reference edit/revoke by
/// the message id and emit no `server_id` (that attr is reaction-only).
pub(crate) fn build_newsletter_edit_node(
    to: &Jid,
    message_id: &str,
    op: NewsletterEdit<'_>,
) -> Node {
    use crate::types::message::EditAttribute;
    let mut plaintext = NodeBuilder::new("plaintext");
    let (edit, stanza_type, body) = match op {
        NewsletterEdit::Edit(m) => {
            if let Some(mt) = wacore::send::media_type_from_message(m) {
                plaintext = plaintext.attr("mediatype", mt);
            }
            (
                EditAttribute::AdminEdit,
                wacore::send::stanza_type_from_message(m),
                waproto::codec::message_to_vec(m),
            )
        }
        NewsletterEdit::Revoke => (EditAttribute::AdminRevoke, "text", Vec::new()),
    };
    NodeBuilder::new("message")
        .attr("to", to)
        .attr("id", message_id)
        .attr("type", stanza_type)
        .attr("edit", edit.to_string_val())
        .children([plaintext.bytes(body).build()])
        .build()
}

/// Build a message edit in WA Web's wire shape: a top-level
/// protocolMessage(type=MESSAGE_EDIT) carrying the new content under
/// editedMessage, same as build_revoke_message and our own receive path. The
/// top-level Message.editedMessage FutureProofMessage is the history/storage
/// form, not what WA Web sends on the wire.
pub(crate) fn build_edit_message(
    remote_jid: &Jid,
    message_id: String,
    participant: Option<String>,
    new_content: wa::Message,
    timestamp_ms: i64,
) -> wa::Message {
    wa::Message {
        protocol_message: buffa::MessageField::some(wa::message::ProtocolMessage {
            key: buffa::MessageField::some(wa::MessageKey {
                remote_jid: Some(remote_jid.to_string()),
                from_me: Some(true),
                id: Some(message_id),
                participant,
            }),
            r#type: Some(wa::message::protocol_message::Type::MessageEdit),
            edited_message: buffa::MessageField::some(new_content),
            timestamp_ms: Some(timestamp_ms),
            ..Default::default()
        }),
        ..Default::default()
    }
}

impl Client {
    /// Send a message to a user, group, or newsletter.
    ///
    /// Newsletter messages are sent as plaintext (no E2E encryption).
    /// For status/story updates use [`Client::status()`] instead.
    pub fn send_message(
        &self,
        to: impl Into<Jid>,
        message: wa::Message,
    ) -> impl Future<Output = Result<SendResult, SendError>> + '_ {
        // Sync-prologue box: a plain async fn would hold the ~1 KB message
        // by value in every embedder's frame.
        let to = to.into();
        let message = Box::new(message);
        async move {
            // Box::pin: the inner future carries ~1 KB of pre-encrypt locals.
            Box::pin(self.send_message_with_options_inner(to, message, SendOptions::default()))
                .await
        }
    }

    /// Plain-text convenience over [`Client::send_message`].
    pub fn send_text(
        &self,
        to: impl Into<Jid>,
        text: impl Into<String>,
    ) -> impl Future<Output = Result<SendResult, SendError>> + '_ {
        use wacore::proto_helpers::MessageBuilderExt;
        let to = to.into();
        let message = Box::new(wa::Message::text(text));
        async move {
            Box::pin(self.send_message_with_options_inner(to, message, SendOptions::default()))
                .await
        }
    }

    /// Forward an existing message to a chat.
    ///
    /// Builds a forward-ready copy of `message` (sets `is_forwarded`, bumps the
    /// forwarding score, strips the reply/quote chain, and drops the source
    /// `message_secret`) via
    /// [`MessageExt::prepare_for_forward`](wacore::proto_helpers::MessageExt::prepare_for_forward),
    /// then sends it.
    /// `message` may be a received body or a wrapper (ephemeral/view-once); the
    /// inner content is unwrapped before forwarding. Existing media is relayed
    /// from the same CDN blob rather than re-uploaded.
    pub fn forward_message(
        &self,
        to: impl Into<Jid>,
        message: &wa::Message,
    ) -> impl Future<Output = Result<SendResult, SendError>> + '_ {
        use wacore::proto_helpers::MessageExt;
        let to = to.into();
        let body = message.get_base_message().prepare_for_forward();
        async move {
            Box::pin(self.send_message_with_options_inner(to, body, SendOptions::default())).await
        }
    }

    /// Send a message with additional options.
    pub fn send_message_with_options(
        &self,
        to: impl Into<Jid>,
        message: wa::Message,
        options: SendOptions,
    ) -> impl Future<Output = Result<SendResult, SendError>> + '_ {
        // Thin generic shim: the large async body below stays monomorphic so
        // each `Into<Jid>` instantiation does not duplicate the state machine.
        // Sync-prologue box + Box::pin as in send_message.
        let to = to.into();
        let message = Box::new(message);
        async move { Box::pin(self.send_message_with_options_inner(to, message, options)).await }
    }

    #[cfg_attr(
        feature = "tracing",
        tracing::instrument(
            name = "wa.send.message",
            level = "debug",
            skip_all,
            fields(
                to = %to.observe(),
                lid = tracing::field::Empty,
                pn = tracing::field::Empty
            ),
            err(Debug)
        )
    )]
    async fn send_message_with_options_inner(
        &self,
        to: Jid,
        mut message: Box<wa::Message>,
        options: SendOptions,
    ) -> Result<SendResult, SendError> {
        #[cfg(feature = "tracing")]
        self.record_identity_on_span(&tracing::Span::current());

        validate_extra_stanza_nodes(&options.extra_stanza_nodes)?;
        if options.message_id.as_ref().is_some_and(String::is_empty) {
            return Err(SendError::InvalidRequest(
                "message ID must not be empty".into(),
            ));
        }

        let _t = wacore::telemetry::timer(wacore::telemetry::SEND_DURATION);
        self.stats.record_message_sent();
        wacore::telemetry::send(match to.server {
            Server::Group => "group",
            Server::Broadcast => "status",
            Server::Newsletter => "newsletter",
            _ => "dm",
        });
        if let Some(exp) = options.ephemeral_expiration
            && exp > 0
        {
            use wacore::proto_helpers::MessageExt;
            if !message.set_ephemeral_expiration(exp) {
                // Bare `conversation` messages have no contextInfo field.
                log::warn!("Could not set contextInfo.expiration on this message type");
            }
        }

        let stanza_type_override = options.stanza_type_override;
        let group_metadata_freshness = options.group_metadata_freshness;
        let device_freshness = options.device_freshness;
        let sent_at = SendInstant::now();
        let request_id = match options.message_id {
            Some(id) => id,
            None => self.generate_message_id_at(sent_at.unix_secs_u64()),
        };
        // Both paths below consume `to`, so save a copy for the result. The id
        // is not copied: it is lent to the pipeline as `&str` and moved into
        // the result once the send returns.
        let result_to = to.clone();

        // Newsletters are not E2E encrypted — send as plaintext via SMAX stanza.
        // Matches WA Web's OutMessagePublishNewsletterRequest + ContentType mixins.
        if to.is_newsletter() {
            let stanza_type = stanza_type_override
                .map(StanzaType::as_wire)
                .unwrap_or_else(|| wacore::send::stanza_type_from_message(&message));
            let (_, meta_node) = infer_stanza_metadata(&message);
            let mut plaintext_builder = NodeBuilder::new("plaintext");
            if let Some(mt) = wacore::send::media_type_from_message(&message) {
                plaintext_builder = plaintext_builder.attr("mediatype", mt);
            }
            let mut children = vec![
                plaintext_builder
                    .bytes(waproto::codec::message_to_vec(&message))
                    .build(),
            ];
            children.extend(meta_node);
            children.extend(options.extra_stanza_nodes);
            let stanza = NodeBuilder::new("message")
                .attr("to", to)
                .attr("type", stanza_type)
                .attr("id", &request_id)
                .children(children)
                .build();
            self.send_node(stanza).await?;
            return Ok(SendResult {
                message_id: request_id,
                to: result_to,
            });
        }

        let (edit, inferred_meta) = infer_stanza_metadata(&message);
        let biz = infer_biz_node(&message, sent_at.unix_secs_u64());

        let extra_nodes =
            build_extra_stanza_nodes(&to, inferred_meta, biz, options.extra_stanza_nodes);
        // send_message_impl now boxes each branch future itself, so its own
        // frame (prologue + epilogue) embeds here without a second box; the
        // shim's Box::pin above still keeps `send_message`'s future
        // pointer-sized for callers embedding it in their own futures.
        self.send_message_impl(
            to,
            &message,
            SendPipelineOptions {
                sent_at: Some(sent_at),
                request_id: Some(&request_id),
                edit,
                extra_stanza_nodes: extra_nodes,
                stanza_type: stanza_type_override,
                group_metadata_freshness,
                device_freshness,
                ..Default::default()
            },
        )
        .await
        .map_err(SendError::from_anyhow)?;
        Ok(SendResult {
            message_id: request_id,
            to: result_to,
        })
    }

    /// Send a status/story update using sender-key encryption.
    ///
    /// Status uses LID addressing (matches `WAWebEncryptAndSendStatusMsg`):
    /// LID recipients pass through, PN recipients are resolved to LID via
    /// `Client::get_lid_pn_entry` (cache-aside), and unresolvable recipients
    /// are skipped silently. The resulting `GroupInfo` carries
    /// `AddressingMode::Lid`; `prepare_group_stanza` signs with `own_lid`
    /// and emits `addressing_mode="lid"` on the stanza. Errors only if no
    /// recipient could be resolved.
    #[cfg_attr(feature = "tracing", tracing::instrument(name = "wa.send.status", level = "debug", skip_all, fields(count = recipients.len()), err(Debug)))]
    pub(crate) async fn send_status_message(
        &self,
        message: wa::Message,
        recipients: &[Jid],
        mut options: crate::features::status::StatusSendOptions,
    ) -> Result<SendResult, SendError> {
        use wacore::client::context::GroupInfo;
        use wacore_binary::builder::NodeBuilder;

        if recipients.is_empty() {
            return Err(SendError::InvalidRequest(
                "cannot send status with no recipients".into(),
            ));
        }
        validate_extra_stanza_nodes(&options.extra_stanza_nodes)?;
        validate_status_message_id(&message, options.message_id.as_deref())?;

        // Status posts don't go through send_message_with_options, so count them here.
        let _t = wacore::telemetry::timer(wacore::telemetry::SEND_DURATION);
        self.stats.record_message_sent();
        wacore::telemetry::send("status");

        let to = Jid::status_broadcast();
        let request_id = options
            .message_id
            .take()
            .unwrap_or_else(|| self.generate_message_id());

        // Borrow from the held snapshot: no field clones, the Arc keeps it alive.
        let device_snapshot = self.persistence_manager.get_device_snapshot();
        let account_info = &device_snapshot.account;
        let own_jid = device_snapshot.pn.as_ref().ok_or(SendError::NotLoggedIn)?;
        // Status is LID-addressed (matches WA Web post-LID-migration). Without
        // a real device LID we can't sign or fan out correctly; refuse rather
        // than silently emit `addressing_mode="lid"` with a PN sender.
        let own_lid = device_snapshot.lid.as_ref().ok_or_else(|| {
            SendError::InvalidRequest(
                "cannot send status: device has no LID yet. Finish pairing / LID \
                 migration before posting status."
                    .into(),
            )
        })?;

        // Fail fast for any JID that isn't a user (PN or LID). Mirrors WA
        // Web's `asUserWidOrThrow` inside `toUserLid`: non-user inputs are a
        // programming bug, not something to silently drop during resolution.
        for jid in recipients {
            if !(jid.is_pn() || jid.is_lid()) {
                return Err(SendError::InvalidRequest(format!(
                    "invalid status recipient {jid}: must be a user JID (PN or LID), \
                     not a group/broadcast/newsletter/hosted/etc."
                )));
            }
        }

        use futures::StreamExt;
        use std::collections::HashMap;
        // Resolve recipient LIDs concurrently (a status audience can be hundreds of
        // contacts, each a cold-cache DB read). Stream over indices and rebuild
        // `resolved` in order — assemble_status_participants is position-sensitive.
        const STATUS_LID_RESOLVE_CONCURRENCY: usize = 16;
        let resolved_indexed: Vec<(usize, Option<Jid>)> =
            futures::stream::iter(0..recipients.len())
                .map(|i| async move { (i, self.resolve_recipient_to_lid(&recipients[i]).await) })
                .buffer_unordered(STATUS_LID_RESOLVE_CONCURRENCY)
                .collect()
                .await;
        let mut resolved: Vec<Option<Jid>> = vec![None; recipients.len()];
        let mut lid_to_pn_map: HashMap<wacore_binary::CompactString, Jid> =
            HashMap::with_capacity(recipients.len() + 1);
        for (i, lid) in resolved_indexed {
            if let Some(lid_jid) = lid {
                if recipients[i].is_pn() {
                    lid_to_pn_map.insert(lid_jid.user.clone(), recipients[i].to_non_ad());
                }
                resolved[i] = Some(lid_jid);
            }
        }
        lid_to_pn_map.insert(own_lid.user.clone(), own_jid.to_non_ad());

        let participants = wacore::send::assemble_status_participants(resolved, own_lid)?;
        let mut group_info =
            GroupInfo::with_lid_to_pn_map(participants, AddressingMode::Lid, lid_to_pn_map);

        // One encode feeds retry cache and wire; mci-hoist re-encodes (folded context).
        let shared_content = message
            .message_context_info
            .is_unset()
            .then(|| std::sync::Arc::new(waproto::codec::message_to_vec(&message)));
        self.add_recent_message(&to, &request_id, &message, shared_content.clone())
            .await;

        let device_store_arc = self.persistence_manager.get_device_arc().await;
        let to_str = to.to_string();
        let distribution_guard = self.group_distribution_lock(&to).await;

        let force_skdm = {
            use wacore::libsignal::store::sender_key_name::SenderKeyName;
            // Sender key name tracks the addressing mode of the group stanza.
            // Since status now uses LID addressing (see send_status_message
            // header), the key is stored under own_lid, matching the address
            // prepare_group_stanza derives internally.
            let sender_address = own_lid.to_protocol_address();
            let sender_key_name = SenderKeyName::from_parts(&to_str, sender_address.as_str());

            let key_exists = self
                .signal_cache
                .get_sender_key(&sender_key_name, &*device_snapshot.backend)
                .await?
                .is_some();

            if !key_exists {
                self.reset_sender_key_device_tracking(&to_str).await?;
            }

            !key_exists
        };

        let mut store_adapter = self.signal_adapter_from(device_store_arc.clone());
        let mut stores = store_adapter.as_signal_stores();

        // Determine which devices need SKDM using the unified per-device map.
        // Status keeps the prior phash behavior, so we drop the full device set
        // and only use the SKDM-target subset.
        let skdm_target_devices =
            if !force_skdm || options.device_freshness == crate::cache::Freshness::Refresh {
                self.resolve_status_skdm_targets(
                    &to_str,
                    &group_info,
                    own_lid,
                    options.device_freshness,
                    force_skdm,
                )
                .await?
            } else {
                None
            };

        // prepare_group_stanza and ensure_status_participants both read the
        // participant list and expect self present. Done after SKDM resolution
        // to preserve the prior ordering (resolve ran without self appended).
        let own_status_base = own_lid.to_non_ad();
        if !group_info
            .participants
            .iter()
            .any(|participant| participant.is_same_user_as(&own_status_base))
        {
            group_info.participants.push(own_status_base);
        }

        // `<meta status_setting>` describes the POSTER's privacy on their own
        // status. Reactions go through WA Web's addon path and never visit
        // `WAWebEncryptAndSendStatusMsg`; attaching the meta on a reaction
        // gets the stanza NACK'd with 479 (SmaxInvalid). Revokes also skip it.
        let mut extra_stanza_nodes = options.extra_stanza_nodes;
        if wacore::send::status_carries_privacy_meta(&message) {
            extra_stanza_nodes.push(
                NodeBuilder::new("meta")
                    .attr("status_setting", options.privacy.as_str())
                    .build(),
            );
        }

        let prepared = match wacore::send::prepare_group_stanza(
            &*self.runtime,
            &mut stores,
            self,
            wacore::send::GroupStanzaRequest {
                group: &group_info,
                own_jid,
                own_lid,
                account: account_info.as_deref(),
                to: &to,
                message: &message,
                message_id: &request_id,
                force_distribution: force_skdm,
                distribution_targets: skdm_target_devices,
                distribution_policy: wacore::send::SenderKeyDistributionPolicy::BestEffort,
                phash_devices: None,
                edit: None,
                extra_nodes: &extra_stanza_nodes,
                pre_encoded: shared_content.as_deref().map(Vec::as_slice),
            },
        )
        .await
        {
            Ok(prepared) => prepared,
            Err(e) => {
                if let Some(SignalProtocolError::NoSenderKeyState(_)) =
                    e.downcast_ref::<SignalProtocolError>()
                {
                    log::warn!("No sender key for status broadcast, forcing distribution.");

                    self.reset_sender_key_device_tracking(&to_str).await?;

                    let mut store_adapter_retry =
                        self.signal_adapter_from(device_store_arc.clone());
                    let mut stores_retry = store_adapter_retry.as_signal_stores();

                    wacore::send::prepare_group_stanza(
                        &*self.runtime,
                        &mut stores_retry,
                        self,
                        wacore::send::GroupStanzaRequest {
                            group: &group_info,
                            own_jid,
                            own_lid,
                            account: account_info.as_deref(),
                            to: &to,
                            message: &message,
                            message_id: &request_id,
                            force_distribution: true,
                            distribution_targets: None,
                            distribution_policy:
                                wacore::send::SenderKeyDistributionPolicy::BestEffort,
                            phash_devices: None,
                            edit: None,
                            extra_nodes: &extra_stanza_nodes,
                            pre_encoded: shared_content.as_deref().map(Vec::as_slice),
                        },
                    )
                    .await?
                } else {
                    return Err(e.into());
                }
            }
        };

        let stanza = self
            .ensure_status_participants(prepared.node, &group_info)
            .await?;

        // Gate the stanza on the sender-key ratchet advance being durable
        // (same rule as the DM/group send path); a failure aborts the send.
        self.persist_signal_state_pre_wire().await?;

        let ack = stanza
            .attrs()
            .optional_string("phash")
            .map(|s| wacore_binary::CompactString::from(s.as_ref()));
        if let Some(phash) = ack.clone() {
            self.register_phash_waiter(&request_id, phash, to.clone(), true);
        }

        if let Err(e) = self.send_node(stanza).await {
            if ack.is_some() {
                self.response_waiters_guard().remove(&request_id);
            }
            return Err(e.into());
        }

        self.update_sender_key_devices(&to_str, &prepared.skdm_devices)
            .await;
        drop(distribution_guard);

        for user in &prepared.stale_device_users {
            self.invalidate_device_cache(user).await;
        }

        Ok(SendResult {
            message_id: request_id,
            to,
        })
    }

    /// Resolve the group's device set for a warm/partial send. Returns
    /// `None` when device resolution fails (caller falls back to the full
    /// `force_skdm` path), otherwise `Some((all_devices, needs_skdm))` where
    /// `all_devices` is the complete resolved set (feeds the phash) and
    /// `needs_skdm` is the subset still missing the sender key (feeds SKDM
    /// distribution). `needs_skdm` may be empty (fully warm send).
    ///
    /// For LID mode, uses `group_info.phone_jid_for_lid_user` to query devices
    /// via PN when available (LID usync is unreliable for own JID), then
    /// converts the result back to LID. Same fallback as `prepare_group_stanza`.
    /// Load (or lazily build) the per-group sender-key device map.
    ///
    /// Atomic get-or-init: if another task invalidated the cache during our
    /// DB read, get_or_init's single-flight guarantee means the stale data
    /// won't be inserted — the invalidation wins and the next caller re-inits.
    async fn skdm_device_map(
        &self,
        group_jid: &str,
    ) -> std::sync::Arc<crate::sender_key_device_cache::SenderKeyDeviceMap> {
        use crate::sender_key_device_cache::SenderKeyDeviceMap;
        let pm = self.persistence_manager.clone();
        self.sender_key_device_cache
            .get_or_init(group_jid, async {
                let db_rows = pm
                    .get_sender_key_devices(group_jid)
                    .await
                    .unwrap_or_else(|e| {
                        log::warn!(
                            "Failed to read sender key devices for {}: {:?}",
                            group_jid,
                            e
                        );
                        vec![]
                    });
                std::sync::Arc::new(SenderKeyDeviceMap::from_db_rows(&db_rows))
            })
            .await
    }

    /// Filter the resolved device set down to the subset still needing SKDM.
    ///
    /// No empty-cache early-exit: WA Web iterates an empty `senderKey` Map
    /// as `false` per participant, so the filter must run unconditionally.
    fn filter_skdm_targets(
        &self,
        group_jid: &str,
        all_devices: &[Jid],
        cached_map: &crate::sender_key_device_cache::SenderKeyDeviceMap,
        own_sending_jid: &Jid,
    ) -> Vec<Jid> {
        let needs_skdm: Vec<Jid> = all_devices
            .iter()
            .filter(|device| {
                if device.is_hosted() {
                    return false;
                }
                if device.user == own_sending_jid.user && device.device == own_sending_jid.device {
                    return false;
                }
                // WA Web parity (ParticipantStore.js skDistribList): a device is
                // warm only when it AND its primary (device 0) hold the key, so a
                // forgotten primary redistributes the whole user while a forgotten
                // companion redistributes only itself. One inner-map resolution
                // per device (single user-string hash) instead of two.
                !cached_map.device_and_primary_warm(&device.user, device.device)
            })
            .cloned()
            .collect();

        log::debug!(
            "Resolved {} devices ({} need SKDM) for {}",
            all_devices.len(),
            needs_skdm.len(),
            group_jid
        );
        needs_skdm
    }

    /// SKDM target resolution for the status path, whose `GroupInfo` is built
    /// fresh per send (no stable identity to memoize against).
    #[cfg_attr(feature = "tracing", tracing::instrument(name = "wa.send.resolve_skdm_targets", level = "debug", skip_all, fields(group = %wacore_binary::jid::observe_str(group_jid))))]
    async fn resolve_status_skdm_targets(
        &self,
        group_jid: &str,
        group_info: &wacore::client::context::GroupInfo,
        own_sending_jid: &Jid,
        freshness: crate::cache::Freshness,
        force_distribution: bool,
    ) -> Result<Option<Vec<Jid>>, anyhow::Error> {
        let cached_map = if force_distribution {
            None
        } else {
            Some(self.skdm_device_map(group_jid).await)
        };

        let is_lid_mode = group_info.addressing_mode == AddressingMode::Lid;
        let jids_to_resolve: Vec<Jid> = group_info
            .participants
            .iter()
            .map(|jid| {
                if is_lid_mode
                    && jid.is_lid()
                    && let Some(pn) = group_info.phone_jid_for_lid_user(&jid.user)
                {
                    return pn.to_non_ad();
                }
                jid.to_non_ad()
            })
            .collect();

        let resolved = match freshness {
            crate::cache::Freshness::CachePreferred => {
                self.get_user_devices_owned(jids_to_resolve).await
            }
            crate::cache::Freshness::Refresh => self.refresh_user_devices(jids_to_resolve).await,
        };
        match resolved {
            Ok(mut devices) => {
                if is_lid_mode {
                    for device in &mut devices {
                        *device = group_info.phone_device_jid_into_lid(std::mem::take(device));
                    }
                }
                if force_distribution {
                    wacore::send::retain_skdm_distribution_targets(&mut devices, own_sending_jid);
                } else if let Some(cached_map) = cached_map {
                    devices.retain(|device| {
                        !device.is_hosted()
                            && !(device.user == own_sending_jid.user
                                && device.device == own_sending_jid.device)
                            && !cached_map.device_and_primary_warm(&device.user, device.device)
                    });
                }
                log::debug!(
                    "Resolved {} status devices needing SKDM for {}",
                    devices.len(),
                    group_jid
                );
                Ok(Some(devices))
            }
            Err(error) if freshness == crate::cache::Freshness::CachePreferred => {
                log::warn!(
                    "Failed to resolve devices for SKDM check in {}: {:?}",
                    group_jid,
                    error
                );
                Ok(None)
            }
            Err(error) => Err(error),
        }
    }

    /// SKDM target resolution for cached-group sends: the full device set
    /// comes from the per-group memo (`resolve_group_devices_memoized`), so a
    /// warm repeat send skips the per-member registry fan-out entirely.
    #[cfg_attr(feature = "tracing", tracing::instrument(name = "wa.send.resolve_skdm_targets_memoized", level = "debug", skip_all, fields(group = %group_jid)))]
    async fn resolve_skdm_targets_memoized(
        &self,
        group: &Jid,
        group_jid: &str,
        group_info: &std::sync::Arc<wacore::client::context::GroupInfo>,
        own_sending_jid: &Jid,
    ) -> Option<(std::sync::Arc<wacore::send::ResolvedGroupDevices>, Vec<Jid>)> {
        let cached_map = self.skdm_device_map(group_jid).await;
        match self
            .resolve_group_devices_memoized(group, group_info, own_sending_jid)
            .await
        {
            Ok(all_devices) => {
                // Load after the resolve await (so a cold flip during it is
                // visible to the hit check below) but BEFORE the filter: a
                // flip racing the filter stamps the inserted memo as already
                // stale (its stored generation lags the map's), so the next
                // read re-runs it. A flip after this load and before the send
                // is the same bounded one-send window the unmemoized filter
                // has, recovered by the retry-receipt resend.
                let cached_map_gen = cached_map.generation();
                // Skip the O(devices) filter_skdm_targets scan when the same
                // (devices, sender-key-map) Arc pair, generation AND sending
                // identity were already warm, reusing the memoized targets
                // (empty, or the own devices that re-receive their SKDM every
                // send). The devices Arc swaps on membership change; the
                // cached-map Arc swaps on a warm-mark invalidation; the
                // generation catches an in-place cold flip that keeps the
                // same Arc; the memoized needs are a pure function of that
                // identity.
                if self.device_memos_enabled
                    && let Some((dw, cw, memo_gen, memo_sender, memo_needs)) =
                        self.skdm_warm_memo.get(group).await
                    && std::ptr::eq(dw.as_ptr(), std::sync::Arc::as_ptr(&all_devices))
                    && std::ptr::eq(cw.as_ptr(), std::sync::Arc::as_ptr(&cached_map))
                    && memo_gen == cached_map_gen
                    && &memo_sender == own_sending_jid
                {
                    return Some((all_devices, memo_needs));
                }
                let needs_skdm = self.filter_skdm_targets(
                    group_jid,
                    all_devices.devices(),
                    &cached_map,
                    own_sending_jid,
                );
                if self.device_memos_enabled
                    && (needs_skdm.is_empty() || {
                        let snapshot = self.persistence_manager.get_device_snapshot();
                        skdm_needs_only_own_devices(
                            &needs_skdm,
                            snapshot.pn.as_ref(),
                            snapshot.lid.as_ref(),
                        )
                    })
                {
                    self.skdm_warm_memo
                        .insert(
                            group.clone(),
                            (
                                std::sync::Arc::downgrade(&all_devices),
                                std::sync::Arc::downgrade(&cached_map),
                                cached_map_gen,
                                own_sending_jid.clone(),
                                needs_skdm.clone(),
                            ),
                        )
                        .await;
                }
                Some((all_devices, needs_skdm))
            }
            Err(e) => {
                log::warn!(
                    "Failed to resolve devices for SKDM check in {}: {:?}",
                    group_jid,
                    e
                );
                None
            }
        }
    }

    /// Update sender key device tracking after a successful group/status send.
    ///
    /// Called AFTER `send_node()` succeeds (WA Web: `markHasSenderKey` after server ACK).
    /// On full distribution, clears old state and marks the provided device list.
    /// On partial, marks only the specific SKDM recipients.
    ///
    /// The `all_resolved_devices` parameter carries the exact device list resolved
    /// for the stanza, avoiding a redundant `resolve_devices` call and preventing
    /// the clear-then-fail race where a transient resolver failure leaves the map empty.
    /// Mark devices as `has_key=true` after successful SKDM distribution.
    ///
    /// Excludes our own devices (`exclude_own_devices=true`), mirroring WA Web's
    /// `ParticipantStore` helper, which guards every `markHasSenderKey` mutation
    /// with `!isMeDevice`. Own companions are therefore never memoized as warm, so
    /// `filter_skdm_targets` re-distributes their SKDM on every send — the same
    /// reason WA Web can't orphan its own companions. Marking them here instead
    /// would be one-directional: the retry-receipt forget path also excludes own
    /// devices (to stop an inbound retry tearing down our own session), so an own
    /// companion whose one SKDM encryption failed could never be re-sent one.
    pub(crate) async fn update_sender_key_devices(&self, group_jid: &str, devices: &[Jid]) {
        if devices.is_empty() {
            return;
        }

        // No invalidation on success: set_sender_key_status_for_devices
        // already drops the cached map when it actually writes new warm marks,
        // and an own-devices-only set (never memoized) writes nothing —
        // invalidating for it would force a DB re-read on every warm send.
        if let Err(e) = self
            .set_sender_key_status_for_devices(group_jid, devices, true, true)
            .await
        {
            log::warn!(
                "Failed to update sender key devices for {}: {:?}",
                group_jid,
                e
            );
            // A failed write may still have partially landed (backend
            // implementations are not required to be atomic), so drop the
            // cached map rather than risk serving pre-write state.
            self.sender_key_device_cache.invalidate(group_jid).await;
        }
    }

    /// Cold path of the phash check: the server's phash disagreed with ours, so
    /// invalidate the relevant device/group caches and (for groups) force
    /// sender-key redistribution. Spawned only on a mismatch, which is why the
    /// common path costs a string comparison on the read loop.
    #[cfg_attr(feature = "tracing", tracing::instrument(name = "wa.send.phash_mismatch", level = "debug", skip_all, fields(jid = %jid.observe())))]
    pub(crate) async fn handle_phash_mismatch(
        &self,
        jid: &Jid,
        our_phash: &str,
        server_phash: &str,
        invalidate_group_cache: bool,
    ) {
        log::warn!(
            "Phash mismatch for {}: ours={our_phash}, server={server_phash}. Invalidating caches.",
            jid.observe()
        );
        // DM phash covers both recipient + own devices
        // (WA Web: syncDeviceListJob([recipient, me]))
        if !jid.is_group() && !jid.is_status_broadcast() {
            self.invalidate_device_cache(&jid.user).await;
            if let Some(own_pn) = &self.persistence_manager.get_device_snapshot().pn {
                self.invalidate_device_cache(&own_pn.user).await;
            }
        }
        let jid_str = jid.to_string();
        // Cache-only invalidation re-reads the same stale rows on the next send.
        // Drop the persisted state too so the next send takes the full-
        // distribution path. If the clear fails, fall back to deleting the bot's
        // own sender key for the chat — the next send will see `!key_exists` and
        // force_skdm without depending on the tracker.
        let mut flush_fallback = false;
        if jid.is_group() || jid.is_status_broadcast() {
            let distribution_guard = self.group_distribution_lock(jid).await;
            if let Err(e) = self.reset_sender_key_device_tracking(&jid_str).await {
                log::warn!(
                    "phash mismatch: clear_sender_key_devices failed: {e} — \
                     deleting own sender key as fallback to force redistribution"
                );
                use wacore::libsignal::store::sender_key_name::SenderKeyName;
                use wacore::types::jid::JidExt;
                let snapshot = self.persistence_manager.get_device_snapshot();
                for own in snapshot.lid.iter().chain(snapshot.pn.iter()) {
                    let sk =
                        SenderKeyName::from_parts(&jid_str, own.to_protocol_address().as_str());
                    self.signal_cache.delete_sender_key(sk.cache_key()).await;
                }
                flush_fallback = true;
            }
            drop(distribution_guard);
        } else {
            self.sender_key_device_cache.invalidate(&jid_str).await;
        }
        if flush_fallback {
            let _ = self
                .flush_signal_cache_batch_safe_logged("phash-mismatch-fallback", None)
                .await;
        }
        if invalidate_group_cache {
            self.lock_group_metadata(jid).await.invalidate().await;
        }
    }

    /// Ensure the status stanza has a <participants> node listing all recipient
    /// user JIDs. WhatsApp Web's `participantList` uses bare USER JIDs (not
    /// device JIDs) — `<to jid="user@s.whatsapp.net"/>` — to tell the server
    /// which users should receive the skmsg. The SKDM distribution list
    /// (already in <participants>) uses device JIDs with <enc> children.
    async fn ensure_status_participants(
        &self,
        stanza: Node,
        group_info: &wacore::client::context::GroupInfo,
    ) -> Result<Node, anyhow::Error> {
        Ok(wacore::send::ensure_status_participants(stanza, group_info))
    }

    #[cfg_attr(feature = "tracing", tracing::instrument(name = "wa.send.impl", level = "debug", skip_all, fields(to = %to.observe()), err(Debug)))]
    pub(crate) async fn send_message_impl(
        &self,
        to: Jid,
        message: &wa::Message,
        options: SendPipelineOptions<'_>,
    ) -> Result<(), anyhow::Error> {
        let SendPipelineOptions {
            sent_at,
            request_id: request_id_override,
            peer,
            force_key_distribution,
            edit,
            extra_stanza_nodes,
            stanza_type: stanza_type_override,
            group_metadata_freshness,
            device_freshness,
            borrowed_message_id,
        } = options;
        // Callers that already stamped their message hand the instant down; the
        // rest sample here so the pipeline below still has exactly one.
        let sent_at = sent_at.unwrap_or_else(SendInstant::now);
        validate_extra_stanza_nodes(&extra_stanza_nodes)?;
        if request_id_override.is_some_and(str::is_empty) {
            return Err(SendError::InvalidRequest("message ID must not be empty".into()).into());
        }
        // Newsletters are plaintext channels and never use the E2E path. Text
        // sends go through the <plaintext> branch in send_message_with_options;
        // edit/revoke have dedicated plaintext methods (newsletter().edit_message
        // / revoke_message). A newsletter JID here is a mis-routed pin/edit/revoke
        // (pin is not a channel op), so reject it.
        if to.is_newsletter() {
            return Err(SendError::InvalidRequest(
                "newsletter JIDs are not valid on the E2E send path; use \
                 newsletter().edit_message/revoke_message (pin is unsupported on channels)"
                    .into(),
            )
            .into());
        }

        // status@broadcast reactions fan out pairwise to the author's devices;
        // status posts keep going through send_status_message (owns recipients).
        let (to, is_status_addon) = if to.is_status_broadcast() {
            let author = message
                .reaction_message
                .as_option()
                .and_then(|rm| rm.key.as_option())
                .and_then(|k| k.participant.as_ref())
                .and_then(|p| p.parse::<Jid>().ok())
                .filter(|jid| jid.is_pn() || jid.is_lid())
                .ok_or_else(|| {
                    SendError::InvalidRequest(
                        "send_message to status@broadcast requires \
                         reaction_message.key.participant = status author (user JID). \
                         Use client.status() for posting new statuses."
                            .into(),
                    )
                })?;
            (author, true)
        } else {
            (to, false)
        };

        // Generate request ID early (doesn't need lock). This frame owns the
        // only copy for the whole send: the branch builders, the phash waiter
        // and the messageSecret persistence all borrow it, so a send names its
        // message exactly once no matter how many stages read that name.
        let generated_request_id;
        let request_id: &str = match request_id_override {
            Some(id) => id,
            None => {
                generated_request_id = self.generate_message_id_at(sent_at.unix_secs_u64());
                &generated_request_id
            }
        };
        let tc_issue_target = to.clone();

        // Dispatch to a concrete boxed future per branch: this function's own
        // frame stays small (prologue + epilogue), and a DM send never
        // allocates the group branch's state machine, which dominated the old
        // single-future layout.
        let SendBranchOutput {
            node: stanza_to_send,
            msg_secret: outbound_msg_secret,
            group_sender_identity: outbound_group_sender_identity,
            skdm_update,
            distribution_guard,
            issue_tc_token_after_send: should_issue_tc_token_after_send,
            dm_phash,
        } = if peer && !to.is_group() {
            box_send_branch(self.send_peer_branch(to, message, request_id)).await?
        } else if to.is_group() {
            box_send_branch(self.send_group_branch(GroupBranchRequest {
                to,
                message,
                request_id,
                force_key_distribution,
                edit,
                extra_stanza_nodes: &extra_stanza_nodes,
                group_metadata_freshness,
                device_freshness,
                borrowed_message_id,
            }))
            .await?
        } else {
            box_send_branch(self.send_dm_branch(DmBranchRequest {
                to,
                message,
                request_id,
                sent_at,
                edit,
                extra_stanza_nodes,
                is_status_addon,
                device_freshness,
                borrowed_message_id,
            }))
            .await?
        };

        // The outbound advance must be durable BEFORE the stanza hits the wire:
        // reusing an outbound counter reuses its message key + IV. Counters are
        // leased in batches (see `SessionRecord::reserve_sender_chain_counters`),
        // so most sends are already covered by a durable lease and only
        // schedule the coalesced write-behind; a send that raised either lease
        // flushes synchronously, and a persistence failure must abort the send
        // rather than transmit an advance we couldn't save.
        self.persist_signal_state_pre_wire().await?;

        // A borrowed id must not register a phash ack-waiter: the waiter map is
        // keyed by outer stanza id, so it would overwrite the original send's
        // waiter (either ack could resolve the wrong send, and the older timeout
        // could remove the replacement). The edit's own ack is best-effort.
        // Registered before the stanza goes out: the ack can arrive while
        // send_node is still returning, and a waiter installed afterwards would
        // miss it.
        // Keying the waiter off `request_id` rather than re-reading the stanza
        // is only sound while every branch stamps the id it was handed; assert
        // that instead of paying an owned copy of an attribute we already have.
        debug_assert_eq!(
            stanza_to_send.attrs().optional_string("id").as_deref(),
            Some(request_id),
            "branch stanza must carry the id this send was named with"
        );
        let ack_message_id = if !borrowed_message_id && let Some(phash) = dm_phash {
            // Group sends also invalidate group cache on mismatch: the server's
            // participant set diverged, so the next send needs a fresh query.
            let invalidate_group = tc_issue_target.is_group();
            self.register_phash_waiter(
                request_id,
                phash,
                tc_issue_target.clone(),
                invalidate_group,
            );
            Some(request_id)
        } else {
            None
        };

        // Server expects the outer `to` as the broadcast chat even though
        // encryption targeted the author's devices (mirrors incoming `from`).
        let mut stanza_to_send = stanza_to_send;
        if is_status_addon {
            stanza_to_send.attrs.insert("to", Jid::status_broadcast());
        }
        if let Some(t) = stanza_type_override {
            stanza_to_send.attrs.insert("type", t.as_wire());
        }

        if let Err(e) = self.send_node(stanza_to_send).await {
            if let Some(msg_id) = ack_message_id {
                self.response_waiters_guard().remove(msg_id);
            }
            return Err(e.into());
        }
        // Skip when the stanza id is borrowed from another message: binding the
        // outbound secret under the borrowed id would overwrite the original
        // message's secret (breaking later reactions/poll votes on it).
        if !borrowed_message_id && let Some(secret) = outbound_msg_secret.as_ref() {
            let sender = match outbound_group_sender_identity {
                Some(s) => Some(s),
                None => self.dm_sender_identity_for(&tc_issue_target).await,
            };
            if let Some(sender) = sender {
                let is_bot_chat = tc_issue_target.is_bot();
                let class = wacore::msg_secret::classify(message, is_bot_chat);
                self.persist_outbound_msg_secret(
                    &tc_issue_target,
                    &sender,
                    request_id,
                    secret,
                    class,
                    sent_at,
                )
                .await;
            }
        }

        if let Some(update) = skdm_update {
            self.update_sender_key_devices(&update.to_str, &update.devices)
                .await;
            for user in &update.stale_users {
                self.invalidate_device_cache(user).await;
            }
        }
        // Warm marking is visible; a waiting cold send may now re-resolve.
        drop(distribution_guard);

        // Issue new tc token after send if a bucket boundary was crossed.
        // Fire-and-forget so send_message returns without waiting for the IQ
        if should_issue_tc_token_after_send {
            if let Some(client) = self.self_weak.get().and_then(|w| w.upgrade()) {
                let target = tc_issue_target;
                self.runtime
                    .spawn(Box::pin(async move {
                        client.issue_tc_token_after_send(&target).await;
                    }))
                    .detach();
            } else {
                log::debug!(target: "Client/TcToken", "Skipping fire-and-forget issuance: client dropped");
            }
        }

        Ok(())
    }

    /// Peer branch of [`Self::send_message_impl`]: own-device sync messages,
    /// never groups.
    async fn send_peer_branch(
        &self,
        to: Jid,
        message: &wa::Message,
        request_id: &str,
    ) -> Result<SendBranchOutput, anyhow::Error> {
        let node = {
            // Peer messages are only valid for individual users, not groups
            // Resolve encryption JID and acquire lock ONLY for encryption
            let encryption_jid = self.resolve_encryption_jid(&to).await;
            let signal_addr = encryption_jid.to_protocol_address();

            let session_mutex = self.session_lock_for(signal_addr.as_str()).await;
            let _session_guard = session_mutex.lock().await;

            let mut store_adapter = self.signal_adapter().await;

            let device_snapshot = self.persistence_manager.get_device_snapshot();
            wacore::send::prepare_peer_stanza(
                &mut store_adapter.session_store,
                &mut store_adapter.identity_store,
                to,
                &signal_addr,
                message,
                request_id,
                device_snapshot.account.as_deref(),
            )
            .await?
        };
        Ok(SendBranchOutput::stanza_only(node))
    }

    /// Group branch of [`Self::send_message_impl`]: sender-key encryption,
    /// SKDM distribution and the cold/rotation single-flight.
    async fn send_group_branch(
        &self,
        request: GroupBranchRequest<'_>,
    ) -> Result<SendBranchOutput, anyhow::Error> {
        let GroupBranchRequest {
            to,
            message,
            request_id,
            force_key_distribution,
            edit,
            extra_stanza_nodes,
            group_metadata_freshness,
            device_freshness,
            borrowed_message_id,
        } = request;
        // Every arm of the prepare match below assigns these three.
        let outbound_msg_secret: Option<[u8; 32]>;
        let outbound_group_sender_identity: Option<Jid>;
        let skdm_update: Option<SkdmUpdate>;
        let mut distribution_guard: Option<async_lock::MutexGuardArc<()>> = None;
        let node = {
            // No send-level lock: encrypt_group_message serializes the
            // sender-key chain advance per (group, sender) at the cipher.
            let group_info = self
                .groups()
                .query_info_with_freshness(&to, group_metadata_freshness)
                .await?;

            // Borrow from the held snapshot: no field clones, the Arc keeps it alive.
            let device_snapshot = self.persistence_manager.get_device_snapshot();
            let account_info = &device_snapshot.account;
            let own_jid = device_snapshot
                .pn
                .as_ref()
                .ok_or(ClientError::NotLoggedIn)?;
            let own_lid = device_snapshot
                .lid
                .as_ref()
                .ok_or_else(|| anyhow!("LID not set, cannot send to group"))?;

            // One encode feeds retry cache and wire; mci-hoist re-encodes (folded context).
            let shared_content = message
                .message_context_info
                .is_unset()
                .then(|| std::sync::Arc::new(waproto::codec::message_to_vec(message)));
            // Store serialized message bytes for retry (lightweight). Skip when
            // the id is borrowed: it would replace the original message's
            // retry-cache entry, so a retry receipt for it returns this edit.
            if !borrowed_message_id {
                self.add_recent_message(&to, request_id, message, shared_content.clone())
                    .await;
            }

            let device_store_arc = self.persistence_manager.get_device_arc().await;
            let to_str = to.to_string();

            let (own_sending_jid, _) = match group_info.addressing_mode {
                AddressingMode::Lid => (own_lid.clone(), "lid"),
                AddressingMode::Pn => (own_jid.clone(), "pn"),
            };

            // Memo identity must be the CACHED Arc: ensure_self_in_group clones
            // a fresh GroupInfo whenever self is absent from the snapshot, which
            // would make the memo miss on every send to such groups. The memoized
            // resolver applies the same self-append internally.
            let group_info_for_memo = std::sync::Arc::clone(&group_info);
            let refreshed_devices = if device_freshness == crate::cache::Freshness::Refresh {
                Some(
                    self.resolve_group_devices_uncached(
                        &group_info_for_memo,
                        &own_sending_jid,
                        crate::cache::Freshness::Refresh,
                    )
                    .await?,
                )
            } else {
                None
            };
            // resolve_skdm_targets and prepare_group_stanza both read the
            // participant list and expect self to be present.
            let group_info = ensure_self_in_group(group_info, &own_sending_jid);

            // Side-effect-free cold check: does the sender key record exist,
            // and has its chain advanced past the rotation threshold? Reads
            // the record without deleting anything, so a false positive (a
            // concurrent send already rotating/recreating) costs only the
            // re-check under the lock below.
            use wacore::libsignal::store::sender_key_name::SenderKeyName;
            let sender_address = own_sending_jid.to_protocol_address();
            let sender_key_name = SenderKeyName::from_parts(&to_str, sender_address.as_str());
            // WA Web posts SenderKeyExpired with `PERIODIC_ROTATION` after
            // a chain advances past a threshold. Captured-js doesn't show
            // the value; 1000 mirrors common Signal hygiene defaults.
            const SENDER_KEY_ROTATION_THRESHOLD: u32 = 1000;
            let read_sender_key_state = || async {
                let record = self
                    .signal_cache
                    .get_sender_key(&sender_key_name, &*device_snapshot.backend)
                    .await?;
                let key_exists = record.is_some();
                // Read the chain iteration through the shared `Arc` without cloning
                // the record: borrow the current state instead of `*_mut().cloned()`.
                let needs_rotation = record
                    .as_ref()
                    .and_then(|r| r.sender_key_state().ok())
                    .and_then(|state| state.sender_chain_key())
                    .map(|ck| ck.iteration())
                    .is_some_and(|iter| iter >= SENDER_KEY_ROTATION_THRESHOLD);
                Ok::<(bool, bool), anyhow::Error>((key_exists, needs_rotation))
            };

            let (key_exists, needs_rotation) = read_sender_key_state().await?;
            let mut force_skdm = force_key_distribution || !key_exists || needs_rotation;
            if force_skdm {
                // Serialize the whole rotation/redistribution under the
                // per-group guard and RE-CHECK once inside it: a send that
                // merely raced the winner's delete->recreate window sees the
                // fresh record here and downgrades to a warm send instead of
                // redistributing to every member again.
                distribution_guard = Some(self.group_distribution_lock(&to).await);
                let (key_exists, needs_rotation) = read_sender_key_state().await?;
                force_skdm = force_key_distribution || !key_exists || needs_rotation;
                if !key_exists || needs_rotation {
                    self.reset_sender_key_device_tracking(&to_str).await?;
                }
                if needs_rotation {
                    log::info!(
                        "Periodic sender-key rotation for {} (chain iteration >= {SENDER_KEY_ROTATION_THRESHOLD})",
                        to.observe()
                    );
                    self.signal_cache
                        .delete_sender_key(sender_key_name.cache_key())
                        .await;
                }
                if !force_skdm {
                    distribution_guard = None;
                }
            }

            let mut store_adapter = self.signal_adapter_from(device_store_arc.clone());

            let mut stores = store_adapter.as_signal_stores();

            // Determine which devices need SKDM distribution using the unified
            // per-device sender key map (matches WA Web's participant.senderKey Map).
            // `all_devices_for_phash` carries the FULL resolved set so the phash
            // covers every device + self even on a warm send (WA Web sends a
            // phash on every group send); `skdm_target_devices` is the subset
            // still missing the key. On the cold/`force_skdm` path both are
            // `None` and `prepare_group_stanza` resolves the set itself.
            let (all_devices_for_phash, skdm_target_devices): (
                Option<GroupDeviceSnapshot>,
                Option<Vec<Jid>>,
            ) = if force_skdm {
                match refreshed_devices {
                    Some(mut targets) => {
                        wacore::send::retain_skdm_distribution_targets(
                            &mut targets,
                            &own_sending_jid,
                        );
                        (None, Some(targets))
                    }
                    None => (None, None),
                }
            } else {
                let initial_targets = match refreshed_devices {
                    Some(all) => {
                        let all = GroupDeviceSnapshot::Owned(
                            wacore::send::ResolvedGroupDevices::new(all),
                        );
                        let cached_map = self.skdm_device_map(&to_str).await;
                        let needs = self.filter_skdm_targets(
                            &to_str,
                            all.as_ref().devices(),
                            &cached_map,
                            &own_sending_jid,
                        );
                        Some((all, needs))
                    }
                    None => self
                        .resolve_skdm_targets_memoized(
                            &to,
                            &to_str,
                            &group_info_for_memo,
                            &own_sending_jid,
                        )
                        .await
                        .map(|(all, needs)| (GroupDeviceSnapshot::Shared(all), needs)),
                };
                match initial_targets {
                    Some((all, needs)) if needs.is_empty() => (Some(all), Some(needs)),
                    // Own devices are never memoized warm, so they re-receive
                    // their SKDM on every send by design — own-only needs IS
                    // the warm steady state, not a cold group: no distribution
                    // guard, no cache invalidation, no re-resolve.
                    Some((all, needs))
                        if skdm_needs_only_own_devices(&needs, Some(own_jid), Some(own_lid)) =>
                    {
                        (Some(all), Some(needs))
                    }
                    Some((first_all, first_needs)) => {
                        // Cold: wait for any in-flight distribution, then
                        // re-resolve. The loser usually finds every device
                        // already marked warm by the winner and downgrades to a
                        // plain skmsg send; if the winner failed, the targets
                        // are still cold and this send distributes normally.
                        distribution_guard = Some(self.group_distribution_lock(&to).await);
                        // Force a DB re-read: a concurrent warm send may have
                        // started the cache init before the winner's marking
                        // landed and then published that stale (empty) map,
                        // which would otherwise turn this into a full
                        // re-distribution to every member.
                        self.sender_key_device_cache.invalidate(&to_str).await;
                        match self
                            .resolve_skdm_targets_memoized(
                                &to,
                                &to_str,
                                &group_info_for_memo,
                                &own_sending_jid,
                            )
                            .await
                        {
                            Some((all, needs)) => {
                                // Fully warm OR down to the own-only steady
                                // state: nothing left that needs the
                                // single-flight, release it before the send.
                                if needs.is_empty()
                                    || skdm_needs_only_own_devices(
                                        &needs,
                                        Some(own_jid),
                                        Some(own_lid),
                                    )
                                {
                                    distribution_guard = None;
                                }
                                (Some(GroupDeviceSnapshot::Shared(all)), Some(needs))
                            }
                            // Transient re-resolve failure: keep the first
                            // resolve's targets rather than silently sending
                            // without the distribution it already knew was
                            // needed.
                            None => (Some(first_all), Some(first_needs)),
                        }
                    }
                    None => (None, None),
                }
            };

            match wacore::send::prepare_group_stanza(
                &*self.runtime,
                &mut stores,
                self,
                wacore::send::GroupStanzaRequest {
                    group: &group_info,
                    own_jid,
                    own_lid,
                    account: account_info.as_deref(),
                    to: &to,
                    message,
                    message_id: request_id,
                    force_distribution: force_skdm,
                    distribution_targets: skdm_target_devices,
                    distribution_policy: wacore::send::SenderKeyDistributionPolicy::BestEffort,
                    phash_devices: all_devices_for_phash.as_ref().map(AsRef::as_ref),
                    edit: edit.as_ref(),
                    extra_nodes: extra_stanza_nodes,
                    pre_encoded: shared_content.as_deref().map(Vec::as_slice),
                },
            )
            .await
            {
                Ok(prepared) => {
                    skdm_update = Some(SkdmUpdate {
                        to_str: to_str.clone(),
                        devices: prepared.skdm_devices,
                        stale_users: prepared.stale_device_users,
                    });
                    outbound_msg_secret = prepared.message_secret;
                    outbound_group_sender_identity = Some(prepared.sender_identity);
                    prepared.node
                }
                Err(e) => {
                    if let Some(SignalProtocolError::NoSenderKeyState(_)) =
                        e.downcast_ref::<SignalProtocolError>()
                    {
                        log::warn!(
                            "No sender key for group {}, forcing distribution.",
                            to.observe()
                        );

                        // This retry redistributes, so it needs the same
                        // single-flight guard as a cold send (a warm send that
                        // lost its sender key arrives here without one).
                        if distribution_guard.is_none() {
                            distribution_guard = Some(self.group_distribution_lock(&to).await);
                        }

                        // Re-check under the guard: a concurrent retry may have
                        // already recreated the key and marked the devices, in
                        // which case this send retries warm instead of clearing
                        // the tracking and redistributing to every member again.
                        let (key_recreated, _) = read_sender_key_state().await?;
                        let warm_targets = if key_recreated {
                            self.sender_key_device_cache.invalidate(&to_str).await;
                            self.resolve_skdm_targets_memoized(
                                &to,
                                &to_str,
                                &group_info_for_memo,
                                &own_sending_jid,
                            )
                            .await
                        } else {
                            None
                        };
                        let (retry_force, retry_targets, retry_all) = match warm_targets {
                            Some((all, needs)) => {
                                (false, Some(needs), Some(GroupDeviceSnapshot::Shared(all)))
                            }
                            None => {
                                self.reset_sender_key_device_tracking(&to_str).await?;
                                (true, None, None)
                            }
                        };

                        let mut store_adapter_retry =
                            self.signal_adapter_from(device_store_arc.clone());
                        let mut stores_retry = store_adapter_retry.as_signal_stores();

                        let retry_prepared = wacore::send::prepare_group_stanza(
                            &*self.runtime,
                            &mut stores_retry,
                            self,
                            wacore::send::GroupStanzaRequest {
                                group: &group_info,
                                own_jid,
                                own_lid,
                                account: account_info.as_deref(),
                                to: &to,
                                message,
                                message_id: request_id,
                                force_distribution: retry_force,
                                distribution_targets: retry_targets,
                                distribution_policy:
                                    wacore::send::SenderKeyDistributionPolicy::BestEffort,
                                phash_devices: retry_all.as_ref().map(AsRef::as_ref),
                                edit: edit.as_ref(),
                                extra_nodes: extra_stanza_nodes,
                                pre_encoded: shared_content.as_deref().map(Vec::as_slice),
                            },
                        )
                        .await?;

                        skdm_update = Some(SkdmUpdate {
                            to_str,
                            devices: retry_prepared.skdm_devices,
                            stale_users: retry_prepared.stale_device_users,
                        });
                        outbound_msg_secret = retry_prepared.message_secret;
                        outbound_group_sender_identity = Some(retry_prepared.sender_identity);
                        retry_prepared.node
                    } else {
                        return Err(e);
                    }
                }
            }
        };
        Ok(SendBranchOutput {
            node,
            msg_secret: outbound_msg_secret,
            group_sender_identity: outbound_group_sender_identity,
            skdm_update,
            distribution_guard,
            issue_tc_token_after_send: false,
            dm_phash: None,
        })
    }

    /// DM branch of [`Self::send_message_impl`]: pairwise Signal encryption
    /// with device fan-out (also used by status-reaction add-ons).
    async fn send_dm_branch(
        &self,
        request: DmBranchRequest<'_>,
    ) -> Result<SendBranchOutput, anyhow::Error> {
        let DmBranchRequest {
            to,
            message,
            request_id,
            sent_at,
            edit,
            extra_stanza_nodes,
            is_status_addon,
            device_freshness,
            borrowed_message_id,
        } = request;
        let mut should_issue_tc_token_after_send = false;
        let prepared = {
            // Per-device locking to match decrypt path (message.rs:684),
            // preventing ratchet desync on concurrent send/receive.

            // One encode feeds retry cache and wire; mci-hoist re-encodes (folded context).
            let shared_content = message
                .message_context_info
                .is_unset()
                .then(|| std::sync::Arc::new(waproto::codec::message_to_vec(message)));
            // Status reaction retries arrive with `from=status@broadcast`;
            // cache under the broadcast chat so take_recent_message hits. Skip
            // for a borrowed id: it would replace the original message's
            // retry-cache entry (a retry receipt for it would return this edit).
            if !borrowed_message_id {
                if is_status_addon {
                    self.add_recent_message(
                        &Jid::status_broadcast(),
                        request_id,
                        message,
                        shared_content.clone(),
                    )
                    .await;
                } else {
                    self.add_recent_message(&to, request_id, message, shared_content.clone())
                        .await;
                }
            }

            let device_snapshot = self.persistence_manager.get_device_snapshot();
            let own_jid = device_snapshot
                .pn
                .as_ref()
                .ok_or(ClientError::NotLoggedIn)?;

            // PN→LID mapping (WA Web: ManagePhoneNumberMappingJob)
            if to.is_pn() && self.lid_pn_cache.get_current_lid(&to.user).await.is_none() {
                let sid = self.generate_request_id();
                let spec = wacore::iq::usync::LidQuerySpec::new(vec![to.to_non_ad()], sid);
                // Best-effort: WA Web also catches and warns on failure
                match self.execute(spec).await {
                    Ok(resp) => {
                        for mapping in &resp.lid_mappings {
                            if let Err(e) = self
                                .add_lid_pn_mapping(
                                    &mapping.lid,
                                    &mapping.phone_number,
                                    crate::lid_pn_cache::LearningSource::Usync,
                                )
                                .await
                            {
                                log::warn!(
                                    "Failed to persist LID mapping {} -> {}: {e:?}",
                                    mapping.phone_number,
                                    mapping.lid
                                );
                            }
                        }
                    }
                    Err(e) => {
                        log::warn!(
                            "LID query failed for {}, falling back to PN: {e:?}",
                            to.observe()
                        );
                    }
                }
            }

            // The LID-vs-PN wire namespace is an account-level decision: the
            // server 400-nacks LID-addressed DMs from accounts that are not
            // 1:1-LID-migrated (issue #941).
            let recipient_bare = self.resolve_dm_wire_jid(&to).await;

            let stanza_to = dm_stanza_to(&recipient_bare, &to);

            // DM fanout, memoized per recipient: a warm repeat DM skips both
            // registry lookups, the list rebuild and the phash. See
            // `resolve_dm_devices_memoized` for its freshness contract.
            let dm_devices = self
                .resolve_dm_devices_memoized(
                    &to,
                    &recipient_bare,
                    own_jid,
                    device_snapshot.lid.as_ref(),
                    device_freshness,
                )
                .await?;

            self.ensure_e2e_sessions(dm_devices.devices()).await?;

            let mut extra_stanza_nodes = extra_stanza_nodes;
            // tctoken applies to 1:1 chats; status reactions share the fanout
            // path but WA Web does not attach tctokens to them.
            if !to.is_group() && !to.is_newsletter() && !is_status_addon {
                should_issue_tc_token_after_send = self
                    .maybe_include_tc_token(&to, &mut extra_stanza_nodes, sent_at)
                    .await;
            }
            if should_issue_tc_token_after_send {
                debug!(target: "Client/TcToken", "Scheduled tc token issuance after send for {}", to.observe());
            }

            let lock_jids = self.build_session_lock_keys(dm_devices.devices()).await;
            let _session_guards = self.session_guards_for(&lock_jids).await;

            let mut store_adapter = self.signal_adapter().await;

            let mut stores = store_adapter.as_signal_stores();

            wacore::send::prepare_dm_stanza(
                &*self.runtime,
                &mut stores,
                self,
                wacore::send::DmStanzaRequest {
                    own_jid,
                    account: device_snapshot.account.as_deref(),
                    to: &stanza_to,
                    message,
                    message_id: request_id,
                    edit: edit.as_ref(),
                    extra_nodes: &extra_stanza_nodes,
                    devices: &dm_devices,
                    pre_encoded: shared_content.as_deref().map(Vec::as_slice),
                },
            )
            .await?
        };
        Ok(SendBranchOutput {
            node: prepared.node,
            msg_secret: prepared.message_secret,
            group_sender_identity: None,
            skdm_update: None,
            distribution_guard: None,
            issue_tc_token_after_send: should_issue_tc_token_after_send,
            dm_phash: prepared.phash,
        })
    }

    /// Persist a generated `MessageContextInfo.message_secret` keyed by
    /// `(chat_non_ad, sender_non_ad, msg_id)`. The sender identity must
    /// match what `<meta target_sender_jid>` echoes back at GET time —
    /// LID for bot chats and LID-mode groups, PN otherwise.
    pub(crate) async fn persist_outbound_msg_secret(
        &self,
        chat: &Jid,
        sender: &Jid,
        msg_id: &str,
        secret: &[u8; wacore::reporting_token::MESSAGE_SECRET_SIZE],
        class: wacore::msg_secret::RetentionClass,
        sent_at: SendInstant,
    ) {
        let policy = self.cache_config.msg_secret_policy;
        if !policy.persists() {
            return;
        }
        // BotOnly keeps only bot-context secrets; a group message that invokes a
        // bot classifies as Bot, so its reply can still be decrypted.
        if policy.bot_only() && class != wacore::msg_secret::RetentionClass::Bot {
            return;
        }
        // Outbound secrets are minted with the parent event, so the send's own
        // instant IS the parent event time.
        let now = sent_at.unix_secs();
        let expires_at = wacore::msg_secret::expires_at(
            policy,
            &self.cache_config.msg_secret_retention,
            class,
            u64::try_from(now).ok(),
            now,
        );
        let entry = wacore::store::traits::MsgSecretEntry::new(
            chat, sender, msg_id, *secret, expires_at, now,
        );
        // Same write-behind buffer as inbound captures: visible immediately,
        // flushed off the send path (msmsg replies read buffer-first).
        self.msg_secret_buffer.queue_one(entry).await;
    }

    /// Decide the identity (LID vs PN) under which an outbound DM's
    /// `messageSecret` should be persisted. Group sends should use
    /// `PreparedGroupStanza.sender_identity` directly instead of this.
    pub(crate) async fn dm_sender_identity_for(&self, to: &Jid) -> Option<Jid> {
        if to.server == Server::Bot {
            self.lid()
        } else {
            self.pn()
        }
    }

    /// Build sorted, deduplicated per-device session lock keys.
    /// INVARIANT: Keys are sorted to prevent deadlocks when acquiring multiple
    /// session locks (e.g. DM sends that encrypt for recipient + own devices).
    /// Resolve encryption JIDs and sort for deadlock-free lock acquisition.
    pub(crate) async fn build_session_lock_keys(&self, device_jids: &[Jid]) -> Vec<Jid> {
        let mut keys: Vec<Jid> = Vec::with_capacity(device_jids.len());
        for jid in device_jids {
            keys.push(self.resolve_encryption_jid(jid).await);
        }
        keys.sort_unstable_by(wacore::types::jid::cmp_for_lock_order);
        keys.dedup_by(|a, b| wacore::types::jid::cmp_for_lock_order(a, b).is_eq());
        keys
    }

    /// Take every per-device session lock, in `jids` order.
    ///
    /// INVARIANT: acquisition order IS `jids` order, and callers pass keys from
    /// [`Self::build_session_lock_keys`], which sorts them. That single order is
    /// what keeps two sends overlapping on a device from deadlocking, so a
    /// change here has to preserve it.
    ///
    /// Each mutex is locked as it is resolved rather than resolving the whole
    /// set first: the handles exist only to be locked, so the vector holding
    /// them was pure staging. The guards themselves must still be collected —
    /// they are what keeps the locks held for the caller's scope.
    pub(crate) async fn session_guards_for(
        &self,
        jids: &[Jid],
    ) -> Vec<async_lock::MutexGuardArc<()>> {
        // A duplicate key would have this loop await a lock it already holds,
        // which is a silent self-deadlock rather than a panic: the send just
        // never returns. Every caller goes through `build_session_lock_keys`,
        // which sorts and dedups, so this only fires if a future path forgets
        // to.
        debug_assert!(
            jids.windows(2).all(|pair| pair[0] != pair[1]),
            "session lock keys must be deduped before acquisition, or the loop deadlocks on itself"
        );

        let mut guards = Vec::with_capacity(jids.len());
        // A `ProtocolAddress` IS the "{name}.0" string the lock map is keyed by,
        // and it holds it inline, so the whole loop names its keys without
        // allocating a formatting buffer.
        let mut addr = wacore::types::jid::make_reusable_protocol_address();
        for jid in jids {
            jid.reset_protocol_address(&mut addr);
            let mutex = self.session_lock_for(addr.as_str()).await;
            guards.push(mutex.lock_arc().await);
        }
        guards
    }

    /// The mutexes [`Self::session_guards_for`] would take, without taking
    /// them. Only tests need this: production code always wants the guards, and
    /// resolving handles it does not lock is what this commit removed.
    #[cfg(test)]
    pub(crate) async fn session_mutexes_for(
        &self,
        jids: &[Jid],
    ) -> Vec<std::sync::Arc<async_lock::Mutex<()>>> {
        let mut mutexes = Vec::with_capacity(jids.len());
        let mut addr = wacore::types::jid::make_reusable_protocol_address();
        for jid in jids {
            jid.reset_protocol_address(&mut addr);
            mutexes.push(self.session_lock_for(addr.as_str()).await);
        }
        mutexes
    }
}

/// Self-DM detection: appending an own-device lookup on top of the
/// recipient's list would address each physical device twice (LID + PN),
/// which the server rejects with `ack error="400"`.
/// WAWebDBDeviceListFanout never re-fetches the own list for the same account.
pub(crate) fn is_self_dm_recipient(
    recipient_bare: &Jid,
    own_pn: &Jid,
    own_lid: Option<&Jid>,
) -> bool {
    match recipient_bare.server {
        Server::Lid => own_lid.is_some_and(|lid| recipient_bare.user == lid.user),
        Server::Pn => recipient_bare.user == own_pn.user,
        _ => false,
    }
}

/// The outer `<message to>`, the DeviceSentMessage destinationJid, and the
/// reporting-token remote jid must share the participants' namespace.
/// WAWebSendMsgCreateFanoutStanza builds the whole stanza from one CHAT_JID
/// (always a bare user wid), so the `to` is the resolved wire jid whenever
/// the caller's namespace differs from it (LID upgrade, or PN downgrade on
/// an unmigrated account), and a device-qualified caller jid is normalized
/// to the bare chat jid. A `to` mixing namespaces with the participants is
/// rejected wholesale by the server with `ack error="400"`.
pub(crate) fn dm_stanza_to(recipient_bare: &Jid, to: &Jid) -> Jid {
    if recipient_bare.is_lid() || to.is_lid() {
        recipient_bare.clone()
    } else {
        to.to_non_ad()
    }
}

#[cfg(test)]
#[allow(clippy::disallowed_methods)]
mod tests {
    use super::*;
    use crate::test_utils::wait_for_lock_waiter;
    use std::str::FromStr;

    #[test]
    fn status_revoke_requires_a_distinct_outer_stanza_id() {
        let target_id = "3EB0REVOKETARGET";
        let revoke = wa::Message {
            protocol_message: buffa::MessageField::some(wa::message::ProtocolMessage {
                r#type: Some(wa::message::protocol_message::Type::Revoke),
                key: buffa::MessageField::some(wa::MessageKey {
                    id: Some(target_id.into()),
                    ..Default::default()
                }),
                ..Default::default()
            }),
            ..Default::default()
        };

        assert!(matches!(
            validate_status_message_id(&revoke, Some(target_id)),
            Err(SendError::InvalidRequest(_))
        ));
        assert!(validate_status_message_id(&revoke, Some("3EB0NEWSTANZAID")).is_ok());
        assert!(validate_status_message_id(&revoke, None).is_ok());
    }

    #[test]
    fn dm_stanza_to_follows_resolved_wire_namespace() {
        let pn: Jid = "5511987650001@s.whatsapp.net".parse().unwrap();
        let lid: Jid = "111000011112222@lid".parse().unwrap();

        // PN caller, PN wire (unmigrated or unmapped): caller jid preserved.
        assert_eq!(dm_stanza_to(&pn, &pn), pn);
        // PN caller upgraded to LID wire: `to` must be the LID.
        assert_eq!(dm_stanza_to(&lid, &pn), lid);
        // LID caller kept on LID wire: unchanged.
        assert_eq!(dm_stanza_to(&lid, &lid), lid);
        // LID caller downgraded to PN wire (unmigrated account): `to` must be
        // the PN — reusing the caller's LID would mix namespaces.
        assert_eq!(dm_stanza_to(&pn, &lid), pn);
        // Device-qualified caller jid is normalized to the bare chat jid.
        let pn_device: Jid = "5511987650001:5@s.whatsapp.net".parse().unwrap();
        assert_eq!(dm_stanza_to(&pn, &pn_device), pn);
    }

    #[test]
    fn ensure_self_in_group_shares_when_present_and_appends_when_absent() {
        use wacore::client::context::GroupInfo;
        use wacore::types::message::AddressingMode;

        let own: Jid = "999999999999@s.whatsapp.net".parse().unwrap();
        let other: Jid = "111111111111@s.whatsapp.net".parse().unwrap();

        // Self already a member (the common case): the shared Arc passes through
        // untouched, with no deep clone of the participant list.
        let with_self = Arc::new(GroupInfo::new(
            vec![other.to_non_ad(), own.to_non_ad()],
            AddressingMode::Pn,
        ));
        let out = ensure_self_in_group(with_self.clone(), &own);
        assert!(Arc::ptr_eq(&with_self, &out));

        // Self missing: a fresh GroupInfo is built with self appended.
        let without_self = Arc::new(GroupInfo::new(vec![other.to_non_ad()], AddressingMode::Pn));
        let out = ensure_self_in_group(without_self.clone(), &own);
        assert!(!Arc::ptr_eq(&without_self, &out));
        assert_eq!(out.participants.len(), 2);
        assert!(out.participants.iter().any(|p| p.is_same_user_as(&own)));
    }

    // The group SKDM pairwise fan-out must hold the SAME per-device session mutex
    // the DM path locks, so the two can't advance a shared device's ratchet at
    // once. Acquiring the group lock must block the DM per-device lock.
    #[tokio::test]
    async fn group_skdm_lock_shares_dm_per_device_session_mutex() {
        use wacore::client::context::SendContextResolver;

        let client = crate::test_utils::create_test_client().await;
        let device: Jid = "15551234567:3@s.whatsapp.net".parse().unwrap();

        // The exact mutex the DM send path would lock for this device.
        let keys = client
            .build_session_lock_keys(std::slice::from_ref(&device))
            .await;
        let dm_mutexes = client.session_mutexes_for(&keys).await;
        assert_eq!(dm_mutexes.len(), 1);
        assert!(
            dm_mutexes[0].try_lock().is_some(),
            "uncontended before the group lock"
        );

        // Hold the group SKDM lock for the same device.
        let guard = client
            .lock_device_sessions(std::slice::from_ref(&device))
            .await;
        assert!(
            dm_mutexes[0].try_lock().is_none(),
            "group SKDM fan-out must block the DM per-device session lock"
        );

        drop(guard);
        assert!(
            dm_mutexes[0].try_lock().is_some(),
            "the per-device session lock releases when the group guard drops"
        );
    }

    #[tokio::test]
    async fn send_message_to_status_without_reaction_errors() {
        let client = crate::test_utils::create_test_client().await;
        let to = Jid::status_broadcast();
        let err = client
            .send_message(
                to,
                wa::Message {
                    conversation: Some("hi".into()),
                    ..Default::default()
                },
            )
            .await
            .expect_err("status@broadcast without reaction must error");
        let msg = format!("{err}");
        assert!(
            msg.contains("reaction_message") || msg.contains("status"),
            "unexpected error: {msg}"
        );
    }

    #[tokio::test]
    async fn status_send_waits_for_distribution_guard() {
        let client = crate::test_utils::create_test_client().await;
        let own_pn: Jid = "15551234001@s.whatsapp.net".parse().unwrap();
        let own_lid: Jid = "100000000000001@lid".parse().unwrap();
        client
            .persistence_manager
            .process_command(DeviceCommand::SetId(Some(own_pn)))
            .await;
        client
            .persistence_manager
            .process_command(DeviceCommand::SetLid(Some(own_lid)))
            .await;

        let status = Jid::status_broadcast();
        let held = client.group_distribution_lock(&status).await;
        let lock = client
            .group_distribution_locks
            .get(&status)
            .await
            .expect("cached distribution lock");
        let lock_refs = Arc::strong_count(&lock);
        let mut task = tokio::spawn({
            let client = client.clone();
            async move {
                let recipient: Jid = "100000000000002@lid".parse().unwrap();
                client
                    .send_status_message(
                        wa::Message {
                            conversation: Some("serialized status".into()),
                            ..Default::default()
                        },
                        std::slice::from_ref(&recipient),
                        crate::features::status::StatusSendOptions::default(),
                    )
                    .await
            }
        });

        wait_for_lock_waiter(&lock, lock_refs).await;
        assert!(!task.is_finished(), "status send must wait for the lane");
        drop(held);

        let _ = tokio::time::timeout(std::time::Duration::from_secs(5), &mut task)
            .await
            .expect("status send must resume")
            .expect("status task");
    }

    // A logged-out send goes through send_message_impl, whose internal
    // `ClientError::NotLoggedIn` is threaded as `anyhow`. The wrapper must
    // surface the typed `SendError::NotLoggedIn`, not the `Internal` catch-all,
    // so callers can match it (regression test for r3432644890).
    #[tokio::test]
    async fn send_message_logged_out_dm_returns_not_logged_in() {
        let client = crate::test_utils::create_test_client().await;
        let to: Jid = "111111111111@s.whatsapp.net".parse().unwrap();
        let err = client
            .send_message(
                to,
                wa::Message {
                    conversation: Some("hi".into()),
                    ..Default::default()
                },
            )
            .await
            .expect_err("logged-out DM send must error");
        assert!(
            matches!(err, SendError::NotLoggedIn),
            "expected SendError::NotLoggedIn, got: {err:?}"
        );
    }

    // Edit path resolves the sender before the wire, so a logged-out DM edit
    // must surface the typed NotLoggedIn (not the Internal catch-all).
    #[tokio::test]
    async fn edit_message_logged_out_dm_returns_not_logged_in() {
        let client = crate::test_utils::create_test_client().await;
        let to: Jid = "111111111111@s.whatsapp.net".parse().unwrap();
        let err = client
            .edit_message(
                to,
                "ORIG_ID",
                wa::Message {
                    conversation: Some("x".into()),
                    ..Default::default()
                },
            )
            .await
            .expect_err("logged-out DM edit must error");
        assert!(
            matches!(err, SendError::NotLoggedIn),
            "expected SendError::NotLoggedIn, got: {err:?}"
        );
    }

    // An empty EditOptions::stanza_id must land in request_id and be rejected as
    // InvalidRequest — doubles as a guard that stanza_id actually reaches the id.
    #[tokio::test]
    async fn edit_message_with_empty_stanza_id_returns_invalid_request() {
        let client = crate::test_utils::create_test_client().await;
        seed_pn(&client, "222222222222@s.whatsapp.net").await;
        let to: Jid = "111111111111@s.whatsapp.net".parse().unwrap();
        let err = client
            .edit_message_with_options(
                to,
                "ORIG_ID",
                wa::Message {
                    conversation: Some("x".into()),
                    ..Default::default()
                },
                EditOptions {
                    stanza_id: Some(String::new()),
                },
            )
            .await
            .expect_err("empty stanza_id must error");
        assert!(
            matches!(err, SendError::InvalidRequest(_)),
            "expected SendError::InvalidRequest, got: {err:?}"
        );
    }

    #[tokio::test]
    async fn send_message_to_status_reaction_rejects_non_user_participant() {
        let client = crate::test_utils::create_test_client().await;
        let to = Jid::status_broadcast();
        let err = client
            .send_message(
                to,
                wa::Message {
                    reaction_message: buffa::MessageField::some(wa::message::ReactionMessage {
                        key: buffa::MessageField::some(wa::MessageKey {
                            remote_jid: Some("status@broadcast".into()),
                            from_me: Some(false),
                            id: Some("ORIGID".into()),
                            participant: Some("120363040237990503@g.us".into()),
                        }),
                        text: Some("❤️".into()),
                        sender_timestamp_ms: Some(1),
                        ..Default::default()
                    }),
                    ..Default::default()
                },
            )
            .await
            .expect_err("group JID as participant must error");
        assert!(
            format!("{err}").contains("user JID"),
            "expected user-JID error, got: {err}"
        );
    }

    #[tokio::test]
    async fn send_message_to_status_reaction_without_participant_errors() {
        let client = crate::test_utils::create_test_client().await;
        let to = Jid::status_broadcast();
        let err = client
            .send_message(
                to,
                wa::Message {
                    reaction_message: buffa::MessageField::some(wa::message::ReactionMessage {
                        key: buffa::MessageField::some(wa::MessageKey {
                            remote_jid: Some("status@broadcast".into()),
                            from_me: Some(false),
                            id: Some("ORIGID".into()),
                            participant: None,
                        }),
                        text: Some("❤️".into()),
                        sender_timestamp_ms: Some(1),
                        ..Default::default()
                    }),
                    ..Default::default()
                },
            )
            .await
            .expect_err("reaction without key.participant must error");
        assert!(
            format!("{err}").contains("participant"),
            "expected participant error, got: {err}"
        );
    }

    #[test]
    fn test_revoke_type_default_is_sender() {
        // RevokeType::Sender is the default (for deleting own messages)
        let revoke_type = RevokeType::default();
        assert_eq!(revoke_type, RevokeType::Sender);
    }

    #[test]
    fn test_force_skdm_only_for_admin_revoke() {
        // Admin revokes require force_skdm=true to get proper message structure
        // with phash, <participants>, and <device-identity> that WhatsApp Web uses.
        // Without this, the server returns error 479.
        let sender_jid = Jid::from_str("123456@s.whatsapp.net").unwrap();

        let sender_revoke = RevokeType::Sender;
        let admin_revoke = RevokeType::Admin {
            original_sender: sender_jid,
        };

        // This matches the logic in revoke_message()
        let force_skdm_sender = matches!(sender_revoke, RevokeType::Admin { .. });
        let force_skdm_admin = matches!(admin_revoke, RevokeType::Admin { .. });

        assert!(!force_skdm_sender, "Sender revoke should NOT force SKDM");
        assert!(force_skdm_admin, "Admin revoke MUST force SKDM");
    }

    #[test]
    fn test_sender_revoke_message_key_structure() {
        // Sender revoke (edit="7"): from_me=true, participant=None
        // The sender is identified by from_me=true, no participant field needed
        let to = Jid::from_str("120363040237990503@g.us").unwrap();
        let message_id = "3EB0ABC123".to_string();

        let (from_me, participant, edit_attr) = match RevokeType::Sender {
            RevokeType::Sender => (true, None, EditAttribute::SenderRevoke),
            RevokeType::Admin { original_sender } => (
                false,
                Some(original_sender.to_non_ad_string()),
                EditAttribute::AdminRevoke,
            ),
        };

        assert!(from_me, "Sender revoke must have from_me=true");
        assert!(
            participant.is_none(),
            "Sender revoke must NOT set participant"
        );
        assert_eq!(edit_attr.to_string_val(), "7");

        let revoke_message = build_revoke_message(&to, from_me, message_id.clone(), participant);

        let proto_msg = revoke_message.protocol_message.into_option().unwrap();
        let key = proto_msg.key.into_option().unwrap();
        assert_eq!(key.from_me, Some(true));
        assert_eq!(key.participant, None);
        assert_eq!(key.id, Some(message_id));
    }

    #[test]
    fn test_admin_revoke_message_key_structure() {
        // Admin revoke (edit="8"): from_me=false, participant=original_sender
        // The participant field identifies whose message is being deleted
        let to = Jid::from_str("120363040237990503@g.us").unwrap();
        let message_id = "3EB0ABC123".to_string();
        let original_sender = Jid::from_str("236395184570386:22@lid").unwrap();

        let revoke_type = RevokeType::Admin {
            original_sender: original_sender.clone(),
        };
        let (from_me, participant, edit_attr) = match revoke_type {
            RevokeType::Sender => (true, None, EditAttribute::SenderRevoke),
            RevokeType::Admin { original_sender } => (
                false,
                Some(original_sender.to_non_ad_string()),
                EditAttribute::AdminRevoke,
            ),
        };

        assert!(!from_me, "Admin revoke must have from_me=false");
        assert!(
            participant.is_some(),
            "Admin revoke MUST set participant to original sender"
        );
        assert_eq!(edit_attr.to_string_val(), "8");

        let revoke_message =
            build_revoke_message(&to, from_me, message_id.clone(), participant.clone());

        let proto_msg = revoke_message.protocol_message.into_option().unwrap();
        let key = proto_msg.key.into_option().unwrap();
        assert_eq!(key.from_me, Some(false));
        // Participant should be the original sender with device number stripped
        assert_eq!(key.participant, Some("236395184570386@lid".to_string()));
        assert_eq!(key.id, Some(message_id));
    }

    // Fictitious JIDs (not real PII):
    //   own PN user = "5500000000000"
    //   own LID user = "111111111111111"
    //   other LID user = "222222222222222"
    const SELF_PN: &str = "5500000000000";
    const SELF_LID: &str = "111111111111111";
    const SELF_DEVICE: u16 = 7;
    const OTHER_LID: &str = "222222222222222";

    #[test]
    fn self_dm_lid_recipient_matches_own_lid() {
        let own_pn = Jid::pn_device(SELF_PN, SELF_DEVICE);
        let own_lid = Jid::lid_device(SELF_LID, SELF_DEVICE);
        let recipient = Jid::lid(SELF_LID);

        assert!(is_self_dm_recipient(&recipient, &own_pn, Some(&own_lid)));
    }

    #[test]
    fn self_dm_pn_recipient_matches_own_pn() {
        // Self-DM addressed in PN namespace (no LID mapping resolved yet).
        let own_pn = Jid::pn_device(SELF_PN, SELF_DEVICE);
        let own_lid = Jid::lid_device(SELF_LID, SELF_DEVICE);
        let recipient = Jid::pn(SELF_PN);

        assert!(is_self_dm_recipient(&recipient, &own_pn, Some(&own_lid)));
    }

    #[test]
    fn self_dm_pn_recipient_self_dm_even_without_own_lid() {
        // PN-keyed self-detection does not require an own_lid to be known.
        let own_pn = Jid::pn_device(SELF_PN, SELF_DEVICE);
        let recipient = Jid::pn(SELF_PN);

        assert!(is_self_dm_recipient(&recipient, &own_pn, None));
    }

    #[test]
    fn non_self_lid_recipient_is_not_self_dm() {
        let own_pn = Jid::pn_device(SELF_PN, SELF_DEVICE);
        let own_lid = Jid::lid_device(SELF_LID, SELF_DEVICE);
        let recipient = Jid::lid(OTHER_LID);

        assert!(!is_self_dm_recipient(&recipient, &own_pn, Some(&own_lid)));
    }

    #[test]
    fn lid_recipient_without_own_lid_is_not_self_dm() {
        // WAWebUserPrefsMeUser.isMeAccount keys on isSameAccountAndAddressingMode;
        // PN-string equality across namespaces must NOT trigger.
        let own_pn = Jid::pn_device(SELF_PN, SELF_DEVICE);
        let recipient = Jid::lid(SELF_PN);

        assert!(!is_self_dm_recipient(&recipient, &own_pn, None));
    }

    #[test]
    fn group_or_broadcast_recipient_is_not_self_dm() {
        // Defensive: only PN/LID DMs ever take the self-DM short-circuit.
        let own_pn = Jid::pn_device(SELF_PN, SELF_DEVICE);
        let own_lid = Jid::lid_device(SELF_LID, SELF_DEVICE);

        assert!(!is_self_dm_recipient(
            &Jid::group("120363000000000000"),
            &own_pn,
            Some(&own_lid),
        ));
        assert!(!is_self_dm_recipient(
            &Jid::status_broadcast(),
            &own_pn,
            Some(&own_lid),
        ));
    }

    #[test]
    fn self_dm_with_no_recipient_cache_still_appends_own_devices() {
        // Edge case raised in PR review: if `recipient_cached` ends up `None`
        // (cache eviction + warmup failed), the self-DM short-circuit must
        // still let `own_cached` populate the fanout. Otherwise the bare-JID
        // fallback drops every companion device.
        let own_pn = Jid::pn_device(SELF_PN, SELF_DEVICE);
        let own_lid = Jid::lid_device(SELF_LID, SELF_DEVICE);
        let recipient_bare = Jid::lid(SELF_LID);
        assert!(is_self_dm_recipient(
            &recipient_bare,
            &own_pn,
            Some(&own_lid)
        ));

        let recipient_cached: Option<Vec<Jid>> = None;
        let own_cached_pn: Vec<Jid> = [0u16, 3, SELF_DEVICE]
            .into_iter()
            .map(|d| Jid::pn_device(SELF_PN, d))
            .collect();

        // Mirrors the call-site logic: we keep own_cached when recipient_cached is None
        // even in a self-DM.
        let keep_own = recipient_cached.is_none();
        assert!(keep_own);

        let mut all_dm_jids = match recipient_cached {
            Some(devices) => devices,
            None => vec![recipient_bare],
        };
        if keep_own {
            all_dm_jids.extend(own_cached_pn.iter().cloned());
        }
        all_dm_jids.retain(|j| {
            let is_sender = (j.is_same_user_as(&own_pn) && j.device == own_pn.device)
                || (j.is_same_user_as(&own_lid) && j.device == own_lid.device);
            !is_sender
        });
        wacore::types::jid::sort_dedup_by_device(&mut all_dm_jids);

        // Must contain the bare LID plus the two non-sender PN companion devices.
        assert!(
            all_dm_jids.iter().any(|j| j.is_lid()),
            "bare recipient LID must remain"
        );
        assert_eq!(
            all_dm_jids.iter().filter(|j| j.is_pn()).count(),
            2,
            "companion PN devices must survive when recipient_cached is None"
        );
    }

    #[test]
    fn old_merge_produced_lid_pn_duplicates_for_self_dm() {
        // Pinning regression: the OLD merge path (recipient_cached LID ++
        // own_cached PN, then sort_dedup_by_device) left every device listed
        // twice for a self-DM, which the server rejects with ack error="400".
        let own_pn = Jid::pn_device(SELF_PN, SELF_DEVICE);
        let own_lid = Jid::lid_device(SELF_LID, SELF_DEVICE);
        let recipient_bare = Jid::lid(SELF_LID);

        let devices = [0u16, 3, 5, SELF_DEVICE];
        let recipient_cached: Vec<Jid> = devices
            .iter()
            .map(|&d| Jid::lid_device(SELF_LID, d))
            .collect();
        let own_cached: Vec<Jid> = devices
            .iter()
            .map(|&d| Jid::pn_device(SELF_PN, d))
            .collect();

        let retain_non_sender = |j: &Jid| {
            let is_sender = (j.is_same_user_as(&own_pn) && j.device == own_pn.device)
                || (j.is_same_user_as(&own_lid) && j.device == own_lid.device);
            !is_sender
        };

        let mut buggy = recipient_cached.clone();
        buggy.extend(own_cached.clone());
        buggy.retain(retain_non_sender);
        wacore::types::jid::sort_dedup_by_device(&mut buggy);
        assert_eq!(buggy.len(), (devices.len() - 1) * 2);

        assert!(is_self_dm_recipient(
            &recipient_bare,
            &own_pn,
            Some(&own_lid)
        ));

        let mut fixed = recipient_cached;
        fixed.retain(retain_non_sender);
        wacore::types::jid::sort_dedup_by_device(&mut fixed);
        assert_eq!(fixed.len(), devices.len() - 1);
        for j in &fixed {
            assert!(j.is_lid());
        }
    }

    #[test]
    fn test_admin_revoke_preserves_lid_format() {
        // LID JIDs must NOT be converted to PN (phone number) format.
        // This was a bug that caused error 479 - the participant field must
        // preserve the original JID format exactly (with device stripped).
        let lid_sender = Jid::from_str("236395184570386:22@lid").unwrap();
        let participant_str = lid_sender.to_non_ad_string();

        // Must preserve @lid suffix, device number stripped
        assert_eq!(participant_str, "236395184570386@lid");
        assert!(
            participant_str.ends_with("@lid"),
            "LID participant must preserve @lid suffix"
        );
    }

    // SKDM Recipient Filtering Tests - validates DeviceKey-based filtering

    #[test]
    fn test_skdm_recipient_filtering_basic() {
        use std::collections::HashSet;

        let known_recipients: Vec<Jid> = [
            "1234567890:0@s.whatsapp.net",
            "1234567890:5@s.whatsapp.net",
            "9876543210:0@s.whatsapp.net",
        ]
        .into_iter()
        .map(|s| Jid::from_str(s).unwrap())
        .collect();

        let all_devices: Vec<Jid> = [
            "1234567890:0@s.whatsapp.net",
            "1234567890:5@s.whatsapp.net",
            "9876543210:0@s.whatsapp.net",
            "5555555555:0@s.whatsapp.net", // new
        ]
        .into_iter()
        .map(|s| Jid::from_str(s).unwrap())
        .collect();

        let known_set: HashSet<DeviceKey<'_>> =
            known_recipients.iter().map(|j| j.device_key()).collect();

        let new_devices: Vec<Jid> = all_devices
            .into_iter()
            .filter(|device| !known_set.contains(&device.device_key()))
            .collect();

        assert_eq!(new_devices.len(), 1);
        assert_eq!(new_devices[0].user, "5555555555");
    }

    #[test]
    fn test_skdm_recipient_filtering_lid_jids() {
        use std::collections::HashSet;

        let known_recipients: Vec<Jid> = [
            "236395184570386:91@lid",
            "129171292463295:0@lid",
            "45857667830004:14@lid",
        ]
        .into_iter()
        .map(|s| Jid::from_str(s).unwrap())
        .collect();

        let all_devices: Vec<Jid> = [
            "236395184570386:91@lid",
            "129171292463295:0@lid",
            "45857667830004:14@lid",
            "45857667830004:15@lid", // new
        ]
        .into_iter()
        .map(|s| Jid::from_str(s).unwrap())
        .collect();

        let known_set: HashSet<DeviceKey<'_>> =
            known_recipients.iter().map(|j| j.device_key()).collect();

        let new_devices: Vec<Jid> = all_devices
            .into_iter()
            .filter(|device| !known_set.contains(&device.device_key()))
            .collect();

        assert_eq!(new_devices.len(), 1);
        assert_eq!(new_devices[0].user, "45857667830004");
        assert_eq!(new_devices[0].device, 15);
    }

    #[test]
    fn test_skdm_recipient_filtering_all_known() {
        use std::collections::HashSet;

        let known_recipients: Vec<Jid> =
            ["1234567890:0@s.whatsapp.net", "1234567890:5@s.whatsapp.net"]
                .into_iter()
                .map(|s| Jid::from_str(s).unwrap())
                .collect();

        let all_devices: Vec<Jid> = ["1234567890:0@s.whatsapp.net", "1234567890:5@s.whatsapp.net"]
            .into_iter()
            .map(|s| Jid::from_str(s).unwrap())
            .collect();

        let known_set: HashSet<DeviceKey<'_>> =
            known_recipients.iter().map(|j| j.device_key()).collect();

        let new_devices: Vec<Jid> = all_devices
            .into_iter()
            .filter(|device| !known_set.contains(&device.device_key()))
            .collect();

        assert!(new_devices.is_empty());
    }

    #[test]
    fn test_skdm_recipient_filtering_all_new() {
        use std::collections::HashSet;

        let known_recipients: Vec<Jid> = vec![];

        let all_devices: Vec<Jid> = ["1234567890:0@s.whatsapp.net", "9876543210:0@s.whatsapp.net"]
            .into_iter()
            .map(|s| Jid::from_str(s).unwrap())
            .collect();

        let known_set: HashSet<DeviceKey<'_>> =
            known_recipients.iter().map(|j| j.device_key()).collect();

        let new_devices: Vec<Jid> = all_devices
            .clone()
            .into_iter()
            .filter(|device| !known_set.contains(&device.device_key()))
            .collect();

        assert_eq!(new_devices.len(), all_devices.len());
    }

    #[test]
    fn test_device_key_comparison() {
        // Jid parse/display normalizes :0 (omitted in Display, missing ':N' parses as device 0).
        // This test ensures DeviceKey comparisons work correctly under that normalization.
        let test_cases = [
            (
                "1234567890:0@s.whatsapp.net",
                "1234567890@s.whatsapp.net",
                true,
            ),
            (
                "1234567890:5@s.whatsapp.net",
                "1234567890:5@s.whatsapp.net",
                true,
            ),
            (
                "1234567890:5@s.whatsapp.net",
                "1234567890:6@s.whatsapp.net",
                false,
            ),
            ("236395184570386:91@lid", "236395184570386:91@lid", true),
            ("236395184570386:0@lid", "236395184570386@lid", true),
            ("user1@s.whatsapp.net", "user2@s.whatsapp.net", false),
        ];

        for (jid1_str, jid2_str, should_match) in test_cases {
            let jid1: Jid = jid1_str.parse().expect("should parse jid1");
            let jid2: Jid = jid2_str.parse().expect("should parse jid2");

            let key1 = jid1.device_key();
            let key2 = jid2.device_key();

            assert_eq!(
                key1 == key2,
                should_match,
                "DeviceKey comparison failed for '{}' vs '{}': expected match={}, got match={}",
                jid1_str,
                jid2_str,
                should_match,
                key1 == key2
            );

            assert_eq!(
                jid1.device_eq(&jid2),
                should_match,
                "device_eq failed for '{}' vs '{}'",
                jid1_str,
                jid2_str
            );
        }
    }

    #[test]
    fn empty_sender_key_device_map_marks_all_devices_for_skdm() {
        use crate::sender_key_device_cache::SenderKeyDeviceMap;

        let map = SenderKeyDeviceMap::from_db_rows(&[]);
        assert_eq!(map.device_has_key("271060335329480", 0), None);

        let all_resolved_devices: Vec<Jid> = [
            "271060335329480@lid",
            "77610646245392@lid",
            "276661023027320:5@lid",
        ]
        .into_iter()
        .map(|s| Jid::from_str(s).unwrap())
        .collect();

        let needs_skdm: Vec<&Jid> = all_resolved_devices
            .iter()
            .filter(|device| {
                !map.device_has_key(&device.user, device.device)
                    .unwrap_or(false)
                    || !map.device_has_key(&device.user, 0).unwrap_or(false)
            })
            .collect();

        assert_eq!(needs_skdm.len(), all_resolved_devices.len());
    }

    /// Fails if the empty-cache early-exit is reintroduced.
    #[tokio::test]
    async fn resolve_skdm_targets_distributes_when_cache_empty_but_devices_known() {
        use wacore::client::context::GroupInfo;
        use wacore::store::traits::{DeviceInfo, DeviceListRecord};
        use wacore::types::message::AddressingMode;

        let client = crate::test_utils::create_test_client().await;
        let group_jid = "120363161500776365@g.us";
        let own_lid = Jid::from_str("193832511623409:13@lid").unwrap();

        let participant_users = ["271060335329480", "77610646245392", "276661023027320"];

        // Pre-populate so `resolve_devices` succeeds without a transport.
        for user in &participant_users {
            let record = DeviceListRecord {
                user: (*user).into(),
                devices: vec![DeviceInfo::new(0, None)],
                timestamp: wacore::time::now_secs(),
                phash: None,
                raw_id: None,
            };
            client
                .device_registry_cache
                .raw_insert_for_tests((*user).into(), Arc::new(record))
                .await;
        }

        let participants: Vec<Jid> = participant_users
            .iter()
            .map(|u| Jid::from_str(&format!("{u}@lid")).unwrap())
            .collect();

        let group_info = GroupInfo::new(participants.clone(), AddressingMode::Lid);

        let needs_skdm = client
            .resolve_status_skdm_targets(
                group_jid,
                &group_info,
                &own_lid,
                crate::cache::Freshness::CachePreferred,
                false,
            )
            .await
            .expect("device resolution must succeed")
            .expect("missing targets means device resolution failed");

        // Empty cache → every participant needs SKDM, and the full set equals
        // the target set on this cold path.
        assert_eq!(needs_skdm.len(), participants.len());
        for user in &participant_users {
            assert!(needs_skdm.iter().any(|j| j.user == *user));
        }
    }

    #[test]
    fn single_forgotten_row_keeps_full_distribution() {
        use crate::sender_key_device_cache::SenderKeyDeviceMap;

        let map = SenderKeyDeviceMap::from_db_rows(&[("271060335329480@lid".to_string(), false)]);
        assert_eq!(map.device_has_key("271060335329480", 0), Some(false));

        let all_resolved_devices: Vec<Jid> = [
            "271060335329480@lid",
            "77610646245392@lid",
            "276661023027320:5@lid",
        ]
        .into_iter()
        .map(|s| Jid::from_str(s).unwrap())
        .collect();

        let needs_skdm: Vec<&Jid> = all_resolved_devices
            .iter()
            .filter(|device| {
                !map.device_has_key(&device.user, device.device)
                    .unwrap_or(false)
                    || !map.device_has_key(&device.user, 0).unwrap_or(false)
            })
            .collect();

        assert_eq!(
            needs_skdm.len(),
            3,
            "after retry inserts one row, ALL devices correctly flagged for SKDM \
             (this is what unblocks redistribution on the SECOND message)"
        );
    }

    /// WA Web primary-device gate (ParticipantStore.js): a companion is warm only
    /// when it AND its primary (device 0) hold the key. A forgotten companion
    /// redistributes only itself (no per-user amplification); a forgotten primary
    /// redistributes the whole user. Drives the real `filter_skdm_targets`.
    #[tokio::test]
    async fn filter_skdm_targets_uses_primary_device_gate() {
        use crate::sender_key_device_cache::SenderKeyDeviceMap;

        let client = crate::test_utils::create_test_client().await;
        let group = "120363161500776365@g.us";
        let own = Jid::from_str("999999999999999:1@lid").unwrap();

        // Companion forgotten, primary warm: only the companion redistributes.
        let map = SenderKeyDeviceMap::from_db_rows(&[
            ("100100100100100@lid".to_string(), true),
            ("100100100100100:5@lid".to_string(), false),
        ]);
        let devices = [
            Jid::from_str("100100100100100@lid").unwrap(),
            Jid::from_str("100100100100100:5@lid").unwrap(),
        ];
        let needs = client.filter_skdm_targets(group, &devices, &map, &own);
        assert_eq!(needs.len(), 1, "warm primary keeps the keyed companion out");
        assert_eq!(needs[0].device, 5);

        // Primary forgotten, companion warm: the whole user redistributes (WA Web
        // marks a companion cold when its primary is cold).
        let map = SenderKeyDeviceMap::from_db_rows(&[
            ("200200200200200@lid".to_string(), false),
            ("200200200200200:5@lid".to_string(), true),
        ]);
        let devices = [
            Jid::from_str("200200200200200@lid").unwrap(),
            Jid::from_str("200200200200200:5@lid").unwrap(),
        ];
        let needs = client.filter_skdm_targets(group, &devices, &map, &own);
        assert_eq!(needs.len(), 2, "cold primary redistributes the whole user");

        // Companion warm but the primary row is absent (None): WA Web's `?? false`
        // treats a missing primary as cold, so the companion still redistributes.
        let map = SenderKeyDeviceMap::from_db_rows(&[("300300300300300:5@lid".to_string(), true)]);
        let devices = [Jid::from_str("300300300300300:5@lid").unwrap()];
        let needs = client.filter_skdm_targets(group, &devices, &map, &own);
        assert_eq!(
            needs.len(),
            1,
            "absent primary is cold, companion redistributes"
        );
    }

    /// End-to-end: after a send marks its SKDM targets, our own companion is NOT
    /// memoized (WA Web `!isMeDevice` guard on `markHasSenderKey`), so the next send
    /// re-distributes its SKDM — it can't be orphaned by a one-off encryption
    /// failure (the retry/forget path also excludes own devices). An external member
    /// stays warm and is not re-targeted.
    #[tokio::test]
    async fn own_companion_is_never_memoized_so_it_redistributes_every_send() {
        use crate::sender_key_device_cache::SenderKeyDeviceMap;

        let client = crate::test_utils::create_test_client().await;
        let own_lid = Jid::from_str("888000888000888:1@lid").unwrap();
        client
            .persistence_manager
            .process_command(DeviceCommand::SetLid(Some(own_lid.clone())))
            .await;

        let group = "120363000000000009@g.us";
        let own_companion = Jid::from_str("888000888000888:5@lid").unwrap();
        let member = Jid::from_str("111000111000111@lid").unwrap();

        // A send marks its full target set warm (own companion + external member).
        client
            .update_sender_key_devices(group, &[own_companion.clone(), member.clone()])
            .await;

        // Persisted: the external member is warm; our own companion was skipped.
        let rows = client
            .persistence_manager
            .get_sender_key_devices(group)
            .await
            .unwrap();
        let map = SenderKeyDeviceMap::from_db_rows(&rows);
        assert_eq!(
            map.device_has_key("111000111000111", 0),
            Some(true),
            "external member is memoized warm"
        );
        assert_eq!(
            map.device_has_key("888000888000888", 5),
            None,
            "own companion is never memoized (re-distributed every send)"
        );

        // Next send: only the own companion is re-targeted; the member stays warm.
        let devices = [own_companion.clone(), member];
        let needs = client.filter_skdm_targets(group, &devices, &map, &own_lid);
        assert_eq!(
            needs,
            vec![own_companion],
            "own companion redistributes; external member stays warm"
        );
    }

    /// The own-only re-distribution set is the warm steady state (WA Web:
    /// `getGroupSenderKeyList` reads the in-memory map with no storage
    /// re-read), so marking it after a send must NOT drop the cached device
    /// map. A set with an external member writes a new warm mark and must.
    #[tokio::test]
    async fn own_only_skdm_mark_keeps_device_map_cached() {
        use crate::sender_key_device_cache::SenderKeyDeviceMap;
        use std::sync::Arc;

        let client = crate::test_utils::create_test_client().await;
        let own_lid = Jid::from_str("888000888000888:1@lid").unwrap();
        client
            .persistence_manager
            .process_command(DeviceCommand::SetLid(Some(own_lid.clone())))
            .await;

        let group = "120363000000000010@g.us";
        let own_primary = Jid::from_str("888000888000888:0@lid").unwrap();
        let member = Jid::from_str("111000111000111:0@lid").unwrap();

        client
            .sender_key_device_cache
            .get_or_init(group, async {
                Arc::new(SenderKeyDeviceMap::from_db_rows(&[]))
            })
            .await;

        // Own-only mark (the every-send steady state): nothing written, cache kept.
        client
            .update_sender_key_devices(group, std::slice::from_ref(&own_primary))
            .await;
        client
            .sender_key_device_cache
            .get_or_init(group, async {
                panic!("own-only SKDM mark must not invalidate the device map")
            })
            .await;

        // An external member writes a warm mark: the cached map must drop so
        // the next send re-reads the new state.
        client
            .update_sender_key_devices(group, &[own_primary, member])
            .await;
        let rebuilt = std::sync::atomic::AtomicBool::new(false);
        client
            .sender_key_device_cache
            .get_or_init(group, async {
                rebuilt.store(true, std::sync::atomic::Ordering::Relaxed);
                Arc::new(SenderKeyDeviceMap::from_db_rows(&[]))
            })
            .await;
        assert!(
            rebuilt.load(std::sync::atomic::Ordering::Relaxed),
            "a new external warm mark must invalidate the device map"
        );
    }

    /// `skdm_needs_only_own_devices` gates the warm fast path in
    /// `send_group_branch`: own-only sets qualify, anything external (or an
    /// empty set, which has its own arm) does not.
    #[test]
    fn skdm_needs_only_own_devices_classification() {
        let own_pn = Jid::from_str("5511999990000:2@s.whatsapp.net").unwrap();
        let own_lid = Jid::from_str("888000888000888:2@lid").unwrap();
        let own_primary_lid = Jid::from_str("888000888000888:0@lid").unwrap();
        let own_primary_pn = Jid::from_str("5511999990000:0@s.whatsapp.net").unwrap();
        let member = Jid::from_str("111000111000111:0@lid").unwrap();

        assert!(skdm_needs_only_own_devices(
            &[own_primary_lid.clone(), own_primary_pn],
            Some(&own_pn),
            Some(&own_lid)
        ));
        assert!(
            !skdm_needs_only_own_devices(
                &[own_primary_lid, member.clone()],
                Some(&own_pn),
                Some(&own_lid)
            ),
            "an external member must take the cold path"
        );
        assert!(
            !skdm_needs_only_own_devices(&[], Some(&own_pn), Some(&own_lid)),
            "the empty set is handled by its own (fully warm) arm"
        );
        assert!(
            !skdm_needs_only_own_devices(&[member], Some(&own_pn), Some(&own_lid)),
            "external-only must take the cold path"
        );
    }

    #[test]
    fn test_skdm_filtering_large_group() {
        use std::collections::HashSet;

        let mut known_recipients: Vec<Jid> = Vec::with_capacity(1000);
        let mut all_devices: Vec<Jid> = Vec::with_capacity(1010);

        for i in 0..1000i64 {
            let jid_str = format!("{}:1@lid", 100000000000000i64 + i);
            let jid = Jid::from_str(&jid_str).unwrap();
            known_recipients.push(jid.clone());
            all_devices.push(jid);
        }

        for i in 1000i64..1010i64 {
            let jid_str = format!("{}:1@lid", 100000000000000i64 + i);
            all_devices.push(Jid::from_str(&jid_str).unwrap());
        }

        let known_set: HashSet<DeviceKey<'_>> =
            known_recipients.iter().map(|j| j.device_key()).collect();

        let new_devices: Vec<Jid> = all_devices
            .into_iter()
            .filter(|device| !known_set.contains(&device.device_key()))
            .collect();

        assert_eq!(new_devices.len(), 10);
    }

    mod infer_stanza {
        use super::*;

        #[test]
        fn regular_message_returns_none() {
            let msg = wa::Message {
                conversation: Some("hello".into()),
                ..Default::default()
            };
            let (edit, node) = infer_stanza_metadata(&msg);
            assert!(edit.is_none());
            assert!(node.is_none());
        }

        #[test]
        fn pin_returns_edit_attribute() {
            let msg = wa::Message {
                pin_in_chat_message: buffa::MessageField::some(Default::default()),
                ..Default::default()
            };
            let (edit, node) = infer_stanza_metadata(&msg);
            assert_eq!(edit, Some(EditAttribute::PinInChat));
            assert!(node.is_none());
        }

        #[test]
        fn poll_creation_v3_returns_meta_node() {
            let msg = wa::Message {
                poll_creation_message_v3: buffa::MessageField::some(Default::default()),
                ..Default::default()
            };
            let (edit, node) = infer_stanza_metadata(&msg);
            assert!(edit.is_none());
            let node = node.expect("should have meta node");
            assert_eq!(node.tag, "meta");
            let mut attrs = node.attrs();
            assert_eq!(
                attrs.optional_string("polltype").unwrap().as_ref(),
                "creation"
            );
        }

        #[test]
        fn event_returns_meta_node() {
            let msg = wa::Message {
                event_message: buffa::MessageField::some(Default::default()),
                ..Default::default()
            };
            let (edit, node) = infer_stanza_metadata(&msg);
            assert!(edit.is_none());
            let node = node.expect("should have meta node");
            assert_eq!(node.tag, "meta");
            let mut attrs = node.attrs();
            assert_eq!(
                attrs.optional_string("event_type").unwrap().as_ref(),
                "creation"
            );
        }

        #[test]
        fn empty_message_returns_none() {
            let (edit, node) = infer_stanza_metadata(&wa::Message::default());
            assert!(edit.is_none());
            assert!(node.is_none());
        }

        #[test]
        fn member_label_set_returns_member_tag_user_update() {
            let msg = wacore::send::build_member_label_message("VIP".to_string(), 1_700_000_000);
            let (_, node) = infer_stanza_metadata(&msg);
            let node = node.expect("member_label should have meta node");
            let mut attrs = node.attrs();
            assert_eq!(
                attrs.optional_string("appdata").unwrap().as_ref(),
                "member_tag"
            );
            assert_eq!(
                attrs.optional_string("tag_reason").unwrap().as_ref(),
                "user_update"
            );
        }

        #[test]
        fn member_label_clear_returns_user_delete() {
            // Empty label = clearing the tag → tag_reason "user_delete".
            let msg = wacore::send::build_member_label_message(String::new(), 1_700_000_000);
            let (_, node) = infer_stanza_metadata(&msg);
            let node = node.expect("member_label should have meta node");
            let mut attrs = node.attrs();
            assert_eq!(
                attrs.optional_string("appdata").unwrap().as_ref(),
                "member_tag"
            );
            assert_eq!(
                attrs.optional_string("tag_reason").unwrap().as_ref(),
                "user_delete"
            );
        }

        #[test]
        fn poll_creation_v1_returns_meta_node() {
            let msg = wa::Message {
                poll_creation_message: buffa::MessageField::some(Default::default()),
                ..Default::default()
            };
            let (edit, node) = infer_stanza_metadata(&msg);
            assert!(edit.is_none());
            let node = node.expect("should have meta node");
            assert_eq!(node.tag, "meta");
            let mut attrs = node.attrs();
            assert_eq!(
                attrs.optional_string("polltype").unwrap().as_ref(),
                "creation"
            );
        }

        #[test]
        fn poll_creation_v2_returns_meta_node() {
            let msg = wa::Message {
                poll_creation_message_v2: buffa::MessageField::some(Default::default()),
                ..Default::default()
            };
            let (edit, node) = infer_stanza_metadata(&msg);
            assert!(edit.is_none());
            let node = node.expect("should have meta node");
            assert_eq!(node.tag, "meta");
            let mut attrs = node.attrs();
            assert_eq!(
                attrs.optional_string("polltype").unwrap().as_ref(),
                "creation"
            );
        }

        #[test]
        fn poll_vote_returns_meta_node() {
            let msg = wa::Message {
                poll_update_message: buffa::MessageField::some(wa::message::PollUpdateMessage {
                    vote: buffa::MessageField::some(wa::message::PollEncValue::default()),
                    ..Default::default()
                }),
                ..Default::default()
            };
            let (edit, node) = infer_stanza_metadata(&msg);
            assert!(edit.is_none());
            let node = node.expect("should have meta node");
            assert_eq!(node.tag, "meta");
            let mut attrs = node.attrs();
            assert_eq!(attrs.optional_string("polltype").unwrap().as_ref(), "vote");
        }

        #[test]
        fn view_once_image_emits_view_once_meta() {
            let msg = wa::Message {
                image_message: buffa::MessageField::some(wa::message::ImageMessage {
                    view_once: Some(true),
                    ..Default::default()
                }),
                ..Default::default()
            };
            let (_, node) = infer_stanza_metadata(&msg);
            let node = node.expect("view-once image should emit meta");
            assert_eq!(node.tag, "meta");
            assert_eq!(
                node.attrs().optional_string("view_once").unwrap().as_ref(),
                "true"
            );
        }

        #[test]
        fn plain_image_emits_no_meta() {
            let msg = wa::Message {
                image_message: buffa::MessageField::some(wa::message::ImageMessage::default()),
                ..Default::default()
            };
            assert!(infer_stanza_metadata(&msg).1.is_none());
        }

        #[test]
        fn event_response_returns_meta_node() {
            let msg = wa::Message {
                enc_event_response_message: buffa::MessageField::some(Default::default()),
                ..Default::default()
            };
            let (edit, node) = infer_stanza_metadata(&msg);
            assert!(edit.is_none());
            let node = node.expect("should have meta node");
            assert_eq!(node.tag, "meta");
            let mut attrs = node.attrs();
            assert_eq!(
                attrs.optional_string("event_type").unwrap().as_ref(),
                "response"
            );
        }

        #[test]
        fn poll_update_without_vote_returns_none() {
            let msg = wa::Message {
                poll_update_message: buffa::MessageField::some(wa::message::PollUpdateMessage {
                    ..Default::default()
                }),
                ..Default::default()
            };
            let (edit, node) = infer_stanza_metadata(&msg);
            assert!(edit.is_none());
            assert!(node.is_none());
        }

        #[test]
        fn revoked_reaction_returns_sender_revoke() {
            let msg = wa::Message {
                reaction_message: buffa::MessageField::some(wa::message::ReactionMessage {
                    text: Some(String::new()),
                    ..Default::default()
                }),
                ..Default::default()
            };
            let (edit, _) = infer_stanza_metadata(&msg);
            assert_eq!(edit, Some(EditAttribute::SenderRevoke));
        }

        #[test]
        fn keep_in_chat_undo_returns_sender_revoke() {
            let msg = wa::Message {
                keep_in_chat_message: buffa::MessageField::some(wa::message::KeepInChatMessage {
                    key: buffa::MessageField::some(wa::MessageKey {
                        from_me: Some(true),
                        ..Default::default()
                    }),
                    keep_type: Some(wa::KeepType::UndoKeepForAll),
                    ..Default::default()
                }),
                ..Default::default()
            };
            let (edit, _) = infer_stanza_metadata(&msg);
            assert_eq!(edit, Some(EditAttribute::SenderRevoke));
        }

        #[test]
        fn secret_encrypted_message_edit_returns_message_edit() {
            let msg = wa::Message {
                secret_encrypted_message: buffa::MessageField::some(
                    wa::message::SecretEncryptedMessage {
                        secret_enc_type: Some(
                            wa::message::secret_encrypted_message::SecretEncType::MessageEdit,
                        ),
                        ..Default::default()
                    },
                ),
                ..Default::default()
            };
            let (edit, _) = infer_stanza_metadata(&msg);
            assert_eq!(edit, Some(EditAttribute::MessageEdit));
        }

        #[test]
        fn secret_encrypted_event_edit_emits_both_edit_attr_and_meta_node() {
            // EVENT_EDIT is the one case where the edit attribute AND the
            // meta node both fire: `event_type=edit` meta + `edit="1"` attr.
            let msg = wa::Message {
                secret_encrypted_message: buffa::MessageField::some(
                    wa::message::SecretEncryptedMessage {
                        secret_enc_type: Some(
                            wa::message::secret_encrypted_message::SecretEncType::EventEdit,
                        ),
                        ..Default::default()
                    },
                ),
                ..Default::default()
            };
            let (edit, node) = infer_stanza_metadata(&msg);
            assert_eq!(edit, Some(EditAttribute::MessageEdit));
            let node = node.expect("should have meta node");
            assert_eq!(
                node.attrs().optional_string("event_type").unwrap().as_ref(),
                "edit"
            );
        }

        #[test]
        fn top_level_edited_message_returns_message_edit() {
            let msg = wa::Message {
                edited_message: buffa::MessageField::some(wa::message::FutureProofMessage {
                    message: buffa::MessageField::some(wa::Message::default()),
                }),
                ..Default::default()
            };
            let (edit, _) = infer_stanza_metadata(&msg);
            assert_eq!(edit, Some(EditAttribute::MessageEdit));
        }

        #[test]
        fn build_edit_message_uses_top_level_protocol_message() {
            use std::str::FromStr;
            let to = Jid::from_str("5511999999999@s.whatsapp.net").unwrap();
            let new_content = wa::Message {
                conversation: Some("edited".to_string()),
                ..Default::default()
            };
            let msg = build_edit_message(
                &to,
                "ORIG_ID".to_string(),
                None,
                new_content,
                1_700_000_000_000,
            );

            // Canonical WA Web shape: top-level protocolMessage(type=MESSAGE_EDIT),
            // not the Message.editedMessage FutureProofMessage history wrapper.
            assert!(
                msg.edited_message.is_unset(),
                "edit must not use the FutureProofMessage wrapper"
            );
            let pm = msg
                .protocol_message
                .as_option()
                .expect("top-level protocol_message");
            assert_eq!(
                pm.r#type,
                Some(wa::message::protocol_message::Type::MessageEdit)
            );
            assert_eq!(
                pm.key.as_option().and_then(|k| k.id.as_deref()),
                Some("ORIG_ID")
            );
            assert_eq!(pm.key.as_option().and_then(|k| k.from_me), Some(true));
            assert_eq!(
                pm.edited_message
                    .as_option()
                    .and_then(|m| m.conversation.as_deref()),
                Some("edited")
            );
            // The send path still derives the edit attribute from this shape.
            assert_eq!(
                infer_stanza_metadata(&msg).0,
                Some(EditAttribute::MessageEdit)
            );
        }
    }

    mod biz_node_tests {
        use super::*;
        use std::str::FromStr;
        use wa::message::interactive_message::{
            self, NativeFlowMessage, native_flow_message::NativeFlowButton,
        };

        // Fixed unix seconds for deterministic privacy_mode_ts assertions.
        const FIXED_NOW: u64 = 1_700_000_000;
        // FIXED_NOW - BIZ_PRIVACY_MODE_TS_OFFSET = 1_700_000_000 - 77_980_457
        const EXPECTED_PRIVACY_TS: &str = "1622019543";

        fn msg_with_native_flow_button(button_name: &str) -> wa::Message {
            wa::Message {
                interactive_message: buffa::MessageField::some(wa::message::InteractiveMessage {
                    interactive_message: Some(
                        interactive_message::InteractiveMessage::NativeFlowMessage(Box::new(
                            NativeFlowMessage {
                                buttons: vec![NativeFlowButton {
                                    name: Some(button_name.to_string()),
                                    button_params_json: None,
                                }],
                                message_version: Some(1),
                                message_params_json: None,
                            },
                        )),
                    ),
                    ..Default::default()
                }),
                ..Default::default()
            }
        }

        fn assert_biz_common_attrs(node: &Node, ctx: &str) {
            assert_eq!(node.tag, "biz", "{ctx}");
            let mut a = node.attrs();
            assert_eq!(
                a.optional_string("actual_actors").unwrap().as_ref(),
                "2",
                "{ctx}"
            );
            assert_eq!(
                a.optional_string("host_storage").unwrap().as_ref(),
                "2",
                "{ctx}"
            );
            assert_eq!(
                a.optional_string("privacy_mode_ts").unwrap().as_ref(),
                EXPECTED_PRIVACY_TS,
                "{ctx}"
            );
        }

        fn assert_nested_biz(node: &Node, expected_flow_name: &str, ctx: &str) {
            assert_biz_common_attrs(node, ctx);
            assert!(
                node.attrs().optional_string("native_flow_name").is_none(),
                "{ctx}: nested form has no native_flow_name attr"
            );
            let interactive = node
                .get_optional_child("interactive")
                .unwrap_or_else(|| panic!("{ctx}: missing <interactive>"));
            let mut ia = interactive.attrs();
            assert_eq!(
                ia.optional_string("type").unwrap().as_ref(),
                "native_flow",
                "{ctx}"
            );
            assert_eq!(ia.optional_string("v").unwrap().as_ref(), "1", "{ctx}");

            let nf = interactive
                .get_optional_child("native_flow")
                .unwrap_or_else(|| panic!("{ctx}: missing <native_flow>"));
            let mut nfa = nf.attrs();
            assert_eq!(nfa.optional_string("v").unwrap().as_ref(), "9", "{ctx}");
            assert_eq!(
                nfa.optional_string("name").unwrap().as_ref(),
                expected_flow_name,
                "{ctx}"
            );

            let qc = node
                .get_optional_child("quality_control")
                .unwrap_or_else(|| panic!("{ctx}: missing <quality_control>"));
            assert_eq!(
                qc.attrs().optional_string("source_type").unwrap().as_ref(),
                "third_party",
                "{ctx}"
            );
        }

        /// Payment-family buttons emit the flat `<biz>` form with
        /// `native_flow_name` as an attr and NO children.
        #[test]
        fn payment_simple_form() {
            let cases: &[(&str, &str)] = &[
                ("payment_info", "payment_info"),
                ("review_and_pay", "order_details"),
                ("review_order", "order_status"),
                ("order_status", "order_status"),
                ("payment_status", "payment_status"),
                ("payment_method", "payment_method"),
                ("payment_reminder", "payment_reminder"),
            ];
            for (button, expected_flow) in cases {
                let biz = infer_biz_node(&msg_with_native_flow_button(button), FIXED_NOW)
                    .unwrap_or_else(|| panic!("{button}: should produce biz"));
                assert_biz_common_attrs(&biz, button);
                assert_eq!(
                    biz.attrs()
                        .optional_string("native_flow_name")
                        .unwrap()
                        .as_ref(),
                    *expected_flow,
                    "{button}: native_flow_name attr"
                );
                assert!(
                    biz.children().unwrap_or(&[]).is_empty(),
                    "{button}: PaymentSimple has no children"
                );
            }
        }

        /// Every non-payment button name announces itself as `mixed`. The
        /// eight names that used to keep their own flow name are the ones
        /// #1132 measured as universally refused (473/405), so they must not
        /// regain a bespoke shape without fresh live evidence.
        #[test]
        fn formerly_named_buttons_now_route_through_mixed() {
            let cases: &[&str] = &[
                "cta_url",
                "cta_catalog",
                "catalog_message",
                "galaxy_message",
                "booking_confirmation",
                "call_permission_request",
                "open_webview",
                "message_with_link_status",
            ];
            for button in cases {
                let biz = infer_biz_node(&msg_with_native_flow_button(button), FIXED_NOW)
                    .unwrap_or_else(|| panic!("{button}: should produce biz"));
                assert_nested_biz(&biz, "mixed", button);
            }
        }

        /// quick_reply / cta_copy / cta_call / single_select / send_location
        /// and unknown future button names route through `name="mixed"`.
        #[test]
        fn mixed_form_for_dropped_buttons() {
            let cases: &[&str] = &[
                "quick_reply",
                "cta_copy",
                "cta_call",
                "single_select",
                "send_location",
                "future_button_xyz",
            ];
            for button in cases {
                let biz = infer_biz_node(&msg_with_native_flow_button(button), FIXED_NOW)
                    .unwrap_or_else(|| panic!("{button}: should produce biz"));
                assert_nested_biz(&biz, "mixed", button);
            }
        }

        /// Non-interactive messages produce no `<biz>` (no fan-out into the
        /// extra_stanza_nodes path).
        #[test]
        fn no_interactive_returns_none() {
            let msg = wa::Message {
                conversation: Some("hello".into()),
                ..Default::default()
            };
            assert!(infer_biz_node(&msg, FIXED_NOW).is_none());
        }

        fn carousel_msg(im: wa::message::InteractiveMessage) -> wa::Message {
            wa::Message {
                interactive_message: buffa::MessageField::some(wa::message::InteractiveMessage {
                    interactive_message: Some(
                        interactive_message::InteractiveMessage::CarouselMessage(Default::default()),
                    ),
                    ..im
                }),
                ..Default::default()
            }
        }

        fn body(text: &str) -> buffa::MessageField<interactive_message::Body> {
            buffa::MessageField::some(interactive_message::Body {
                text: Some(text.to_string()),
            })
        }

        /// #1133: a carousel's buttons live on its cards, so the button rule
        /// never fires. Without this the message left with no `<biz>` at all —
        /// accepted, acked, and then invisible on the recipient's handset.
        #[test]
        fn carousel_with_body_emits_mixed_biz() {
            let msg = carousel_msg(wa::message::InteractiveMessage {
                body: body("pick a plan"),
                ..Default::default()
            });
            let biz = infer_biz_node(&msg, FIXED_NOW).expect("carousel should produce biz");
            assert_nested_biz(&biz, "mixed", "carousel");
        }

        /// A header title alone is enough, and so is a footer — WA Web's rule
        /// is a disjunction over body / title / footer / header image.
        #[test]
        fn carousel_envelope_variants_emit_mixed_biz() {
            let title = carousel_msg(wa::message::InteractiveMessage {
                header: buffa::MessageField::some(interactive_message::Header {
                    title: Some("Our menu".into()),
                    ..Default::default()
                }),
                ..Default::default()
            });
            assert_nested_biz(
                &infer_biz_node(&title, FIXED_NOW).expect("title should produce biz"),
                "mixed",
                "header title",
            );

            let footer = carousel_msg(wa::message::InteractiveMessage {
                footer: buffa::MessageField::some(interactive_message::Footer {
                    text: Some("tap to order".into()),
                    ..Default::default()
                }),
                ..Default::default()
            });
            assert_nested_biz(
                &infer_biz_node(&footer, FIXED_NOW).expect("footer should produce biz"),
                "mixed",
                "footer",
            );

            let image = carousel_msg(wa::message::InteractiveMessage {
                header: buffa::MessageField::some(interactive_message::Header {
                    media: Some(interactive_message::header::Media::ImageMessage(
                        Default::default(),
                    )),
                    ..Default::default()
                }),
                ..Default::default()
            });
            assert_nested_biz(
                &infer_biz_node(&image, FIXED_NOW).expect("header image should produce biz"),
                "mixed",
                "header image",
            );
        }

        /// An empty envelope announces nothing, so there is nothing to mark.
        #[test]
        fn carousel_without_envelope_returns_none() {
            assert!(infer_biz_node(&carousel_msg(Default::default()), FIXED_NOW).is_none());
        }

        /// An empty-string body is not an envelope: WA Web tests `length > 0`,
        /// not presence.
        #[test]
        fn empty_body_text_is_not_an_envelope() {
            let msg = carousel_msg(wa::message::InteractiveMessage {
                body: body(""),
                ..Default::default()
            });
            assert!(infer_biz_node(&msg, FIXED_NOW).is_none());
        }

        /// Storefronts are excluded from the envelope rule, as in WA Web.
        #[test]
        fn shop_storefront_returns_none() {
            let msg = wa::Message {
                interactive_message: buffa::MessageField::some(wa::message::InteractiveMessage {
                    body: body("visit our shop"),
                    interactive_message: Some(
                        interactive_message::InteractiveMessage::ShopStorefrontMessage(
                            Default::default(),
                        ),
                    ),
                    ..Default::default()
                }),
                ..Default::default()
            };
            assert!(infer_biz_node(&msg, FIXED_NOW).is_none());
        }

        /// Interactive but not native-flow (e.g. CollectionMessage) yields None.
        #[test]
        fn interactive_without_native_flow_returns_none() {
            let msg = wa::Message {
                interactive_message: buffa::MessageField::some(wa::message::InteractiveMessage {
                    interactive_message: Some(
                        interactive_message::InteractiveMessage::CollectionMessage(
                            Default::default(),
                        ),
                    ),
                    ..Default::default()
                }),
                ..Default::default()
            };
            assert!(infer_biz_node(&msg, FIXED_NOW).is_none());
        }

        /// NativeFlow with empty button list yields None — no signal to classify.
        #[test]
        fn native_flow_without_buttons_returns_none() {
            let msg = wa::Message {
                interactive_message: buffa::MessageField::some(wa::message::InteractiveMessage {
                    interactive_message: Some(
                        interactive_message::InteractiveMessage::NativeFlowMessage(Box::new(
                            NativeFlowMessage {
                                buttons: vec![],
                                message_version: Some(1),
                                message_params_json: None,
                            },
                        )),
                    ),
                    ..Default::default()
                }),
                ..Default::default()
            };
            assert!(infer_biz_node(&msg, FIXED_NOW).is_none());
        }

        /// Button with `name = None` is treated as missing classifier → None.
        #[test]
        fn button_without_name_returns_none() {
            let msg = wa::Message {
                interactive_message: buffa::MessageField::some(wa::message::InteractiveMessage {
                    interactive_message: Some(
                        interactive_message::InteractiveMessage::NativeFlowMessage(Box::new(
                            NativeFlowMessage {
                                buttons: vec![NativeFlowButton {
                                    name: None,
                                    button_params_json: None,
                                }],
                                message_version: Some(1),
                                message_params_json: None,
                            },
                        )),
                    ),
                    ..Default::default()
                }),
                ..Default::default()
            };
            assert!(infer_biz_node(&msg, FIXED_NOW).is_none());
        }

        /// Messages wrapped in `documentWithCaptionMessage` still pick up the
        /// native_flow payload from the inner message.
        #[test]
        fn document_with_caption_wrapper() {
            let inner = wa::Message {
                interactive_message: buffa::MessageField::some(wa::message::InteractiveMessage {
                    interactive_message: Some(
                        interactive_message::InteractiveMessage::NativeFlowMessage(Box::new(
                            NativeFlowMessage {
                                buttons: vec![NativeFlowButton {
                                    name: Some("quick_reply".into()),
                                    button_params_json: None,
                                }],
                                message_version: Some(1),
                                message_params_json: None,
                            },
                        )),
                    ),
                    ..Default::default()
                }),
                ..Default::default()
            };
            let msg = wa::Message {
                document_with_caption_message: buffa::MessageField::some(
                    wa::message::FutureProofMessage {
                        message: buffa::MessageField::some(inner),
                    },
                ),
                ..Default::default()
            };
            let biz = infer_biz_node(&msg, FIXED_NOW)
                .expect("doc-with-caption wrapper should propagate the inner native_flow");
            assert_nested_biz(&biz, "mixed", "doc-with-caption/quick_reply");
        }

        // -- build_extra_stanza_nodes assembly tests --

        fn quick_reply_biz() -> Node {
            infer_biz_node(&msg_with_native_flow_button("quick_reply"), FIXED_NOW)
                .expect("quick_reply produces biz")
        }

        fn payment_biz() -> Node {
            infer_biz_node(&msg_with_native_flow_button("payment_info"), FIXED_NOW)
                .expect("payment_info produces biz")
        }

        fn jid(s: &str) -> Jid {
            Jid::from_str(s).expect("valid jid in test")
        }

        /// DM: `<bot biz_bot="1"/>` is prepended before the `<biz>`. The
        /// order matters because it is part of the wire shape.
        #[test]
        fn dm_emits_bot_before_biz() {
            let nodes = build_extra_stanza_nodes(
                &jid("5511999999999@s.whatsapp.net"),
                None,
                Some(quick_reply_biz()),
                vec![],
            );
            assert_eq!(nodes.len(), 2, "expected [<bot>, <biz>]");
            assert_eq!(nodes[0].tag, "bot");
            assert_eq!(
                nodes[0]
                    .attrs()
                    .optional_string("biz_bot")
                    .unwrap()
                    .as_ref(),
                "1"
            );
            assert_eq!(nodes[1].tag, "biz");
        }

        /// Group: `<bot>` is NOT emitted; only `<biz>`.
        #[test]
        fn group_omits_bot() {
            let nodes = build_extra_stanza_nodes(
                &jid("120363000000000001@g.us"),
                None,
                Some(quick_reply_biz()),
                vec![],
            );
            assert_eq!(nodes.len(), 1);
            assert_eq!(nodes[0].tag, "biz");
        }

        /// LID DM (non-group): `<bot>` is still emitted.
        #[test]
        fn lid_dm_emits_bot() {
            let nodes = build_extra_stanza_nodes(
                &jid("100000000000001@lid"),
                None,
                Some(payment_biz()),
                vec![],
            );
            assert_eq!(nodes.len(), 2);
            assert_eq!(nodes[0].tag, "bot");
        }

        /// No biz + no meta → user nodes pass through untouched.
        #[test]
        fn no_biz_no_meta_passthrough() {
            let user_nodes = vec![NodeBuilder::new("custom").build()];
            let nodes =
                build_extra_stanza_nodes(&jid("X@s.whatsapp.net"), None, None, user_nodes.clone());
            assert_eq!(nodes.len(), 1);
            assert_eq!(nodes[0].tag, "custom");
        }

        /// Full ordering: [meta, bot, biz, user_nodes...].
        #[test]
        fn full_ordering_meta_bot_biz_user() {
            let meta = NodeBuilder::new("meta").attr("appdata", "default").build();
            let user_a = NodeBuilder::new("user_a").build();
            let user_b = NodeBuilder::new("user_b").build();
            let nodes = build_extra_stanza_nodes(
                &jid("X@s.whatsapp.net"),
                Some(meta),
                Some(quick_reply_biz()),
                vec![user_a, user_b],
            );
            assert_eq!(nodes.len(), 5);
            assert_eq!(nodes[0].tag, "meta");
            assert_eq!(nodes[1].tag, "bot");
            assert_eq!(nodes[2].tag, "biz");
            assert_eq!(nodes[3].tag, "user_a");
            assert_eq!(nodes[4].tag, "user_b");
        }

        /// Meta-only (no biz) preserves order: meta then user nodes; no bot.
        #[test]
        fn meta_only_preserves_order() {
            let meta = NodeBuilder::new("meta").build();
            let user = NodeBuilder::new("u").build();
            let nodes =
                build_extra_stanza_nodes(&jid("X@s.whatsapp.net"), Some(meta), None, vec![user]);
            assert_eq!(nodes.len(), 2);
            assert_eq!(nodes[0].tag, "meta");
            assert_eq!(nodes[1].tag, "u");
        }
    }

    #[test]
    fn structural_extra_children_are_rejected_before_send_work() {
        for tag in RESERVED_EXTRA_STANZA_CHILDREN {
            let error = validate_extra_stanza_nodes(&[NodeBuilder::new(tag).build()])
                .expect_err("send-owned child must be rejected");
            assert!(error.to_string().contains(tag));
        }

        validate_extra_stanza_nodes(&[
            NodeBuilder::new("meta").build(),
            NodeBuilder::new("biz").build(),
            NodeBuilder::new("custom-extension").build(),
        ])
        .expect("non-structural protocol extensions remain available");
    }

    /// Regression tests for #462: send path session lock keys must match decrypt path.
    mod session_lock_regression {
        use super::*;

        #[tokio::test]
        async fn per_device_lock_keys_cover_all_devices() {
            let client = crate::test_utils::create_test_client().await;

            let devices: Vec<Jid> = [
                "100000012345678@lid",
                "100000012345678:5@lid",
                "100000012345678:33@lid",
            ]
            .iter()
            .map(|s| Jid::from_str(s).unwrap())
            .collect();

            // Uses the production helper (resolve_encryption_jid + sort + dedup)
            let send_lock_keys = client.build_session_lock_keys(&devices).await;

            assert_eq!(send_lock_keys.len(), 3);
            // Sorted by (server, user, device_numeric): 0, 5, 33
            assert_eq!(send_lock_keys[0].device, 0);
            assert_eq!(send_lock_keys[1].device, 5);
            assert_eq!(send_lock_keys[2].device, 33);

            // Send keys must cover every device
            for device_jid in &devices {
                assert!(
                    send_lock_keys.contains(device_jid),
                    "device {device_jid} not in send keys: {send_lock_keys:?}"
                );
            }

            // Bare JID key alone wouldn't protect linked devices
            let bare_key = devices[0].to_protocol_address_string();
            let device5_key = devices[1].to_protocol_address_string();
            assert_ne!(bare_key, device5_key);
        }

        #[tokio::test]
        async fn per_device_lock_serializes_concurrent_session_access() {
            use std::sync::Arc;
            use std::sync::atomic::{AtomicU32, Ordering};

            let session_locks: crate::cache::Cache<String, Arc<async_lock::Mutex<()>>> =
                crate::cache::Cache::builder().max_capacity(100).build();

            let lock_key = "100000012345678:5@lid.0".to_string();
            let access_counter = Arc::new(AtomicU32::new(0));
            let max_concurrent = Arc::new(AtomicU32::new(0));

            let mut handles = Vec::new();
            for _ in 0..10 {
                let locks = session_locks.clone();
                let key = lock_key.clone();
                let counter = access_counter.clone();
                let max = max_concurrent.clone();

                handles.push(tokio::spawn(async move {
                    let mutex: Arc<async_lock::Mutex<()>> = locks
                        .get_with_by_ref(&key, async { Arc::new(async_lock::Mutex::new(())) })
                        .await;
                    // lock_arc() needed: guard must own the Arc since mutex is a local
                    // (production uses lock() with a separate Vec keeping Arcs alive)
                    let _guard = mutex.lock_arc().await;

                    let active = counter.fetch_add(1, Ordering::SeqCst) + 1;
                    max.fetch_max(active, Ordering::SeqCst);
                    tokio::task::yield_now().await;
                    counter.fetch_sub(1, Ordering::SeqCst);
                }));
            }

            for handle in handles {
                handle.await.unwrap();
            }

            assert_eq!(max_concurrent.load(Ordering::SeqCst), 1);
        }

        #[tokio::test]
        async fn different_device_locks_are_independent() {
            use std::sync::Arc;
            use std::sync::atomic::{AtomicU32, Ordering};

            let session_locks: crate::cache::Cache<String, Arc<async_lock::Mutex<()>>> =
                crate::cache::Cache::builder().max_capacity(100).build();

            let max_concurrent = Arc::new(AtomicU32::new(0));
            let counter = Arc::new(AtomicU32::new(0));
            let barrier = Arc::new(tokio::sync::Barrier::new(2));

            let keys = ["100000012345678@lid.0", "100000012345678:5@lid.0"];

            let mut handles = Vec::new();
            for key in keys {
                let locks = session_locks.clone();
                let key = key.to_string();
                let c = counter.clone();
                let m = max_concurrent.clone();
                let b = barrier.clone();

                handles.push(tokio::spawn(async move {
                    let mutex: Arc<async_lock::Mutex<()>> = locks
                        .get_with_by_ref(&key, async { Arc::new(async_lock::Mutex::new(())) })
                        .await;
                    // lock_arc(): same reason as above
                    let _guard = mutex.lock_arc().await;

                    let active = c.fetch_add(1, Ordering::SeqCst) + 1;
                    m.fetch_max(active, Ordering::SeqCst);
                    b.wait().await;
                    c.fetch_sub(1, Ordering::SeqCst);
                }));
            }

            for handle in handles {
                handle.await.unwrap();
            }

            assert_eq!(max_concurrent.load(Ordering::SeqCst), 2);
        }

        /// Regression: 1:1 DM recipient must use bare Signal address matching
        /// the receive path. Starts from device-specific JID and verifies
        /// to_non_ad() normalization produces the correct bare key.
        #[tokio::test]
        async fn dm_recipient_uses_bare_address() {
            let client = crate::test_utils::create_test_client().await;

            // Start from device-specific JID, exercise the production path
            let recipient_device33 = Jid::from_str("100000012345678:33@lid").unwrap();
            let own_device_5 = Jid::from_str("999999999999:5@s.whatsapp.net").unwrap();

            // Same normalization as send_message_impl
            let recipient_bare = client
                .resolve_encryption_jid(&recipient_device33)
                .await
                .to_non_ad();

            let all_dm_jids = vec![recipient_bare.clone(), own_device_5.clone()];
            let lock_jids = client.build_session_lock_keys(&all_dm_jids).await;

            // Recipient lock key must be BARE (device 0), matching decrypt path
            assert_eq!(
                recipient_bare.to_protocol_address_string(),
                "100000012345678@lid.0"
            );
            assert!(lock_jids.contains(&recipient_bare));

            // Own device lock key must be device-specific
            assert!(lock_jids.contains(&own_device_5));

            // Device-specific recipient key must NOT be present
            assert!(
                !lock_jids.contains(&recipient_device33),
                "recipient must NOT use device-specific address"
            );
        }

        /// Verify bare normalization deduplicates multiple recipient devices.
        #[test]
        fn bare_normalization_deduplicates_recipient_devices() {
            let devices: Vec<Jid> = [
                "100000012345678@lid",
                "100000012345678:5@lid",
                "100000012345678:33@lid",
            ]
            .iter()
            .map(|s| Jid::from_str(s).unwrap())
            .collect();

            // All collapse to the same bare JID
            let bare: Vec<Jid> = devices.iter().map(|j| j.to_non_ad()).collect();
            assert!(bare.windows(2).all(|w| w[0] == w[1]));
            assert_eq!(
                bare[0].to_protocol_address_string(),
                "100000012345678@lid.0"
            );
        }

        /// Every key handed in ends up locked, and every one is released when
        /// the guards are dropped. The device counts are the three the DM path
        /// actually produces: none (a fan-out that resolved to nothing), one
        /// (a steady 1:1) and several (companion devices in play).
        #[tokio::test]
        async fn taking_guards_locks_every_key_and_releasing_them_frees_every_key() {
            let client = crate::test_utils::create_test_client_with_name("guards_cover").await;

            for count in [0usize, 1, 3] {
                let devices: Vec<Jid> = (0..count)
                    .map(|i| Jid::from_str(&format!("10000001234567{i}:5@lid")).unwrap())
                    .collect();
                let keys = client.build_session_lock_keys(&devices).await;
                assert_eq!(keys.len(), count, "one key per device at count {count}");
                let mutexes = client.session_mutexes_for(&keys).await;

                let guards = client.session_guards_for(&keys).await;
                assert_eq!(guards.len(), count, "one guard per key at count {count}");
                for (i, mutex) in mutexes.iter().enumerate() {
                    assert!(
                        mutex.try_lock().is_none(),
                        "key {i} of {count} must be held while the guards live"
                    );
                }

                drop(guards);
                for (i, mutex) in mutexes.iter().enumerate() {
                    assert!(
                        mutex.try_lock().is_some(),
                        "key {i} of {count} must be free once the guards are dropped"
                    );
                }
            }
        }

        /// The keys are locked in the order given, which is the sorted order
        /// `build_session_lock_keys` produces. That single global order is the
        /// only thing keeping two sends that overlap on a device from
        /// deadlocking, so acquiring out of order must be observable.
        ///
        /// Blocking the SECOND key and then waiting for the FIRST to become
        /// contended is what pins the order down: a taker that went second-first
        /// would park on the blocked key and never touch the first one.
        #[tokio::test]
        async fn keys_are_locked_in_the_order_they_are_given() {
            let client = crate::test_utils::create_test_client_with_name("guards_order").await;

            let devices: Vec<Jid> = ["100000012345670:5@lid", "100000012345671:5@lid"]
                .iter()
                .map(|s| Jid::from_str(s).unwrap())
                .collect();
            let keys = client.build_session_lock_keys(&devices).await;
            assert_eq!(keys.len(), 2);
            let mutexes = client.session_mutexes_for(&keys).await;

            let blocker = mutexes[1].lock_arc().await;

            let mut taker = tokio::spawn({
                let client = client.clone();
                let keys = keys.clone();
                async move { client.session_guards_for(&keys).await.len() }
            });

            // Bounded work, not a deadline: yield until the first key is taken.
            let mut polls = 0;
            while mutexes[0].try_lock().is_some() {
                polls += 1;
                assert!(
                    polls < 10_000,
                    "the first key was never taken, so acquisition did not start there"
                );
                tokio::task::yield_now().await;
            }
            assert!(
                futures::poll!(&mut taker).is_pending(),
                "the taker must still be parked on the second key"
            );

            drop(blocker);
            assert_eq!(taker.await.expect("taker finishes"), 2);
        }
    }

    // ---- outbound messageSecret capture ---------------------------------

    use crate::store::commands::DeviceCommand;
    use std::sync::Arc;

    async fn seed_pn(client: &Arc<Client>, pn: &str) {
        client
            .persistence_manager
            .process_command(DeviceCommand::SetId(Some(pn.parse().expect("pn"))))
            .await;
    }

    async fn seed_pn_and_lid(client: &Arc<Client>, pn: &str, lid: &str) {
        client
            .persistence_manager
            .process_command(DeviceCommand::SetId(Some(pn.parse().expect("pn"))))
            .await;
        client
            .persistence_manager
            .process_command(DeviceCommand::SetLid(Some(lid.parse().expect("lid"))))
            .await;
    }

    fn peer_test_account_proto() -> wa::ADVSignedDeviceIdentity {
        wa::ADVSignedDeviceIdentity {
            details: Some(vec![0u8; 32]),
            account_signature_key: Some(vec![0u8; 32]),
            account_signature: Some(vec![0u8; 64]),
            device_signature: Some(vec![0u8; 64]),
        }
    }

    async fn seed_peer_send_state(client: &Arc<Client>, peer: &Jid) {
        use wacore::libsignal::protocol::{
            IdentityKeyPair, KeyPair, PreKeyBundle, SignalProtocolError, UsePQRatchet,
            process_prekey_bundle,
        };

        client
            .persistence_manager
            .process_command(DeviceCommand::SetAccount(Some(peer_test_account_proto())))
            .await;

        let bundle =
            tokio::task::spawn_blocking(|| -> Result<PreKeyBundle, SignalProtocolError> {
                let mut rng = rand::make_rng::<rand::rngs::StdRng>();
                let receiver = IdentityKeyPair::generate(&mut rng);
                let spk = KeyPair::generate(&mut rng);
                let opk = KeyPair::generate(&mut rng);
                let sig = receiver
                    .private_key()
                    .calculate_signature(&spk.public_key.serialize(), &mut rng)?;

                PreKeyBundle::new(
                    1,
                    1u32.into(),
                    Some((1u32.into(), opk.public_key)),
                    1u32.into(),
                    spk.public_key,
                    sig.to_vec(),
                    *receiver.identity_key(),
                )
            })
            .await
            .expect("prekey bundle task")
            .expect("prekey bundle");

        let mut adapter = client.signal_adapter().await;
        let mut rng = rand::make_rng::<rand::rngs::StdRng>();
        process_prekey_bundle(
            &peer.to_protocol_address(),
            &mut adapter.session_store,
            &mut adapter.identity_store,
            &bundle,
            &mut rng,
            UsePQRatchet::No,
        )
        .await
        .expect("peer session");
    }

    fn pdo_request_message(request_type: wa::message::PeerDataOperationRequestType) -> wa::Message {
        wa::Message {
            protocol_message: buffa::MessageField::some(wa::message::ProtocolMessage {
                r#type: Some(wa::message::protocol_message::Type::PeerDataOperationRequestMessage),
                peer_data_operation_request_message: buffa::MessageField::some(
                    wa::message::PeerDataOperationRequestMessage {
                        peer_data_operation_request_type: Some(request_type),
                        ..Default::default()
                    },
                ),
                ..Default::default()
            }),
            ..Default::default()
        }
    }

    #[tokio::test]
    async fn peer_pdo_send_path_stamps_history_sync_options() {
        let client = crate::test_utils::create_test_client_with_name("peer_pdo_attrs").await;
        let peer: Jid = "100000000000001@s.whatsapp.net".parse().unwrap();
        seed_peer_send_state(&client, &peer).await;

        let request_id = "PDO_PEER_ATTRS_1";
        let waiter = client
            .wait_for_sent_node(crate::client::NodeFilter::tag("message").attr("id", request_id));
        let msg =
            pdo_request_message(wa::message::PeerDataOperationRequestType::HistorySyncOnDemand);

        let result = client
            .send_message_impl(
                peer,
                &msg,
                SendPipelineOptions {
                    request_id: Some(request_id),
                    peer: true,
                    ..Default::default()
                },
            )
            .await;
        assert!(
            result.is_err(),
            "test client has no socket; send should fail after stanza capture"
        );

        let node = tokio::time::timeout(std::time::Duration::from_secs(1), waiter)
            .await
            .expect("sent node should be captured")
            .expect("sent node waiter should resolve");
        assert_eq!(
            node.attrs().optional_string("category").unwrap().as_ref(),
            "peer"
        );
        assert_eq!(
            node.attrs()
                .optional_string("push_priority")
                .unwrap()
                .as_ref(),
            "high_force"
        );
        assert_eq!(
            node.attrs()
                .optional_string("privacy_sensitive")
                .unwrap()
                .as_ref(),
            "1"
        );
    }

    #[tokio::test]
    async fn stanza_type_override_sets_wire_type_attr() {
        let client = crate::test_utils::create_test_client_with_name("stanza_type_override").await;
        let peer: Jid = "100000000000003@s.whatsapp.net".parse().unwrap();
        seed_peer_send_state(&client, &peer).await;

        let request_id = "STANZA_TYPE_OVERRIDE_1";
        let waiter = client
            .wait_for_sent_node(crate::client::NodeFilter::tag("message").attr("id", request_id));
        let msg =
            pdo_request_message(wa::message::PeerDataOperationRequestType::HistorySyncOnDemand);

        // Poll is never the type for this message; it can only come from the override.
        let result = client
            .send_message_impl(
                peer,
                &msg,
                SendPipelineOptions {
                    request_id: Some(request_id),
                    peer: true,
                    stanza_type: Some(StanzaType::Poll),
                    ..Default::default()
                },
            )
            .await;
        assert!(
            result.is_err(),
            "test client has no socket; send should fail after stanza capture"
        );

        let node = tokio::time::timeout(std::time::Duration::from_secs(1), waiter)
            .await
            .expect("sent node should be captured")
            .expect("sent node waiter should resolve");
        assert_eq!(
            node.attrs().optional_string("type").unwrap().as_ref(),
            StanzaType::Poll.as_wire()
        );
    }

    /// Shared setup for the DM wire-namespace regression tests: own PN/LID +
    /// account, the peer's LID mapping, device-registry entries for both peer
    /// namespaces and self, offline-sync completion, and a seeded Signal
    /// session for the peer's LID device so the offline fanout can encrypt
    /// without a socket. Returns `(peer_pn, peer_lid)`.
    async fn seed_dm_wire_namespace_state(client: &Arc<Client>) -> (Jid, Jid) {
        use wacore::libsignal::protocol::{
            IdentityKeyPair, KeyPair, PreKeyBundle, SignalProtocolError, UsePQRatchet,
            process_prekey_bundle,
        };

        // A LID-addressed DM requires the device's own PN and LID to be known.
        let own_pn: Jid = "111111111111@s.whatsapp.net".parse().unwrap();
        let own_lid: Jid = "222222222222@lid".parse().unwrap();
        client
            .persistence_manager
            .process_command(DeviceCommand::SetId(Some(own_pn.clone())))
            .await;
        client
            .persistence_manager
            .process_command(DeviceCommand::SetLid(Some(own_lid)))
            .await;
        client
            .persistence_manager
            .process_command(DeviceCommand::SetAccount(Some(peer_test_account_proto())))
            .await;

        // The peer is LID-mapped: the wire namespace is then decided solely by
        // the account's migration state.
        let peer_pn: Jid = "100000000000777@s.whatsapp.net".parse().unwrap();
        let peer_lid: Jid = "555000000000777@lid".parse().unwrap();
        client
            .add_lid_pn_mapping(
                peer_lid.user.as_str(),
                peer_pn.user.as_str(),
                crate::lid_pn_cache::LearningSource::Usync,
            )
            .await
            .expect("seed lid mapping");

        // Pre-seed the device registry for the peer (both namespaces) and self
        // so the offline send resolves the fanout from cache instead of
        // blocking on a network device-list fetch (which would time out with
        // no socket).
        for user in [
            peer_lid.user.to_string(),
            peer_pn.user.to_string(),
            own_pn.user.to_string(),
        ] {
            client
                .update_device_list(wacore::store::traits::DeviceListRecord {
                    user,
                    devices: vec![wacore::store::traits::DeviceInfo::new(0, None)],
                    timestamp: wacore::time::now_secs(),
                    phash: None,
                    raw_id: None,
                })
                .await
                .expect("seed device registry");
        }

        // The test client never connects, so the send's `ensure_e2e_sessions`
        // would otherwise block on `wait_for_offline_delivery_end` until
        // timeout. Enter live state synchronously (the real finisher now runs
        // as a spawned task).
        client.enter_live_mode_for_tests();

        // Seed a Signal session for the peer's LID device so the offline fanout
        // can encrypt without fetching prekeys over the (absent) socket. The
        // session lives under the LID address in both tests: Signal addressing
        // is LID-first regardless of the wire namespace (WAWebSignalAddress).
        let lid_addr = peer_lid.to_non_ad();
        let bundle =
            tokio::task::spawn_blocking(|| -> Result<PreKeyBundle, SignalProtocolError> {
                let mut rng = rand::make_rng::<rand::rngs::StdRng>();
                let receiver = IdentityKeyPair::generate(&mut rng);
                let spk = KeyPair::generate(&mut rng);
                let opk = KeyPair::generate(&mut rng);
                let sig = receiver
                    .private_key()
                    .calculate_signature(&spk.public_key.serialize(), &mut rng)?;
                PreKeyBundle::new(
                    1,
                    1u32.into(),
                    Some((1u32.into(), opk.public_key)),
                    1u32.into(),
                    spk.public_key,
                    sig.to_vec(),
                    *receiver.identity_key(),
                )
            })
            .await
            .expect("prekey bundle task")
            .expect("prekey bundle");
        {
            let mut adapter = client.signal_adapter().await;
            let mut rng = rand::make_rng::<rand::rngs::StdRng>();
            process_prekey_bundle(
                &lid_addr.to_protocol_address(),
                &mut adapter.session_store,
                &mut adapter.identity_store,
                &bundle,
                &mut rng,
                UsePQRatchet::No,
            )
            .await
            .expect("peer lid session");
        }

        (peer_pn, peer_lid)
    }

    /// Regression for #730: on a 1:1-LID-migrated account, a DM to a
    /// LID-mapped peer must address the outer `<message to>` by LID, matching
    /// the LID `<participants>`. Pre-fix the outer `to` kept the caller's PN,
    /// so a PN-to over LID participants was rejected wholesale by the server
    /// with `ack error="400"` and never delivered (while the send still
    /// returned Ok). WAWebSendMsgCreateFanoutStanza builds the whole stanza
    /// from one CHAT_JID (the LID after migration).
    #[tokio::test]
    async fn dm_to_lid_mapped_peer_addresses_outer_to_by_lid() {
        let client = crate::test_utils::create_test_client_with_name("lid_dm_to").await;
        let (peer_pn, peer_lid) = seed_dm_wire_namespace_state(&client).await;

        // LID wire addressing is gated on the account being 1:1-LID-migrated.
        client
            .persistence_manager
            .process_command(DeviceCommand::SetLidMigrated(true))
            .await;

        let request_id = "LID_DM_TO_1";
        let waiter = client
            .wait_for_sent_node(crate::client::NodeFilter::tag("message").attr("id", request_id));
        let msg = wa::Message {
            conversation: Some("hi".into()),
            ..Default::default()
        };
        // Caller passes the PN form; the resolved namespace must win on the wire.
        let result = client
            .send_message_impl(
                peer_pn,
                &msg,
                SendPipelineOptions {
                    request_id: Some(request_id),
                    ..Default::default()
                },
            )
            .await;
        assert!(
            result.is_err(),
            "test client has no socket; send captures the stanza then errors"
        );

        let node = tokio::time::timeout(std::time::Duration::from_secs(1), waiter)
            .await
            .expect("sent node should be captured")
            .expect("sent node waiter should resolve");

        // The fix: outer `<message to>` is the LID, not the caller's PN.
        let to_str = node
            .attrs()
            .optional_string("to")
            .expect("message has a to")
            .into_owned();
        let to_jid: Jid = to_str.parse().expect("to parses");
        assert!(
            to_jid.is_lid(),
            "outer <message to> must be LID to match the LID participants, got {to_str}"
        );
        assert_eq!(
            to_jid.user.as_str(),
            peer_lid.user.as_str(),
            "outer to user must be the peer LID"
        );

        // Uniformity guard: every <participants>/<to> is LID too (no mix).
        let participants = node
            .get_optional_child("participants")
            .expect("stanza has participants");
        let entries = participants.children().expect("participants has children");
        assert!(
            !entries.is_empty(),
            "fanout must target at least the recipient"
        );
        for entry in entries {
            let pj: Jid = entry
                .attrs()
                .optional_string("jid")
                .expect("participant jid")
                .parse()
                .expect("participant jid parses");
            assert!(
                pj.is_lid(),
                "participant {pj} must be LID (uniform namespace)"
            );
        }
    }

    /// Regression for #941: an account that is NOT 1:1-LID-migrated must keep
    /// DM wire addressing on PN even with a cached LID mapping — the server
    /// 400-nacks LID-addressed DMs from unmigrated accounts. WA Web only
    /// addresses 1:1 chats by LID once `Lid1X1MigrationUtils.isLidMigrated()`.
    #[tokio::test]
    async fn dm_from_unmigrated_account_addresses_outer_to_by_pn() {
        let client = crate::test_utils::create_test_client_with_name("pn_dm_to").await;
        let (peer_pn, _peer_lid) = seed_dm_wire_namespace_state(&client).await;

        let request_id = "PN_DM_TO_1";
        let waiter = client
            .wait_for_sent_node(crate::client::NodeFilter::tag("message").attr("id", request_id));
        let msg = wa::Message {
            conversation: Some("hi".into()),
            ..Default::default()
        };
        let result = client
            .send_message_impl(
                peer_pn.clone(),
                &msg,
                SendPipelineOptions {
                    request_id: Some(request_id),
                    ..Default::default()
                },
            )
            .await;
        assert!(
            result.is_err(),
            "test client has no socket; send captures the stanza then errors"
        );

        let node = tokio::time::timeout(std::time::Duration::from_secs(1), waiter)
            .await
            .expect("sent node should be captured")
            .expect("sent node waiter should resolve");

        let to_str = node
            .attrs()
            .optional_string("to")
            .expect("message has a to")
            .into_owned();
        let to_jid: Jid = to_str.parse().expect("to parses");
        assert!(
            to_jid.is_pn(),
            "outer <message to> must stay PN on an unmigrated account, got {to_str}"
        );
        assert_eq!(
            to_jid.user.as_str(),
            peer_pn.user.as_str(),
            "outer to user must be the peer PN"
        );

        // Uniformity guard: every <participants>/<to> is PN too (no mix).
        let participants = node
            .get_optional_child("participants")
            .expect("stanza has participants");
        let entries = participants.children().expect("participants has children");
        assert!(
            !entries.is_empty(),
            "fanout must target at least the recipient"
        );
        for entry in entries {
            let pj: Jid = entry
                .attrs()
                .optional_string("jid")
                .expect("participant jid")
                .parse()
                .expect("participant jid parses");
            assert!(
                pj.is_pn(),
                "participant {pj} must be PN (uniform namespace)"
            );
        }
    }

    /// Newsletter JIDs must be rejected at the E2E send path root (covers the
    /// mis-routed pin/edit/revoke producers that call send_message_impl directly).
    #[tokio::test]
    async fn newsletter_jid_rejected_on_e2e_send_path() {
        let client = crate::test_utils::create_test_client_with_name("newsletter_e2e_guard").await;
        let channel: Jid = "120363000000000001@newsletter".parse().unwrap();
        let msg = wa::Message {
            conversation: Some("x".to_string()),
            ..Default::default()
        };
        let err = client
            .send_message_impl(channel, &msg, SendPipelineOptions::default())
            .await
            .expect_err("newsletter JID must be rejected on the E2E send path");
        assert!(
            err.to_string().to_lowercase().contains("newsletter"),
            "error should name the newsletter mis-route, got: {err}"
        );
    }

    /// The pin producer routes through send_message_impl, so a newsletter pin is
    /// rejected rather than building an encrypted fanout against a channel.
    #[tokio::test]
    async fn pin_message_rejects_newsletter() {
        let client = crate::test_utils::create_test_client_with_name("newsletter_pin_guard").await;
        let channel: Jid = "120363000000000002@newsletter".parse().unwrap();
        let key = wa::MessageKey {
            remote_jid: Some(channel.to_string()),
            from_me: Some(true),
            id: Some("MID".to_string()),
            participant: None,
        };
        let err = client
            .pin_message(channel, key, PinDuration::Days7)
            .await
            .expect_err("pinning a newsletter message must be rejected");
        assert!(
            err.to_string().to_lowercase().contains("newsletter"),
            "error should name the newsletter mis-route, got: {err}"
        );
    }

    /// Newsletter edit: plaintext `<message edit="3">` keyed by server_id, with the
    /// new content in `<plaintext>`. Keyed by the message id STRING (not server_id),
    /// and a text edit carries no mediatype.
    #[test]
    fn build_newsletter_edit_node_emits_plaintext_edit() {
        use buffa::Message as _;
        let to: Jid = "120363000000000001@newsletter".parse().unwrap();
        let content = wa::Message {
            conversation: Some("edited text".to_string()),
            ..Default::default()
        };
        let node =
            build_newsletter_edit_node(&to, "3EB0EDITTARGET", NewsletterEdit::Edit(&content));

        let mut a = node.attrs();
        assert_eq!(a.optional_string("id").unwrap().as_ref(), "3EB0EDITTARGET");
        assert_eq!(a.optional_string("type").unwrap().as_ref(), "text");
        assert_eq!(a.optional_string("edit").unwrap().as_ref(), "3");

        let pt = node
            .get_optional_child("plaintext")
            .expect("plaintext child");
        assert!(
            pt.attrs().optional_string("mediatype").is_none(),
            "a text edit must not carry a mediatype attr"
        );
        let bytes = match pt.content.as_ref() {
            Some(wacore_binary::NodeContent::Bytes(b)) => b.clone(),
            other => panic!("expected plaintext bytes, got {other:?}"),
        };
        let decoded = wa::Message::decode_from_slice(bytes.as_slice()).expect("decode plaintext");
        assert_eq!(decoded.conversation.as_deref(), Some("edited text"));
    }

    /// Media newsletter edit: type="media" + `<plaintext mediatype="image">`.
    #[test]
    fn build_newsletter_edit_node_media_edit() {
        let to: Jid = "120363000000000001@newsletter".parse().unwrap();
        let content = wa::Message {
            image_message: buffa::MessageField::some(wa::message::ImageMessage {
                caption: Some("new caption".to_string()),
                ..Default::default()
            }),
            ..Default::default()
        };
        let node = build_newsletter_edit_node(&to, "3EB0MEDIA", NewsletterEdit::Edit(&content));

        let mut a = node.attrs();
        assert_eq!(a.optional_string("id").unwrap().as_ref(), "3EB0MEDIA");
        assert_eq!(a.optional_string("type").unwrap().as_ref(), "media");
        assert_eq!(a.optional_string("edit").unwrap().as_ref(), "3");
        let pt = node
            .get_optional_child("plaintext")
            .expect("plaintext child");
        assert_eq!(
            pt.attrs().optional_string("mediatype").unwrap().as_ref(),
            "image"
        );
    }

    /// Newsletter revoke: plaintext `<message type="text" edit="8">` keyed by the
    /// message id STRING, with an empty `<plaintext>`.
    #[test]
    fn build_newsletter_edit_node_revoke_is_empty_plaintext() {
        let to: Jid = "120363000000000002@newsletter".parse().unwrap();
        let node = build_newsletter_edit_node(&to, "3EB0REVOKETARGET", NewsletterEdit::Revoke);

        let mut a = node.attrs();
        assert_eq!(
            a.optional_string("id").unwrap().as_ref(),
            "3EB0REVOKETARGET"
        );
        assert_eq!(a.optional_string("type").unwrap().as_ref(), "text");
        assert_eq!(a.optional_string("edit").unwrap().as_ref(), "8");

        let pt = node
            .get_optional_child("plaintext")
            .expect("plaintext child");
        let empty = match pt.content.as_ref() {
            None => true,
            Some(wacore_binary::NodeContent::Bytes(b)) => b.is_empty(),
            _ => false,
        };
        assert!(empty, "revoke must carry an empty plaintext");
    }

    /// The public newsletter().edit_message wrapper emits the plaintext edit stanza
    /// keyed by the message id it was given.
    #[tokio::test]
    async fn newsletter_edit_message_wrapper_sends_plaintext_edit() {
        let client = crate::test_utils::create_test_client_with_name("nl_edit_wrap").await;
        let channel: Jid = "120363000000000001@newsletter".parse().unwrap();
        let waiter =
            client.wait_for_sent_node(crate::client::NodeFilter::tag("message").attr("edit", "3"));
        let content = wa::Message {
            conversation: Some("edited".to_string()),
            ..Default::default()
        };
        // No socket on the test client: send_node captures the node, then errors.
        let _ = client
            .newsletter()
            .edit_message(&channel, "TARGETMID", content)
            .await;

        let node = tokio::time::timeout(std::time::Duration::from_secs(1), waiter)
            .await
            .expect("sent node captured")
            .expect("waiter resolves");
        let mut a = node.attrs();
        assert_eq!(a.optional_string("id").unwrap().as_ref(), "TARGETMID");
        assert_eq!(a.optional_string("edit").unwrap().as_ref(), "3");
    }

    /// The newsletter edit/revoke methods reject non-newsletter JIDs, so a misuse
    /// cannot send plaintext content to a DM/group (it would not be E2E-encrypted).
    #[tokio::test]
    async fn newsletter_edit_revoke_reject_non_newsletter_jid() {
        let client = crate::test_utils::create_test_client_with_name("nl_reject_nonchannel").await;
        let dm: Jid = "5511999999999@s.whatsapp.net".parse().unwrap();
        let group: Jid = "120363000000000009@g.us".parse().unwrap();

        let e1 = client
            .newsletter()
            .edit_message(
                &dm,
                "MID",
                wa::Message {
                    conversation: Some("x".to_string()),
                    ..Default::default()
                },
            )
            .await
            .expect_err("edit_message must reject a DM JID");
        assert!(e1.to_string().to_lowercase().contains("newsletter"));

        let e2 = client
            .newsletter()
            .revoke_message(&group, "MID")
            .await
            .expect_err("revoke_message must reject a group JID");
        assert!(e2.to_string().to_lowercase().contains("newsletter"));
    }

    /// An empty message_id (NewsletterMessage.message_id may be empty if the server
    /// omitted the id) is rejected rather than sending a target-less id="" stanza.
    #[tokio::test]
    async fn newsletter_edit_revoke_reject_empty_message_id() {
        let client = crate::test_utils::create_test_client_with_name("nl_reject_empty_id").await;
        let channel: Jid = "120363000000000001@newsletter".parse().unwrap();

        let e1 = client
            .newsletter()
            .edit_message(
                &channel,
                "",
                wa::Message {
                    conversation: Some("x".to_string()),
                    ..Default::default()
                },
            )
            .await
            .expect_err("edit_message must reject an empty message_id");
        assert!(e1.to_string().to_lowercase().contains("message_id"));

        let e2 = client
            .newsletter()
            .revoke_message(&channel, "")
            .await
            .expect_err("revoke_message must reject an empty message_id");
        assert!(e2.to_string().to_lowercase().contains("message_id"));
    }

    #[tokio::test]
    async fn persist_outbound_msg_secret_writes_under_chat_sender_id() {
        let client = crate::test_utils::create_test_client_with_name("secret_chat_id").await;
        seed_pn(&client, "5511000000001:0@s.whatsapp.net").await;
        let chat: Jid = "5511777776666@s.whatsapp.net".parse().unwrap();
        let sender: Jid = "5511000000001:0@s.whatsapp.net".parse().unwrap();
        let secret = [0x55u8; 32];
        client
            .persist_outbound_msg_secret(
                &chat,
                &sender,
                "MID_1",
                &secret,
                wacore::msg_secret::RetentionClass::Text,
                SendInstant::now(),
            )
            .await;
        client.msg_secret_buffer.wait_flushed().await;
        let got = client
            .persistence_manager
            .backend()
            .get_msg_secret(
                "5511777776666@s.whatsapp.net",
                "5511000000001@s.whatsapp.net",
                "MID_1",
            )
            .await
            .expect("get");
        assert_eq!(got.as_deref(), Some(&secret[..]));
    }

    #[tokio::test]
    async fn persist_outbound_msg_secret_strips_devices_in_key() {
        let client = crate::test_utils::create_test_client_with_name("secret_strip").await;
        let chat_with_dev: Jid = "5511777776666:7@s.whatsapp.net".parse().unwrap();
        let sender_with_dev: Jid = "5511000000001:3@s.whatsapp.net".parse().unwrap();
        client
            .persist_outbound_msg_secret(
                &chat_with_dev,
                &sender_with_dev,
                "MID_4",
                &[2u8; 32],
                wacore::msg_secret::RetentionClass::Text,
                SendInstant::now(),
            )
            .await;
        client.msg_secret_buffer.wait_flushed().await;
        let got = client
            .persistence_manager
            .backend()
            .get_msg_secret(
                "5511777776666@s.whatsapp.net",
                "5511000000001@s.whatsapp.net",
                "MID_4",
            )
            .await
            .unwrap();
        assert_eq!(
            got.as_deref(),
            Some(&[2u8; 32][..]),
            "chat and sender must be stored non-AD"
        );
    }

    #[tokio::test]
    async fn dm_sender_identity_picks_lid_for_bot_else_pn() {
        let client = crate::test_utils::create_test_client_with_name("dm_id_pick").await;
        seed_pn_and_lid(
            &client,
            "5511000000001:0@s.whatsapp.net",
            "999888777666555:0@lid",
        )
        .await;
        let bot_chat: Jid = "867051314767696@bot".parse().unwrap();
        let pn_chat: Jid = "5511777776666@s.whatsapp.net".parse().unwrap();
        let lid_chat: Jid = "111222333444555@lid".parse().unwrap();
        assert_eq!(
            client
                .dm_sender_identity_for(&bot_chat)
                .await
                .map(|j| j.to_non_ad_string()),
            Some("999888777666555@lid".to_string()),
            "bot chats must resolve to our LID"
        );
        assert_eq!(
            client
                .dm_sender_identity_for(&pn_chat)
                .await
                .map(|j| j.to_non_ad_string()),
            Some("5511000000001@s.whatsapp.net".to_string()),
            "PN chats must resolve to our PN"
        );
        // LID-DM is presently routed under PN; flagged as a follow-up only
        // because production hasn't surfaced it. Documented behaviour.
        assert_eq!(
            client
                .dm_sender_identity_for(&lid_chat)
                .await
                .map(|j| j.to_non_ad_string()),
            Some("5511000000001@s.whatsapp.net".to_string()),
        );
    }

    /// Regression for Codex P2 (LID-mode group bot replies): the persisted
    /// sender must match whatever `prepare_group_stanza` picked for the
    /// group's addressing_mode, surfaced via `PreparedGroupStanza.sender_identity`.
    #[tokio::test]
    async fn persist_uses_group_sender_identity_for_lid_mode_groups() {
        let client = crate::test_utils::create_test_client_with_name("secret_lid_group").await;
        seed_pn_and_lid(
            &client,
            "5511000000001:0@s.whatsapp.net",
            "999888777666555:0@lid",
        )
        .await;
        // Simulate a LID-mode group: addressing identity is our LID, not PN.
        let group_chat: Jid = "120363021033254949@g.us".parse().unwrap();
        let lid_sender: Jid = "999888777666555:0@lid".parse().unwrap();
        let secret = [0x4Du8; 32];
        client
            .persist_outbound_msg_secret(
                &group_chat,
                &lid_sender,
                "GROUP_MID",
                &secret,
                wacore::msg_secret::RetentionClass::Text,
                SendInstant::now(),
            )
            .await;
        client.msg_secret_buffer.wait_flushed().await;
        let got = client
            .persistence_manager
            .backend()
            .get_msg_secret(
                "120363021033254949@g.us",
                "999888777666555@lid",
                "GROUP_MID",
            )
            .await
            .unwrap();
        assert_eq!(
            got.as_deref(),
            Some(&secret[..]),
            "LID-mode group secrets must key under our LID, not PN"
        );
        let under_pn = client
            .persistence_manager
            .backend()
            .get_msg_secret(
                "120363021033254949@g.us",
                "5511000000001@s.whatsapp.net",
                "GROUP_MID",
            )
            .await
            .unwrap();
        assert!(
            under_pn.is_none(),
            "LID-mode group must NOT key under our PN"
        );
    }

    /// Regression: `wacore::send::prepare_dm_stanza` mints the
    /// `message_secret` on a CLONE of the caller's message. Verify the secret
    /// is surfaced via `PreparedDmStanza.message_secret` so the post-send hook
    /// can persist it -- without this an original-message-based check would
    /// miss every ordinary outbound bot prompt.
    #[test]
    fn prepared_dm_stanza_exposes_generated_message_secret() {
        use wacore::reporting_token::generate_reporting_token;

        let msg = wa::Message {
            conversation: Some("hi bot".into()),
            ..Default::default()
        };
        let to: Jid = "867051314767696@bot".parse().unwrap();
        let result = generate_reporting_token(&msg, "MID_X", &to, &to, None);
        assert!(
            result.is_some(),
            "ordinary text messages must produce a reporting token + secret"
        );
        let result = result.unwrap();
        assert_eq!(result.message_secret.len(), 32);
        // PreparedDmStanza/PreparedGroupStanza now carry this exact array
        // through to send_message_impl which calls persist_outbound_msg_secret.
        let prepared = wacore::send::PreparedDmStanza {
            node: NodeBuilder::new("message").build(),
            phash: None,
            message_secret: Some(result.message_secret),
        };
        assert_eq!(prepared.message_secret.as_ref().unwrap().len(), 32);
    }

    /// A send names its message once and every downstream stage reads that same
    /// name: the wire stanza, the phash ack-waiter, the outbound messageSecret
    /// and the returned `SendResult`. A non-ASCII id is used on purpose — a
    /// truncating or byte-indexing copy anywhere in that chain would show up
    /// here and nowhere else.
    #[tokio::test]
    async fn one_id_names_the_stanza_the_waiter_the_secret_and_the_result() {
        let (client, _transport) = crate::test_utils::create_iq_test_client().await;
        let (peer_pn, _peer_lid) = seed_dm_wire_namespace_state(&client).await;

        let message_id = "ID_ünïcødé_✅_ONE";
        let result = client
            .send_message_with_options(
                peer_pn.clone(),
                wa::Message {
                    conversation: Some("hi".into()),
                    ..Default::default()
                },
                SendOptions::default().with_message_id(message_id),
            )
            .await
            .expect("connected test client should complete the send");

        assert_eq!(
            result.message_id, message_id,
            "result carries the caller id"
        );
        assert_eq!(result.to, peer_pn, "result carries the caller target");

        let waiters = client.response_waiters_guard();
        assert!(
            waiters.contains_key(message_id),
            "the phash ack-waiter must be keyed by the send's own id"
        );
        assert_eq!(waiters.len(), 1, "no second entry under another spelling");
        drop(waiters);

        let secret = client.msg_secret_buffer.lookup(
            &peer_pn.to_non_ad_string(),
            &client.pn().expect("own pn").to_non_ad_string(),
            message_id,
        );
        assert!(
            secret.is_some(),
            "the outbound messageSecret must be bound to the same id"
        );
    }

    /// The waiter is installed before the stanza reaches the socket (a fast ack
    /// can land while `send_node` is still returning), so a send that fails on
    /// the wire has to take it back out — under the id it registered. Removing
    /// under anything else leaks an entry that a later ack could resolve.
    #[tokio::test]
    async fn a_failed_send_takes_its_phash_waiter_back_out() {
        let client = crate::test_utils::create_test_client_with_name("phash_waiter_rollback").await;
        let (peer_pn, _peer_lid) = seed_dm_wire_namespace_state(&client).await;

        let message_id = "ID_ünïcødé_✅_ROLLBACK";
        let result = client
            .send_message_impl(
                peer_pn,
                &wa::Message {
                    conversation: Some("hi".into()),
                    ..Default::default()
                },
                SendPipelineOptions {
                    request_id: Some(message_id),
                    ..Default::default()
                },
            )
            .await;
        assert!(result.is_err(), "no socket: the send must fail on the wire");
        assert_eq!(
            client.response_waiters_guard().len(),
            0,
            "a failed send must leave no waiter behind, under any key"
        );
    }

    /// A borrowed id belongs to another message: registering a waiter under it
    /// would overwrite the original send's waiter, and binding a secret under it
    /// would overwrite the original's secret.
    #[tokio::test]
    async fn a_borrowed_id_registers_no_waiter_and_binds_no_secret() {
        let (client, _transport) = crate::test_utils::create_iq_test_client().await;
        let (peer_pn, _peer_lid) = seed_dm_wire_namespace_state(&client).await;

        let message_id = "ID_BORROWED_1";
        client
            .send_message_impl(
                peer_pn.clone(),
                &wa::Message {
                    conversation: Some("hi".into()),
                    ..Default::default()
                },
                SendPipelineOptions {
                    request_id: Some(message_id),
                    borrowed_message_id: true,
                    ..Default::default()
                },
            )
            .await
            .expect("connected test client should complete the send");

        assert_eq!(
            client.response_waiters_guard().len(),
            0,
            "a borrowed id must not claim the waiter slot"
        );
        assert!(
            client
                .msg_secret_buffer
                .lookup(
                    &peer_pn.to_non_ad_string(),
                    &client.pn().expect("own pn").to_non_ad_string(),
                    message_id,
                )
                .is_none(),
            "a borrowed id must not claim the secret slot"
        );
    }

    /// An empty id would name nothing: it must be refused at both entry points
    /// before any state is stamped with it.
    #[tokio::test]
    async fn an_empty_id_is_refused_at_both_entry_points() {
        let client = crate::test_utils::create_test_client_with_name("empty_send_id").await;
        let peer: Jid = "100000000000777@s.whatsapp.net".parse().unwrap();
        let msg = wa::Message {
            conversation: Some("hi".into()),
            ..Default::default()
        };

        let public = client
            .send_message_with_options(
                peer.clone(),
                msg.clone(),
                SendOptions::default().with_message_id(""),
            )
            .await;
        assert!(
            matches!(public, Err(SendError::InvalidRequest(_))),
            "public send must reject an empty id, got {public:?}"
        );

        let internal = client
            .send_message_impl(
                peer,
                &msg,
                SendPipelineOptions {
                    request_id: Some(""),
                    ..Default::default()
                },
            )
            .await;
        let internal = internal.expect_err("internal send must reject an empty id");
        assert!(
            internal
                .to_string()
                .contains("message ID must not be empty"),
            "unexpected error: {internal}"
        );
    }

    /// The plaintext newsletter branch returns before the E2E pipeline, so it
    /// builds its own result; it must still hand back the id it stamped and the
    /// channel it addressed.
    #[tokio::test]
    async fn the_newsletter_branch_returns_the_id_and_target_it_stamped() {
        let (client, _transport) = crate::test_utils::create_iq_test_client().await;
        let channel: Jid = "123456789@newsletter".parse().unwrap();

        let message_id = "ID_ünïcødé_✅_NEWS";
        let waiter = client
            .wait_for_sent_node(crate::client::NodeFilter::tag("message").attr("id", message_id));
        let result = client
            .send_message_with_options(
                channel.clone(),
                wa::Message {
                    conversation: Some("hi".into()),
                    ..Default::default()
                },
                SendOptions::default().with_message_id(message_id),
            )
            .await
            .expect("newsletter send is plaintext and needs no session");

        assert_eq!(result.message_id, message_id);
        assert_eq!(result.to, channel);

        let node = waiter.await.expect("the stanza should be captured");
        assert_eq!(
            node.attrs().optional_string("id").as_deref(),
            Some(message_id),
            "the wire id must be the same one the result reports"
        );
    }
}

#[cfg(test)]
mod jid_into_convention {
    use super::*;

    /// Compile-time guard for the `impl Into<Jid>` convention: every core
    /// method must accept BOTH an owned `Jid` (move, zero copy) and a `&Jid`
    /// (one clone via `From<&Jid>`). Never executed; compilation is the test.
    #[allow(dead_code)]
    async fn both_call_styles_compile(client: &Client, jid: Jid) {
        let msg = wa::Message::default();
        let _ = client.send_message(&jid, msg.clone()).await;
        let _ = client
            .send_message_with_options(&jid, msg.clone(), SendOptions::default())
            .await;
        let _ = client.forward_message(&jid, &msg).await;
        let _ = client
            .edit_message(&jid, "ID", wa::Message::default())
            .await;
        let _ = client.revoke_message(&jid, "ID", RevokeType::Sender).await;
        let _ = client
            .pin_message(&jid, wa::MessageKey::default(), PinDuration::default())
            .await;
        let _ = client.unpin_message(&jid, wa::MessageKey::default()).await;
        let _ = client
            .send_reaction(&jid, wa::MessageKey::default(), "x")
            .await;
        let _ = client
            .keep_message(&jid, wa::MessageKey::default(), true)
            .await;
        // Owned style: moves, no clone. Each method consumes its own copy so
        // the whole core surface is pinned, not just send_message.
        let _ = client.send_message(jid.clone(), msg.clone()).await;
        let _ = client
            .send_message_with_options(jid.clone(), msg.clone(), SendOptions::default())
            .await;
        let _ = client.forward_message(jid.clone(), &msg).await;
        let _ = client
            .edit_message(jid.clone(), "ID", wa::Message::default())
            .await;
        let _ = client
            .revoke_message(jid.clone(), "ID", RevokeType::Sender)
            .await;
        let _ = client
            .pin_message(
                jid.clone(),
                wa::MessageKey::default(),
                PinDuration::default(),
            )
            .await;
        let _ = client
            .unpin_message(jid.clone(), wa::MessageKey::default())
            .await;
        let _ = client
            .send_reaction(jid.clone(), wa::MessageKey::default(), "x")
            .await;
        let _ = client
            .keep_message(jid, wa::MessageKey::default(), true)
            .await;
    }
}

#[cfg(test)]
mod future_size_tests {
    /// The public send futures embed in every event-handler and spawned-task
    /// frame, so their size is a per-event heap cost. Keep them pointer-scale
    /// (measured 64-128 B; the bound leaves slack only for layout drift).
    #[tokio::test]
    async fn send_futures_stay_small() {
        let client = crate::test_utils::create_test_client().await;
        let jid: wacore_binary::jid::Jid = "5511999990000@s.whatsapp.net".parse().unwrap();
        let msg = waproto::whatsapp::Message::default();

        let f = client.send_message(jid.clone(), msg.clone());
        assert!(size_of_val(&f) <= 192, "send_message future grew");
        drop(f);
        let f = client.send_text(jid.clone(), "x");
        assert!(size_of_val(&f) <= 192, "send_text future grew");
        drop(f);
        let f = client.forward_message(jid.clone(), &msg);
        assert!(size_of_val(&f) <= 192, "forward_message future grew");
        drop(f);
        let f = client.send_message_with_options(jid, msg, Default::default());
        assert!(
            size_of_val(&f) <= 192,
            "send_message_with_options future grew"
        );
        drop(f);
    }
}

#[cfg(test)]
mod clock_budget_tests {
    use super::*;
    use crate::store::commands::DeviceCommand;
    use std::sync::Arc;
    use wacore::time::clock_reads;

    const OWN_PN: &str = "15551234001";
    const PEER_PN: &str = "5511900000001";
    const PEER_LID: &str = "100000000000079";

    /// Budget for one steady-state DM send, in clock reads. On wasm32 and
    /// embedded targets every read leaves the module, so this is a real cost of
    /// the send path and not just an instruction count.
    const SEND_WALL_READS: u64 = 1;
    const SEND_MONOTONIC_READS: u64 = 2;

    async fn seed_devices(client: &Arc<Client>, user: &str) {
        client
            .update_device_list(wacore::store::traits::DeviceListRecord {
                user: user.into(),
                devices: vec![wacore::store::traits::DeviceInfo::new(0, None)],
                timestamp: wacore::time::now_secs(),
                phash: None,
                raw_id: None,
            })
            .await
            .expect("seed device list");
    }

    /// Registry, LID mapping and Signal sessions already seeded, so a send
    /// queries nothing over the wire.
    async fn cold_send_client() -> (
        Arc<Client>,
        Arc<crate::transport::mock::CapturingMockTransport>,
        Jid,
    ) {
        let (client, transport) = crate::test_utils::create_iq_test_client().await;
        client
            .persistence_manager
            .process_command(DeviceCommand::SetId(Some(
                format!("{OWN_PN}@s.whatsapp.net").parse().expect("own pn"),
            )))
            .await;
        client
            .persistence_manager
            .process_command(DeviceCommand::SetLid(Some(
                "100000000000001@lid".parse().expect("own lid"),
            )))
            .await;

        let peer = Jid::pn(PEER_PN);
        seed_devices(&client, PEER_PN).await;
        seed_devices(&client, OWN_PN).await;
        seed_devices(&client, PEER_LID).await;
        client
            .add_lid_pn_mapping(
                PEER_LID,
                PEER_PN,
                crate::lid_pn_cache::LearningSource::Usync,
            )
            .await
            .expect("lid mapping");
        crate::test_utils::seed_peer_session(&client, &peer).await;
        crate::test_utils::seed_peer_session(
            &client,
            &format!("{PEER_LID}@lid").parse().expect("lid jid"),
        )
        .await;

        (client, transport, peer)
    }

    /// [`cold_send_client`] plus a first send, which drains the once-per-peer
    /// privacy-token issuance so the next send is steady state.
    async fn warm_send_client() -> (
        Arc<Client>,
        Arc<crate::transport::mock::CapturingMockTransport>,
        Jid,
    ) {
        let (client, transport, peer) = cold_send_client().await;
        client.send_text(peer.clone(), "warm").await.expect("warm");
        // The privacy token is issued off the send path, so wait for its frame
        // rather than let it land inside the measured window.
        crate::test_utils::poll_until("the privacy token to be issued", || {
            transport.sent_count() >= 2
        })
        .await;
        crate::test_utils::wait_for_outbound_tasks(&client).await;
        (client, transport, peer)
    }

    /// Paused time so the unanswered IQs this harness leaves behind do not cost
    /// their real timeouts.
    #[tokio::test(start_paused = true)]
    async fn dm_send_stays_within_its_clock_budget() {
        let (client, transport, peer) = warm_send_client().await;
        let frames_before = transport.sent_count();

        let base = clock_reads::snapshot();
        client.send_text(peer, "hello").await.expect("send");
        let reads = clock_reads::since(base);

        assert_eq!(
            transport.sent_count() - frames_before,
            1,
            "the budget only describes a send that writes exactly one frame"
        );
        assert!(
            reads.total() > 0,
            "a zero count means the flow did not run, not that it got free"
        );
        assert!(
            reads.wall <= SEND_WALL_READS,
            "wall-clock reads per DM send rose to {} (budget {SEND_WALL_READS})",
            reads.wall
        );
        assert!(
            reads.monotonic <= SEND_MONOTONIC_READS,
            "monotonic reads per DM send rose to {} (budget {SEND_MONOTONIC_READS})",
            reads.monotonic
        );

        client.disconnect().await;
    }

    /// One send, one instant: the message id, the biz node, the privacy-token
    /// decision and the outbound secret are stamped from a single read, so they
    /// cannot end up describing different seconds for the same message.
    #[tokio::test(start_paused = true)]
    async fn a_send_reads_the_clock_once() {
        let (client, _transport, peer) = warm_send_client().await;

        let base = clock_reads::snapshot();
        let sent = client.send_text(peer.clone(), "hello").await.expect("send");
        assert_eq!(
            clock_reads::since(base).wall,
            1,
            "every stamp on the send path must come from the same read"
        );

        // The outbound secret is one of the stamps, so its presence proves the
        // measured window really covered that write.
        client.msg_secret_buffer.wait_flushed().await;
        let stored = client
            .persistence_manager
            .backend()
            .get_msg_secret(
                &peer.to_non_ad_string(),
                &format!("{OWN_PN}@s.whatsapp.net"),
                &sent.message_id,
            )
            .await
            .expect("msg secret lookup");
        assert!(
            stored.is_some(),
            "the send persisted an outbound secret under the measured instant"
        );

        client.disconnect().await;
    }

    /// The wire timestamp is the one thing the budget must never buy: the
    /// privacy-token IQ a first send emits still carries the real second.
    #[tokio::test(start_paused = true)]
    async fn wire_timestamp_keeps_real_time() {
        let (client, transport, peer) = cold_send_client().await;

        let before = wacore::time::now_secs();
        client.send_text(peer, "hello").await.expect("send");
        crate::test_utils::poll_until("the privacy token to be issued", || {
            transport.sent_count() >= 2
        })
        .await;
        let after = wacore::time::now_secs();

        let mut seen = None;
        for index in 0..transport.sent_count() {
            let node = crate::test_utils::decode_sent_iq(&transport, index).await;
            let node = node.get();
            if node.attrs().optional_string("xmlns").as_deref() != Some("privacy") {
                continue;
            }
            let token = node
                .get_optional_child("tokens")
                .and_then(|t| t.get_optional_child("token"))
                .expect("the privacy IQ carries a <token>");
            seen = token
                .attrs()
                .optional_string("t")
                .and_then(|t| t.parse::<i64>().ok());
            break;
        }

        let stamped = seen.expect("a first send issues a privacy token");
        assert!(
            (before..=after).contains(&stamped),
            "wire timestamp {stamped} outside [{before}, {after}]"
        );

        client.disconnect().await;
    }
}