feat(client): observe what the client sends, not only what it receives

agent_docs/observability.md

@@ -28,10 +28,25 @@ must follow.
 `wacore::stats::SessionStats`, owned by each `Client`. Recorded at exactly two
 chokepoints:
 
-- **Sent**: the noise sender task (`NoiseSocket::with_stats`) after the
+- **Sent**: the noise sender task (`NoiseSocket::with_observers`) after the
   transport write — post-noise wire bytes (frame header + AEAD tag included).
 - **Received**: the read loop (`node_io.rs`) per `DataReceived` batch.
 
+That sent chokepoint is the *only* place every post-handshake frame crosses (the
+XX/IK exchange writes to the transport directly, before this socket exists). Two
+functions reach it — `send_raw_bytes` and `send_raw_bytes_burst` — and everything
+else funnels through one of those: `send_node` and every IQ through the first,
+the ack and delivery-receipt workers through the second. `send_raw_bytes`
+deliberately bypasses node logging and sent-node waiters, so anything that has to
+see everything the client sends on the session socket belongs at the chokepoint
+and nowhere else.
+
+`Event::SentFrame` is the other thing wired into it
+(`Client::acquire_sent_frame_forwarding()`, lease-gated like `RawNode`): it hands
+over the marshaled plaintext of each frame the transport accepted. Both halves
+travel to the socket as `SendObservers`, so the next observer plugs in there
+instead of widening `do_handshake` again.
+
 It also owns the activity timestamps the keepalive dead-socket watchdog reads:
 `last_data_received_ms` (one clock read per received transport event, plus one
 more when that event carries several frames, so a slow drain is not read as
@@ -42,8 +57,8 @@ cost a clock read on every frame written, which is the client's hottest path
 and a call out of the module on wasm32/embedded. `frames_sent` answers "is it
 still sending?" for free. Message-level counters piggyback on the existing
 `telemetry::send`/`recv` chokepoints; reconnect attempts are counted in the
-run loop. VoIP relay sockets pass `None` and are not counted — this is the
-main WA session socket only.
+run loop. VoIP relay sockets pass `SendObservers::default()` and are not counted
+— this is the main WA session socket only.
 
 ### 2. `Client::memory_report()` — retained memory (on demand)
 

src/client.rs

@@ -104,6 +104,113 @@ impl Drop for RawNodeLease {
     }
 }
 
+/// Lease that keeps sent-frame events enabled for one consumer.
+///
+/// Dropping the final lease disables forwarding. The lease holds only a weak
+/// client reference, so it cannot keep the client alive.
+///
+/// It gates, it does not fence, and one aggregate count gates them all rather
+/// than one per lease: a frame captured while any lease was alive can still
+/// arrive just after this one drops, and which handlers receive it is a matter of
+/// subscription, not of who holds a lease. Every gated kind works this way.
+/// Making the drop wait for in-flight dispatches to drain would instead deadlock
+/// an observer that drops its lease from inside its own handler, which is the
+/// natural way to record one frame and stop.
+#[must_use = "dropping the lease immediately releases sent-frame forwarding"]
+pub struct SentFrameLease {
+    client: std::sync::Weak<Client>,
+}
+
+impl Drop for SentFrameLease {
+    fn drop(&mut self) {
+        let Some(client) = self.client.upgrade() else {
+            return;
+        };
+        client.sent_frame_tap.release();
+    }
+}
+
+/// Publishes the plaintext frames that reached the transport as
+/// [`Event::SentFrame`](wacore::types::events::Event::SentFrame).
+///
+/// The client owns it and hands the noise sender a clone of the `Arc`, the same
+/// way it hands over [`SessionStats`](wacore::stats::SessionStats): the gate has
+/// to be readable from the one point every send crosses, and that task cannot
+/// hold the client without keeping it alive.
+pub(crate) struct SentFrameTap {
+    /// Number of consumers currently requesting the event.
+    forwarding: AtomicUsize,
+    bus: wacore::types::events::CoreEventBus,
+    /// Proves the no-lease path builds nothing, rather than only that it
+    /// dispatches nothing.
+    #[cfg(test)]
+    published: AtomicUsize,
+}
+
+impl SentFrameTap {
+    pub(crate) fn new(bus: wacore::types::events::CoreEventBus) -> Self {
+        Self {
+            forwarding: AtomicUsize::new(0),
+            bus,
+            #[cfg(test)]
+            published: AtomicUsize::new(0),
+        }
+    }
+
+    /// Enable forwarding for one consumer. The public door is
+    /// [`Client::acquire_sent_frame_forwarding`], which pairs this with a lease
+    /// that releases it on drop; a caller here owns that pairing itself.
+    pub(crate) fn acquire(&self) {
+        let incremented = self
+            .forwarding
+            .fetch_update(Ordering::Relaxed, Ordering::Relaxed, |count| {
+                count.checked_add(1)
+            })
+            .is_ok();
+        assert!(incremented, "sent-frame forwarding lease counter overflow");
+    }
+
+    pub(crate) fn release(&self) {
+        let previous = self.forwarding.fetch_sub(1, Ordering::Relaxed);
+        debug_assert!(previous > 0, "sent-frame forwarding lease underflow");
+    }
+
+    #[inline]
+    pub(crate) fn enabled(&self) -> bool {
+        self.forwarding.load(Ordering::Relaxed) != 0
+    }
+
+    /// Hand one frame to the observers.
+    ///
+    /// The dispatch is caught: a consumer that only watches must not be able to
+    /// take the send pipeline down with it, and this runs on the noise sender
+    /// task, whose death would end every send on the connection. Containment is
+    /// per dispatch, not per handler, so a panicking observer costs this frame
+    /// for the observers behind it — the bus offers no per-handler isolation for
+    /// any kind, and plugins already wrap their own handlers. A handler that
+    /// *blocks* still stalls sends, the contract every handler has on the read
+    /// loop.
+    pub(crate) fn publish(&self, plaintext: bytes::Bytes) {
+        #[cfg(test)]
+        self.published.fetch_add(1, Ordering::Relaxed);
+        let dispatch = std::panic::AssertUnwindSafe(|| {
+            self.bus.dispatch(Event::SentFrame(
+                wacore::types::events::SentFrame::builder()
+                    .plaintext(plaintext)
+                    .build(),
+            ));
+        });
+        if std::panic::catch_unwind(dispatch).is_err() {
+            warn!("A sent-frame observer panicked; the send pipeline is unaffected.");
+        }
+    }
+
+    #[cfg(test)]
+    pub(crate) fn published(&self) -> usize {
+        self.published.load(Ordering::Relaxed)
+    }
+}
+
 /// Filter for matching incoming stanzas (nodes) by tag and attributes.
 ///
 /// Used with [`Client::wait_for_node`] to wait for specific stanzas.
