rvoip-sip 0.2.4

//! Simplified Media Adapter for rvoip-sip
//!
//! Thin translation layer between media-core and state machine.
//! Focuses only on essential media operations and events.

use crate::adapters::srtp_negotiator::{SrtpNegotiator, SrtpPair};
use crate::api::events::{Event, MediaSecurityKeying, MediaSecurityProfile, MediaSecurityState};
use crate::api::lifecycle::{LifecycleIndex, SessionEventPublisher};
use crate::api::unified::MediaMode;
use crate::cleanup_diag::{self, CleanupStage};
use crate::errors::{Result, SessionError};
use crate::session_store::SessionStore;
use crate::state_table::types::SessionId;
use dashmap::DashMap;
use rvoip_media_core::types::AudioFrame;
use rvoip_media_core::{
    relay::controller::{
        AudioSource, BridgeError, BridgeHandle, MediaConfig, MediaSessionController,
        MediaSessionInfo,
    },
    DialogId,
};
use rvoip_sip_core::sdp::SdpBuilder;
use rvoip_sip_core::types::sdp::{CryptoAttribute, CryptoSuite, ParsedAttribute, SdpSession};
use std::net::{IpAddr, SocketAddr};
use std::str::FromStr;
use std::sync::atomic::{AtomicU64, AtomicU8, Ordering};
use std::sync::Arc;
use tokio::sync::mpsc;

const DIAG_OFF: u8 = 1;
const DIAG_ON: u8 = 2;
const AUDIO_RECEIVER_CHANNEL_FRAMES: usize = 128;

static SRTP_DIAGNOSTICS: AtomicU8 = AtomicU8::new(DIAG_OFF);
static MEDIA_SDP_DIAGNOSTICS: AtomicU8 = AtomicU8::new(DIAG_OFF);

#[cfg(feature = "perf-infra-memory-diagnostics")]
fn spawn_memory_tracked<F>(kind: &'static str, future: F) -> tokio::task::JoinHandle<F::Output>
where
    F: std::future::Future + Send + 'static,
    F::Output: Send + 'static,
{
    rvoip_infra_common::memory_diagnostics::spawn_tracked(kind, future)
}

#[cfg(not(feature = "perf-infra-memory-diagnostics"))]
fn spawn_memory_tracked<F>(_: &'static str, future: F) -> tokio::task::JoinHandle<F::Output>
where
    F: std::future::Future + Send + 'static,
    F::Output: Send + 'static,
{
    tokio::spawn(future)
}

/// NEXT_STEPS B.1 diag — process-global cleanup counter so external
/// observers (the `perf_listener` example, integration tests) can read
/// how many sessions have been fully cleaned up without subscribing to
/// the full tracing log. This is wire-light: an `AtomicU64` per
/// process, incremented once per `cleanup_session` exit. The pair to
/// `perf_listener`'s `accepted_total` lets us see at a glance whether
/// the cleanup path is keeping pace with the accept path.
///
/// The instrumentation is unconditional (no env-gate) because the cost
/// is one atomic add per call hangup. Strip the public getter after
/// the 100-CPS knee is resolved if it turns out to be only diagnostic.
pub mod cleanup_session_diag {
    use std::sync::atomic::{AtomicU64, Ordering};

    static CLEANED: AtomicU64 = AtomicU64::new(0);

    /// Increment the cleanup counter. Called from
    /// [`super::MediaAdapter::cleanup_session`].
    pub fn record_cleanup() {
        CLEANED.fetch_add(1, Ordering::Relaxed);
    }

    /// Read the cumulative count of completed cleanups.
    pub fn cleaned_total() -> u64 {
        CLEANED.load(Ordering::Relaxed)
    }
}

fn sdp_origin_session_id(raw_id: &str) -> String {
    let candidate = raw_id
        .strip_prefix("media-session-")
        .or_else(|| raw_id.strip_prefix("session-"))
        .unwrap_or(raw_id);

    if !candidate.is_empty() && candidate.bytes().all(|b| b.is_ascii_digit()) {
        return candidate.to_string();
    }

    if let Ok(uuid) = uuid::Uuid::parse_str(candidate) {
        let bytes = uuid.as_u128().to_be_bytes();
        let low = u64::from_be_bytes(bytes[8..16].try_into().expect("uuid low bytes"));
        return low.max(1).to_string();
    }

    let mut hash = 14_695_981_039_346_656_037u64;
    for byte in raw_id.bytes() {
        hash ^= byte as u64;
        hash = hash.wrapping_mul(1_099_511_628_211);
    }
    hash.max(1).to_string()
}

fn direction_attribute(direction: crate::types::MediaDirection) -> &'static str {
    match direction {
        crate::types::MediaDirection::SendRecv => "sendrecv",
        crate::types::MediaDirection::SendOnly => "sendonly",
        crate::types::MediaDirection::RecvOnly => "recvonly",
        crate::types::MediaDirection::Inactive => "inactive",
    }
}

fn audio_direction(session: &SdpSession) -> Option<rvoip_sip_core::MediaDirection> {
    session
        .media_descriptions
        .iter()
        .find(|m| m.media == "audio")
        .and_then(|m| m.direction)
        .or(session.direction)
}

fn sip_direction_to_session(
    direction: rvoip_sip_core::MediaDirection,
) -> crate::types::MediaDirection {
    match direction {
        rvoip_sip_core::MediaDirection::SendRecv => crate::types::MediaDirection::SendRecv,
        rvoip_sip_core::MediaDirection::SendOnly => crate::types::MediaDirection::SendOnly,
        rvoip_sip_core::MediaDirection::RecvOnly => crate::types::MediaDirection::RecvOnly,
        rvoip_sip_core::MediaDirection::Inactive => crate::types::MediaDirection::Inactive,
    }
}

fn answer_direction_for_offer(
    offer_direction: &Option<rvoip_sip_core::MediaDirection>,
) -> crate::types::MediaDirection {
    match offer_direction.unwrap_or(rvoip_sip_core::MediaDirection::SendRecv) {
        rvoip_sip_core::MediaDirection::SendRecv => crate::types::MediaDirection::SendRecv,
        rvoip_sip_core::MediaDirection::SendOnly => crate::types::MediaDirection::RecvOnly,
        rvoip_sip_core::MediaDirection::RecvOnly => crate::types::MediaDirection::SendOnly,
        rvoip_sip_core::MediaDirection::Inactive => crate::types::MediaDirection::Inactive,
    }
}

fn local_direction_from_remote_answer(
    answer_direction: &Option<rvoip_sip_core::MediaDirection>,
) -> crate::types::MediaDirection {
    match answer_direction.unwrap_or(rvoip_sip_core::MediaDirection::SendRecv) {
        rvoip_sip_core::MediaDirection::SendRecv => crate::types::MediaDirection::SendRecv,
        rvoip_sip_core::MediaDirection::SendOnly => crate::types::MediaDirection::RecvOnly,
        rvoip_sip_core::MediaDirection::RecvOnly => crate::types::MediaDirection::SendOnly,
        rvoip_sip_core::MediaDirection::Inactive => crate::types::MediaDirection::Inactive,
    }
}

fn srtp_diagnostics_enabled() -> bool {
    SRTP_DIAGNOSTICS.load(Ordering::Relaxed) == DIAG_ON
}

fn media_diagnostics_enabled() -> bool {
    MEDIA_SDP_DIAGNOSTICS.load(Ordering::Relaxed) == DIAG_ON
}

fn sdp_diagnostics_enabled() -> bool {
    srtp_diagnostics_enabled() || media_diagnostics_enabled()
}

pub(crate) fn set_sdp_diagnostics(srtp_enabled: bool, media_enabled: bool) {
    SRTP_DIAGNOSTICS.store(
        if srtp_enabled { DIAG_ON } else { DIAG_OFF },
        Ordering::Relaxed,
    );
    MEDIA_SDP_DIAGNOSTICS.store(
        if media_enabled { DIAG_ON } else { DIAG_OFF },
        Ordering::Relaxed,
    );
}

fn emit_srtp_diag(line: String) {
    eprintln!("SRTP_DIAG {}", line);
    tracing::info!("SRTP_DIAG {}", line);
}

fn emit_media_diag(line: String) {
    eprintln!("MEDIA_DIAG {}", line);
    tracing::info!("MEDIA_DIAG {}", line);
}

fn emit_sdp_diag(line: String) {
    if srtp_diagnostics_enabled() {
        emit_srtp_diag(line.clone());
    }
    if media_diagnostics_enabled() {
        emit_media_diag(line);
    }
}

fn crypto_attribute_diag(count: usize) -> String {
    if count > 0 {
        format!("crypto_attrs={} sdp_attribute=a=crypto", count)
    } else {
        "crypto_attrs=0".to_string()
    }
}

fn audio_transport(session: &SdpSession) -> Option<&str> {
    session
        .media_descriptions
        .iter()
        .find(|m| m.media == "audio")
        .map(|m| m.protocol.as_str())
}

/// NEXT_STEPS C2 — lookup helper from RTP payload type to the
/// `a=rtpmap:` value (without the PT prefix). Returns `None` for
/// payload types we don't know how to advertise; callers should
/// skip emitting an rtpmap for those (legal per RFC 4566 for
/// static PTs 0/8 etc., but a builder convention here is to emit
/// rtpmap for every PT for explicitness).
pub(crate) fn rtpmap_for_pt(pt: u8) -> Option<&'static str> {
    match pt {
        0 => Some("PCMU/8000"),
        8 => Some("PCMA/8000"),
        9 => Some("G722/8000"),
        13 => Some("CN/8000"),
        18 => Some("G729/8000"),
        101 => Some("telephone-event/8000"),
        111 => Some("opus/48000/2"),
        _ => None,
    }
}

/// NEXT_STEPS C2 — `a=fmtp:` value for payload types that require
/// one. Returns `None` for codecs that work fine without an fmtp.
#[cfg(test)]
pub(crate) fn fmtp_for_pt(pt: u8) -> Option<&'static str> {
    fmtp_for_pt_with_g729_annex_b(pt, true)
}

pub(crate) fn fmtp_for_pt_with_g729_annex_b(pt: u8, g729_annex_b: bool) -> Option<&'static str> {
    match pt {
        18 if g729_annex_b => Some("annexb=yes"),
        18 => Some("annexb=no"),
        101 => Some("0-15"),
        // Opus (PT 111) defaults are fine for VoIP without fmtp; a
        // production deployment may want `useinbandfec=1; minptime=10`.
        _ => None,
    }
}

fn parse_annex_b_param(parameters: &str) -> Option<bool> {
    parameters.split(';').find_map(|part| {
        let (name, value) = part.trim().split_once('=')?;
        if !name.trim().eq_ignore_ascii_case("annexb") {
            return None;
        }
        match value.trim().trim_matches('"').to_ascii_lowercase().as_str() {
            "yes" | "true" | "1" => Some(true),
            "no" | "false" | "0" => Some(false),
            _ => None,
        }
    })
}

fn audio_fmtp_annex_b(session: &SdpSession, payload_type: u8) -> Option<bool> {
    let format = payload_type.to_string();
    session
        .media_descriptions
        .iter()
        .find(|m| m.media == "audio")
        .and_then(|m| m.get_fmtp(&format))
        .and_then(|fmtp| parse_annex_b_param(&fmtp.parameters))
}

fn negotiated_g729_annex_b(session: &SdpSession, local_annex_b: bool) -> bool {
    local_annex_b && audio_fmtp_annex_b(session, 18).unwrap_or(true)
}

fn select_primary_audio_payload(formats: &[String]) -> Option<u8> {
    let mut parsed = formats.iter().filter_map(|fmt| fmt.parse::<u8>().ok());
    parsed.find(|pt| !matches!(*pt, 13 | 101)).or_else(|| {
        formats
            .iter()
            .filter_map(|fmt| fmt.parse::<u8>().ok())
            .next()
    })
}

fn select_primary_audio_payload_from_session(session: &SdpSession) -> Option<u8> {
    session
        .media_descriptions
        .iter()
        .find(|m| m.media == "audio")
        .and_then(|m| select_primary_audio_payload(&m.formats))
}

fn codec_name_for_payload(payload_type: u8, g729_annex_b: bool) -> String {
    match payload_type {
        0 => "PCMU",
        8 => "PCMA",
        9 => "G722",
        13 => "CN",
        18 if g729_annex_b => "G729BA",
        18 => "G729A",
        101 => "telephone-event",
        111 => "opus",
        _ => return format!("PT{}", payload_type),
    }
    .to_string()
}

/// Build the SDP answer that declines an offered audio m-line per
/// RFC 3264 §6 / RFC 4568 §7.3. Port=0 signals refusal; the proto is
/// echoed from the offer so the peer can distinguish a policy
/// rejection from a parse error. A single dummy `0` format is
/// included because some peers (and some validators) reject m-lines
/// with an empty `<fmt>` list, even though RFC 3264 allows it.
pub(crate) fn build_port_zero_rejection_sdp(
    origin_session_id: &str,
    origin_version: u64,
    local_ip: &str,
    offered_transport: &str,
) -> Result<String> {
    let version_str = origin_version.to_string();
    let session = SdpBuilder::new("Session")
        .origin("-", origin_session_id, &version_str, "IN", "IP4", local_ip)
        .connection("IN", "IP4", local_ip)
        .time("0", "0")
        .media_audio(0, offered_transport)
        .formats(&["0"])
        .done()
        .build()
        .map_err(|e| {
            SessionError::SDPNegotiationFailed(format!(
                "SdpBuilder failed to build port-zero rejection answer: {}",
                e
            ))
        })?;
    Ok(session.to_string())
}

/// Audio format for recording
#[derive(Debug, Clone, Copy, serde::Serialize, serde::Deserialize)]
pub enum AudioFormat {
    Wav,
    Raw,
    Mp3,
}

/// Recording configuration
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub struct RecordingConfig {
    /// Path where the recording should be saved
    pub file_path: String,

    /// Audio format for the recording
    pub format: AudioFormat,

    /// Sample rate in Hz (e.g., 8000, 16000, 48000)
    pub sample_rate: u32,

    /// Number of channels (1 = mono, 2 = stereo)
    pub channels: u16,

    /// Include mixed audio from both legs (for conference recording)
    pub include_mixed: bool,

    /// Save separate tracks for each leg
    pub separate_tracks: bool,
}

impl Default for RecordingConfig {
    fn default() -> Self {
        Self {
            file_path: "/tmp/recording.wav".to_string(),
            format: AudioFormat::Wav,
            sample_rate: 8000,
            channels: 1,
            include_mixed: false,
            separate_tracks: false,
        }
    }
}

/// Recording status information
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub struct RecordingStatus {
    pub is_recording: bool,
    pub is_paused: bool,
    pub duration_seconds: f64,
    pub file_size_bytes: u64,
}

/// Negotiated media configuration
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub struct NegotiatedConfig {
    pub local_addr: SocketAddr,
    pub remote_addr: SocketAddr,
    pub codec: String,
    pub payload_type: u8,
    pub local_direction: crate::types::MediaDirection,
    pub remote_direction: crate::types::MediaDirection,
}

/// Minimal media adapter - just translates between media-core and state machine
pub struct MediaAdapter {
    /// Media-core controller
    pub(crate) controller: Arc<MediaSessionController>,

    /// Session store for updating IDs
    pub(crate) store: Arc<SessionStore>,

    /// Simple mapping of session IDs to dialog IDs (media-core uses DialogId)
    pub(crate) session_to_dialog: Arc<DashMap<SessionId, DialogId>>,
    pub(crate) dialog_to_session: Arc<DashMap<DialogId, SessionId>>,

    /// Store media session info for SDP generation
    media_sessions: Arc<DashMap<SessionId, MediaSessionInfo>>,

    /// Audio frame channels for receiving decoded audio from media-core
    audio_receivers: Arc<DashMap<SessionId, mpsc::Sender<AudioFrame>>>,

    /// Perf diagnostics for media lifecycle and active audio churn.
    session_create_attempt_total: Arc<AtomicU64>,
    session_create_success_total: Arc<AtomicU64>,
    session_create_failed_total: Arc<AtomicU64>,
    cleanup_attempt_total: Arc<AtomicU64>,
    cleanup_mapped_total: Arc<AtomicU64>,
    cleanup_fallback_total: Arc<AtomicU64>,
    cleanup_media_session_removed_total: Arc<AtomicU64>,
    cleanup_audio_receiver_removed_total: Arc<AtomicU64>,
    audio_subscriber_created_total: Arc<AtomicU64>,
    audio_subscriber_disconnected_total: Arc<AtomicU64>,
    audio_send_frames_total: Arc<AtomicU64>,
    audio_send_samples_total: Arc<AtomicU64>,

    /// Local IP for SDP generation
    local_ip: IpAddr,

    /// Port range for media
    media_port_start: u16,
    media_port_end: u16,

    /// Whether to allocate media-core sessions or generate SDP only.
    media_mode: MediaMode,

    /// Audio mixers for conferences
    audio_mixers: Arc<DashMap<crate::types::MediaSessionId, Vec<crate::types::MediaSessionId>>>,

    // ==== RFC 4568 SDES-SRTP state (Step 2B) ====
    /// Whether to attach `a=crypto:` lines to outgoing offers and to
    /// answer with `RTP/SAVP` when peer offers SRTP. When `false`,
    /// the adapter behaves like the pre-2B baseline (plain RTP/AVP).
    offer_srtp: bool,

    /// When `true`, refuse to fall back to plaintext RTP. UAC: a
    /// remote SDP without acceptable `a=crypto:` causes the
    /// negotiation function to return `Err`. UAS: an offer without
    /// `a=crypto:` is rejected with the same `Err`, which the state
    /// machine surfaces as `488 Not Acceptable Here`.
    srtp_required: bool,

    /// Crypto suites to offer in preference order when `offer_srtp`
    /// is set. Default: AES-CM-128 + HMAC-SHA1-80 then -32 per
    /// RFC 4568 §6.2.1 MTI plus low-bandwidth fallback.
    srtp_offered_suites: Vec<CryptoSuite>,

