diff --git a/crates/rustpbx-media/src/leg.rs b/crates/rustpbx-media/src/leg.rs index b4f113bf3..c0dd35b81 100644 --- a/crates/rustpbx-media/src/leg.rs +++ b/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 diff --git a/e2e/tests/test_ivr_bridge.py b/e2e/tests/test_ivr_bridge.py index 4ec727c00..fc98edacd 100644 --- a/e2e/tests/test_ivr_bridge.py +++ b/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() diff --git a/src/call/app/controller.rs b/src/call/app/controller.rs index cca2883ca..e15ab3b4b 100644 --- a/src/call/app/controller.rs +++ b/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), @@ -163,6 +166,18 @@ impl CallController { Ok(()) } + pub(crate) async fn transfer_await_result( + &self, + target: impl Into, + ) -> 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. diff --git a/src/call/app/event_loop.rs b/src/call/app/event_loop.rs index ae835e439..9c4085f85 100644 --- a/src/call/app/event_loop.rs +++ b/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 }, diff --git a/src/call/app/ivr/common.rs b/src/call/app/ivr/common.rs index 7dd228bcf..2ad4aae9c 100644 --- a/src/call/app/ivr/common.rs +++ b/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 { #[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 diff --git a/src/call/app/ivr/config.rs b/src/call/app/ivr/config.rs index a2d0be8d1..641857af6 100644 --- a/src/call/app/ivr/config.rs +++ b/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, }, Repeat, + Exit, Hangup { #[serde(default)] prompt: Option, @@ -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>, #[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 +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 +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 +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 +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::(invalid).is_err()); + } + + let node: ActionNode = serde_json::from_str(r#"{"type":"exit"}"#).unwrap(); + assert!(matches!(node.action, EntryAction::Exit)); } #[test] diff --git a/src/call/app/ivr/executor.rs b/src/call/app/ivr/executor.rs index f090834e8..05d4880d3 100644 --- a/src/call/app/ivr/executor.rs +++ b/src/call/app/ivr/executor.rs @@ -407,6 +407,7 @@ impl StepIvrApp { EntryAction::Voicemail { .. } => "Voicemail", EntryAction::Play { .. } => "Play", EntryAction::Repeat => "Repeat", + EntryAction::Exit => "Exit", EntryAction::Hangup { .. } => "Hangup", EntryAction::CollectExtension { .. } => "CollectExtension", EntryAction::Collect { .. } => "Collect", @@ -668,17 +669,16 @@ impl StepIvrApp { &mut self, mut event: Option, ) -> anyhow::Result { - // A digit buffered during a non-interruptible step (or while the - // previous provider response was in flight) is delivered before the - // natural event, so caller input is never silently lost. An explicit - // in-band DTMF (interruptible barge-in / provider-driven menu) always - // wins; the buffered digit stays queued for a later step. + // Buffered DTMF may replace only prompt completion. Typed results from + // later steps are authoritative and discard stale buffered digits. if let Some(digit) = self.pending_dtmf.pop_front() { - if matches!(event, Some(ProviderEvent::Dtmf { .. })) { - self.pending_dtmf.push_front(digit); - } else { + if matches!(&event, Some(ProviderEvent::AudioComplete { .. }) | None) { tracing::info!(digit = %digit, "StepIvrApp: delivering buffered DTMF to provider"); event = Some(ProviderEvent::Dtmf { digit }); + } else if matches!(&event, Some(ProviderEvent::Dtmf { .. })) { + self.pending_dtmf.push_front(digit); + } else { + self.pending_dtmf.clear(); } } @@ -852,6 +852,12 @@ impl StepIvrApp { Some(ProviderEvent::DtmfMenuTimeout) => { crate::rwi::TriggerInfo::new("dtmf_menu_timeout") } + Some(ProviderEvent::TransferResult { outcome }) => { + crate::rwi::TriggerInfo::with_detail( + "transfer_result", + serde_json::json!({ "outcome": outcome }), + ) + } None => crate::rwi::TriggerInfo::new("unknown"), }); @@ -894,6 +900,7 @@ impl StepIvrApp { ) -> anyhow::Result { let result = common::execute_action( &node.action, + node.wait_for_result, ctrl, ctx, &mut self.sess, @@ -1292,6 +1299,13 @@ impl CallApp for StepIvrApp { return self.__exec_node(ctrl, context).await; } } + AppEvent::TransferResult { outcome } => { + self.current_node = Some( + self.request_next(Some(ProviderEvent::TransferResult { outcome })) + .await?, + ); + return self.__exec_node(ctrl, context).await; + } AppEvent::Custom { name, data: _ } => { tracing::debug!(event = %name, "StepIvrApp custom event"); } @@ -4005,6 +4019,9 @@ mod tests { prompt_voice: None, min_digits: 11, max_digits: 11, + timeout_ms: 10_000, + inter_digit_timeout_ms: 3_000, + terminator: "#".into(), }); let followup = ActionNode::new(EntryAction::Transfer { target: "2001".into(), @@ -4026,8 +4043,10 @@ mod tests { matches!( c, CallCommand::Play { - source: crate::call::domain::MediaSource::File { path }, .. - } if path == "enter_phone.wav" + source: crate::call::domain::MediaSource::File { path }, + options: Some(options), + .. + } if path == "enter_phone.wav" && options.interrupt_on_dtmf ) }) .await; @@ -4045,6 +4064,27 @@ mod tests { .await; } + #[tokio::test] + async fn typed_result_discards_stale_buffered_dtmf() { + let provider = EventCapturingProvider::new(); + provider.first_call.store(true, Ordering::SeqCst); + let events = provider.captured_events.clone(); + let mut app = StepIvrApp::with_provider(Box::new(provider)); + app.pending_dtmf.push_back("1".to_string()); + + app.request_next(Some(ProviderEvent::PhoneCollected { + number: "10000000000".to_string(), + })) + .await + .unwrap(); + + assert!(app.pending_dtmf.is_empty()); + assert!(matches!( + events.lock().unwrap().as_slice(), + [Some(ProviderEvent::PhoneCollected { number })] if number == "10000000000" + )); + } + // ── Bug 8: Torecord forwards recording complete to provider ────────── #[tokio::test] diff --git a/src/call/app/ivr/provider.rs b/src/call/app/ivr/provider.rs index 3939bbf4a..881d569f1 100644 --- a/src/call/app/ivr/provider.rs +++ b/src/call/app/ivr/provider.rs @@ -1,4 +1,5 @@ use crate::call::app::ivr::config::{ActionNode, EntryAction, IvrProviderConfig}; +use crate::call::domain::TransferOutcome; use async_trait::async_trait; use serde::{Deserialize, Serialize}; use std::collections::HashMap; @@ -132,6 +133,9 @@ pub enum ProviderEvent { digit: String, }, DtmfMenuTimeout, + TransferResult { + outcome: TransferOutcome, + }, } /// Why an IVR session ended. @@ -206,6 +210,7 @@ impl Default for RetryConfig { prompt_text: None, prompt_voice: None, }, + wait_for_result: false, next: None, step_id: None, step_name: None, @@ -445,6 +450,18 @@ impl ActionProvider for StepProvider { mod tests { use super::*; + #[test] + fn transfer_result_uses_minimal_wire_contract() { + let event = ProviderEvent::TransferResult { + outcome: TransferOutcome::NotConnected, + }; + + assert_eq!( + serde_json::to_value(event).unwrap(), + serde_json::json!({"type": "transfer_result", "outcome": "not_connected"}) + ); + } + #[test] fn test_step_provider_endpoint_url_trims_whitespace_and_slash() { let provider = StepProvider::new(" http://127.0.0.1:28080/ivr/step/ "); diff --git a/src/call/app/ivr/third_party.rs b/src/call/app/ivr/third_party.rs index 47ac19131..32d914af3 100644 --- a/src/call/app/ivr/third_party.rs +++ b/src/call/app/ivr/third_party.rs @@ -200,6 +200,9 @@ impl ThirdPartyTreeProvider { prompt_voice: None, min_digits: 11, max_digits: 11, + timeout_ms: 10_000, + inter_digit_timeout_ms: 3_000, + terminator: "#".into(), }, "input_voice" => EntryAction::InputVoice { @@ -270,6 +273,7 @@ impl ThirdPartyTreeProvider { }); ActionNode { action, + wait_for_result: false, next, step_id: None, step_name: None, @@ -468,6 +472,9 @@ fn convert_node_static(node: &ThirdPartyNode) -> EntryAction { prompt_voice: None, min_digits: 11, max_digits: 11, + timeout_ms: 10_000, + inter_digit_timeout_ms: 3_000, + terminator: "#".into(), }, "input_voice" => EntryAction::InputVoice { scene: node.nodename.clone(), diff --git a/src/call/app/ivr/tree_app.rs b/src/call/app/ivr/tree_app.rs index 5af960227..c9f199352 100644 --- a/src/call/app/ivr/tree_app.rs +++ b/src/call/app/ivr/tree_app.rs @@ -698,6 +698,10 @@ impl IvrApp { info!(ivr = %self.definition.name, menu = %current, "IVR repeating menu"); self.enter_menu(¤t, ctrl, ctx).await } + EntryAction::Exit => { + self.state = IvrState::Done; + Ok(AppAction::Exit) + } EntryAction::Hangup { prompt, prompt_text, diff --git a/src/call/app/mod.rs b/src/call/app/mod.rs index efbb30d55..52cf85f33 100644 --- a/src/call/app/mod.rs +++ b/src/call/app/mod.rs @@ -73,6 +73,7 @@ //! 4. Each handler returns an `AppAction` directing the loop what to do next //! 5. When `AppAction::Exit` or `AppAction::Hangup` is returned, `on_exit()` is called +use crate::call::domain::TransferOutcome; use crate::callrecord::CallRecordHangupReason; use async_trait::async_trait; use serde::{Deserialize, Serialize}; @@ -236,6 +237,8 @@ pub enum AppEvent { ParticipantLeft { room_id: String, count: usize }, /// Conference ended by moderator ConferenceEnded { room_id: String }, + /// A transfer retained by this application reached a terminal outcome. + TransferResult { outcome: TransferOutcome }, /// Custom event with arbitrary JSON data Custom { name: String, diff --git a/src/call/domain/command.rs b/src/call/domain/command.rs index fdbcc5981..976b3fd10 100644 --- a/src/call/domain/command.rs +++ b/src/call/domain/command.rs @@ -18,6 +18,13 @@ use tokio::sync::mpsc; use super::{HangupCommand, LegId, MediaSource, RingbackPolicy}; +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum TransferOutcome { + NotConnected, + TargetEnded, +} + /// Type alias for CallCommand sender. pub type CallCommandTx = mpsc::Sender; /// Type alias for CallCommand receiver. @@ -94,6 +101,11 @@ pub enum CallCommand { attended: bool, }, + TransferAwaitResult { + leg_id: LegId, + target: String, + }, + /// Complete an attended transfer TransferComplete { /// The consultation leg @@ -549,6 +561,7 @@ impl CallCommand { | CallCommand::Reject { .. } | CallCommand::Hangup(_) | CallCommand::Transfer { .. } + | CallCommand::TransferAwaitResult { .. } | CallCommand::Hold { music: None, .. } | CallCommand::Unhold { .. } | CallCommand::Trace { .. } diff --git a/src/proxy/proxy_call.rs b/src/proxy/proxy_call.rs index 2d5e9534a..4601125b4 100644 --- a/src/proxy/proxy_call.rs +++ b/src/proxy/proxy_call.rs @@ -12,6 +12,7 @@ use std::time::Instant; use tokio_util::sync::CancellationToken; pub(crate) mod call_meta; +#[cfg(test)] pub(crate) mod dtmf; pub(crate) mod error_catalog; pub(crate) mod ivr_exec_hook; diff --git a/src/proxy/proxy_call/call_meta.rs b/src/proxy/proxy_call/call_meta.rs index a564bc540..9f0e995a5 100644 --- a/src/proxy/proxy_call/call_meta.rs +++ b/src/proxy/proxy_call/call_meta.rs @@ -1,4 +1,4 @@ -use crate::call::domain::{ReturnAppSpec, RtpTimeoutSide}; +use crate::call::domain::{ReturnAppSpec, RtpTimeoutSide, TransferOutcome}; use crate::callrecord::CallRecordHangupReason; use crate::proxy::proxy_call::state::SessionHangupMessage; use rsipstack::dialog::DialogId; @@ -45,6 +45,7 @@ pub struct CallMeta { /// Consumed once by the `CallCommand::StartReturnApp` handler so the /// return is one-shot. pub transfer_return_app: Option, + pub pending_transfer_outcome: Option, /// Ordered diagnostic timeline of the call (ring → answer → ivr → queue → /// transfer → bridge → hold/resume → plays → hangup). Persisted into the /// call-record `metadata["trace"]` array by `record_snapshot`. diff --git a/src/proxy/proxy_call/sip_session.rs b/src/proxy/proxy_call/sip_session.rs index b18533c01..d9e28f086 100644 --- a/src/proxy/proxy_call/sip_session.rs +++ b/src/proxy/proxy_call/sip_session.rs @@ -3219,6 +3219,7 @@ impl SipSession { } DialogState::Terminated(_, reason) => { self.update_leg_state(&LegId::from("caller"), LegState::Ended); + self.meta.pending_transfer_outcome = None; // Our own teardown BYE also emits a Terminated event. Keep an // earlier root cause (for example RTP timeout or autohangup). @@ -4364,7 +4365,7 @@ impl SipSession { info!(index = idx, target = %target.aor, "Trying sequential target"); match self - .try_single_target(target, callee_state_rx, None, None) + .try_single_target(target, callee_state_rx, None, None, None) .await { Ok(()) => { @@ -5278,6 +5279,7 @@ impl SipSession { callee_state_rx: &mut mpsc::UnboundedReceiver, stop_playback_on_answer: Option<&str>, no_trying_timeout: Option, + caller: Option, ) -> Result<(), CalleeError> { use rsipstack::dialog::dialog::DialogState; @@ -5305,6 +5307,9 @@ impl SipSession { let (mut invite_option, callee_uri, callee_call_id) = self.build_target_invite_option(target, None).await?; + if let Some(caller) = caller { + invite_option.caller = caller; + } self.meta.routed_caller = Some(invite_option.caller.to_string()); self.meta.routed_callee = Some(target.aor.to_string()); @@ -8902,11 +8907,38 @@ impl SipSession { ); }; Self::ok_or_failure( - self.handle_transfer(leg_id, target, attended, callee_state_rx) - .await, + self.handle_transfer( + leg_id, + target, + attended, + transfer::TransferDisposition::Detach, + callee_state_rx, + ) + .await, ) } + CallCommand::TransferAwaitResult { leg_id, target } => { + let Some(callee_state_rx) = callee_state_rx.as_deref_mut() else { + self.meta.pending_transfer_outcome = + Some(crate::call::domain::TransferOutcome::NotConnected); + self.deliver_pending_transfer_result(); + return CommandResult::failure( + "No callee state receiver available for transfer".to_string(), + ); + }; + let result = self + .handle_transfer( + leg_id, + target, + false, + transfer::TransferDisposition::AwaitResult, + callee_state_rx, + ) + .await; + Self::ok_or_failure(result) + } + CallCommand::TransferComplete { consult_leg } => { Self::ok_or_failure(self.handle_transfer_complete(consult_leg).await) } @@ -9402,6 +9434,11 @@ impl SipSession { .is_none_or(|d| d.state().is_terminated()); if !caller_alive { + self.meta.pending_transfer_outcome = None; + return CommandResult::success(); + } + + if self.deliver_pending_transfer_result() { return CommandResult::success(); } @@ -9441,6 +9478,25 @@ impl SipSession { CommandResult::success() } + fn deliver_pending_transfer_result(&mut self) -> bool { + let caller_alive = !self + .caller_dialog + .as_ref() + .is_none_or(|dialog| dialog.state().is_terminated()); + if !caller_alive { + self.meta.pending_transfer_outcome = None; + return false; + } + let Some(outcome) = self.meta.pending_transfer_outcome.take() else { + return false; + }; + if outcome == crate::call::domain::TransferOutcome::TargetEnded { + self.bridge.clear(); + } + self.app_event_bridge + .send_app_event(crate::call::app::ControllerEvent::TransferResult(outcome)) + } + /// Send a SIP INFO request to the dialog identified by the given leg. /// Supports `leg_id = "caller"` (the inbound caller dialog) and /// `leg_id = "callee"` (the connected callee dialog). @@ -9497,6 +9553,7 @@ impl SipSession { } async fn handle_hangup(&mut self, cmd: &HangupCommand) -> CommandResult { + self.meta.pending_transfer_outcome = None; let cascade = &cmd.cascade; // Record the system hangup reason (e.g. RtpTimeout from the RTP @@ -10043,10 +10100,8 @@ impl SipSession { info!(%leg_id, "Early media remote description set"); } } - } else { - // 180 Ringing (provisional response with no - // SDP) — notify the session so `on_call_ringing` - // hooks fire (cc_ringing for queue-dialed agents). + } + if resp.status_code == rsipstack::sip::StatusCode::Ringing { info!(session_id = %session_id, %leg_id, "SIP leg ringing (180)"); let _ = cmd_tx.send(CallCommand::LegRinging { leg_id: leg_id.clone(), @@ -12162,6 +12217,7 @@ mod tests { .handle_blind_transfer( LegId::from("caller"), "queue:test-queue".to_string(), + transfer::TransferDisposition::Detach, &mut callee_rx, ) .await; @@ -12233,6 +12289,7 @@ mod tests { .handle_blind_transfer( LegId::from("caller"), "queue:nonexistent".to_string(), + transfer::TransferDisposition::Detach, &mut callee_rx, ) .await; diff --git a/src/proxy/proxy_call/sip_session/transfer.rs b/src/proxy/proxy_call/sip_session/transfer.rs index d44dec9c1..b3765b392 100644 --- a/src/proxy/proxy_call/sip_session/transfer.rs +++ b/src/proxy/proxy_call/sip_session/transfer.rs @@ -1,7 +1,6 @@ use super::SipSession; use crate::call::domain::{CallCommand, LegId, LegState, ReturnAppSpec}; use crate::media::negotiate::MediaNegotiator; -use crate::proxy::proxy_call::dtmf::RtpDtmfDetector; use anyhow::{Result, anyhow}; use futures::{SinkExt, StreamExt}; use rsipstack::dialog::dialog::DialogState; @@ -15,6 +14,51 @@ use rustrtc::media::SampleStreamSource; use std::collections::HashMap; use std::time::Duration; +#[derive(Clone, Copy, PartialEq, Eq)] +pub(super) enum TransferDisposition { + Detach, + AwaitResult, +} + +fn use_b2bua(blind_transfer_use_refer: bool, disposition: TransferDisposition) -> bool { + !blind_transfer_use_refer || disposition == TransferDisposition::AwaitResult +} + +async fn wait_for_bridge_disconnect( + session_cancel: tokio_util::sync::CancellationToken, + bridge_cancel: tokio_util::sync::CancellationToken, + mut forward_handle: tokio::task::JoinHandle<()>, + mut reverse_handle: tokio::task::JoinHandle<()>, +) -> bool { + enum CompletedTask { + Session, + Forward, + Reverse, + } + + let completed = tokio::select! { + biased; + _ = session_cancel.cancelled() => CompletedTask::Session, + _ = &mut forward_handle => CompletedTask::Forward, + _ = &mut reverse_handle => CompletedTask::Reverse, + }; + + bridge_cancel.cancel(); + match completed { + CompletedTask::Session => { + let _ = forward_handle.await; + let _ = reverse_handle.await; + } + CompletedTask::Forward => { + let _ = reverse_handle.await; + } + CompletedTask::Reverse => { + let _ = forward_handle.await; + } + } + !session_cancel.is_cancelled() +} + /// Unified forward sink for the bridge: WS PCM16 → call. Two backing paths: /// - [`BridgeForwardSink::Track`]: a `VoiceEnginePeer` track sender (non-app /// B2BUA path). @@ -107,6 +151,7 @@ pub(crate) enum TransferTarget { Sip { uri: String, return_app: Option, + from_user: Option, }, } @@ -206,6 +251,7 @@ pub(crate) fn parse_transfer_target(target: &str) -> TransferTarget { TransferTarget::Sip { uri: format!("sip:{}", target), return_app: None, + from_user: None, } } else { let mut return_query: Vec<(&str, String)> = Vec::new(); @@ -275,6 +321,7 @@ pub(crate) fn parse_transfer_target(target: &str) -> TransferTarget { format!("sip:{}", uri) }; let mut return_query: Vec<(&str, String)> = Vec::new(); + let mut from_user = None; let clean_uri = if let Some(qpos) = sip.find('?') { let base = &sip[..qpos]; let qs = &sip[qpos + 1..]; @@ -287,7 +334,9 @@ pub(crate) fn parse_transfer_target(target: &str) -> TransferTarget { let key = parts.next().unwrap_or(""); let value = parts.next().unwrap_or(""); let decoded = super::pct_decode_query(value); - if key == "return_app" || key == "return_target" { + if key == "from_user" { + from_user = (!decoded.is_empty()).then_some(decoded); + } else if key == "return_app" || key == "return_target" { return_query.push((key, decoded)); } else if key.starts_with("return_") { return_query.push((key, decoded)); @@ -306,6 +355,7 @@ pub(crate) fn parse_transfer_target(target: &str) -> TransferTarget { TransferTarget::Sip { uri: clean_uri, return_app: ReturnTargetSpec::from_query_pairs(return_query.into_iter()), + from_user, } } }; @@ -315,6 +365,7 @@ pub(crate) fn parse_transfer_target(target: &str) -> TransferTarget { TransferTarget::Sip { uri: format!("sip:{}", target), return_app: None, + from_user: None, } } @@ -324,6 +375,34 @@ impl SipSession { leg_id: LegId, target: String, attended: bool, + disposition: TransferDisposition, + callee_state_rx: &mut mpsc::UnboundedReceiver, + ) -> Result<()> { + if disposition == TransferDisposition::AwaitResult { + self.meta.pending_transfer_outcome = + Some(crate::call::domain::TransferOutcome::NotConnected); + } + + let result = self + .handle_transfer_inner(leg_id, target, attended, disposition, callee_state_rx) + .await; + if disposition == TransferDisposition::AwaitResult { + if result.is_err() { + self.deliver_pending_transfer_result(); + } else { + self.meta.pending_transfer_outcome = + Some(crate::call::domain::TransferOutcome::TargetEnded); + } + } + result + } + + async fn handle_transfer_inner( + &mut self, + leg_id: LegId, + target: String, + attended: bool, + disposition: TransferDisposition, callee_state_rx: &mut mpsc::UnboundedReceiver, ) -> Result<()> { let leg = self.require_leg(&leg_id)?; @@ -346,7 +425,7 @@ impl SipSession { ); } } else { - self.handle_blind_transfer(leg_id, target, callee_state_rx) + self.handle_blind_transfer(leg_id, target, disposition, callee_state_rx) .await?; } @@ -362,13 +441,14 @@ impl SipSession { &mut self, leg_id: LegId, target: String, + disposition: TransferDisposition, callee_state_rx: &mut mpsc::UnboundedReceiver, ) -> Result<()> { self.meta.transfer_in_progress = true; self.sync_rtp_timeout_pause(); let result = self - .handle_blind_transfer_inner(leg_id, target, callee_state_rx) + .handle_blind_transfer_inner(leg_id, target, disposition, callee_state_rx) .await; if result.is_err() { @@ -384,9 +464,17 @@ impl SipSession { &mut self, leg_id: LegId, target: String, + disposition: TransferDisposition, callee_state_rx: &mut mpsc::UnboundedReceiver, ) -> Result<()> { - match parse_transfer_target(&target) { + let target = parse_transfer_target(&target); + if disposition == TransferDisposition::AwaitResult + && !matches!(target, TransferTarget::Sip { .. }) + { + return Err(anyhow!("wait_for_result requires a SIP transfer target")); + } + + match target { TransferTarget::Queue { name, return_app, @@ -428,7 +516,11 @@ impl SipSession { ) .await } - TransferTarget::Sip { uri, return_app } => { + TransferTarget::Sip { + uri, + return_app, + from_user, + } => { self.meta.transfer_return_app = self.resolve_return_app(return_app).await; let realm = self.server.proxy_config.load().select_realm(""); @@ -436,7 +528,10 @@ impl SipSession { let refer_to_uri = rsipstack::sip::Uri::try_from(normalized.as_str()) .map_err(|e| anyhow!("Invalid transfer target URI: {}", e))?; - if !self.server.proxy_config.load().blind_transfer_use_refer { + if use_b2bua( + self.server.proxy_config.load().blind_transfer_use_refer, + disposition, + ) { info!(session_id = %self.id, %leg_id, target = %uri, return_app = ?self.meta.transfer_return_app, "Blind transfer via B-leg INVITE (B2BUA)"); // The transfer target is a NEW peer — invalidate the cached // callee offer so `prepare_callee_media_offer` creates a @@ -445,6 +540,12 @@ impl SipSession { // apply_sdp(answer) for the transferred-to endpoint. self.media.callee_offer = None; self.media.callee_offer_cached_webrtc = None; + let caller = from_user + .map(|user| { + format!("sip:{}@{}", user, refer_to_uri.host_with_port).parse() + }) + .transpose() + .map_err(|e| anyhow!("Invalid transfer caller URI: {}", e))?; let mut location = crate::call::Location { aor: refer_to_uri.clone(), ..Default::default() @@ -488,7 +589,7 @@ impl SipSession { } } let result = self - .try_single_target(&location, callee_state_rx, None, None) + .try_single_target(&location, callee_state_rx, None, None, caller) .await; if result.is_ok() { // The B2BUA blind-transfer path swaps the B leg @@ -1105,9 +1206,15 @@ impl SipSession { let chunk: Vec = buf.drain(..samples_per_frame).collect(); if let BridgeForwardSink::Pcm(tx) = &forward_sink { - if tx.send(chunk).await.is_err() { - info!(%session_id, %leg_id, "Bridge forward: PCM channel closed"); - return; + tokio::select! { + biased; + _ = forward_cancel.cancelled() => return, + result = tx.send(chunk) => { + if result.is_err() { + info!(%session_id, %leg_id, "Bridge forward: PCM channel closed"); + return; + } + } } } else if let BridgeForwardSink::Track(sender) = &forward_sink { let chunk = if ws_sample_rate != enc_sample_rate { @@ -1150,9 +1257,15 @@ impl SipSession { leg_id: leg_id.clone(), digits: digits.to_string(), }; - if tx.send(cmd).await.is_err() { - warn!(session_id = %session_id, %leg_id, "Bridge forward: cmd_tx closed"); - break; + tokio::select! { + biased; + _ = forward_cancel.cancelled() => return, + result = tx.send(cmd) => { + if result.is_err() { + warn!(session_id = %session_id, %leg_id, "Bridge forward: cmd_tx closed"); + break; + } + } } } } @@ -1177,8 +1290,7 @@ impl SipSession { }; // ── 7. Reverse loop: call audio → raw PCM16 → WS + DTMF JSON ─ - // Reads from the PeerConnection's audio track directly so we - // can detect RFC 2833 telephone-event DTMF alongside audio. + // DTMF JSON comes from the session-level deduplicated event channel. let reverse_cancel = cancel_token.child_token(); let reverse_handle = { let leg_id = leg_id.clone(); @@ -1186,8 +1298,6 @@ impl SipSession { crate::utils::spawn(async move { use rustrtc::media::MediaSample; - let mut det = RtpDtmfDetector::default(); - // Capture audio track from PeerConnection let track = loop { if let Some(t) = SipSession::find_audio_receiver_track(&pc).await { @@ -1209,9 +1319,15 @@ impl SipSession { json = dtmf_json_rx.recv() => { match json { Some(json) => { - if ws_write.send(Message::Text(json.into())).await.is_err() { - warn!(session_id = %session_id, %leg_id, "Bridge WS DTMF json write failed"); - break; + tokio::select! { + biased; + _ = reverse_cancel.cancelled() => break, + result = ws_write.send(Message::Text(json.into())) => { + if result.is_err() { + warn!(session_id = %session_id, %leg_id, "Bridge WS DTMF json write failed"); + break; + } + } } } None => break, @@ -1223,21 +1339,7 @@ impl SipSession { let is_dtmf = frame.payload_type .map_or(false, |pt| dtmf_payload_types.contains(&pt)); - if is_dtmf { - // RFC 2833 telephone-event → detect digit - if let Some(digit) = det.observe(&frame.data, frame.rtp_timestamp) { - let json = serde_json::json!({ - "type": "dtmf", - "digit": digit.to_string(), - "leg_id": leg_id, - }); - info!(session_id = %session_id, %leg_id, digit = %digit.to_string(), "Bridge reverse DTMF detected"); - if ws_write.send(Message::Text(json.to_string().into())).await.is_err() { - warn!(session_id = %session_id, %leg_id, "Bridge WS DTMF write failed"); - break; - } - } - } else { + if !is_dtmf { // Regular audio frame — decode to PCM, resample, send as binary let pcm = decoder.decode(&frame.data); let samples = if dec_sample_rate != ws_sample_rate { @@ -1251,9 +1353,15 @@ impl SipSession { for s in &samples { bytes.extend_from_slice(&s.to_ne_bytes()); } - if ws_write.send(Message::Binary(bytes.into())).await.is_err() { - warn!(session_id = %session_id, %leg_id, "Bridge reverse audio write failed"); - break; + tokio::select! { + biased; + _ = reverse_cancel.cancelled() => break, + result = ws_write.send(Message::Binary(bytes.into())) => { + if result.is_err() { + warn!(session_id = %session_id, %leg_id, "Bridge reverse audio write failed"); + break; + } + } } } } @@ -1276,7 +1384,7 @@ impl SipSession { self.conference_bridge = crate::call::runtime::SessionConferenceBridge { bridge_handle: Some(crate::call::runtime::ConferenceBridgeHandle { _tasks: vec![], - cancel_token, + cancel_token: cancel_token.clone(), }), conf_id: Some(format!("bridge-{}", self.id.0)), }; @@ -1286,30 +1394,30 @@ impl SipSession { // handler reads `meta.transfer_return_app` (written here). let has_return_app = return_app.is_some(); self.meta.transfer_return_app = self.resolve_return_app(return_app).await; - if has_return_app { - if let Some(ref cmd_tx) = self.cmd_tx { - let cancel = self.cancel_token.child_token(); - let tx = cmd_tx.clone(); - let mon_session_id = session_id.clone(); - let mon = crate::utils::spawn(async move { - tokio::select! { - biased; - _ = cancel.cancelled() => {} - _ = async { - let _ = forward_handle.await; - let _ = reverse_handle.await; - } => { - if !cancel.is_cancelled() { - info!(session_id = %mon_session_id, "Bridge disconnected; starting return app"); - let cmd = CallCommand::StartReturnApp; - let _ = tx.send(cmd).await; - } + let cancel = self.cancel_token.child_token(); + let tx = self.cmd_tx.clone(); + let mon_session_id = session_id.clone(); + let mon = crate::utils::spawn(async move { + let bridge_disconnected = wait_for_bridge_disconnect( + cancel.clone(), + cancel_token, + forward_handle, + reverse_handle, + ) + .await; + if bridge_disconnected && has_return_app && let Some(tx) = tx { + tokio::select! { + biased; + _ = cancel.cancelled() => {} + result = tx.send(CallCommand::StartReturnApp) => { + if result.is_ok() { + info!(session_id = %mon_session_id, "Bridge disconnected; starting return app"); } } - }); - self.legs.push_task(leg_id.clone(), mon); + } } - } + }); + self.legs.push_task(leg_id.clone(), mon); info!(session_id = %self.id, %leg_id, endpoint = %endpoint, "Bridge established"); // A real media bridge is now active between caller and endpoint — the @@ -1353,8 +1461,13 @@ impl SipSession { format!("{}?Replaces={}", target, encoded_replaces) }; - self.handle_blind_transfer(leg_id, refer_target, callee_state_rx) - .await + self.handle_blind_transfer( + leg_id, + refer_target, + TransferDisposition::Detach, + callee_state_rx, + ) + .await } pub(super) async fn emit_refer_event( @@ -1589,6 +1702,32 @@ impl SipSession { mod tests { use super::*; + #[tokio::test] + async fn bridge_disconnect_cancels_blocked_peer_task() { + let session_cancel = tokio_util::sync::CancellationToken::new(); + let bridge_cancel = session_cancel.child_token(); + let forward = crate::utils::spawn(async {}); + let reverse_cancel = bridge_cancel.child_token(); + let reverse = crate::utils::spawn(async move { + reverse_cancel.cancelled().await; + }); + + let disconnected = tokio::time::timeout( + Duration::from_secs(1), + wait_for_bridge_disconnect(session_cancel, bridge_cancel, forward, reverse), + ) + .await + .expect("bridge monitor must cancel the blocked peer task"); + + assert!(disconnected); + } + + #[test] + fn await_result_forces_b2bua() { + assert!(use_b2bua(true, TransferDisposition::AwaitResult)); + assert!(!use_b2bua(true, TransferDisposition::Detach)); + } + // ------------------------------------------------------------------------- // parse_transfer_target — pure-function dispatch tests // @@ -1785,6 +1924,20 @@ mod tests { TransferTarget::Sip { uri: "sip:1001@pbx.local".to_string(), return_app: None, + from_user: None, + } + ); + } + + #[test] + fn test_parse_transfer_target_extracts_from_user() { + let t = parse_transfer_target("sip:room-123@pbx.example?from_user=relay-caller"); + assert_eq!( + t, + TransferTarget::Sip { + uri: "sip:room-123@pbx.example".to_string(), + return_app: None, + from_user: Some("relay-caller".to_string()), } ); } @@ -1797,6 +1950,7 @@ mod tests { TransferTarget::Sip { uri: "tel:+15551234567".to_string(), return_app: None, + from_user: None, } ); } @@ -1809,6 +1963,7 @@ mod tests { TransferTarget::Sip { uri: "sip:1001".to_string(), return_app: None, + from_user: None, } ); }