From caf4c5ba91748de3d8bc1f76adcf87ada440458c Mon Sep 17 00:00:00 2001 From: "Aaron D. Lee" Date: Sun, 5 Jul 2026 03:31:22 -0400 Subject: [PATCH] feat(trunk): TrunkSession + trunk_driver::drive -- FOB-side trunk-leg tick (slice-5 T4) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The trunk leg's tick function. 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. Caller->FOB direction is a pure mpsc drain; FOB->caller direction is a mpsc push. Slice-4's Reflex

+ LocalVadReflex stack compose identically around the trunk leg's session.pipe -- proving the FOB reflex loop is ingress-agnostic (spec §2.3 -- the architecture's load-bearing claim). TrunkSession is generic over P: AudioPipe + Send so unit tests substitute EchoAudioPipe without constructing a full TapEngine wiring harness; production uses TapAudioPipe. The seam gate holds: crates/rutster-media/src/{loop_driver.rs,rtc_session.rs} stay byte-identical because the trunk leg NEVER enters that code path. The MediaThread dispatches via the new MediaLeg enum (T5). Signed-off-by: Aaron D. Lee --- crates/rutster-trunk/src/lib.rs | 54 +++---- crates/rutster-trunk/src/loop_driver.rs | 193 ++++++++++++++++++++++++ crates/rutster-trunk/src/session.rs | 155 +++++++++++++++++++ 3 files changed, 366 insertions(+), 36 deletions(-) create mode 100644 crates/rutster-trunk/src/loop_driver.rs create mode 100644 crates/rutster-trunk/src/session.rs diff --git a/crates/rutster-trunk/src/lib.rs b/crates/rutster-trunk/src/lib.rs index 7d2a816..848fc29 100644 --- a/crates/rutster-trunk/src/lib.rs +++ b/crates/rutster-trunk/src/lib.rs @@ -8,61 +8,43 @@ //! It owns **no SIP stack**: SIP lives outside the trust boundary, in the rented transport //! (green zone — [ADR-0008](../../../docs/adr/0008-fob-and-green-zone.md)). //! -//! ## Spearhead step 5 — what's here vs what's deferred +//! ## Spearhead step 5 — what's here //! -//! The provider call-control seam ([`provider`]) lands in T2: the -//! `CallControlClient` trait, `MockCallControlClient`, and `TwilioCredentials` (with -//! redacted `Debug`). The live `TwilioCallControlClient` (T6) is feature-gated behind -//! `twilio-live`. The WSS ingress (`g711`, `twilio_media_streams`, `session`, -//! `loop_driver`) lands in the parallel FOB-side tasks (T1/T3/T4) — see the slice-5 spec. +//! FOB-side (dev-c, T1/T3/T4/T5): +//! * `g711` — G711Codec (µ-law encode/decode + 8kHz↔24kHz resampling). +//! * `twilio_media_streams` — TwilioMediaStreamsServer (axum WSS handler). +//! * `session` — TrunkSession (per-trunk-leg struct). +//! * `loop_driver` — trunk_driver::drive (per-tick function). +//! +//! Green-zone (dev-b, T2/T6): +//! * `provider/{mod,mock,twilio}` — CallControlClient trait + Mock + live +//! TwilioCallControlClient (behind `twilio-live` feature). //! //! ## Dependency direction //! -//! `rutster-trunk` → `rutster-call-model` + `rutster-media` (once the FOB-side media path -//! lands): PSTN audio is resampled to the canonical tap format and handed to the media plane +//! `rutster-trunk` → `rutster-call-model` + `rutster-media` + `rutster-tap`: +//! PSTN audio is resampled to the canonical tap format and handed to the media plane //! as a media-leg ingress, parallel to WebRTC. The provider (call-control) side is //! green-zone and depends on no FOB media types — only on `url` + `async-trait` (T2). -pub mod provider; - -// slice-5 fills in the FOB-side rented-transport ingress. Module map: -// -// * g711 -- G711Codec (µ-law encode/decode + 8kHz<->24kHz -// linear-interpolated resampling). T1. -// * mulaw_decode_table -- 256-entry compile-time-generated ITU-T G.711 -// µ-law decode table. T1. -// * mulaw_encode_table -- 65536-entry compile-time-generated ITU-T G.711 -// µ-law encode table. T1. -// -// Future modules landing in subsequent tasks (slice-5 dev-c chain): -// * twilio_media_streams -- TwilioMediaStreamsServer (axum WSS handler). T3. -// * session -- TrunkSession (per-trunk-leg session struct). T4. -// * loop_driver -- trunk_driver::drive (per-tick function). T4. -// -// Green-zone modules (slice-5 dev-b branch): -// * provider/{mod,mock,twilio} -- CallControlClient trait + Mock + live -// TwilioCallControlClient (behind `twilio-live` feature). T2/T6 -- lands -// in dev-b's branch, rebase-merges here at PR time. - pub mod g711; +pub mod loop_driver; mod mulaw_decode_table; mod mulaw_encode_table; +pub mod provider; +pub mod session; pub mod twilio_media_streams; #[cfg(test)] mod tests { - use crate::provider::MockCallControlClient; use crate::g711::G711Codec; + use crate::provider::MockCallControlClient; - /// Stub crates lock boundaries; the compile-test is the lock. Now that - /// the provider module (T2, dev-b) + FOB modules (T1, dev-c) both land, - /// this test guards the crate-level re-exports compile against the - /// combined public API surface. + /// Stub boundary smoke test, retained after the FOB modules land. Also + /// exercises the public codec + provider API surface once. #[test] fn crate_compiles() { - // Touch the provider public API (T2). let _mock = MockCallControlClient::new(); - // Touch the codec public API (T1). let frame = G711Codec::decode_mulaw_to_pcm(&[0xFF; 160]); assert_eq!(frame.samples.len(), 480); } diff --git a/crates/rutster-trunk/src/loop_driver.rs b/crates/rutster-trunk/src/loop_driver.rs new file mode 100644 index 0000000..4c4d739 --- /dev/null +++ b/crates/rutster-trunk/src/loop_driver.rs @@ -0,0 +1,193 @@ +//! # 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 tracing::{info, warn}; + +use crate::session::{TrunkSession, IDLE_TIMEOUT, OUTBOUND_TICK}; + +/// 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( + session: &mut TrunkSession