    /// UAC-side state held between `generate_sdp_offer` and
    /// `negotiate_sdp_as_uac`. The offerer-role `SrtpNegotiator`
    /// holds our locally-generated keys keyed by tag.
    pending_srtp_offerers: Arc<DashMap<SessionId, SrtpNegotiator>>,

    /// Negotiated SRTP context pairs keyed by session. Phase 2B.2
    /// will read these out and hand them to media-core's
    /// `start_secure_media`.
    pub(crate) negotiated_srtp: Arc<DashMap<SessionId, SrtpPair>>,

    /// Global event coordinator for publishing RFC 4733 DTMF events
    /// onto the rvoip-sip API event bus. Populated at boot via
    /// [`Self::set_global_coordinator`]; `None` in tests that bypass
    /// the full wiring.
    pub(crate) global_coordinator: Arc<
        tokio::sync::RwLock<
            Option<Arc<rvoip_infra_common::events::coordinator::GlobalEventCoordinator>>,
        >,
    >,

    /// App-level event publisher that updates lifecycle before bus delivery.
    pub(crate) app_event_publisher: Arc<tokio::sync::RwLock<Option<SessionEventPublisher>>>,

    /// Sprint 3 A6 — public RTP-side address advertised in SDP `c=` /
    /// `o=` / `m=audio` lines. Set at coordinator boot from either
    /// `Config::media_public_addr` (static override) or a successful
    /// `Config::stun_server` probe. `None` falls back to `local_ip` +
    /// the per-session local RTP port (today's behaviour).
    public_rtp_addr: std::sync::RwLock<Option<SocketAddr>>,

    /// Sprint 3 C1 — when `true`, generated offers and answers
    /// advertise PT 13 (RFC 3389 Comfort Noise) alongside the
    /// PCMU + PCMA + telephone-event format set. Set at coordinator
    /// boot from `Config::comfort_noise_enabled`.
    comfort_noise_enabled: bool,

    /// Sprint 3.5 C2 swap — when `true` (default), the answer's
    /// format list is the strict RFC 3264 §6 intersection of the
    /// offered formats and our supported set, in offerer-preference
    /// order. When `false`, the answer always advertises our full
    /// supported set (legacy pre-Sprint-3.5 behaviour). Set at
    /// coordinator boot from `Config::strict_codec_matching`.
    strict_codec_matching: bool,

    /// RTP payload types to advertise in outgoing offers and accept on inbound
    /// matching. Default `[0, 8, 101]` (PCMU + PCMA + telephone-event) is the
    /// beta full-media set. Comfort Noise (PT 13) is independently controlled
    /// by `comfort_noise_enabled` for back-compat with existing callers and is
    /// inserted into the list automatically when enabled.
    ///
    /// This low-level adapter can still render SDP metadata for additional
    /// payload types in tests, but `Config::validate` rejects codec sets that
    /// media-core cannot encode/decode for beta full-media operation.
    offered_codecs: Vec<u8>,

    /// G.729 Annex B VAD/DTX/CNG SDP policy. When PT 18 is offered, we emit
    /// `a=fmtp:18 annexb=yes` when this is true and `annexb=no` when false.
    /// Answers disable Annex B when either side advertises `annexb=no`.
    g729_annex_b: bool,
}

impl MediaAdapter {
    /// Create a new media adapter (no SRTP — equivalent to the
    /// pre-Step-2B behaviour).
    pub fn new(
        controller: Arc<MediaSessionController>,
        store: Arc<SessionStore>,
        local_ip: IpAddr,
        port_start: u16,
        port_end: u16,
    ) -> Self {
        Self {
            controller,
            store,
            session_to_dialog: Arc::new(DashMap::new()),
            dialog_to_session: Arc::new(DashMap::new()),
            media_sessions: Arc::new(DashMap::new()),
            audio_receivers: Arc::new(DashMap::new()),
            session_create_attempt_total: Arc::new(AtomicU64::new(0)),
            session_create_success_total: Arc::new(AtomicU64::new(0)),
            session_create_failed_total: Arc::new(AtomicU64::new(0)),
            cleanup_attempt_total: Arc::new(AtomicU64::new(0)),
            cleanup_mapped_total: Arc::new(AtomicU64::new(0)),
            cleanup_fallback_total: Arc::new(AtomicU64::new(0)),
            cleanup_media_session_removed_total: Arc::new(AtomicU64::new(0)),
            cleanup_audio_receiver_removed_total: Arc::new(AtomicU64::new(0)),
            audio_subscriber_created_total: Arc::new(AtomicU64::new(0)),
            audio_subscriber_disconnected_total: Arc::new(AtomicU64::new(0)),
            audio_send_frames_total: Arc::new(AtomicU64::new(0)),
            audio_send_samples_total: Arc::new(AtomicU64::new(0)),
            local_ip,
            media_port_start: port_start,
            media_port_end: port_end,
            media_mode: MediaMode::Enabled,
            audio_mixers: Arc::new(DashMap::new()),
            offer_srtp: false,
            srtp_required: false,
            srtp_offered_suites: vec![
                CryptoSuite::AesCm128HmacSha1_80,
                CryptoSuite::AesCm128HmacSha1_32,
            ],
            pending_srtp_offerers: Arc::new(DashMap::new()),
            negotiated_srtp: Arc::new(DashMap::new()),
            global_coordinator: Arc::new(tokio::sync::RwLock::new(None)),
            app_event_publisher: Arc::new(tokio::sync::RwLock::new(None)),
            public_rtp_addr: std::sync::RwLock::new(None),
            comfort_noise_enabled: false,
            strict_codec_matching: true,
            offered_codecs: vec![0, 8, 101],
            g729_annex_b: true,
        }
    }

    /// Sprint 3 C1 — enable RFC 3389 Comfort Noise advertisement on
    /// outgoing offers and answers. Wired from
    /// `Config::comfort_noise_enabled` at coordinator boot. Mutates
    /// in place, mirroring the `set_srtp_policy` shape.
    ///
    /// Sprint 3.6 follow-up: also propagates the toggle into the
    /// underlying `MediaSessionController` so the VAD-driven
    /// `CnGate` activates on the audio TX path. With this wired,
    /// outbound PCM frames that VAD classifies as silence stop
    /// becoming G.711 packets and are replaced by periodic CN
    /// (PT 13) packets per RFC 3389 §4.1.
    pub fn set_comfort_noise(&mut self, enabled: bool) {
        self.comfort_noise_enabled = enabled;
        self.controller.set_comfort_noise_enabled(enabled);
    }

    /// Set media allocation behavior.
    pub fn set_media_mode(&mut self, mode: MediaMode) {
        self.media_mode = mode;
    }

    /// Return the current RTP receive packet count for a SIP session, when a
    /// media-core RTP session exists for it.
    pub(crate) async fn rtp_packets_received(&self, session_id: &SessionId) -> Option<u64> {
        let dialog_id = self.session_to_dialog.get(session_id)?.clone();
        self.controller
            .get_session_info(&dialog_id)
            .await
            .and_then(|info| info.rtp_stats.map(|stats| stats.packets_received))
    }

    /// Sprint 3.5 C2 swap — enable strict RFC 3264 §6 SDP-answer
    /// matching. Wired from `Config::strict_codec_matching` at
    /// coordinator boot.
    pub fn set_strict_codec_matching(&mut self, strict: bool) {
        self.strict_codec_matching = strict;
    }

    /// Set the list of RTP payload types this UA advertises in offers and
    /// accepts in answers. The adapter assumes the caller already ran
    /// `Config::validate`; tests may still use this setter directly for
    /// parser/SDP coverage outside the beta full-media contract.
    pub fn set_offered_codecs(&mut self, codecs: Vec<u8>) {
        self.offered_codecs = codecs;
    }

    /// Set local G.729 Annex B preference for PT 18 SDP offer/answer.
    pub fn set_g729_annex_b(&mut self, enabled: bool) {
        self.g729_annex_b = enabled;
    }

    /// Feature-gated retained-object counts for perf leak investigations.
    #[cfg(feature = "perf-tests")]
    pub(crate) fn perf_diagnostic_counts(&self) -> serde_json::Value {
        let audio_receiver_queue_frames: usize = self
            .audio_receivers
            .iter()
            .map(|entry| {
                let sender = entry.value();
                sender.max_capacity().saturating_sub(sender.capacity())
            })
            .sum();
        let audio_receiver_capacity_frames: usize = self
            .audio_receivers
            .iter()
            .map(|entry| entry.value().max_capacity())
            .sum();
        #[cfg(feature = "perf-media-diagnostics")]
        let controller_diagnostics = self.controller.diagnostic_counts();
        #[cfg(not(feature = "perf-media-diagnostics"))]
        let controller_diagnostics = serde_json::json!({
            "enabled": false,
            "compiled": false,
            "feature": "perf-media-diagnostics",
        });

        serde_json::json!({
            "session_to_dialog": self.session_to_dialog.len(),
            "dialog_to_session": self.dialog_to_session.len(),
            "media_sessions": self.media_sessions.len(),
            "audio_receivers": self.audio_receivers.len(),
            "audio_receiver_queue_frames": audio_receiver_queue_frames,
            "audio_receiver_capacity_frames": audio_receiver_capacity_frames,
            "pending_srtp_offerers": self.pending_srtp_offerers.len(),
            "negotiated_srtp": self.negotiated_srtp.len(),
            "audio_mixers": self.audio_mixers.len(),
            "controller": controller_diagnostics,
            "lifecycle": {
                "session_create_attempt_total": self.session_create_attempt_total.load(Ordering::Relaxed),
                "session_create_success_total": self.session_create_success_total.load(Ordering::Relaxed),
                "session_create_failed_total": self.session_create_failed_total.load(Ordering::Relaxed),
                "cleanup_attempt_total": self.cleanup_attempt_total.load(Ordering::Relaxed),
                "cleanup_mapped_total": self.cleanup_mapped_total.load(Ordering::Relaxed),
                "cleanup_fallback_total": self.cleanup_fallback_total.load(Ordering::Relaxed),
                "cleanup_media_session_removed_total": self.cleanup_media_session_removed_total.load(Ordering::Relaxed),
                "cleanup_audio_receiver_removed_total": self.cleanup_audio_receiver_removed_total.load(Ordering::Relaxed),
                "audio_subscriber_created_total": self.audio_subscriber_created_total.load(Ordering::Relaxed),
                "audio_subscriber_disconnected_total": self.audio_subscriber_disconnected_total.load(Ordering::Relaxed),
            },
            "audio_send": {
                "frames_total": self.audio_send_frames_total.load(Ordering::Relaxed),
                "samples_total": self.audio_send_samples_total.load(Ordering::Relaxed),
            },
        })
    }

    /// Compose the effective offer-format list: configured
    /// `offered_codecs` with comfort-noise (PT 13) inserted in front
    /// of DTMF (PT 101) when enabled, preserving the legacy ordering
    /// the byte-fixture tests pin.
    fn effective_offered_formats(&self) -> Vec<u8> {
        if !self.comfort_noise_enabled {
            return self.offered_codecs.clone();
        }
        let mut out = Vec::with_capacity(self.offered_codecs.len() + 1);
        for pt in &self.offered_codecs {
            if *pt == 101 {
                out.push(13);
            }
            out.push(*pt);
        }
        if !out.contains(&13) {
            // No DTMF in the list — append CN at the end.
            out.push(13);
        }
        out
    }

    async fn apply_negotiated_media_config(
        &self,
        dialog_id: &DialogId,
        remote_addr: SocketAddr,
        codec: &str,
    ) -> Result<()> {
        let mut config = self
            .controller
            .get_session_info(dialog_id)
            .await
            .ok_or_else(|| {
                SessionError::MediaError(format!("No media session for dialog {}", dialog_id))
            })?
            .config;
        config.remote_addr = Some(remote_addr);
        config.preferred_codec = Some(codec.to_string());

        self.controller
            .update_media(dialog_id.clone(), config)
            .await
            .map_err(|e| {
                SessionError::MediaError(format!("Failed to apply negotiated media config: {}", e))
            })
    }

    /// Set the public RTP address advertised in SDP. Called at
    /// coordinator boot from `Config::media_public_addr` (static
    /// override) or a successful STUN probe. Idempotent — subsequent
    /// calls overwrite. The IP address goes into `c=`/`o=` lines and
    /// the port (when set) replaces `info.rtp_port` in `m=audio`.
    pub fn set_public_rtp_addr(&self, addr: Option<SocketAddr>) {
        if let Ok(mut guard) = self.public_rtp_addr.write() {
            *guard = addr;
        }
    }

    /// Read the current public RTP address override (used by tests
    /// and by SDP generation).
    pub(crate) fn public_rtp_addr(&self) -> Option<SocketAddr> {
        self.public_rtp_addr.read().ok().and_then(|g| *g)
    }

    /// Local IP address bound by the adapter. Used by the Sprint 3
    /// A6 STUN probe to bind its temp socket on the same interface.
    pub fn local_ip(&self) -> IpAddr {
        self.local_ip
    }

    /// Install the global event coordinator so the adapter can publish
    /// RFC 4733 DTMF events onto the rvoip-sip API event stream.
    /// Idempotent — a later call replaces any prior coordinator.
    pub async fn set_global_coordinator(
        &self,
        coordinator: Arc<rvoip_infra_common::events::coordinator::GlobalEventCoordinator>,
    ) {
        *self.global_coordinator.write().await = Some(coordinator.clone());
        let mut publisher = self.app_event_publisher.write().await;
        if publisher.is_none() {
            *publisher = Some(SessionEventPublisher::new(
                coordinator,
                LifecycleIndex::new(),
            ));
        }
    }

    /// Install the app-level event publisher. This is preferred over direct
    /// global-coordinator publication because it updates lifecycle first.
    pub(crate) async fn set_app_event_publisher(&self, publisher: SessionEventPublisher) {
        *self.app_event_publisher.write().await = Some(publisher);
    }

    /// Configure the SRTP offer policy. Called by `UnifiedCoordinator`
    /// when constructing the adapter from a [`Config`](crate::api::unified::Config) that has
    /// `offer_srtp` / `srtp_required` / `srtp_offered_suites` set.
    /// Mutates in place rather than returning a new adapter so the
    /// existing constructor signature stays unchanged.
    pub fn set_srtp_policy(
        &mut self,
        offer_srtp: bool,
        srtp_required: bool,
        suites: Vec<CryptoSuite>,
    ) {
        self.offer_srtp = offer_srtp;
        self.srtp_required = srtp_required;
        if !suites.is_empty() {
            self.srtp_offered_suites = suites;
        }
    }

    // ===== Outbound Actions (from state machine) =====

    /// Start a media session
    pub async fn start_session(&self, session_id: &SessionId) -> Result<()> {
        // Check if session already exists
        if let Some(dialog_id) = self
            .session_to_dialog
            .get(session_id)
            .map(|entry| entry.value().clone())
        {
            // Session already exists, check if it's started in media-core
            if self.controller.get_session_info(&dialog_id).await.is_some() {
                tracing::debug!("Media session already started for session {}", session_id.0);
                return Ok(());
            }
        }

        // If not, create it - delegate to create_session
        let _media_id = self.create_session(session_id).await?;
        Ok(())
    }

    /// Generate SDP offer (for UAC).
    ///
    /// Built via `sip-core`'s typed `SdpBuilder` (RFC 8866). The
    /// previous format-string implementation produced byte-identical
    /// output to this version when the `offer_srtp` knob is not set —
    /// Extract the `a=crypto:` attributes from the audio m-section of
    /// a parsed SDP. Empty result means the peer offered no SRTP.
    fn extract_audio_crypto(session: &SdpSession) -> Vec<CryptoAttribute> {
        session
            .media_descriptions
            .iter()
            .find(|m| m.media == "audio")
            .map(|m| {
                m.generic_attributes
                    .iter()
                    .filter_map(|a| match a {
                        ParsedAttribute::Crypto(c) => Some(c.clone()),
                        _ => None,
                    })
                    .collect()
            })
            .unwrap_or_default()
    }

