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

agent_docs/observability.md

@@ -32,6 +32,17 @@ chokepoints:
   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 outbound frame crosses — five
+distinct send paths reach it (`send_node`, `send_raw_bytes`,
+`send_raw_bytes_burst`, and the ack/receipt workers through the burst), and
+`send_raw_bytes` deliberately bypasses node logging and sent-node waiters — so
+anything that has to see *everything* the client sends belongs there 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

src/client.rs

@@ -103,6 +103,105 @@ 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.
+#[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.
@@ -1453,6 +1552,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

@@ -315,6 +315,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,
@@ -507,6 +508,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),
@@ -823,7 +825,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

@@ -10,7 +10,9 @@ impl Client {
     /// cause the server to close the connection.
     ///
     /// 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> {
         let noise_socket = self.get_noise_socket()?;
         // Wire bytes and the last-sent timestamp are recorded by the noise