alee/rutster
0
0
Fork 0
Code Issues Pull Requests 2 Actions Packages Projects Releases Wiki Activity

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

Browse Source 
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>
This commit is contained in:
A.D.Lee
2026-07-05 03:40:54 -04:00
parent b76f0dcc4b
commit 346bd163fe
5 changed files with 259 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
Cargo.lock generated
View File

@@ -1194,6 +1194,7 @@ dependencies = [
"rutster-media",
"rutster-tap",
"rutster-tap-echo",
"rutster-trunk",
"serde",
"serde_json",
"thiserror 1.0.69",

1
crates/rutster/Cargo.toml
View File

@@ -11,6 +11,7 @@ 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" }
rutster-trunk = { path = "../rutster-trunk" }
axum = { workspace = true }
tokio = { workspace = true }
tokio-tungstenite = { workspace = true }

36
crates/rutster/src/main.rs
View File

@@ -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;
})

162
crates/rutster/src/media_thread.rs
View File

@@ -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);
}
}
}

60
crates/rutster/src/routes.rs
View File

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