    /// Process SDP answer and negotiate (for UAC)
    pub async fn negotiate_sdp_as_uac(
        &self,
        session_id: &SessionId,
        remote_sdp: &str,
    ) -> Result<NegotiatedConfig> {
        // Parse remote SDP to extract IP and port
        let (remote_ip, remote_port) = self.parse_sdp_connection(remote_sdp)?;
        let srtp_diagnostics = srtp_diagnostics_enabled();
        if sdp_diagnostics_enabled() {
            if let Ok(parsed) = SdpSession::from_str(remote_sdp) {
                emit_sdp_diag(format!(
                    "remote_sdp_answer session={} media={}:{} transport={} {}",
                    session_id.0,
                    remote_ip,
                    remote_port,
                    audio_transport(&parsed).unwrap_or("unknown"),
                    crypto_attribute_diag(Self::extract_audio_crypto(&parsed).len())
                ));
            }
        }

        // SDES answer-side handling (RFC 4568 §7.5).
        // The state machine path that calls us doesn't expose a
        // failure-with-487 hook today; if `srtp_required` and the
        // answer can't satisfy SRTP, we surface `SDPNegotiationFailed`
        // which the executor turns into terminal `CallFailed`.
        if let Some((_, offerer_state)) = self.pending_srtp_offerers.remove(session_id) {
            // We did offer SRTP. Look for a matching `a=crypto:` in the answer.
            let parsed = SdpSession::from_str(remote_sdp).map_err(|e| {
                SessionError::SDPNegotiationFailed(format!(
                    "Failed to parse remote SDP for SDES answer extraction: {}",
                    e
                ))
            })?;
            let attrs = Self::extract_audio_crypto(&parsed);
            if let Some(chosen) = attrs.first() {
                let pair = offerer_state.accept_answer(chosen)?;
                self.negotiated_srtp.insert(session_id.clone(), pair);
                tracing::info!(
                    "SDES answer accepted for session {}: tag {} suite {:?}",
                    session_id.0,
                    chosen.tag,
                    chosen.suite
                );
                if srtp_diagnostics {
                    emit_srtp_diag(format!(
                        "sdes_answer_accepted session={} suite={:?}",
                        session_id.0, chosen.suite
                    ));
                }
            } else if self.srtp_required {
                return Err(SessionError::SDPNegotiationFailed(
                    "srtp_required is set but the SDP answer carries no a=crypto: line".into(),
                ));
            } else {
                tracing::warn!(
                    "Session {} offered SRTP but the answer didn't accept it; \
                     proceeding plaintext (Config::srtp_required = false)",
                    session_id.0
                );
                let _ = offerer_state; // dropped — keys discarded
            }
        }

        let parsed_answer = SdpSession::from_str(remote_sdp).ok();
        let answer_direction = parsed_answer.as_ref().and_then(audio_direction);
        let payload_type = parsed_answer
            .as_ref()
            .and_then(select_primary_audio_payload_from_session)
            .unwrap_or(0);
        let negotiated_annex_b = parsed_answer
            .as_ref()
            .filter(|_| payload_type == 18)
            .map(|answer| negotiated_g729_annex_b(answer, self.g729_annex_b))
            .unwrap_or(false);
        let negotiated_codec = codec_name_for_payload(payload_type, negotiated_annex_b);

        // Update media session with remote address. SRTP contexts (if
        // negotiated in 2B.1) must be installed *between* updating the
        // remote address and starting the audio transmitter — the
        // transmitter spawns a send loop and we don't want any
        // plaintext packets going out before the encrypt-side
        // SrtpContext is in place.
        let signaling_only_port = self.signaling_only_local_port();
        let dialog_id = if signaling_only_port.is_some() {
            None
        } else {
            self.session_to_dialog
                .get(session_id)
                .map(|entry| entry.value().clone())
        };
        if let Some(dialog_id) = dialog_id {
            let remote_addr = SocketAddr::new(remote_ip, remote_port);

            self.apply_negotiated_media_config(&dialog_id, remote_addr, &negotiated_codec)
                .await?;

            // RFC 4568 SDES: install per-direction contexts before the
            // first wire packet flows.
            if let Some((_, pair)) = self.negotiated_srtp.remove(session_id) {
                let suite = pair.suite;
                self.controller
                    .install_srtp_contexts(&dialog_id, pair.send_ctx, pair.recv_ctx)
                    .await
                    .map_err(|e| {
                        SessionError::MediaError(format!("Failed to install SRTP contexts: {}", e))
                    })?;
                tracing::info!(
                    "🔒 SRTP contexts installed for session {} (suite {:?})",
                    session_id.0,
                    suite
                );
                self.record_media_security_negotiated(session_id, suite, true)
                    .await;
                if srtp_diagnostics {
                    emit_srtp_diag(format!(
                        "srtp_contexts_installed session={} role=uac suite={:?}",
                        session_id.0, suite
                    ));
                }
            }

            // Establish media flow (this starts audio transmission)
            self.controller
                .establish_media_flow(&dialog_id, remote_addr)
                .await
                .map_err(|e| {
                    SessionError::MediaError(format!("Failed to establish media flow: {}", e))
                })?;

            tracing::info!(
                "✅ Updated RTP remote address to {} for session {}",
                remote_addr,
                session_id.0
            );
        } else if signaling_only_port.is_some() {
            if let Some((_, pair)) = self.negotiated_srtp.remove(session_id) {
                self.record_media_security_negotiated(session_id, pair.suite, false)
                    .await;
            }
        }

        let local_port = match signaling_only_port {
            Some(port) => port,
            None => self.get_local_port(session_id)?,
        };
        let config = NegotiatedConfig {
            local_addr: SocketAddr::new(self.local_ip, local_port),
            remote_addr: SocketAddr::new(remote_ip, remote_port),
            codec: negotiated_codec,
            payload_type,
            local_direction: local_direction_from_remote_answer(&answer_direction),
            remote_direction: answer_direction
                .map(sip_direction_to_session)
                .unwrap_or(crate::types::MediaDirection::SendRecv),
        };

        // Event publishing will be handled by SessionCrossCrateEventHandler

        Ok(config)
    }

    /// Generate SDP answer and negotiate (for UAS)
    pub async fn negotiate_sdp_as_uas(
        &self,
        session_id: &SessionId,
        remote_sdp: &str,
    ) -> Result<(String, NegotiatedConfig)> {
        // Parse remote SDP — typed parse for both connection extraction
        // and SDES handling.
        let parsed_offer = SdpSession::from_str(remote_sdp).map_err(|e| {
            SessionError::SDPNegotiationFailed(format!("Failed to parse remote SDP: {}", e))
        })?;
        let (remote_ip, remote_port) = self.parse_sdp_connection(remote_sdp)?;
        let srtp_diagnostics = srtp_diagnostics_enabled();
        if sdp_diagnostics_enabled() {
            emit_sdp_diag(format!(
                "remote_sdp_offer session={} media={}:{} transport={} {}",
                session_id.0,
                remote_ip,
                remote_port,
                audio_transport(&parsed_offer).unwrap_or("unknown"),
                crypto_attribute_diag(Self::extract_audio_crypto(&parsed_offer).len())
            ));
        }

        // SDES UAS-side handling. Per RFC 4568 §7.3, if we require
        // SRTP and the offer doesn't include any `a=crypto:` lines we
        // must reject — the state-machine path turns the
        // `SDPNegotiationFailed` into a `488 Not Acceptable Here`
        // (decision D10).
        let offered_crypto = Self::extract_audio_crypto(&parsed_offer);
        let offered_transport = audio_transport(&parsed_offer)
            .unwrap_or("RTP/AVP")
            .to_string();
        let (answer_attr, srtp_pair, reject_with_port_zero) =
            if !offered_crypto.is_empty() && self.offer_srtp {
                // Both sides want SRTP — negotiate.
                let answerer = SrtpNegotiator::new_answerer();
                let (chosen, pair) = answerer.process_offer(&offered_crypto)?;
                self.negotiated_srtp.insert(session_id.clone(), pair);
                tracing::info!(
                    "SDES offer accepted for session {}: tag {} suite {:?}",
                    session_id.0,
                    chosen.tag,
                    chosen.suite
                );
                if srtp_diagnostics {
                    emit_srtp_diag(format!(
                        "sdes_offer_accepted session={} suite={:?}",
                        session_id.0, chosen.suite
                    ));
                }
                (Some(chosen), true, false)
            } else if offered_crypto.is_empty() && self.srtp_required {
                return Err(SessionError::SDPNegotiationFailed(
                    "srtp_required is set but the SDP offer carries no a=crypto: line".into(),
                ));
            } else if !offered_crypto.is_empty() && !self.offer_srtp {
                // RFC 3264 §6 + RFC 4568 §7.3: peer offered SRTP but our
                // policy is plaintext. Reject the m-line by setting port=0
                // in the answer, preserving the offered proto so the peer
                // can distinguish a rejection from a parse error.
                tracing::info!(
                    "Session {} received SRTP offer but local policy is offer_srtp=false; \
                 rejecting m-line with port=0 per RFC 3264 §6",
                    session_id.0
                );
                if srtp_diagnostics {
                    emit_srtp_diag(format!(
                        "sdes_offer_rejected session={} reason=local_policy",
                        session_id.0
                    ));
                }
                (None, false, true)
            } else {
                (None, false, false)
            };
        let _ = srtp_pair; // suppress unused warning — value retained via DashMap insert

        // Port-zero rejection short-circuit: build a minimal RFC 3264
        // §6 declined m-line answer and return without setting up any
        // media flow. The peer is responsible for either re-offering
        // (with plaintext) or terminating the dialog.
        if reject_with_port_zero {
            let (origin_session_id, origin_version) = self.next_sdp_origin(session_id).await?;
            let advertised_ip = self
                .public_rtp_addr()
                .map(|sa| sa.ip())
                .unwrap_or(self.local_ip);
            let sdp_answer = build_port_zero_rejection_sdp(
                &origin_session_id,
                origin_version,
                &advertised_ip.to_string(),
                offered_transport.as_str(),
            )?;
            let config = NegotiatedConfig {
                local_addr: SocketAddr::new(advertised_ip, 0),
                remote_addr: SocketAddr::new(remote_ip, remote_port),
                codec: "PCMU".to_string(),
                payload_type: 0,
                local_direction: crate::types::MediaDirection::Inactive,
                remote_direction: crate::types::MediaDirection::Inactive,
            };
            return Ok((sdp_answer, config));
        }

        let signaling_only_port = self.signaling_only_local_port();
        let local_port = match signaling_only_port {
            Some(port) => port,
            None => self.get_local_port(session_id)?,
        };

        let formats = compute_answer_formats(
            &parsed_offer,
            &self.effective_offered_formats(),
            self.strict_codec_matching,
            self.offer_srtp,
            self.srtp_required,
        )?;
        let negotiated_payload_type = select_primary_audio_payload(&formats).unwrap_or(0);
        let negotiated_annex_b = if negotiated_payload_type == 18 {
            negotiated_g729_annex_b(&parsed_offer, self.g729_annex_b)
        } else {
            false
        };
        let negotiated_codec = codec_name_for_payload(negotiated_payload_type, negotiated_annex_b);
        let offered_direction = audio_direction(&parsed_offer);
        let answer_direction = answer_direction_for_offer(&offered_direction);

        // Update media session with remote address. SRTP contexts must
        // be installed BEFORE establish_media_flow starts the audio
        // transmitter — see `negotiate_sdp_as_uac` for the same
        // ordering rationale.
        let dialog_id = if signaling_only_port.is_some() {
            None
        } else {
            self.session_to_dialog
                .get(session_id)
                .map(|entry| entry.value().clone())
        };
        if let Some(dialog_id) = dialog_id {
            let remote_addr = SocketAddr::new(remote_ip, remote_port);

            self.apply_negotiated_media_config(&dialog_id, remote_addr, &negotiated_codec)
                .await?;

            if let Some((_, pair)) = self.negotiated_srtp.remove(session_id) {
                let suite = pair.suite;
                self.controller
                    .install_srtp_contexts(&dialog_id, pair.send_ctx, pair.recv_ctx)
                    .await
                    .map_err(|e| {
                        SessionError::MediaError(format!(
                            "Failed to install SRTP contexts (UAS): {}",
                            e
                        ))
                    })?;
                tracing::info!(
                    "🔒 SRTP contexts installed for session {} (UAS, suite {:?})",
                    session_id.0,
                    suite
                );
                self.record_media_security_negotiated(session_id, suite, true)
                    .await;
                if srtp_diagnostics {
                    emit_srtp_diag(format!(
                        "srtp_contexts_installed session={} role=uas suite={:?}",
                        session_id.0, suite
                    ));
                }
            }

            self.controller
                .establish_media_flow(&dialog_id, remote_addr)
                .await
                .map_err(|e| {
                    SessionError::MediaError(format!("Failed to establish media flow: {}", e))
                })?;

            tracing::info!(
                "✅ Updated RTP remote address to {} for session {} (UAS)",
                remote_addr,
                session_id.0
            );
        } else if signaling_only_port.is_some() {
            if let Some((_, pair)) = self.negotiated_srtp.remove(session_id) {
                self.record_media_security_negotiated(session_id, pair.suite, false)
                    .await;
            }
        }

        // Generate the SDP answer.
        //
        // Sprint 3.5 — `negotiate_sdp_as_uas` now consumes the
        // generic RFC 3264 §6 matcher in
        // `rvoip_sip_dialog::sdp::match_offer`. The strict path
        // (default) intersects offered formats with our supported
        // set in offerer-preference order; the permissive path
        // (`Config::strict_codec_matching = false`) preserves the
        // pre-Sprint-3.5 "always answer with full set" behaviour
        // for deployments that depend on it.
        let (origin_session_id, origin_version) = self.next_sdp_origin(session_id).await?;
        let origin_version = origin_version.to_string();
        // Sprint 3 A6 — same public-address override as the offer
        // path, so answers carry the discovered/configured public
        // mapping when one is set.
        let public = self.public_rtp_addr();
        let advertised_ip = public.map(|sa| sa.ip()).unwrap_or(self.local_ip);
        let local_ip_str = advertised_ip.to_string();
        let advertised_port = public
            .filter(|sa| sa.port() != 0)
            .map(|sa| sa.port())
            .unwrap_or(local_port);
        let answer_transport = if answer_attr.is_some() {
            "RTP/SAVP"
        } else {
            "RTP/AVP"
        };

        if sdp_diagnostics_enabled() {
            emit_sdp_diag(format!(
                "local_sdp_answer session={} media={}:{} transport={} {} direction={}",
                session_id.0,
                advertised_ip,
                advertised_port,
                answer_transport,
                crypto_attribute_diag(usize::from(answer_attr.is_some())),
                direction_attribute(answer_direction)
            ));
        }

        let formats_str: Vec<&str> = formats.iter().map(|s| s.as_str()).collect();
        let mut media_builder = SdpBuilder::new("Session")
            .origin(
                "-",
                &origin_session_id,
                &origin_version,
                "IN",
                "IP4",
                &local_ip_str,
            )
            .connection("IN", "IP4", &local_ip_str)
            .time("0", "0")
            .media_audio(advertised_port, answer_transport)
            .formats(&formats_str);
        // Emit rtpmap/fmtp ONLY for the formats we kept. In the
        // permissive branch this is the full set; in the strict
        // branch the matcher's intersection has already filtered.
        // NEXT_STEPS C2 — routed through `rtpmap_for_pt` so adding a
        // new codec only requires extending the helper.
        for fmt in &formats {
            let Ok(pt) = fmt.parse::<u8>() else {
                continue;
            };
            if let Some(rtpmap) = rtpmap_for_pt(pt) {
                media_builder = media_builder.rtpmap(fmt.as_str(), rtpmap);
            }
            if let Some(fmtp) = fmtp_for_pt_with_g729_annex_b(pt, negotiated_annex_b) {
                media_builder = media_builder.fmtp(fmt.as_str(), fmtp);
            }
        }
        if let Some(attr) = answer_attr {
            media_builder = media_builder.crypto_attribute(attr);
        }
        let session = media_builder
            .attribute(direction_attribute(answer_direction), None::<String>)
            .done()
            .build()
            .map_err(|e| {
                SessionError::SDPNegotiationFailed(format!(
                    "SdpBuilder failed to build answer: {}",
                    e
                ))
            })?;
        let sdp_answer = session.to_string();

        let config = NegotiatedConfig {
            local_addr: SocketAddr::new(self.local_ip, local_port),
            remote_addr: SocketAddr::new(remote_ip, remote_port),
            codec: negotiated_codec,
            payload_type: negotiated_payload_type,
            local_direction: answer_direction,
            remote_direction: offered_direction
                .map(sip_direction_to_session)
                .unwrap_or(crate::types::MediaDirection::SendRecv),
        };

        // Event publishing will be handled by SessionCrossCrateEventHandler

        // Media flow is already represented by MediaStreamStarted above

        Ok((sdp_answer, config))
    }

    async fn record_media_security_negotiated(
        &self,
        session_id: &SessionId,
        suite: CryptoSuite,
        contexts_installed: bool,
    ) {
        let state = MediaSecurityState {
            keying: MediaSecurityKeying::Sdes,
            suite,
            profile: MediaSecurityProfile::RtpSavp,
            contexts_installed,
        };

        match self.store.get_session(session_id).await {
            Ok(mut session) => {
                session.media_security = Some(state.clone());
                if let Err(e) = self.store.update_session(session).await {
                    tracing::warn!(
                        "Failed to persist media security state for session {}: {}",
                        session_id.0,
                        e
                    );
                }
            }
            Err(e) => {
                tracing::debug!(
                    "Session {} was gone before media security state could be persisted: {}",
                    session_id.0,
                    e
                );
            }
        }

        let api_event = Event::MediaSecurityNegotiated {
            call_id: session_id.clone(),
            keying: state.keying,
            suite: state.suite,
            profile: state.profile,
            contexts_installed: state.contexts_installed,
        };

        if let Some(publisher) = self.app_event_publisher.read().await.clone() {
            publisher.publish(api_event);
        } else if let Some(coordinator) = self.global_coordinator.read().await.clone() {
            let wrapped = crate::adapters::SessionApiCrossCrateEvent::new(api_event);
            spawn_memory_tracked(
                "sip.media_adapter.media_security_publish_task",
                async move {
                    if let Err(e) = coordinator.publish(wrapped).await {
                        tracing::warn!("Failed to publish MediaSecurityNegotiated event: {}", e);
                    }
                },
            );
        } else {
            tracing::debug!(
                "MediaSecurityNegotiated publish skipped for session {}: no event publisher yet",
                session_id.0
            );
        }
    }

    /// Play an audio file to the remote party
    pub async fn play_audio_file(&self, session_id: &SessionId, file_path: &str) -> Result<()> {
        // Get the dialog ID for this session
        let _dialog_id = self
            .session_to_dialog
            .get(session_id)
            .ok_or_else(|| {
                SessionError::MediaError(format!("No media session for {}", session_id.0))
            })?
            .clone();

        // Send play file command to media controller
        // Note: The actual media-core API might differ
        tracing::info!(
            "Playing audio file {} for session {}",
            file_path,
            session_id.0
        );

        // In a real implementation, this would send the file path to the media relay
        // For now, we'll just log it
        // Send a media event (using MediaError as a workaround for now)
        // In production, we'd have proper event types for these
        tracing::debug!("Audio playback started: {}", file_path);

        Ok(())
    }

