Cargo.lock

@@ -143,6 +143,7 @@ dependencies = [
  "async-trait",
  "axum-core",
  "axum-macros",
+ "base64",
  "bytes",
  "futures-util",
  "http",
@@ -161,8 +162,10 @@ dependencies = [
  "serde_json",
  "serde_path_to_error",
  "serde_urlencoded",
+ "sha1",
  "sync_wrapper",
  "tokio",
+ "tokio-tungstenite",
  "tower",
  "tower-layer",
  "tower-service",
@@ -1511,10 +1514,22 @@ name = "rutster-trunk"
 version = "0.0.0"
 dependencies = [
  "async-trait",
+ "axum",
+ "base64",
+ "futures-util",
  "reqwest",
+ "rutster-brain-realtime",
+ "rutster-call-model",
+ "rutster-media",
+ "rutster-tap",
+ "serde",
  "serde_json",
+ "thiserror 1.0.69",
  "tokio",
+ "tokio-tungstenite",
+ "tower",
  "tracing",
+ "tracing-subscriber",
  "url",
 ]
 

crates/rutster-trunk/Cargo.toml

@@ -8,43 +8,35 @@ repository.workspace = true
 description = "Rented carrier transport — CPaaS media-leg ingress; no first-party SIP (spearhead step 5, ADR-0007)."
 
 [dependencies]
-# async-trait: lets us declare `async fn` in the CallControlClient trait while
-# still using it as a trait object (`Box<dyn CallControlClient>`) in the route
-# handlers. Native async-fn-in-traits (stable since 1.75) does NOT support
-# `dyn` dispatch without manual desugaring — async_trait rewrites the signature
-# to `fn -> Pin<Box<dyn Future + Send>>` for us (spec §3.4).
+# FOB-side (dev-c, T1/T3/T4/T5):
+rutster-media = { path = "../rutster-media" }
+rutster-call-model = { path = "../rutster-call-model" }
+rutster-tap = { path = "../rutster-tap" }
+tokio = { workspace = true, features = ["macros", "rt-multi-thread", "sync", "time"] }
+axum = { workspace = true, features = ["ws"] }
+serde = { workspace = true, features = ["derive"] }
+serde_json = { workspace = true }
+tracing = { workspace = true }
+thiserror = { workspace = true }
+base64 = { workspace = true }
+
+# Green-zone (dev-b, T2/T6):
 async-trait = { workspace = true }
-# url: the `TwilioCredentials::webhook_base` field is a `url::Url` (the
-# operator's public base URL Twilio calls back). Parsed in `config::twilio_credentials`.
 url = { workspace = true }
-# reqwest + tracing + serde_json are OPTIONAL — only pulled in when the
-# `twilio-live` feature is enabled (the live `TwilioCallControlClient`, T6).
-# This keeps the default CI build (default-features-off) free of reqwest's
-# transitive dep tree, so per-PR `cargo deny check` stays lean + cargo's
-# resolve is fast for the common case. The maintainer's manual
-# `cargo test --features=twilio-live` + the twilio-live CI job (T10) pull them in.
-# `serde_json` parses the Calls.json REST response body (the `sid` field).
+# reqwest is OPTIONAL — only pulled in when `twilio-live` is enabled (the
+# live `TwilioCallControlClient`, T6). Keeps default CI build lean.
 reqwest = { workspace = true, optional = true }
-tracing = { workspace = true, optional = true }
-serde_json = { workspace = true, optional = true }
 
 [dev-dependencies]
-# tokio: only the dev-dep is needed for the mock's `#[tokio::test]` attribute
-# (`#[tokio::test]` expands to a `#[test]` that spins up a current-thread
-# runtime + drives the async fn to completion). The library itself is
-# runtime-agnostic: async_trait's desugared futures poll on whatever runtime
-# the caller drives them with — the mock's bodies are synchronous (lock + push,
-# no `.await`), so no tokio dependency leaks into the library's public surface.
-tokio = { workspace = true }
+tower = { workspace = true }
+futures-util = { workspace = true }
+tokio-tungstenite = { workspace = true }
+tracing-subscriber = { workspace = true }
+rutster-brain-realtime = { path = "../rutster-brain-realtime", features = ["mock"] }
 
 [features]
 default = []
-# The live `TwilioCallControlClient` (T6) is feature-gated behind `twilio-live`
-# so the routine CI gate stays feature-default-off: `MockCallControlClient` is
-# the per-PR test surface; the maintainer runs `cargo test --features=twilio-live`
-# + the e2e suite only when validating a release (against real Twilio creds).
-# Spec §1.2 + plan T6 + ADR-0009 (credentials never reach the brain).
-# `dep:reqwest` + `dep:tracing` are the optional-dep enable syntax
-# (`feature = "dep:foo"` enables `foo` without exposing it as an implicit
-# feature named `foo` — the explicit form is forward-compatible + unambiguous).
-twilio-live = ["dep:reqwest", "dep:tracing", "dep:serde_json"]
+# The live `TwilioCallControlClient` (T6) is feature-gated behind `twilio-live`.
+# `MockCallControlClient` is the per-PR test surface; the maintainer runs
+# `cargo test --features=twilio-live` manually pre-release (ADR-0009).
+twilio-live = ["dep:reqwest"]

crates/rutster-trunk/src/g711.rs

@@ -0,0 +1,266 @@
+//! # G.711 µ-law codec + 8 kHz <-> 24 kHz linear-interpolated resampling
+//!
+//! `G711Codec` is the boundary codec between Twilio Media Streams raw-audio
+//! forks (8 kHz µ-law, base64-encoded JSON envelopes) and rutster's canonical
+//! 24 kHz mono PCM `PcmFrame` (slice-1 spec §3.9 -- 480 samples / 20 ms).
+//!
+//! # Why in-core, not a `g711` crate dep
+//!
+//! µ-law is ~30 lines of table-driven code. The codec has been standard
+//! since 1972 (ITU-T G.711); it has not changed. Pulling a dependency for a
+//! constant-mapping table would be a FOB hygiene violation (ADR-0008 -- link
+//! mature OSS only when the complexity is non-trivial; this is neither). The
+//! implementation is also learner-facing per AGENTS.md "Code style (Rust)":
+//! the µ-law companding formula is a piece of telephony history worth teaching.
+//!
+//! # The hot path
+//!
+//! [`G711Codec::decode_mulaw_to_pcm`] runs inside the WSS pump task for every
+//! inbound Twilio "media" frame (one per 20 ms tick). [`G711Codec::encode_pcm_to_mulaw`]
+//! runs in the WSS pump's outbound send loop for every brain reply the std
+//! thread produces. Both call paths are pure-function over `&[u8]` -- no I/O
+//! inside, no allocation beyond the result buffer's `Vec` capacity hint.
+//!
+//! # Why 3x upsample by linear interpolation (not `rubato` / speexdsp)
+//!
+//! 24 kHz / 8 kHz = exactly 3. Each input sample becomes 3 output samples:
+//!
+//! ```text
+//!   out[3i]     = input[i]
+//!   out[3i + 1] = (2*input[i] + input[i+1]) / 3
+//!   out[3i + 2] = (input[i] + 2*input[i+1]) / 3
+//! ```
+//!
+//! These weights place the two interpolated samples at 1/3 and 2/3 of the
+//! 8 kHz sample interval -- a straight line between each pair of input
+//! samples. Linear interpolation is the cheap correct-enough answer for the
+//! spearhead MVP; the aliasing is below `LocalVadReflex`'s RMS threshold.
+//!
+//! Linear interpolation is the cheap correct-enough answer for the spearhead
+//! MVP -- the artifacts are below the audibility threshold for `LocalVadReflex`'s
+//! barge-in trigger, which only needs RMS energy above `VAD_RMS_THRESHOLD`
+//! (slice-4 spec §3.4). A production-grade resampler (`rubato` or `speexdsp`)
+//! would polish the high-frequency aliasing further, but doing so now would
+//! pull in a transitive dep we don't need for the wedge claim ("the FOB reflex
+//! loop works against real phone audio"). `rubato` lands in a post-spearhead
+//! refinement if a downstream consumer needs it (slice-5 spec §6.6 / §8.1).
+//!
+//! # Resampling round-trip drift budget
+//!
+//! The encode -> decode round-trip energy drift is bounded by µ-law's intrinsic
+//! segment quantization error (3 dB worst-case at segment transitions, well
+//! under the 12% energy-drift budget in slice-5 spec §6.5). The drift is
+//! verified by `tests::decode_then_encode_round_trips_a_loud_signal_within_12pct_energy_drift`.
+
+use rutster_media::{PcmFrame, SAMPLES_PER_FRAME};
+
+use crate::mulaw_decode_table::MULAW_TO_LINEAR;
+use crate::mulaw_encode_table::LINEAR_TO_MULAW;
+
+/// Zero-state codec. The methods are pure functions -- the codec holds no
+/// per-session state because the µ-law encode/decode tables are global
+/// `static` arrays (compile-time-generated, see `mulaw_decode_table.rs`
+/// and `mulaw_encode_table.rs`). The struct exists so callers can build
+/// router / handler signatures that explicitly thread the codec concept
+/// (matching the slice-5 spec §3.1 surface we extend in later tasks).
+pub struct G711Codec;
+
+impl G711Codec {
+    /// Decode a Twilio Media Streams "Media" frame payload (already
+    /// base64-decoded bytes of µ-law samples at 8 kHz) into a 24 kHz
+    /// `PcmFrame` (slice-1 canonical format). 3x linear upsample.
+    ///
+    /// # Frame size contract
+    ///
+    /// Twilio Media Streams delivers 160 µ-law bytes per 20 ms tick
+    /// (8000 samples/sec * 0.020 sec = 160). 3x upsampled, that becomes
+    /// 480 samples -- exactly one `PcmFrame` of slice-1 spec §3.9.
+    ///
+    /// # Hot-path policy
+    ///
+    /// A malformed frame (wrong byte count) does NOT crash the WSS pump
+    /// task: `debug_assert!` surfaces protocol drift in test builds; in
+    /// release the function returns a zeroed `PcmFrame` and lets the
+    /// caller drop + observe (spec §3.1).
+    pub fn decode_mulaw_to_pcm(mulaw: &[u8]) -> PcmFrame {
+        // `debug_assert!` catches protocol drift in test builds; release
+        // builds take the safe fallback so the WSS pump task never crashes
+        // on a malformed envelope (hot-path policy, AGENTS.md).
+        debug_assert_eq!(
+            mulaw.len(),
+            160,
+            "expected 160 µ-law bytes per 20ms frame (got {})",
+            mulaw.len()
+        );
+
+        // Malformed input is dropped on the hot path: return silence so the
+        // loop tick continues safely. A correct Twilio Media Streams frame
+        // always carries exactly 160 µ-law bytes per 20 ms tick.
+        if mulaw.len() != 160 {
+            return PcmFrame::zeroed();
+        }
+
+        // Decode the 160 µ-law bytes into a fixed-size 8 kHz linear buffer.
+        // 160 * sizeof(i16) = 320 bytes -- small enough to live on the stack.
+        let mut linear_8k = [0i16; 160];
+        for (i, &byte) in mulaw.iter().enumerate() {
+            linear_8k[i] = MULAW_TO_LINEAR[byte as usize];
+        }
+
+        let mut samples = [0i16; SAMPLES_PER_FRAME];
+        for i in 0..160 {
+            let current = linear_8k[i];
+            // For the final 8 kHz sample there is no next sample to
+            // interpolate toward, so hold the last value -- no extrapolation
+            // past the frame boundary.
+            let next = if i < 159 { linear_8k[i + 1] } else { current };
+
+            let base = 3 * i;
+            samples[base] = current;
+            samples[base + 1] = ((2 * current as i32 + next as i32) / 3) as i16;
+            samples[base + 2] = ((current as i32 + 2 * next as i32) / 3) as i16;
+        }
+
+        PcmFrame { samples }
+    }
+
+    /// Encode a 24 kHz `PcmFrame` into 8 kHz µ-law bytes for Twilio Media
+    /// Streams. Inverse of [`decode_mulaw_to_pcm`]: 3x downsample by
+    /// decimation (take every 3rd sample) plus a `LINEAR_TO_MULAW` table
+    /// lookup. The result is a 160-byte `Vec<u8>`, one per 20 ms tick.
+    ///
+    /// # Why decimation (not averaging)
+    ///
+    /// Decimation simply drops the samples at positions 3i+1 and 3i+2. The
+    /// aliasing energy this introduces is below `LocalVadReflex`'s RMS
+    /// threshold (slice-4 spec §3.4); the brain never cares about
+    /// frequencies above 4 kHz (its speech model is band-limited below
+    /// 8 kHz Nyquist). `rubato` would deliver better quality at the cost
+    /// of a transitive dep + slower hot path -- not warranted for the
+    /// spearhead MVP (spec §6.6 + §8.1).
+    pub fn encode_pcm_to_mulaw(frame: &PcmFrame) -> Vec<u8> {
+        let mut mulaw = Vec::with_capacity(160);
+        for i in 0..160 {
+            // `i16 as u16` reinterprets the bit pattern -- i16 -1 -> u16 65535
+            // -> table index 65535. This is the correct index for
+            // LINEAR_TO_MULAW, which is laid out `linear_to_mulaw(i as i16)` for
+            // all `i in 0..65536` -- the bit-pattern identity holds.
+            let sample = frame.samples[3 * i];
+            mulaw.push(LINEAR_TO_MULAW[(sample as u16) as usize]);
+        }
+        mulaw
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn decode_160_byte_frame_yields_480_samples() {
+        // 160 µ-law bytes is one 20 ms Twilio Media Streams tick; decoding
+        // must yield exactly the slice-1 canonical 480-sample PcmFrame.
+        // (Input is `0xFF` = ITU-T G.711 silence per spec; decode is the
+        // `mulaw_to_linear(0xFF)` value, repeated 480 times after upsample.)
+        let mulaw = vec![0xFFu8; 160];
+        let frame = G711Codec::decode_mulaw_to_pcm(&mulaw);
+        assert_eq!(frame.samples.len(), SAMPLES_PER_FRAME);
+    }
+
+    #[test]
+    fn decode_silence_yields_all_zero_pcm() {
+        // ITU-T G.711 µ-law silence is the byte 0xFF. The reference decoder's
+        // BIAS add + subtract on the positive branch yields exactly 0 PCM --
+        // this is the "mid-tread zero" property that keeps silences silent.
+        let mulaw = vec![0xFFu8; 160];
+        let frame = G711Codec::decode_mulaw_to_pcm(&mulaw);
+        assert!(
+            frame.samples.iter().all(|&s| s == 0),
+            "decoded µ-law silence (0xFF repeated) must be all-zero PCM; got non-zero"
+        );
+    }
+
+    #[test]
+    fn encode_pcm_to_mulaw_emits_160_bytes_per_frame() {
+        let frame = PcmFrame::zeroed();
+        let mulaw = G711Codec::encode_pcm_to_mulaw(&frame);
+        assert_eq!(mulaw.len(), 160);
+    }
+
+    #[test]
+    fn encode_silence_emits_mu_law_silence_0xff() {
+        // Encoding a zero PCM frame must yield 160 bytes of 0xFF (the µ-law
+        // silence encoding) -- inverse of `decode_silence_yields_all_zero_pcm`.
+        let frame = PcmFrame::zeroed();
+        let mulaw = G711Codec::encode_pcm_to_mulaw(&frame);
+        assert!(
+            mulaw.iter().all(|&b| b == 0xFF),
+            "encoding zero PCM must yield 0xFF µ-law silence; got non-0xFF"
+        );
+    }
+
+    #[test]
+    fn decode_then_encode_round_trips_a_loud_signal_within_12pct_energy_drift() {
+        // The wedge claim: a real PSTN caller's voice survives the
+        // µ-law encode -> And the decode round trip with RMS energy preserved
+        // within the spec's 12% budget (slice-5 spec §6.5 -- the barge-in
+        // trigger only needs RMS energy above VAD_RMS_THRESHOLD; µ-law's
+        // quantization error is well below that floor).
+        //
+        // A constant-amplitude loud signal (10_000 amplitude i16, ~30% of
+        // full scale) is the harshest single-tone test case -- the µ-law
+        // quantization step is most visible at flat amplitudes (segment
+        // transitions are zero, so no dynamic range advantage accrues).
+        let mut input_frame = PcmFrame::zeroed();
+        for s in &mut input_frame.samples {
+            *s = 10_000;
+        }
+
+        let mulaw = G711Codec::encode_pcm_to_mulaw(&input_frame);
+        assert_eq!(mulaw.len(), 160, "encoded frame must be 160 µ-law bytes");
+
+        let decoded = G711Codec::decode_mulaw_to_pcm(&mulaw);
+
+        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,
+            "µ-law round-trip energy drift {:.2}% > 12% budget (spec §6.5); \
+             orig_rms={}, dec_rms={}",
+            drift * 100.0,
+            orig_rms,
+            dec_rms
+        );
+    }
+
+    #[test]
+    fn encode_then_decode_preserves_polarity() {
+        // A loud negative sample (silent cues, DTMF, or echo-cancel tails)
+        // must round-trip without sign flip. Verifies the XOR mask sign
+        // convention (positive -> MSB 1; negative -> MSB 0) is consistent
+        // across encode + decode.
+        let mut input_frame = PcmFrame::zeroed();
+        for s in &mut input_frame.samples {
+            *s = -10_000;
+        }
+        let mulaw = G711Codec::encode_pcm_to_mulaw(&input_frame);
+        let decoded = G711Codec::decode_mulaw_to_pcm(&mulaw);
+
+        // All decoded samples must be negative (no polarity flip).
+        assert!(
+            decoded.samples.iter().all(|&s| s <= 0),
+            "negative-amplitude input must decode to non-positive samples; \
+             observed a positive sample (sign convention bug)"
+        );
+    }
+
+    /// RMS energy over a sample buffer -- the same shape `LocalVadReflex`
+    /// uses for its barge-in trigger (slice-4 spec §3.4). Local helper here
+    /// to keep this test module self-contained + the assertion readable.
+    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()
+    }
+}

crates/rutster-trunk/src/lib.rs

@@ -8,33 +8,44 @@
 //! 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 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::g711::G711Codec;
     use crate::provider::MockCallControlClient;
 
-    /// Stub crates lock boundaries; the compile-test is the lock. Now that the
-    /// provider module is populated (T2), this test also guards the crate-level
-    /// re-exports compile against the 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 public API so the test exercises the re-export wiring.
         let _mock = MockCallControlClient::new();
+        let frame = G711Codec::decode_mulaw_to_pcm(&[0xFF; 160]);
+        assert_eq!(frame.samples.len(), 480);
     }
 }

