rutster/crates/rutster-trunk/src/loop_driver.rs

//! # trunk_driver::drive -- the FOB-side trunk-leg tick function (spec §3.3)
//!
//! Parallels `crates/rutster-media/src/loop_driver.rs` minus the str0m/Opus/
//! RTP machinery. There is no RTP to decode, no Opus to encode, no str0m poll
//! loop to drain. The caller->FOB direction is a pure mpsc drain; the
//! FOB->caller direction is a mpsc push.
//!
//! # The seam (sacred -- slice-4 Task 10 pinned-blob CI gate)
//!
//! This function is a SEPARATE file in a SEPARATE crate. `loop_driver.rs`
//! in `rutster-media` stays byte-identical because the trunk leg never
//! enters that code path. The MediaThread's tick (T5) dispatches via the
//! `MediaLeg` enum:
//!
//! ```ignore
//! match leg {
//!     MediaLeg::WebRTC(s) => rutster_media::loop_driver::drive(s, now),  // unchanged slice-4
//!     MediaLeg::Trunk(s)   => rutster_trunk::loop_driver::drive(s, now),  // NEW this slice
//! }
//! ```

use std::time::{Duration, Instant};

use rutster_media::{AudioSink, AudioSource};
use tracing::{info, warn};

use crate::session::{IDLE_TIMEOUT, OUTBOUND_TICK, TrunkSession};