    /// Start recording the media session
    pub async fn start_recording_old(&self, session_id: &SessionId) -> Result<String> {
        // Get the dialog ID for this session
        let _dialog_id = self
            .session_to_dialog
            .get(session_id)
            .ok_or_else(|| {
                SessionError::MediaError(format!("No media session for {}", session_id.0))
            })?
            .clone();

        // Generate a unique recording filename
        let recording_path = format!("/tmp/recording_{}.wav", session_id.0);

        tracing::info!(
            "Starting recording for session {} at {}",
            session_id.0,
            recording_path
        );

        // In a real implementation, this would start recording through the media relay
        // For now, just log the recording start
        tracing::debug!("Recording started at: {}", recording_path);

        // Store recording path in session if needed
        if let Ok(session) = self.store.get_session(session_id).await {
            // Could add a recording_path field to SessionState if needed
            let _ = self.store.update_session(session).await;
        }

        Ok(recording_path)
    }

    /// Create a media bridge between two sessions (for peer-to-peer conferencing)
    pub async fn create_bridge(&self, session1: &SessionId, session2: &SessionId) -> Result<()> {
        // Get dialog IDs for both sessions
        let _dialog1 = self
            .session_to_dialog
            .get(session1)
            .ok_or_else(|| {
                SessionError::MediaError(format!("No media session for {}", session1.0))
            })?
            .clone();
        let _dialog2 = self
            .session_to_dialog
            .get(session2)
            .ok_or_else(|| {
                SessionError::MediaError(format!("No media session for {}", session2.0))
            })?
            .clone();

        tracing::info!(
            "Creating media bridge between {} and {}",
            session1.0,
            session2.0
        );

        // In a real implementation, this would configure the media relay to bridge RTP streams
        // For now, we'll just update the session states
        if let Ok(mut session1_state) = self.store.get_session(session1).await {
            session1_state.bridged_to = Some(session2.clone());
            let _ = self.store.update_session(session1_state).await;
        }

        if let Ok(mut session2_state) = self.store.get_session(session2).await {
            session2_state.bridged_to = Some(session1.clone());
            let _ = self.store.update_session(session2_state).await;
        }

        // Log bridge creation
        tracing::debug!("Bridge created between {} and {}", session1.0, session2.0);

        Ok(())
    }

    /// Swap the audio source on the running transmitter for a session.
    /// Used by early-media flows to replace silence with a ringback tone,
    /// hold announcement, or custom samples during `EarlyMedia`.
    ///
    /// The media session must already have an active transmitter — callers
    /// typically invoke this right after `send_early_media` (which has
    /// `PrepareEarlyMediaSDP` + `establish_media_flow` set one up).
    pub async fn set_audio_source(
        &self,
        session_id: &SessionId,
        source: AudioSource,
    ) -> Result<()> {
        let dialog_id = self
            .session_to_dialog
            .get(session_id)
            .ok_or_else(|| {
                SessionError::MediaError(format!("No media session for {}", session_id.0))
            })?
            .clone();

        self.controller
            .set_audio_source(&dialog_id, source)
            .await
            .map_err(|e| SessionError::MediaError(format!("Failed to set audio source: {}", e)))
    }

    /// Bridge the RTP streams of two sessions at the media-core layer.
    ///
    /// Resolves each `SessionId` to its underlying `DialogId` and delegates
    /// to `MediaSessionController::bridge_sessions`. Transparent packet-level
    /// relay — both legs must have negotiated the same codec and reached the
    /// `Active` state (remote RTP address known).
    ///
    /// Dropping the returned [`BridgeHandle`] tears the bridge down.
    pub async fn bridge_rtp_sessions(
        &self,
        session_a: &SessionId,
        session_b: &SessionId,
    ) -> std::result::Result<BridgeHandle, BridgeError> {
        let dialog_a = self
            .session_to_dialog
            .get(session_a)
            .ok_or_else(|| BridgeError::SessionNotFound(session_a.0.clone()))?
            .clone();
        let dialog_b = self
            .session_to_dialog
            .get(session_b)
            .ok_or_else(|| BridgeError::SessionNotFound(session_b.0.clone()))?
            .clone();

        let handle = self.controller.bridge_sessions(dialog_a, dialog_b).await?;

        // Keep the legacy session-store `bridged_to` pointers in sync so
        // anything that queries session state sees the pairing.
        if let Ok(mut a_state) = self.store.get_session(session_a).await {
            a_state.bridged_to = Some(session_b.clone());
            let _ = self.store.update_session(a_state).await;
        }
        if let Ok(mut b_state) = self.store.get_session(session_b).await {
            b_state.bridged_to = Some(session_a.clone());
            let _ = self.store.update_session(b_state).await;
        }

        Ok(handle)
    }

    /// Destroy a media bridge
    pub async fn destroy_bridge(&self, session_id: &SessionId) -> Result<()> {
        // Get the bridged session
        let bridged_session = if let Ok(session) = self.store.get_session(session_id).await {
            session.bridged_to.clone()
        } else {
            None
        };

        if let Some(other_session) = bridged_session {
            tracing::info!(
                "Destroying bridge between {} and {}",
                session_id.0,
                other_session.0
            );

            // Clear bridge information from both sessions
            if let Ok(mut session1_state) = self.store.get_session(session_id).await {
                session1_state.bridged_to = None;
                let _ = self.store.update_session(session1_state).await;
            }

            if let Ok(mut session2_state) = self.store.get_session(&other_session).await {
                session2_state.bridged_to = None;
                let _ = self.store.update_session(session2_state).await;
            }

            // Log bridge destruction
            tracing::debug!(
                "Bridge destroyed between {} and {}",
                session_id.0,
                other_session.0
            );
        }

        Ok(())
    }

    /// Stop recording the media session
    pub async fn stop_recording_old(&self, session_id: &SessionId) -> Result<()> {
        // Get the dialog ID for this session
        let _dialog_id = self
            .session_to_dialog
            .get(session_id)
            .ok_or_else(|| {
                SessionError::MediaError(format!("No media session for {}", session_id.0))
            })?
            .clone();

        tracing::info!("Stopping recording for session {}", session_id.0);

        // In a real implementation, this would stop recording through the media relay
        tracing::debug!("Recording stopped");

        Ok(())
    }

    // ===== AUDIO FRAME API - The Missing Core Functionality =====

    /// Send an audio frame for encoding and transmission
    /// This is the equivalent of the legacy MediaControl::send_audio_frame() API.
    pub async fn send_audio_frame(
        &self,
        session_id: &SessionId,
        audio_frame: AudioFrame,
    ) -> Result<()> {
        // Get the dialog ID for this session
        let dialog_id = self
            .session_to_dialog
            .get(session_id)
            .ok_or_else(|| {
                SessionError::MediaError(format!("No media session for {}", session_id.0))
            })?
            .clone();

        tracing::trace!(
            "📤 Sending audio frame for session {} ({} samples) via RTP",
            session_id.0,
            audio_frame.samples.len()
        );

        // Move the PCM buffer into media-core instead of cloning it. At soak
        // scale, the old clone created one extra Vec allocation per outbound
        // frame and showed up as sustained RSS/CPU pressure.
        let AudioFrame {
            samples: pcm_samples,
            timestamp,
            ..
        } = audio_frame;
        self.audio_send_frames_total.fetch_add(1, Ordering::Relaxed);
        self.audio_send_samples_total
            .fetch_add(pcm_samples.len() as u64, Ordering::Relaxed);

        self.controller
            .encode_and_send_audio_frame(&dialog_id, pcm_samples, timestamp)
            .await
            .map_err(|e| {
                SessionError::MediaError(format!("Failed to send audio frame via RTP: {}", e))
            })?;

        tracing::trace!(
            "✅ Audio frame sent successfully via RTP for session {}",
            session_id.0
        );
        Ok(())
    }

    /// Create a new media session
    pub async fn create_session(
        &self,
        session_id: &SessionId,
    ) -> Result<crate::types::MediaSessionId> {
        self.session_create_attempt_total
            .fetch_add(1, Ordering::Relaxed);
        // Create dialog ID for media-core
        let dialog_id = DialogId::new(format!("media-{}", session_id.0));

        tracing::info!(
            "🚀 Creating media session for session {} with dialog ID {}",
            session_id.0,
            dialog_id
        );

        // Store mappings
        self.session_to_dialog
            .insert(session_id.clone(), dialog_id.clone());
        self.dialog_to_session
            .insert(dialog_id.clone(), session_id.clone());

        if self.signaling_only_local_port().is_some() {
            tracing::info!(
                "signaling-only media mode: skipped media-core allocation for session {}",
                session_id.0
            );
            self.session_create_success_total
                .fetch_add(1, Ordering::Relaxed);
            return Ok(dialog_id);
        }

        // Create media config with our settings
        let media_config = MediaConfig {
            local_addr: SocketAddr::new(self.local_ip, 0), // Let media-core allocate port
            remote_addr: None,                             // Will be set when we get remote SDP
            preferred_codec: Some("PCMU".to_string()),     // G.711 µ-law as default
            parameters: std::collections::HashMap::new(),
        };

        // Start the media session in media-core
        self.controller
            .start_media(dialog_id.clone(), media_config)
            .await
            .map_err(|e| {
                self.session_create_failed_total
                    .fetch_add(1, Ordering::Relaxed);
                SessionError::MediaError(format!("Failed to start media session: {}", e))
            })?;

        // Install RFC 4733 DTMF callback so incoming PT 101 frames
        // surface as `Event::DtmfReceived` on the public API bus.
        // Fires once per digit (media-core dedupes the three §2.5.1.3
        // retransmits) so app consumers don't see duplicate digits.
        self.install_dtmf_callback(session_id.clone(), dialog_id.clone())
            .await;

        // Get and store session info
        if let Some(info) = self.controller.get_session_info(&dialog_id).await {
            self.media_sessions.insert(session_id.clone(), info.clone());

            // `MediaSessionId` is a type alias for
            // `rvoip_media_core::DialogId` (P5), so the value we hand
            // back to session-core is identical to the controller's
            // dialog id — no `from_dialog` reconstruction needed.
            // `store_session_mapping` separately wants media-core's
            // **internal** `MediaSessionId` type (still distinct), so
            // that conversion stays.
            let media_id = dialog_id.clone();
            self.controller.store_session_mapping(
                session_id.0.clone(),
                rvoip_media_core::MediaSessionId::from_dialog(&dialog_id),
            );

            tracing::info!(
                "✅ Media session created successfully for dialog {}",
                dialog_id
            );
            self.session_create_success_total
                .fetch_add(1, Ordering::Relaxed);
            return Ok(media_id);
        }

        self.session_create_failed_total
            .fetch_add(1, Ordering::Relaxed);
        Err(SessionError::MediaError(
            "Failed to get session info after creation".to_string(),
        ))
    }

    /// Generate the local SDP offer. **Sole** SDP-offer generator — this
    /// is the only entry point used by the state-machine's
    /// `Action::GenerateLocalSDP`. Always uses `SdpBuilder`, always
    /// advertises PCMU + PCMA + RFC 4733 telephone-event, and conditionally
    /// attaches `a=crypto:` lines when [`Config::offer_srtp`](crate::api::unified::Config) is set.
    ///
    /// Profile selection per RFC 4568 §3.1.4: `RTP/SAVP` when offering
    /// SDES, `RTP/AVP` otherwise. SRTP master keys are generated via
    /// [`SrtpNegotiator::new_offerer`] and stashed in
    /// `pending_srtp_offerers` keyed by `session_id` so
    /// [`Self::negotiate_sdp_as_uac`] can drive `accept_answer` against
    /// the matching answer.
    ///
    /// **DTMF advertisement (P2 fix).** Pre-Sprint 2.5 the non-SRTP path
    /// silently omitted PT 101 / `a=fmtp:101 0-15`, leaving plaintext
    /// callers unable to negotiate DTMF. The unified path always emits
    /// the telephone-event rtpmap + fmtp regardless of profile —
    /// `offer_advertises_telephone_event_on_plaintext` in the test
    /// module locks this in.
    pub async fn generate_local_sdp(&self, session_id: &SessionId) -> Result<String> {
        self.generate_local_sdp_offer(session_id, crate::types::MediaDirection::SendRecv)
            .await
    }

    /// Generate a local SDP offer for the requested media direction.
    pub async fn generate_local_sdp_offer(
        &self,
        session_id: &SessionId,
        direction: crate::types::MediaDirection,
    ) -> Result<String> {
        // Resolve dialog id and prime the cached session info. Both
        // the SRTP and plaintext paths used to do this; doing it once
        // up front lets the rest of the body share one shape.
        let dialog_id = self
            .session_to_dialog
            .get(session_id)
            .ok_or_else(|| {
                SessionError::SessionNotFound(format!(
                    "No dialog mapping for session {}",
                    session_id.0
                ))
            })?
            .clone();
        if self.signaling_only_local_port().is_some() {
            return self
                .generate_signaling_only_sdp_offer(session_id, direction)
                .await;
        }
        let info = self
            .controller
            .get_session_info(&dialog_id)
            .await
            .ok_or_else(|| {
                SessionError::MediaError(format!(
                    "Failed to get session info for dialog {}",
                    dialog_id
                ))
            })?;
        self.media_sessions.insert(session_id.clone(), info.clone());

        // Sprint 3 A6 — when a public RTP address has been configured
        // (static override or STUN-discovered), advertise that in the
        // SDP `c=` / `o=` / `m=audio` lines instead of the bind-address.
        // The static override's port wins when set; otherwise we keep
        // the per-session local RTP port (most NATs don't preserve
        // ports across the binding, but absent better info the local
        // port is our best guess and symmetric-RTP latching covers
        // the rest).
        let public = self.public_rtp_addr();
        let port = public
            .filter(|sa| sa.port() != 0)
            .map(|sa| sa.port())
            .unwrap_or_else(|| info.rtp_port.unwrap_or(info.config.local_addr.port()));
        let (origin_session_id, origin_version) = self.next_sdp_origin(session_id).await?;
        let origin_version = origin_version.to_string();
        let advertised_ip = public.map(|sa| sa.ip()).unwrap_or(self.local_ip);
        let local_ip_str = advertised_ip.to_string();

        // Profile + crypto. RFC 4568 §3.1.4 — `RTP/SAVP` is mandatory
        // when offering SDES.
        let (transport, crypto_attrs) = if self.offer_srtp {
            let (negotiator, attrs) = SrtpNegotiator::new_offerer(&self.srtp_offered_suites)?;
            self.pending_srtp_offerers
                .insert(session_id.clone(), negotiator);
            ("RTP/SAVP", attrs)
        } else {
            ("RTP/AVP", Vec::new())
        };
        let crypto_attr_count = crypto_attrs.len();
        if sdp_diagnostics_enabled() {
            emit_sdp_diag(format!(
                "local_sdp_offer session={} media={}:{} transport={} {} direction={}",
                session_id.0,
                advertised_ip,
                port,
                transport,
                crypto_attribute_diag(crypto_attr_count),
                direction_attribute(direction)
            ));
        }

        // NEXT_STEPS C2 — iterate the configured `offered_codecs` and
        // emit one rtpmap (+ fmtp where required) per PT, instead of
        // hard-coding PCMU/PCMA/DTMF. Comfort Noise (PT 13) is folded
        // in by `effective_offered_formats` so existing callers that
        // only flip the `comfort_noise_enabled` switch keep working.
        // Crypto attrs follow rtpmap/fmtp so ordering matches what
        // carriers expect; sendrecv goes last so the byte-fixture
        // tests stay stable.
        let format_pts = self.effective_offered_formats();
        let format_strings: Vec<String> = format_pts.iter().map(|pt| pt.to_string()).collect();
        let formats_ref: Vec<&str> = format_strings.iter().map(|s| s.as_str()).collect();
        let mut media_builder = SdpBuilder::new("Session")
            .origin(
                "-",
                &origin_session_id,
                &origin_version,
                "IN",
                "IP4",
                &local_ip_str,
            )
            .connection("IN", "IP4", &local_ip_str)
            .time("0", "0")
            .media_audio(port, transport)
            .formats(&formats_ref);
        for (pt, pt_str) in format_pts.iter().zip(format_strings.iter()) {
            if let Some(rtpmap) = rtpmap_for_pt(*pt) {
                media_builder = media_builder.rtpmap(pt_str.as_str(), rtpmap);
            }
            if let Some(fmtp) = fmtp_for_pt_with_g729_annex_b(*pt, self.g729_annex_b) {
                media_builder = media_builder.fmtp(pt_str.as_str(), fmtp);
            }
        }
        for attr in crypto_attrs {
            media_builder = media_builder.crypto_attribute(attr);
        }
        let session = media_builder
            .attribute(direction_attribute(direction), None::<String>)
            .done()
            .build()
            .map_err(|e| {
                SessionError::SDPNegotiationFailed(format!(
                    "SdpBuilder failed to build offer: {}",
                    e
                ))
            })?;

        let sdp = session.to_string();
        tracing::info!(
            "✅ Generated SDP for session {} with local port {} direction {}",
            session_id.0,
            port,
            direction_attribute(direction)
        );
        Ok(sdp)
    }

    /// Install the RFC 4733 DTMF bridge: registers a callback with
    /// media-core so PT 101 packets (already deduped to one-per-digit
    /// on the first end-of-event frame) are published as
    /// `Event::DtmfReceived { call_id, digit }` on the rvoip-sip
    /// public API event stream. No-op if the app event publisher/global
    /// coordinator has not been installed yet (e.g. isolated unit tests).
    async fn install_dtmf_callback(&self, session_id: SessionId, dialog_id: DialogId) {
        let publisher = self.app_event_publisher.read().await.clone();
        let coordinator = self.global_coordinator.read().await.clone();
        if publisher.is_none() && coordinator.is_none() {
            tracing::debug!(
                "DTMF callback install skipped for session {}: no event publisher yet",
                session_id.0
            );
            return;
        };

        let (tx, mut rx) = mpsc::channel::<rvoip_media_core::DtmfNotification>(32);
        if let Err(e) = self
            .controller
            .set_dtmf_callback(dialog_id.clone(), tx)
            .await
        {
            tracing::warn!(
                "Failed to register DTMF callback for session {} (dialog {}): {}",
                session_id.0,
                dialog_id,
                e
            );
            return;
        }

        // Consumer task: forwards each DTMF notification from media-core
        // onto the session-core API event bus. Exits cleanly when the
        // sender end of the channel is dropped (media session stopped).
        let sid = session_id.clone();
        let did = dialog_id.clone();
        spawn_memory_tracked("sip.media_adapter.dtmf_bridge_task", async move {
            while let Some(notification) = rx.recv().await {
                let api_event = crate::api::events::Event::DtmfReceived {
                    call_id: sid.clone(),
                    digit: notification.digit,
                };
                if let Some(publisher) = publisher.as_ref() {
                    publisher.publish(api_event);
                } else if let Some(coordinator) = coordinator.as_ref() {
                    let wrapped = crate::adapters::SessionApiCrossCrateEvent::new(api_event);
                    if let Err(e) = coordinator.publish(wrapped).await {
                        tracing::warn!(
                            "Failed to publish DtmfReceived for session {}: {}",
                            sid.0,
                            e
                        );
                    }
                }
                tracing::info!(
                    "📢 Published DtmfReceived digit='{}' for session {}",
                    notification.digit,
                    sid.0
                );
            }
            tracing::debug!(
                "DTMF bridge task exited for session {} (dialog {})",
                sid.0,
                did
            );
        });
    }