crates/rutster-trunk/src/loop_driver.rs

@@ -0,0 +1,192 @@
+//! # 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
+        ));
+    }
+}

crates/rutster-trunk/src/mulaw_decode_table.rs

@@ -0,0 +1,101 @@
+//! 8-bit µ-law to 16-bit linear decode table -- ITU-T G.711 (1972, reaffirmed 2000).
+//!
+//! The table is computed at compile time by [`mulaw_to_linear`], a `const fn`
+//! that encodes the standard piecewise-linear decoding formula. The 256-entry
+//! table fits in L1; the table lookup is the decode fast path -- no branchy
+//! segment arithmetic at decode call time.
+//!
+//! # The µ-law decode formula (ITU-T G.711 reference, teachable-moment)
+//!
+//! µ-law bytes are stored bit-inverted on the wire (the encoder finishes with
+//! a bitwise XOR mask, see `mulaw_encode_table.rs`). The decoder's first move
+//! is therefore `let toggled = !u;` -- undo the encoder's terminal inversion
+//! and recover the raw exponent/mantissa/sign bits.
+//!
+//! After toggling:
+//!   * bit 7 (MSB) = 1  --> original PCM was NEGATIVE.
+//!   * bits 6..4        --> 3-bit segment exponent (0..7); larger = louder.
+//!   * bits 3..0        --> 4-bit mantissa (offset within segment).
+//!
+//! The biased magnitude is reconstructed as
+//!   `t = ((mantissa << 3) + BIAS) << exponent`  (BIAS = 0x84 = 132),
+//! then the reference decoder applies a final sign-aware adjustment:
+//!   * positive --> return `t - BIAS`
+//!   * negative --> return `BIAS - t`
+//!
+//! This BIAS add/subtract on both sides of the encode/decode is the companding
+//! "compression mid-tread" property: zero PCM in, zero PCM out (silent input
+//! encodes to 0xFF, decodes back to 0). The asymmetry `BIAS - t` for the
+//! negative branch is what gives µ-law its slightly larger dynamic range on
+//! the negative side (one extra notch below zero that the positive side lacks).
+//!
+//! # Why NOT a `g711` crate dep
+//!
+//! The codec has been standard since 1972 and is ~30 lines of table-driven
+//! code. Pulling a dependency for a constant-mapping table violates the FOB
+//! hygiene rule (ADR-0008 -- link mature OSS only when the complexity is
+//! non-trivial; this is neither). It's also learner-facing: this file
+//! TEACHES the telephony history, which earns the project's "learns Rust
+//! from this codebase" goal a real win.
+
+/// Decode bias per ITU-T G.711 §2.2 (the Sun Microsystems reference impl uses
+/// bias = 0x84 = 132 for both decode and encode; the magic number traces
+/// directly to the µ-law segment boundary at `BIAS << exponent`, which keeps
+/// segment 0's t-value range `[BIAS, BIAS + 0x70)` exactly proportional to
+/// the µ-law quantization step).
+const BIAS: i32 = 0x84;
+
+/// Convert one 8-bit µ-law byte to a 16-bit linear PCM sample.
+///
+/// Generated at compile time so the [`MULAW_TO_LINEAR`] table is a `static`
+/// array -- no runtime cost beyond the L1 cache for the 256-byte lookup.
+///
+/// # Algorithm
+///
+/// Mirrors the ITU-T G.711 reference decoder (Sun Microsystems variant,
+/// public domain). See the module-level docs for the formula walkthrough.
+const fn mulaw_to_linear(u: u8) -> i16 {
+    // Undo the encoder's terminal bitwise XOR mask (0xFF for positive,
+    // 0x7F for negative). NOT is wholesale inversion -- recovers the raw
+    // sign | exponent | mantissa bits the encoder assembled pre-XOR.
+    let toggled = !u;
+
+    // After NOT: MSB = 1 iff the original PCM sample was NEGATIVE.
+    let is_negative = (toggled & 0x80) != 0;
+
+    let exponent = ((toggled >> 4) & 0x07) as i32;
+    let mantissa = (toggled & 0x0F) as i32;
+
+    // Reconstruct the biased magnitude: `((mant << 3) + BIAS) << exp`.
+    // The `<< 3` aligns the 4-bit mantissa to the segment's 8-sample step;
+    // the `+ BIAS` shifts every segment's range up by BIAS so the lowest
+    // segment (exp=0) covers `[BIAS, BIAS + 0x70)` (the µ-law near-zero band).
+    let t = ((mantissa << 3) + BIAS) << exponent;
+
+    // Sign-aware final adjustment -- the reference decoder does NOT just
+    // subtract the bias from `t` symmetrically: the negative branch inverts
+    // the subtraction (`BIAS - t`), preserving the µ-law "mid-tread" zero.
+    // With this, an input of zero PCM decodes back through `mulaw_to_linear`
+    // (the encoder maps 0 -> 0xFF) and yields zero, not BIAS.
+    if is_negative {
+        (BIAS - t) as i16
+    } else {
+        (t - BIAS) as i16
+    }
+}
+
+/// 256-entry µ-law decode table (one slot per possible 8-bit byte).
+///
+/// `#[rustfmt::skip]` keeps rustfmt from reflowing the const-eval initializer
+/// (a `while` loop initializing 256 entries) -- the canonical idiom for
+/// compile-time table generation in stable Rust (no `const_for` needed).
+#[rustfmt::skip]
+pub static MULAW_TO_LINEAR: [i16; 256] = {
+    let mut t = [0i16; 256];
+    let mut i = 0;
+    while i < 256 {
+        t[i] = mulaw_to_linear(i as u8);
+        i += 1;
+    }
+    t
+};

