feat(binary): MediaLeg enum + MediaCmd::RegisterTrunk variant + POST /v1/trunk/{sessions,webhook} routes + TwilioMediaStreamsServer router mount (slice-5 T5 Step A)

Step A -- minimal-viable T5 surface (no ThreadSession refactor):
  * MediaLeg enum { WebRTC(RtcSession), Trunk(TrunkSession<TapAudioPipe>) }
    + as_webrtc_mut + channel_id + channel_state + is_closed + set_tap_handle
    helpers (dead_code-suppressed per AGENTS.md with documented rationale;
    Step B will wire them).
  * run_per_leg_tick dispatch pub fn matching on MediaLeg: WebRTC => media
    crate's loop_driver::drive (unchanged), Trunk => rutster_trunk's
    loop_driver::drive (NEW slice-5). The seam gate holds: trunk leg never
    enters the media crate's loop_driver.rs.
  * MediaCmd::RegisterTrunk variant + placeholder arm (drops mpsc ends,
    drops oneshot reply without sending -- pump task's reply_rx.await
    returns Err + closes WSS gracefully per the existing cmd-rejection
    posture for Register admission control).
  * Two HTTP routes: POST /v1/trunk/sessions (503 stub -- the originate
    handler needs the CallControlClient trait threaded into AppState,
    Step B or dev-b handover) + POST /v1/trunk/webhook (returns TwiML
    with hardcoded loopback Media Streams fork URL -- T5 Step B + dev-b's
    TwilioCredentials env parser will populate webhook_base).
  * main.rs: spawned trunk_register relay task that converts
    RegisterTrunkInboundChannel -> MediaCmd::RegisterTrunk + awaits
    oneshot reply, plus mounts TwilioMediaStreamsServer::router alongside
    the existing /v1/sessions routes.

Step B (deferred to follow-up commit / dev-b handover):
  * Full ThreadSession refactor to use leg: MediaLeg instead of rtc:
    RtcSession for the WebRTC code path's existing tests (4 tests).
  * Actual RegisterTrunk handler body (spawn_tap_engine + Reflex +
    LocalVadReflex composition + TrunkSession construction).
  * ADR-0009 honoring verification (TwilioCredentials never flow through
    this module -- the call_sid + tap_url fields are operational
    correlation IDs, not credentials).
  * T7 reflex-on-trunk + T8 PSTN-sim e2e integration tests, which
    pin behavior end-to-end + force the Step B refactor's correctness.

ADR-0007 honored: zero SIP bytes parsed; wire surface in this commit is
JSON Twilio Media Streams protocol + HTTP/REST + axum router merge.

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

crates/rutster/Cargo.toml

@@ -11,6 +11,15 @@ description = "Rutster binary: axum signaling + media driver + static browser te
 rutster-media = { path = "../rutster-media" }
 rutster-call-model = { path = "../rutster-call-model" }
 rutster-tap = { path = "../rutster-tap" }
+<<<<<<< HEAD
+# rutster-trunk: the rented-transport crate owns `TwilioCredentials` (T2) +
+# the env parser's return type. The binary's `config::twilio_credentials`
+# (T6) constructs it from env vars + hands it to the live
+# `TwilioCallControlClient` (when `--features=twilio-live` is enabled on the
+# trunk crate). ADR-0009: `TwilioCredentials` is imported here but never
+# re-exported through the workspace root / never sent over the tap WS protocol.
+=======
+>>>>>>> 346bd16 (feat(binary): MediaLeg enum + MediaCmd::RegisterTrunk variant + POST /v1/trunk/{sessions,webhook} routes + TwilioMediaStreamsServer router mount (slice-5 T5 Step A))
 # rutster-trunk: the rented-transport crate owns `TwilioCredentials` (T2) +
 # the env parser's return type. The binary's `config::twilio_credentials`
 # (T6) constructs it from env vars + hands it to the live

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::*;