actr_runtime/wire/webrtc/

coordinator.rs

// WebRTC Signaling Coordinator - Coordinates WebRTC P2P connection establishment

#[allow(dead_code)]
fn is_ipv4_candidate_allowed(cand: &str) -> bool {
    // Only filter out IPv6 candidates (link-local and other IPv6 addresses)
    // Allow all IPv4 candidates (private and public IPs)
    if cand.contains("fe80::") || cand.contains(" udp6 ") || cand.contains("::") {
        return false;
    }

    // Accept all IPv4 candidates by default
    // This includes: loopback (127.x), private (10.x, 172.x, 192.168.x), and public IPs
    true
}

// Responsibilities:
// - Listen to WebRTC signaling messages from SignalingClient
// - Handle Offer/Answer/ICE candidate exchanges
// - Establish and manage RTCPeerConnection instances
// - Create and cache WebRtcConnection instances
// - Aggregate messages from all peers

use super::connection::WebRtcConnection;
use super::negotiator::WebRtcNegotiator;
#[cfg(feature = "opentelemetry")]
use super::trace;
use super::{SignalingClient, WebRtcConfig};
use crate::INITIAL_CONNECTION_TIMEOUT;
use crate::error::{RuntimeError, RuntimeResult};
use crate::inbound::MediaFrameRegistry;
use crate::lifecycle::CredentialState;
use crate::transport::connection_event::{ConnectionEvent, ConnectionEventBroadcaster};
use actr_framework::Bytes;
use actr_protocol::ActrIdExt;
use actr_protocol::prost::Message as ProstMessage;
use actr_protocol::{
    ActrId, ActrRelay, PayloadType, RoleAssignment, RoleNegotiation, SignalingEnvelope, actr_relay,
    session_description::Type as SdpType, signaling_envelope,
};
use std::collections::HashMap;
use std::{
    sync::Arc,
    time::{Duration, Instant},
};
use tokio::sync::{Mutex, RwLock, mpsc, oneshot};
use tokio::task::JoinHandle;
use tokio_util::sync::CancellationToken;
use tracing::Instrument;
#[cfg(feature = "opentelemetry")]
use tracing_opentelemetry::OpenTelemetrySpanExt;
use webrtc::data_channel::RTCDataChannel;
use webrtc::ice_transport::ice_candidate::RTCIceCandidate;
use webrtc::ice_transport::ice_gathering_state::RTCIceGatheringState;
use webrtc::peer_connection::{RTCPeerConnection, peer_connection_state::RTCPeerConnectionState};
use webrtc::track::track_local::TrackLocalWriter;

const ICE_RESTART_MAX_RETRIES: u32 = 5;
const ICE_RESTART_TIMEOUT: Duration = Duration::from_secs(5);
const ICE_RESTART_INITIAL_BACKOFF_MS: u64 = 5000;
const ICE_RESTART_MAX_BACKOFF_MS: u64 = 60000;
const ICE_RESTART_MAX_TOTAL_DURATION: Duration = Duration::from_secs(60);
const ICE_GATHERING_TIMEOUT: Duration = Duration::from_secs(10);
const ROLE_WAIT_TIMEOUT: Duration = Duration::from_secs(10);

// Health check constants
const HEALTH_CHECK_INTERVAL: Duration = Duration::from_secs(10);
const MAX_FAILED_DURATION: Duration = Duration::from_secs(60); // 1 minute

/// Per-peer negotiation state (role, ready signals)
/// Consolidates multiple related fields into a single lock to reduce contention.
#[derive(Default)]
struct PeerNegotiationState {
    /// Role negotiation responder
    role_tx: Option<oneshot::Sender<bool>>,
    /// Ready notifier for answerer path
    ready_tx: Option<oneshot::Sender<()>>,
    /// Ready receiver for proactive offerer path
    ready_rx: Option<oneshot::Receiver<()>>,
}

/// Simple exponential backoff iterator for retries
#[derive(Debug)]
struct ExponentialBackoff {
    current_retries: u32,
    max_retries: Option<u32>,
    initial_delay: Duration,
    max_delay: Duration,
    /// Optional total duration limit (across all retries)
    max_total_duration: Option<Duration>,
    /// Start time for tracking total duration
    start_time: Option<Instant>,
}

impl ExponentialBackoff {
    pub fn new(initial_delay: Duration, max_delay: Duration, max_retries: Option<u32>) -> Self {
        Self {
            current_retries: 0,
            max_retries,
            initial_delay,
            max_delay,
            max_total_duration: None,
            start_time: None,
        }
    }

    /// Create a new ExponentialBackoff with total duration limit
    pub fn with_total_duration(
        initial_delay: Duration,
        max_delay: Duration,
        max_retries: Option<u32>,
        max_total_duration: Duration,
    ) -> Self {
        Self {
            current_retries: 0,
            max_retries,
            initial_delay,
            max_delay,
            max_total_duration: Some(max_total_duration),
            start_time: Some(Instant::now()),
        }
    }

    /// Check if total duration has been exceeded
    fn is_duration_exceeded(&self) -> bool {
        if let (Some(max_duration), Some(start)) = (self.max_total_duration, self.start_time) {
            start.elapsed() > max_duration
        } else {
            false
        }
    }
}

impl Iterator for ExponentialBackoff {
    type Item = Duration;

    fn next(&mut self) -> Option<Duration> {
        // Initialize start_time on first call if max_total_duration is set
        if self.max_total_duration.is_some() && self.start_time.is_none() {
            self.start_time = Some(Instant::now());
        }

        // Check total duration limit first
        if self.is_duration_exceeded() {
            return None;
        }

        let delay = self.initial_delay;

        // Check max retries
        if let Some(max_retries) = self.max_retries {
            self.current_retries += 1;
            if self.current_retries > max_retries {
                return None;
            }
        }

        self.initial_delay = (self.initial_delay * 2).min(self.max_delay);
        Some(delay)
    }
}

/// Type alias for message receiver (from all peers)
type MessageRx = Arc<Mutex<mpsc::UnboundedReceiver<(Vec<u8>, Bytes, PayloadType)>>>;

/// Peer connection state
struct PeerState {
    /// RTCPeerConnection (for receiving ICE candidates)
    peer_connection: Arc<RTCPeerConnection>,

    /// WebRtcConnection (for business message transmission)
    webrtc_conn: WebRtcConnection,

    /// Connection ready notification (for initiate_connection to wait)
    ready_tx: Option<oneshot::Sender<()>>,

    /// Whether we are the offerer for the current session (affects ICE restart handling)
    is_offerer: bool,

    /// Whether ICE restart is in progress (controls buffering and retries)
    ice_restart_inflight: bool,

    /// Restart attempts counter (resets on success)
    ice_restart_attempts: u32,

    /// In-flight ICE restart task handle (for de-duplication and lifecycle management)
    restart_task_handle: Option<JoinHandle<()>>,

    /// Last state change timestamp (for health check)
    last_state_change: std::time::Instant,

    /// Current connection state (for health check)
    current_state: RTCPeerConnectionState,
}

/// WebRTC signaling coordinator
pub struct WebRtcCoordinator {
    /// Local Actor ID
    local_id: ActrId,

    /// Local credentials
    credential_state: CredentialState,

    /// SignalingClient (for sending ICE/SDP)
    signaling_client: Arc<dyn SignalingClient>,

    /// WebRTC negotiator
    negotiator: WebRtcNegotiator,

    /// Peer state mapping (ActrId → PeerState)
    peers: Arc<RwLock<HashMap<ActrId, PeerState>>>,

    /// Pending ICE candidates (received before remote description is set)
    /// ActrId → Vec<candidate_string>
    pending_candidates: Arc<RwLock<HashMap<ActrId, Vec<String>>>>,

    /// Message receive channel (aggregated from all peers)
    /// (from: ActrId bytes, data: Bytes)
    /// Format: (sender_id_bytes, message_data, payload_type)
    message_rx: MessageRx,
    message_tx: mpsc::UnboundedSender<(Vec<u8>, Bytes, PayloadType)>,

    /// MediaTrack callback registry (for WebRTC native media streams)
    media_frame_registry: Arc<MediaFrameRegistry>,

    /// Per-peer negotiation state (role, ready signals, restart tasks)
    /// Single lock consolidating pending_role, pending_ready, pending_ready_wait, and in_flight_restarts
    peer_negotiation: Arc<Mutex<HashMap<ActrId, PeerNegotiationState>>>,

    /// Connection event broadcaster for notifying all layers
    event_broadcaster: ConnectionEventBroadcaster,

    /// Root tracing contexts for connection initiation (ActrId → Context)
    #[cfg(feature = "opentelemetry")]
    root_context_map: Arc<RwLock<HashMap<ActrId, opentelemetry::Context>>>,
}

impl WebRtcCoordinator {
    /// Create new coordinator
    pub fn new(
        local_id: ActrId,
        credential_state: CredentialState,
        signaling_client: Arc<dyn SignalingClient>,
        webrtc_config: WebRtcConfig,
        realm_id: u32,
        media_frame_registry: Arc<MediaFrameRegistry>,
    ) -> Self {
        let (message_tx, message_rx) = mpsc::unbounded_channel();
        let negotiator = WebRtcNegotiator::new(webrtc_config, realm_id, credential_state.clone());

        Self {
            local_id,
            credential_state,
            signaling_client,
            negotiator,
            peers: Arc::new(RwLock::new(HashMap::new())),
            pending_candidates: Arc::new(RwLock::new(HashMap::new())),
            message_rx: Arc::new(Mutex::new(message_rx)),
            message_tx,
            media_frame_registry,
            peer_negotiation: Arc::new(Mutex::new(HashMap::new())),
            event_broadcaster: ConnectionEventBroadcaster::new(),
            #[cfg(feature = "opentelemetry")]
            root_context_map: Arc::new(RwLock::new(HashMap::new())),
        }
    }

    /// Get a subscriber for connection events
    pub fn subscribe_events(&self) -> tokio::sync::broadcast::Receiver<ConnectionEvent> {
        self.event_broadcaster.subscribe()
    }

    /// Get the event sender for sharing with WebRtcConnection instances
    pub fn event_sender(&self) -> tokio::sync::broadcast::Sender<ConnectionEvent> {
        self.event_broadcaster.sender()
    }

    /// Trigger ICE restart for all connections in Failed/Disconnected state
    pub async fn retry_failed_connections(self: &Arc<Self>) {
        let peers = self.peers.read().await;
        // Collect peers that need restart to avoid holding lock during async operations
        let mut targets = Vec::new();

        for (peer_id, state) in peers.iter() {
            match state.current_state {
                RTCPeerConnectionState::Failed | RTCPeerConnectionState::Disconnected => {
                    if !state.ice_restart_inflight {
                        targets.push(peer_id.clone());
                    }
                }
                _ => {
                    #[cfg(test)]
                    tracing::debug!(
                        "Actor {:?} is in state {:?}, test restart",
                        peer_id,
                        state.current_state
                    );
                    targets.push(peer_id.clone());
                }
            }
        }
        drop(peers); // Release lock

        for peer_id in targets {
            tracing::info!("♻️ Auto-retrying failed connection to actor {:?}", peer_id);
            if let Err(e) = self.restart_ice(&peer_id).await {
                tracing::error!("❌ Failed to restart ICE for {:?}: {}", peer_id, e);
            }
        }
    }

    /// Clear pending ICE restart attempts (called on network loss)
    pub async fn clear_pending_restarts(&self) {
        let mut peers = self.peers.write().await;
        for (peer_id, state) in peers.iter_mut() {
            if state.ice_restart_inflight {
                tracing::info!("🛑 Aborting in-flight ICE restart for {:?}", peer_id);
                // Cancel restart task if it exists
                if let Some(handle) = state.restart_task_handle.take() {
                    handle.abort();
                }
                state.ice_restart_inflight = false;
                // Note: We don't reset attempts here, as we might want to resume counting if network comes back quickly?
                // Actually, resetting makes sense as it's a fresh network start.
                state.ice_restart_attempts = 0;
            }
        }
    }