, + now: Instant, +) -> Option { + // === 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 + )); + } +} diff --git a/crates/rutster-trunk/src/session.rs b/crates/rutster-trunk/src/session.rs new file mode 100644 index 0000000..e474cc7 --- /dev/null +++ b/crates/rutster-trunk/src/session.rs @@ -0,0 +1,155 @@ +//! # TrunkSession -- the per-trunk-leg session (spec §3.3) +//! +//! Parallels slice-1's `RtcSession` minus the str0m/Opus/UDP-socket footprint. +//! The wrapped pipe stack is REUSED from slice-4 verbatim: +//! `LocalVadReflex>`, where `P: AudioPipe + Send` is the inner pipe +//! (production: `TapAudioPipe`; tests: `EchoAudioPipe`). The composition site +//! (MediaThread's `RegisterTrunk` handler -- T5) constructs the full stack. + +use std::time::{Duration, Instant}; + +use rutster_call_model::Channel; +use rutster_media::PcmFrame; +use tokio::sync::mpsc; + +/// 60 s of silence (no inbound PCM from Twilio's "media" frames) means +/// "the call leg is dead" -- the WSS pump task has likely crashed or the +/// Twilio Media Streams fork has been silently closed by the cloud. Same +/// threshold as slice-1's RtcSession idle timeout (spec §4.5). +pub const IDLE_TIMEOUT: Duration = Duration::from_secs(60); + +/// 20 ms outbound encode tick (matches slice-1's PCM frame size: 480 samples +/// @ 24 kHz = 20 ms). The trunk driver pulls a single PCM frame from the +/// pipe's `next_pcm_frame` source on each tick onset + pushes to the +/// outbound WSS mpsc. Mirrors `rutster_media::loop_driver::OUTBOUND_TICK`. +pub const OUTBOUND_TICK: Duration = Duration::from_millis(20); + +/// The per-trunk-leg session. Generic over `P: AudioPipe + Send` so tests +/// substitute a stub inner pipe (e.g. `EchoAudioPipe`) without constructing +/// the full TapEngine wiring (TapConn, WS handshake, mpsc pairs). +/// +/// # Production type instantiation +/// +/// `TrunkSession` -- MediaThread::RegisterTrunk (T5) +/// composes: +/// 1. `let (tap_pipe, tap_conn) = spawn_tap_engine(...)` (slice-2). +/// 2. `let reflex = Reflex::new(tap_pipe, advisory_rx, metrics)` (slice-4). +/// 3. `let wrapped = LocalVadReflex::new(reflex, advisory_tx)` (slice-4). +/// 4. `TrunkSession { pipe: wrapped, ... }`. +/// +/// The `(P: AudioPipe + Send)` bound mirrors the slice-4 trait constraints: +/// the std thread holds the session behind `&mut TrunkSession` exclusively, +/// but Send is required when the binary moves sessions across thread-task +/// boundaries for teardown or migration. +pub struct TrunkSession { + /// The call-channel the FOB created for this trunk leg. + pub channel: Channel, + /// The wrapped reflex stack: `LocalVadReflex>`. Constructed + /// by MediaThread::RegisterTrunk (T5), REUSED slice-4 task-6 composition. + pub pipe: rutster_media::LocalVadReflex>, + /// Caller PCM from Twilio (decoded at 24 kHz by the WSS pump task). + pub inbound_from_twilio_rx: mpsc::Receiver, + /// FOB outbound PCM for Twilio (the WSS pump encodes µ-law + sends back). + pub outbound_to_twilio_tx: mpsc::Sender, + /// Last Instant a caller PCM frame was received. Used to detect the + /// 60 s idle timeout (WSS pump gone silent -> close). + pub last_idle_rx: Instant, + /// Cached next-deadline Instant (mirrors `RtcSession::next_timeout`). + pub next_timeout: Option, + /// Last Instant we wrote an outbound PCM frame (mirrors + /// `RtcSession::last_outbound_at`). Paces the 20 ms outbound tick. + pub last_outbound_at: Instant, +} + +impl TrunkSession

{ + /// Construct a session with the provided wrapped pipe + mpsc pair. The + /// MediaThread::RegisterTrunk (T5) handler owns this construction site + /// in production; tests call this directly with stub pipes. + pub fn new( + channel: Channel, + pipe: rutster_media::LocalVadReflex>, + inbound_from_twilio_rx: mpsc::Receiver, + outbound_to_twilio_tx: mpsc::Sender, + now: Instant, + ) -> Self { + Self { + channel, + pipe, + inbound_from_twilio_rx, + outbound_to_twilio_tx, + last_idle_rx: now, + next_timeout: None, + last_outbound_at: now, + } + } + + /// Slice-4's `ChannelState::Closed` predicate (mirrors the slice-1 + /// `RtcSession::is_closed` API). Used by the std thread's tick loop to + /// evict closed sessions from the session map. + pub fn is_closed(&self) -> bool { + matches!(self.channel.state, rutster_call_model::ChannelState::Closed) + } +} + +#[cfg(test)] +pub mod tests { + use super::*; + use rutster_media::{EchoAudioPipe, Reflex, ReflexMetrics, LocalVadReflex}; + use tokio::sync::mpsc; + + /// Build a `TrunkSession` with no real brain wiring + /// (no TapEngine, no WS endpoint). The materialized wrapped pipe is the + /// same composition the FOB constructs in production: + /// `LocalVadReflex>`. The advisory mpsc pair is + /// fresh and never receives events in these tests. + pub fn trunk_session_with_echo_pipe(now: Instant) -> TrunkSession { + let (advisory_tx, advisory_rx) = mpsc::channel(16); + let metrics = ReflexMetrics::new(); + let reflex = Reflex::new(EchoAudioPipe::new(), advisory_rx, metrics); + let wrapped = LocalVadReflex::new(reflex, advisory_tx); + let (inbound_tx, inbound_rx) = mpsc::channel(16); + let (outbound_tx, outbound_rx) = mpsc::channel(16); + let channel = Channel::new_inbound(); + let mut session = TrunkSession::new(channel, wrapped, inbound_rx, outbound_tx, now); + // Expose the tx side to the test via a leak into a multi-call test. + session.last_idle_rx = now; + session + } + + /// EchoAudioPipe ownership partner of [`trunk_session_with_echo_pipe`]: + /// returns the sender + receiver ends separately for test scenarios. + pub fn trunk_session_with_echo_pipe_and_mpsc( + now: Instant, + ) -> ( + TrunkSession, + mpsc::Sender, + mpsc::Receiver, + ) { + let (advisory_tx, advisory_rx) = mpsc::channel(16); + let metrics = ReflexMetrics::new(); + let reflex = Reflex::new(EchoAudioPipe::new(), advisory_rx, metrics); + let wrapped = LocalVadReflex::new(reflex, advisory_tx); + let (inbound_tx, inbound_rx) = mpsc::channel(16); + let (outbound_tx, outbound_rx) = mpsc::channel(16); + let channel = Channel::new_inbound(); + let session = TrunkSession::new(channel, wrapped, inbound_rx, outbound_tx, now); + (session, inbound_tx, outbound_rx) + } + + #[test] + fn new_session_starts_in_new_state() { + let now = Instant::now(); + let session = trunk_session_with_echo_pipe(now); + assert!(matches!( + session.channel.state, + rutster_call_model::ChannelState::New + )); + } + + #[test] + fn is_closed_false_for_fresh_session() { + let now = Instant::now(); + let session = trunk_session_with_echo_pipe(now); + assert!(!session.is_closed()); + } +}