crates/rutster-trunk/src/mulaw_encode_table.rs

@@ -0,0 +1,113 @@
+//! 16-bit linear to 8-bit µ-law encode table -- ITU-T G.711 (1972, reaffirmed 2000).
+//!
+//! The 65536-entry / 64 KB table is computed at compile time by
+//! [`linear_to_mulaw`], a `const fn` encoding the standard piecewise-linear
+//! companding formula. The encode hot path is a single indexed lookup --
+//! branchless on the 20 ms tick.
+//!
+//! # The µ-law encode formula (ITU-T G.711 reference, teachable-moment)
+//!
+//! µ-law compresses 16-bit linear PCM to 8-bit logarithmic PCM via piecewise
+//! companding -- 8 segments, each twice the quantization step of the prior.
+//! The fundamental transform tracks a biased magnitude:
+//!
+//!   * For `s >= 0` (non-negative):  `biased = s + BIAS`
+//!   * For `s < 0`  (negative):      `biased = BIAS - s`  (= BIAS + magnitude)
+//!
+//! where `BIAS = 0x84 = 132`. The bias has a subtle reason: the lowest
+//! segment (the "near-zero" band, encoded as mantissa range [0..15]) starts
+//! at `BIAS`, which guarantees segment 0 has a symmetric, centered range
+//! around the zero PCM point. Without BIAS, the 8-segment decoder would
+//! produce an asymmetric "near-zero" band that biases silences toward +0.
+//!
+//! The biased magnitude is clipped to `0x7FFF` (32767) to bound the segment
+//! search -- this matches the upper end of segment 7's range (`seg_uend[7] = 0x7FFF`).
+//!
+//! The segment exponent is then `floor(log2(biased)) - 7`, clamped to `[0, 7]`.
+//! The mantissa is the top 4 bits of `biased >> (exponent + 3)` -- i.e. the
+//! high-nibble of the segment-local position.
+//!
+//! Finally the encoder assembles `uval = (exponent << 4) | mantissa` (7 bits,
+//! MSB unset) and applies a terminal XOR mask that inverts 7 (negative) or
+//! 8 (positive) bits. The mask trick is what makes µ-law "self-clocking" on
+//! the wire: every consecutive byte has at least one transition (no 0x00
+//! runs dominate silence on T1/E1 trunks historically).
+//!
+//! # Why NOT a `g711` crate dep
+//!
+//! Same argument as `mulaw_decode_table.rs` -- standard, ~30 lines, no
+//! complexity a maintained dep would solve; learner-facing.
+
+/// Encode bias per ITU-T G.711 §2.2 (matches `mulaw_decode_table::BIAS`).
+const BIAS: i32 = 0x84;
+
+/// Convert one 16-bit linear PCM sample to an 8-bit µ-law byte.
+///
+/// `const fn` so the [`LINEAR_TO_MULAW`] table is built at compile time and
+/// stored as a 64 KB `static` -- the encode hot path is one indexed lookup,
+/// no branchy segment search at call time.
+const fn linear_to_mulaw(s: i16) -> u8 {
+    let is_negative = s < 0;
+
+    // Bias the magnitude: positive samples get `s + BIAS`; negatives get
+    // `BIAS - s` = `BIAS + |s|`. Both branches yield `biased >= BIAS` (>= 132),
+    // so the segment search below never sees the degenerate biased = 0 case.
+    let biased: i32 = if is_negative {
+        BIAS - (s as i32)
+    } else {
+        (s as i32) + BIAS
+    };
+
+    // Clip to 0x7FFF (32767) per the Sun reference impl. After clipping the
+    // largest biased value lives in segment 7's range (`seg_uend[7] = 0x7FFF`),
+    // bounding the segment search to [0, 7].
+    let biased: i32 = if biased > 0x7FFF { 0x7FFF } else { biased };
+
+    // Segment exponent = floor(log2(biased)) - 7, clamped to [0, 7].
+    //
+    // `leading_zeros` on a positive `u32` returns `31 - floor(log2(x))` for
+    // x > 0; we cast `biased` (provably positive -- no sign-bit ambiguity)
+    // to `u32` first to avoid clippy::cast_sign_loss on the i32 path.
+    let log2_biased = 31 - (biased as u32).leading_zeros() as i32;
+    // `i32::max` / `i32::min` aren't const fn stable as of Rust 1.85; the
+    // equivalent clamp via if/else is the canonical const-eval workaround
+    // (and unavoidable here -- no const trait-method dispatch available
+    // yet without unstable features).
+    let exponent = if log2_biased < 7 {
+        0
+    } else if log2_biased > 14 {
+        7
+    } else {
+        log2_biased - 7
+    };
+
+    // Mantissa = top 4 bits of the segment-local position. Shifting right by
+    // `exponent + 3` aligns the segment's lowest 3 bits to the bottom of the
+    // u32 (discarded), then the next 4 bits are the mantissa.
+    let mantissa = ((biased >> (exponent + 3)) & 0x0F) as u8;
+    let uval = ((exponent as u8) << 4) | mantissa;
+
+    // Terminal XOR mask: 0xFF for positive (inverts all 8 bits -- MSB
+    // becomes 1), 0x7F for negative (inverts bits 0..6, preserves MSB as 0).
+    // This is the µ-law wire convention: stored byte MSB = 1 iff the original
+    // PCM sample was non-negative (the encoder's guaranteed-transition trick
+    // keeps T1/E1 line codes alive on silence runs).
+    let mask: u8 = if is_negative { 0x7F } else { 0xFF };
+    uval ^ mask
+}
+
+/// 65536-entry µ-law encode table (one slot per possible 16-bit PCM value).
+///
+/// The 64 KB binary footprint is trivial on modern hardware; the L1 trade-off
+/// favors hot-path speed (one indexed lookup vs ~8 iterations of a segment
+/// search). The const-eval initializer runs at compile time -- no runtime cost.
+#[rustfmt::skip]
+pub static LINEAR_TO_MULAW: [u8; 65536] = {
+    let mut t = [0u8; 65536];
+    let mut i = 0;
+    while i < 65536 {
+        t[i] = linear_to_mulaw(i as i16);
+        i += 1;
+    }
+    t
+};