/// One iteration of the trunk-leg driver. Mirrors `rutster_media::loop_driver::drive`
/// minus str0m. Returns the sleep-until-next-tick budget.
///
/// # Hot-path policy (AGENTS.md "Error handling")
///
/// Never `?`-propagate. `try_recv` / `try_send` return `Err` on empty/full,
/// neither of which halts the tick -- log + move on. A dropped caller PCM
/// frame is recoverable; a crashed std thread is not.
pub fn drive<P: rutster_media::AudioPipe + Send>(
    session: &mut TrunkSession<P>,
    now: Instant,
) -> Option<Duration> {
    // === Step 1: drain inbound caller PCM from Twilio ===
    // (The WebRTC equivalent reads UDP + str0m::Input::Receive; the trunk
    //  equivalent drains a tokio mpsc the WSS pump task fills.)
    while let Ok(frame) = session.inbound_from_twilio_rx.try_recv() {
        // Push the caller's PCM through the wrapped AudioPipe stack:
        // LocalVadReflex::on_pcm_frame inspects for VAD (slice-4 §3.4 --
        // primary trigger path), then delegates inward -> Reflex::
        // on_pcm_frame -> the inner P (TapAudioPipe in production) ->
        // tx_pcm_in -> TapEngine -> brain WS as `audio_in`.
        session.pipe.on_pcm_frame(frame);
        session.last_idle_rx = now;
    }

    // === Step 2: outbound encode tick (every 20 ms) ===
    if now.duration_since(session.last_outbound_at) >= OUTBOUND_TICK {
        if let Some(reply_frame) = session.pipe.next_pcm_frame() {
            // Brain reply PCM; push to the WSS pump task for µ-law encode
            // + Twilio Media Streams JSON "media" frame emission. try_send;
            // drop + observe on full (hot-path policy).
            if session.outbound_to_twilio_tx.try_send(reply_frame).is_err() {
                warn!(
                    channel_id = %session.channel.id,
                    "outbound_to_twilio_tx full; dropping brain reply frame"
                );
            }
        }
        session.last_outbound_at = now;
    }

    // === Step 3: idle timeout (mirrors slice-1 §4.5) ===
    if now.duration_since(session.last_idle_rx) > IDLE_TIMEOUT {
        info!(
            channel_id = %session.channel.id,
            "trunk idle timeout (60s no RX); closing session"
        );
        session.channel.state = rutster_call_model::ChannelState::Closed;
        return None;
    }

    // === Step 4: sleep budget ===
    // The std thread's meta-tick is 10 ms; this duration tells the next
    // tick's worth of sleep (mirrors loop_driver::drive's contract).
    session.next_timeout = Some(now + OUTBOUND_TICK);
    Some(OUTBOUND_TICK)
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::session::tests::trunk_session_with_echo_pipe_and_mpsc;
    use rutster_media::PcmFrame;

    #[tokio::test]
    async fn tick_drains_inbound_mpsc_and_pushes_to_pipe_sink() {
        // Given: a TrunkSession with an EchoAudioPipe inner + an inbound
        // mpsc carrying one PCM frame.
        let now = Instant::now();
        let (mut session, inbound_tx, _out_rx) = trunk_session_with_echo_pipe_and_mpsc(now);

        let mut frame = PcmFrame::zeroed();
        frame.samples[0] = 4242;
        inbound_tx.send(frame).await.unwrap();

        // When: drive() runs one tick.
        let _ = drive(&mut session, now);

        // Then: the EchoAudioPipe's internal queue holds the frame (and
        // the VAD path's RMS threshold for a constant 4242 amplitude is
        // below the barge-in trigger threshold, so the frame passes
        // through unmuted). We verify by reading it back via the same
        // pipe's `next_pcm_frame` -- but `next_pcm_frame` is downstream
        // of the LocalVadReflex's muted state, so we pull the inner
        // pipe's queue contents indirectly by sending a fresh formatted
        // frame. (Slice-4 §3.4 covers the unmuted drain identity held
        // when caller energy is below VAD_RMS_THRESHOLD.)
        let out_frame = session.pipe.next_pcm_frame();
        // The EchoAudioPipe holds exactly one frame (the one we pushed).
        // If the tick route worked, we get Some back; if the inbound
        // mpsc was never drained, we get None (the echopipe is empty).
        assert!(
            out_frame.is_some(),
            "tick must drain inbound mpsc to the pipe"
        );
    }

    #[tokio::test]
    async fn tick_pulls_from_pipe_source_and_pushes_to_outbound() {
        // Given: a TrunkSession with an EchoAudioPipe inner that already
        // holds a frame (pushed via on_pcm_frame), and the outbound mpsc
        // ready to receive.
        let now = Instant::now();
        let (mut session, _in_tx, mut out_rx) = trunk_session_with_echo_pipe_and_mpsc(now);

        // Prime the inner echo pipe with a frame.
        let mut frame = PcmFrame::zeroed();
        frame.samples[0] = 9001;
        session.pipe.on_pcm_frame(frame);

        // Force the outbound tick to fire (last_outbound_at = now - 100ms).
        session.last_outbound_at = now - Duration::from_millis(100);

        // When: drive() runs one tick.
        let _ = drive(&mut session, now);

        // Then: outbound mpsc has the frame (try_recv returns it).
        let drained = out_rx.try_recv();
        assert!(
            drained.is_ok(),
            "tick must push brain reply PCM to outbound mpsc"
        );
    }

    #[tokio::test]
    async fn idle_timeout_closes_session_after_60s_silence() {
        // Given: a session whose last_idle_rx is 90s in the past.
        let now = Instant::now();
        let (mut session, _in_tx, _out_rx) = trunk_session_with_echo_pipe_and_mpsc(now);
        session.last_idle_rx = now - Duration::from_secs(90);

        // When: drive() runs one tick.
        let next = drive(&mut session, now);

        // Then: the channel is Closed and the tick returns None (halting
        // the std thread's tick loop for this leg).
        assert!(matches!(
            session.channel.state,
            rutster_call_model::ChannelState::Closed
        ));
        assert_eq!(next, None, "drive returns None on close");
    }

    #[tokio::test]
    async fn drive_advances_last_idle_rx_on_inbound_frame() {
        // Given: a TrunkSession with an inbound frame ready; last_idle_rx
        // is currently in the past.
        let now = Instant::now();
        let (mut session, inbound_tx, _out_rx) = trunk_session_with_echo_pipe_and_mpsc(now);
        session.last_idle_rx = now - Duration::from_secs(30);

        inbound_tx.send(PcmFrame::zeroed()).await.unwrap();

        // When: drive() runs one tick.
        let _ = drive(&mut session, now);

        // Then: last_idle_rx advanced to now (idle timeout didn't fire).
        assert_eq!(session.last_idle_rx, now);
        assert!(matches!(
            session.channel.state,
            rutster_call_model::ChannelState::New
        ));
    }
}