    /// Start internal event listener for handling connection close events
    ///
    /// This listens for ConnectionClosed and DataChannelClosed events and triggers
    /// cleanup of WebRtcCoordinator's internal resources (peers map, pending candidates, etc.)
    fn spawn_internal_event_listener(self: &Arc<Self>) -> tokio::task::JoinHandle<()> {
        let mut event_rx = self.event_broadcaster.subscribe();
        let coordinator = Arc::downgrade(self);

        tokio::spawn(async move {
            loop {
                match event_rx.recv().await {
                    Ok(event) => {
                        if let Some(coord) = coordinator.upgrade() {
                            // Extract peer_id and check if cleanup is needed
                            let peer_id_to_cleanup = match &event {
                                ConnectionEvent::DataChannelClosed {
                                    peer_id,
                                    payload_type,
                                } => {
                                    // Only cleanup if peer still exists (avoid duplicate cleanup)
                                    if coord.peers.read().await.contains_key(peer_id) {
                                        tracing::warn!(
                                            "⚠️ DataChannel closed for peer {}, payload_type={:?}; triggering coordinator cleanup",
                                            peer_id.serial_number,
                                            payload_type
                                        );
                                        Some(peer_id.clone())
                                    } else {
                                        tracing::debug!(
                                            "ℹ️ DataChannel closed for peer {} but already cleaned up",
                                            peer_id.serial_number
                                        );
                                        None
                                    }
                                }
                                ConnectionEvent::ConnectionClosed { peer_id } => {
                                    if coord.peers.read().await.contains_key(peer_id) {
                                        tracing::warn!(
                                            "⚠️ Connection closed for peer {}; triggering coordinator cleanup",
                                            peer_id.serial_number
                                        );
                                        Some(peer_id.clone())
                                    } else {
                                        tracing::debug!(
                                            "ℹ️ Connection closed for peer {} but already cleaned up",
                                            peer_id.serial_number
                                        );
                                        None
                                    }
                                }
                                ConnectionEvent::StateChanged { peer_id, state } => {
                                    use crate::transport::connection_event::ConnectionState;
                                    if matches!(state, ConnectionState::Closed) {
                                        if coord.peers.read().await.contains_key(peer_id) {
                                            tracing::warn!(
                                                "⚠️ PeerConnection state changed to Closed for peer {}; triggering coordinator cleanup",
                                                peer_id.serial_number
                                            );
                                            Some(peer_id.clone())
                                        } else {
                                            tracing::debug!(
                                                "ℹ️ PeerConnection Closed for peer {} but already cleaned up",
                                                peer_id.serial_number
                                            );
                                            None
                                        }
                                    } else {
                                        None
                                    }
                                }
                                _ => None,
                            };

                            // Cleanup outside the match to avoid holding read lock
                            if let Some(peer_id) = peer_id_to_cleanup {
                                coord.cleanup_cancelled_connection(&peer_id).await;
                            }
                        } else {
                            // Coordinator dropped, exit
                            tracing::debug!(
                                "🔌 WebRtcCoordinator internal event listener stopping (coordinator dropped)"
                            );
                            break;
                        }
                    }
                    Err(tokio::sync::broadcast::error::RecvError::Lagged(n)) => {
                        tracing::warn!(
                            "⚠️ WebRtcCoordinator internal event listener lagged by {} events",
                            n
                        );
                    }
                    Err(tokio::sync::broadcast::error::RecvError::Closed) => {
                        tracing::debug!(
                            "🔌 WebRtcCoordinator internal event listener stopped (channel closed)"
                        );
                        break;
                    }
                }
            }
        })
    }

    /// Start health check task to clean up stale connections
    ///
    /// Periodically checks peer connection states and cleans up:
    /// - Connections in Failed/Closed state for too long (> 1 minutes)
    ///
    /// Note: Disconnected states and ICE restart failures are handled automatically
    /// by the existing ICE restart mechanism, so we only check terminal states here.
    fn spawn_health_check_task(self: &Arc<Self>) -> JoinHandle<()> {
        let coordinator = Arc::downgrade(self);

        tokio::spawn(async move {
            let mut interval = tokio::time::interval(HEALTH_CHECK_INTERVAL);
            interval.tick().await; // Skip first immediate tick

            loop {
                interval.tick().await;

                if let Some(coord) = coordinator.upgrade() {
                    coord.check_and_cleanup_stale_connections().await;
                } else {
                    tracing::debug!("🔌 Health check task stopping (coordinator dropped)");
                    break;
                }
            }

            tracing::info!("🛑 Health check task exited");
        })
    }

    /// Check and cleanup stale peer connections
    ///
    /// This method identifies peers that should be cleaned up based on:
    /// - Failed/Closed state duration exceeding threshold
    ///
    /// Note: ICE restart failures and Disconnected states are handled automatically
    /// by the ICE restart mechanism, so we don't need to check them here.
    async fn check_and_cleanup_stale_connections(&self) {
        let peers_to_cleanup: Vec<(ActrId, String)> = {
            let peers = self.peers.read().await;
            let now = std::time::Instant::now();

            peers
                .iter()
                .filter_map(|(peer_id, state)| {
                    // Get current real-time state from RTCPeerConnection
                    let current_state = state.peer_connection.connection_state();
                    let duration_since_change = now.duration_since(state.last_state_change);

                    // Cleanup condition: Failed/Closed state for too long
                    // These are terminal states that won't recover automatically
                    if matches!(
                        current_state,
                        RTCPeerConnectionState::Failed | RTCPeerConnectionState::Closed
                    ) && duration_since_change > MAX_FAILED_DURATION
                    {
                        let reason = format!(
                            "{:?} for {}s",
                            current_state,
                            duration_since_change.as_secs()
                        );

                        tracing::warn!(
                            "🧹 Marking peer {} for cleanup: {}",
                            peer_id.serial_number,
                            reason
                        );

                        Some((peer_id.clone(), reason))
                    } else {
                        None
                    }
                })
                .collect()
        };

        // Cleanup marked peers
        if !peers_to_cleanup.is_empty() {
            tracing::info!(
                "🧹 Health check: cleaning up {} stale connection(s)",
                peers_to_cleanup.len()
            );

            for (peer_id, reason) in peers_to_cleanup {
                tracing::info!(
                    "🧹 Cleaning up stale connection for peer {}: {}",
                    peer_id.serial_number,
                    reason
                );
                self.cleanup_cancelled_connection(&peer_id).await;
            }
        }
    }

    /// Start signaling coordinator (listen for ActrRelay messages)
    ///
    /// This method starts a background task that continuously listens for messages from SignalingClient
    /// and handles WebRTC-related signaling (Offer/Answer/ICE)
    pub async fn start(self: Arc<Self>) -> RuntimeResult<()> {
        tracing::info!("🚀 WebRtcCoordinator starting signaling loop");

        // Start internal event listener for connection close handling
        self.spawn_internal_event_listener();

        // Start health check task for cleaning up stale connections
        self.spawn_health_check_task();

        let coordinator = self.clone();
        tokio::spawn(async move {
            loop {
                // 1. Receive message from SignalingClient
                match coordinator.signaling_client.receive_envelope().await {
                    Ok(Some(envelope)) => {
                        #[cfg(feature = "opentelemetry")]
                        let (span, remote_ctx) = {
                            let remote_ctx = trace::extract_trace_context(&envelope);
                            let span = tracing::info_span!(
                                "signaling.handle_envelope",
                                envelope_id = envelope.envelope_id,
                                reply_for = ?envelope.reply_for
                            );
                            span.set_parent(remote_ctx.clone());
                            (span, remote_ctx)
                        };

                        let handle_envelope_fut = coordinator.handle_envelope(
                            envelope,
                            #[cfg(feature = "opentelemetry")]
                            remote_ctx,
                        );
                        #[cfg(feature = "opentelemetry")]
                        let handle_envelope_fut = handle_envelope_fut.instrument(span);
                        handle_envelope_fut.await;
                    }
                    Ok(None) => {
                        tracing::info!(
                            "🔌 SignalingClient connection closed, exiting signaling loop"
                        );
                        break;
                    }
                    Err(e) => {
                        tracing::error!("❌ Signaling receive error: {}", e);
                        // Continue loop, don't exit (may be temporary error)
                    }
                }
            }

            tracing::info!("🛑 WebRtcCoordinator signaling loop exited");
        });

        Ok(())
    }