@@ -1459,6 +1566,10 @@ pub struct Client {
     /// forwarding.
     decrypted_payload_forwarding: AtomicUsize,
 
+    /// Gate and publisher for `Event::SentFrame`. Behind an `Arc` because the
+    /// noise sender task reads it; see [`SentFrameTap`].
+    pub(crate) sent_frame_tap: Arc<SentFrameTap>,
+
     /// Stanza interceptors, behind the same copy-on-write snapshot the event
     /// bus uses: reading one costs a refcount bump, so the read loop allocates
     /// nothing per stanza. Registering is the rare side, and pays the copy.

src/client/accessors.rs

@@ -88,6 +88,33 @@ impl Client {
         self.decrypted_payload_forwarding.load(Ordering::Relaxed) != 0
     }
 
+    /// Acquire sent-frame forwarding for one consumer.
+    ///
+    /// [`Event::SentFrame`] stays enabled until every acquired lease is dropped.
+    /// While none is held nothing is emitted and nothing is cloned: the send
+    /// path costs one relaxed atomic load.
+    ///
+    /// This is the outbound counterpart of
+    /// [`acquire_raw_node_forwarding`](Self::acquire_raw_node_forwarding), and
+    /// unlike [`wait_for_sent_node`](Self::wait_for_sent_node) it is neither
+    /// filtered nor one-shot and covers every send path, including the ones that
+    /// never build a `Node`.
+    ///
+    /// [`Event::SentFrame`]: wacore::types::events::Event::SentFrame
+    pub fn acquire_sent_frame_forwarding(self: &Arc<Self>) -> SentFrameLease {
+        self.sent_frame_tap.acquire();
+        SentFrameLease {
+            client: Arc::downgrade(self),
+        }
+    }
+
+    /// Only tests ask this: the send path reads the gate through the tap the
+    /// noise sender already holds, not through the client.
+    #[cfg(test)]
+    pub(crate) fn sent_frame_forwarding_enabled(&self) -> bool {
+        self.sent_frame_tap.enabled()
+    }
+
     /// Register an interceptor that sees each decoded stanza before the
     /// built-in pipeline, and may take it.
     ///

src/client/lifecycle.rs

@@ -317,6 +317,7 @@ impl Client {
         let (tx, rx) = async_channel::bounded(32);
 
         let device_topology = device_topology::DeviceTopology::new();
+        let sent_frame_tap = Arc::new(SentFrameTap::new(core.event_bus.clone()));
         let this = Self {
             runtime: runtime.clone(),
             core,
@@ -509,6 +510,7 @@ impl Client {
             alloc_meter: std::sync::OnceLock::new(),
             raw_node_forwarding: AtomicUsize::new(0),
             decrypted_payload_forwarding: AtomicUsize::new(0),
+            sent_frame_tap,
             stanza_interceptors: std::sync::RwLock::new(Arc::new(Vec::new())),
             stanza_interceptor_count: AtomicUsize::new(0),
             next_interceptor_id: AtomicU64::new(0),
@@ -918,7 +920,8 @@ impl Client {
             &self.ik_handshake_failures,
             transport.clone(),
             &mut transport_events,
-            Some(self.stats.clone()),
+            crate::socket::noise_socket::SendObservers::with_stats(self.stats.clone())
+                .with_sent_frames(self.sent_frame_tap.clone()),
         )
         .await
         {

src/client/messaging.rs

@@ -14,7 +14,9 @@ impl Client {
     /// the connection, which is why the format byte is checked here.
     ///
     /// This bypasses node logging and `sent_node_waiter` resolution — use
-    /// [`send_node`](Client::send_node) for normal stanza sending.
+    /// [`send_node`](Client::send_node) for normal stanza sending. It is still
+    /// observed: `Event::SentFrame` is emitted from the noise sender, past every
+    /// bypass here.
     pub async fn send_raw_bytes(&self, plaintext: Vec<u8>) -> Result<(), ClientError> {
         wacore_binary::util::check_plain_payload(&plaintext).map_err(SocketError::Marshal)?;
         let noise_socket = self.get_noise_socket()?;