crates/rutster-trunk/src/session.rs

@@ -0,0 +1,152 @@
+//! # 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<Reflex<P>>`, 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<TapAudioPipe>` -- 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<P: rutster_media::AudioPipe + Send> {
+    /// The call-channel the FOB created for this trunk leg.
+    pub channel: Channel,
+    /// The wrapped reflex stack: `LocalVadReflex<Reflex<P>>`. Constructed
+    /// by MediaThread::RegisterTrunk (T5), REUSED slice-4 task-6 composition.
+    pub pipe: rutster_media::LocalVadReflex<rutster_media::Reflex<P>>,
+    /// Caller PCM from Twilio (decoded at 24 kHz by the WSS pump task).
+    pub inbound_from_twilio_rx: mpsc::Receiver<PcmFrame>,
+    /// FOB outbound PCM for Twilio (the WSS pump encodes µ-law + sends back).
+    pub outbound_to_twilio_tx: mpsc::Sender<PcmFrame>,
+    /// 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<Instant>,
+    /// 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<P: rutster_media::AudioPipe + Send> TrunkSession<P> {
+    /// 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<rutster_media::Reflex<P>>,
+        inbound_from_twilio_rx: mpsc::Receiver<PcmFrame>,
+        outbound_to_twilio_tx: mpsc::Sender<PcmFrame>,
+        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, LocalVadReflex, Reflex, ReflexMetrics};
+    use tokio::sync::mpsc;
+
+    /// Build a `TrunkSession<EchoAudioPipe>` with no real brain wiring
+    /// (no TapEngine, no WS endpoint). The materialized wrapped pipe is the
+    /// same composition the FOB constructs in production:
+    /// `LocalVadReflex<Reflex<EchoAudioPipe>>`. The advisory mpsc pair is
+    /// fresh and never receives events in these tests.
+    pub fn trunk_session_with_echo_pipe(now: Instant) -> TrunkSession<EchoAudioPipe> {
+        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();
+        TrunkSession::new(channel, wrapped, inbound_rx, outbound_tx, now)
+    }
+
+    /// 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<EchoAudioPipe>,
+        mpsc::Sender<PcmFrame>,
+        mpsc::Receiver<PcmFrame>,
+    ) {
+        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());
+    }
+}

crates/rutster-trunk/src/twilio_media_streams.rs

@@ -0,0 +1,352 @@
+//! # TwilioMediaStreamsServer -- the inbound WSS pump task.
+
+use axum::{
+    Router,
+    extract::{
+        State,
+        ws::{Message, WebSocket, WebSocketUpgrade},
+    },
+    response::IntoResponse,
+    routing::get,
+};
+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()
+    }
+}

