feat(tap): TapAudioPipe + TapMetrics (spec §4.1)

- TapMetrics: atomic counters (drop + observe per slice-1 §3.8).
- TapAudioPipe: AudioSource + AudioSink impl over mpsc + VecDeque ring.
  Drop-oldest on overflow (lowest-latency-correct); silence on underflow.
- TAP_PLAYOUT_FRAMES = 5 (100ms @ 20ms/frame; tunable constant per spec §4.1).
- 6 unit tests cover underflow, overflow drops-oldest, sink-full drops,
  disconnected engine, frame round-trip, empty ring.

Spec ref: 2026-06-28-slice-2-agent-tap-design.md §4.1.

crates/rutster-tap/src/lib.rs

@@ -27,18 +27,22 @@
 //! — that would invert the canonical-home of `PcmFrame` and pull the
 //! loopback peer into the tap story.
 
+pub mod metrics;
 pub mod protocol;
+pub mod tap_audio_pipe;
 
 // Re-export PcmFrame so `rutster-tap` consumers (the binary, the echo brain,
 // future brains) get it from one canonical home (spec §3.1).
 pub use rutster_media::PcmFrame;
 
+pub use metrics::{MetricsSnapshot, TapMetrics, TAP_PLAYOUT_FRAMES};
 pub use protocol::{
     decode_envelope, decode_pcm, encode_audio_in, encode_audio_out, encode_bye, encode_error,
     encode_hello, encode_pcm, encode_session_end, AudioPayload, DecodedFrame, DecodedPayload,
     Envelope, ErrorPayload, FrameKind, HelloPayload, Payload, ReasonPayload, SessionEndPayload,
     TapProtoError, PROTOCOL_VERSION, SAMPLES_PER_FRAME as WIRE_SAMPLES_PER_FRAME,
 };