    /// Subscribe to receive decoded audio frames from RTP
    /// This is the equivalent of the legacy MediaControl::subscribe_to_audio_frames() API.
    pub async fn subscribe_to_audio_frames(
        &self,
        session_id: &SessionId,
    ) -> Result<crate::types::AudioFrameSubscriber> {
        // Get the dialog ID for this session
        let dialog_id = self
            .session_to_dialog
            .get(session_id)
            .ok_or_else(|| {
                SessionError::SessionNotFound(format!("No media session for {}", session_id.0))
            })?
            .clone();

        tracing::info!(
            "🎧 Setting up audio subscription for session {} (dialog: {})",
            session_id.0,
            dialog_id
        );

        // Keep decoded-audio buffering bounded for real-time media. At 50 fps,
        // 128 frames is ~2.5 seconds; the old 1000-frame buffer retained up to
        // 20 seconds of stale audio per active call.
        let (tx, rx) = mpsc::channel(AUDIO_RECEIVER_CHANNEL_FRAMES);

        // Register the callback with MediaSessionController to receive audio frames
        self.controller
            .set_audio_frame_callback(dialog_id.clone(), tx.clone())
            .await
            .map_err(|e| {
                SessionError::MediaError(format!("Failed to set audio callback: {}", e))
            })?;

        // Store the sender for this session for cleanup
        self.audio_receivers.insert(session_id.clone(), tx);
        self.audio_subscriber_created_total
            .fetch_add(1, Ordering::Relaxed);

        tracing::info!(
            "🎧 Created audio frame subscriber for session {} with dialog {}",
            session_id.0,
            dialog_id
        );

        // Return our types::AudioFrameSubscriber
        Ok(crate::types::AudioFrameSubscriber::new(
            session_id.clone(),
            rx,
        ))
    }

    /// Internal method to forward received audio frames to subscribers
    /// This should be called by the media event handler when audio frames are received
    #[allow(dead_code)]
    pub(crate) async fn forward_audio_frame_to_subscriber(
        &self,
        session_id: &SessionId,
        audio_frame: rvoip_media_core::types::AudioFrame,
    ) -> Result<()> {
        if let Some(tx) = self.audio_receivers.get(session_id) {
            if let Err(_) = tx.send(audio_frame).await {
                // Receiver has been dropped, clean up
                self.audio_receivers.remove(session_id);
                self.audio_subscriber_disconnected_total
                    .fetch_add(1, Ordering::Relaxed);
                tracing::debug!(
                    "Audio frame subscriber disconnected for session {}",
                    session_id.0
                );
            }
        }
        Ok(())
    }

    // ===== New Methods for CallController and ConferenceManager =====

    /// Create a media session
    pub async fn create_media_session(&self) -> Result<crate::types::MediaSessionId> {
        let media_id = crate::types::MediaSessionId::new_v4();
        Ok(media_id)
    }

    /// Stop a media session
    pub async fn stop_media_session(&self, _media_id: crate::types::MediaSessionId) -> Result<()> {
        // For now, just return Ok
        Ok(())
    }

    /// Set media direction (for hold/resume)
    pub async fn set_media_direction(
        &self,
        media_id: crate::types::MediaSessionId,
        direction: crate::types::MediaDirection,
    ) -> Result<()> {
        let media_direction = match direction {
            crate::types::MediaDirection::SendRecv => {
                rvoip_media_core::types::MediaDirection::SendRecv
            }
            crate::types::MediaDirection::SendOnly => {
                rvoip_media_core::types::MediaDirection::SendOnly
            }
            crate::types::MediaDirection::RecvOnly => {
                rvoip_media_core::types::MediaDirection::RecvOnly
            }
            crate::types::MediaDirection::Inactive => {
                rvoip_media_core::types::MediaDirection::Inactive
            }
        };
        self.controller
            .set_media_direction(&media_id, media_direction)
            .await
            .map_err(|e| SessionError::MediaError(format!("Failed to set media direction: {}", e)))
    }

    /// Create hold SDP for an established session.
    pub async fn create_hold_sdp_for_session(&self, session_id: &SessionId) -> Result<String> {
        self.generate_local_sdp_offer(session_id, crate::types::MediaDirection::SendOnly)
            .await
    }

    /// Create active SDP for an established session.
    pub async fn create_active_sdp_for_session(&self, session_id: &SessionId) -> Result<String> {
        self.generate_local_sdp_offer(session_id, crate::types::MediaDirection::SendRecv)
            .await
    }

    /// Create hold SDP without a session context.
    ///
    /// Prefer [`Self::create_hold_sdp_for_session`]. This fallback exists for
    /// older internal call sites and tests; it advertises the configured media
    /// start port rather than disabling the m-line.
    pub async fn create_hold_sdp(&self) -> Result<String> {
        self.create_directional_fallback_sdp(crate::types::MediaDirection::SendOnly)
    }

    /// Create active SDP without a session context.
    ///
    /// Prefer [`Self::create_active_sdp_for_session`].
    pub async fn create_active_sdp(&self) -> Result<String> {
        self.create_directional_fallback_sdp(crate::types::MediaDirection::SendRecv)
    }

    fn create_directional_fallback_sdp(
        &self,
        direction: crate::types::MediaDirection,
    ) -> Result<String> {
        let local_ip = self.local_ip.to_string();
        let formats: &[&str] = if self.comfort_noise_enabled {
            &["0", "8", "13", "101"]
        } else {
            &["0", "8", "101"]
        };
        let mut media_builder = SdpBuilder::new("Session")
            .origin("-", "0", "0", "IN", "IP4", &local_ip)
            .connection("IN", "IP4", &local_ip)
            .time("0", "0")
            .media_audio(self.media_port_start, "RTP/AVP")
            .formats(formats)
            .rtpmap("0", "PCMU/8000")
            .rtpmap("8", "PCMA/8000");
        if self.comfort_noise_enabled {
            media_builder = media_builder.rtpmap("13", "CN/8000");
        }
        let session = media_builder
            .rtpmap("101", "telephone-event/8000")
            .fmtp("101", "0-15")
            .attribute(direction_attribute(direction), None::<String>)
            .done()
            .build()
            .map_err(|e| {
                SessionError::SDPNegotiationFailed(format!(
                    "SdpBuilder failed to build {} SDP: {}",
                    direction_attribute(direction),
                    e
                ))
            })?;
        Ok(session.to_string())
    }

    async fn next_sdp_origin(&self, session_id: &SessionId) -> Result<(String, u64)> {
        let mut session = self.store.get_session(session_id).await.map_err(|e| {
            SessionError::SessionNotFound(format!(
                "No session state for SDP origin {}: {}",
                session_id.0, e
            ))
        })?;
        if session.sdp_origin_session_id.is_empty() {
            session.sdp_origin_session_id = sdp_origin_session_id(&session.session_id.0);
        }
        session.sdp_origin_version = session.sdp_origin_version.saturating_add(1);
        let out = (
            session.sdp_origin_session_id.clone(),
            session.sdp_origin_version,
        );
        self.store.update_session(session).await.map_err(|e| {
            SessionError::MediaError(format!("Failed to persist SDP origin version: {}", e))
        })?;
        Ok(out)
    }

    /// Send DTMF digit (legacy `media_id` signature used by the state
    /// machine's CallController path). Delegates to
    /// [`Self::send_dtmf_rfc4733`] with a 100 ms duration.
    pub async fn send_dtmf(
        &self,
        media_id: crate::types::MediaSessionId,
        digit: char,
    ) -> Result<()> {
        // `MediaSessionId` is now a type alias for media-core's
        // `DialogId` (P5), so the value passed in IS the dialog id we
        // need — no reconstruction.
        let dialog_id = media_id;
        self.controller
            .send_dtmf_packet(&dialog_id, digit, 100)
            .await
            .map_err(|e| SessionError::MediaError(format!("DTMF send failed: {}", e)))?;
        tracing::debug!("☎️  Queued DTMF '{}' for media_id {:?}", digit, dialog_id);
        Ok(())
    }

    /// Send RFC 4733 DTMF by session id — preferred public API, used
    /// by [`UnifiedCoordinator::send_dtmf`](crate::api::unified::UnifiedCoordinator::send_dtmf).
    pub async fn send_dtmf_rfc4733(
        &self,
        session_id: &SessionId,
        digit: char,
        duration_ms: u32,
    ) -> Result<()> {
        let dialog_id = self
            .session_to_dialog
            .get(session_id)
            .ok_or_else(|| {
                SessionError::SessionNotFound(format!("No media session for {}", session_id.0))
            })?
            .clone();

        self.controller
            .send_dtmf_packet(&dialog_id, digit, duration_ms)
            .await
            .map_err(|e| SessionError::MediaError(format!("DTMF send failed: {}", e)))?;

        tracing::info!(
            "☎️  Queued DTMF '{}' for session {} (dialog {}, duration={}ms)",
            digit,
            session_id.0,
            dialog_id,
            duration_ms
        );
        Ok(())
    }

    /// Set mute state
    pub async fn set_mute(
        &self,
        media_id: crate::types::MediaSessionId,
        muted: bool,
    ) -> Result<()> {
        // TODO: Implement mute control
        tracing::debug!(
            "Setting mute state to {} for media session {:?}",
            muted,
            media_id
        );
        Ok(())
    }

    /// Start recording for media session
    pub async fn start_recording_media(
        &self,
        media_id: crate::types::MediaSessionId,
    ) -> Result<()> {
        // TODO: Implement recording
        tracing::info!("Starting recording for media session {:?}", media_id);
        Ok(())
    }

    /// Stop recording for media session
    pub async fn stop_recording_media(&self, media_id: crate::types::MediaSessionId) -> Result<()> {
        // TODO: Implement recording stop
        tracing::info!("Stopping recording for media session {:?}", media_id);
        Ok(())
    }

    // ===== Conference Methods =====

    /// Create an audio mixer for a conference
    pub async fn create_audio_mixer(&self) -> Result<crate::types::MediaSessionId> {
        let mixer_id = crate::types::MediaSessionId::new_v4();
        self.audio_mixers.insert(mixer_id.clone(), Vec::new());
        tracing::info!("Created audio mixer {:?}", mixer_id);
        Ok(mixer_id)
    }

    /// Redirect audio to a mixer
    pub async fn redirect_to_mixer(
        &self,
        media_id: crate::types::MediaSessionId,
        mixer_id: crate::types::MediaSessionId,
    ) -> Result<()> {
        if let Some(mut mixer) = self.audio_mixers.get_mut(&mixer_id) {
            mixer.push(media_id.clone());
        }
        tracing::debug!("Redirected media {:?} to mixer {:?}", media_id, mixer_id);
        Ok(())
    }

    /// Remove audio from a mixer
    pub async fn remove_from_mixer(
        &self,
        media_id: crate::types::MediaSessionId,
        mixer_id: crate::types::MediaSessionId,
    ) -> Result<()> {
        if let Some(mut mixer) = self.audio_mixers.get_mut(&mixer_id) {
            mixer.retain(|id| id != &media_id);
        }
        tracing::debug!("Removed media {:?} from mixer {:?}", media_id, mixer_id);
        Ok(())
    }

    /// Destroy an audio mixer
    pub async fn destroy_mixer(&self, mixer_id: crate::types::MediaSessionId) -> Result<()> {
        self.audio_mixers.remove(&mixer_id);
        tracing::info!("Destroyed audio mixer {:?}", mixer_id);
        Ok(())
    }

    /// Clean up all mappings and resources for a session.
    ///
    /// Idempotent — safe to call multiple times. Always removes the audio
    /// frame callback from media-core (so subscriber `rx.recv()` calls can
    /// return `None` and exit their loops) as long as the dialog mapping is
    /// still present.
    pub async fn cleanup_session(&self, session_id: &SessionId) -> Result<()> {
        let guard = cleanup_diag::stage_guard(CleanupStage::MediaCleanup, &session_id.0);
        self.cleanup_attempt_total.fetch_add(1, Ordering::Relaxed);
        // Resolve dialog_id — prefer the mapping populated by create_session
        // (which is authoritative once set) but fall back to the deterministic
        // form `media-<session_id>` when the mapping has been lost (e.g. on a
        // second CleanupMedia after a Dialog3xxRedirect transition's actions
        // already cleared it). The fallback ensures media-core's sessions map
        // is always freed so a subsequent CreateMediaSession can reuse the
        // same dialog_id.
        let removed = self.session_to_dialog.remove(session_id);
        let dialog_id = match removed {
            Some((_, d)) => {
                self.cleanup_mapped_total.fetch_add(1, Ordering::Relaxed);
                Some(d)
            }
            None => {
                // Fallback to deterministic form used by create_session.
                self.cleanup_fallback_total.fetch_add(1, Ordering::Relaxed);
                Some(DialogId::new(format!("media-{}", session_id.0)))
            }
        };

        if let Some(dialog_id) = dialog_id {
            if self.signaling_only_local_port().is_none() {
                let _ = self
                    .controller
                    .remove_audio_frame_callback(&dialog_id)
                    .await;
                // stop_media may return "session not found" on the fallback path
                // (media-core already cleaned up) — expected; ignore.
                let _ = self.controller.stop_media(&dialog_id).await;
            }
            self.dialog_to_session.remove(&dialog_id);
        }

        if self.media_sessions.remove(session_id).is_some() {
            self.cleanup_media_session_removed_total
                .fetch_add(1, Ordering::Relaxed);
        }
        // Drop our own clone of the tx as well (the other lived in media-core).
        if self.audio_receivers.remove(session_id).is_some() {
            self.cleanup_audio_receiver_removed_total
                .fetch_add(1, Ordering::Relaxed);
        }

        // NEXT_STEPS B diag — bump a process-global counter so the
        // perf_listener example can poll cleanup throughput without
        // grepping logs at 100+ msg/sec. Cleared & checked in the
        // dockerized sipp comparison harness.
        cleanup_session_diag::record_cleanup();

        tracing::debug!(
            "Cleaned up media adapter resources for session {}",
            session_id.0
        );
        guard.finish_success();
        Ok(())
    }

    // ===== Helper Methods =====

    /// Get local RTP port for a session
    fn get_local_port(&self, session_id: &SessionId) -> Result<u16> {
        self.media_sessions
            .get(session_id)
            .and_then(|info| info.rtp_port)
            .ok_or_else(|| {
                SessionError::SessionNotFound(format!("No local port for session {}", session_id.0))
            })
    }

    async fn generate_signaling_only_sdp_offer(
        &self,
        session_id: &SessionId,
        direction: crate::types::MediaDirection,
    ) -> Result<String> {
        let (origin_session_id, origin_version) = self.next_sdp_origin(session_id).await?;
        let origin_version = origin_version.to_string();
        let advertised_ip = self
            .public_rtp_addr()
            .map(|sa| sa.ip())
            .unwrap_or(self.local_ip);
        let local_ip_str = advertised_ip.to_string();
        let port = self.signaling_only_local_port().ok_or_else(|| {
            SessionError::MediaError(
                "signaling-only SDP requested while media mode is enabled".to_string(),
            )
        })?;
        let (transport, crypto_attrs) = if self.offer_srtp {
            let (negotiator, attrs) = SrtpNegotiator::new_offerer(&self.srtp_offered_suites)?;
            self.pending_srtp_offerers
                .insert(session_id.clone(), negotiator);
            ("RTP/SAVP", attrs)
        } else {
            ("RTP/AVP", Vec::new())
        };

        let format_pts = self.effective_offered_formats();
        let format_strings: Vec<String> = format_pts.iter().map(|pt| pt.to_string()).collect();
        let formats_ref: Vec<&str> = format_strings.iter().map(|s| s.as_str()).collect();
        let mut media_builder = SdpBuilder::new("Session")
            .origin(
                "-",
                &origin_session_id,
                &origin_version,
                "IN",
                "IP4",
                &local_ip_str,
            )
            .connection("IN", "IP4", &local_ip_str)
            .time("0", "0")
            .media_audio(port, transport)
            .formats(&formats_ref);
        for (pt, pt_str) in format_pts.iter().zip(format_strings.iter()) {
            if let Some(rtpmap) = rtpmap_for_pt(*pt) {
                media_builder = media_builder.rtpmap(pt_str.as_str(), rtpmap);
            }
            if let Some(fmtp) = fmtp_for_pt_with_g729_annex_b(*pt, self.g729_annex_b) {
                media_builder = media_builder.fmtp(pt_str.as_str(), fmtp);
            }
        }
        for attr in crypto_attrs {
            media_builder = media_builder.crypto_attribute(attr);
        }
        let session = media_builder
            .attribute(direction_attribute(direction), None::<String>)
            .done()
            .build()
            .map_err(|e| {
                SessionError::SDPNegotiationFailed(format!(
                    "SdpBuilder failed to build signaling-only offer: {}",
                    e
                ))
            })?;

        tracing::info!(
            "signaling-only media mode: generated SDP for session {} with advertised port {}",
            session_id.0,
            port
        );
        Ok(session.to_string())
    }

