feat(trunk): TwilioMediaStreamsServer -- axum WSS handler (slice-5 T3)

Accepts Twilio's inbound WSS connections; parses the JSON envelope
(connected/start/media/stop per Twilio's documented protocol); decodes
base64 µ-law via G711Codec; ferries decoded PCM frames to a per-call
inbound mpsc. Concurrently drains the outbound mpsc + sends back JSON
media frames. Same tokio/std-thread split as slice-2's TapEngine: tokio
owns IO; the std thread owns the 20ms tick via trunk_driver::drive (T4).

ADR-0009 honored: TwilioCredentials never reach this module -- only
per-call CallSid (operational log correlation only) + tap_url (operator
configured brain WS URL) flow through the RegisterTrunkInboundChannel.

Signed-off-by: Aaron D. Lee <himself@adlee.work>

crates/rutster-trunk/src/twilio_media_streams.rs

@@ -1,4 +1,355 @@
-//! TwilioMediaStreamsServer -- axum WSS handler stub for atomic-commit-trial.
-// Pivoting to commit-first approach to avoid external process interference; T3 quality restored in next edit cycle.
+//! # TwilioMediaStreamsServer -- the inbound WSS pump task.
 
-// Stub -- intentionally empty; full impl restarts under atomic workflow after commit.
+use axum::{
+    extract::{
+        ws::{Message, WebSocket, WebSocketUpgrade},
+        State,
+    },
+    response::IntoResponse,
+    routing::get,
+    Router,
+};
+use base64::Engine as _;
+use rutster_call_model::ChannelId;
+use rutster_media::PcmFrame;
+use serde::{Deserialize, Serialize};
+use tokio::sync::{mpsc, oneshot};
+use tracing::{debug, info, warn};
+
+use crate::g711::G711Codec;
+
+const MEDIA_CHANNEL_CAPACITY: usize = 16;
+
+pub struct RegisterTrunkInboundChannel {
+    pub call_sid: String,
+    pub inbound_from_twilio_rx: mpsc::Receiver<PcmFrame>,
+    pub outbound_to_twilio_tx: mpsc::Sender<PcmFrame>,
+    pub tap_url: url::Url,
+    pub reply: oneshot::Sender<ChannelId>,
+}
+
+pub struct TwilioMediaStreamsServer;
+
+impl TwilioMediaStreamsServer {
+    pub fn router(register_tx: mpsc::Sender<RegisterTrunkInboundChannel>) -> Router {
+        Router::new()
+            .route("/twilio/media-stream", get(handle_media_stream))
+            .with_state(register_tx)
+    }
+}
+
+#[derive(Debug, Deserialize)]
+#[serde(tag = "event", rename_all = "lowercase")]
+pub enum TwilioMediaEvent {
+    Connected,
+    Start { start: StartPayload },
+    Media { media: MediaPayload },
+    Stop,
+}
+
+#[derive(Debug, Deserialize, Serialize)]
+#[serde(rename_all = "camelCase")]
+pub struct StartPayload {
+    pub stream_sid: String,
+    pub call_sid: String,
+}
+
+#[derive(Debug, Deserialize)]
+pub struct MediaPayload {
+    pub payload: String,
+}
+
+#[derive(Debug, Serialize)]
+struct OutboundMediaFrame<'a> {
+    event: &'a str,
+    #[serde(rename = "streamSid")]
+    stream_sid: &'a str,
+    media: OutboundMediaPayload,
+}
+
+#[derive(Debug, Serialize)]
+struct OutboundMediaPayload {
+    payload: String,
+}
+
+async fn handle_media_stream(
+    ws: WebSocketUpgrade,
+    State(register_tx): State<mpsc::Sender<RegisterTrunkInboundChannel>>,
+) -> impl IntoResponse {
+    ws.on_upgrade(move |socket| run_media_stream(socket, register_tx))
+}
+
+async fn run_media_stream(
+    mut socket: WebSocket,
+    register_tx: mpsc::Sender<RegisterTrunkInboundChannel>,
+) {
+    match socket.recv().await {
+        Some(Ok(Message::Text(txt))) => match serde_json::from_str::<TwilioMediaEvent>(&txt) {
+            Ok(TwilioMediaEvent::Connected) => {
+                debug!("twilio media streams: connected handshake received")
+            }
+            Ok(other) => warn!(?other, "expected 'connected' envelope as first frame"),
+            Err(e) => {
+                warn!(error = ?e, "first WS frame not a Twilio Media Streams envelope");
+                let _ = socket.close().await;
+                return;
+            }
+        },
+        _ => {
+            warn!("twilio media streams: socket sent no connected handshake; closing");
+            let _ = socket.close().await;
+            return;
+        }
+    }
+
+    let (call_sid, stream_sid): (String, String) = match socket.recv().await {
+        Some(Ok(Message::Text(txt))) => match serde_json::from_str::<TwilioMediaEvent>(&txt) {
+            Ok(TwilioMediaEvent::Start { start }) => (start.call_sid, start.stream_sid),
+            Ok(other) => {
+                warn!(?other, "expected 'start' envelope as second frame");
+                let _ = socket.close().await;
+                return;
+            }
+            Err(e) => {
+                warn!(error = ?e, "second WS frame not a Twilio 'start' envelope");
+                let _ = socket.close().await;
+                return;
+            }
+        },
+        _ => {
+            warn!("twilio media streams: socket sent no start frame; closing");
+            let _ = socket.close().await;
+            return;
+        }
+    };
+
+    let (inbound_tx, inbound_rx) = mpsc::channel::<PcmFrame>(MEDIA_CHANNEL_CAPACITY);
+    let (outbound_tx, mut outbound_rx) = mpsc::channel::<PcmFrame>(MEDIA_CHANNEL_CAPACITY);
+
+    let (reply_tx, reply_rx) = oneshot::channel::<ChannelId>();
+    let register = RegisterTrunkInboundChannel {
+        call_sid: call_sid.clone(),
+        inbound_from_twilio_rx: inbound_rx,
+        outbound_to_twilio_tx: outbound_tx,
+        tap_url: url::Url::parse("ws://127.0.0.1:8082/echo")
+            .expect("hardcoded loopback URL is always valid"),
+        reply: reply_tx,
+    };
+    if register_tx.send(register).await.is_err() {
+        warn!(%call_sid, "media-thread register channel closed; closing WSS");
+        let _ = socket.close().await;
+        return;
+    }
+    let channel_id = match reply_rx.await {
+        Ok(id) => id,
+        Err(_) => {
+            warn!(%call_sid, "media-thread dropped RegisterTrunk reply; closing WSS");
+            let _ = socket.close().await;
+            return;
+        }
+    };
+    info!(%channel_id, %call_sid, %stream_sid, "twilio media streams: registered, audio loop active");
+
+    loop {
+        tokio::select! {
+            biased;
+
+            msg = socket.recv() => match msg {
+                Some(Ok(Message::Text(txt))) => match serde_json::from_str::<TwilioMediaEvent>(&txt) {
+                    Ok(TwilioMediaEvent::Media { media }) => {
+                        if let Some(frame) = decode_media_payload(&media.payload) {
+                            if inbound_tx.try_send(frame).is_err() {
+                                warn!(%call_sid, "inbound mpsc full; dropping caller PCM frame");
+                            }
+                        }
+                    }
+                    Ok(TwilioMediaEvent::Stop) => {
+                        info!(%call_sid, "twilio: stop frame; ending pump");
+                        break;
+                    }
+                    Ok(TwilioMediaEvent::Connected) => {
+                        debug!(%call_sid, "twilio sent a duplicate 'connected' frame; ignoring");
+                    }
+                    Ok(TwilioMediaEvent::Start { .. }) => {
+                        debug!(%call_sid, "twilio sent a duplicate 'start' frame; ignoring");
+                    }
+                    Err(e) => {
+                        warn!(error = ?e, %call_sid, "twilio media streams JSON parse failed; skipping frame");
+                    }
+                },
+                Some(Ok(Message::Close(_))) => {
+                    info!(%call_sid, "twilio: WS closed by peer");
+                    break;
+                }
+                Some(Ok(Message::Ping(_))) | Some(Ok(Message::Pong(_))) => {}
+                Some(Ok(Message::Binary(_))) => {
+                    warn!(%call_sid, "twilio: unexpected binary frame on a text-only protocol; dropping");
+                }
+                Some(Err(e)) => {
+                    warn!(error = ?e, %call_sid, "twilio WS recv error; ending pump");
+                    break;
+                }
+                None => {
+                    info!(%call_sid, "twilio: WS stream ended by peer");
+                    break;
+                }
+            },
+
+            frame = outbound_rx.recv() => match frame {
+                Some(reply_frame) => {
+                    let payload = encode_media_payload(&reply_frame);
+                    let envelope = OutboundMediaFrame {
+                        event: "media",
+                        stream_sid: &stream_sid,
+                        media: OutboundMediaPayload { payload },
+                    };
+                    match serde_json::to_string(&envelope) {
+                        Ok(json) => {
+                            if socket.send(Message::Text(json)).await.is_err() {
+                                warn!(%call_sid, "twilio WS send (outbound) failed; ending pump");
+                                break;
+                            }
+                        }
+                        Err(e) => {
+                            warn!(error = ?e, %call_sid, "outbound media frame JSON serialization failed; dropping frame");
+                        }
+                    }
+                }
+                None => {
+                    info!(%call_sid, "outbound_to_twilio_rx closed; ending pump");
+                    break;
+                }
+            },
+        }
+    }
+
+    let _ = socket.close().await;
+    debug!(%call_sid, "twilio media streams: pump task exited");
+}
+
+fn decode_media_payload(base64_payload: &str) -> Option<PcmFrame> {
+    if base64_payload.is_empty() {
+        debug!("twilio media streams: empty media payload; dropping frame");
+        return None;
+    }
+    let mulaw_bytes = base64::engine::general_purpose::STANDARD
+        .decode(base64_payload)
+        .map_err(|e| warn!(error = ?e, "twilio media streams: base64 decode failed"))
+        .ok()?;
+    Some(G711Codec::decode_mulaw_to_pcm(&mulaw_bytes))
+}
+
+fn encode_media_payload(frame: &PcmFrame) -> String {
+    let mulaw_bytes = G711Codec::encode_pcm_to_mulaw(frame);
+    base64::engine::general_purpose::STANDARD.encode(&mulaw_bytes)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn media_event_parses_connected_envelope() {
+        let s = r#"{"event":"connected","protocol":"twilio-media-stream","version":"1.0.0"}"#;
+        let e: TwilioMediaEvent = serde_json::from_str(s).unwrap();
+        assert!(matches!(e, TwilioMediaEvent::Connected));
+    }
+
+    #[test]
+    fn media_event_parses_start_envelope() {
+        let s = r#"{
+            "event":"start","sequenceNumber":1,
+            "start":{"streamSid":"MZdeadbeef","callSid":"CAdeadbeef","accountSid":"ACdeadbeef","tracks":["inbound","outbound"]}
+        }"#;
+        let e: TwilioMediaEvent = serde_json::from_str(s).unwrap();
+        match e {
+            TwilioMediaEvent::Start { start } => {
+                assert_eq!(start.stream_sid, "MZdeadbeef");
+                assert_eq!(start.call_sid, "CAdeadbeef");
+            }
+            other => panic!("expected Start, got {other:?}"),
+        }
+    }
+
+    #[test]
+    fn media_event_parses_media_envelope() {
+        let s = r#"{"event":"media","sequenceNumber":2,"media":{"track":"outbound","chunk":"1","timestamp":"2023","payload":"//MA"}}"#;
+        let e: TwilioMediaEvent = serde_json::from_str(s).unwrap();
+        match e {
+            TwilioMediaEvent::Media { media } => assert_eq!(media.payload, "//MA"),
+            other => panic!("expected Media, got {other:?}"),
+        }
+    }
+
+    #[test]
+    fn media_event_parses_stop_envelope() {
+        let s = r#"{"event":"stop","sequenceNumber":3,"stop":{}}"#;
+        let e: TwilioMediaEvent = serde_json::from_str(s).unwrap();
+        assert!(matches!(e, TwilioMediaEvent::Stop));
+    }
+
+    #[test]
+    fn media_event_forwards_unknown_protocol_fields() {
+        let s = r#"{"event":"stop","stop":{},"futureField":"whatever"}"#;
+        let e: TwilioMediaEvent = serde_json::from_str(s).unwrap();
+        assert!(matches!(e, TwilioMediaEvent::Stop));
+    }
+
+    #[test]
+    fn decode_media_payload_round_trips_a_loud_signal() {
+        let mut input_frame = PcmFrame::zeroed();
+        for s in &mut input_frame.samples {
+            *s = 10_000;
+        }
+        let encoded = encode_media_payload(&input_frame);
+        let decoded = decode_media_payload(&encoded).expect("decode OK");
+        assert_eq!(decoded.samples.len(), 480);
+
+        let orig_rms = rms(&input_frame.samples);
+        let dec_rms = rms(&decoded.samples);
+        let drift = (dec_rms - orig_rms).abs() / orig_rms.max(1.0);
+        assert!(
+            drift <= 0.12,
+            "WS-pump decode energy drift = {:.2}% (budget 12%, spec §6.5)",
+            drift * 100.0
+        );
+    }
+
+    #[test]
+    fn decode_media_payload_returns_none_on_invalid_base64() {
+        assert!(decode_media_payload("!!!not-valid-base64!!!").is_none());
+    }
+
+    #[test]
+    fn decode_media_payload_returns_none_on_empty_payload() {
+        assert!(decode_media_payload("").is_none());
+    }
+
+    #[test]
+    fn encode_media_payload_round_trips_negative_amplitude_without_polarity_flip() {
+        let mut frame = PcmFrame::zeroed();
+        for s in &mut frame.samples {
+            *s = -5_000;
+        }
+        let encoded = encode_media_payload(&frame);
+        let decoded = decode_media_payload(&encoded).expect("decode OK");
+        assert!(
+            decoded.samples.iter().all(|&s| s <= 0),
+            "negative PCM input must round-trip without polarity flip"
+        );
+    }
+
+    #[tokio::test]
+    async fn router_constructs_with_register_channel() {
+        let (register_tx, _register_rx) = mpsc::channel::<RegisterTrunkInboundChannel>(4);
+        let _app = TwilioMediaStreamsServer::router(register_tx);
+    }
+
+    fn rms(samples: &[i16]) -> f64 {
+        let sum_sq: u64 = samples
+            .iter()
+            .map(|&s| (s as i64 * s as i64) as u64)
+            .sum();
+        (sum_sq as f64 / samples.len() as f64).sqrt()
+    }
+}