crates/rutster-trunk/tests/reflex_on_trunk.rs

@@ -0,0 +1,79 @@
+// crates/rutster-trunk/tests/reflex_on_trunk.rs
+//
+// T7 — Reflex-on-trunk-leg verification test (slice-5 spec §7).
+//
+// Proves that slice-4's `Reflex<TapAudioPipe>` + `LocalVadReflex` decorate the
+// trunk leg's `TapAudioPipe` identically to a WebRTC leg. A PSTN caller's loud
+// PCM triggers the same local-VAD state machine and the same `barge-in` kill
+// that a WebRTC caller's audio does.
+//
+// This is a *unit* integration test: it constructs the wrapped pipe stack
+// directly, without the binary's MediaThread or a real WSS brain. The heavier
+// end-to-end WSS sim lives in `sim_integ.rs` (T8).
+
+use std::sync::atomic::Ordering;
+
+use rutster_media::{
+    AudioSink, AudioSource, LocalVadReflex, PcmFrame, Reflex, ReflexMetrics, VAD_DEBOUNCE_FRAMES,
+};
+use rutster_tap::TapAudioPipe;
+use tokio::sync::mpsc;
+
+/// Build a loud 24 kHz PCM frame whose RMS energy is well above
+/// `VAD_RMS_THRESHOLD`. The mock caller uses a constant amplitude of 1000,
+/// the same value slice-4's reflex unit tests use.
+fn loud_frame() -> PcmFrame {
+    let mut frame = PcmFrame::zeroed();
+    for s in frame.samples.iter_mut() {
+        *s = 1000;
+    }
+    frame
+}
+
+#[tokio::test]
+async fn local_vad_on_trunk_pipe_kills_playout_and_resumes_on_fresh_brain_audio() {
+    // Given: a TapAudioPipe + the same Reflex<LocalVadReflex> composition the
+    // FOB builds for every leg (WebRTC or trunk).
+    let (tx_pcm_in, rx_pcm_in) = mpsc::channel(32);
+    let (tx_audio_out, rx_audio_out) = mpsc::channel(32);
+    let _rx_pcm_in = rx_pcm_in; // brain side; not used in this unit test
+    let tap_metrics = rutster_tap::TapMetrics::new();
+    let tap_pipe = TapAudioPipe::new(tx_pcm_in, rx_audio_out, tap_metrics);
+
+    let (advisory_tx, advisory_rx) = mpsc::channel(16);
+    let metrics = ReflexMetrics::new();
+    let reflex = Reflex::new(tap_pipe, advisory_rx, metrics.clone());
+    let mut wrapped_pipe = LocalVadReflex::new(reflex, advisory_tx);
+
+    // Pre-load one brain audio_out frame so the playout ring is non-empty.
+    tx_audio_out.send(PcmFrame::zeroed()).await.unwrap();
+
+    // When: the PSTN caller speaks for `VAD_DEBOUNCE_FRAMES` consecutive ticks.
+    let loud = loud_frame();
+    for _ in 0..VAD_DEBOUNCE_FRAMES {
+        wrapped_pipe.on_pcm_frame(loud.clone());
+    }
+
+    // Then: the third `next_pcm_frame` drains the local-VAD advisory, mutes the
+    // pipe, flushes the ring, and returns `None` (playout is killed).
+    let killed = wrapped_pipe.next_pcm_frame();
+    assert!(
+        killed.is_none(),
+        "local VAD must barge-in and suppress playout on the trunk leg"
+    );
+    assert_eq!(
+        metrics.barge_in_count.load(Ordering::Relaxed),
+        1,
+        "barge-in must fire exactly once for the first loud utterance"
+    );
+
+    // And when: a fresh brain reply arrives after the barge.
+    tx_audio_out.send(PcmFrame::zeroed()).await.unwrap();
+
+    // Then: the Reflex un-mutes and returns the fresh frame.
+    let resumed = wrapped_pipe.next_pcm_frame();
+    assert!(
+        resumed.is_some(),
+        "first fresh audio_out post-barge must resume trunk-leg playout"
+    );
+}

crates/rutster-trunk/tests/sim_integ.rs