    fn signaling_only_local_port(&self) -> Option<u16> {
        match self.media_mode {
            MediaMode::Enabled => None,
            MediaMode::SignalingOnly { sdp_rtp_port } => Some(
                self.public_rtp_addr()
                    .filter(|addr| addr.port() != 0)
                    .map(|addr| addr.port())
                    .unwrap_or(sdp_rtp_port),
            ),
        }
    }

    /// Parse SDP to extract connection info from the audio m= section.
    ///
    /// Uses sip-core's typed `SdpSession::from_str` parser instead of
    /// the previous bespoke line-scanner so that future SDP work
    /// (`a=crypto:` for SDES, `a=fingerprint:`/`a=setup:` for
    /// DTLS-SRTP, video m= sections, RFC 8866 conformance) gets
    /// validation for free.
    ///
    /// Per RFC 8866 §5.7 the m-section's own `c=` line (if present)
    /// overrides the session-level `c=`. We honour that.
    fn parse_sdp_connection(&self, sdp: &str) -> Result<(IpAddr, u16)> {
        let session = SdpSession::from_str(sdp).map_err(|e| {
            SessionError::SDPNegotiationFailed(format!("Failed to parse SDP: {}", e))
        })?;

        let media = session
            .media_descriptions
            .iter()
            .find(|m| m.media == "audio")
            .ok_or_else(|| {
                SessionError::SDPNegotiationFailed("SDP has no audio m= section".into())
            })?;

        let port = media.port as u16;

        // Prefer the per-media c= line; fall back to session-level.
        let conn = media
            .connection_info
            .as_ref()
            .or(session.connection_info.as_ref())
            .ok_or_else(|| {
                SessionError::SDPNegotiationFailed(
                    "SDP has no c= line at session or audio level".into(),
                )
            })?;

        let ip = match &conn.connection_address {
            host_str => host_str.parse::<IpAddr>().map_err(|e| {
                SessionError::SDPNegotiationFailed(format!(
                    "SDP c= address {:?} is not a valid IP: {}",
                    host_str, e
                ))
            })?,
        };

        Ok((ip, port))
    }

    // ===== Event handling removed - now centralized in SessionCrossCrateEventHandler ====="

    // ===== Recording Management =====

    /// Start recording for a session (simple version for backward compatibility)
    pub async fn start_recording(&self, session_id: &SessionId) -> Result<String> {
        // Use default config for backward compatibility
        let config = RecordingConfig::default();
        self.start_recording_with_config(session_id, config).await
    }

    /// Start recording for a session with specific config
    pub async fn start_recording_with_config(
        &self,
        session_id: &SessionId,
        config: RecordingConfig,
    ) -> Result<String> {
        // Get dialog ID
        let _dialog_id = self
            .session_to_dialog
            .get(session_id)
            .ok_or_else(|| {
                SessionError::MediaError(format!("No dialog for session {}", session_id.0))
            })?
            .clone();

        tracing::info!(
            "Starting recording for session {} with path: {}",
            session_id.0,
            config.file_path
        );

        // For now, we'll generate a simple recording ID
        // In a real implementation, this would interact with media-core's recording API
        let recording_id = format!("rec_{}_{}", session_id.0, chrono::Utc::now().timestamp());

        // TODO: When media-core adds recording support, implement this properly
        // self.controller.start_recording(&dialog_id, config).await

        tracing::info!(
            "✅ Recording started for session {} with ID: {}",
            session_id.0,
            recording_id
        );
        Ok(recording_id)
    }

    /// Stop recording for a session (simple version for backward compatibility)
    pub async fn stop_recording(&self, session_id: &SessionId) -> Result<()> {
        // For backward compatibility, we don't have a recording_id
        // Just log the action
        tracing::info!("Stopping recording for session {}", session_id.0);
        Ok(())
    }

    /// Stop recording for a session with specific recording ID
    pub async fn stop_recording_with_id(
        &self,
        session_id: &SessionId,
        recording_id: &str,
    ) -> Result<()> {
        // Get dialog ID
        let _dialog_id = self
            .session_to_dialog
            .get(session_id)
            .ok_or_else(|| {
                SessionError::MediaError(format!("No dialog for session {}", session_id.0))
            })?
            .clone();

        tracing::info!(
            "Stopping recording {} for session {}",
            recording_id,
            session_id.0
        );

        // TODO: When media-core adds recording support, implement this properly
        // self.controller.stop_recording(&dialog_id, recording_id).await

        tracing::info!("✅ Recording stopped for session {}", session_id.0);
        Ok(())
    }

    /// Pause recording for a session
    pub async fn pause_recording(&self, session_id: &SessionId, recording_id: &str) -> Result<()> {
        // Get dialog ID
        let _dialog_id = self
            .session_to_dialog
            .get(session_id)
            .ok_or_else(|| {
                SessionError::MediaError(format!("No dialog for session {}", session_id.0))
            })?
            .clone();

        tracing::debug!(
            "Pausing recording {} for session {}",
            recording_id,
            session_id.0
        );

        // TODO: When media-core adds recording support, implement this properly
        // self.controller.pause_recording(&dialog_id, recording_id).await

        Ok(())
    }

    /// Resume a paused recording
    pub async fn resume_recording(&self, session_id: &SessionId, recording_id: &str) -> Result<()> {
        // Get dialog ID
        let _dialog_id = self
            .session_to_dialog
            .get(session_id)
            .ok_or_else(|| {
                SessionError::MediaError(format!("No dialog for session {}", session_id.0))
            })?
            .clone();

        tracing::debug!(
            "Resuming recording {} for session {}",
            recording_id,
            session_id.0
        );

        // TODO: When media-core adds recording support, implement this properly
        // self.controller.resume_recording(&dialog_id, recording_id).await

        Ok(())
    }

    /// Get recording status
    pub async fn get_recording_status(
        &self,
        session_id: &SessionId,
        _recording_id: &str,
    ) -> Result<RecordingStatus> {
        // Get dialog ID
        let _dialog_id = self
            .session_to_dialog
            .get(session_id)
            .ok_or_else(|| {
                SessionError::MediaError(format!("No dialog for session {}", session_id.0))
            })?
            .clone();

        // TODO: When media-core adds recording support, implement this properly
        // For now, return a mock status
        Ok(RecordingStatus {
            is_recording: false,
            is_paused: false,
            duration_seconds: 0.0,
            file_size_bytes: 0,
        })
    }

    /// Start recording for a bridged session pair
    pub async fn start_bridge_recording(
        &self,
        session1: &SessionId,
        session2: &SessionId,
        config: RecordingConfig,
    ) -> Result<String> {
        tracing::info!(
            "Starting bridge recording for sessions {} <-> {}",
            session1.0,
            session2.0
        );

        // Start recording on both sessions with mixed audio
        let mut recording_config = config;
        recording_config.include_mixed = true;
        recording_config.separate_tracks = true;

        // Start recording on the first session (will capture both legs if bridged)
        self.start_recording_with_config(session1, recording_config)
            .await
    }

    /// Enable/disable recording for all conference sessions
    pub async fn set_conference_recording_enabled(&self, enabled: bool) -> Result<()> {
        // This would be stored in a shared configuration
        // For now, we'll just log the intent
        tracing::info!("Conference recording enabled: {}", enabled);

        // TODO: Store this in a shared configuration that conference sessions check
        // when they are created

        Ok(())
    }
}

impl Clone for MediaAdapter {
    fn clone(&self) -> Self {
        Self {
            controller: self.controller.clone(),
            store: self.store.clone(),
            session_to_dialog: self.session_to_dialog.clone(),
            dialog_to_session: self.dialog_to_session.clone(),
            media_sessions: self.media_sessions.clone(),
            audio_receivers: self.audio_receivers.clone(),
            session_create_attempt_total: self.session_create_attempt_total.clone(),
            session_create_success_total: self.session_create_success_total.clone(),
            session_create_failed_total: self.session_create_failed_total.clone(),
            cleanup_attempt_total: self.cleanup_attempt_total.clone(),
            cleanup_mapped_total: self.cleanup_mapped_total.clone(),
            cleanup_fallback_total: self.cleanup_fallback_total.clone(),
            cleanup_media_session_removed_total: self.cleanup_media_session_removed_total.clone(),
            cleanup_audio_receiver_removed_total: self.cleanup_audio_receiver_removed_total.clone(),
            audio_subscriber_created_total: self.audio_subscriber_created_total.clone(),
            audio_subscriber_disconnected_total: self.audio_subscriber_disconnected_total.clone(),
            audio_send_frames_total: self.audio_send_frames_total.clone(),
            audio_send_samples_total: self.audio_send_samples_total.clone(),
            audio_mixers: self.audio_mixers.clone(),
            local_ip: self.local_ip,
            media_port_start: self.media_port_start,
            media_port_end: self.media_port_end,
            media_mode: self.media_mode,
            offer_srtp: self.offer_srtp,
            srtp_required: self.srtp_required,
            srtp_offered_suites: self.srtp_offered_suites.clone(),
            pending_srtp_offerers: self.pending_srtp_offerers.clone(),
            negotiated_srtp: self.negotiated_srtp.clone(),
            global_coordinator: self.global_coordinator.clone(),
            app_event_publisher: self.app_event_publisher.clone(),
            public_rtp_addr: std::sync::RwLock::new(self.public_rtp_addr()),
            comfort_noise_enabled: self.comfort_noise_enabled,
            strict_codec_matching: self.strict_codec_matching,
            offered_codecs: self.offered_codecs.clone(),
            g729_annex_b: self.g729_annex_b,
        }
    }
}

/// Sprint 3.5 — compute the answer's `m=audio` format list from the
/// offer + our policy flags. Pure (no `MediaAdapter` state) so unit
/// tests can exercise the strict-vs-permissive logic without standing
/// up a coordinator.
///
/// Returns the formats in the order they should appear on the wire.
/// Caller is responsible for emitting the matching `a=rtpmap:` /
/// `a=fmtp:` lines.
///
/// `Err(SDPNegotiationFailed)` when:
/// - Strict mode + offer carries no overlap with our supported set
///   → state machine surfaces this as `488 Not Acceptable Here`.
/// - Strict mode + matcher rejects on SRTP policy (e.g. `require_srtp`
///   set + offer is plain RTP/AVP).
pub(crate) fn compute_answer_formats(
    offer: &SdpSession,
    offered_codecs: &[u8],
    strict: bool,
    offer_srtp: bool,
    srtp_required: bool,
) -> Result<Vec<String>> {
    let supported: Vec<String> = offered_codecs.iter().map(|pt| pt.to_string()).collect();

    if !strict {
        // Permissive — answer with our full set regardless. Matches
        // the pre-Sprint-3.5 shape.
        return Ok(supported);
    }

    let caps = rvoip_sip_dialog::sdp::AnswerCapabilities {
        supported_formats: supported,
        accept_srtp: offer_srtp,
        require_srtp: srtp_required,
    };
    let m = rvoip_sip_dialog::sdp::match_offer(offer, &caps)
        .map_err(|e| SessionError::SDPNegotiationFailed(format!("{}", e)))?;
    let line = m
        .media_lines
        .iter()
        .find(|l| l.media == "audio")
        .ok_or_else(|| {
            SessionError::SDPNegotiationFailed("matcher returned no audio media line".into())
        })?;
    if !line.accepted {
        return Err(SessionError::SDPNegotiationFailed(
            "no codec overlap with offer".into(),
        ));
    }
    Ok(line.negotiated_formats.clone())
}

#[cfg(test)]
mod sdp_format_tests {
    //! Byte-fixture regression tests for the format-strings →
    //! `SdpBuilder` refactor (Step 2B.1, decision D11). Builds the same
    //! SDP via the typed builder and asserts byte-identical output to
    //! what the previous `format!` block would have produced.
    //!
    //! When the SRTP offer landing in 2B.2 changes the m= transport to
    //! `RTP/SAVP` and adds `a=crypto:` lines, these tests will need a
    //! second fixture for that case.

    use super::*;

    #[test]
    fn offer_direction_maps_to_correct_answer_direction() {
        use rvoip_sip_core::MediaDirection as SipDirection;

        assert_eq!(
            answer_direction_for_offer(&Some(SipDirection::SendRecv)),
            crate::types::MediaDirection::SendRecv
        );
        assert_eq!(
            answer_direction_for_offer(&Some(SipDirection::SendOnly)),
            crate::types::MediaDirection::RecvOnly
        );
        assert_eq!(
            answer_direction_for_offer(&Some(SipDirection::RecvOnly)),
            crate::types::MediaDirection::SendOnly
        );
        assert_eq!(
            answer_direction_for_offer(&Some(SipDirection::Inactive)),
            crate::types::MediaDirection::Inactive
        );
        assert_eq!(
            answer_direction_for_offer(&None),
            crate::types::MediaDirection::SendRecv
        );
    }

    #[test]
    fn remote_answer_direction_maps_to_local_direction() {
        use rvoip_sip_core::MediaDirection as SipDirection;

        assert_eq!(
            local_direction_from_remote_answer(&Some(SipDirection::SendRecv)),
            crate::types::MediaDirection::SendRecv
        );
        assert_eq!(
            local_direction_from_remote_answer(&Some(SipDirection::SendOnly)),
            crate::types::MediaDirection::RecvOnly
        );
        assert_eq!(
            local_direction_from_remote_answer(&Some(SipDirection::RecvOnly)),
            crate::types::MediaDirection::SendOnly
        );
        assert_eq!(
            local_direction_from_remote_answer(&Some(SipDirection::Inactive)),
            crate::types::MediaDirection::Inactive
        );
        assert_eq!(
            local_direction_from_remote_answer(&None),
            crate::types::MediaDirection::SendRecv
        );
    }

    #[test]
    fn g729_fmtp_reflects_annex_b_flag() {
        assert_eq!(fmtp_for_pt_with_g729_annex_b(18, true), Some("annexb=yes"));
        assert_eq!(fmtp_for_pt_with_g729_annex_b(18, false), Some("annexb=no"));
        assert_eq!(fmtp_for_pt(101), Some("0-15"));
    }

    #[test]
    fn g729_annex_b_negotiation_honors_remote_no() {
        let sdp = "v=0\r\n\
o=- 1 1 IN IP4 127.0.0.1\r\n\
s=Session\r\n\
c=IN IP4 127.0.0.1\r\n\
t=0 0\r\n\
m=audio 17000 RTP/AVP 18 101\r\n\
a=rtpmap:18 G729/8000\r\n\
a=fmtp:18 annexb=no\r\n\
a=rtpmap:101 telephone-event/8000\r\n\
a=fmtp:101 0-15\r\n";
        let session = SdpSession::from_str(sdp).expect("sdp parses");

        assert!(!negotiated_g729_annex_b(&session, true));
        assert!(!negotiated_g729_annex_b(&session, false));
        assert_eq!(
            select_primary_audio_payload_from_session(&session),
            Some(18)
        );
        assert_eq!(codec_name_for_payload(18, false), "G729A");
    }

    #[test]
    fn audio_direction_prefers_media_level_direction() {
        use rvoip_sip_core::MediaDirection as SipDirection;

        let sdp = SdpBuilder::new("Session")
            .origin("-", "1", "0", "IN", "IP4", "127.0.0.1")
            .connection("IN", "IP4", "127.0.0.1")
            .time("0", "0")
            .direction(SipDirection::SendRecv)
            .media_audio(16000, "RTP/AVP")
            .formats(&["0"])
            .rtpmap("0", "PCMU/8000")
            .direction(SipDirection::SendOnly)
            .done()
            .build()
            .expect("offer builds");

        assert_eq!(audio_direction(&sdp), Some(SipDirection::SendOnly));
    }

    /// Build the offer the same way `generate_local_sdp` does, but with
    /// fixed inputs so the output is deterministic. Mirrors the
    /// production shape: PCMU + PCMA + telephone-event, RTP/AVP profile.
    fn build_offer(dialog_id: &str, elapsed_secs: u64, ip: &str, port: u16) -> String {
        let elapsed = elapsed_secs.to_string();
        let origin_session_id = sdp_origin_session_id(dialog_id);
        SdpBuilder::new("Session")
            .origin("-", &origin_session_id, &elapsed, "IN", "IP4", ip)
            .connection("IN", "IP4", ip)
            .time("0", "0")
            .media_audio(port, "RTP/AVP")
            .formats(&["0", "8", "101"])
            .rtpmap("0", "PCMU/8000")
            .rtpmap("8", "PCMA/8000")
            .rtpmap("101", "telephone-event/8000")
            .fmtp("101", "0-15")
            .attribute("sendrecv", None::<String>)
            .done()
            .build()
            .expect("offer builds")
            .to_string()
    }

    /// Same as `build_offer` but for the answer (different sess_version).
    fn build_answer(sess_id: &str, ip: &str, port: u16) -> String {
        SdpBuilder::new("Session")
            .origin("-", sess_id, "0", "IN", "IP4", ip)
            .connection("IN", "IP4", ip)
            .time("0", "0")
            .media_audio(port, "RTP/AVP")
            .formats(&["0", "8", "101"])
            .rtpmap("0", "PCMU/8000")
            .rtpmap("8", "PCMA/8000")
            .rtpmap("101", "telephone-event/8000")
            .fmtp("101", "0-15")
            .attribute("sendrecv", None::<String>)
            .done()
            .build()
            .expect("answer builds")
            .to_string()
    }

    /// Reference output for the unified offer (Sprint 2.5 P2): PCMU +
    /// PCMA + telephone-event on every offer regardless of SRTP. Pre-P2
    /// the non-SRTP path emitted only `0 8` (no DTMF) and the SRTP path
    /// only `0 101` (no PCMA); both have been merged into the unified
    /// shape below.
    fn legacy_offer(dialog_id: &str, elapsed_secs: u64, ip: &str, port: u16) -> String {
        let origin_session_id = sdp_origin_session_id(dialog_id);
        format!(
            "v=0\r\n\
             o=- {} {} IN IP4 {}\r\n\
             s=Session\r\n\
             c=IN IP4 {}\r\n\
             t=0 0\r\n\
             m=audio {} RTP/AVP 0 8 101\r\n\
             a=rtpmap:0 PCMU/8000\r\n\
             a=rtpmap:8 PCMA/8000\r\n\
             a=rtpmap:101 telephone-event/8000\r\n\
             a=fmtp:101 0-15\r\n\
             a=sendrecv\r\n",
            origin_session_id, elapsed_secs, ip, ip, port,
        )
    }

