feat: support interactive transfers and resilient media bridging

crates/rustpbx-media/src/leg.rs

@@ -865,6 +865,9 @@ impl Drop for LegInner {
 fn build_rtc_config(cfg: &LegConfig) -> RtcConfiguration {
     RtcConfiguration {
         transport_mode: cfg.transport.clone(),
+        external_ip: cfg.external_ip.clone(),
+        rtp_start_port: cfg.rtp_port_range.map(|(start, _)| start),
+        rtp_end_port: cfg.rtp_port_range.map(|(_, end)| end),
         buffer_drop_strategy: BufferDropStrategy::DropOldest,
         // ICE pre-ready buffering: packets are buffered only until the RTP
         // transport is set up, and DropOldest means depth stays tiny in steady
@@ -1030,6 +1033,19 @@ mod tests {
         a.stop();
     }
 
+    #[test]
+    fn rtp_leg_uses_configured_network_constraints() {
+        let mut cfg = LegConfig::rtp_pcmu();
+        cfg.external_ip = Some("203.0.113.10".to_string());
+        cfg.rtp_port_range = Some((20000, 20100));
+
+        let rtc = build_rtc_config(&cfg);
+
+        assert_eq!(rtc.external_ip.as_deref(), Some("203.0.113.10"));
+        assert_eq!(rtc.rtp_start_port, Some(20000));
+        assert_eq!(rtc.rtp_end_port, Some(20100));
+    }
+
     #[tokio::test]
     async fn webrtc_leg_generates_dtls_offer() {
         // A WebRTC (DTLS-SRTP) leg must produce a real WebRTC offer with a

e2e/tests/test_ivr_bridge.py

@@ -179,7 +179,7 @@ async def test_tree_ivr_bridge_audio_accuracy(pbx, sipbot_pool, tmp_path, ws_bri
 
 @pytest.mark.asyncio
 async def test_tree_ivr_bridge_dtmf_json(pbx, sipbot_pool, tmp_path, ws_bridge_server):
-    """Caller DTMF during bridge → forwarded as JSON text frames over the WS."""
+    """Each caller DTMF press is forwarded once over the bridge WebSocket."""
     greeting = tmp_path / "bridge_greeting.wav"
     h.generate_sine_wav(greeting, 440.0, 2.0, 8000, 0.5)
     pbx.config_builder.add_ivr(
@@ -209,7 +209,24 @@ async def test_tree_ivr_bridge_dtmf_json(pbx, sipbot_pool, tmp_path, ws_bridge_s
     deadline = asyncio.get_event_loop().time() + 12
     while asyncio.get_event_loop().time() < deadline:
         frames = ws_bridge_server.capture.dtmf_frames()
-        if any("dtmf" in f and ("5" in f) for f in frames):
+        matching = [
+            json.loads(frame)
+            for frame in frames
+            if json.loads(frame).get("type") == "dtmf"
+            and json.loads(frame).get("digit") == "5"
+        ]
+        if matching:
+            await asyncio.sleep(0.5)
+            frames = ws_bridge_server.capture.dtmf_frames()
+            matching = [
+                json.loads(frame)
+                for frame in frames
+                if json.loads(frame).get("type") == "dtmf"
+                and json.loads(frame).get("digit") == "5"
+            ]
+            assert len(matching) == 1, (
+                f"one DTMF press produced {len(matching)} bridge frames: {matching}"
+            )
             return
         await asyncio.sleep(0.5)
     frames = ws_bridge_server.capture.dtmf_frames()

src/call/app/controller.rs

@@ -77,6 +77,9 @@ pub enum ControllerEvent {
     /// Recording finished.
     RecordingComplete(RecordingInfo),
 
+    /// An application-owned transfer reached a terminal outcome.
+    TransferResult(crate::call::domain::TransferOutcome),
+
     /// Call hung up.
     Hangup(Option<CallRecordHangupReason>),
 
@@ -163,6 +166,18 @@ impl CallController {
         Ok(())
     }
 
+    pub(crate) async fn transfer_await_result(
+        &self,
+        target: impl Into<String>,
+    ) -> anyhow::Result<()> {
+        self.session
+            .send_command(CallCommand::TransferAwaitResult {
+                leg_id: LegId::from("caller"),
+                target: target.into(),
+            })?;
+        Ok(())
+    }
+
     /// Play an audio file.
     ///
     /// The `interruptible` flag determines if DTMF input should stop playback.

src/call/app/event_loop.rs

@@ -158,6 +158,19 @@ impl AppEventLoop {
                             }
                         }
                     }
+                    Some(ControllerEvent::TransferResult(outcome)) => {
+                        match Self::await_or_cancel(&self.cancel_token, self.app.on_external_event(
+                            AppEvent::TransferResult { outcome },
+                            &mut self.controller,
+                            &self.context
+                        )).await {
+                            WaitResult::Completed(res) => res,
+                            WaitResult::Cancelled => {
+                                self.app.on_exit(ExitReason::Cancelled).await?;
+                                Ok(AppAction::Exit)
+                            }
+                        }
+                    }
                     Some(ControllerEvent::Custom(name, data)) => {
                         match Self::await_or_cancel(&self.cancel_token, self.app.on_external_event(
                             AppEvent::Custom { name, data },

src/call/app/ivr/common.rs

@@ -45,6 +45,7 @@ pub enum WaitEvent {
         text: String,
         confidence: f32,
     },
+    TransferResult,
     /// Audio source was requested (e.g. via tts_text) but no TTS service is available.
     /// The caller should inform the provider and request a fallback action.
     NoAudio,
@@ -224,6 +225,7 @@ fn extract_tts_text(value: &serde_json::Value) -> Option<String> {
 #[allow(clippy::too_many_arguments)]
 pub async fn execute_action(
     action: &EntryAction,
+    wait_for_result: bool,
     ctrl: &mut CallController,
     ctx: &ApplicationContext,
     sess: &mut SessionData,
@@ -257,6 +259,10 @@ pub async fn execute_action(
                 t.push('?');
                 t.push_str(&query);
             }
+            if wait_for_result {
+                ctrl.transfer_await_result(t).await?;
+                return Ok(ActionResult::WaitFor(WaitEvent::TransferResult));
+            }
             Ok(ActionResult::Terminal(TerminalAction::Transfer(t)))
         }
         EntryAction::Queue {
@@ -447,6 +453,9 @@ pub async fn execute_action(
             prompt_voice,
             min_digits,
             max_digits,
+            timeout_ms,
+            inter_digit_timeout_ms,
+            terminator,
         } => {
             let audio = resolve_audio(
                 prompt.as_deref(),
@@ -456,15 +465,15 @@ pub async fn execute_action(
             )
             .await;
             if let Some(a) = audio {
-                ctrl.play_audio(a, false).await?;
+                ctrl.play_audio(a, true).await?;
             }
             let config = DtmfCollectConfig {
                 min_digits: *min_digits,
                 max_digits: *max_digits,
-                timeout: Duration::from_millis(10000),
-                terminator: Some('#'),
+                timeout: Duration::from_millis(*timeout_ms),
+                terminator: terminator.chars().next(),
                 play_prompt: None,
-                inter_digit_timeout: Some(Duration::from_millis(3000)),
+                inter_digit_timeout: Some(Duration::from_millis(*inter_digit_timeout_ms)),
             };
             let digits = ctrl.collect_dtmf(config).await?;
             sess.variables.insert("phone_number".into(), digits.clone());
@@ -662,6 +671,8 @@ pub async fn execute_action(
             }))
         }
 
+        EntryAction::Exit => Ok(ActionResult::Terminal(TerminalAction::Exit)),
+
         EntryAction::Play { .. }
         | EntryAction::Menu { .. }
         | EntryAction::Repeat

src/call/app/ivr/config.rs

@@ -1,4 +1,4 @@
-use serde::{Deserialize, Serialize};
+use serde::{Deserialize, Deserializer, Serialize, de::Error as _};
 use std::collections::HashMap;
 
 /// Top-level wrapper for the TOML file (`[ivr]` table).
@@ -186,6 +186,7 @@ pub enum EntryAction {
         prompt_voice: Option<String>,
     },
     Repeat,
+    Exit,
     Hangup {
         #[serde(default)]
         prompt: Option<String>,
@@ -309,6 +310,21 @@ pub enum EntryAction {
         min_digits: usize,
         #[serde(default = "default_phone_digits")]
         max_digits: usize,
+        #[serde(
+            default = "default_input_phone_timeout_ms",
+            deserialize_with = "deserialize_input_phone_timeout_ms"
+        )]
+        timeout_ms: u64,
+        #[serde(
+            default = "default_input_phone_inter_digit_timeout_ms",
+            deserialize_with = "deserialize_input_phone_inter_digit_timeout_ms"
+        )]
+        inter_digit_timeout_ms: u64,
+        #[serde(
+            default = "default_input_phone_terminator",
+            deserialize_with = "deserialize_input_phone_terminator"
+        )]
+        terminator: String,
     },
 
     InputVoice {
@@ -389,6 +405,8 @@ impl EntryAction {
 pub struct ActionNode {
     #[serde(flatten)]
     pub action: EntryAction,
+    #[serde(default, skip_serializing_if = "is_false")]
+    pub wait_for_result: bool,
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub next: Option<Box<ActionNode>>,
     #[serde(default, skip_serializing_if = "Option::is_none")]
@@ -403,6 +421,7 @@ impl ActionNode {
     pub fn new(action: EntryAction) -> Self {
         Self {
             action,
+            wait_for_result: false,
             next: None,
             step_id: None,
             step_name: None,
@@ -413,6 +432,7 @@ impl ActionNode {
     pub fn with_next(action: EntryAction, next: ActionNode) -> Self {
         Self {
             action,
+            wait_for_result: false,
             next: Some(Box::new(next)),
             step_id: None,
             step_name: None,
@@ -624,6 +644,9 @@ fn default_timeout_ms() -> u64 {
 fn default_max_retries() -> u32 {
     3
 }
+fn is_false(value: &bool) -> bool {
+    !*value
+}
 fn default_min_digits() -> usize {
     3
 }
@@ -639,6 +662,71 @@ fn default_webhook_timeout() -> u64 {
 fn default_phone_digits() -> usize {
     11
 }
+const MAX_INPUT_PHONE_TIMEOUT_MS: u64 = 300_000;
+const MAX_INPUT_PHONE_INTER_DIGIT_TIMEOUT_MS: u64 = 60_000;
+
+fn default_input_phone_timeout_ms() -> u64 {
+    10_000
+}
+fn default_input_phone_inter_digit_timeout_ms() -> u64 {
+    3_000
+}
+fn default_input_phone_terminator() -> String {
+    "#".to_string()
+}
+
+fn deserialize_input_phone_timeout_ms<'de, D>(deserializer: D) -> Result<u64, D::Error>
+where
+    D: Deserializer<'de>,
+{
+    deserialize_bounded_u64(deserializer, MAX_INPUT_PHONE_TIMEOUT_MS, "input_phone timeout_ms")
+}
+
+fn deserialize_input_phone_inter_digit_timeout_ms<'de, D>(
+    deserializer: D,
+) -> Result<u64, D::Error>
+where
+    D: Deserializer<'de>,
+{
+    deserialize_bounded_u64(
+        deserializer,
+        MAX_INPUT_PHONE_INTER_DIGIT_TIMEOUT_MS,
+        "input_phone inter_digit_timeout_ms",
+    )
+}
+
+fn deserialize_bounded_u64<'de, D>(
+    deserializer: D,
+    max: u64,
+    field: &str,
+) -> Result<u64, D::Error>
+where
+    D: Deserializer<'de>,
+{
+    let value = u64::deserialize(deserializer)?;
+    if (1..=max).contains(&value) {
+        Ok(value)
+    } else {
+        Err(D::Error::custom(format!("{field} must be between 1 and {max}")))
+    }
+}
+
+fn deserialize_input_phone_terminator<'de, D>(deserializer: D) -> Result<String, D::Error>
+where
+    D: Deserializer<'de>,
+{
+    let value = String::deserialize(deserializer)?;
+    let mut chars = value.chars();
+    let Some(terminator) = chars.next() else {
+        return Err(D::Error::custom("input_phone terminator must not be empty"));
+    };
+    if chars.next().is_some() || !matches!(terminator, '0'..='9' | '*' | '#' | 'A'..='D') {
+        return Err(D::Error::custom(
+            "input_phone terminator must be one DTMF character",
+        ));
+    }
+    Ok(value)
+}
 
 impl EntryAction {
     pub fn is_dtmf_menu(&self) -> bool {
@@ -878,8 +966,9 @@ action = { type = "menu", menu = "root" }
 
     #[test]
     fn test_transfer_return_to_ivr_alias_deserialize() {
-        let json = r#"{"type":"transfer","target":"1001","return_to_ivr":"step-ivr"}"#;
+        let json = r#"{"type":"transfer","target":"1001","return_to_ivr":"step-ivr","wait_for_result":true}"#;
         let node: ActionNode = serde_json::from_str(json).unwrap();
+        assert!(node.wait_for_result);
         match node.action {
             EntryAction::Transfer { return_target, .. } => {
                 assert_eq!(return_target.as_deref(), Some("step-ivr"));
@@ -1041,9 +1130,29 @@ action = { type = "menu", menu = "root" }
         ));
 
         // input_phone
-        let node: ActionNode =
-            serde_json::from_str(r#"{"type":"input_phone","prompt":"input_phone.wav"}"#).unwrap();
-        assert!(matches!(node.action, EntryAction::InputPhone { .. }));
+        let node: ActionNode = serde_json::from_str(
+            r##"{"type":"input_phone","prompt":"input_phone.wav","timeout_ms":25000,"inter_digit_timeout_ms":4000,"terminator":"#"}"##,
+        )
+        .unwrap();
+        assert!(matches!(
+            node.action,
+            EntryAction::InputPhone {
+                timeout_ms: 25_000,
+                inter_digit_timeout_ms: 4_000,
+                ref terminator,
+                ..
+            } if terminator == "#"
+        ));
+        for invalid in [
+            r##"{"type":"input_phone","timeout_ms":0}"##,
+            r##"{"type":"input_phone","inter_digit_timeout_ms":0}"##,
+            r##"{"type":"input_phone","terminator":"X"}"##,
+        ] {
+            assert!(serde_json::from_str::<ActionNode>(invalid).is_err());
+        }
+
+        let node: ActionNode = serde_json::from_str(r#"{"type":"exit"}"#).unwrap();
+        assert!(matches!(node.action, EntryAction::Exit));
     }
 
     #[test]