@@ -0,0 +1,348 @@
+// crates/rutster-trunk/tests/sim_integ.rs
+//
+// T8 — PSTN sim end-to-end integration test (slice-5 spec §7).
+//
+// Drives a synthetic PSTN caller through the FOB reflex loop end-to-end:
+//   loud PCM -> local VAD trips -> barge kills playout -> brain replies
+//   -> un-mute -> caller hangup -> session closes.
+//
+// The test is in `rutster-trunk` so it can construct `TrunkSession` directly.
+// It cannot use the binary crate's `spawn_tap_engine` / `MediaThread` (circular
+// dev-dependency), so it builds a minimal test-only tap engine task that calls
+// `rutster_tap::tap_client::run_tap_client` against the same BrainShim surface
+// slice-4's `barge_in_integration.rs` uses.
+
+use std::sync::Arc;
+use std::sync::atomic::{AtomicUsize, Ordering};
+use std::time::{Duration, Instant};
+
+use futures_util::{SinkExt, StreamExt};
+use rutster_brain_realtime::mock::MockRealtimeBrain;
+use rutster_brain_realtime::openai_client::run_openai_pump;
+use rutster_call_model::Channel;
+use rutster_media::{LocalVadReflex, PcmFrame, Reflex, ReflexMetrics};
+
+use rutster_tap::{
+    DecodedPayload, FunctionCallEvent, FunctionCallOutputEvent, TapAudioPipe, TapMetrics,
+    decode_envelope, encode_hello, tap_client::run_tap_client,
+};
+use rutster_trunk::loop_driver::drive;
+use rutster_trunk::session::TrunkSession;
+use tokio::net::TcpListener;
+use tokio::sync::{mpsc, oneshot};
+use tokio_tungstenite::tungstenite::Message;
+use tokio_tungstenite::tungstenite::client::IntoClientRequest;
+use tracing::info;
+use url::Url;
+
+// === Brain-shim helpers (mirrored from slice-4 barge_in_integration.rs) ===
+// The brain process's accept loop is inlined so the test exercises the real
+// OpenAI-client pump (`run_openai_pump`) against the mock brain without
+// spawning a subprocess or depending on private helpers in another test.
+
+/// Handle returned by `start_brain_shim`. Drop to tear down.
+struct BrainShim {
+    addr: std::net::SocketAddr,
+    shutdown: Option<oneshot::Sender<()>>,
+    join: tokio::task::JoinHandle<()>,
+}
+
+impl Drop for BrainShim {
+    fn drop(&mut self) {
+        if let Some(tx) = self.shutdown.take() {
+            let _ = tx.send(());
+        }
+        self.join.abort();
+    }
+}
+
+/// Start an in-process brain-process-equivalent WS server on an ephemeral port.
+async fn start_brain_shim(mock_url: String) -> BrainShim {
+    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+    let addr = listener.local_addr().unwrap();
+    let (shutdown_tx, shutdown_rx) = oneshot::channel::<()>();
+    let join = tokio::spawn(async move {
+        brain_accept_loop(listener, shutdown_rx, mock_url).await;
+    });
+    BrainShim {
+        addr,
+        shutdown: Some(shutdown_tx),
+        join,
+    }
+}
+
+/// Accept loop: spawns a per-connection task for each tap WS dial.
+async fn brain_accept_loop(
+    listener: TcpListener,
+    mut shutdown: oneshot::Receiver<()>,
+    mock_url: String,
+) {
+    loop {
+        tokio::select! {
+            biased;
+            _ = &mut shutdown => {
+                info!("brain_shim accept loop shutting down");
+                return;
+            }
+            res = listener.accept() => {
+                let Ok((stream, peer)) = res else { continue };
+                let url = mock_url.clone();
+                tokio::spawn(async move {
+                    if let Err(e) = handle_tap_connection(stream, peer, &url).await {
+                        info!(%peer, error = ?e, "brain_shim connection ended");
+                    }
+                });
+            }
+        }
+    }
+}
+
+/// Handle one tap WS connection: handshake, split sink+stream, dial the mock
+/// OpenAI side, and run `run_openai_pump`.
+async fn handle_tap_connection(
+    stream: tokio::net::TcpStream,
+    peer: std::net::SocketAddr,
+    openai_url: &str,
+) -> Result<(), Box<dyn std::error::Error>> {
+    let mut tap_ws = tokio_tungstenite::accept_async(stream).await?;
+    info!(%peer, "brain_shim tap WS connection accepted");
+
+    let hello_in = tap_ws
+        .next()
+        .await
+        .ok_or("tap connection closed before hello")??;
+    let hello_text = hello_in.into_text().map_err(|_| "hello not text")?;
+    let decoded = decode_envelope(&hello_text)?;
+    let session_id = match decoded.payload {
+        DecodedPayload::Hello(p) => p.session_id,
+        _ => return Err("first tap frame not hello".into()),
+    };
+    info!(%peer, %session_id, "brain_shim tap hello received");
+    let ack = encode_hello(&session_id, 0, 0)?;
+    tap_ws.send(Message::Text(ack)).await?;
+
+    let (tap_via, pump_tap_rx) = mpsc::channel::<String>(64);
+    let (pump_tap_tx, mut tap_out_rx) = mpsc::channel::<String>(64);
+    let (mut tap_sink, mut tap_stream) = tap_ws.split();
+
+    let in_fwd = tokio::spawn(async move {
+        while let Some(msg_res) = tap_stream.next().await {
+            if let Ok(m) = msg_res {
+                if let Ok(text) = m.into_text() {
+                    if tap_via.send(text).await.is_err() {
+                        break;
+                    }
+                }
+            }
+        }
+    });
+    let out_fwd = tokio::spawn(async move {
+        while let Some(text) = tap_out_rx.recv().await {
+            if tap_sink.send(Message::Text(text)).await.is_err() {
+                break;
+            }
+        }
+    });
+
+    let request = openai_url.into_client_request()?;
+    let (openai_ws, _resp) = tokio_tungstenite::connect_async(request).await?;
+    info!(%peer, %openai_url, "brain_shim OpenAI side connected");
+
+    let pump_result =
+        run_openai_pump(openai_ws, pump_tap_rx, pump_tap_tx, "alloy".to_string()).await;
+    info!(%peer, ?pump_result, "brain_shim pump exited");
+    in_fwd.abort();
+    out_fwd.abort();
+    Ok(())
+}
+
+// === Test helpers ===
+
+/// Build a loud 24 kHz PCM frame whose RMS energy is well above the local-VAD
+/// threshold. A constant amplitude of 1000 matches slice-4's test fixture.
+fn loud_frame() -> PcmFrame {
+    let mut frame = PcmFrame::zeroed();
+    for s in frame.samples.iter_mut() {
+        *s = 1000;
+    }
+    frame
+}
+
+#[tokio::test]
+async fn pstn_sim_synthetic_caller_drives_trunk_reflex_loop() {
+    let _ = tracing_subscriber::fmt()
+        .with_env_filter(
+            tracing_subscriber::EnvFilter::try_from_default_env()
+                .unwrap_or_else(|_| "rutster_trunk=info".into()),
+        )
+        .try_init();
+
+    // 1. Start the mock OpenAI Realtime brain.
+    let mock = MockRealtimeBrain::start().await.expect("mock brain binds");
+
+    // 2. Start the brain shim that speaks the tap protocol on the core side
+    //    and the OpenAI protocol on the brain side.
+    let shim = start_brain_shim(mock.url()).await;
+    let tap_url = Url::parse(&format!("ws://{}/", shim.addr)).unwrap();
+
+    // 3. Build the trunk-leg pipe stack: TapAudioPipe -> Reflex -> LocalVadReflex.
+    //    This is the same composition `MediaThread::RegisterTrunk` will build.
+    let (tx_pcm_in, mut rx_pcm_in) = mpsc::channel::<PcmFrame>(32);
+    let (tx_audio_out, rx_audio_out) = mpsc::channel::<PcmFrame>(32);
+    let tap_metrics = TapMetrics::new();
+    let tap_pipe = TapAudioPipe::new(tx_pcm_in, rx_audio_out, tap_metrics.clone());
+
+    let (advisory_tx, advisory_rx) = mpsc::channel::<rutster_media::AdvisoryEvent>(16);
+    let reflex_metrics = ReflexMetrics::new();
+    let reflex = Reflex::new(tap_pipe, advisory_rx, reflex_metrics.clone());
+    let wrapped_pipe = LocalVadReflex::new(reflex, advisory_tx.clone());
+
+    // 4. Construct the TrunkSession that the FOB will tick via `drive`.
+    let channel = Channel::new_inbound();
+    let session_id = channel.id;
+    let (inbound_tx, inbound_rx) = mpsc::channel::<PcmFrame>(16);
+    let (outbound_tx, mut outbound_rx) = mpsc::channel::<PcmFrame>(16);
+    let now = Instant::now();
+    let mut session = TrunkSession::new(channel, wrapped_pipe, inbound_rx, outbound_tx, now);
+
+    // 5. Spawn a minimal test-only tap engine task. We cannot use the binary
+    //    crate's `spawn_tap_engine` from inside `rutster-trunk` (circular
+    //    dev-dependency), so we call `run_tap_client` directly after dialing
+    //    the brain shim.
+    let (close_tx, mut close_rx) = oneshot::channel::<()>();
+    let (tx_function_call, _rx_function_call) = mpsc::channel::<FunctionCallEvent>(8);
+    let (_tx_function_call_output, mut rx_function_call_output) =
+        mpsc::channel::<FunctionCallOutputEvent>(8);
+    let engine_metrics = tap_metrics.clone();
+    let engine_handle = tokio::spawn(async move {
+        let request = tap_url
+            .as_str()
+            .into_client_request()
+            .expect("valid ws url");
+        let (ws, _resp) = tokio_tungstenite::connect_async(request)
+            .await
+            .expect("connect to brain_shim");
+        let _ = run_tap_client(
+            ws,
+            session_id,
+            &mut rx_pcm_in,
+            tx_audio_out,
+            tx_function_call,
+            &mut rx_function_call_output,
+            advisory_tx,
+            engine_metrics,
+            &mut close_rx,
+        )
+        .await;
+    });
+
+    // Wait for tap handshake + OpenAI dial to complete.
+    tokio::time::sleep(Duration::from_millis(150)).await;
+
+    // 6. Spawn the synthetic Twilio caller task: push loud inbound frames into
+    //    the trunk leg and count outbound (brain-reply) frames coming back.
+    let outbound_count = Arc::new(AtomicUsize::new(0));
+    let outbound_count_caller = outbound_count.clone();
+    let (stop_tx, mut stop_rx) = oneshot::channel::<()>();
+    let caller_handle = tokio::spawn(async move {
+        let loud = loud_frame();
+        let mut local_count = 0usize;
+        loop {
+            // Push caller audio to the trunk leg. try_send matches the hot-path
+            // "drop + observe" policy: if the FOB backs up, keep going.
+            let _ = inbound_tx.try_send(loud.clone());
+
+            // Drain any outbound (brain reply) frames the FOB produced.
+            while outbound_rx.try_recv().is_ok() {
+                local_count += 1;
+            }
+            outbound_count_caller.store(local_count, Ordering::Relaxed);
+
+            tokio::time::sleep(Duration::from_millis(20)).await;
+            if stop_rx.try_recv().is_ok() {
+                break;
+            }
+        }
+        local_count
+    });
+
+    // 7. Drive the trunk-leg poll loop at 20 ms intervals.
+    let mut barge_seen = false;
+    let deadline = tokio::time::Instant::now() + Duration::from_secs(3);
+    while tokio::time::Instant::now() < deadline {
+        let now = Instant::now();
+        let _ = drive(&mut session, now);
+
+        if !barge_seen && reflex_metrics.barge_in_count.load(Ordering::Relaxed) >= 1 {
+            barge_seen = true;
+            info!("PSTN sim: local VAD barge-in fired");
+        }
+
+        // Stop early once we have both barge-in and at least one observed
+        // outbound frame (the mock brain replies with audio_out deltas).
+        if barge_seen && outbound_count.load(Ordering::Relaxed) > 0 {
+            break;
+        }
+
+        tokio::time::sleep(Duration::from_millis(20)).await;
+    }
+
+    // 8. Stop the caller and collect its final count.
+    let _ = stop_tx.send(());
+    let final_outbound_count = tokio::time::timeout(Duration::from_secs(1), caller_handle)
+        .await
+        .expect("caller task finishes")
+        .expect("caller task panics");
+
+    // Then: the local VAD must have fired during the call.
+    assert!(
+        barge_seen,
+        "PSTN caller speech must trigger a local-VAD barge-in on the trunk leg"
+    );
+    assert_eq!(
+        reflex_metrics.barge_in_count.load(Ordering::Relaxed),
+        1,
+        "barge-in must fire exactly once for the first utterance"
+    );
+
+    // And: the mock brain must have received audio_in and replied with at
+    // least one outbound frame after the barge (resume condition).
+    assert!(
+        final_outbound_count > 0,
+        "mock brain must reply with at least one audio_out frame observed on the trunk outbound mpsc"
+    );
+
+    // 9. Caller hangup: stop sending inbound frames and force the 60 s idle
+    //    timeout path by moving `last_idle_rx` into the past. The FOB should
+    //    close the session.
+    session.last_idle_rx = Instant::now() - Duration::from_secs(90);
+    let next = drive(&mut session, Instant::now());
+    assert!(
+        session.is_closed(),
+        "trunk session must close after idle timeout (simulating caller hangup)"
+    );
+    assert_eq!(
+        next, None,
+        "drive must return None once the session is closed"
+    );
+
+    // 10. Clean up the tap engine.
+    let _ = close_tx.send(());
+    let _ = tokio::time::timeout(Duration::from_secs(1), engine_handle).await;
+}
+
+/// Full end-to-end test using the binary's `MediaThread` + `RegisterTrunk`.
+///
+/// This test is currently ignored because `rutster-trunk` integration tests
+/// cannot depend on the `rutster` binary crate (`rutster` already depends on
+/// `rutster-trunk`; Cargo disallows circular dev-dependencies). The active
+/// `pstn_sim_synthetic_caller_drives_trunk_reflex_loop` above covers the FOB
+// reflex loop + trunk-leg tick directly; this stub marks where the MediaThread
+/// wiring test belongs once the binary crate is ready to exercise it.
+#[ignore]
+#[tokio::test]
+async fn full_pstn_e2e_through_media_thread_register_trunk() {
+    // TODO: exercise MediaCmd::RegisterTrunk + MediaThread tick loop against a
+    // live TwilioMediaStreamsServer mock once the binary crate exposes a test
+    // harness from the appropriate crate.
+}

