feat(core): expose typed stanza responses and retries

src/client.rs

@@ -1044,56 +1044,90 @@ fn build_pong(to: String, id: Option<&str>) -> wacore_binary::Node {
     builder.build()
 }
 
+/// Compare decoded attribute values by their wire display without allocating.
+#[inline]
+fn value_refs_display_equal(
+    left: &wacore_binary::node::ValueRef<'_>,
+    right: &wacore_binary::node::ValueRef<'_>,
+) -> bool {
+    use wacore_binary::node::ValueRef;
+
+    match (left, right) {
+        (ValueRef::String(left), ValueRef::String(right)) => left == right,
+        (ValueRef::Jid(left), ValueRef::Jid(right)) => left.display_eq_jid(right),
+        (ValueRef::String(left), ValueRef::Jid(right)) => right.display_eq(left),
+        (ValueRef::Jid(left), ValueRef::String(right)) => left.display_eq(right),
+    }
+}
+
+#[derive(Clone, Copy)]
+enum AckParticipantPolicy {
+    Preserve,
+    OmitReceiptDestinationDuplicate,
+}
+
+#[inline]
+fn ack_participant<'node, 'data>(
+    node: &'node wacore_binary::NodeRef<'data>,
+    from: &wacore_binary::node::ValueRef<'data>,
+    policy: AckParticipantPolicy,
+) -> Option<&'node wacore_binary::node::ValueRef<'data>> {
+    node.get_attr("participant")
+        .filter(|participant| match policy {
+            AckParticipantPolicy::Preserve => true,
+            AckParticipantPolicy::OmitReceiptDestinationDuplicate => {
+                node.tag != "receipt" || !value_refs_display_equal(participant, from)
+            }
+        })
+}
+
 /// Build an `<ack/>` for the given stanza, matching WA Web / whatsmeow behavior:
 ///
 /// - `class` = original stanza tag
-/// - `id`, `to` (flipped from `from`), `participant` copied from original
+/// - `id`, `to` (flipped from `from`) copied from original
+/// - `participant` follows the generic or receipt-specialized policy
 /// - `from` = own device PN, only for message acks
-/// - `type` echoed for non-message stanzas (whatsmeow: `node.Tag != "message"`),
-///   except `notification type="encrypt"` with `<identity/>` child (WA Web drops type there).
+/// - `type` echoed when present, except `notification type="encrypt"` with
+///   an `<identity/>` child
 ///
 /// For receipt acks, WA Web uses `MAYBE_CUSTOM_STRING(ackString)` where
 /// `ackString = maybeAttrString("type")` — so `type` is only included when
 /// explicitly present on the incoming receipt (delivery receipts normally
 /// have no type attribute, meaning the ack also has no type).
