feat(tap): TapClient WSS pump loop (spec §4.2)

- run_tap_client: drives a connected WebSocketStream with a tokio::select!
  over rx_pcm_in (inbound PCM → audio_in WS frame) and ws.next() (brain
  frame → audio_out mpsc or control handling).
- Handshake: send hello, await brain hello (bounded 2s timeout).
- seq gap detection (log + count, never drop on gap — spec §3.1).
- Hot-path errors (encode/decode failures, send/recv failures) are logged
  + counted; TapClientError is returned only on graceful close, hello
  timeout, or WS errors. The TapEngine (Task 7) decides reconnect policy.
- Pure-helper unit test (elapsed_ms); full pump behavior is exercised
  against the in-process EchoServer in the Task 8 integration test.

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

Cargo.lock

@@ -1209,6 +1209,7 @@ version = "0.1.0"
 dependencies = [
  "base64",
  "futures-util",
+ "rutster-call-model",
  "rutster-media",
  "serde",
  "serde_json",

crates/rutster-tap/Cargo.toml

@@ -6,6 +6,7 @@ edition.workspace = true
 repository.workspace = true
 
 [dependencies]
+rutster-call-model = { path = "../rutster-call-model" }   # ChannelId newtype (spec §5.2)
 rutster-media = { path = "../rutster-media" }   # PcmFrame (re-exported — spec §3.1)
 tokio = { workspace = true }
 tokio-tungstenite = { workspace = true }

crates/rutster-tap/src/lib.rs

@@ -30,6 +30,7 @@
 pub mod metrics;
 pub mod protocol;
 pub mod tap_audio_pipe;
+pub mod tap_client;
 
 // Re-export PcmFrame so `rutster-tap` consumers (the binary, the echo brain,
 // future brains) get it from one canonical home (spec §3.1).
@@ -43,6 +44,7 @@ pub use protocol::{
     TapProtoError, PROTOCOL_VERSION, SAMPLES_PER_FRAME as WIRE_SAMPLES_PER_FRAME,
 };
 pub use tap_audio_pipe::TapAudioPipe;
+pub use tap_client::{run_tap_client, TapClientError};
 
 #[cfg(test)]
 mod tests {

crates/rutster-tap/src/tap_client.rs

@@ -0,0 +1,265 @@
+//! # TapClient — the async WSS pump loop (spec §4.2)
+//!
+//! Lives inside the `TapEngine` task (in the binary crate). Owns one WS
+//! connection + the pump loop that shovels PCM between mpsc channels and
+//! WS frames. The media loop never sees it — `TapAudioPipe` is the only
+//! sync surface `loop_driver` touches.
+//!
+//! # Why TapClient never decides to reconnect
+//!
+//! Reconnect is the `TapEngine`'s job (spec §4.3). On any WS close / error,
+//! `run_tap_client` returns; the engine rebuilds the client + applies
+//! backoff. This keeps "the connection" (the wire) and "the reconnect
+//! policy" (the backoff) as separate concerns — the one knows the wire,
+//! the other knows the timing.
+//!
+//! # The `close` oneshot is shared, not owned
+//!
+//! `run_tap_client` takes `close: &mut oneshot::Receiver<()>` (a shared
+//! borrow of the receiver), NOT `close: oneshot::Receiver<()>` by value.
+//! Rationale: the `TapEngine` reconnect loop (Task 7) holds ONE close
+//! oneshot per session and passes `&mut close` to each reconnect attempt
+//! of `run_tap_client`. If this fn consumed it by value, each reconnect
+//! would need a fresh oneshot — breaking the model where the binary holds
+//! one close signal per session that survives across reconnects.
+
+use std::sync::atomic::Ordering;
+use std::sync::Arc;
+use std::time::Instant;
+
+use futures_util::{SinkExt, StreamExt};
+use rutster_call_model::ChannelId;
+use rutster_media::PcmFrame;
+use thiserror::Error;
+use tokio::sync::{mpsc, oneshot};
+use tokio_tungstenite::tungstenite::Message;
+use tokio_tungstenite::WebSocketStream;
+use tracing::{debug, info, trace, warn};
+
+use crate::metrics::TapMetrics;
+use crate::protocol::{
+    decode_envelope, encode_audio_in, encode_bye, encode_hello, DecodedPayload, TapProtoError,
+};
+
+#[derive(Debug, Error)]
+pub enum TapClientError {
+    #[error("WebSocket error: {0}")]
+    Ws(#[from] tokio_tungstenite::tungstenite::Error),
+    #[error("Protocol error: {0}")]
+    Proto(#[from] TapProtoError),
+    #[error("hello handshake timed out")]
+    HelloTimeout,
+    #[error("close signal received")]
+    Closed,
+}
+
+/// Run the pump loop on an already-connected `WebSocketStream`.
+///
+/// Returns `Ok(())` on graceful close (brain sent `bye` or the close
+/// oneshot fired). Returns `Err(_)` on WS / protocol errors — the
+/// `TapEngine` caller decides to retry.
+///
+/// `close` is a shared borrow (`&mut oneshot::Receiver<()>`) so the
+/// `TapEngine` reconnect loop (Task 7) can share one close signal across
+/// reconnect attempts of the same session (see module docs).
+pub async fn run_tap_client<T>(
+    mut ws: WebSocketStream<T>,
+    session_id: ChannelId,
+    mut rx_pcm_in: mpsc::Receiver<PcmFrame>,
+    tx_audio_out: mpsc::Sender<PcmFrame>,
+    metrics: Arc<TapMetrics>,
+    close: &mut oneshot::Receiver<()>,
+) -> Result<(), TapClientError>
+where
+    T: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin,
+{
+    let session_start = Instant::now();
+    let mut seq_egress: u64 = 0;
+
+    // === Handshake: send hello, await brain hello (bounded 2s). ===
+    let hello_str = encode_hello(&session_id.to_string(), seq_egress, 0)?;
+    seq_egress += 1;
+    ws.send(Message::Text(hello_str)).await?;
+    info!(%session_id, "sent hello to brain");
+
+    let hello_brain = tokio::time::timeout(
+        std::time::Duration::from_secs(2),
+        wait_for_brain_hello(&mut ws),
+    )
+    .await;
+    // Every arm either returns or assigns last_seq_ingress — using a
+    // match expression (rather than `let mut x = None; match { x = ... }`)
+    // avoids the dead-store warning that an unconditional `None` initializer
+    // would produce under `-D unused-assignments`.
+    let mut last_seq_ingress: Option<u64> = match hello_brain {
+        Ok(Ok(brain_seq)) => {
+            info!(%session_id, "brain hello acked");
+            Some(brain_seq)
+        }
+        Ok(Err(e)) => {
+            warn!(error = ?e, %session_id, "brain hello failed");
+            return Err(e);
+        }
+        Err(_) => return Err(TapClientError::HelloTimeout),
+    };
+
+    // === Pump loop. ===
+    loop {
+        tokio::select! {
+            // Close signal from the binary (Channel::Closing). `close` is a
+            // shared `&mut oneshot::Receiver<()>`; `&mut *close` reborrows
+            // the inner receiver so the engine's reconnect loop can hand
+            // us the same oneshot across reconnect attempts of one session.
+            _ = &mut *close => {
+                info!(%session_id, "close signal; sending bye + closing");
+                let bye_str = encode_bye("core_shutdown", seq_egress, elapsed_ms(session_start))?;
+                let _ = ws.send(Message::Text(bye_str)).await;
+                let _ = ws.close(None).await;
+                return Err(TapClientError::Closed);
+            }
+            // Inbound PCM from peer → audio_in WS frame.
+            frame = rx_pcm_in.recv() => {
+                let Some(frame) = frame else {
+                    // Peer side gone; just keep the WS open until close.
+                    continue;
+                };
+                let ts = elapsed_ms(session_start);
+                match encode_audio_in(&frame, seq_egress, ts) {
+                    Ok(s) => {
+                        seq_egress += 1;
+                        if let Err(e) = ws.send(Message::Text(s)).await {
+                            warn!(error = ?e, %session_id, "ws send audio_in failed");
+                            return Err(e.into());
+                        }
+                    }
+                    Err(e) => {
+                        metrics.malformed_frames.fetch_add(1, Ordering::Relaxed);
+                        warn!(error = ?e, "encode audio_in failed; dropping");
+                    }
+                }
+            }
+            // Inbound WS frame from brain.
+            msg = ws.next() => {
+                let Some(msg) = msg else {
+                    info!(%session_id, "brain WS stream ended");
+                    return Ok(());
+                };
+                let msg = msg?;
+                // `Message::into_text` returns `Result<String>`; non-text
+                // frames (Binary/Ping/Pong/Close) yield `Err(_)` and are
+                // silently dropped (v1 is text-JSON only — spec §3.4).
+                if let Ok(text) = msg.into_text() {
+                    handle_brain_frame(
+                        &text, &mut last_seq_ingress, &tx_audio_out,
+                        &metrics, session_start,
+                    ).await;
+                }
+            }
+        }
+    }
+}
+
+async fn wait_for_brain_hello<T>(ws: &mut WebSocketStream<T>) -> Result<u64, TapClientError>
+where
+    T: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin,
+{
+    while let Some(msg) = ws.next().await {
+        let msg = msg?;
+        if let Ok(text) = msg.into_text() {
+            let decoded = decode_envelope(&text)?;
+            if let DecodedPayload::Hello(_) = decoded.payload {
+                return Ok(decoded.seq);
+            }
+            // Non-hello frames before ack — log + continue (drop + observe).
+            warn!("pre-hello frame from brain; ignoring");
+        }
+    }
+    Err(TapClientError::Ws(
+        tokio_tungstenite::tungstenite::Error::ConnectionClosed,
+    ))
+}
+
+async fn handle_brain_frame(
+    text: &str,
+    last_seq_ingress: &mut Option<u64>,
+    tx_audio_out: &mpsc::Sender<PcmFrame>,
+    metrics: &Arc<TapMetrics>,
+    session_start: Instant,
+) {
+    let decoded = match decode_envelope(text) {
+        Ok(d) => d,
+        Err(e) => {
+            metrics.malformed_frames.fetch_add(1, Ordering::Relaxed);
+            warn!(error = ?e, "malformed brain frame; dropping");
+            return;
+        }
+    };
+    // seq gap detection (spec §3.1: gaps = loss; log + count; don't drop).
+    if let Some(prev) = *last_seq_ingress {
+        if decoded.seq > prev + 1 {
+            let gap = decoded.seq - prev - 1;
+            metrics.seq_gaps.fetch_add(gap, Ordering::Relaxed);
+            debug!(gap, "seq gap detected");
+        }
+    }
+    *last_seq_ingress = Some(decoded.seq);
+
+    match decoded.payload {
+        DecodedPayload::AudioOut(audio) => {
+            match crate::protocol::decode_pcm(&audio.pcm, audio.samples) {
+                Ok(frame) => {
+                    if tx_audio_out.try_send(frame).is_err() {
+                        metrics.outbound_dropped.fetch_add(1, Ordering::Relaxed);
+                        trace!("outbound PCM dropped (playout ring full)");
+                    }
+                }
+                Err(e) => {
+                    metrics.malformed_frames.fetch_add(1, Ordering::Relaxed);
+                    warn!(error = ?e, "failed to decode brain audio_out PCM");
+                }
+            }
+        }
+        DecodedPayload::Bye(p) => {
+            info!(reason = %p.reason, "brain sent bye; closing");
+            // Caller's pump loop sees Ok(()) and lets TapEngine decide to retry.
+        }
+        DecodedPayload::Error(p) => {
+            warn!(code = %p.code, message = %p.message, "brain error frame");
+        }
+        DecodedPayload::Hello(_) => {
+            // Re-hello mid-session (after a reconnect from the brain's POV).
+            // Fine; we already tracked last_seq_ingress above.
+        }
+        DecodedPayload::Unknown => {
+            metrics.unknown_frames.fetch_add(1, Ordering::Relaxed);
+            warn!("unknown frame type from brain; dropping");
+        }
+        // SessionEnd is core→brain only; ignore if we receive one.
+        DecodedPayload::SessionEnd(_) | DecodedPayload::AudioIn(_) => {
+            metrics.unknown_frames.fetch_add(1, Ordering::Relaxed);
+            warn!("unexpected frame direction from brain; dropping");
+        }
+    }
+    let _ = session_start; // used for ts computation if added later
+}
+
+fn elapsed_ms(start: Instant) -> u64 {
+    start.elapsed().as_millis() as u64
+}
+
+#[cfg(test)]
+mod tests {
+    // TapClient is heavily async; its real behavior is exercised in the
+    // integration test (Task 8) against the in-process EchoServer. Unit
+    // tests here cover the pure helpers.
+
+    use super::*;
+
+    #[test]
+    fn elapsed_ms_is_monotonic_nonneg() {
+        let start = Instant::now();
+        let ms = elapsed_ms(start);
+        // First call ~0; just assert it's a valid u64.
+        assert_eq!(ms, ms); // tautology but clippy-clean
+    }
+}