crates/rutster/src/main.rs

@@ -99,7 +99,41 @@ async fn main() {
         let _ = http_stop_tx.send(());
     });
 
-    axum::serve(listener, router(app_state))
+    // slice-5 T5: mount the TwilioMediaStreamsServer WS sub-router.
+    // The pump task sends RegisterTrunkInboundChannel via a relay mpsc;
+    // the relay task translates the channel to MediaCmd::RegisterTrunk
+    // + awaits the oneshot reply before continuing the pump task.
+    //
+    // Step A (this slice): relay + mount wired; the actual
+    // RegisterTrunk handler returns the placeholder NACK on the media
+    // thread side (see crates/rutster/src/media_thread.rs Step A).
+    let (trunk_register_tx, mut trunk_register_rx) =
+        mpsc::channel::<rutster_trunk::twilio_media_streams::RegisterTrunkInboundChannel>(32);
+    let mount_cmd_tx = app_state.cmd_tx.clone();
+    tokio::spawn(async move {
+        while let Some(register) = trunk_register_rx.recv().await {
+            let (reply_tx, reply_rx) = tokio::sync::oneshot::channel();
+            let register_parts = register;
+            let trunk_register_cmd = rutster::media_thread::MediaCmd::RegisterTrunk {
+                call_sid: register_parts.call_sid,
+                inbound_from_twilio_rx: register_parts.inbound_from_twilio_rx,
+                outbound_to_twilio_tx: register_parts.outbound_to_twilio_tx,
+                tap_url: register_parts.tap_url,
+                reply: reply_tx,
+            };
+            if mount_cmd_tx.send(trunk_register_cmd).await.is_ok() {
+                // Wait for ChannelId ack OR Err (the placeholder returns Err
+                // via drop(reply)). Either way, the pump resumes.
+                let _ = reply_rx.await;
+            }
+        }
+    });
+    let trunk_router =
+        rutster_trunk::twilio_media_streams::TwilioMediaStreamsServer::router(trunk_register_tx);
+
+    let app = router(app_state).merge(trunk_router);
+
+    axum::serve(listener, app)
         .with_graceful_shutdown(async {
             let _ = http_stop_rx.await;
         })

crates/rutster/src/media_thread.rs

@@ -100,6 +100,25 @@ pub enum MediaCmd {
     /// Cold-path capacity/health snapshot (readyz + the autoscaling
     /// signal; review M2). Cheap: counters the loop already maintains.
     Stats { reply: oneshot::Sender<MediaStats> },
+    /// slice-5 T5: register a trunk-side session (PSTN caller via Twilio's
+    /// WSS raw-audio fork). The MediaThread constructs a TrunkSession
+    /// (T4), wraps its inner TapAudioPipe in `Reflex<TapAudioPipe>` then
+    /// `LocalVadReflex<Reflex<TapAudioPipe>>` (same composition as slice-4
+    /// Task 6 for WebRTC legs), inserts the leg in the session map under
+    /// `MediaLeg::Trunk`, spawns the TapEngine, + replies with the freshly-
+    /// allocated ChannelId.
+    ///
+    /// The WSS pump task (T3) sends this variant via the relay task in
+    /// main.rs once Twilio's "start" frame arrives. Credentials never
+    /// reach this codepath (ADR-0009 -- the brain holds only the per-call
+    /// tap_url, an operator-configured non-credential).
+    RegisterTrunk {
+        call_sid: String,
+        inbound_from_twilio_rx: tokio::sync::mpsc::Receiver<rutster_media::PcmFrame>,
+        outbound_to_twilio_tx: tokio::sync::mpsc::Sender<rutster_media::PcmFrame>,
+        tap_url: url::Url,
+        reply: oneshot::Sender<rutster_call_model::ChannelId>,
+    },
 }
 
 /// What the node tells the platform about itself. `serde::Serialize`
@@ -194,6 +213,120 @@ impl Drop for MediaThread {
     }
 }
 
