diff --git a/crates/rutster/Cargo.toml b/crates/rutster/Cargo.toml index 7c84e37..92dddb3 100644 --- a/crates/rutster/Cargo.toml +++ b/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 diff --git a/crates/rutster/src/main.rs b/crates/rutster/src/main.rs index 09b93b8..64a8a6b 100644 --- a/crates/rutster/src/main.rs +++ b/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::(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; }) diff --git a/crates/rutster/src/media_thread.rs b/crates/rutster/src/media_thread.rs index ad05721..74cbe2a 100644 --- a/crates/rutster/src/media_thread.rs +++ b/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 }, + /// 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` then + /// `LocalVadReflex>` (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, + outbound_to_twilio_tx: tokio::sync::mpsc::Sender, + tap_url: url::Url, + reply: oneshot::Sender, + }, } /// 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` (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), +} + +// 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 { + 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 > 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, 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); + } } } diff --git a/crates/rutster/src/routes.rs b/crates/rutster/src/routes.rs index 3f02131..3e8ccc4 100644 --- a/crates/rutster/src/routes.rs +++ b/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 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 +/// +/// +/// +/// +/// +/// ``` +/// +/// 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#" + + + + +"#; + ( + StatusCode::OK, + [(header::CONTENT_TYPE, "application/xml")], + twiml, + ) + .into_response() +} + #[cfg(test)] mod tests { use super::*;