    fn legacy_answer(sess_id: &str, ip: &str, port: u16) -> String {
        format!(
            "v=0\r\n\
             o=- {} {} IN IP4 {}\r\n\
             s=Session\r\n\
             c=IN IP4 {}\r\n\
             t=0 0\r\n\
             m=audio {} RTP/AVP 0 8 101\r\n\
             a=rtpmap:0 PCMU/8000\r\n\
             a=rtpmap:8 PCMA/8000\r\n\
             a=rtpmap:101 telephone-event/8000\r\n\
             a=fmtp:101 0-15\r\n\
             a=sendrecv\r\n",
            sess_id, 0u64, ip, ip, port,
        )
    }

    #[test]
    fn offer_matches_legacy_format_byte_for_byte() {
        let dialog_id = "test-dialog-uuid";
        let elapsed = 42u64;
        let ip = "127.0.0.1";
        let port = 16000;
        let new = build_offer(dialog_id, elapsed, ip, port);
        let old = legacy_offer(dialog_id, elapsed, ip, port);
        assert_eq!(
            new, old,
            "SdpBuilder offer drifted from legacy format-string output"
        );
    }

    #[test]
    fn answer_matches_legacy_format_byte_for_byte() {
        let sess_id = "1234567890";
        let ip = "192.168.1.42";
        let port = 16002;
        let new = build_answer(sess_id, ip, port);
        let old = legacy_answer(sess_id, ip, port);
        assert_eq!(
            new, old,
            "SdpBuilder answer drifted from legacy format-string output"
        );
    }

    #[test]
    fn offer_round_trips_through_typed_parser() {
        // Build → parse → assert key fields. Catches CRLF / spacing
        // issues that would also break peer interop.
        let sdp_str = build_offer("d", 0, "10.0.0.1", 5004);
        let parsed = SdpSession::from_str(&sdp_str).expect("parses back");
        assert_eq!(parsed.session_name, "Session");
        assert_eq!(parsed.media_descriptions.len(), 1);
        let m = &parsed.media_descriptions[0];
        assert_eq!(m.media, "audio");
        assert_eq!(m.port, 5004);
        assert_eq!(m.protocol, "RTP/AVP");
        assert_eq!(m.formats, vec!["0", "8", "101"]);
    }

    #[test]
    fn offer_origin_uses_numeric_session_id() {
        let sdp = build_offer(
            "media-session-3e071bea-bda5-4758-bf05-8bce16c690e6",
            0,
            "10.0.0.1",
            5004,
        );
        let origin = sdp
            .lines()
            .find(|line| line.starts_with("o="))
            .expect("origin line");
        let fields: Vec<&str> = origin.trim_start_matches("o=").split_whitespace().collect();

        assert_eq!(fields.len(), 6, "origin line was: {}", origin);
        assert!(
            fields[1].bytes().all(|b| b.is_ascii_digit()),
            "SDP sess-id must be numeric for PJMEDIA/Asterisk interop: {}",
            origin
        );
        assert!(
            fields[1].len() <= 20,
            "SDP sess-id should fit common 64-bit parser limits: {}",
            origin
        );
    }

    /// Sprint 2.5 P2 regression: every plaintext (RTP/AVP) offer must
    /// advertise PT 101 telephone-event + the RFC 4733 fmtp param
    /// range. Pre-P2 the non-SRTP code path emitted `m=audio … RTP/AVP
    /// 0 8` with no DTMF rtpmap, which silently broke DTMF negotiation
    /// for any plaintext call. The unified `generate_local_sdp` emits
    /// the full PCMU + PCMA + 101 set on every offer.
    #[test]
    fn offer_advertises_telephone_event_on_plaintext() {
        let sdp = build_offer("d", 0, "127.0.0.1", 16000);
        assert!(
            sdp.contains("m=audio 16000 RTP/AVP 0 8 101\r\n"),
            "plaintext offer must advertise PT 101 alongside PCMU + PCMA:\n{}",
            sdp
        );
        assert!(
            sdp.contains("a=rtpmap:101 telephone-event/8000\r\n"),
            "plaintext offer must carry the RFC 4733 telephone-event rtpmap:\n{}",
            sdp
        );
        assert!(
            sdp.contains("a=fmtp:101 0-15\r\n"),
            "plaintext offer must carry the RFC 4733 fmtp 0-15 range:\n{}",
            sdp
        );
    }

    /// Build an SRTP-flavoured offer (RFC 4568 §3.1.4: m= profile is
    /// `RTP/SAVP`) directly via the builder so we can assert the
    /// shape without standing up a full MediaAdapter. Mirrors the
    /// unified `generate_local_sdp` shape — PCMU + PCMA + telephone-event
    /// — with crypto lines appended.
    fn build_srtp_offer(ip: &str, port: u16, attrs: Vec<CryptoAttribute>) -> String {
        let mut media_builder = SdpBuilder::new("Session")
            .origin("-", "1", "0", "IN", "IP4", ip)
            .connection("IN", "IP4", ip)
            .time("0", "0")
            .media_audio(port, "RTP/SAVP")
            .formats(&["0", "8", "101"])
            .rtpmap("0", "PCMU/8000")
            .rtpmap("8", "PCMA/8000")
            .rtpmap("101", "telephone-event/8000")
            .fmtp("101", "0-15");
        for attr in attrs {
            media_builder = media_builder.crypto_attribute(attr);
        }
        media_builder
            .attribute("sendrecv", None::<String>)
            .done()
            .build()
            .expect("srtp offer builds")
            .to_string()
    }

    #[test]
    fn srtp_offer_uses_savp_profile_and_carries_crypto_lines() {
        // RFC 4568 §3.1.4 — m= line MUST be RTP/SAVP when offering SDES.
        use crate::adapters::srtp_negotiator::SrtpNegotiator;
        let suites = vec![
            CryptoSuite::AesCm128HmacSha1_80,
            CryptoSuite::AesCm128HmacSha1_32,
        ];
        let (_, attrs) = SrtpNegotiator::new_offerer(&suites).unwrap();
        let sdp = build_srtp_offer("127.0.0.1", 16000, attrs);

        // Wire-level checks.
        assert!(
            sdp.contains("m=audio 16000 RTP/SAVP 0 8 101\r\n"),
            "SRTP offer should use RTP/SAVP profile per RFC 4568 §3.1.4 \
             with the unified PCMU+PCMA+telephone-event format set:\n{}",
            sdp
        );
        assert!(
            sdp.contains("a=crypto:1 AES_CM_128_HMAC_SHA1_80 inline:"),
            "SRTP offer should carry tag-1 _80 crypto line:\n{}",
            sdp
        );
        assert!(
            sdp.contains("a=crypto:2 AES_CM_128_HMAC_SHA1_32 inline:"),
            "SRTP offer should carry tag-2 _32 crypto line:\n{}",
            sdp
        );

        // Round-trip: parse back via the typed parser, assert the
        // crypto attributes survive both directions.
        let parsed = SdpSession::from_str(&sdp).expect("parses");
        let m = &parsed.media_descriptions[0];
        assert_eq!(m.protocol, "RTP/SAVP");
        let crypto_count = m
            .generic_attributes
            .iter()
            .filter(|a| matches!(a, ParsedAttribute::Crypto(_)))
            .count();
        assert_eq!(crypto_count, 2);
    }

    #[test]
    fn extract_audio_crypto_finds_both_offered_lines() {
        use crate::adapters::srtp_negotiator::SrtpNegotiator;
        let (_, attrs) = SrtpNegotiator::new_offerer(&[
            CryptoSuite::AesCm128HmacSha1_80,
            CryptoSuite::AesCm128HmacSha1_32,
        ])
        .unwrap();
        let sdp = build_srtp_offer("127.0.0.1", 16000, attrs);
        let parsed = SdpSession::from_str(&sdp).expect("parses");
        let extracted = MediaAdapter::extract_audio_crypto(&parsed);
        assert_eq!(extracted.len(), 2);
        assert_eq!(extracted[0].tag, 1);
        assert_eq!(extracted[1].tag, 2);
    }

    #[test]
    fn extract_audio_crypto_ignores_unknown_crypto_suite_and_keeps_supported_lines() {
        let sdp = concat!(
            "v=0\r\n",
            "o=- 1 0 IN IP4 127.0.0.1\r\n",
            "s=Session\r\n",
            "c=IN IP4 127.0.0.1\r\n",
            "t=0 0\r\n",
            "m=audio 16000 RTP/SAVP 0 8 101\r\n",
            "a=rtpmap:0 PCMU/8000\r\n",
            "a=crypto:99 AEAD_AES_128_GCM inline:AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA==\r\n",
            "a=crypto:1 AES_CM_128_HMAC_SHA1_80 inline:AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA\r\n",
            "a=sendrecv\r\n",
        );
        let parsed = SdpSession::from_str(sdp).expect("unknown crypto suite should not fail SDP");
        let extracted = MediaAdapter::extract_audio_crypto(&parsed);
        assert_eq!(extracted.len(), 1);
        assert_eq!(extracted[0].tag, 1);
        assert_eq!(extracted[0].suite, CryptoSuite::AesCm128HmacSha1_80);
    }

    #[test]
    fn extract_audio_crypto_parses_asterisk_default_aes256_name() {
        let sdp = concat!(
            "v=0\r\n",
            "o=- 1 0 IN IP4 127.0.0.1\r\n",
            "s=Session\r\n",
            "c=IN IP4 127.0.0.1\r\n",
            "t=0 0\r\n",
            "m=audio 16000 RTP/SAVP 0 8 101\r\n",
            "a=rtpmap:0 PCMU/8000\r\n",
            "a=crypto:1 AES_CM_128_HMAC_SHA1_80 inline:AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA\r\n",
            "a=crypto:2 AES_256_CM_HMAC_SHA1_80 inline:AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA==\r\n",
            "a=sendrecv\r\n",
        );
        let parsed = SdpSession::from_str(sdp).expect("Asterisk AES-256 crypto name parses");
        let extracted = MediaAdapter::extract_audio_crypto(&parsed);
        assert_eq!(extracted.len(), 2);
        assert_eq!(extracted[0].suite, CryptoSuite::AesCm128HmacSha1_80);
        assert_eq!(extracted[1].suite, CryptoSuite::AesCm256HmacSha1_80);
    }

    /// Sprint 3 A6 — when a public RTP address is configured (static
    /// or STUN-discovered), the offer's c=/o=/m= lines must advertise
    /// it instead of the local interface IP/port. Mirrors what the
    /// generate_local_sdp body does — `local_ip_str` resolves to
    /// `public.ip()` when set, and `port` to `public.port()` when
    /// non-zero, else falls back to the per-session local port.
    #[test]
    fn public_rtp_addr_override_replaces_local_ip_and_port_in_offer() {
        let public: SocketAddr = "203.0.113.42:30000".parse().unwrap();
        let local_fallback: std::net::IpAddr = "192.168.1.10".parse().unwrap();
        let local_port_fallback: u16 = 16000;

        // Replicate the override branch the way generate_local_sdp does it.
        let public_opt = Some(public);
        let advertised_ip = public_opt.map(|sa| sa.ip()).unwrap_or(local_fallback);
        let port = public_opt
            .filter(|sa| sa.port() != 0)
            .map(|sa| sa.port())
            .unwrap_or(local_port_fallback);

        let dialog_id = "dlg";
        let origin_session_id = sdp_origin_session_id(dialog_id);
        let sdp = build_offer(dialog_id, 0, &advertised_ip.to_string(), port);
        assert!(
            sdp.contains("c=IN IP4 203.0.113.42\r\n"),
            "c= must carry public IP when override set:\n{}",
            sdp
        );
        assert!(
            sdp.contains(&format!(
                "o=- {} 0 IN IP4 203.0.113.42\r\n",
                origin_session_id
            )),
            "o= must carry public IP when override set:\n{}",
            sdp
        );
        assert!(
            sdp.contains("m=audio 30000 RTP/AVP"),
            "m=audio must carry public port when override set:\n{}",
            sdp
        );
    }

    #[test]
    fn public_rtp_addr_unset_falls_back_to_local_ip_and_local_port() {
        let public_opt: Option<SocketAddr> = None;
        let local_fallback: std::net::IpAddr = "192.168.1.10".parse().unwrap();
        let local_port_fallback: u16 = 16000;

        let advertised_ip = public_opt.map(|sa| sa.ip()).unwrap_or(local_fallback);
        let port = public_opt
            .filter(|sa| sa.port() != 0)
            .map(|sa| sa.port())
            .unwrap_or(local_port_fallback);

        let sdp = build_offer("dlg", 0, &advertised_ip.to_string(), port);
        assert!(
            sdp.contains("c=IN IP4 192.168.1.10\r\n"),
            "c= falls back to local_ip when no override:\n{}",
            sdp
        );
        assert!(
            sdp.contains("m=audio 16000 RTP/AVP"),
            "m=audio falls back to local_port when no override:\n{}",
            sdp
        );
    }

    /// Sprint 3 C1 — when `comfort_noise_enabled` is set, the SDP
    /// offer's `m=audio` line lists `13` and the body carries an
    /// `a=rtpmap:13 CN/8000` line. The order must be `0 8 13 101` so
    /// telephone-event remains last (Sprint 2.5 P2 fixture stability).
    #[test]
    fn cn_enabled_offer_advertises_pt13_and_rtpmap() {
        let ip = "127.0.0.1";
        let port = 16000;
        let sdp = SdpBuilder::new("Session")
            .origin("-", "dlg", "0", "IN", "IP4", ip)
            .connection("IN", "IP4", ip)
            .time("0", "0")
            .media_audio(port, "RTP/AVP")
            .formats(&["0", "8", "13", "101"])
            .rtpmap("0", "PCMU/8000")
            .rtpmap("8", "PCMA/8000")
            .rtpmap("13", "CN/8000")
            .rtpmap("101", "telephone-event/8000")
            .fmtp("101", "0-15")
            .attribute("sendrecv", None::<String>)
            .done()
            .build()
            .expect("offer builds")
            .to_string();

        assert!(
            sdp.contains("m=audio 16000 RTP/AVP 0 8 13 101\r\n"),
            "format list must include 13 between PCMA and telephone-event:\n{}",
            sdp
        );
        assert!(
            sdp.contains("a=rtpmap:13 CN/8000\r\n"),
            "RFC 3389 CN rtpmap must appear:\n{}",
            sdp
        );
        // Sanity: existing PTs still present in the right shape.
        assert!(sdp.contains("a=rtpmap:0 PCMU/8000\r\n"));
        assert!(sdp.contains("a=rtpmap:8 PCMA/8000\r\n"));
        assert!(sdp.contains("a=rtpmap:101 telephone-event/8000\r\n"));
    }

    /// Sprint 3.5 — strict matching answers with the intersection
    /// only. Offer = `0 101` (no PCMA); answer must carry `0 101`,
    /// not the legacy full `0 8 101` set.
    #[test]
    fn strict_default_answers_with_intersection_only() {
        let offer_sdp = SdpBuilder::new("Session")
            .origin("-", "1", "0", "IN", "IP4", "127.0.0.1")
            .connection("IN", "IP4", "127.0.0.1")
            .time("0", "0")
            .media_audio(16000, "RTP/AVP")
            .formats(&["0", "101"])
            .rtpmap("0", "PCMU/8000")
            .rtpmap("101", "telephone-event/8000")
            .fmtp("101", "0-15")
            .attribute("sendrecv", None::<String>)
            .done()
            .build()
            .expect("offer builds")
            .to_string();
        let offer = SdpSession::from_str(&offer_sdp).expect("offer parses");

        let formats = compute_answer_formats(
            &offer,
            &[0, 8, 101], /*strict*/
            true,         /*offer_srtp*/
            false,        /*srtp_required*/
            false,
        )
        .expect("strict-mode match succeeds");
        assert_eq!(
            formats,
            vec!["0".to_string(), "101".to_string()],
            "strict answer must drop PCMA when the offer didn't list it"
        );
    }

    /// Sprint 3.5 — permissive mode preserves the legacy
    /// pre-Sprint-3.5 "always full set" answer shape.
    #[test]
    fn permissive_mode_answers_with_full_set() {
        let offer_sdp = SdpBuilder::new("Session")
            .origin("-", "1", "0", "IN", "IP4", "127.0.0.1")
            .connection("IN", "IP4", "127.0.0.1")
            .time("0", "0")
            .media_audio(16000, "RTP/AVP")
            .formats(&["0", "101"])
            .rtpmap("0", "PCMU/8000")
            .rtpmap("101", "telephone-event/8000")
            .fmtp("101", "0-15")
            .attribute("sendrecv", None::<String>)
            .done()
            .build()
            .expect("offer builds")
            .to_string();
        let offer = SdpSession::from_str(&offer_sdp).expect("offer parses");

        let formats = compute_answer_formats(
            &offer,
            &[0, 8, 101], /*strict*/
            false,        /*offer_srtp*/
            false,        /*srtp_required*/
            false,
        )
        .expect("permissive mode never errors on overlap");
        assert_eq!(
            formats,
            vec!["0".to_string(), "8".to_string(), "101".to_string()],
            "permissive answer keeps the full legacy set"
        );
    }