+///
 /// Encode an ack stanza directly to bytes, bypassing Node + marshal_auto.
 /// Acks are the most frequent outbound stanza (~1 per inbound message).
 fn encode_ack_bytes(
     node: &wacore_binary::NodeRef<'_>,
     own_device_pn: Option<&Jid>,
-) -> Result<Option<Vec<u8>>, wacore_binary::error::BinaryError> {
+    participant_policy: AckParticipantPolicy,
+) -> Result<Vec<u8>, crate::features::StanzaResponseError> {
     use wacore_binary::encoder::{ByteWriter, EncodeNode, Encoder};
 
-    let Some(id_val) = node.get_attr("id") else {
-        return Ok(None);
-    };
-    let Some(from_val) = node.get_attr("from") else {
-        return Ok(None);
-    };
-    // WAWebReceiptAck: `participant: r && r !== e ? DEVICE_JID(r) : DROP_ATTR`.
-    // Drop the attribute when it would duplicate `to` (which is the flipped `from`).
-    let participant_val = node.get_attr("participant").filter(|p| {
-        let p_str = p.as_str();
-        let from_str = from_val.as_str();
-        p_str.as_ref() != from_str.as_ref()
-    });
+    let id_val = crate::features::required_stanza_attr(node, "id")?;
+    let from_val = crate::features::required_stanza_attr(node, "from")?;
+    let tag = node.tag.as_ref();
+    let participant_val = ack_participant(node, from_val, participant_policy);
     // Server expects `recipient` echoed back so it can route the ack to the
     // origin companion/device (hosted-companion, peer, LID-routed stanzas).
     // Dropping it makes the server close the stream with `<stream:error><ack/>`.
     let recipient_val = node.get_attr("recipient");
-    let tag = node.tag.as_ref();
 
-    let typ_val = if tag != "message" && !is_encrypt_identity_notification(node) {
+    let typ_val = if !is_encrypt_identity_notification(node) {
         node.get_attr("type")
     } else {
         None
     };
 
-    let include_from = tag == "message" && own_device_pn.is_some();
+    let own_device_pn = if tag == "message" {
+        Some(own_device_pn.ok_or(crate::features::StanzaResponseError::MissingLocalIdentity)?)
+    } else {
+        None
+    };
 
     // Count attrs: class + id + to + optional(from, participant, recipient, type)
     let attr_count = 3
-        + usize::from(include_from)
+        + usize::from(own_device_pn.is_some())
         + usize::from(participant_val.is_some())
         + usize::from(recipient_val.is_some())
         + usize::from(typ_val.is_some());
@@ -1161,15 +1195,15 @@ fn encode_ack_bytes(
         participant: participant_val,
         recipient: recipient_val,
         typ: typ_val,
-        own_pn: if include_from { own_device_pn } else { None },
+        own_pn: own_device_pn,
         tag_str: tag,
         attr_count,
     };
 
     let mut buf = Vec::with_capacity(64);
     let mut encoder = Encoder::new_vec(&mut buf)?;
     encoder.write_node(&ack)?;
-    Ok(Some(buf))
+    Ok(buf)
 }
 
 /// Minimal `<message>` stanza carrying the attrs `encode_ack_bytes` needs,
@@ -1195,20 +1229,21 @@ fn message_ack_source_node(info: &crate::types::message::MessageInfo) -> Node {
     builder.build()
 }
 
-/// Build an ack Node (used in tests for structure verification).
+/// Build an automatic ack Node (used in tests for structure verification).
 #[cfg(test)]
 fn build_ack_node(node: &wacore_binary::NodeRef<'_>, own_device_pn: Option<&Jid>) -> Option<Node> {
     let id = node.get_attr("id")?.to_node_value();
     let from_ref = node.get_attr("from")?;
     let from = from_ref.to_node_value();
-    // Drop participant when it duplicates `to` (the flipped `from`).
-    let participant = node
-        .get_attr("participant")
-        .filter(|p| p.as_str().as_ref() != from_ref.as_str().as_ref())
-        .map(|v| v.to_node_value());
-    let recipient = node.get_attr("recipient").map(|v| v.to_node_value());
     let tag = node.tag.as_ref();
-    let typ = if tag != "message" && !is_encrypt_identity_notification(node) {
+    let participant = ack_participant(
+        node,
+        from_ref,
+        AckParticipantPolicy::OmitReceiptDestinationDuplicate,
+    )
+    .map(|value| value.to_node_value());
+    let recipient = node.get_attr("recipient").map(|v| v.to_node_value());
+    let typ = if !is_encrypt_identity_notification(node) {
         node.get_attr("type").map(|v| v.to_node_value())
     } else {
         None
@@ -1243,7 +1278,7 @@ fn is_encrypt_identity_notification(node: &wacore_binary::NodeRef<'_>) -> bool {
     node.tag == "notification"
         && node
             .get_attr("type")
-            .is_some_and(|v| v.as_str() == "encrypt")
+            .is_some_and(|value| value == "encrypt")
         && node.get_optional_child("identity").is_some()
 }
 

src/client/lifecycle.rs

@@ -990,9 +990,8 @@ impl Client {
         self.is_connected.load(Ordering::Acquire)
     }
 
-    /// Force the connected flag (tests only): the facade's connect path now gates on `is_connected`,
-    /// so a unit test driving `spawn_call`/`place_call` must mark the client connected first.
-    #[cfg(all(test, feature = "voip-runtime"))]
+    /// Force the connected flag for tests that exercise connected-only operations.
+    #[cfg(test)]
     pub(crate) fn set_connected_for_test(&self, connected: bool) {
         self.is_connected.store(connected, Ordering::Release);
     }

src/client/node_io.rs

@@ -589,6 +589,18 @@ impl Client {
         }
     }
 
+    #[inline]
+    fn encode_ack_from_snapshot(
+        &self,
+        node: &wacore_binary::NodeRef<'_>,
+        participant_policy: AckParticipantPolicy,
+    ) -> Result<Vec<u8>, crate::features::StanzaResponseError> {
+        let device = self.persistence_manager.get_device_snapshot();
+        let encoded = encode_ack_bytes(node, device.pn.as_ref(), participant_policy);
+        drop(device);
+        encoded
+    }
+
     /// Build and send an <ack/> node corresponding to the given stanza.
     #[cfg_attr(
         feature = "tracing",
@@ -604,10 +616,10 @@ impl Client {
         if !self.is_connected() {
             return Err(ClientError::NotConnected);
         }
-        let own_pn = self.get_pn();
-        let buf = match encode_ack_bytes(node, own_pn.as_ref()) {
-            Ok(Some(buf)) => buf,
-            Ok(None) => return Ok(()),
+        let buf = match self
+            .encode_ack_from_snapshot(node, AckParticipantPolicy::OmitReceiptDestinationDuplicate)
+        {
+            Ok(buf) => buf,
             Err(e) => {
                 log::warn!("Failed to encode ack: {e}");
                 return Ok(());
@@ -616,21 +628,39 @@ impl Client {
         self.send_raw_bytes(buf).await
     }
 
+    /// Confirm a received stanza using its original borrowed node.
+    ///
+    /// Unlike the tolerant automatic receive path, malformed input is returned
+    /// to the caller and no successful outcome is reported unless the response
+    /// reaches the transport.
+    #[cfg_attr(
+        feature = "tracing",
+        tracing::instrument(name = "wa.conn.ack_explicit", level = "debug", skip_all, err(Debug))
+    )]
+    pub async fn acknowledge_stanza(
+        &self,
+        stanza: &wacore_binary::NodeRef<'_>,
+    ) -> Result<(), crate::features::StanzaResponseError> {
+        let bytes = self.encode_ack_from_snapshot(stanza, AckParticipantPolicy::Preserve)?;
+        self.send_raw_bytes(bytes).await?;
+        Ok(())
+    }
+
     /// Send a transport ack so the server stops replaying a stanza from the
     /// offline queue. Awaitable so callers can order it after a retry receipt
     /// in a single flushed task.
     pub(crate) async fn send_transport_ack(&self, info: &crate::types::message::MessageInfo) {
         let source = message_ack_source_node(info);
-        let own_pn = self.get_pn();
-        match encode_ack_bytes(&source.as_node_ref(), own_pn.as_ref()) {
-            Ok(Some(buf)) => {
+        let encoded =
+            self.encode_ack_from_snapshot(&source.as_node_ref(), AckParticipantPolicy::Preserve);
+        match encoded {
+            Ok(buf) => {
                 if let Err(e) = self.send_raw_bytes(buf).await
                     && !e.is_transport_unavailable()
                 {
                     log::warn!("Failed to send transport ack for undecryptable message: {e:?}");
                 }
             }
-            Ok(None) => {}
             Err(e) => log::warn!("Failed to encode transport ack: {e}"),
         }
     }
@@ -655,10 +685,8 @@ impl Client {
         self: &Arc<Self>,
         node: &wacore_binary::NodeRef<'_>,
     ) {
-        let own_pn = self.get_pn();
-        let buf = match encode_ack_bytes(node, own_pn.as_ref()) {
-            Ok(Some(b)) => b,
-            Ok(None) => return,
+        let buf = match self.encode_ack_from_snapshot(node, AckParticipantPolicy::Preserve) {
+            Ok(buf) => buf,
             Err(e) => {
                 log::warn!("Failed to encode node transport ack: {e}");
                 return;