+pub use tap_audio_pipe::TapAudioPipe;
 
 #[cfg(test)]
 mod tests {

crates/rutster-tap/src/metrics.rs

@@ -0,0 +1,62 @@
+//! # TapMetrics — the "observe" half of "drop + observe" (slice-1 §3.8)
+//!
+//! Atomic counters shared between `TapAudioPipe` (hot path — inbound/outbound
+//! drops, playout overflow/underflow) and `TapClient` (seq gaps, unknown
+//! frames, malformed frames, reconnect attempts). All ops are
+//! `fetch_add(_, Ordering::Relaxed)` — we're using the counts for
+//! observability, not synchronization, so relaxed ordering is correct +
+//! cheapest. A snapshot read is taken at log time (e.g. once per second or
+//! on tap-disconnect).
+
+use std::sync::atomic::{AtomicU64, Ordering};
+use std::sync::Arc;
+
+/// Bounded playout ring capacity (spec §4.1: 5 frames = 100 ms at 20 ms/frame).
+/// Tunable *constant* (no runtime config in slice-2; a future-rung concern).
+pub const TAP_PLAYOUT_FRAMES: usize = 5;
+
+#[derive(Debug, Default)]
+pub struct TapMetrics {
+    pub inbound_dropped: AtomicU64,
+    pub outbound_dropped: AtomicU64,
+    pub playout_overflow: AtomicU64,
+    pub playout_underflow: AtomicU64,
+    pub seq_gaps: AtomicU64,
+    pub unknown_frames: AtomicU64,
+    pub malformed_frames: AtomicU64,
+    pub reconnect_attempts: AtomicU64,
+}
+
+impl TapMetrics {
+    pub fn new() -> Arc<Self> {
+        Arc::new(Self::default())
+    }
+
+    /// Read a consistent-ish snapshot. Not atomic across fields (each is
+    /// independently read), but for log/observability purposes that's fine —
+    /// we're reporting "approximately this many X happened," not synchronizing.
+    pub fn snapshot(&self) -> MetricsSnapshot {
+        MetricsSnapshot {
+            inbound_dropped: self.inbound_dropped.load(Ordering::Relaxed),
+            outbound_dropped: self.outbound_dropped.load(Ordering::Relaxed),
+            playout_overflow: self.playout_overflow.load(Ordering::Relaxed),
+            playout_underflow: self.playout_underflow.load(Ordering::Relaxed),
+            seq_gaps: self.seq_gaps.load(Ordering::Relaxed),
+            unknown_frames: self.unknown_frames.load(Ordering::Relaxed),
+            malformed_frames: self.malformed_frames.load(Ordering::Relaxed),
+            reconnect_attempts: self.reconnect_attempts.load(Ordering::Relaxed),
+        }
+    }
+}
+
+#[derive(Debug, Clone, Default)]
+pub struct MetricsSnapshot {
+    pub inbound_dropped: u64,
+    pub outbound_dropped: u64,
+    pub playout_overflow: u64,
+    pub playout_underflow: u64,
+    pub seq_gaps: u64,
+    pub unknown_frames: u64,
+    pub malformed_frames: u64,
+    pub reconnect_attempts: u64,
+}

crates/rutster-tap/src/tap_audio_pipe.rs

@@ -0,0 +1,215 @@
+//! # TapAudioPipe — the seam object (spec §4.1)
+//!
+//! The sync object `RtcSession` holds and `loop_driver` calls via the
+//! `AudioSource`/`AudioSink` trait seam. It's a thin wrapper over two
+//! `tokio::sync::mpsc` channels + a bounded `VecDeque` playout ring:
+//!
+//! ```text
+//!   peer mic → Opus decode → on_pcm_frame() → tx_pcm_in → TapClient → audio_in (WS)
+//!   brain audio_out (WS) → TapClient → rx_audio_out → [playout ring] → next_pcm_frame() → Opus encode → peer
+//! ```
+//!
+//! # Why `VecDeque` for the playout ring (not `mpsc`?)
+//! The playout ring has a specific policy on overflow (drop *oldest*, not
+//! newest — spec §4.1). `mpsc` drops the *newest* on `try_send` when
+//! full (the receiver would see a gap). For a real-time media path, we
+//! want stale (old) frames dropped, not late (new) frames — drop-oldest
+//! keeps the buffer at-or-behind real-time. A `VecDeque` under our
+//! manual `push_back_bounded` policy is the smallest structure that fits.
+
+use std::collections::VecDeque;
+use std::sync::atomic::Ordering;
+use std::sync::Arc;
+
+use rutster_media::{AudioSink, AudioSource, PcmFrame};
+use tokio::sync::mpsc;
+use tracing::{debug, trace};
+
+use crate::metrics::{TapMetrics, TAP_PLAYOUT_FRAMES};
+
+pub struct TapAudioPipe {
+    // Core → brain (inbound decoded PCM from peer):
+    tx_pcm_in: mpsc::Sender<PcmFrame>,
+    // Brain → core (playout buffer):
+    rx_audio_out: mpsc::Receiver<PcmFrame>,
+    // Bounded ring between rx_audio_out and next_pcm_frame (spec §4.1).
+    playout_ring: VecDeque<PcmFrame>,
+    metrics: Arc<TapMetrics>,
+}
+
+impl TapAudioPipe {
+    pub fn new(
+        tx_pcm_in: mpsc::Sender<PcmFrame>,
+        rx_audio_out: mpsc::Receiver<PcmFrame>,
+        metrics: Arc<TapMetrics>,
+    ) -> Self {
+        Self {
+            tx_pcm_in,
+            rx_audio_out,
+            playout_ring: VecDeque::with_capacity(TAP_PLAYOUT_FRAMES),
+            metrics,
+        }
+    }
+
+    /// Drain the inbound mpsc into the playout ring. Called by
+    /// `next_pcm_frame` (one drain per source tick). Returns the count
+    /// drained. **Hot-path policy:** every drop is counted (drop + observe).
+    fn drain_inbound(&mut self) -> usize {
+        let mut drained = 0;
+        loop {
+            match self.rx_audio_out.try_recv() {
+                Ok(frame) => {
+                    // push_back_bounded: drop oldest if full (drop-oldest policy).
+                    if self.playout_ring.len() >= TAP_PLAYOUT_FRAMES {
+                        self.playout_ring.pop_front();
+                        self.metrics
+                            .playout_overflow
+                            .fetch_add(1, Ordering::Relaxed);
+                        debug!(overflow = true, "playout ring overflow; dropped oldest");
+                    }
+                    self.playout_ring.push_back(frame);
+                    drained += 1;
+                }
+                Err(mpsc::error::TryRecvError::Empty) => break,
+                Err(mpsc::error::TryRecvError::Disconnected) => {
+                    // Engine task gone. Silence until TapClient reconnects
+                    // via a fresh mpsc (handled by TapEngine's reconnect loop).
+                    trace!("rx_audio_out disconnected; silence until reconnect");
+                    break;
+                }
+            }
+        }
+        drained
+    }
+}
+
+impl AudioSource for TapAudioPipe {
+    /// Take the next brain-proposed PCM frame to send to the peer.
+    /// `None` = silence (loop_driver emits Opus silence on None — slice-1).
+    fn next_pcm_frame(&mut self) -> Option<PcmFrame> {
+        // First drain any frames the engine task has queued; then pop one.
+        self.drain_inbound();
+        match self.playout_ring.pop_front() {
+            Some(frame) => Some(frame),
+            None => {
+                // Underflow: brain slower than the 20ms tick. Silence.
+                self.metrics
+                    .playout_underflow
+                    .fetch_add(1, Ordering::Relaxed);
+                None
+            }
+        }
+    }
+}
+
+impl AudioSink for TapAudioPipe {
+    /// Receive a decoded PCM frame from the peer. Must not block
+    /// (slice-1 §3.3 contract — `on_pcm_frame` runs in the 20ms loop).
+    /// `try_send` (not `send`) — if the channel is full (engine task slow
+    /// or gone), drop + count (hot-path "drop + observe, don't crash").
+    fn on_pcm_frame(&mut self, frame: PcmFrame) {
+        if self.tx_pcm_in.try_send(frame).is_err() {
+            self.metrics.inbound_dropped.fetch_add(1, Ordering::Relaxed);
+            trace!("inbound PCM dropped (channel full)");
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[allow(clippy::type_complexity)] // test helper: 5-tuple of channel ends; not worth a struct.
+    fn channels() -> (
+        mpsc::Sender<PcmFrame>,
+        mpsc::Receiver<PcmFrame>,
+        mpsc::Sender<PcmFrame>,
+        mpsc::Receiver<PcmFrame>,
+        Arc<TapMetrics>,
+    ) {
+        // tx_pcm_in (inbound PCM → engine), rx_audio_out (engine → playout ring)
+        let (tx_pcm_in, rx_pcm_in) = mpsc::channel(64);
+        let (tx_audio_out, rx_audio_out) = mpsc::channel(64);
+        let metrics = TapMetrics::new();
+        (tx_pcm_in, rx_pcm_in, tx_audio_out, rx_audio_out, metrics)
+    }
+
+    #[test]
+    fn source_returns_none_on_empty_ring() {
+        let (_, _, tx_audio_out, rx_audio_out, metrics) = channels();
+        let mut pipe = TapAudioPipe::new(tx_audio_out, rx_audio_out, metrics);
+        assert!(pipe.next_pcm_frame().is_none());
+    }
+
+    #[test]
+    fn source_returns_frame_pushed_by_engine() {
+        let (_, _, tx_audio_out, rx_audio_out, metrics) = channels();
+        let mut pipe = TapAudioPipe::new(tx_audio_out.clone(), rx_audio_out, metrics.clone());
+        // Engine task (simulated): push an audio_out frame.
+        let mut frame = PcmFrame::zeroed();
+        frame.samples[0] = 42;
+        tx_audio_out.blocking_send(frame).unwrap();
+        // Source should now drain + return it.
+        let got = pipe.next_pcm_frame().expect("frame present");
+        assert_eq!(got.samples[0], 42);
+        assert_eq!(metrics.playout_underflow.load(Ordering::Relaxed), 0);
+    }
+
+    #[test]
+    fn overflow_drops_oldest_not_newest() {
+        let (_, _, tx_audio_out, rx_audio_out, metrics) = channels();
+        let mut pipe = TapAudioPipe::new(tx_audio_out.clone(), rx_audio_out, metrics.clone());
+        // Push TAP_PLAYOUT_FRAMES + 3 frames. The first 3 should be dropped
+        // (oldest), keeping the last TAP_PLAYOUT_FRAMES.
+        for i in 0..(TAP_PLAYOUT_FRAMES + 3) {
+            let mut f = PcmFrame::zeroed();
+            f.samples[0] = i as i16;
+            tx_audio_out.blocking_send(f).unwrap();
+        }
+        // Drain into the ring via next_pcm_frame's drain step (returns the
+        // first popped, so we call N+1 times to fully drain + assert).
+        let first = pipe.next_pcm_frame().expect("first frame after drain");
+        assert_eq!(first.samples[0], 3); // first 3 (0,1,2) were dropped
+        assert_eq!(metrics.playout_overflow.load(Ordering::Relaxed), 3);
+    }
+
+    #[test]
+    fn underflow_increments_counter() {
+        let (_, _, tx_audio_out, rx_audio_out, metrics) = channels();
+        let mut pipe = TapAudioPipe::new(tx_audio_out, rx_audio_out, metrics.clone());
+        pipe.next_pcm_frame(); // None -> underflow
+        pipe.next_pcm_frame(); // None -> underflow
+        assert_eq!(metrics.playout_underflow.load(Ordering::Relaxed), 2);
+    }
+
+    #[test]
+    fn sink_full_drops_and_counts() {
+        // Tiny inbound channel to force overflow.
+        let (tx_pcm_in, _rx_pcm_in, _tx_audio_out, rx_audio_out, metrics) = channels();
+        let mut pipe = TapAudioPipe::new(tx_pcm_in, rx_audio_out, metrics.clone());
+        // Channel capacity 64 — push 100 frames.
+        for _ in 0..100 {
+            pipe.on_pcm_frame(PcmFrame::zeroed());
+        }
+        assert!(metrics.inbound_dropped.load(Ordering::Relaxed) > 0);
+    }
+
+    #[test]
+    fn disconnected_engine_returns_none() {
+        let (_, _, tx_audio_out, rx_audio_out, metrics) = channels();
+        let mut pipe = TapAudioPipe::new(tx_audio_out, rx_audio_out, metrics);
+        // Drop the sender → rx_audio_out sees Disconnected → None (silence).
+        // (Implicit drop at scope exit here; explicit next_pcm_frame below
+        // reads Disconnected.)
+        assert!(pipe.next_pcm_frame().is_none());
+    }
+
+    #[test]
+    fn disconnected_engine_returns_none_after_close() {
+        let (tx_pcm_in, _rx_pcm_in, tx_audio_out, rx_audio_out, metrics) = channels();
+        let mut pipe = TapAudioPipe::new(tx_pcm_in, rx_audio_out, metrics);
+        drop(tx_audio_out); // close the engine→playout-ring direction
+                            // Now next_pcm_frame should return None (silence) — Disconnected path.
+        assert!(pipe.next_pcm_frame().is_none());
+    }
+}