    /// Sprint 3.5 — strict mode + zero overlap returns
    /// `SDPNegotiationFailed`. The state machine turns this into
    /// `488 Not Acceptable Here` (the same path `srtp_required`
    /// already uses on a plain offer).
    #[test]
    fn strict_default_no_overlap_returns_negotiation_failed() {
        let offer_sdp = SdpBuilder::new("Session")
            .origin("-", "1", "0", "IN", "IP4", "127.0.0.1")
            .connection("IN", "IP4", "127.0.0.1")
            .time("0", "0")
            .media_audio(16000, "RTP/AVP")
            .formats(&["97", "98"])
            .rtpmap("97", "VP8/90000")
            .rtpmap("98", "VP9/90000")
            .attribute("sendrecv", None::<String>)
            .done()
            .build()
            .expect("offer builds")
            .to_string();
        let offer = SdpSession::from_str(&offer_sdp).expect("offer parses");

        let err = compute_answer_formats(
            &offer,
            &[0, 8, 101], /*strict*/
            true,         /*offer_srtp*/
            false,        /*srtp_required*/
            false,
        )
        .unwrap_err();
        assert!(
            matches!(err, SessionError::SDPNegotiationFailed(_)),
            "no overlap must surface as SDPNegotiationFailed → 488 NAH; got {:?}",
            err
        );
    }

    /// Sprint 3.5 — strict matching preserves CN advertisement
    /// when both peers offer it. Offer `0 13 101`, our caps include
    /// `13` (comfort_noise_enabled=true), answer carries `0 13 101`.
    #[test]
    fn strict_with_cn_enabled_keeps_cn_in_intersection() {
        let offer_sdp = SdpBuilder::new("Session")
            .origin("-", "1", "0", "IN", "IP4", "127.0.0.1")
            .connection("IN", "IP4", "127.0.0.1")
            .time("0", "0")
            .media_audio(16000, "RTP/AVP")
            .formats(&["0", "13", "101"])
            .rtpmap("0", "PCMU/8000")
            .rtpmap("13", "CN/8000")
            .rtpmap("101", "telephone-event/8000")
            .fmtp("101", "0-15")
            .attribute("sendrecv", None::<String>)
            .done()
            .build()
            .expect("offer builds")
            .to_string();
        let offer = SdpSession::from_str(&offer_sdp).expect("offer parses");

        let formats = compute_answer_formats(
            &offer,
            &[0, 8, 13, 101], /*strict*/
            true,             /*offer_srtp*/
            false,            /*srtp_required*/
            false,
        )
        .expect("CN-on-both-sides match succeeds");
        assert_eq!(
            formats,
            vec!["0".to_string(), "13".to_string(), "101".to_string()],
            "intersection must include 13 when both sides advertise it"
        );
    }

    #[test]
    fn cn_disabled_offer_omits_pt13_and_rtpmap() {
        // The pre-Sprint-3 baseline shape — no `13`, no CN rtpmap.
        let sdp = build_offer("dlg", 0, "127.0.0.1", 16000);
        assert!(
            sdp.contains("m=audio 16000 RTP/AVP 0 8 101\r\n"),
            "default offer must keep the pre-Sprint-3 format set:\n{}",
            sdp
        );
        assert!(
            !sdp.contains("CN/8000"),
            "default offer must not advertise CN: \n{}",
            sdp
        );
    }

    #[tokio::test]
    async fn opus_offered_codec_appears_in_generated_offer() {
        // NEXT_STEPS C2 — when Opus (PT 111) is in the configured
        // offered_codecs, the generated SDP offer MUST carry
        // `a=rtpmap:111 opus/48000/2` and list `111` in the m-line
        // format set. The legacy PCMU/PCMA/DTMF still go out alongside
        // when included in the list.
        use crate::session_store::SessionStore;
        use crate::state_table::types::Role;
        use rvoip_media_core::relay::controller::MediaSessionController;
        use std::net::Ipv4Addr;

        let controller = Arc::new(MediaSessionController::new());
        let store = Arc::new(SessionStore::new());
        let session_id = SessionId("opus-offer-test".to_string());
        store
            .create_session(session_id.clone(), Role::UAC, false)
            .await
            .expect("create session");

        let mut adapter = MediaAdapter::new(
            controller,
            store.clone(),
            IpAddr::V4(Ipv4Addr::LOCALHOST),
            16000,
            16100,
        );
        adapter.set_offered_codecs(vec![0, 8, 111, 101]);
        adapter
            .start_session(&session_id)
            .await
            .expect("start session");

        let sdp = adapter
            .generate_local_sdp_offer(&session_id, crate::types::MediaDirection::SendRecv)
            .await
            .expect("offer builds");

        assert!(
            sdp.contains("a=rtpmap:111 opus/48000/2"),
            "Opus rtpmap must appear in the offer when 111 is in offered_codecs:\n{}",
            sdp
        );
        assert!(
            sdp.contains("a=rtpmap:0 PCMU/8000"),
            "legacy PCMU rtpmap must still appear:\n{}",
            sdp
        );
        assert!(
            sdp.contains(" 111 ") || sdp.contains(" 111\r\n"),
            "m-line format list must include PT 111:\n{}",
            sdp
        );
    }

    #[tokio::test]
    async fn g729_offer_advertises_annex_b_yes() {
        use crate::session_store::SessionStore;
        use crate::state_table::types::Role;
        use rvoip_media_core::relay::controller::MediaSessionController;
        use std::net::Ipv4Addr;

        let controller = Arc::new(MediaSessionController::new());
        let store = Arc::new(SessionStore::new());
        let session_id = SessionId("g729-annexb-yes-offer-test".to_string());
        store
            .create_session(session_id.clone(), Role::UAC, false)
            .await
            .expect("create session");

        let mut adapter = MediaAdapter::new(
            controller,
            store.clone(),
            IpAddr::V4(Ipv4Addr::LOCALHOST),
            16000,
            16100,
        );
        adapter.set_offered_codecs(vec![18, 101]);
        adapter.set_g729_annex_b(true);
        adapter
            .start_session(&session_id)
            .await
            .expect("start session");

        let sdp = adapter
            .generate_local_sdp_offer(&session_id, crate::types::MediaDirection::SendRecv)
            .await
            .expect("offer builds");

        assert!(
            sdp.contains(" RTP/AVP 18 101\r\n"),
            "m-line must advertise PT18:\n{}",
            sdp
        );
        assert!(
            sdp.contains("a=rtpmap:18 G729/8000"),
            "G.729 rtpmap must appear:\n{}",
            sdp
        );
        assert!(
            sdp.contains("a=fmtp:18 annexb=yes"),
            "Annex B enabled offer must advertise annexb=yes:\n{}",
            sdp
        );
    }

    #[tokio::test]
    async fn g729_offer_advertises_annex_b_no() {
        use crate::session_store::SessionStore;
        use crate::state_table::types::Role;
        use rvoip_media_core::relay::controller::MediaSessionController;
        use std::net::Ipv4Addr;

        let controller = Arc::new(MediaSessionController::new());
        let store = Arc::new(SessionStore::new());
        let session_id = SessionId("g729-annexb-no-offer-test".to_string());
        store
            .create_session(session_id.clone(), Role::UAC, false)
            .await
            .expect("create session");

        let mut adapter = MediaAdapter::new(
            controller,
            store.clone(),
            IpAddr::V4(Ipv4Addr::LOCALHOST),
            16000,
            16100,
        );
        adapter.set_offered_codecs(vec![18, 101]);
        adapter.set_g729_annex_b(false);
        adapter
            .start_session(&session_id)
            .await
            .expect("start session");

        let sdp = adapter
            .generate_local_sdp_offer(&session_id, crate::types::MediaDirection::SendRecv)
            .await
            .expect("offer builds");

        assert!(
            sdp.contains("a=fmtp:18 annexb=no"),
            "Annex B disabled offer must advertise annexb=no:\n{}",
            sdp
        );
    }

    #[cfg(feature = "g729")]
    #[tokio::test]
    async fn g729_uas_negotiation_updates_media_core_codec() {
        use crate::session_store::SessionStore;
        use crate::state_table::types::Role;
        use rvoip_media_core::relay::controller::MediaSessionController;
        use std::net::Ipv4Addr;

        let controller = Arc::new(MediaSessionController::new());
        let store = Arc::new(SessionStore::new());
        let session_id = SessionId("g729-negotiated-media-config-test".to_string());
        store
            .create_session(session_id.clone(), Role::UAS, false)
            .await
            .expect("create session");

        let mut adapter = MediaAdapter::new(
            controller.clone(),
            store.clone(),
            IpAddr::V4(Ipv4Addr::LOCALHOST),
            16000,
            16100,
        );
        adapter.set_offered_codecs(vec![18, 101]);
        adapter.set_g729_annex_b(false);
        adapter
            .start_session(&session_id)
            .await
            .expect("start session");

        let offer_sdp = SdpBuilder::new("Session")
            .origin("-", "1", "0", "IN", "IP4", "127.0.0.1")
            .connection("IN", "IP4", "127.0.0.1")
            .time("0", "0")
            .media_audio(35000, "RTP/AVP")
            .formats(&["18", "101"])
            .rtpmap("18", "G729/8000")
            .fmtp("18", "annexb=no")
            .rtpmap("101", "telephone-event/8000")
            .fmtp("101", "0-15")
            .attribute("sendrecv", None::<String>)
            .done()
            .build()
            .expect("offer builds")
            .to_string();

        let (answer_sdp, config) = adapter
            .negotiate_sdp_as_uas(&session_id, &offer_sdp)
            .await
            .expect("G.729 offer negotiates");

        assert!(
            answer_sdp.contains("a=fmtp:18 annexb=no"),
            "G.729A answer must keep annexb=no:\n{}",
            answer_sdp
        );
        assert_eq!(config.codec, "G729A");
        assert_eq!(config.payload_type, 18);

        let dialog_id = adapter
            .session_to_dialog
            .get(&session_id)
            .expect("dialog mapping exists")
            .value()
            .clone();
        let info = controller
            .get_session_info(&dialog_id)
            .await
            .expect("media session exists");
        assert_eq!(info.config.preferred_codec, Some("G729A".to_string()));
        assert_eq!(
            info.config.remote_addr,
            Some(SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 35000))
        );

        adapter
            .cleanup_session(&session_id)
            .await
            .expect("cleanup media session");
    }

    #[tokio::test]
    async fn default_offered_codecs_omit_opus() {
        // Regression guard: the C2 default (PCMU + PCMA + DTMF) must
        // not silently start advertising Opus. Adding Opus support
        // when media-core has no `opus` feature would negotiate a
        // codec the encoder can't produce.
        use crate::session_store::SessionStore;
        use crate::state_table::types::Role;
        use rvoip_media_core::relay::controller::MediaSessionController;
        use std::net::Ipv4Addr;

        let controller = Arc::new(MediaSessionController::new());
        let store = Arc::new(SessionStore::new());
        let session_id = SessionId("default-codecs-test".to_string());
        store
            .create_session(session_id.clone(), Role::UAC, false)
            .await
            .expect("create session");

        let adapter = MediaAdapter::new(
            controller,
            store.clone(),
            IpAddr::V4(Ipv4Addr::LOCALHOST),
            16000,
            16100,
        );
        adapter
            .start_session(&session_id)
            .await
            .expect("start session");

        let sdp = adapter
            .generate_local_sdp_offer(&session_id, crate::types::MediaDirection::SendRecv)
            .await
            .expect("offer builds");

        assert!(
            !sdp.contains("opus"),
            "default offer must not advertise Opus:\n{}",
            sdp
        );
    }

    #[tokio::test]
    async fn next_sdp_origin_increments_version_across_calls() {
        // RFC 3264 §8 — every fresh local SDP body on a session must
        // carry a strictly greater `o=` version. The adapter centralises
        // this through `next_sdp_origin`; the unit test below proves
        // the increment, the audit (NEXT_STEPS.md Area C1.2) proves all
        // production builder paths route through this helper.
        use crate::session_store::SessionStore;
        use crate::state_table::types::Role;
        use rvoip_media_core::relay::controller::MediaSessionController;
        use std::net::Ipv4Addr;

        let controller = Arc::new(MediaSessionController::new());
        let store = Arc::new(SessionStore::new());
        let session_id = SessionId("test-version-bump".to_string());
        store
            .create_session(session_id.clone(), Role::UAC, false)
            .await
            .expect("create session");

        let adapter = MediaAdapter::new(
            controller,
            store.clone(),
            IpAddr::V4(Ipv4Addr::LOCALHOST),
            16000,
            16100,
        );

        let (sess1, v_initial) = adapter
            .next_sdp_origin(&session_id)
            .await
            .expect("initial offer");
        let (sess2, v_hold) = adapter
            .next_sdp_origin(&session_id)
            .await
            .expect("hold re-INVITE");
        let (sess3, v_resume) = adapter
            .next_sdp_origin(&session_id)
            .await
            .expect("resume re-INVITE");

        // o= session-id must stay stable across re-INVITEs on the same
        // dialog (RFC 3264 §6 + §8) — only the version advances.
        assert_eq!(
            sess1, sess2,
            "session-id must not change between offer and hold re-INVITE"
        );
        assert_eq!(
            sess1, sess3,
            "session-id must not change between offer and resume re-INVITE"
        );

        // Strict version monotonicity, RFC 3264 §8.
        assert!(
            v_hold > v_initial,
            "hold re-INVITE version ({}) must exceed initial offer ({})",
            v_hold,
            v_initial
        );
        assert!(
            v_resume > v_hold,
            "resume re-INVITE version ({}) must exceed hold ({})",
            v_resume,
            v_hold
        );
    }

    #[test]
    fn port_zero_rejection_emits_m_audio_zero_with_offered_proto() {
        // RFC 3264 §6 / RFC 4568 §7.3: when we decline an offered
        // m-line (here: peer offered SRTP, our policy is plaintext),
        // the answer's m-line port must be 0 and the proto must echo
        // the offer's proto so the peer distinguishes a policy
        // rejection from a parse error.
        let sdp = build_port_zero_rejection_sdp("1234", 7, "192.0.2.10", "RTP/SAVP")
            .expect("rejection SDP builds");

        assert!(
            sdp.contains("m=audio 0 RTP/SAVP"),
            "rejection answer must carry port=0 and the offered proto:\n{}",
            sdp
        );
        assert!(
            sdp.contains("o=- 1234 7 IN IP4 192.0.2.10"),
            "rejection answer must carry the supplied origin line:\n{}",
            sdp
        );
        assert!(
            sdp.contains("c=IN IP4 192.0.2.10"),
            "rejection answer must include a c= line:\n{}",
            sdp
        );
        assert!(
            !sdp.contains("a=crypto"),
            "rejection answer must not advertise any crypto attributes:\n{}",
            sdp
        );
    }

    #[test]
    fn port_zero_rejection_echoes_avp_proto_when_offered() {
        // If peer ever offers plaintext but we still want to decline
        // (e.g. require_srtp branch wired through this helper later),
        // the proto echoed must be `RTP/AVP`, not the SAVP default.
        let sdp = build_port_zero_rejection_sdp("1", 0, "10.0.0.1", "RTP/AVP")
            .expect("rejection SDP builds");

        assert!(
            sdp.contains("m=audio 0 RTP/AVP"),
            "rejection must echo the offered proto verbatim:\n{}",
            sdp
        );
    }

    #[test]
    fn public_rtp_addr_with_zero_port_keeps_local_port() {
        // The override semantics: when `media_public_addr` carries an
        // IP-only mapping (port 0), advertise the public IP but keep
        // the per-session local port. Useful for SBC-fronted setups
        // where the port doesn't change but the IP does.
        let public: SocketAddr = "203.0.113.42:0".parse().unwrap();
        let public_opt = Some(public);
        let local_fallback: std::net::IpAddr = "192.168.1.10".parse().unwrap();
        let local_port_fallback: u16 = 16000;

        let advertised_ip = public_opt.map(|sa| sa.ip()).unwrap_or(local_fallback);
        let port = public_opt
            .filter(|sa| sa.port() != 0)
            .map(|sa| sa.port())
            .unwrap_or(local_port_fallback);

        assert_eq!(advertised_ip.to_string(), "203.0.113.42");
        assert_eq!(port, 16000, "zero port must defer to local_port_fallback");
    }

    #[tokio::test]
    async fn uas_plain_avp_answer_uses_allocated_rtp_port() {
        use crate::session_store::SessionStore;
        use crate::state_table::types::Role;
        use rvoip_media_core::relay::controller::MediaSessionController;
        use std::net::Ipv4Addr;

        let controller = Arc::new(MediaSessionController::new());
        let store = Arc::new(SessionStore::new());
        let session_id = SessionId("uas-plain-avp-answer-test".to_string());
        store
            .create_session(session_id.clone(), Role::UAS, false)
            .await
            .expect("create session");

        let adapter = MediaAdapter::new(
            controller,
            store,
            IpAddr::V4(Ipv4Addr::LOCALHOST),
            16000,
            16100,
        );
        adapter
            .start_session(&session_id)
            .await
            .expect("start media session");

        let offer_sdp = SdpBuilder::new("Session")
            .origin("-", "1", "0", "IN", "IP4", "127.0.0.1")
            .connection("IN", "IP4", "127.0.0.1")
            .time("0", "0")
            .media_audio(35000, "RTP/AVP")
            .formats(&["0"])
            .rtpmap("0", "PCMU/8000")
            .attribute("sendrecv", None::<String>)
            .done()
            .build()
            .expect("offer builds")
            .to_string();

        let (answer_sdp, config) = adapter
            .negotiate_sdp_as_uas(&session_id, &offer_sdp)
            .await
            .expect("plain RTP offer negotiates");
        let parsed = SdpSession::from_str(&answer_sdp).expect("answer parses");
        let media = parsed
            .media_descriptions
            .iter()
            .find(|media| media.media == "audio")
            .expect("audio media line");

        assert_eq!(media.protocol, "RTP/AVP");
        assert_eq!(media.formats, vec!["0".to_string()]);
        assert_eq!(media.port, config.local_addr.port());
        assert_ne!(
            media.port, 5060,
            "SDP answer must advertise the allocated RTP port, not the SIP listener port"
        );

        adapter
            .cleanup_session(&session_id)
            .await
            .expect("cleanup media session");
    }
}