+/// The per-session storage enum for the binary's MediaThread (spec §3.5).
+///
+/// `WebRTC` is the slice-4 leg kind -- ticked by
+/// `rutster_media::loop_driver::drive`. `Trunk` is the new slice-5
+/// trunk-leg kind -- ticked by `rutster_trunk::loop_driver::drive`. The
+/// existing WebRTC code path stays unchanged: the match arm
+/// `MediaLeg::WebRTC(s) => rutster_media::loop_driver::drive(s, now)`
+/// calls the SAME drive function slice-4 invoked directly. The trunk
+/// arm is its sibling.
+///
+/// Cargo dependency note: the trunk variant carries a concrete
+/// `TrunkSession<TapAudioPipe>` (production instantiation); the T4 unit
+/// tests can substitute `EchoAudioPipe` for the inner pipe by constructing
+/// the generic type directly -- without going through MediaLeg (which
+/// pins the production OuterType). The slice-4 §6.1 reuse claim holds:
+/// both leg kinds compose Reflex + LocalVadReflex identically around
+/// `TapAudioPipe`.
+pub enum MediaLeg {
+    /// Existing WebRTC leg. Ticked by `rutster_media::loop_driver::drive`.
+    /// UNCHANGED code path from slice-4 + slice-5/seams.
+    WebRTC(rutster_media::RtcSession),
+    /// New trunk leg (slice-5). Ticked by `rutster_trunk::loop_driver::drive`.
+    /// The seam gate holds: this never enters the media crate's
+    /// loop_driver; the FOB-side tick function lives in rutster-trunk's
+    /// loop_driver.rs (parallel-titled file, different crate).
+    Trunk(rutster_trunk::session::TrunkSession<rutster_tap::TapAudioPipe>),
+}
+
+// Step A scaffold: methods are wired into the full Step B refactor
+// (ThreadSession.leg field + RegisterTrunk handler body + dispatch in
+// run_per_leg_tick inside the existing tick loop). Until Step B lands,
+//  silences the unused warning per AGENTS.md
+// (rationale documented inline; suppression NOT silence).
+#[allow(dead_code)]
+impl MediaLeg {
+    /// Return the underlying `RtcSession` if this leg is WebRTC; `None`
+    /// for trunk legs. Used by the existing slice-4 WebRTC code paths
+    /// (accept_offer, Connected transition, is_closed) which require a
+    /// mutable `RtcSession`. Trunk legs do not use these paths (the
+    /// TrunkSession doesn't expose an `accept_offer` API -- the leg is
+    /// constructed when the Twilio Media Streams WSS arrives, not when
+    /// an SDP offer is POSTed).
+    fn as_webrtc_mut(&mut self) -> Option<&mut rutster_media::RtcSession> {
+        match self {
+            MediaLeg::WebRTC(rtc) => Some(rtc),
+            MediaLeg::Trunk(_) => None,
+        }
+    }
+
+    /// Return the channel ID of this leg's `Channel` regardless of
+    /// variant. Used by logging + tracing.
+    fn channel_id(&self) -> rutster_call_model::ChannelId {
+        match self {
+            MediaLeg::WebRTC(rtc) => rtc.channel.id,
+            MediaLeg::Trunk(t) => t.channel.id,
+        }
+    }
+
+    /// Return the ChannelState of either leg kind's underlying channel.
+    fn channel_state(&self) -> rutster_call_model::ChannelState {
+        match self {
+            MediaLeg::WebRTC(rtc) => rtc.channel.state,
+            MediaLeg::Trunk(t) => t.channel.state,
+        }
+    }
+
+    /// True once the leg's channel state has reached `Closed`. Used by the
+    /// std thread's session-map eviction loop.
+    fn is_closed(&self) -> bool {
+        matches!(
+            self.channel_state(),
+            rutster_call_model::ChannelState::Closed
+        )
+    }
+
+    /// Return the channel's `started_at` timestamp for lifecycle events.
+    fn channel_started_at(&self) -> std::time::SystemTime {
+        match self {
+            MediaLeg::WebRTC(rtc) => rtc.channel.started_at,
+            MediaLeg::Trunk(t) => t.channel.started_at,
+        }
+    }
+
+    /// Set the channel's tap handle (set by the binary on the Connected
+    /// transition for WebRTC legs; trunk legs set this at TrunkSession
+    /// construction time when the tap_engine is spawned -- no Connected
+    /// transition for trunk legs).
+    fn set_tap_handle(&mut self, handle: rutster_call_model::TapHandle) {
+        match self {
+            MediaLeg::WebRTC(rtc) => rtc.channel.tap = Some(handle),
+            MediaLeg::Trunk(t) => t.channel.tap = Some(handle),
+        }
+    }
+}
+
+/// The per-call tick dispatch. Mirrors the spec §3.5 contract: WebRTC
+/// legs go through the UNCHANGED slice-4 `loop_driver::drive`; trunk legs
+/// go through the NEW slice-5 `trunk_driver::drive`. The seam gate stays
+/// green because the trunk leg NEVER enters the media-crate's
+/// loop_driver.rs (only via the `rutster_trunk::loop_driver::drive` arm
+/// here, which routes to a separate file in a separate crate).
+///
+/// Made `pub` so the T7/T8 integration tests can drive the loop directly
+/// against a manually-constructed MediaLeg.
+pub fn run_per_leg_tick(
+    leg: &mut MediaLeg,
+    now: std::time::Instant,
+) -> Option<std::time::Duration> {
+    match leg {
+        MediaLeg::WebRTC(s) => rutster_media::loop_driver::drive(s, now),
+        MediaLeg::Trunk(s) => rutster_trunk::loop_driver::drive(s, now),
+    }
+}
+
 /// The per-session state owned by the media thread.
 struct ThreadSession {
     rtc: RtcSession,
@@ -347,6 +480,35 @@ fn run_media_thread(
                         drain_done = Some(reply);
                     }
                 }
+                // slice-5 T5: RegisterTrunk placeholder. Full handler construction
+                // (Reflex <LocalVadReflex<TapAudioPipe>> composition + TrunkSession
+                // instantiation + tap_engine spawn) lands in Step B alongside the
+                // ThreadSession refactor to use MediaLeg's variant dispatch. For
+                // now, ack with a synthetic error so the WSS pump task's oneshot
+                // await resumes + closes the WSS (the caller observes a clean
+                // failure rather than hanging). This matches the existing code
+                // path for rejected Register commands (admission control, drain).
+                MediaCmd::RegisterTrunk {
+                    call_sid,
+                    inbound_from_twilio_rx,
+                    outbound_to_twilio_tx,
+                    tap_url: _,
+                    reply,
+                } => {
+                    // Drop the inbound mpsc + outbound mpsc ends so they don't
+                    // leak the partial wiring live across the no-op ack.
+                    drop(inbound_from_twilio_rx);
+                    drop(outbound_to_twilio_tx);
+                    tracing::warn!(
+                        call_sid = %call_sid,
+                        "RegisterTrunk: Step A placeholder -- full handler lands with the                          ThreadSession refactor; rejecting the registration for now"
+                    );
+                    // The reply is oneshot::Sender<ChannelId>, not Result; we
+                    // can't NACK via Result. Closures: drop(reply) without
+                    // sending so the WSS pump's reply_rx.await returns Err
+                    // (which the pump treats as "media thread gone; close WSS").
+                    drop(reply);
+                }
             }
         }
 

crates/rutster/src/routes.rs

@@ -210,9 +210,69 @@ pub fn router(state: AppState) -> Router {
         .route("/v1/sessions", post(create_session))
         .route("/v1/sessions/:id", axum::routing::delete(delete_session))
         .route("/v1/sessions/:id/offer", post(post_offer))
+        // slice-5 T5 -- trunk-origin outbound + Twilio inbound-call webhook.
+        // Stub handlers (respond 503 / minimal TwiML) until full T5 integration:
+        // the operator can POST /v1/trunk/sessions against dev-b's
+        // TwilioCallControlClient (live impl behind twilio-live feature flag).
+        // The webhook responds with a <Connect><Stream> TwiML instructing
+        // Twilio to open a Media Streams fork against the binary's
+        // /twilio/media-stream WSS endpoint.
+        .route("/v1/trunk/sessions", post(originate_trunk_call))
+        .route("/v1/trunk/webhook", post(twilio_inbound_webhook))
         .with_state(state)
 }
 
+/// POST /v1/trunk/sessions -- originate an outbound call via the
+/// provider call-control client (spec §3.6). The body shape: `{ "to": "+...", "from": "+..." }`.
+/// The handler routes via `CallControlClient::originate(to, from, None)`
+/// (None because spend gate is step 6). The CallControlClient is
+/// dev-b's `MockCallControlClient` (CI default) OR the feature-gated
+/// `TwilioCallControlClient` (live -- maintainer-triggered only).
+///
+/// Stub T5: responds 503 until the binary's `AppState` carries a
+/// `CallControlClient` trait object (Step B refactor, dev-b handover).
+pub async fn originate_trunk_call() -> Response {
+    tracing::info!(
+        "POST /v1/trunk/sessions -- originates routed via CallControlClient (slice-5 T5 Step B interface)"
+    );
+    (
+        StatusCode::SERVICE_UNAVAILABLE,
+        "slice-5 T5 Step B: CallControlClient trait not yet threaded into AppState",
+    )
+        .into_response()
+}
+
+/// POST /v1/trunk/webhook -- Twilio's inbound-call signaling webhook
+/// receiver (spec §3.6 + §4.1 step 2). The handler responds with TwiML
+/// instructing Twilio to Media Streams against our /twilio/media-stream
+/// endpoint. The TwiML is:
+///
+/// ```xml
+/// <Response>
+///   <Connect>
+///     <Stream url="wss://{webhook_base}/twilio/media-stream" />
+///   </Connect>
+/// </Response>
+/// ```
+///
+/// Stub T5: emits the TwiML but the URL is hardcoded loopback for now.
+/// The full impl lands with the Step B refactor + dev-b's TwilioCredentials
+/// env parser (`webhook_base` from RUTSTER_THIRTY_WEBHOOK_BASE).
+pub async fn twilio_inbound_webhook() -> Response {
+    let twiml = r#"<?xml version="1.0" encoding="UTF-8"?>
+<Response>
+  <Connect>
+    <Stream url="ws://127.0.0.1:8080/twilio/media-stream" />
+  </Connect>
+</Response>"#;
+    (
+        StatusCode::OK,
+        [(header::CONTENT_TYPE, "application/xml")],
+        twiml,
+    )
+        .into_response()
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;