    /// Handle received signaling envelope
    async fn handle_envelope(
        self: &Arc<Self>,
        envelope: SignalingEnvelope,
        #[cfg(feature = "opentelemetry")] remote_ctx: opentelemetry::Context,
    ) {
        // Decode SignalingEnvelope
        match envelope.flow {
            Some(signaling_envelope::Flow::ActrRelay(relay)) => {
                let source = relay.source;
                let target = relay.target;
                #[cfg(feature = "opentelemetry")]
                self.root_context_map
                    .write()
                    .await
                    .insert(source.clone(), remote_ctx);
                match relay.payload {
                    Some(actr_relay::Payload::SessionDescription(sd)) => match sd.r#type() {
                        SdpType::Offer => {
                            tracing::info!("📥 Received Offer from {:?}", source.serial_number);
                            if let Err(e) = self.handle_offer(&source, sd.sdp).await {
                                tracing::error!("❌ Failed to handle Offer: {}", e);
                            }
                        }
                        SdpType::Answer => {
                            tracing::info!("📥 Received Answer from {:?}", source.serial_number);
                            if let Err(e) = self.handle_answer(&source, sd.sdp).await {
                                tracing::error!("❌ Failed to handle Answer: {}", e);
                            }
                        }
                        SdpType::RenegotiationOffer => {
                            tracing::warn!("⚠️ Received RenegotiationOffer, not supported yet");
                        }
                        SdpType::IceRestartOffer => {
                            tracing::info!(
                                "♻️ Received ICE Restart Offer from {:?}",
                                source.serial_number
                            );
                            if let Err(e) = self.handle_ice_restart_offer(&source, sd.sdp).await {
                                tracing::error!("❌ Failed to handle ICE Restart Offer: {}", e);
                            }
                        }
                    },
                    Some(actr_relay::Payload::RoleAssignment(assign)) => {
                        tracing::info!(
                            "🎭 Received RoleAssignment from {:?}, is_offerer={} (source peer)",
                            source.serial_number,
                            assign.is_offerer,
                        );
                        let peer = if source == self.local_id {
                            target.clone()
                        } else {
                            source.clone()
                        };
                        self.handle_role_assignment(assign.clone(), peer).await;
                    }
                    Some(actr_relay::Payload::IceCandidate(ice)) => {
                        tracing::debug!(
                            "📥 Received ICE Candidate from {:?}",
                            source.serial_number
                        );
                        if let Err(e) = self.handle_ice_candidate(&source, ice.candidate).await {
                            tracing::error!("❌ Failed to handle ICE Candidate: {}", e);
                        }
                    }
                    Some(actr_relay::Payload::RoleNegotiation(_)) => {
                        tracing::trace!(
                            "📥 Received RoleNegotiation payload; ignored by WebRtcCoordinator"
                        );
                    }
                    None => {
                        tracing::warn!("⚠️ ActrRelay missing payload");
                    }
                }
            }
            Some(other_flow) => {
                tracing::warn!("⚠️ Ignoring non-ActrRelay flow: {:?}", other_flow);
            }
            None => {
                tracing::warn!("⚠️ SignalingEnvelope missing flow");
            }
        }
    }

    /// Close all peer connections and clear internal peer state.
    ///
    /// This is typically called during shutdown to ensure that all
    /// RTCPeerConnection instances are closed and associated state
    /// (pending ICE candidates, WebRtcConnection state) is dropped.
    pub async fn close_all_peers(&self) -> RuntimeResult<()> {
        tracing::info!("🔻 Closing all WebRTC peer connections");

        // Take snapshot of peers and clear map
        let peers_snapshot: Vec<Arc<RTCPeerConnection>> = {
            let mut peers = self.peers.write().await;
            let conns: Vec<Arc<RTCPeerConnection>> =
                peers.values().map(|p| p.peer_connection.clone()).collect();
            peers.clear();
            conns
        };

        // Clear pending ICE candidates
        {
            let mut pending = self.pending_candidates.write().await;
            pending.clear();
        }

        // Clear root tracing contexts (if enabled)
        #[cfg(feature = "opentelemetry")]
        self.root_context_map.write().await.clear();

        // Close each RTCPeerConnection
        for pc in peers_snapshot {
            tracing::info!("🔻 Closing PeerConnection");

            if let Err(e) = pc.close().await {
                tracing::warn!("⚠️ Failed to close PeerConnection: {}", e);
            } else {
                tracing::info!("✅ PeerConnection closed");
            }
        }

        Ok(())
    }

    /// Send ActrRelay message (internal helper method)
    #[cfg_attr(
        feature = "opentelemetry",
        tracing::instrument(level = "info", skip_all, fields(target = %target.to_string_repr()))
    )]
    async fn send_actr_relay(
        &self,
        target: &ActrId,
        payload: actr_relay::Payload,
    ) -> RuntimeResult<()> {
        let credential = self.credential_state.credential().await;
        let relay = ActrRelay {
            source: self.local_id.clone(),
            credential,
            target: target.clone(),
            payload: Some(payload),
        };

        let flow = signaling_envelope::Flow::ActrRelay(relay);

        let envelope = SignalingEnvelope {
            envelope_version: 1,
            envelope_id: uuid::Uuid::new_v4().to_string(),
            reply_for: None,
            timestamp: prost_types::Timestamp {
                seconds: chrono::Utc::now().timestamp(),
                nanos: 0,
            },
            traceparent: None,
            tracestate: None,
            flow: Some(flow),
        };

        self.signaling_client
            .send_envelope(envelope)
            .await
            .map_err(|e| RuntimeError::Unavailable {
                message: format!("Signaling server unavailable: {e}"),
                target: None,
            })?;

        Ok(())
    }

    /// Initiate connection (create Offer)
    ///
    /// Acts as the initiator, sending a WebRTC connection request to the target peer
    #[cfg_attr(
        feature = "opentelemetry",
        tracing::instrument(level = "info", skip_all, fields(target_id = %target.to_string_repr()))
    )]
    pub async fn initiate_connection(
        self: &Arc<Self>,
        target: &ActrId,
    ) -> RuntimeResult<oneshot::Receiver<()>> {
        tracing::info!(
            "🚀 Initiating P2P connection to {}",
            target.to_string_repr()
        );

        // Role negotiation: determine if we should be offerer or answerer
        let role_result =
            tokio::time::timeout(Duration::from_secs(15), self.negotiate_role(target)).await;

        let is_offerer = match role_result {
            Ok(Ok(v)) => v,
            Ok(Err(e)) => {
                self.peer_negotiation.lock().await.remove(target);
                return Err(e);
            }
            Err(_) => {
                self.peer_negotiation.lock().await.remove(target);
                return Err(RuntimeError::DeadlineExceeded {
                    message: "Role negotiation timeout".to_string(),
                    timeout_ms: 5000,
                });
            }
        };
        tracing::debug!(
            "Role negotiation decided we are {:?} for {}",
            if is_offerer { "offerer" } else { "answerer" },
            target.serial_number
        );
        if !is_offerer {
            let (tx, rx) = oneshot::channel();
            self.peer_negotiation
                .lock()
                .await
                .entry(target.clone())
                .or_default()
                .ready_tx = Some(tx);
            return Ok(rx);
        }

        self.start_offer_connection(target, true).await
    }

    /// Create and send an offer (offerer path). If `skip_negotiation` is true, assumes角色已确定。
    /// This method includes retry logic for initial connection failures.
    #[cfg_attr(
        feature = "opentelemetry",
        tracing::instrument(skip_all, fields(target_id = ?target.to_string_repr()))
    )]
    async fn start_offer_connection(
        self: &Arc<Self>,
        target: &ActrId,
        skip_negotiation: bool,
    ) -> RuntimeResult<oneshot::Receiver<()>> {
        if !skip_negotiation {
            let role_result =
                tokio::time::timeout(Duration::from_secs(15), self.negotiate_role(target)).await;

            let role_result = match role_result {
                Ok(Ok(v)) => v,
                Ok(Err(e)) => {
                    self.peer_negotiation.lock().await.remove(target);
                    return Err(e);
                }
                Err(_) => {
                    self.peer_negotiation.lock().await.remove(target);
                    return Err(RuntimeError::DeadlineExceeded {
                        message: "Role negotiation timeout".to_string(),
                        timeout_ms: 5000,
                    });
                }
            };

            if !role_result {
                tracing::info!(
                    "🎭 Role negotiation decided we are answerer for {}, waiting for offer",
                    target.serial_number
                );
                let (tx, rx) = oneshot::channel();
                self.peer_negotiation
                    .lock()
                    .await
                    .entry(target.clone())
                    .or_default()
                    .ready_tx = Some(tx);
                return Ok(rx);
            }
        }

        // Single connection attempt (no retry)
        tracing::info!("🔄 Starting connection to serial={}", target.serial_number);

        match self.do_single_offer_connection(target).await {
            Ok((ready_rx, webrtc_conn)) => {
                // Wait for connection to be ready with timeout
                match tokio::time::timeout(INITIAL_CONNECTION_TIMEOUT, ready_rx).await {
                    Ok(Ok(())) => {
                        tracing::info!(
                            "✅ Connection established to serial={}",
                            target.serial_number
                        );
                        // Return a new channel that's already signaled
                        let (tx, rx) = oneshot::channel();
                        let _ = tx.send(());
                        return Ok(rx);
                    }
                    Ok(Err(_)) => {
                        tracing::warn!(
                            "⚠️ Connection failed (channel closed) for serial={}",
                            target.serial_number
                        );
                        // Cleanup failed connection attempt
                        self.cleanup_failed_connection(target, webrtc_conn).await;
                        return Err(RuntimeError::Other(anyhow::anyhow!(
                            "Connection ready channel closed"
                        )));
                    }
                    Err(_) => {
                        tracing::warn!(
                            "⚠️ Connection timed out for serial={}",
                            target.serial_number
                        );
                        // Cleanup failed connection attempt
                        self.cleanup_failed_connection(target, webrtc_conn).await;
                        return Err(RuntimeError::DeadlineExceeded {
                            message: "Initial connection timeout".to_string(),
                            timeout_ms: INITIAL_CONNECTION_TIMEOUT.as_millis() as u64,
                        });
                    }
                }
            }
            Err(e) => {
                tracing::warn!(
                    "⚠️ Connection failed for serial={}: {}",
                    target.serial_number,
                    e
                );
                return Err(e);
            }
        }
    }

    /// Cleanup a failed connection attempt
    ///
    /// NOTE: Releases the write lock BEFORE calling close() to avoid blocking
    /// other operations on `peers` during potentially slow close operations.
    async fn cleanup_failed_connection(&self, target: &ActrId, webrtc_conn: WebRtcConnection) {
        // Remove from peers map first, then close outside the lock
        let state_to_close = {
            let mut peers = self.peers.write().await;
            peers.remove(target)
        }; // Lock released here

        if let Some(state) = state_to_close {
            if let Err(e) = state.peer_connection.close().await {
                tracing::warn!(
                    "⚠️ Failed to close peer_connection during cleanup for {}: {}",
                    target.serial_number,
                    e
                );
            }
        }

        // Close WebRtcConnection
        if let Err(e) = webrtc_conn.close().await {
            tracing::warn!(
                "⚠️ Failed to close WebRtcConnection during cleanup for {}: {}",
                target.serial_number,
                e
            );
        }

        // Clear pending candidates
        {
            let mut pending = self.pending_candidates.write().await;
            pending.remove(target);
        }

        tracing::debug!(
            "🧹 Cleaned up failed connection attempt for serial={}",
            target.serial_number
        );
    }

    /// Cleanup a cancelled connection attempt (simpler version without WebRtcConnection)
    ///
    /// Used when connection creation is cancelled before completion.
    ///
    /// IMPORTANT: This method must release all locks before calling close() methods
    /// to avoid deadlock, since close() may trigger events that call this method again.
    async fn cleanup_cancelled_connection(&self, target: &ActrId) {
        tracing::debug!(
            "🧹 Starting cleanup for cancelled connection serial={}",
            target.serial_number
        );

        // 1. Remove from peers map FIRST, release lock, THEN close
        //    This avoids deadlock: close() sends events that may trigger this method again
        let state_to_close = {
            let mut peers = self.peers.write().await;
            peers.remove(target)
        }; // Lock released here

        // Now close outside the lock (close() may send ConnectionClosed event)
        if let Some(state) = state_to_close {
            if let Err(e) = state.peer_connection.close().await {
                tracing::warn!(
                    "⚠️ Failed to close peer_connection during cancel cleanup for {}: {}",
                    target.serial_number,
                    e
                );
            }
            if let Err(e) = state.webrtc_conn.close().await {
                tracing::warn!(
                    "⚠️ Failed to close webrtc_conn during cancel cleanup for {}: {}",
                    target.serial_number,
                    e
                );
            }
        }

        // 2. Clear pending candidates
        self.pending_candidates.write().await.remove(target);

        // 3. Clear negotiation state (role negotiation only, no restart_handle)
        if self.peer_negotiation.lock().await.remove(target).is_some() {
            tracing::debug!(
                "🧹 Clearing negotiation state for serial={}",
                target.serial_number
            );
        }

        // 4. Cancel in-flight restart task if any (from peers lock)
        // Note: We need to abort the task before removing the peer state
        if let Some(peer_state) = self.peers.read().await.get(target) {
            if let Some(ref handle) = peer_state.restart_task_handle {
                handle.abort();
                tracing::debug!(
                    "🧹 Aborted restart task for serial={}",
                    target.serial_number
                );
            }
        }

        tracing::debug!(
            "🧹 Cleaned up cancelled connection for serial={}",
            target.serial_number
        );
    }

    /// Perform a single offer connection attempt (without retry logic)
    async fn do_single_offer_connection(
        self: &Arc<Self>,
        target: &ActrId,
    ) -> RuntimeResult<(oneshot::Receiver<()>, WebRtcConnection)> {
        let peer_connection = self.negotiator.create_peer_connection().await?;
        let peer_connection_arc = Arc::new(peer_connection);

        // 2. Create WebRtcConnection (shares Arc<RTCPeerConnection>) and
        //    install state-change handler with ICE-restart wiring.
        let webrtc_conn = WebRtcConnection::new(
            target.clone(),
            Arc::clone(&peer_connection_arc),
            self.event_broadcaster.sender(),
        );
        self.install_restart_handler(
            webrtc_conn.clone(),
            Arc::clone(&peer_connection_arc),
            target.clone(),
        );

        // 2.5. CRITICAL: Insert peer state early as placeholder to prevent race conditions
        // Create ready channel now, will be populated in step 8
        let (ready_tx, ready_rx) = oneshot::channel();
        {
            let mut peers = self.peers.write().await;
            peers.insert(
                target.clone(),
                PeerState {
                    peer_connection: peer_connection_arc.clone(),
                    webrtc_conn: webrtc_conn.clone(),
                    ready_tx: Some(ready_tx),
                    is_offerer: true,
                    ice_restart_inflight: false,
                    ice_restart_attempts: 0,
                    restart_task_handle: None,
                    last_state_change: std::time::Instant::now(),
                    current_state: RTCPeerConnectionState::New,
                },
            );
            tracing::debug!(
                "🔒 Inserted placeholder peer state for {} (offerer)",
                target.to_string_repr()
            );
        } // Release lock immediately

        // 3. Pre-create negotiated DataChannel for Reliable to trigger ICE gathering
        let _reliable_lane = webrtc_conn
            .get_lane(actr_protocol::PayloadType::RpcReliable)
            .await?;
        tracing::debug!("Pre-created Reliable DataChannel for ICE gathering");

        // 3.5. Pre-create media tracks for sending (MUST be done before creating Offer)
        let _video_track = webrtc_conn
            .add_media_track("video-track-1".to_string(), "VP8", "video")
            .await?;
        tracing::debug!("Pre-created video MediaTrack: video-track-1");

        // 4. Register on_track callback for receiving MediaTrack (WebRTC native media)
        let media_registry = Arc::clone(&self.media_frame_registry);
        let sender_id = target.clone();
        peer_connection_arc.on_track(Box::new(move |track, _receiver, _transceiver| {
            let media_registry = Arc::clone(&media_registry);
            let sender_id = sender_id.clone();
            Box::pin(async move {
                let track_id = track.id();
                tracing::info!(
                    "📹 Received MediaTrack: track_id={}, sender={}",
                    track_id,
                    sender_id.to_string_repr()
                );

                tokio::spawn(async move {
                    loop {
                        match track.read_rtp().await {
                            Ok((rtp_packet, _attributes)) => {
                                let payload_data = rtp_packet.payload.clone();
                                let timestamp = rtp_packet.header.timestamp;
                                let codec = "unknown".to_string();
                                let sample = actr_framework::MediaSample {
                                    data: payload_data,
                                    timestamp,
                                    codec,
                                    media_type: actr_framework::MediaType::Video,
                                };
                                media_registry
                                    .dispatch(&track_id, sample, sender_id.clone())
                                    .await;
                            }
                            Err(e) => {
                                tracing::error!(
                                    "❌ Failed to read RTP from track {}: {}",
                                    track_id,
                                    e
                                );
                                break;
                            }
                        }
                    }
                    tracing::info!("🛑 MediaTrack reader task exited for track_id={}", track_id);
                });
            })
        }));

        // 5. Set ICE candidate callback (local ICE candidate collection)
        let coordinator = Arc::downgrade(self);
        let target_id = target.clone();
        #[cfg(feature = "opentelemetry")]
        let root_context_map = self.root_context_map.clone();
        peer_connection_arc.on_ice_candidate(Box::new(
            move |candidate: Option<RTCIceCandidate>| {
                let coordinator = coordinator.clone();
                let target_id = target_id.clone();
                #[cfg(feature = "opentelemetry")]
                let root_context_map = root_context_map.clone();
                Box::pin(async move {
                    if let Some(cand) = candidate {
                        if let Some(coord) = coordinator.upgrade() {
                            let candidate_json = match cand.to_json() {
                                Ok(json) => json.candidate,
                                Err(e) => {
                                    tracing::error!("❌ ICE Candidate serialization failed: {}", e);
                                    return;
                                }
                            };

                            let ice_candidate = actr_protocol::IceCandidate {
                                candidate: candidate_json,
                                sdp_mid: None,
                                sdp_mline_index: None,
                                username_fragment: None,
                            };

                            let payload = actr_relay::Payload::IceCandidate(ice_candidate);

                            // Get root context at callback execution time (not at setup time)
                            #[cfg(feature = "opentelemetry")]
                            let span = {
                                let span = tracing::info_span!(
                                    "send_ice_candidate",
                                    target_id = %target_id.to_string_repr()
                                );
                                if let Some(ctx) =
                                    root_context_map.read().await.get(&target_id).cloned()
                                {
                                    span.set_parent(ctx);
                                } else {
                                    tracing::warn!(
                                        "⚠️ No root context found for target_id={}",
                                        target_id.to_string_repr()
                                    );
                                }
                                span
                            };
                            let send_actr_relay_fut = coord.send_actr_relay(&target_id, payload);
                            #[cfg(feature = "opentelemetry")]
                            let send_actr_relay_fut = send_actr_relay_fut.instrument(span);
                            if let Err(e) = send_actr_relay_fut.await {
                                tracing::error!("❌ Failed to send ICE Candidate: {}", e);
                            } else {
                                tracing::debug!("✅ Sent ICE Candidate");
                            }
                        }
                    } else {
                        tracing::debug!("❌ ICE Candidate is None");
                    }
                })
            },
        ));

        // 6. Create Offer
        let offer_sdp = self.negotiator.create_offer(&peer_connection_arc).await?;

        // 8. Send Offer via signaling server
        let session_desc = actr_protocol::SessionDescription {
            r#type: SdpType::Offer as i32,
            sdp: offer_sdp,
        };
        let payload = actr_relay::Payload::SessionDescription(session_desc);
        self.send_actr_relay(target, payload).await?;

        tracing::info!("✅ Sent Offer to {}", target.to_string_repr());

        // 10. Start receive loop (receive and aggregate messages from this peer)
        self.start_peer_receive_loop(target.clone(), webrtc_conn.clone())
            .await;

        Ok((ready_rx, webrtc_conn))
    }

    /// Handle received Offer (passive side)
    ///
    /// Called when receiving a connection request from another peer.
    /// Supports both initial negotiation and renegotiation.
    #[cfg_attr(
        feature = "opentelemetry",
        tracing::instrument(level = "info", skip_all, fields(remote_id = %from.to_string_repr()))
    )]
    async fn handle_offer(self: &Arc<Self>, from: &ActrId, offer_sdp: String) -> RuntimeResult<()> {
        // ========== PrepareForIncomingOffer: Clean up existing connection if any ==========
        let existing_peer = {
            let peers = self.peers.read().await;
            peers.contains_key(from)
        };

        if existing_peer {
            tracing::info!(
                "🔄 Existing connection found for serial={}, preparing for new Offer",
                from.serial_number
            );

            // Clean up old connection using unified cleanup method
            self.cleanup_cancelled_connection(from).await;
        }
        // ========== PrepareForIncomingOffer END ==========

        tracing::info!("📥 Handling Offer from serial={}", from.serial_number);

        // 1. Create RTCPeerConnection
        let peer_connection = self.negotiator.create_peer_connection().await?;
        let peer_connection_arc = Arc::new(peer_connection);

        // 2. Create WebRtcConnection (shares Arc<RTCPeerConnection>)
        let webrtc_conn = WebRtcConnection::new(
            from.clone(),
            Arc::clone(&peer_connection_arc),
            self.event_broadcaster.sender(),
        );

        // CRITICAL: Insert peer state immediately as a placeholder to prevent race conditions.
        // This prevents ensure_connection from creating a duplicate connection while we're
        // still setting up callbacks and negotiating the connection.
        // The state will be updated later after Answer is sent (step 6).
        {
            let mut peers = self.peers.write().await;
            peers.insert(
                from.clone(),
                PeerState {
                    peer_connection: peer_connection_arc.clone(),
                    webrtc_conn: webrtc_conn.clone(),
                    ready_tx: None,
                    is_offerer: false,
                    ice_restart_inflight: false,
                    ice_restart_attempts: 0,
                    restart_task_handle: None,
                    last_state_change: std::time::Instant::now(),
                    current_state: RTCPeerConnectionState::New,
                },
            );
            tracing::debug!(
                "🔒 Inserted placeholder peer state for {} (answerer)",
                from.to_string_repr()
            );
        } // Release lock immediately

        // 3. Register state change handler (combines cleanup + ready notification)
        // NOTE: on_peer_connection_state_change can only have ONE callback, so we combine:
        //   - WebRtcConnection.handle_state_change() for cleanup on terminal states
        //   - Ready notification when Connected (answerer side)
        let webrtc_conn_for_state = webrtc_conn.clone();
        let coord_weak_for_state = Arc::downgrade(self);
        let from_id_for_state = from.clone();
        peer_connection_arc.on_peer_connection_state_change(Box::new(
            move |state: RTCPeerConnectionState| {
                let webrtc_conn = webrtc_conn_for_state.clone();
                let coord_weak = coord_weak_for_state.clone();
                let peer_id = from_id_for_state.clone();
                Box::pin(async move {
                    // First: run WebRtcConnection's state change handler (cleanup logic)
                    webrtc_conn.handle_state_change(state).await;

                    // Update state tracking for health check
                    if let Some(coord) = coord_weak.upgrade() {
                        let mut peers = coord.peers.write().await;
                        if let Some(peer_state) = peers.get_mut(&peer_id) {
                            peer_state.current_state = state;
                            peer_state.last_state_change = std::time::Instant::now();
                        }
                        drop(peers); // Release lock
                    }
                })
            },
        ));

        // 4. Register on_data_channel handler to reuse negotiated channels created by the offerer
        let conn_for_data_channel = webrtc_conn.clone();

        let from_id_for_data_channel = from.clone();
        let coord_weak_for_state = Arc::downgrade(self);
        let message_tx = self.message_tx.clone();
        peer_connection_arc.on_data_channel(Box::new(move |dc: Arc<RTCDataChannel>| {
            let conn = conn_for_data_channel.clone();
            let coord_weak = coord_weak_for_state.clone();
            let peer_id = from_id_for_data_channel.clone();
            let message_tx = message_tx.clone();
            Box::pin(async move {
                let channel_id = dc.id();
                let label = dc.label();
                let dc_for_registration = Arc::clone(&dc);

                let payload_type = PayloadType::from_str_name(&label);

                if let Some(coord) = coord_weak.upgrade() {
                    let ready_tx = {
                        let mut neg = coord.peer_negotiation.lock().await;
                        neg.get_mut(&peer_id).and_then(|s| s.ready_tx.take())
                    };
                    if let Some(tx) = ready_tx {
                        tracing::info!(
                            "✅ [Answerer] Connection ready, sending notification for {}",
                            peer_id.serial_number
                        );
                        let _ = tx.send(());
                    }
                }

                match payload_type {
                    Some(pt) => {
                        if let Err(e) = conn
                            .register_received_data_channel(dc_for_registration, pt, message_tx)
                            .await
                        {
                            tracing::warn!(
                                "❌ Failed to register received DataChannel label={} id={}: {}",
                                label,
                                channel_id,
                                e
                            );
                        } else {
                            tracing::debug!(
                                "📨 Registered DataChannel from offerer label={} id={}",
                                label,
                                channel_id
                            );
                        }
                    }
                    None => {
                        tracing::warn!(
                            "❓ Ignoring DataChannel with unmapped id={} label={}",
                            channel_id,
                            label
                        );
                    }
                }
            })
        }));

        // 3.5. Pre-create media tracks for sending (MUST be done before creating Answer)
        // Create a default video track for demo purposes
        let _video_track = webrtc_conn
            .add_media_track("video-track-1".to_string(), "VP8", "video")
            .await?;
        tracing::debug!("Pre-created video MediaTrack: video-track-1 (answerer)");

        // 4. Register on_track callback for receiving MediaTrack (WebRTC native media)
        let media_registry = Arc::clone(&self.media_frame_registry);
        let sender_id = from.clone();
        peer_connection_arc.on_track(Box::new(move |track, _receiver, _transceiver| {
            let media_registry = Arc::clone(&media_registry);
            let sender_id = sender_id.clone();
            Box::pin(async move {
                let track_id = track.id();
                tracing::info!(
                    "📹 Received MediaTrack: track_id={}, sender={}",
                    track_id,
                    sender_id.to_string_repr()
                );

                // Spawn task to read RTP packets from track
                tokio::spawn(async move {
                    loop {
                        // Read RTP packet from track
                        match track.read_rtp().await {
                            Ok((rtp_packet, _attributes)) => {
                                // Extract payload and timestamp
                                let payload_data = rtp_packet.payload.clone();
                                let timestamp = rtp_packet.header.timestamp;

                                // TODO: Extract codec from track (for now use placeholder)
                                let codec = "unknown".to_string();

                                // Create MediaSample
                                let sample = actr_framework::MediaSample {
                                    data: payload_data,
                                    timestamp,
                                    codec,
                                    media_type: actr_framework::MediaType::Video, // TODO: detect from track
                                };

                                // Dispatch to registered callback
                                media_registry
                                    .dispatch(&track_id, sample, sender_id.clone())
                                    .await;
                            }
                            Err(e) => {
                                tracing::error!(
                                    "❌ Failed to read RTP from track {}: {}",
                                    track_id,
                                    e
                                );
                                break;
                            }
                        }
                    }
                    tracing::info!("🛑 MediaTrack reader task exited for track_id={}", track_id);
                });
            })
        }));

        // 5. Set ICE candidate callback (local ICE candidate collection)
        let coordinator = Arc::downgrade(self);
        let target_id = from.clone();
        #[cfg(feature = "opentelemetry")]
        let root_context_map = self.root_context_map.clone();
        peer_connection_arc.on_ice_candidate(Box::new(
            move |candidate: Option<RTCIceCandidate>| {
                let coordinator = coordinator.clone();
                let target_id = target_id.clone();
                #[cfg(feature = "opentelemetry")]
                let root_context_map = root_context_map.clone();
                Box::pin(async move {
                    if let Some(cand) = candidate {
                        if let Some(coord) = coordinator.upgrade() {
                            // Convert RTCIceCandidate to JSON string (webrtc crate's standard method)
                            let candidate_json = match cand.to_json() {
                                Ok(json) => json.candidate,
                                Err(e) => {
                                    tracing::error!("❌ ICE Candidate serialization failed: {}", e);
                                    return;
                                }
                            };

                            let ice_candidate = actr_protocol::IceCandidate {
                                candidate: candidate_json,
                                sdp_mid: None,
                                sdp_mline_index: None,
                                username_fragment: None,
                            };

                            let payload = actr_relay::Payload::IceCandidate(ice_candidate);

                            // Get root context at callback execution time (not at setup time)
                            #[cfg(feature = "opentelemetry")]
                            let span = {
                                let span = tracing::info_span!(
                                    "send_ice_candidate",
                                    target_id = %target_id.to_string_repr()
                                );
                                if let Some(ctx) =
                                    root_context_map.read().await.get(&target_id).cloned()
                                {
                                    span.set_parent(ctx);
                                } else {
                                    tracing::warn!(
                                        "⚠️ No root context found for target_id={}",
                                        target_id.to_string_repr()
                                    );
                                }
                                span
                            };
                            let send_actr_relay_fut = coord.send_actr_relay(&target_id, payload);
                            #[cfg(feature = "opentelemetry")]
                            let send_actr_relay_fut = send_actr_relay_fut.instrument(span);
                            if let Err(e) = send_actr_relay_fut.await {
                                tracing::error!("❌ Failed to send ICE Candidate: {}", e);
                            }
                            tracing::debug!(
                                "🔄 Handle offer Sent ICE Candidate to serial={}",
                                target_id.serial_number
                            );
                        }
                    }
                })
            },
        ));

        // 5. Create Answer
        let answer_sdp = self
            .negotiator
            .create_answer(&peer_connection_arc, offer_sdp)
            .await?;

        // 7. Send Answer via signaling server
        let session_desc = actr_protocol::SessionDescription {
            r#type: SdpType::Answer as i32,
            sdp: answer_sdp,
        };
        let payload = actr_relay::Payload::SessionDescription(session_desc);
        self.send_actr_relay(from, payload).await?;

        tracing::info!("✅ Sent Answer to {}", from.to_string_repr());

        // 8. Flush any buffered ICE candidates (remote description is now set)
        self.flush_pending_candidates(from, &peer_connection_arc)
            .await?;

        // Note: ready notification is sent in on_data_channel callback
        // when DataChannel is actually registered (see above)

        Ok(())
    }

    /// Handle received Answer (initiator side)
    ///
    /// Supports both initial negotiation and renegotiation answers.
    #[cfg_attr(
        feature = "opentelemetry",
        tracing::instrument(
            level = "info",
            skip_all,
            fields(
                remote.id = %from.to_string_repr(),
                answer_len = answer_sdp.len()
            )
        )
    )]
    async fn handle_answer(
        self: &Arc<Self>,
        from: &ActrId,
        answer_sdp: String,
    ) -> RuntimeResult<()> {
        // Get corresponding PeerConnection and ready_tx
        let (peer_connection, ready_tx, is_renegotiation) = {
            let mut peers = self.peers.write().await;
            tracing::info!(
                "🔍 [LOOKUP] Searching for: id={}, total peers={}",
                from.to_string_repr(),
                peers.len()
            );
            for (k, _) in peers.iter() {
                tracing::info!("   📌 [LOOKUP] Stored: id={}", k.to_string_repr());
            }
            let state = peers.get_mut(from).ok_or_else(|| {
                RuntimeError::Other(anyhow::anyhow!("Peer not found: {}", from.to_string_repr()))
            })?;

            let pc = state.peer_connection.clone();
            let tx = state.ready_tx.take();
            let is_reneg = tx.is_none(); // If ready_tx already taken, this is renegotiation
            (pc, tx, is_reneg)
        };

        if is_renegotiation {
            tracing::info!(
                "🔄 Handling renegotiation Answer from {}",
                from.to_string_repr()
            );
        } else {
            tracing::info!("📥 Handling initial Answer from {}", from.to_string_repr());
        }

        // Handle Answer (set remote SDP)
        self.negotiator
            .handle_answer(&peer_connection, answer_sdp)
            .await?;

        // Flush any buffered ICE candidates (remote description is now set)
        self.flush_pending_candidates(from, &peer_connection)
            .await?;

        tracing::info!(
            "✅ WebRTC connection negotiation completed: {}",
            from.to_string_repr()
        );

        // Wait for PeerConnection to actually connect (max 5 seconds)
        let pc_clone = peer_connection.clone();
        let peers = Arc::clone(&self.peers);
        let from_id = from.clone();
        tokio::spawn(async move {
            let start = tokio::time::Instant::now();
            loop {
                let state = pc_clone.connection_state();
                if state == webrtc::peer_connection::peer_connection_state::RTCPeerConnectionState::Connected {
                    tracing::info!("✅ PeerConnection fully connected");
                    // Mark ICE restart attempt complete
                    let mut peers_guard = peers.write().await;
                    if let Some(s) = peers_guard.get_mut(&from_id) {
                        s.ice_restart_inflight = false;
                        s.ice_restart_attempts = 0;
                    }
                    break;
                }
                if state == webrtc::peer_connection::peer_connection_state::RTCPeerConnectionState::Failed {
                    tracing::error!("❌ PeerConnection failed");
                    return;
                }
                if start.elapsed() > std::time::Duration::from_secs(5) {
                    tracing::warn!("⚠️ PeerConnection connection timeout (5s)");
                    break;
                }
                tokio::time::sleep(std::time::Duration::from_millis(10)).await;
            }

            // Notify initiate_connection that connection is ready
            if let Some(tx) = ready_tx {
                let _ = tx.send(());
            }
        });

        Ok(())
    }

    /// Flush buffered ICE candidates for a peer
    ///
    /// Called after remote description is set, to add any candidates that arrived early
    async fn flush_pending_candidates(
        &self,
        peer_id: &ActrId,
        peer_connection: &RTCPeerConnection,
    ) -> RuntimeResult<()> {
        // Extract buffered candidates for this peer
        let candidates = {
            let mut pending = self.pending_candidates.write().await;
            pending.remove(peer_id)
        };

        if let Some(candidates) = candidates {
            tracing::debug!(
                "🔄 Flushing {} buffered ICE candidates for {:?}",
                candidates.len(),
                peer_id
            );

            for candidate in candidates {
                if let Err(e) = self
                    .negotiator
                    .add_ice_candidate(peer_connection, candidate)
                    .await
                {
                    tracing::warn!("⚠️ Failed to add buffered ICE candidate: {}", e);
                }
            }
        }

        Ok(())
    }

    /// Handle received ICE Candidate
    #[cfg_attr(
        feature = "opentelemetry",
        tracing::instrument(
            level = "trace",
            skip_all,
            fields(
                remote.id = %from.to_string_repr(),
                candidate_len = candidate.len()
            )
        )
    )]
    async fn handle_ice_candidate(
        self: &Arc<Self>,
        from: &ActrId,
        candidate: String,
    ) -> RuntimeResult<()> {
        tracing::trace!("📥 Received ICE Candidate from {}", from.to_string_repr());

        // DEBUG: Temporarily disable candidate filtering for local testing
        // TODO: Re-enable proper filtering for production
        // if !is_ipv4_candidate_allowed(&candidate) {
        //     tracing::debug!("🚫 Ignoring ICE candidate from {:?}: {}", from, candidate);
        //     return Ok(());
        // }

        // Try to get peer and check if remote description is set
        let peer_opt = {
            let peers = self.peers.read().await;
            peers.get(from).map(|state| state.peer_connection.clone())
        };

        match peer_opt {
            Some(peer_connection) => {
                // Check if remote description is set
                if peer_connection.remote_description().await.is_some() {
                    // Can add candidate immediately
                    self.negotiator
                        .add_ice_candidate(&peer_connection, candidate)
                        .await?;
                    tracing::trace!("✅ Added ICE Candidate from {}", from.to_string_repr());
                } else {
                    // Buffer for later (remote description not yet set)
                    self.pending_candidates
                        .write()
                        .await
                        .entry(from.clone())
                        .or_insert_with(Vec::new)
                        .push(candidate);
                    tracing::debug!(
                        "🔖 Buffered ICE candidate from {:?} (remote description not yet set)",
                        from
                    );
                }
            }
            None => {
                // Buffer for when peer is created
                self.pending_candidates
                    .write()
                    .await
                    .entry(from.clone())
                    .or_insert_with(Vec::new)
                    .push(candidate);
                tracing::debug!(
                    "🔖 Buffered ICE candidate from {:?} (peer not yet created)",
                    from
                );
            }
        }

        Ok(())
    }

    /// Start peer receive loop
    ///
    /// Starts a background task for each peer to receive messages from WebRtcConnection and aggregate to a unified message_tx
    ///
    /// IMPORTANT: We need to listen to ALL PayloadTypes, not just RpcReliable:
    /// - RpcReliable, RpcSignal: for RPC messages
    /// - StreamReliable, StreamLatencyFirst: for DataStream messages
    async fn start_peer_receive_loop(&self, peer_id: ActrId, webrtc_conn: WebRtcConnection) {
        let message_tx = self.message_tx.clone();

        // Listen to all relevant PayloadTypes
        let payload_types = vec![
            PayloadType::RpcReliable,
            PayloadType::RpcSignal,
            PayloadType::StreamReliable,
            PayloadType::StreamLatencyFirst,
        ];

        for payload_type in payload_types {
            let message_tx_clone = message_tx.clone();
            let peer_id_clone = peer_id.clone();
            let webrtc_conn_clone = webrtc_conn.clone();

            tokio::spawn(async move {
                tracing::debug!(
                    "📡 Starting receive loop for peer {:?}, PayloadType: {:?}",
                    peer_id_clone,
                    payload_type
                );

                // Get Lane for this PayloadType
                let lane = match webrtc_conn_clone.get_lane(payload_type).await {
                    Ok(l) => l,
                    Err(e) => {
                        tracing::error!(
                            "❌ Failed to get Lane for {:?}, PayloadType {:?}: {}",
                            peer_id_clone,
                            payload_type,
                            e
                        );
                        return;
                    }
                };

                // Continuously receive messages
                loop {
                    match lane.recv().await {
                        Ok(data) => {
                            tracing::debug!(
                                "📨 Received message from {:?} (PayloadType: {:?}): {} bytes",
                                peer_id_clone,
                                payload_type,
                                data.len()
                            );

                            // Serialize peer_id as bytes
                            let peer_id_bytes = peer_id_clone.encode_to_vec();

                            // Send to aggregation channel (include PayloadType)
                            if let Err(e) =
                                message_tx_clone.send((peer_id_bytes, data, payload_type))
                            {
                                tracing::error!("❌ Message aggregation failed: {:?}", e);
                                break;
                            }
                        }
                        Err(e) => {
                            tracing::warn!(
                                "❌ Peer {:?} message receive failed (PayloadType: {:?}): {}",
                                peer_id_clone,
                                payload_type,
                                e
                            );
                            break;
                        }
                    }
                }

                tracing::debug!(
                    "📡 Receive loop exited for peer {:?}, PayloadType: {:?}",
                    peer_id_clone,
                    payload_type
                );
            });
        }
    }

    /// Send message to specified peer
    ///
    /// If connection doesn't exist, automatically initiates WebRTC connection and waits for it to be ready.
    /// Supports retry with exponential backoff on transient errors.
    #[cfg_attr(
        feature = "opentelemetry",
        tracing::instrument(skip_all, fields(target_id = ?target.to_string_repr()))
    )]
    pub(crate) async fn send_message(
        self: &Arc<Self>,
        target: &ActrId,
        data: &[u8],
    ) -> RuntimeResult<()> {
        const MAX_RETRIES: u32 = 3;
        const OVERALL_TIMEOUT: Duration = Duration::from_secs(30);

        tracing::debug!("📤 Sending message to {:?}: {} bytes", target, data.len());

        // Wrap entire operation with overall timeout
        let result = tokio::time::timeout(
            OVERALL_TIMEOUT,
            self.send_message_with_retry(target, data, MAX_RETRIES),
        )
        .await;

        match result {
            Ok(inner_result) => inner_result,
            Err(_) => {
                tracing::error!(
                    "⏰ Overall timeout ({}s) exceeded for send_message to {}",
                    OVERALL_TIMEOUT.as_secs(),
                    target.to_string_repr()
                );
                self.cleanup_cancelled_connection(target).await;
                Err(RuntimeError::DeadlineExceeded {
                    message: format!(
                        "send_message overall timeout ({}s)",
                        OVERALL_TIMEOUT.as_secs()
                    ),
                    timeout_ms: OVERALL_TIMEOUT.as_millis() as u64,
                })
            }
        }
    }

    /// Inner implementation of send_message with retry logic
    async fn send_message_with_retry(
        self: &Arc<Self>,
        target: &ActrId,
        data: &[u8],
        max_retries: u32,
    ) -> RuntimeResult<()> {
        let mut backoff = ExponentialBackoff::new(
            Duration::from_millis(1), // initial delay
            Duration::from_secs(10),  // max delay
            None,                     // no limit (we control manually)
        );

        let mut last_error = None;

        for attempt in 0..=max_retries {
            // Wait before retry (skip first attempt)
            if attempt > 0 {
                let delay = backoff.next().unwrap_or(Duration::from_secs(5));
                tracing::info!(
                    "🔄 Retrying send_message to {} (attempt {}/{}, delay {:?})",
                    target.to_string_repr(),
                    attempt + 1,
                    max_retries + 1,
                    delay
                );
                tokio::time::sleep(delay).await;
            }

            match self.try_send_message_once(target, data).await {
                Ok(()) => return Ok(()),
                Err(e) => {
                    // Only retry on transient errors
                    let should_retry = matches!(
                        &e,
                        RuntimeError::DeadlineExceeded { .. } | RuntimeError::Other(_)
                    );

                    if !should_retry {
                        return Err(e);
                    }

                    tracing::warn!(
                        "⚠️ send_message attempt {}/{} failed: {}",
                        attempt + 1,
                        max_retries + 1,
                        e
                    );
                    last_error = Some(e);

                    // Cleanup connection before retry (might be stale)
                    self.cleanup_cancelled_connection(target).await;
                }
            }
        }

        // All retries exhausted
        Err(last_error.unwrap_or_else(|| {
            RuntimeError::Other(anyhow::anyhow!("send_message failed after all retries"))
        }))
    }

    /// Single attempt to send a message
    async fn try_send_message_once(
        self: &Arc<Self>,
        target: &ActrId,
        data: &[u8],
    ) -> RuntimeResult<()> {
        // Check if connection exists or is being established
        let has_connection = loop {
            let state = {
                let peers = self.peers.read().await;
                peers
                    .get(target)
                    .map(|s| (s.current_state, s.last_state_change))
            };

            match state {
                Some((
                    RTCPeerConnectionState::New | RTCPeerConnectionState::Connecting,
                    started,
                )) => {
                    // Connection is being established, check if it's still fresh
                    if started.elapsed() < INITIAL_CONNECTION_TIMEOUT {
                        // Wait a bit and check again
                        tracing::debug!(
                            "⏳ Connection to {} is being established, waiting...",
                            target.to_string_repr()
                        );
                        tokio::time::sleep(Duration::from_millis(100)).await;
                        continue;
                    } else {
                        // Connecting timeout, treat as not connected
                        tracing::warn!(
                            "⏰ Connection to {} timed out while connecting",
                            target.to_string_repr()
                        );
                        break false;
                    }
                }
                Some((RTCPeerConnectionState::Connected, _)) => {
                    // Connection exists and is ready
                    break true;
                }
                Some(_) => {
                    // Connection exists but in other state (Disconnected/Failed/Closed)
                    // Let initiate_connection handle it
                    break false;
                }
                None => {
                    // No connection exists
                    break false;
                }
            }
        };

        #[cfg(feature = "opentelemetry")]
        let _ = self
            .root_context_map
            .write()
            .await
            .insert(target.clone(), tracing::Span::current().context());

        // If connection doesn't exist, initiate connection
        if !has_connection {
            tracing::info!(
                "🔗 First send to {:?}, initiating role negotiation + WebRTC connection",
                target.serial_number
            );

            let ready_rx = self.initiate_connection(target).await?;
            tracing::debug!(?ready_rx, "ready_rx");

            // Wait for connection to be ready (10s timeout for single attempt)
            match tokio::time::timeout(Duration::from_secs(10), ready_rx).await {
                Ok(Ok(())) => {
                    tracing::info!("✅ WebRTC connection ready: {}", target.to_string_repr());
                }
                Ok(Err(_)) => {
                    return Err(RuntimeError::Other(anyhow::anyhow!(
                        "Connection establishment failed (channel closed)"
                    )));
                }
                Err(_) => {
                    return Err(RuntimeError::DeadlineExceeded {
                        message: "Connection establishment timeout".to_string(),
                        timeout_ms: 10000,
                    });
                }
            }
        }

        // Get corresponding WebRtcConnection
        let webrtc_conn = {
            let peers = self.peers.read().await;
            peers
                .get(target)
                .map(|state| state.webrtc_conn.clone())
                .ok_or_else(|| {
                    RuntimeError::Other(anyhow::anyhow!("Peer connection not found: {target:?}"))
                })?
        };

        // Get Reliable Lane
        let lane = webrtc_conn
            .get_lane(PayloadType::RpcReliable)
            .await
            .map_err(|e| RuntimeError::Other(anyhow::anyhow!("Failed to get Lane: {e}")))?;

        // Send message (convert to Bytes)
        lane.send(Bytes::copy_from_slice(data))
            .await
            .map_err(|e| RuntimeError::Other(anyhow::anyhow!("Failed to send message: {e}")))?;

        Ok(())
    }

    /// Receive message (aggregated from all peers)
    /// Receive message with PayloadType information
    ///
    /// Returns: Option<(sender_id_bytes, message_data, payload_type)>
    pub async fn receive_message(&self) -> RuntimeResult<Option<(Vec<u8>, Bytes, PayloadType)>> {
        let mut rx = self.message_rx.lock().await;
        Ok(rx.recv().await)
    }

    /// Create WebRTC connection (factory method)
    ///
    /// For ConnectionFactory, creates a WebRTC connection to the specified Dest.
    /// If connection already exists, returns it directly; otherwise initiates new connection and waits for it to be ready.
    /// Supports retry with exponential backoff on timeout or channel errors.
    /// The entire method has a 30-second overall timeout.
    ///
    /// # Arguments
    /// - `dest`: destination (must be Actor type)
    /// - `cancel_token`: optional cancellation token to terminate the operation
    ///
    /// # Returns
    /// - `Ok(WebRtcConnection)`: ready WebRTC connection
    /// - `Err`: WebRTC only supports Actor targets, connection cancelled, or connection establishment failed
    #[cfg_attr(
        feature = "opentelemetry",
        tracing::instrument(skip_all, fields(target_id = ?dest.as_actor_id().map(|id| id.to_string_repr())))
    )]
    pub async fn create_connection(
        self: &Arc<Self>,
        dest: &crate::transport::Dest,
        cancel_token: Option<CancellationToken>,
    ) -> RuntimeResult<WebRtcConnection> {
        // Overall timeout for the entire create_connection operation
        const OVERALL_TIMEOUT: Duration = Duration::from_secs(30);

        // Extract target_id first (before timeout wrapper) for cleanup
        let target_id = dest.as_actor_id().ok_or_else(|| {
            RuntimeError::ConfigurationError(
                "WebRTC only supports Actor targets, not Shell".to_string(),
            )
        })?;

        // Wrap the entire operation with overall timeout
        let result = tokio::time::timeout(
            OVERALL_TIMEOUT,
            self.create_connection_inner(dest, cancel_token.clone()),
        )
        .await;

        match result {
            Ok(inner_result) => inner_result,
            Err(_) => {
                // Overall timeout exceeded
                tracing::error!(
                    "⏰ [Factory] Overall timeout ({}s) exceeded for connection to {}",
                    OVERALL_TIMEOUT.as_secs(),
                    target_id.to_string_repr()
                );
                self.cleanup_cancelled_connection(target_id).await;
                Err(RuntimeError::DeadlineExceeded {
                    message: format!(
                        "WebRTC connection creation overall timeout ({}s)",
                        OVERALL_TIMEOUT.as_secs()
                    ),
                    timeout_ms: OVERALL_TIMEOUT.as_millis() as u64,
                })
            }
        }
    }

    /// Inner implementation of create_connection without overall timeout
    async fn create_connection_inner(
        self: &Arc<Self>,
        dest: &crate::transport::Dest,
        cancel_token: Option<CancellationToken>,
    ) -> RuntimeResult<WebRtcConnection> {
        // Check cancellation at entry
        if let Some(ref token) = cancel_token {
            if token.is_cancelled() {
                return Err(RuntimeError::Other(anyhow::anyhow!(
                    "Connection creation cancelled before starting"
                )));
            }
        }

        // 1. Check if dest is Actor
        let target_id = dest.as_actor_id().ok_or_else(|| {
            RuntimeError::ConfigurationError(
                "WebRTC only supports Actor targets, not Shell".to_string(),
            )
        })?;

        tracing::debug!(
            "🏭 [Factory] Creating WebRTC connection to {:?}",
            target_id.to_string_repr()
        );

        // 2. Check if connection already exists
        {
            let peers = self.peers.read().await;
            if let Some(state) = peers.get(target_id) {
                tracing::debug!(
                    "♻️ [Factory] Reusing existing WebRTC connection: {:?}",
                    target_id.to_string_repr()
                );
                return Ok(state.webrtc_conn.clone());
            }
        }

        // 3. Retry loop with exponential backoff (max 3 retries)
        const MAX_RETRIES: u32 = 3;
        let mut backoff = ExponentialBackoff::new(
            Duration::from_secs(5),  // initial delay
            Duration::from_secs(15), // max delay
            None,                    // no limit (we control manually)
        );

        let mut last_error = None;

        for attempt in 0..=MAX_RETRIES {
            // Check cancellation before each attempt
            if let Some(ref token) = cancel_token {
                if token.is_cancelled() {
                    return Err(RuntimeError::Other(anyhow::anyhow!(
                        "Connection creation cancelled"
                    )));
                }
            }

            // Wait before retry (skip first attempt)
            if attempt > 0 {
                let delay = backoff.next().unwrap_or(Duration::from_secs(10));
                tracing::info!(
                    "🔄 [Factory] Retrying connection to {} (attempt {}/{}, delay {:?})",
                    target_id.to_string_repr(),
                    attempt + 1,
                    MAX_RETRIES + 1,
                    delay
                );

                // Interruptible sleep with cancellation
                if let Some(ref token) = cancel_token {
                    tokio::select! {
                        biased;
                        _ = token.cancelled() => {
                            self.cleanup_cancelled_connection(target_id).await;
                            return Err(RuntimeError::Other(anyhow::anyhow!(
                                "Connection creation cancelled during retry wait"
                            )));
                        }
                        _ = tokio::time::sleep(delay) => {}
                    }
                } else {
                    tokio::time::sleep(delay).await;
                }
            } else {
                tracing::info!(
                    "🔨 [Factory] Initiating new WebRTC connection: {:?}",
                    target_id.to_string_repr()
                );
            }

            // Attempt connection
            match self
                .try_create_connection_once(target_id, cancel_token.as_ref())
                .await
            {
                Ok(conn) => return Ok(conn),
                Err(e) => {
                    // Check if this is a cancellation error - don't retry
                    if let Some(ref token) = cancel_token {
                        if token.is_cancelled() {
                            return Err(e);
                        }
                    }

                    // Only retry on timeout or transient errors
                    let should_retry = matches!(
                        &e,
                        RuntimeError::DeadlineExceeded { .. } | RuntimeError::Other(_)
                    );

                    if !should_retry {
                        return Err(e);
                    }

                    tracing::warn!(
                        "⚠️ [Factory] Connection attempt {}/{} failed: {}",
                        attempt + 1,
                        MAX_RETRIES + 1,
                        e
                    );
                    last_error = Some(e);

                    // Cleanup failed connection before retry
                    self.cleanup_cancelled_connection(target_id).await;
                }
            }
        }

        // All retries exhausted
        Err(last_error.unwrap_or_else(|| {
            RuntimeError::Other(anyhow::anyhow!("Connection failed after all retries"))
        }))
    }

    /// Single attempt to create a WebRTC connection
    async fn try_create_connection_once(
        self: &Arc<Self>,
        target_id: &ActrId,
        cancel_token: Option<&CancellationToken>,
    ) -> RuntimeResult<WebRtcConnection> {
        #[cfg(feature = "opentelemetry")]
        self.root_context_map
            .write()
            .await
            .insert(target_id.clone(), tracing::Span::current().context());

        let ready_rx = self.initiate_connection(target_id).await?;

        // Check cancellation after initiation
        if let Some(token) = cancel_token {
            if token.is_cancelled() {
                self.cleanup_cancelled_connection(target_id).await;
                return Err(RuntimeError::Other(anyhow::anyhow!(
                    "Connection creation cancelled after initiation"
                )));
            }
        }

        // Wait for connection to be ready (10s timeout) with cancellation support
        let timeout_duration = std::time::Duration::from_secs(10);

        let wait_result = if let Some(token) = cancel_token {
            tokio::select! {
                biased;
                _ = token.cancelled() => {
                    self.cleanup_cancelled_connection(target_id).await;
                    return Err(RuntimeError::Other(anyhow::anyhow!(
                        "Connection creation cancelled while waiting"
                    )));
                }
                _ = tokio::time::sleep(timeout_duration) => {
                    Err(RuntimeError::DeadlineExceeded {
                        message: "WebRTC connection establishment timeout".to_string(),
                        timeout_ms: 10000,
                    })
                }
                result = ready_rx => {
                    result.map_err(|_| RuntimeError::Other(anyhow::anyhow!(
                        "Connection establishment failed (channel closed)"
                    )))
                }
            }
        } else {
            tokio::time::timeout(timeout_duration, ready_rx)
                .await
                .map_err(|_| RuntimeError::DeadlineExceeded {
                    message: "WebRTC connection establishment timeout".to_string(),
                    timeout_ms: 10000,
                })?
                .map_err(|_| {
                    RuntimeError::Other(anyhow::anyhow!(
                        "Connection establishment failed (channel closed)"
                    ))
                })
        };

        wait_result?;

        tracing::info!(
            "✅ [Factory] WebRTC connection ready: {:?}",
            target_id.to_string_repr()
        );

        // Final cancellation check
        if let Some(token) = cancel_token {
            if token.is_cancelled() {
                self.cleanup_cancelled_connection(target_id).await;
                return Err(RuntimeError::Other(anyhow::anyhow!(
                    "Connection creation cancelled after ready"
                )));
            }
        }

        // Get and return WebRtcConnection
        let peers = self.peers.read().await;
        peers
            .get(target_id)
            .map(|state| state.webrtc_conn.clone())
            .ok_or_else(|| {
                RuntimeError::Other(anyhow::anyhow!(
                    "Peer not found after connection establishment"
                ))
            })
    }

    /// Send media sample to target Actor via WebRTC Track
    ///
    /// # Arguments
    /// - `target`: Target Actor ID
    /// - `track_id`: Media track identifier
    /// - `sample`: Media sample to send
    ///
    /// # Returns
    /// Ok(()) if sent successfully
    pub async fn send_media_sample(
        &self,
        target: &actr_protocol::ActrId,
        track_id: &str,
        sample: actr_framework::MediaSample,
    ) -> RuntimeResult<()> {
        use webrtc::rtp::header::Header as RtpHeader;
        use webrtc::rtp::packet::Packet as RtpPacket;

        // 1. Get PeerState for target
        let peers = self.peers.read().await;
        let peer_state = peers.get(target).ok_or_else(|| {
            RuntimeError::Other(anyhow::anyhow!(
                "No connection to target: {}",
                target.to_string_repr()
            ))
        })?;

        // 2. Get Track from WebRtcConnection
        let track = peer_state
            .webrtc_conn
            .get_media_track(track_id)
            .await
            .ok_or_else(|| RuntimeError::Other(anyhow::anyhow!("Track not found: {track_id}")))?;

        // 3. Get next sequence number for this track
        let sequence_number = peer_state
            .webrtc_conn
            .next_sequence_number(track_id)
            .await
            .ok_or_else(|| {
                RuntimeError::Other(anyhow::anyhow!(
                    "Sequence number not found for track: {track_id}"
                ))
            })?;

        // 4. Get SSRC for this track
        let ssrc = peer_state
            .webrtc_conn
            .get_ssrc(track_id)
            .await
            .ok_or_else(|| {
                RuntimeError::Other(anyhow::anyhow!("SSRC not found for track: {track_id}"))
            })?;

        // 5. Construct RTP packet from MediaSample
        let rtp_packet = RtpPacket {
            header: RtpHeader {
                version: 2,
                padding: false,
                extension: false,
                marker: true,     // Mark each sample (simplified)
                payload_type: 96, // Dynamic payload type (simplified - TODO: codec-specific)
                sequence_number,  // Per-track sequence number (wraps at 65535)
                timestamp: sample.timestamp,
                ssrc, // Unique SSRC per track (randomly generated)
                ..Default::default()
            },
            payload: sample.data,
        };

        // 6. Send RTP packet via track
        track
            .write_rtp(&rtp_packet)
            .await
            .map_err(|e| RuntimeError::Other(anyhow::anyhow!("Failed to write RTP: {e}")))?;

        tracing::debug!(
            "📤 Sent MediaSample: track_id={}, seq={}, ssrc=0x{:08x}, timestamp={}, size={}",
            track_id,
            sequence_number,
            ssrc,
            sample.timestamp,
            rtp_packet.payload.len()
        );

        Ok(())
    }

    /// Add dynamic media track and trigger SDP renegotiation
    ///
    /// # Arguments
    /// - `target`: Target Actor ID
    /// - `track_id`: Media track identifier
    /// - `codec`: Codec name (e.g., "VP8", "H264", "OPUS")
    /// - `media_type`: Media type ("video" or "audio")
    ///
    /// # Returns
    /// Ok(()) if track added and renegotiation completed successfully
    ///
    /// # Note
    /// This triggers SDP renegotiation on the existing PeerConnection.
    /// The connection remains active and existing tracks continue transmitting.
    pub async fn add_dynamic_track(
        &self,
        target: &actr_protocol::ActrId,
        track_id: String,
        codec: &str,
        media_type: &str,
    ) -> RuntimeResult<()> {
        tracing::info!(
            "🎬 Adding dynamic track: track_id={}, codec={}, type={}, target={}",
            track_id,
            codec,
            media_type,
            target.to_string_repr()
        );

        // 1. Get existing peer state and extract needed parts
        let (webrtc_conn, peer_connection) = {
            let peers = self.peers.read().await;
            let state = peers.get(target).ok_or_else(|| {
                RuntimeError::Other(anyhow::anyhow!(
                    "No connection to target: {}",
                    target.to_string_repr()
                ))
            })?;
            (state.webrtc_conn.clone(), state.peer_connection.clone())
        };

        // 2. Add track to existing PeerConnection
        webrtc_conn
            .add_media_track(track_id.clone(), codec, media_type)
            .await?;

        tracing::info!("✅ Added track to PeerConnection: {}", track_id);

        // 3. Trigger SDP renegotiation
        let root_span = tracing::info_span!("add_track", target_id = %target.to_string_repr());
        #[cfg(feature = "opentelemetry")]
        self.root_context_map
            .write()
            .await
            .insert(target.clone(), root_span.context());

        self.renegotiate_connection(target, &peer_connection)
            .instrument(root_span)
            .await?;

        tracing::info!("✅ Dynamic track added successfully: {}", track_id);

        Ok(())
    }

    /// Renegotiate SDP with existing peer
    ///
    /// Creates new Offer with updated track list and exchanges SDP.
    /// ICE connection remains active (no restart).
    async fn renegotiate_connection(
        &self,
        target: &actr_protocol::ActrId,
        peer_connection: &Arc<RTCPeerConnection>,
    ) -> RuntimeResult<()> {
        tracing::info!(
            "🔄 Starting SDP renegotiation with {}",
            target.to_string_repr()
        );

        // 1. Create new Offer (includes all tracks: old + new)
        let offer = peer_connection.create_offer(None).await.map_err(|e| {
            RuntimeError::Other(anyhow::anyhow!("Failed to create renegotiation offer: {e}"))
        })?;
        let offer_sdp = offer.sdp.clone();

        // 2. Set local description
        peer_connection
            .set_local_description(offer)
            .await
            .map_err(|e| {
                RuntimeError::Other(anyhow::anyhow!("Failed to set local description: {e}"))
            })?;

        tracing::debug!(
            "📝 Created renegotiation Offer (SDP length: {})",
            offer_sdp.len()
        );

        // 3. Send Offer via signaling server
        let session_desc = actr_protocol::SessionDescription {
            r#type: SdpType::Offer as i32,
            sdp: offer_sdp,
        };
        let payload = actr_relay::Payload::SessionDescription(session_desc);
        self.send_actr_relay(target, payload).await?;

        tracing::info!("✅ Sent renegotiation Offer to {}", target.to_string_repr());

        // 4. Answer will be handled by existing handle_answer() method
        // Note: We don't wait for Answer here to avoid blocking.
        // The renegotiation completes asynchronously when Answer arrives.

        Ok(())
    }

    /// Initiate ICE restart on an existing connection (offerer side).
    /// Uses atomic state management within peers lock for complete de-duplication.
    /// If ICE restart fails after all retries, attempts to establish a new connection.
    pub async fn restart_ice(
        self: &Arc<Self>,
        target: &actr_protocol::ActrId,
    ) -> RuntimeResult<()> {
        // All ICE restart state management is now unified within the peers lock
        // This provides atomic de-duplication and eliminates race conditions
        let (peer_connection, should_start_restart) = {
            let mut peers = self.peers.write().await;
            if let Some(state) = peers.get_mut(target) {
                // 1. Check if restart is already in-flight using restart_task_handle
                if let Some(ref handle) = state.restart_task_handle {
                    if !handle.is_finished() {
                        tracing::debug!(
                            "🚫 ICE restart already in-flight for serial={}, skipping (task not finished)",
                            target.serial_number
                        );
                        return Ok(());
                    }
                }

                // 2. Also check ice_restart_inflight flag as a backup
                if state.ice_restart_inflight {
                    tracing::debug!(
                        "🚫 ICE restart already in-flight for serial={}, skipping (ice_restart_inflight=true)",
                        target.serial_number
                    );
                    return Ok(());
                }

                // 3. Check if we are the offerer
                if !state.is_offerer {
                    tracing::warn!(
                        "🚫 Skip ICE restart to serial={}: we are not the offerer",
                        target.serial_number
                    );
                    return Ok(());
                }

                // 4. Set flag to prevent concurrent restarts
                state.ice_restart_inflight = true;
                // Note: ice_restart_attempts will be managed by do_ice_restart_inner
                // Note: restart_task_handle will be set after spawning the task

                (state.peer_connection.clone(), true)
            } else {
                tracing::warn!(
                    "🚫 Skip ICE restart to serial={}: peer not found",
                    target.serial_number
                );
                return Ok(());
            }
        };

        if !should_start_restart {
            return Ok(());
        }

        tracing::info!(
            "♻️ Initiating ICE restart to serial={}",
            target.serial_number
        );

        // Spawn restart task
        let target_clone = target.clone();
        let peers = Arc::clone(&self.peers);
        let negotiator = self.negotiator.clone();
        let local_id = self.local_id.clone();
        let credential_state = self.credential_state.clone();
        let signaling_client = Arc::clone(&self.signaling_client);
        let coordinator_weak = Arc::downgrade(self);

        let handle = tokio::spawn(async move {
            let restart_result = Self::do_ice_restart_inner(
                &target_clone,
                &peers,
                peer_connection,
                &negotiator,
                &local_id,
                credential_state,
                &signaling_client,
            )
            .await;

            match restart_result {
                Ok(true) => {
                    tracing::info!(
                        "✅ ICE restart succeeded for serial={}",
                        target_clone.serial_number
                    );
                }
                Ok(false) => {
                    // ICE restart failed after all retries, clean up and try to establish new connection
                    tracing::warn!(
                        "⚠️ ICE restart exhausted for serial={}, cleaning up and attempting fresh connection",
                        target_clone.serial_number
                    );

                    if let Some(coord) = coordinator_weak.upgrade() {
                        // First, clean up the old connection resources
                        tracing::info!(
                            "🧹 Cleaning up old connection after ICE restart failure for serial={}",
                            target_clone.serial_number
                        );
                        coord.cleanup_cancelled_connection(&target_clone).await;
                    }
                }
                Err(e) => {
                    tracing::error!(
                        "❌ ICE restart failed for serial={}: {}",
                        target_clone.serial_number,
                        e
                    );

                    // Clean up resources on error
                    if let Some(coord) = coordinator_weak.upgrade() {
                        tracing::info!(
                            "🧹 Cleaning up connection after ICE restart error for serial={}",
                            target_clone.serial_number
                        );
                        coord.cleanup_cancelled_connection(&target_clone).await;
                    }
                }
            }

            // Cleanup restart_task_handle registration - now in peers lock
            {
                let mut peers_guard = peers.write().await;
                if let Some(state) = peers_guard.get_mut(&target_clone) {
                    state.restart_task_handle = None;
                }
            }
        });

        // Store the restart handle immediately in the same peers lock
        // This completes the atomic state transition
        {
            let mut peers = self.peers.write().await;
            if let Some(state) = peers.get_mut(target) {
                state.restart_task_handle = Some(handle);
            }
        }

        Ok(())
    }

    /// Internal ICE restart implementation with retries
    /// Returns Ok(true) if restart succeeded, Ok(false) if all retries exhausted
    async fn do_ice_restart_inner(
        target: &ActrId,
        peers: &Arc<RwLock<HashMap<ActrId, PeerState>>>,
        peer_connection: Arc<RTCPeerConnection>,
        negotiator: &WebRtcNegotiator,
        local_id: &ActrId,
        credential_state: CredentialState,
        signaling_client: &Arc<dyn SignalingClient>,
    ) -> RuntimeResult<bool> {
        // Use enhanced backoff with total duration limit
        let backoff = ExponentialBackoff::with_total_duration(
            Duration::from_millis(ICE_RESTART_INITIAL_BACKOFF_MS),
            Duration::from_millis(ICE_RESTART_MAX_BACKOFF_MS),
            Some(ICE_RESTART_MAX_RETRIES),
            ICE_RESTART_MAX_TOTAL_DURATION,
        );

        let mut restart_ok = false;
        let mut gathering_started_at: Option<Instant> = None;

        for delay in backoff {
            // ========== Guard 1: Check signaling state ==========
            if !signaling_client.is_connected() {
                tracing::debug!(
                    "🔄 Signaling not ready for ICE restart to serial={}, will retry after {:?}",
                    target.serial_number,
                    delay
                );
                tokio::time::sleep(delay).await;
                continue; // Skip this iteration, don't create offer
            }

            // ========== Guard 2: Check ICE gathering state (with timeout detection) ==========
            let gathering_state = peer_connection.ice_gathering_state();
            if gathering_state == RTCIceGatheringState::Gathering {
                let started = gathering_started_at.get_or_insert_with(Instant::now);
                let gathering_duration = started.elapsed();

                if gathering_duration > ICE_GATHERING_TIMEOUT {
                    tracing::error!(
                        "❌ ICE gathering stuck for {:?}, aborting ICE restart for serial={}",
                        gathering_duration,
                        target.serial_number
                    );
                    // Close peer connection to stop gathering
                    let _ = peer_connection.close().await;
                    return Ok(false);
                }

                tracing::debug!(
                    "⏳ ICE gathering in progress ({:?} elapsed), will retry after {:?}",
                    gathering_duration,
                    delay
                );
                tokio::time::sleep(delay).await;
                continue; // Skip this iteration, wait for gathering to complete
            } else {
                // Not gathering, reset timer
                gathering_started_at = None;
            }

            // ========== Both guards passed, safe to create offer ==========
            let (offer_sdp, attempt) = {
                let mut peers_guard = peers.write().await;
                let state = match peers_guard.get_mut(target) {
                    Some(s) => s,
                    None => {
                        tracing::warn!(
                            "🚫 Peer state not found during ICE restart for serial={}",
                            target.serial_number
                        );
                        return Ok(false);
                    }
                };

                if !state.is_offerer {
                    tracing::warn!(
                        "🚫 Skip ICE restart to serial={}: we are not the offerer",
                        target.serial_number
                    );
                    state.ice_restart_inflight = false;
                    state.ice_restart_attempts = 0;
                    return Ok(false);
                }

                // ice_restart_inflight is already set to true in restart_ice()
                // Just increment the attempt counter here
                state.ice_restart_attempts += 1;
                let attempt = state.ice_restart_attempts;

                let offer_sdp = negotiator
                    .create_ice_restart_offer(&peer_connection)
                    .await?;

                (offer_sdp, attempt)
            };

            // Send ICE restart offer
            let relay = ActrRelay {
                source: local_id.clone(),
                credential: credential_state.credential().await,
                target: target.clone(),
                payload: Some(actr_relay::Payload::SessionDescription(
                    actr_protocol::SessionDescription {
                        r#type: SdpType::IceRestartOffer as i32,
                        sdp: offer_sdp,
                    },
                )),
            };

            let envelope = SignalingEnvelope {
                envelope_version: 1,
                envelope_id: uuid::Uuid::new_v4().to_string(),
                reply_for: None,
                timestamp: prost_types::Timestamp {
                    seconds: chrono::Utc::now().timestamp(),
                    nanos: 0,
                },
                flow: Some(signaling_envelope::Flow::ActrRelay(relay)),
                traceparent: None,
                tracestate: None,
            };

            if let Err(e) = signaling_client.send_envelope(envelope).await {
                tracing::error!(
                    "❌ Failed to send ICE restart offer to serial={}: {}",
                    target.serial_number,
                    e
                );
                // Mark inflight as false and continue to next retry
                let mut peers_guard = peers.write().await;
                if let Some(state) = peers_guard.get_mut(target) {
                    state.ice_restart_inflight = false;
                }
                tokio::time::sleep(delay).await;
                continue;
            }

            tracing::info!(
                "♻️ ICE restart attempt {} sent to serial={}",
                attempt,
                target.serial_number
            );

            // Wait for restart completion
            let success =
                Self::wait_for_restart_completion_static(peers, target, ICE_RESTART_TIMEOUT).await;

            if success {
                restart_ok = true;
                break;
            }

            tracing::warn!(
                "⚠️ ICE restart attempt {} timed out for serial={}",
                attempt,
                target.serial_number
            );

            // Mark current attempt ended
            {
                let mut peers_guard = peers.write().await;
                if let Some(state) = peers_guard.get_mut(target) {
                    state.ice_restart_inflight = false;
                }
            }

            // Exponential backoff before retrying
            tracing::info!(
                "⏳ Waiting {:?} before next ICE restart attempt to serial={}",
                delay,
                target.serial_number
            );
            tokio::time::sleep(delay).await;
        }

        if !restart_ok {
            tracing::warn!(
                "⚠️ Backoff iterator exhausted for serial={}, stopping retries and dropping peer",
                target.serial_number
            );
            Self::drop_peer_connection_static(peers, target).await;
            return Ok(false);
        }

        Ok(true)
    }

    /// Static version of wait_for_restart_completion for use in spawned task
    /// Uses read lock for checking status to avoid blocking other peers
    async fn wait_for_restart_completion_static(
        peers: &Arc<RwLock<HashMap<ActrId, PeerState>>>,
        target: &ActrId,
        timeout: Duration,
    ) -> bool {
        let mut interval = tokio::time::interval(Duration::from_millis(100));
        let timeout_sleep = tokio::time::sleep(timeout);
        tokio::pin!(timeout_sleep);

        loop {
            tokio::select! {
                _ = &mut timeout_sleep => {
                    return false;
                }
                _ = interval.tick() => {
                    // Use read lock to check status (allows concurrent access)
                    let is_done = {
                        let peers_guard = peers.read().await;
                        match peers_guard.get(target) {
                            Some(state) => !state.ice_restart_inflight,
                            None => return false,
                        }
                    };

                    if is_done {
                        // Only acquire write lock when actually need to reset counter
                        let mut peers_guard = peers.write().await;
                        if let Some(state) = peers_guard.get_mut(target) {
                            state.ice_restart_attempts = 0;
                        }
                        return true;
                    }
                }
            }
        }
    }

    /// Static version of drop_peer_connection for use in spawned task
    ///
    /// NOTE: Releases the write lock BEFORE calling close() to avoid blocking
    /// other operations on `peers` during potentially slow close operations.
    async fn drop_peer_connection_static(
        peers: &Arc<RwLock<HashMap<ActrId, PeerState>>>,
        target: &ActrId,
    ) {
        // Remove peer from map first, then close outside the lock
        let state_to_close = {
            let mut peers_guard = peers.write().await;
            peers_guard.remove(target)
        }; // Lock released here

        if let Some(state) = state_to_close {
            if let Err(e) = state.peer_connection.close().await {
                tracing::warn!(
                    "⚠️ Failed to close peer_connection for {}: {}",
                    target.serial_number,
                    e
                );
            }
            if let Err(e) = state.webrtc_conn.close().await {
                tracing::warn!(
                    "⚠️ Failed to close WebRtcConnection for {}: {}",
                    target.serial_number,
                    e
                );
            }
            tracing::info!("🧹 Dropped peer connection for {}", target.serial_number);
        } else {
            tracing::warn!(
                "⚠️ drop_peer_connection: peer not found {}",
                target.serial_number
            );
        }
    }

    /// Handle renegotiation Offer (existing connection)
    ///
    /// Called when receiving an Offer on an already-established connection.
    /// This happens when the remote peer adds/removes tracks dynamically.
    #[allow(dead_code)]
    async fn handle_renegotiation_offer(
        &self,
        from: &ActrId,
        offer_sdp: String,
    ) -> RuntimeResult<()> {
        tracing::info!(
            "🔄 Processing renegotiation Offer from {}",
            from.to_string_repr()
        );

        // 1. Get existing peer connection
        let peer_connection = {
            let peers = self.peers.read().await;
            let state = peers.get(from).ok_or_else(|| {
                RuntimeError::Other(anyhow::anyhow!("Peer state not found for renegotiation"))
            })?;
            state.peer_connection.clone()
        };

        // 2. Set remote description (new Offer)
        let offer =
            webrtc::peer_connection::sdp::session_description::RTCSessionDescription::offer(
                offer_sdp,
            )
            .map_err(|e| {
                RuntimeError::Other(anyhow::anyhow!("Failed to parse renegotiation offer: {e}"))
            })?;
        peer_connection
            .set_remote_description(offer)
            .await
            .map_err(|e| {
                RuntimeError::Other(anyhow::anyhow!("Failed to set remote description: {e}"))
            })?;

        tracing::debug!("✅ Set remote description (renegotiation Offer)");

        // 3. Create Answer
        let answer = peer_connection.create_answer(None).await.map_err(|e| {
            RuntimeError::Other(anyhow::anyhow!(
                "Failed to create renegotiation answer: {e}"
            ))
        })?;
        let answer_sdp = answer.sdp.clone();

        // 4. Set local description
        peer_connection
            .set_local_description(answer)
            .await
            .map_err(|e| {
                RuntimeError::Other(anyhow::anyhow!("Failed to set local description: {e}"))
            })?;

        tracing::debug!(
            "✅ Created renegotiation Answer (SDP length: {})",
            answer_sdp.len()
        );

        // 5. Send Answer via signaling server
        let session_desc = actr_protocol::SessionDescription {
            r#type: SdpType::Answer as i32,
            sdp: answer_sdp,
        };
        let payload = actr_relay::Payload::SessionDescription(session_desc);
        self.send_actr_relay(from, payload).await?;

        tracing::info!("✅ Sent renegotiation Answer to {}", from.to_string_repr());

        // Note: on_track callback will automatically trigger for new remote tracks
        // No need to manually handle track additions here

        Ok(())
    }

    /// Handle ICE restart Offer on an existing connection.
    /// Only the answerer should accept restart; offerer-side restarts are initiated locally.
    async fn handle_ice_restart_offer(
        self: &Arc<Self>,
        from: &ActrId,
        offer_sdp: String,
    ) -> RuntimeResult<()> {
        // Locate peer state and ensure we are not the offerer
        let (peer_connection, is_offerer) = {
            let peers = self.peers.read().await;
            let state = peers.get(from).ok_or_else(|| {
                RuntimeError::Other(anyhow::anyhow!(
                    "ICE restart offer received for unknown peer"
                ))
            })?;
            (state.peer_connection.clone(), state.is_offerer)
        };

        if is_offerer {
            tracing::warn!(
                "🚫 Ignoring ICE restart offer from {:?}: we are current offerer",
                from.serial_number
            );
            return Ok(());
        }

        // Apply remote restart offer and generate answer
        let answer_sdp = self
            .negotiator
            .create_answer(&peer_connection, offer_sdp)
            .await?;

        // Send restart answer back
        let session_desc = actr_protocol::SessionDescription {
            r#type: SdpType::Answer as i32,
            sdp: answer_sdp,
        };
        let payload = actr_relay::Payload::SessionDescription(session_desc);
        self.send_actr_relay(from, payload).await?;

        // Flush any buffered ICE candidates collected before remote description was set
        self.flush_pending_candidates(from, &peer_connection)
            .await?;

        tracing::info!(
            "✅ Completed ICE restart answer to serial={}",
            from.serial_number
        );

        Ok(())
    }

    /// Remove peer connection and clear associated cached state.

    /// Handle role assignment result
    #[cfg_attr(
        feature = "opentelemetry",
        tracing::instrument(skip_all, fields(peer_id = ?peer.to_string_repr()))
    )]
    async fn handle_role_assignment(self: &Arc<Self>, assign: RoleAssignment, peer: ActrId) {
        tracing::debug!(?assign, ?peer, "handle_role_assignment");

        // ========== Check for role change to offerer and clean up if needed ==========
        // Only clean up when becoming offerer (we need to initiate a new connection)
        // If becoming answerer, we just wait for the peer's offer
        if assign.is_offerer {
            let has_connection = self.peers.read().await.contains_key(&peer);

            // Clean up if we have an existing connection (reconnection scenario)
            if has_connection {
                tracing::info!(
                    "🔄 Assigned as offerer for {} (has_connection={}), cleaning up for new connection",
                    peer.serial_number,
                    has_connection
                );

                // Spawn cleanup in background to avoid blocking signaling loop
                // cleanup_cancelled_connection may call close() which sends events,
                // and those events may trigger transport cleanup that could block
                let this = Arc::clone(self);
                let peer_clone = peer.clone();
                tokio::spawn(async move {
                    this.cleanup_cancelled_connection(&peer_clone).await;
                });
            }
        }
        // ========== End role change check ==========

        // 先尝试唤醒等待的协商
        let role_sender = {
            let mut neg = self.peer_negotiation.lock().await;
            neg.get_mut(&peer).and_then(|s| s.role_tx.take())
        };
        if let Some(sender) = role_sender {
            if sender.send(assign.is_offerer).is_ok() {
                return;
            }
        }

        tracing::debug!(
            ?assign,
            ?peer,
            "handle_role_assignment: no pending negotiation"
        );
        // 如果目前还没有连接，根据角色立即行动，避免依赖 send_message 才触发
        let has_connection = self.peers.read().await.contains_key(&peer);
        if has_connection {
            return;
        }
        if assign.is_offerer {
            tracing::info!(
                "🎭 Acting as offerer to {} per assignment (no pending negotiation)",
                peer.serial_number
            );
            // Spawn the offer connection in background to avoid blocking signaling loop
            let this = Arc::clone(self);
            let peer_clone = peer.clone();
            #[cfg(feature = "opentelemetry")]
            let current_span = tracing::Span::current();
            tokio::spawn(async move {
                let start_offer_fut = this.start_offer_connection(&peer_clone, true);
                #[cfg(feature = "opentelemetry")]
                let start_offer_fut = start_offer_fut.instrument(current_span);
                match start_offer_fut.await {
                    Ok(ready_rx) => {
                        this.peer_negotiation
                            .lock()
                            .await
                            .entry(peer_clone.clone())
                            .or_default()
                            .ready_rx = Some(ready_rx);
                    }
                    Err(e) => {
                        tracing::warn!(
                            "⚠️ Failed to start proactive offer connection to {}: {}",
                            peer_clone.serial_number,
                            e
                        );
                    }
                }
            });
        } else {
            tracing::debug!(
                "🎭 Assignment marks us as answerer for {}, waiting for offer (no pending negotiation)",
                peer.serial_number
            );
            let (tx, _rx) = oneshot::channel();
            self.peer_negotiation
                .lock()
                .await
                .entry(peer.clone())
                .or_default()
                .ready_tx = Some(tx);

            // 防止长时间等不到 offer：超时后主动重新协商/建链
            let weak = Arc::downgrade(self);
            let peer_clone = peer.clone();
            #[cfg(feature = "opentelemetry")]
            let current_span = tracing::Span::current();
            tokio::spawn(async move {
                tokio::time::sleep(ROLE_WAIT_TIMEOUT).await;
                if let Some(coord) = weak.upgrade() {
                    // 如果已经有连接或 ready 被消费则退出
                    if coord.peers.read().await.contains_key(&peer_clone) {
                        return;
                    }
                    let pending = {
                        let mut neg = coord.peer_negotiation.lock().await;
                        neg.get_mut(&peer_clone).and_then(|s| s.ready_tx.take())
                    };
                    if pending.is_none() {
                        return;
                    }
                    tracing::warn!(
                        "⏳ Waiting for offer from {} timed out, force acting as offerer",
                        peer_clone.serial_number
                    );
                    let start_offer_fut = coord.start_offer_connection(&peer_clone, true);
                    #[cfg(feature = "opentelemetry")]
                    let start_offer_fut = start_offer_fut.instrument(current_span);
                    match start_offer_fut.await {
                        Ok(ready_rx) => {
                            coord
                                .peer_negotiation
                                .lock()
                                .await
                                .entry(peer_clone.clone())
                                .or_default()
                                .ready_rx = Some(ready_rx);
                        }
                        Err(e) => {
                            tracing::warn!(
                                "⚠️ Failed to start offer connection after timeout to {}: {}",
                                peer_clone.serial_number,
                                e
                            );
                        }
                    }
                }
            });
        }
    }

    /// Initiate role negotiation and await assignment
    #[cfg_attr(
        feature = "opentelemetry",
        tracing::instrument(skip_all, fields(target_id = ?target.to_string_repr()))
    )]
    async fn negotiate_role(&self, target: &ActrId) -> RuntimeResult<bool> {
        let (tx, rx) = oneshot::channel();
        // 按目标 ActorId 记录等待的角色分配
        self.peer_negotiation
            .lock()
            .await
            .entry(target.clone())
            .or_default()
            .role_tx = Some(tx);

        let payload = actr_relay::Payload::RoleNegotiation(RoleNegotiation {
            from: self.local_id.clone(),
            to: target.clone(),
            realm_id: self.local_id.realm.realm_id,
        });

        tracing::debug!(
            "🔄 Sending role negotiation to serial={}",
            target.serial_number
        );
        self.send_actr_relay(target, payload).await?;

        rx.await.map_err(|_| {
            RuntimeError::Other(anyhow::anyhow!(
                "Role negotiation channel closed before assignment"
            ))
        })
    }

    /// Install a state change handler to auto-trigger ICE restart on disconnection (offerer only).
    fn install_restart_handler(
        self: &Arc<Self>,
        webrtc_conn: WebRtcConnection,
        peer_connection: Arc<RTCPeerConnection>,
        target: ActrId,
    ) {
        let coord = Arc::downgrade(self);
        peer_connection.on_peer_connection_state_change(Box::new(
            move |state: RTCPeerConnectionState| {
                let coord = coord.clone();
                let target = target.clone();
                let webrtc_conn = webrtc_conn.clone();
                Box::pin(async move {
                    // First run the base WebRtcConnection cleanup.
                    webrtc_conn.handle_state_change(state).await;

                    tracing::info!(
                        "📡 PeerConnection state for {} -> {:?}",
                        target.serial_number,
                        state
                    );

                    // Update state tracking for health check
                    if let Some(c) = coord.upgrade() {
                        let mut peers = c.peers.write().await;
                        if let Some(peer_state) = peers.get_mut(&target) {
                            peer_state.current_state = state;
                            peer_state.last_state_change = std::time::Instant::now();
                        }
                        drop(peers); // Release lock before potentially long-running operations
                    }

                    if matches!(
                        state,
                        RTCPeerConnectionState::Disconnected | RTCPeerConnectionState::Failed
                    ) {
                        if let Some(c) = coord.upgrade() {
                            if let Err(e) = c.restart_ice(&target).await {
                                tracing::warn!(
                                    "⚠️ Failed to auto restart ICE to {}: {}",
                                    target.serial_number,
                                    e
                                );
                            }
                        }
                    }
                })
            },
        ));
    }
}