openrtc 1.0.1

//! Transport-upgrade config and native WebRTC signaling hooks for `Client`.
//!
//! Phase 3 keeps this path explicitly feature-gated so default builds stay
//! unchanged while native hosts can opt into WebRTC datachannel upgrades.

use super::*;
#[cfg(all(not(target_arch = "wasm32"), feature = "transport-moq"))]
use crate::native_moq_policy::{
    is_data_ready as native_moq_is_data_ready,
    is_repairable_state as native_moq_is_repairable_state, setup_status_report,
    should_close_before_restart as native_moq_should_close_before_restart,
    should_preserve_active_moq_on_iroh_send,
    should_reuse_existing_setup as native_moq_should_reuse_existing_setup,
    successful_send_status_report,
};
#[cfg(not(target_arch = "wasm32"))]
use crate::native_send_policy::{peer_send_order, NativePeerSendOrder};
#[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
use crate::native_webrtc_policy::{
    inbound_signal_can_bootstrap_responder, native_signal_send_failure_is_terminal_for_attempt,
    native_webrtc_session_is_active, should_ignore_stale_inbound_signal_recovery,
    should_immediately_retire_active_webrtc_for_reason, NativeWebRTCAttemptReservation,
    NativeWebRTCBrowserSignalTrigger, NativeWebRTCNativeTrigger, NativeWebRTCRecoveryOptions,
    NativeWebRTCRecoveryTrigger,
};

#[cfg(not(target_arch = "wasm32"))]
type NativeMainSendFuture<'a> =
    std::pin::Pin<Box<dyn std::future::Future<Output = Result<(), String>> + Send + 'a>>;

#[cfg(not(target_arch = "wasm32"))]
fn native_control_stream_rank(stream_id: iroh::endpoint::StreamId) -> u64 {
    let initiator_rank = match stream_id.initiator() {
        iroh::endpoint::Side::Client => 0,
        iroh::endpoint::Side::Server => 1,
    };
    stream_id.index().saturating_mul(2) + initiator_rank
}

#[cfg(not(target_arch = "wasm32"))]
fn should_replace_native_control_stream(
    existing: NativeControlStreamOwner,
    incoming: NativeControlStreamOwner,
) -> bool {
    existing.transport_stable_id != incoming.transport_stable_id
        || incoming.stream_rank < existing.stream_rank
}

#[cfg(not(target_arch = "wasm32"))]
pub(crate) async fn retire_losing_native_control_candidate(
    send: iroh::endpoint::SendStream,
    recv: iroh::endpoint::RecvStream,
) -> std::io::Result<()> {
    let delivery = crate::application_crypto_streams::PeerSendStream::plain(send)
        .finish_and_wait_for_peer(std::time::Duration::from_secs(2))
        .await;
    drop(recv);
    delivery
}

#[cfg(not(target_arch = "wasm32"))]
pub(crate) async fn retire_replaced_native_control_send(
    send: Arc<tokio::sync::Mutex<iroh::endpoint::SendStream>>,
) -> std::io::Result<()> {
    let mut send = send.lock().await;
    send.finish()
        .map_err(|error| std::io::Error::other(format!("finish replaced stream: {error}")))?;
    match tokio::time::timeout(std::time::Duration::from_secs(2), send.stopped()).await {
        Ok(Ok(_)) => Ok(()),
        Ok(Err(error)) => Err(std::io::Error::other(format!(
            "wait for peer acknowledgement: {error}"
        ))),
        Err(_) => Err(std::io::Error::new(
            std::io::ErrorKind::TimedOut,
            "timed out waiting for peer acknowledgement",
        )),
    }
}

#[cfg(not(target_arch = "wasm32"))]
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum NativeBleSwitchAdmission {
    Ready,
    Wait,
    Reject,
}

#[cfg(not(target_arch = "wasm32"))]
fn classify_native_ble_switch_admission(
    block: Option<&(bool, String)>,
) -> NativeBleSwitchAdmission {
    match block {
        None => NativeBleSwitchAdmission::Ready,
        Some((true, _)) => NativeBleSwitchAdmission::Reject,
        Some((false, _)) => NativeBleSwitchAdmission::Wait,
    }
}

#[cfg(not(target_arch = "wasm32"))]
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum NativeOptionalRouteKind {
    WebRtc,
    Moq,
}

#[cfg(not(target_arch = "wasm32"))]
impl crate::client::NativeOptionalRouteGenerations {
    fn get(
        &self,
        kind: NativeOptionalRouteKind,
    ) -> Option<crate::client::NativeOptionalRouteGeneration> {
        match kind {
            NativeOptionalRouteKind::WebRtc => self.webrtc,
            NativeOptionalRouteKind::Moq => self.moq,
        }
    }

    fn set(
        &mut self,
        kind: NativeOptionalRouteKind,
        generation: Option<crate::client::NativeOptionalRouteGeneration>,
    ) {
        match kind {
            NativeOptionalRouteKind::WebRtc => self.webrtc = generation,
            NativeOptionalRouteKind::Moq => self.moq = generation,
        }
    }

    fn is_empty(&self) -> bool {
        self.webrtc.is_none() && self.moq.is_none()
    }
}

#[cfg(not(target_arch = "wasm32"))]
fn native_optional_route_generation_matches(
    record: Option<&crate::connection_manager::ConnectionRecord>,
    expected: Option<crate::client::NativeOptionalRouteGeneration>,
    route_instance_id: usize,
) -> bool {
    let (Some(record), Some(expected)) = (record, expected) else {
        return false;
    };
    expected.route_instance_id == route_instance_id
        && record.transport_stable_id == Some(expected.transport_stable_id)
        && record.transport_generation == expected.transport_generation
        && record.route_generation == expected.route_generation
        && matches!(
            record.state,
            crate::connection_manager::ConnectionState::Connecting
                | crate::connection_manager::ConnectionState::Connected
        )
}

impl Client {
    #[cfg(not(target_arch = "wasm32"))]
    pub(crate) async fn refresh_native_optional_route_generation(&self, connection_id: &str) {
        let Some(record) = self
            .connection_manager
            .get_by_connection_id(connection_id)
            .await
        else {
            self.native_optional_route_generations
                .write()
                .await
                .remove(connection_id);
            return;
        };

        let current_webrtc_instance = {
            #[cfg(feature = "transport-webrtc")]
            {
                self.native_webrtc_sessions
                    .read()
                    .await
                    .get(connection_id)
                    .map(|session| Arc::as_ptr(session) as usize)
            }
            #[cfg(not(feature = "transport-webrtc"))]
            {
                None
            }
        };
        let current_moq_instance = {
            #[cfg(feature = "transport-moq")]
            {
                self.native_moq_sessions
                    .read()
                    .await
                    .get(connection_id)
                    .map(|session| Arc::as_ptr(session) as usize)
            }
            #[cfg(not(feature = "transport-moq"))]
            {
                None
            }
        };

        let mut installed = self.native_optional_route_generations.write().await;
        let Some(generations) = installed.get_mut(connection_id) else {
            return;
        };
        for (kind, current_instance) in [
            (NativeOptionalRouteKind::WebRtc, current_webrtc_instance),
            (NativeOptionalRouteKind::Moq, current_moq_instance),
        ] {
            let Some(mut generation) = generations.get(kind) else {
                continue;
            };
            let still_installed = current_instance == Some(generation.route_instance_id)
                && record.transport_stable_id == Some(generation.transport_stable_id)
                && record.transport_generation == generation.transport_generation;
            if still_installed {
                generation.route_generation = record.route_generation;
                generations.set(kind, Some(generation));
            } else {
                generations.set(kind, None);
            }
        }
        if generations.is_empty() {
            installed.remove(connection_id);
        }
    }

    #[cfg(not(target_arch = "wasm32"))]
    async fn install_native_optional_route_generation(
        &self,
        connection_id: &str,
        kind: NativeOptionalRouteKind,
        route_instance_id: usize,
    ) -> Option<crate::client::NativeOptionalRouteGeneration> {
        let record = self
            .connection_manager
            .get_by_connection_id(connection_id)
            .await?;
        if !matches!(
            record.state,
            crate::connection_manager::ConnectionState::Connecting
                | crate::connection_manager::ConnectionState::Connected
        ) {
            return None;
        }
        let transport_stable_id = record.transport_stable_id?;
        let generation = crate::client::NativeOptionalRouteGeneration {
            transport_stable_id,
            transport_generation: record.transport_generation,
            route_generation: record.route_generation,
            route_instance_id,
        };
        self.native_optional_route_generations
            .write()
            .await
            .entry(connection_id.to_string())
            .or_default()
            .set(kind, Some(generation));
        Some(generation)
    }

    #[cfg(not(target_arch = "wasm32"))]
    async fn clear_native_optional_route_generation_if_current(
        &self,
        connection_id: &str,
        kind: NativeOptionalRouteKind,
        route_instance_id: usize,
    ) {
        let mut installed = self.native_optional_route_generations.write().await;
        let Some(generations) = installed.get_mut(connection_id) else {
            return;
        };
        if generations
            .get(kind)
            .is_some_and(|generation| generation.route_instance_id == route_instance_id)
        {
            generations.set(kind, None);
        }
        if generations.is_empty() {
            installed.remove(connection_id);
        }
    }

    #[cfg(not(target_arch = "wasm32"))]
    async fn native_optional_route_generation_for_instance(
        &self,
        connection_id: &str,
        kind: NativeOptionalRouteKind,
        route_instance_id: usize,
    ) -> Option<crate::client::NativeOptionalRouteGeneration> {
        self.native_optional_route_generations
            .read()
            .await
            .get(connection_id)
            .and_then(|generations| generations.get(kind))
            .filter(|generation| generation.route_instance_id == route_instance_id)
    }

    #[cfg(not(target_arch = "wasm32"))]
    async fn native_optional_route_generation_is_current(
        &self,
        connection_id: &str,
        kind: NativeOptionalRouteKind,
        route_instance_id: usize,
    ) -> bool {
        let record = self
            .connection_manager
            .get_by_connection_id(connection_id)
            .await;
        let expected = self
            .native_optional_route_generation_for_instance(connection_id, kind, route_instance_id)
            .await;
        native_optional_route_generation_matches(record.as_ref(), expected, route_instance_id)
    }

    #[cfg(all(not(target_arch = "wasm32"), feature = "transport-moq"))]
    pub(crate) const NATIVE_MOQ_SEND_READY_RETRY_WINDOW_MS: u64 = 6_000;
    #[cfg(all(not(target_arch = "wasm32"), feature = "transport-moq"))]
    pub(crate) const NATIVE_MOQ_SEND_READY_RETRY_DELAY_MS: u64 = 250;
    #[cfg(all(not(target_arch = "wasm32"), feature = "transport-moq"))]
    pub(crate) const NATIVE_MOQ_SEND_REPAIR_RETRY_MIN_INTERVAL_MS: u64 = 1_000;
    #[cfg(all(not(target_arch = "wasm32"), feature = "transport-moq"))]
    pub(crate) const NATIVE_MOQ_SEND_MAX_REPAIR_REQUESTS: u32 = 3;
    #[cfg(all(not(target_arch = "wasm32"), feature = "transport-moq"))]
    pub(crate) const NATIVE_MOQ_DATA_READY_TIMEOUT_MS: u64 = 5_000;
    #[cfg(all(not(target_arch = "wasm32"), feature = "transport-moq"))]
    pub(crate) const NATIVE_MOQ_DATA_READY_STABLE_POLL_MS: u64 = 125;
    #[cfg(all(not(target_arch = "wasm32"), feature = "transport-moq"))]
    pub(crate) const NATIVE_MOQ_DATA_READY_STABLE_SAMPLES: u32 = 8;

    #[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
    pub(crate) const NATIVE_WEBRTC_ADMISSION_FAILURE_BACKOFF_MS: i64 =
        crate::runtime_policy::NATIVE_WEBRTC_ADMISSION_FAILURE_BACKOFF_MS;
    #[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
    pub(crate) const NATIVE_WEBRTC_CONNECT_TIMEOUT_BACKOFF_MS: i64 =
        crate::runtime_policy::NATIVE_WEBRTC_CONNECT_TIMEOUT_BACKOFF_MS;
    #[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
    pub(crate) const NATIVE_WEBRTC_CONNECT_TIMEOUT_RETRY_DELAY_MS: u64 =
        crate::runtime_policy::NATIVE_WEBRTC_CONNECT_TIMEOUT_RETRY_DELAY_MS;
    #[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
    pub(crate) const NATIVE_WEBRTC_DIRECT_QUIC_SUPPRESSION_MS: i64 =
        crate::runtime_policy::NATIVE_WEBRTC_DIRECT_QUIC_SUPPRESSION_MS;
    #[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
    pub(crate) const NATIVE_WEBRTC_MAX_UPGRADE_ATTEMPTS: u32 =
        crate::runtime_policy::NATIVE_WEBRTC_MAX_UPGRADE_ATTEMPTS;
    #[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
    pub(crate) const NATIVE_WEBRTC_SIGNAL_SEND_TIMEOUT_MS: u64 =
        crate::runtime_policy::NATIVE_WEBRTC_SIGNAL_SEND_TIMEOUT_MS;
    #[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
    pub(crate) const NATIVE_WEBRTC_SIGNAL_SEND_RETRY_DELAY_MS: u64 =
        crate::runtime_policy::NATIVE_WEBRTC_SIGNAL_SEND_RETRY_DELAY_MS;
    #[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
    pub(crate) const NATIVE_WEBRTC_CONNECT_TIMEOUT_MS: u64 =
        crate::runtime_policy::NATIVE_WEBRTC_CONNECT_TIMEOUT_MS;
    /// If an attempt has been in Connecting state for this long without ever
    /// receiving a remote SDP (offer or answer), treat it as hung and abort
    /// early so the retry loop can open a fresh attempt instead of burning
    /// the full 45s connect timeout waiting on signals that will never land.
    #[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
    pub(crate) const NATIVE_WEBRTC_NO_SIGNAL_ABORT_MS: u64 =
        crate::runtime_policy::NATIVE_WEBRTC_NO_SIGNAL_ABORT_MS;
    #[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
    fn native_webrtc_heartbeat_config() -> crate::heartbeat::HeartbeatConfig {
        crate::heartbeat::HeartbeatConfig {
            tick_interval: std::time::Duration::from_secs(
                crate::runtime_policy::native_webrtc_heartbeat_tick_secs(),
            ),
            suspect_after: std::time::Duration::from_secs(
                crate::runtime_policy::native_webrtc_heartbeat_suspect_after_secs(),
            ),
            stale_after: std::time::Duration::from_secs(
                crate::runtime_policy::native_webrtc_heartbeat_stale_after_secs(),
            ),
            send_timeout: std::time::Duration::from_secs(
                crate::runtime_policy::native_webrtc_heartbeat_send_timeout_secs(),
            ),
        }
    }

    #[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
    async fn current_native_webrtc_route_identity_matches(
        &self,
        connection_id: &str,
        source: &Arc<crate::transport::NativeWebRTCDataChannel>,
    ) -> bool {
        let route_instance_id = Arc::as_ptr(source) as usize;
        let is_current_session = self
            .native_webrtc_sessions
            .read()
            .await
            .get(connection_id)
            .is_some_and(|current| Arc::ptr_eq(current, source));
        is_current_session
            && self
                .native_optional_route_generation_is_current(
                    connection_id,
                    NativeOptionalRouteKind::WebRtc,
                    route_instance_id,
                )
                .await
    }

    #[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
    async fn current_native_webrtc_route_matches(
        &self,
        connection_id: &str,
        source: &Arc<crate::transport::NativeWebRTCDataChannel>,
    ) -> bool {
        native_webrtc_session_is_active(source.state())
            && self
                .current_native_webrtc_route_identity_matches(connection_id, source)
                .await
    }

    #[cfg(all(not(target_arch = "wasm32"), feature = "transport-moq"))]
    async fn current_native_moq_route_identity_matches(
        &self,
        connection_id: &str,
        source: &Arc<crate::transport::NativeMoQSession>,
    ) -> bool {
        let route_instance_id = Arc::as_ptr(source) as usize;
        let is_current_session = self
            .native_moq_sessions
            .read()
            .await
            .get(connection_id)
            .is_some_and(|current| Arc::ptr_eq(current, source));
        is_current_session
            && self
                .native_optional_route_generation_is_current(
                    connection_id,
                    NativeOptionalRouteKind::Moq,
                    route_instance_id,
                )
                .await
    }

    #[cfg(all(not(target_arch = "wasm32"), feature = "transport-moq"))]
    async fn current_native_moq_route_matches(
        &self,
        connection_id: &str,
        source: &Arc<crate::transport::NativeMoQSession>,
    ) -> bool {
        matches!(
            source.state(),
            crate::transport::NativeMoQState::Connecting
                | crate::transport::NativeMoQState::Connected
        ) && self
            .current_native_moq_route_identity_matches(connection_id, source)
            .await
    }

    #[cfg(all(
        not(target_arch = "wasm32"),
        any(feature = "transport-webrtc", feature = "transport-moq")
    ))]
    pub(crate) async fn handle_native_transport_peer_data(
        &self,
        connection_id: &str,
        remote_node_id: Option<&str>,
        transport: &str,
        generation: crate::client::NativePeerDataGeneration,
        payload: bytes::Bytes,
    ) -> anyhow::Result<()> {
        let payload = if transport == "moq" {
            self.open_inbound_direct_moq_payload(connection_id, &payload)?
        } else {
            self.open_inbound_application_payload(connection_id, &payload)?
        };
        self.emit_native_peer_data(crate::client::NativePeerDataEvent {
            connection_id: connection_id.to_string(),
            remote_node_id: remote_node_id
                .map(str::trim)
                .filter(|value| !value.is_empty())
                .map(ToOwned::to_owned),
            transport: transport.to_string(),
            transport_stable_id: generation.transport_stable_id,
            transport_generation: generation.transport_generation,
            route_generation: generation.route_generation,
            payload,
        });
        Ok(())
    }

    #[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
    pub(crate) async fn handle_current_native_webrtc_peer_data(
        &self,
        connection_id: &str,
        remote_node_id: Option<&str>,
        source: &Arc<crate::transport::NativeWebRTCDataChannel>,
        payload: bytes::Bytes,
    ) -> anyhow::Result<bool> {
        let is_current = self
            .current_native_webrtc_route_matches(connection_id, source)
            .await;
        if !is_current || source.state() != crate::transport::NativeWebRTCState::Connected {
            println!(
                "[NativeWebRTC] stale peer-data ignored connection_id={} remote_node_id={:?} negotiation_id={}",
                connection_id,
                remote_node_id,
                source.negotiation_id(),
            );
            return Ok(false);
        }

        let route_instance_id = Arc::as_ptr(source) as usize;
        let Some(expected) = self
            .native_optional_route_generation_for_instance(
                connection_id,
                NativeOptionalRouteKind::WebRtc,
                route_instance_id,
            )
            .await
        else {
            return Ok(false);
        };
        let Some(generation) = self
            .current_native_peer_data_generation(connection_id, Some(expected.transport_stable_id))
            .await
        else {
            return Ok(false);
        };
        if generation.transport_generation != expected.transport_generation
            || generation.route_generation != expected.route_generation
            || !self
                .current_native_webrtc_route_matches(connection_id, source)
                .await
        {
            return Ok(false);
        }

        self.handle_native_transport_peer_data(
            connection_id,
            remote_node_id,
            "webrtc",
            generation,
            payload,
        )
        .await?;
        Ok(true)
    }

    #[cfg(all(not(target_arch = "wasm32"), feature = "transport-moq"))]
    pub(crate) async fn handle_current_native_moq_peer_data(
        &self,
        connection_id: &str,
        remote_node_id: Option<&str>,
        source: &Arc<crate::transport::NativeMoQSession>,
        payload: bytes::Bytes,
    ) -> anyhow::Result<bool> {
        let is_current = self
            .current_native_moq_route_matches(connection_id, source)
            .await;
        if !is_current || source.state() != crate::transport::NativeMoQState::Connected {
            println!(
                "[NativeMoQ] stale peer-data ignored connection_id={} remote_node_id={:?}",
                connection_id, remote_node_id,
            );
            return Ok(false);
        }

        let route_instance_id = Arc::as_ptr(source) as usize;
        let Some(expected) = self
            .native_optional_route_generation_for_instance(
                connection_id,
                NativeOptionalRouteKind::Moq,
                route_instance_id,
            )
            .await
        else {
            return Ok(false);
        };
        let Some(generation) = self
            .current_native_peer_data_generation(connection_id, Some(expected.transport_stable_id))
            .await
        else {
            return Ok(false);
        };
        if generation.transport_generation != expected.transport_generation
            || generation.route_generation != expected.route_generation
            || !self
                .current_native_moq_route_matches(connection_id, source)
                .await
        {
            return Ok(false);
        }

        self.handle_native_transport_peer_data(
            connection_id,
            remote_node_id,
            "moq",
            generation,
            payload,
        )
        .await?;
        Ok(true)
    }

    #[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
    fn create_native_webrtc_negotiation_id(connection_id: &str) -> String {
        format!(
            "native-{}-{}",
            connection_id.replace(':', "-"),
            now_millis_i64()
        )
    }

    #[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
    async fn native_webrtc_session_for_connection(
        &self,
        connection_id: &str,
    ) -> Option<Arc<crate::transport::NativeWebRTCDataChannel>> {
        self.native_webrtc_sessions
            .read()
            .await
            .get(connection_id)
            .cloned()
    }

    #[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
    pub(crate) async fn suppress_native_webrtc_restarts(
        &self,
        connection_id: &str,
        duration_ms: i64,
        reason: impl Into<String>,
    ) {
        if connection_id.trim().is_empty() || duration_ms <= 0 {
            return;
        }

        let reason = reason.into();
        let until_ms = now_millis_i64().saturating_add(duration_ms);
        self.native_webrtc_suppressions.write().await.insert(
            connection_id.to_string(),
            crate::client::NativeWebRTCSuppression {
                until_ms,
                reason: reason.clone(),
            },
        );

        println!(
            "[NativeWebRTC] suppressing restart connection_id={} duration_ms={} until_ms={} reason={}",
            connection_id,
            duration_ms,
            until_ms,
            reason,
        );
    }

    #[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
    async fn active_native_webrtc_suppression(
        &self,
        connection_id: &str,
    ) -> Option<crate::client::NativeWebRTCSuppression> {
        let now = now_millis_i64();
        let mut suppressions = self.native_webrtc_suppressions.write().await;
        let suppression = suppressions.get(connection_id).cloned();
        match suppression {
            Some(value) if value.until_ms > now => Some(value),
            Some(_) => {
                suppressions.remove(connection_id);
                None
            }
            None => None,
        }
    }

    /// Safety expiry for an in-flight attempt record: upgrade connect timeout
    /// plus a small grace window. If an attempt hasn't concluded (by clearing
    /// itself) within this window, we assume it is abandoned and let a new
    /// trigger proceed.
    #[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
    const NATIVE_WEBRTC_ATTEMPT_IN_FLIGHT_GRACE_MS: i64 =
        crate::runtime_policy::NATIVE_WEBRTC_ATTEMPT_IN_FLIGHT_GRACE_MS;

    /// Returns the current in-flight attempt record for a connection, evicting
    /// it if its safety expiry has passed.
    #[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
    async fn active_native_webrtc_attempt(
        &self,
        connection_id: &str,
    ) -> Option<crate::client::NativeWebRTCAttemptInFlight> {
        let now = now_millis_i64();
        let mut attempts = self.native_webrtc_attempts_in_flight.write().await;
        match attempts.get(connection_id).cloned() {
            Some(value) if value.expires_at_ms > now => Some(value),
            Some(_) => {
                attempts.remove(connection_id);
                None
            }
            None => None,
        }
    }

    /// Records that a new upgrade attempt is starting. Replaces any prior
    /// record for the same connection (the caller is the sole authorized start
    /// path after the in-flight gate has approved it).
    #[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
    async fn record_native_webrtc_attempt_started(
        &self,
        connection_id: &str,
        negotiation_id: &str,
        preserve_on_direct_path: bool,
    ) {
        let now = now_millis_i64();
        let expires_at_ms = now
            .saturating_add(Self::NATIVE_WEBRTC_CONNECT_TIMEOUT_MS as i64)
            .saturating_add(Self::NATIVE_WEBRTC_ATTEMPT_IN_FLIGHT_GRACE_MS);
        self.native_webrtc_attempts_in_flight.write().await.insert(
            connection_id.to_string(),
            crate::client::NativeWebRTCAttemptInFlight {
                negotiation_id: negotiation_id.to_string(),
                started_at_ms: now,
                expires_at_ms,
                preserve_on_direct_path,
                last_skip_log_ms: 0,
            },
        );
    }

    /// Throttle for the "skipped: attempt in flight" log line. First skip is
    /// always logged; subsequent skips for the same attempt log at most once
    /// per this interval.
    #[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
    const NATIVE_WEBRTC_SKIP_LOG_INTERVAL_MS: i64 =
        crate::runtime_policy::NATIVE_WEBRTC_SKIP_LOG_INTERVAL_MS;

    /// Returns true if the caller should log a skipped-attempt-in-flight
    /// line for the given connection right now, updating the last-log
    /// timestamp atomically. Throttled to one log per attempt every
    /// NATIVE_WEBRTC_SKIP_LOG_INTERVAL_MS.
    #[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
    async fn should_log_native_webrtc_skip(&self, connection_id: &str) -> bool {
        let now = now_millis_i64();
        let mut attempts = self.native_webrtc_attempts_in_flight.write().await;
        let Some(entry) = attempts.get_mut(connection_id) else {
            return true;
        };
        if entry.last_skip_log_ms == 0
            || now.saturating_sub(entry.last_skip_log_ms)
                >= Self::NATIVE_WEBRTC_SKIP_LOG_INTERVAL_MS
        {
            entry.last_skip_log_ms = now;
            return true;
        }
        false
    }

    /// Clears the in-flight record for a connection iff it matches the given
    /// negotiation_id. Used at terminal outcomes so a later, unrelated attempt
    /// cannot be accidentally cleared.
    #[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
    pub(crate) async fn clear_native_webrtc_attempt_if_current(
        &self,
        connection_id: &str,
        negotiation_id: &str,
    ) {
        let mut attempts = self.native_webrtc_attempts_in_flight.write().await;
        if let Some(existing) = attempts.get(connection_id) {
            if existing.negotiation_id == negotiation_id {
                attempts.remove(connection_id);
            }
        }
    }

    /// Unconditionally clears the in-flight record (used on explicit
    /// renegotiate or full connection teardown).
    #[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
    pub(crate) async fn clear_native_webrtc_attempt(&self, connection_id: &str) {
        self.native_webrtc_attempts_in_flight
            .write()
            .await
            .remove(connection_id);
    }

    #[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
    pub(crate) async fn clear_native_webrtc_suppression(
        &self,
        connection_id: &str,
        reason_prefix: Option<&str>,
    ) -> bool {
        let mut suppressions = self.native_webrtc_suppressions.write().await;
        match suppressions.get(connection_id) {
            Some(value)
                if reason_prefix.map(|prefix| value.reason.starts_with(prefix)) != Some(false) =>
            {
                suppressions.remove(connection_id);
                true
            }
            _ => false,
        }
    }

    #[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
    pub(crate) async fn reset_native_webrtc_attempt_budget(
        &self,
        connection_id: &str,
        reason: &str,
    ) {
        let removed = self
            .native_webrtc_attempt_counts
            .write()
            .await
            .remove(connection_id);
        let cleared_suppression = self
            .clear_native_webrtc_suppression(connection_id, Some("upgrade-attempts-exhausted"))
            .await;
        if removed.is_some() || cleared_suppression {
            println!(
                "[NativeWebRTC] reset attempt budget connection_id={} reason={} had_counter={} cleared_exhausted_suppression={}",
                connection_id,
                reason,
                removed.is_some(),
                cleared_suppression,
            );
        }
    }

    #[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
    async fn close_native_webrtc_session_for_replacement(
        &self,
        connection_id: &str,
        reason: &str,
    ) -> bool {
        let session = self
            .native_webrtc_session_for_connection(connection_id)
            .await;
        let Some(session) = session else {
            self.evict_native_signal_stream(connection_id, reason).await;
            return false;
        };

        session.close();
        let removed = self
            .remove_native_webrtc_session_if_current(connection_id, &session, reason)
            .await;
        // This helper is used only when the logical peer or its physical Iroh
        // leg is being replaced. Unlike ordinary WebRTC repair, replacement
        // also retires the native-main route owned by that peer generation.
        self.evict_native_signal_stream(connection_id, reason).await;
        if removed {
            self.finalize_deferred_managed_retirement_if_needed(connection_id)
                .await;
        }
        removed
    }

    #[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
    async fn retire_replaced_peer_connection(
        &self,
        record: crate::connection_manager::ConnectionRecord,
        reason: &str,
    ) {
        self.forget_session_connection(&record.connection_id);
        self.cancel_native_webrtc_retry(&record.connection_id).await;
        self.reset_native_webrtc_attempt_budget(&record.connection_id, reason)
            .await;
        let _ = self
            .clear_native_webrtc_suppression(&record.connection_id, None)
            .await;
        let _ = self
            .close_native_webrtc_session_for_replacement(&record.connection_id, reason)
            .await;

        if let Some(endpoint_str) = record.endpoint_id.as_deref().or(record.node_id.as_deref()) {
            if let Ok(endpoint_id) = endpoint_str.parse::<iroh::EndpointId>() {
                if let Some(connection) = self.get_connection(endpoint_id).await {
                    connection.close(0u8.into(), reason.as_bytes());
                }
            }
        }

        self.retire_managed_connection_now(&record.connection_id, Some(reason.to_string()))
            .await;
    }

    #[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
    pub(crate) async fn accept_replacement_peer(
        &self,
        connection_id: &str,
        authoritative_device_id: Option<&str>,
        reason: &str,
    ) {
        // Phase 7 correlation: emit a single structured line at the
        // entry point so log scans can reconstruct the lifecycle without
        // having to grep for the (many) unstructured lines below.
        let phase7_ctx = self.correlation_for_connection(connection_id).await;
        crate::clog!(
            "[NativeWebRTC][replacement-peer]",
            &phase7_ctx,
            "begin reason={} authoritative_device_id={}",
            reason,
            authoritative_device_id.unwrap_or("<none>")
        );
        let current_record = self
            .connection_manager
            .get_by_connection_id(connection_id)
            .await;

        if let Some(device_id) = authoritative_device_id
            .map(str::trim)
            .filter(|value| !value.is_empty())
        {
            let _ = self
                .connection_manager
                .set_device_id(connection_id, device_id.to_string())
                .await;
        }

        let mut stale_records =
            std::collections::HashMap::<String, crate::connection_manager::ConnectionRecord>::new();

        let authoritative_device_id = authoritative_device_id
            .map(str::trim)
            .filter(|value| !value.is_empty())
            .map(ToOwned::to_owned);

        let mut candidate_device_ids = std::collections::HashSet::<String>::new();
        if let Some(device_id) = authoritative_device_id.as_deref() {
            candidate_device_ids.insert(device_id.to_string());
        }
        if let Some(record) = current_record.as_ref() {
            if let Some(device_id) = record.device_id.as_deref() {
                candidate_device_ids.insert(device_id.to_string());
            }
            if let Some(device_id_hint) = record.device_id_hint.as_deref() {
                candidate_device_ids.insert(device_id_hint.to_string());
            }
        }

        for device_id in candidate_device_ids {
            for record in self.connection_manager.get_by_device_id(&device_id).await {
                if record.connection_id != connection_id {
                    stale_records.insert(record.connection_id.clone(), record);
                }
            }
        }

        if stale_records.is_empty() {
            let current_hint = current_record.as_ref().and_then(|record| {
                authoritative_device_id
                    .clone()
                    .or_else(|| record.device_id.clone())
                    .or_else(|| record.device_id_hint.clone())
            });

            if let Some(current_hint) = current_hint {
                for record in self.connection_manager.list_active().await {
                    if record.connection_id == connection_id {
                        continue;
                    }
                    let matches_device_hint = record.device_id.as_deref()
                        == Some(current_hint.as_str())
                        || record.device_id_hint.as_deref() == Some(current_hint.as_str());
                    if matches_device_hint {
                        stale_records.insert(record.connection_id.clone(), record);
                    }
                }
            }
        }

        for (_, record) in stale_records {
            println!(
                "[NativeWebRTC] replacement peer admitted: retiring stale connection_id={} replaced_by={} reason={}",
                record.connection_id,
                connection_id,
                reason,
            );
            self.retire_replaced_peer_connection(record, reason).await;
        }

        self.cancel_native_webrtc_retry(connection_id).await;
        self.reset_native_webrtc_attempt_budget(connection_id, reason)
            .await;
        let _ = self
            .clear_native_webrtc_suppression(connection_id, None)
            .await;
    }

    #[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
    pub(crate) async fn request_native_webrtc_recovery(
        &self,
        connection_id: &str,
        remote_node_id: Option<&str>,
        trigger: NativeWebRTCRecoveryTrigger,
        options: NativeWebRTCRecoveryOptions<'_>,
    ) -> anyhow::Result<()> {
        if connection_id.trim().is_empty() || !self.is_webrtc_transport_enabled().await {
            println!(
                "[NativeWebRTC][lifecycle] request skipped connection_id={} trigger={} reason={}",
                connection_id,
                trigger.as_str(),
                if connection_id.trim().is_empty() {
                    "empty-connection-id"
                } else {
                    "transport-disabled"
                },
            );
            return Ok(());
        }

        let preferred_negotiation_id = options
            .preferred_negotiation_id
            .map(str::trim)
            .filter(|value| !value.is_empty());

        if self
            .connection_manager
            .get_by_connection_id(connection_id)
            .await
            .is_none()
        {
            println!(
                "[NativeWebRTC][lifecycle] request skipped connection_id={} trigger={} reason=missing-connection-record remote_node_id={:?} preferred_negotiation_id={}",
                connection_id,
                trigger.as_str(),
                remote_node_id,
                preferred_negotiation_id.unwrap_or("none"),
            );
            return Ok(());
        }

        if self.session_registry_active() {
            match self.session_admission(connection_id) {
                crate::session_token::SessionAdmission::Pending => {
                    println!(
                        "[NativeWebRTC][lifecycle] request skipped connection_id={} trigger={} reason=admission-pending preferred_negotiation_id={}",
                        connection_id,
                        trigger.as_str(),
                        preferred_negotiation_id.unwrap_or("none"),
                    );
                    return Ok(());
                }
                crate::session_token::SessionAdmission::Rejected { reason } => {
                    println!(
                        "[NativeWebRTC][lifecycle] request skipped connection_id={} trigger={} reason=admission-rejected admission_reason={} preferred_negotiation_id={}",
                        connection_id,
                        trigger.as_str(),
                        reason,
                        preferred_negotiation_id.unwrap_or("none"),
                    );
                    return Ok(());
                }
                crate::session_token::SessionAdmission::Accepted { .. } => {}
            }
        }

        if crate::native_webrtc_policy::native_webrtc_suppression_blocks_recovery(
            preferred_negotiation_id,
        ) {
            if let Some(suppression) = self.active_native_webrtc_suppression(connection_id).await {
                println!(
                    "[NativeWebRTC][lifecycle] request skipped connection_id={} trigger={} reason=restart-suppressed suppression_reason={} remaining_ms={} preferred_negotiation_id={} force_restart={}",
                    connection_id,
                    trigger.as_str(),
                    suppression.reason,
                    suppression.until_ms.saturating_sub(now_millis_i64()),
                    preferred_negotiation_id.unwrap_or("none"),
                    options.force_restart,
                );
                return Ok(());
            }
        } else if let Some(suppression) = self.active_native_webrtc_suppression(connection_id).await
        {
            println!(
                "[NativeWebRTC][lifecycle] request continuing through suppression connection_id={} trigger={} suppression_reason={} remaining_ms={} preferred_negotiation_id={} force_restart={}",
                connection_id,
                trigger.as_str(),
                suppression.reason,
                suppression.until_ms.saturating_sub(now_millis_i64()),
                preferred_negotiation_id.unwrap_or("none"),
                options.force_restart,
            );
        }

        if let Some(existing) = self
            .native_webrtc_session_for_connection(connection_id)
            .await
        {
            let active_negotiation_id = existing.negotiation_id();
            let has_active_attempt = self
                .active_native_webrtc_attempt(connection_id)
                .await
                .is_some();
            if should_ignore_stale_inbound_signal_recovery(
                trigger,
                existing.state(),
                active_negotiation_id.as_str(),
                preferred_negotiation_id,
                has_active_attempt,
            ) {
                println!(
                    "[NativeWebRTC] stale inbound signal ignored connection_id={} trigger={} active_negotiation_id={} signal_negotiation_id={} has_active_attempt={}",
                    connection_id,
                    trigger.as_str(),
                    active_negotiation_id,
                    preferred_negotiation_id.unwrap_or("none"),
                    has_active_attempt,
                );
                return Ok(());
            }
        }

        // Serialize upgrade attempts. Every trigger (inbound-sdp, admission,
        // health-check, relay-path, foreground, handshake, scheduled-retry)
        // funnels through here. Without this gate they race to tear each
        // other's sessions down, producing the "attempt=N/8" churn observed
        // after browser restart. Only `InboundRenegotiate` (explicit fresh
        // session) preempts; everything else either matches the in-flight
        // negotiation_id (no-op) or is dropped.
        if let Some(in_flight) = self.active_native_webrtc_attempt(connection_id).await {
            let matches_in_flight = preferred_negotiation_id
                .map(|value| value == in_flight.negotiation_id)
                .unwrap_or(false);
            let in_flight_age_ms = now_millis_i64().saturating_sub(in_flight.started_at_ms);
            let is_preempting_trigger =
                trigger.can_preempt_in_flight_attempt(preferred_negotiation_id, matches_in_flight);
            if options.force_restart {
                let stale_in_flight =
                    in_flight_age_ms >= Self::NATIVE_WEBRTC_NO_SIGNAL_ABORT_MS as i64;
                if is_preempting_trigger || stale_in_flight {
                    println!(
                        "[NativeWebRTC][lifecycle] force-restart preempting in-flight attempt connection_id={} trigger={} coordinator={:?} in_flight_negotiation_id={} preferred_negotiation_id={} matches_in_flight={} stale_in_flight={} age_ms={}",
                        connection_id,
                        trigger.as_str(),
                        trigger.coordinator(),
                        in_flight.negotiation_id,
                        preferred_negotiation_id.unwrap_or("none"),
                        matches_in_flight,
                        stale_in_flight,
                        in_flight_age_ms,
                    );
                    self.clear_native_webrtc_attempt(connection_id).await;
                } else {
                    println!(
                        "[NativeWebRTC][lifecycle] force-restart deferred to active attempt connection_id={} trigger={} coordinator={:?} in_flight_negotiation_id={} preferred_negotiation_id={} age_ms={}",
                        connection_id,
                        trigger.as_str(),
                        trigger.coordinator(),
                        in_flight.negotiation_id,
                        preferred_negotiation_id.unwrap_or("none"),
                        in_flight_age_ms,
                    );
                    return Ok(());
                }
            } else {
                // If we receive an inbound SDP for a *different* negotiation id while an
                // attempt is in-flight, we must allow it to preempt. This happens after
                // browser refreshes: the browser generates a fresh negotiation id and
                // sends a new offer while the native side still considers its previous
                // attempt "in flight". If we don't preempt here, we drop the inbound SDP
                // and then abort with "no remote SDP".
                if !is_preempting_trigger && !matches_in_flight {
                    if self.should_log_native_webrtc_skip(connection_id).await {
                        println!(
                        "[NativeWebRTC] upgrade skipped: attempt in flight connection_id={} trigger={} coordinator={:?} force_restart={} in_flight_negotiation_id={} preferred_negotiation_id={} age_ms={}",
                        connection_id,
                        trigger.as_str(),
                        trigger.coordinator(),
                        options.force_restart,
                        in_flight.negotiation_id,
                        preferred_negotiation_id.unwrap_or("none"),
                            in_flight_age_ms,
                    );
                    }
                    return Ok(());
                }
                if is_preempting_trigger {
                    println!(
                    "[NativeWebRTC][lifecycle] preempting in-flight attempt connection_id={} trigger={} coordinator={:?} in_flight_negotiation_id={} preferred_negotiation_id={} age_ms={}",
                    connection_id,
                    trigger.as_str(),
                    trigger.coordinator(),
                    in_flight.negotiation_id,
                    preferred_negotiation_id.unwrap_or("none"),
                        in_flight_age_ms,
                );
                    self.clear_native_webrtc_attempt(connection_id).await;
                }
            }
        }

        if trigger.skips_when_session_active() && !options.force_restart {
            if let Some(existing) = self
                .native_webrtc_session_for_connection(connection_id)
                .await
            {
                let negotiation_mismatch = preferred_negotiation_id
                    .map(|value| value != existing.negotiation_id().as_str())
                    .unwrap_or(false);
                if !negotiation_mismatch
                    && matches!(
                        existing.state(),
                        crate::transport::NativeWebRTCState::Connecting
                            | crate::transport::NativeWebRTCState::Connected
                    )
                {
                    println!(
                        "[NativeWebRTC][lifecycle] request skipped connection_id={} trigger={} reason=active-session state={:?} negotiation_id={} preferred_negotiation_id={} force_restart={}",
                        connection_id,
                        trigger.as_str(),
                        existing.state(),
                        existing.negotiation_id(),
                        preferred_negotiation_id.unwrap_or("none"),
                        options.force_restart,
                    );
                    return Ok(());
                }
            }
        }

        println!(
            "[NativeWebRTC] recovery request connection_id={} remote_node_id={:?} trigger={} force_restart={} preferred_negotiation_id={}",
            connection_id,
            remote_node_id,
            trigger.as_str(),
            options.force_restart,
            preferred_negotiation_id.unwrap_or("none"),
        );

        let role_override = options.role_override.or(match trigger {
            NativeWebRTCRecoveryTrigger::BrowserSignal(
                NativeWebRTCBrowserSignalTrigger::InboundSdp,
            ) => Some(crate::transport::NativeWebRTCRole::Responder),
            _ => None,
        });

        self.maybe_start_native_webrtc_upgrade(
            connection_id,
            remote_node_id,
            options.force_restart,
            preferred_negotiation_id,
            role_override,
        )
        .await
    }

    #[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
    pub(crate) fn reserve_native_webrtc_attempt_slot(
        current: u32,
    ) -> NativeWebRTCAttemptReservation {
        crate::native_webrtc_policy::reserve_attempt_slot(
            current,
            Self::NATIVE_WEBRTC_MAX_UPGRADE_ATTEMPTS,
        )
    }

    #[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
    async fn native_webrtc_upgrade_fast_path_skip_reason(
        &self,
        connection_id: &str,
        remote_node_id: Option<&str>,
        force_restart: bool,
        preferred_negotiation_id: Option<&str>,
    ) -> Option<String> {
        if self
            .connection_manager
            .get_by_connection_id(connection_id)
            .await
            .is_none()
        {
            return Some("missing connection record".to_string());
        }

        if self.session_registry_active() {
            match self.session_admission(connection_id) {
                crate::session_token::SessionAdmission::Pending => {
                    return Some("session admission pending".to_string());
                }
                crate::session_token::SessionAdmission::Rejected { reason } => {
                    return Some(format!("session admission rejected: {}", reason));
                }
                crate::session_token::SessionAdmission::Accepted { .. } => {}
            }
        }

        if crate::native_webrtc_policy::native_webrtc_suppression_blocks_recovery(
            preferred_negotiation_id,
        ) {
            if let Some(suppression) = self.active_native_webrtc_suppression(connection_id).await {
                return Some(format!(
                    "restart suppressed reason={} remaining_ms={}",
                    suppression.reason,
                    suppression.until_ms.saturating_sub(now_millis_i64())
                ));
            }
        } else if let Some(suppression) = self.active_native_webrtc_suppression(connection_id).await
        {
            println!(
                "[NativeWebRTC][lifecycle] fast-path continuing through suppression connection_id={} suppression_reason={} remaining_ms={} preferred_negotiation_id={}",
                connection_id,
                suppression.reason,
                suppression.until_ms.saturating_sub(now_millis_i64()),
                preferred_negotiation_id.unwrap_or("none"),
            );
        }

        if preferred_negotiation_id.is_none() {
            let attempt_exhausted = {
                let counts = self.native_webrtc_attempt_counts.read().await;
                counts.get(connection_id).copied().unwrap_or(0)
                    > Self::NATIVE_WEBRTC_MAX_UPGRADE_ATTEMPTS
            };
            if attempt_exhausted {
                return Some("attempt budget exhausted".to_string());
            }
        }

        if preferred_negotiation_id.is_none() {
            if let Some(remote_node_id) = remote_node_id
                .map(str::trim)
                .filter(|value| !value.is_empty())
            {
                let primary_iroh_path = self.iroh_path_kind(remote_node_id).await;
                if matches!(
                    primary_iroh_path,
                    crate::client::IrohPathKind::DirectQuic
                        | crate::client::IrohPathKind::DirectLan
                        | crate::client::IrohPathKind::Ble
                ) {
                    return Some(format!(
                        "optimal iroh path is primary path={}",
                        primary_iroh_path.transport_label()
                    ));
                }
            }
        }

        if let Some(existing) = self
            .native_webrtc_session_for_connection(connection_id)
            .await
        {
            let state = existing.state();
            let negotiation_mismatch = preferred_negotiation_id
                .map(str::trim)
                .filter(|value| !value.is_empty())
                .map(|value| value != existing.negotiation_id().as_str())
                .unwrap_or(false);
            let is_stale = state == crate::transport::NativeWebRTCState::Failed
                || state == crate::transport::NativeWebRTCState::Closed
                || (force_restart
                    && matches!(
                        state,
                        crate::transport::NativeWebRTCState::Connecting
                            | crate::transport::NativeWebRTCState::Connected
                    ))
                || negotiation_mismatch;
            if !is_stale {
                return Some(format!(
                    "existing session active state={:?} negotiation_id={}",
                    state,
                    existing.negotiation_id()
                ));
            }
        }

        None
    }

    #[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
    pub(crate) async fn acquire_native_webrtc_start_gate(
        &self,
        connection_id: &str,
        _remote_node_id: &str,
        _force_restart: bool,
        _preferred_negotiation_id: Option<&str>,
    ) -> Arc<tokio::sync::Notify> {
        loop {
            let wait_for = {
                let mut starts = self.native_webrtc_start_gates.lock().await;
                if let Some(existing) = starts.get(connection_id) {
                    Some(existing.clone())
                } else {
                    let gate = Arc::new(tokio::sync::Notify::new());
                    starts.insert(connection_id.to_string(), gate.clone());
                    return gate;
                }
            };
            wait_for
                .expect("existing start gate must be present")
                .notified()
                .await;
        }
    }

    #[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
    pub(crate) async fn release_native_webrtc_start_gate(
        &self,
        connection_id: &str,
        gate: Arc<tokio::sync::Notify>,
    ) {
        let removed = {
            let mut starts = self.native_webrtc_start_gates.lock().await;
            match starts.get(connection_id) {
                Some(current) if Arc::ptr_eq(current, &gate) => {
                    starts.remove(connection_id);
                    true
                }
                _ => false,
            }
        };

        if removed {
            gate.notify_waiters();
        }
    }

    #[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
    async fn cancel_native_webrtc_retry(&self, connection_id: &str) {
        self.native_webrtc_retry_deadlines
            .lock()
            .await
            .remove(connection_id);
    }

    #[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
    fn schedule_native_webrtc_retry(
        &self,
        connection_id: &str,
        remote_node_id: &str,
        retired_route: crate::client::NativeOptionalRouteGeneration,
        delay_ms: u64,
        reason_prefix: &'static str,
    ) {
        if connection_id.trim().is_empty() || remote_node_id.trim().is_empty() {
            return;
        }

        let connection_id = connection_id.to_string();
        let remote_node_id = remote_node_id.to_string();
        let client = self.clone();
        let due_at_ms = now_millis_i64().saturating_add(delay_ms.min(i64::MAX as u64) as i64);

        let should_schedule = {
            let deadlines = self.native_webrtc_retry_deadlines.clone();
            let connection_id = connection_id.clone();
            async move {
                let now = now_millis_i64();
                let mut deadlines = deadlines.lock().await;
                if deadlines
                    .get(&connection_id)
                    .copied()
                    .is_some_and(|existing| existing > now)
                {
                    return false;
                }
                deadlines.insert(connection_id, due_at_ms);
                true
            }
        };

        tokio::spawn(async move {
            if !should_schedule.await {
                return;
            }

            println!(
                "[NativeWebRTC] scheduling retry connection_id={} remote_node_id={} delay_ms={} reason={}",
                connection_id,
                remote_node_id,
                delay_ms,
                reason_prefix,
            );
            tokio::time::sleep(std::time::Duration::from_millis(delay_ms)).await;

            let should_run = {
                let mut deadlines = client.native_webrtc_retry_deadlines.lock().await;
                match deadlines.get(&connection_id).copied() {
                    Some(current_due_at_ms) if current_due_at_ms == due_at_ms => {
                        deadlines.remove(&connection_id);
                        true
                    }
                    _ => false,
                }
            };
            if !should_run {
                return;
            }

            let base_generation_is_current = client
                .connection_manager
                .get_by_connection_id(&connection_id)
                .await
                .is_some_and(|record| {
                    native_optional_route_generation_matches(
                        Some(&record),
                        Some(retired_route),
                        retired_route.route_instance_id,
                    )
                });
            if !base_generation_is_current
                || client
                    .native_webrtc_session_for_connection(&connection_id)
                    .await
                    .is_some_and(|session| native_webrtc_session_is_active(session.state()))
            {
                return;
            }
            let Ok(endpoint_id) = remote_node_id.parse::<iroh::EndpointId>() else {
                return;
            };
            if let Err(error) = client
                .ensure_connected_with_timeout(endpoint_id, std::time::Duration::from_secs(8))
                .await
            {
                println!(
                    "[NativeWebRTC] scheduled retry skipped: iroh redial unavailable connection_id={} remote_node_id={} reason={} error={}",
                    connection_id,
                    remote_node_id,
                    reason_prefix,
                    error,
                );
                return;
            }

            let base_generation_is_still_current = client
                .connection_manager
                .get_by_connection_id(&connection_id)
                .await
                .is_some_and(|record| {
                    native_optional_route_generation_matches(
                        Some(&record),
                        Some(retired_route),
                        retired_route.route_instance_id,
                    )
                });
            if !base_generation_is_still_current {
                return;
            }

            if matches!(
                client.session_admission(&connection_id),
                crate::session_token::SessionAdmission::Rejected { .. }
            ) {
                return;
            }

            let _ = client
                .clear_native_webrtc_suppression(&connection_id, Some(reason_prefix))
                .await;

            if let Err(error) = client
                .request_native_webrtc_recovery(
                    &connection_id,
                    Some(remote_node_id.as_str()),
                    NativeWebRTCRecoveryTrigger::Native(NativeWebRTCNativeTrigger::ScheduledRetry),
                    NativeWebRTCRecoveryOptions {
                        force_restart: true,
                        preferred_negotiation_id: None,
                        role_override: None,
                    },
                )
                .await
            {
                println!(
                    "[NativeWebRTC] scheduled retry failed connection_id={} remote_node_id={} error={}",
                    connection_id,
                    remote_node_id,
                    error,
                );
            }
        });
    }

    /// Write a framed native-main message over the persistent outbound bi-stream to
    /// `endpoint_id`, opening it on first use and caching the send half in
    /// `native_control_streams`. The stream is intentionally NOT finished — it
    /// stays alive for the connection's lifetime so the remote responder can
    /// keep its inbound-writer handle registered and reply on the same stream.
    ///
    /// On transient write failure the stale cached stream is evicted so the
    /// next call re-opens a fresh one.
    #[cfg(not(target_arch = "wasm32"))]
    pub(crate) fn send_native_main_frame<'a>(
        &'a self,
        connection_id: &'a str,
        endpoint_id: iroh::EndpointId,
        signal_type: &'a str,
        frame_buf: &'a [u8],
    ) -> NativeMainSendFuture<'a> {
        Box::pin(self.send_native_main_frame_inner(
            connection_id,
            endpoint_id,
            signal_type,
            frame_buf,
        ))
    }

    #[cfg(not(target_arch = "wasm32"))]
    async fn send_native_main_frame_inner(
        &self,
        connection_id: &str,
        endpoint_id: iroh::EndpointId,
        signal_type: &str,
        frame_buf: &[u8],
    ) -> Result<(), String> {
        use tokio::io::AsyncWriteExt;

        let current_transport_stable_id = self
            .get_connection(endpoint_id)
            .await
            .map(|connection| connection.stable_id() as u64)
            .ok_or_else(|| format!("missing Iroh connection for endpoint {}", endpoint_id))?;

        // Phase 1: use only a stream owned by the current physical generation.
        let cached = {
            let map = self.native_control_streams.lock().await;
            map.get(connection_id).cloned()
        };

        if let Some(entry) = cached {
            if entry.endpoint_id != endpoint_id
                || entry.owner.transport_stable_id != current_transport_stable_id
            {
                self.evict_native_signal_stream_if_owner(
                    connection_id,
                    entry.owner,
                    "cached-owner-stale",
                )
                .await;
            } else {
                let mut send_guard = entry.send.lock().await;
                match send_guard.write_all(frame_buf).await {
                    Ok(_) => {
                        if let Err(flush_err) = send_guard.flush().await {
                            println!(
                            "[SIGNAL-STREAM] flush-failed-on-cached connection_id={} signal_type={} error={}",
                            connection_id, signal_type, flush_err,
                        );
                            drop(send_guard);
                            self.evict_native_signal_stream_if_owner(
                                connection_id,
                                entry.owner,
                                "cached-flush-failed",
                            )
                            .await;
                        } else {
                            println!(
                            "[SIGNAL-STREAM] wrote-on-cached connection_id={} signal_type={} bytes={}",
                            connection_id,
                            signal_type,
                            frame_buf.len(),
                        );
                            return Ok(());
                        }
                    }
                    Err(err) => {
                        println!(
                        "[SIGNAL-STREAM] write-failed-on-cached connection_id={} signal_type={} error={}",
                        connection_id, signal_type, err,
                    );
                        drop(send_guard);
                        self.evict_native_signal_stream_if_owner(
                            connection_id,
                            entry.owner,
                            "cached-write-failed",
                        )
                        .await;
                    }
                }
            }
        }

        // Phase 2: open a fresh persistent stream.
        let (transport_stable_id, mut send, recv) = self
            .open_bi_internal_with_transport_stable_id(endpoint_id)
            .await
            .map_err(|e| format!("open_bi via iroh failed: {}", e))?;

        println!(
            "[SIGNAL-STREAM] opened-persistent connection_id={} endpoint_id={} signal_type={} stream_id={}",
            connection_id,
            endpoint_id,
            signal_type,
            send.id().index(),
        );

        let label = b"main";
        let label_len = (label.len() as u32).to_be_bytes();
        let mut opening_buf = Vec::with_capacity(1 + 4 + label.len() + frame_buf.len());
        opening_buf.push(0x00);
        opening_buf.extend_from_slice(&label_len);
        opening_buf.extend_from_slice(label);
        opening_buf.extend_from_slice(frame_buf);

        send.write_all(&opening_buf)
            .await
            .map_err(|e| format!("initial write failed: {}", e))?;
        send.flush()
            .await
            .map_err(|e| format!("initial flush failed: {}", e))?;

        println!(
            "[SIGNAL-STREAM] wrote-on-fresh connection_id={} signal_type={} bytes={}",
            connection_id,
            signal_type,
            opening_buf.len(),
        );

        if !self
            .install_native_control_stream(
                connection_id,
                endpoint_id,
                transport_stable_id,
                send,
                recv,
                "opened",
            )
            .await
        {
            return Err(format!(
                "fresh native-main stream lost ownership race connection_id={} transport_stable_id={}",
                connection_id, transport_stable_id
            ));
        }

        Ok(())
    }

    #[cfg(not(target_arch = "wasm32"))]
    pub(crate) async fn install_native_control_stream(
        &self,
        connection_id: &str,
        endpoint_id: iroh::EndpointId,
        transport_stable_id: u64,
        send: iroh::endpoint::SendStream,
        recv: iroh::endpoint::RecvStream,
        source: &str,
    ) -> bool {
        let stream_id = send.id();
        let owner = NativeControlStreamOwner {
            transport_stable_id,
            stream_rank: native_control_stream_rank(stream_id),
        };
        let current_transport_stable_id = self
            .get_connection(endpoint_id)
            .await
            .map(|connection| connection.stable_id() as u64);
        if current_transport_stable_id != Some(transport_stable_id) {
            println!(
                "[OpenRTC][control] stale stream ignored connection_id={} endpoint_id={} transport_stable_id={} current_transport_stable_id={:?} stream_id={} source={}",
                connection_id,
                endpoint_id,
                transport_stable_id,
                current_transport_stable_id,
                stream_id,
                source,
            );
            return false;
        }

        // Keep ownership of a losing candidate until its already-buffered
        // admission verdict is acknowledged. Concurrent bilateral admission
        // can otherwise select the same canonical stream while the losing
        // dialer observes EOF before its approval frame.
        let mut candidate_send = Some(send);
        let (installed, replaced_entry) = {
            let mut streams = self.native_control_streams.lock().await;
            let replace = streams
                .get(connection_id)
                .is_none_or(|existing| should_replace_native_control_stream(existing.owner, owner));
            if replace {
                let replaced = streams.insert(
                    connection_id.to_string(),
                    NativeControlStreamEntry {
                        owner,
                        endpoint_id,
                        send: Arc::new(tokio::sync::Mutex::new(
                            candidate_send
                                .take()
                                .expect("native control candidate must be available"),
                        )),
                    },
                );
                (true, replaced)
            } else {
                (false, None)
            }
        };
        if !installed {
            let delivery_result = retire_losing_native_control_candidate(
                candidate_send
                    .take()
                    .expect("losing native control candidate must be available"),
                recv,
            )
            .await;
            println!(
                "[OpenRTC][control] duplicate stream retired connection_id={} endpoint_id={} transport_stable_id={} stream_id={} stream_rank={} source={} verdict_delivery={}",
                connection_id,
                endpoint_id,
                transport_stable_id,
                stream_id,
                owner.stream_rank,
                source,
                if delivery_result.is_ok() { "acknowledged" } else { "unconfirmed" },
            );
            if let Err(error) = delivery_result {
                eprintln!(
                    "[OpenRTC][control] duplicate stream delivery wait ended connection_id={} endpoint_id={} transport_stable_id={} stream_id={} source={} error={}",
                    connection_id,
                    endpoint_id,
                    transport_stable_id,
                    stream_id,
                    source,
                    error,
                );
            }
            return false;
        }
        let replaced_owner = replaced_entry.as_ref().map(|entry| entry.owner);
        println!(
            "[OpenRTC][control] stream installed connection_id={} endpoint_id={} transport_stable_id={} stream_id={} stream_rank={} source={} replaced_owner={:?}",
            connection_id,
            endpoint_id,
            transport_stable_id,
            stream_id,
            owner.stream_rank,
            source,
            replaced_owner,
        );
        if let Some(replaced) = replaced_entry {
            let connection_id = connection_id.to_string();
            tokio::spawn(async move {
                let delivery_result = retire_replaced_native_control_send(replaced.send).await;
                println!(
                    "[OpenRTC][control] replaced stream retired connection_id={} endpoint_id={} transport_stable_id={} stream_rank={} verdict_delivery={}",
                    connection_id,
                    replaced.endpoint_id,
                    replaced.owner.transport_stable_id,
                    replaced.owner.stream_rank,
                    if delivery_result.is_ok() { "acknowledged" } else { "unconfirmed" },
                );
                if let Err(error) = delivery_result {
                    eprintln!(
                        "[OpenRTC][control] replaced stream delivery wait ended connection_id={} endpoint_id={} transport_stable_id={} stream_rank={} error={}",
                        connection_id,
                        replaced.endpoint_id,
                        replaced.owner.transport_stable_id,
                        replaced.owner.stream_rank,
                        error,
                    );
                }
            });
        }
        self.spawn_native_signal_stream_tasks(connection_id, endpoint_id, owner, recv);
        true
    }

    #[cfg(not(target_arch = "wasm32"))]
    fn spawn_native_signal_stream_tasks(
        &self,
        connection_id: &str,
        endpoint_id: iroh::EndpointId,
        owner: NativeControlStreamOwner,
        recv: iroh::endpoint::RecvStream,
    ) {
        let (frame_tx, frame_rx) =
            tokio::sync::mpsc::channel::<crate::native_protocol::ParsedMainFrame>(32);
        let connection_id_owned = connection_id.to_string();
        let endpoint_id_for_log = endpoint_id.to_string();

        tokio::spawn(Self::read_signal_stream_loop(
            connection_id_owned.clone(),
            endpoint_id_for_log.clone(),
            recv,
            frame_tx,
        ));

        let client = self.clone();
        tokio::spawn(async move {
            client
                .dispatch_signal_stream_frames(
                    connection_id_owned,
                    endpoint_id_for_log,
                    owner,
                    frame_rx,
                )
                .await;
        });
    }

    #[cfg(not(target_arch = "wasm32"))]
    async fn dispatch_signal_stream_frames(
        &self,
        connection_id: String,
        endpoint_id_for_log: String,
        owner: NativeControlStreamOwner,
        mut frame_rx: tokio::sync::mpsc::Receiver<crate::native_protocol::ParsedMainFrame>,
    ) {
        use crate::native_protocol::ParsedMainFrame;
        while let Some(parsed) = frame_rx.recv().await {
            if !self
                .native_signal_stream_is_current(&connection_id, owner)
                .await
            {
                println!(
                    "[SIGNAL-STREAM] stale dispatcher stopped connection_id={} transport_stable_id={} stream_rank={}",
                    connection_id, owner.transport_stable_id, owner.stream_rank,
                );
                break;
            }
            match parsed {
                ParsedMainFrame::TypeScriptJson(json) => {
                    self.maybe_handle_typescript_json_frame(
                        &connection_id,
                        Some(&endpoint_id_for_log),
                        &json,
                    )
                    .await;
                }
                ParsedMainFrame::TypeScriptHandshake(handshake) => {
                    // A TS handshake can legitimately arrive on the signal stream:
                    // when a peer's main control writer is dead after a page reload,
                    // the application-route recovery handshake is re-sent over a
                    // fresh native-signal stream (see `refreshApplicationRouteHandshake`
                    // "retrying over native signal route" on the TS side). Dispatch
                    // it through the same capability / application-crypto handler the
                    // main control reader uses (gated on existing admission) so the
                    // app route re-establishes. Previously this frame was treated as
                    // fatal and closed the signal stream, which tore down the WebRTC
                    // session, stranded the peer on a transport-only route, and
                    // blocked WebRTC promotion after a refresh.
                    if let crate::session_token::SessionAdmission::Rejected { reason } =
                        self.session_admission(&connection_id)
                    {
                        println!(
                            "[SIGNAL-STREAM] handshake-frame ignored connection_id={} endpoint_id={} reason=admission-rejected detail={}",
                            connection_id, endpoint_id_for_log, reason,
                        );
                        continue;
                    }
                    println!(
                        "[SIGNAL-STREAM] handshake-frame dispatched connection_id={} endpoint_id={} action={}",
                        connection_id,
                        endpoint_id_for_log,
                        handshake.action.as_deref().unwrap_or("<none>"),
                    );
                    if let Some(device_id) = handshake.claimed_device_id.as_deref() {
                        let _ = self
                            .bind_session_admission_authoritative_device_id(
                                &connection_id,
                                device_id,
                            )
                            .await;
                    }
                    self.maybe_handle_typescript_handshake_capabilities(
                        &connection_id,
                        Some(&endpoint_id_for_log),
                        &handshake,
                    )
                    .await;
                }
                ParsedMainFrame::NativeMessage(message) => {
                    let serialized = match serde_json::to_vec(&message) {
                        Ok(serialized) => serialized,
                        Err(error) => {
                            println!(
                                "[SIGNAL-STREAM] native-message rejected connection_id={} endpoint_id={} reason=serialize-failed error={}",
                                connection_id, endpoint_id_for_log, error,
                            );
                            continue;
                        }
                    };
                    if let Err(error) = self
                        .inspect_incoming_native_main_frame_for_transport(
                            &connection_id,
                            Some(&endpoint_id_for_log),
                            None,
                            Some(owner.transport_stable_id),
                            &serialized,
                        )
                        .await
                    {
                        println!(
                            "[SIGNAL-STREAM] native-message rejected connection_id={} endpoint_id={} channel={} error={}",
                            connection_id, endpoint_id_for_log, message.channel, error,
                        );
                    }
                }
                ParsedMainFrame::Opaque => {
                    println!(
                        "[SIGNAL-STREAM] frame skipped connection_id={} endpoint_id={} reason=unsupported-on-signal-stream kind=opaque",
                        connection_id, endpoint_id_for_log,
                    );
                }
            }
        }

        println!(
            "[SIGNAL-STREAM] dispatcher-exited connection_id={}",
            connection_id,
        );
        #[cfg(feature = "transport-webrtc")]
        self.cleanup_native_webrtc_session_after_signal_stream_end(
            &connection_id,
            owner,
            "signal-stream-ended",
        )
        .await;
        #[cfg(not(feature = "transport-webrtc"))]
        self.evict_native_signal_stream_if_owner(&connection_id, owner, "control-stream-ended")
            .await;
    }

    #[cfg(not(target_arch = "wasm32"))]
    async fn read_native_main_payload_frame(
        recv: &mut iroh::endpoint::RecvStream,
        frame_len_buf: [u8; 4],
        first_payload_byte: Option<u8>,
    ) -> Result<Option<crate::native_protocol::ParsedMainFrame>, String> {
        let frame_len = u32::from_be_bytes(frame_len_buf) as usize;
        if frame_len == 0 {
            return Ok(None);
        }
        if frame_len > 4 * 1024 * 1024 {
            return Err(format!("frame-too-large: {}", frame_len));
        }

        let mut frame = Vec::with_capacity(frame_len);
        if let Some(first) = first_payload_byte {
            frame.push(first);
        }
        if frame.len() < frame_len {
            let remaining = frame_len - frame.len();
            let mut rest = vec![0u8; remaining];
            recv.read_exact(&mut rest)
                .await
                .map_err(|error| format!("frame-read-failed: {}", error))?;
            frame.extend_from_slice(&rest);
        }

        // The signal stream is a native-main stream; classify every frame and let
        // the dispatcher decide how to handle each kind (webrtc JSON signals,
        // recovery handshakes, etc.). Returning the parsed frame instead of erroring
        // on non-signal kinds keeps the reader alive across a recovery handshake.
        Ok(Some(crate::native_protocol::parse_main_frame(&frame)))
    }

    #[cfg(not(target_arch = "wasm32"))]
    async fn read_native_main_json_frame(
        recv: &mut iroh::endpoint::RecvStream,
    ) -> Result<Option<crate::native_protocol::ParsedMainFrame>, String> {
        let mut protocol_byte = [0u8; 1];
        match recv.read_exact(&mut protocol_byte).await {
            Ok(_) => {}
            Err(error) => {
                return Err(format!("protocol-byte-read-failed: {}", error));
            }
        }
        if protocol_byte[0] != 0x00 {
            return Err(format!("unexpected-protocol-byte: {}", protocol_byte[0]));
        }

        let mut label_len_buf = [0u8; 4];
        recv.read_exact(&mut label_len_buf)
            .await
            .map_err(|error| format!("label-length-read-failed: {}", error))?;
        let label_len = u32::from_be_bytes(label_len_buf) as usize;
        if label_len == 0 || label_len > 256 {
            // Persistent native-main streams write the `[0x00][u32 labelLen]["main"]`
            // label once when the stream opens. Replies on that same stream are
            // plain `[u32 frameLen][frame]`. If we land here, we consumed the
            // first byte of the frame length as `protocol_byte` and the next
            // four bytes as `label_len_buf`; rebuild the length and keep the
            // fourth byte as the first payload byte.
            let frame_len_buf = [
                protocol_byte[0],
                label_len_buf[0],
                label_len_buf[1],
                label_len_buf[2],
            ];
            return Self::read_native_main_payload_frame(
                recv,
                frame_len_buf,
                Some(label_len_buf[3]),
            )
            .await;
        }

        let mut label_buf = vec![0u8; label_len];
        recv.read_exact(&mut label_buf)
            .await
            .map_err(|error| format!("label-read-failed: {}", error))?;
        if String::from_utf8_lossy(&label_buf) != "main" {
            return Err(format!(
                "unexpected-label: {}",
                String::from_utf8_lossy(&label_buf)
            ));
        }

        let mut frame_len_buf = [0u8; 4];
        recv.read_exact(&mut frame_len_buf)
            .await
            .map_err(|error| format!("frame-length-read-failed: {}", error))?;
        Self::read_native_main_payload_frame(recv, frame_len_buf, None).await
    }

    /// Persistent reader on the reverse half of an outbound signaling stream.
    /// Reads [4-byte big-endian len][payload] frames, parses each as JSON, and
    /// forwards them to the dispatcher task through a bounded channel.
    #[cfg(not(target_arch = "wasm32"))]
    async fn read_signal_stream_loop(
        connection_id: String,
        endpoint_id_for_log: String,
        mut recv: iroh::endpoint::RecvStream,
        frame_tx: tokio::sync::mpsc::Sender<crate::native_protocol::ParsedMainFrame>,
    ) {
        println!(
            "[SIGNAL-STREAM] reader-spawned connection_id={} endpoint_id={}",
            connection_id, endpoint_id_for_log,
        );

        loop {
            let parsed = match Self::read_native_main_json_frame(&mut recv).await {
                Ok(Some(parsed)) => parsed,
                Ok(None) => {
                    println!(
                        "[SIGNAL-STREAM] reader-empty-frame connection_id={} endpoint_id={}",
                        connection_id, endpoint_id_for_log,
                    );
                    continue;
                }
                Err(err) => {
                    println!(
                        "[SIGNAL-STREAM] reader-closed connection_id={} endpoint_id={} reason=native-main-frame-read-failed error={}",
                        connection_id, endpoint_id_for_log, err,
                    );
                    return;
                }
            };

            let frame_type = match &parsed {
                crate::native_protocol::ParsedMainFrame::TypeScriptJson(json) => json
                    .get("type")
                    .and_then(|v| v.as_str())
                    .unwrap_or("<none>"),
                crate::native_protocol::ParsedMainFrame::TypeScriptHandshake(_) => "handshake",
                crate::native_protocol::ParsedMainFrame::NativeMessage(_) => "native-message",
                crate::native_protocol::ParsedMainFrame::Opaque => "opaque",
            };
            println!(
                "[SIGNAL-STREAM] reader-frame connection_id={} endpoint_id={} frame_type={}",
                connection_id, endpoint_id_for_log, frame_type,
            );

            if frame_tx.send(parsed).await.is_err() {
                println!(
                    "[SIGNAL-STREAM] reader-closed connection_id={} endpoint_id={} reason=dispatcher-dropped",
                    connection_id, endpoint_id_for_log,
                );
                return;
            }
        }
    }

    /// Evict and drop the persistent signal stream for a connection (called
    /// when the session is removed / retired). Dropping the `SendStream`
    /// triggers a clean reset on the wire, which the remote's reader loop will
    /// observe as EOF.
    #[cfg(not(target_arch = "wasm32"))]
    pub(crate) async fn evict_native_signal_stream(&self, connection_id: &str, reason: &str) {
        let removed = {
            let mut map = self.native_control_streams.lock().await;
            map.remove(connection_id)
        };
        if let Some(entry) = removed {
            if self.native_admission_stream_is_persistent(connection_id) {
                self.invalidate_native_persistent_route_proofs_for_transport(
                    connection_id,
                    entry.owner.transport_stable_id,
                );
            }
            println!(
                "[SIGNAL-STREAM] evicted connection_id={} reason={}",
                connection_id, reason,
            );
        }
    }

    #[cfg(not(target_arch = "wasm32"))]
    async fn native_signal_stream_is_current(
        &self,
        connection_id: &str,
        expected_owner: NativeControlStreamOwner,
    ) -> bool {
        self.native_control_streams
            .lock()
            .await
            .get(connection_id)
            .is_some_and(|entry| entry.owner == expected_owner)
    }

    /// Conditional eviction is the control-stream generation barrier. Reader
    /// tasks outlive physical replacement, so only the task that still owns the
    /// registry entry may remove it.
    #[cfg(not(target_arch = "wasm32"))]
    async fn evict_native_signal_stream_if_owner(
        &self,
        connection_id: &str,
        expected_owner: NativeControlStreamOwner,
        reason: &str,
    ) -> bool {
        let removed = {
            let mut map = self.native_control_streams.lock().await;
            match map.get(connection_id) {
                Some(entry) if entry.owner == expected_owner => {
                    map.remove(connection_id);
                    true
                }
                _ => false,
            }
        };
        if removed {
            if self.native_admission_stream_is_persistent(connection_id) {
                self.invalidate_native_persistent_route_proofs_for_transport(
                    connection_id,
                    expected_owner.transport_stable_id,
                );
            }
            println!(
                "[SIGNAL-STREAM] evicted-current connection_id={} transport_stable_id={} stream_rank={} reason={}",
                connection_id,
                expected_owner.transport_stable_id,
                expected_owner.stream_rank,
                reason,
            );
        } else {
            println!(
                "[SIGNAL-STREAM] stale-eviction-ignored connection_id={} transport_stable_id={} stream_rank={} reason={}",
                connection_id,
                expected_owner.transport_stable_id,
                expected_owner.stream_rank,
                reason,
            );
        }
        removed
    }

    /// Retire a native-main owner and directional proofs that belong to an
    /// older physical Iroh leg. The logical session grant remains accepted;
    /// the existing idempotent token presentation owns route resumption.
    #[cfg(not(target_arch = "wasm32"))]
    pub(crate) async fn retire_stale_native_main_route(
        &self,
        connection_id: &str,
        current_transport_stable_id: u64,
    ) -> bool {
        // The logical grant survives replacement, but both directional route
        // proofs belong to the retired physical generation. Keeping either
        // proof current would suppress reciprocal token presentation and can
        // leave both peers waiting forever for a fresh inbound proof.
        let mut retired = false;
        let stale_owner = self
            .native_control_streams
            .lock()
            .await
            .get(connection_id)
            .map(|entry| entry.owner)
            .filter(|owner| owner.transport_stable_id != current_transport_stable_id);

        if let Some(owner) = stale_owner {
            retired |= self
                .evict_native_signal_stream_if_owner(
                    connection_id,
                    owner,
                    "transport-generation-replaced",
                )
                .await;
        }
        retired |= self.invalidate_native_main_route_proofs_except_transport(
            connection_id,
            current_transport_stable_id,
        );
        if retired {
            // Application keys may remain valid across an in-process network
            // migration, but confirmation belongs to the physical route that
            // proved possession. A relaunched peer can reuse the deterministic
            // connection id with a new ephemeral agreement. Until the existing
            // handshake owner confirms the replacement, keep SDK control open
            // and product traffic closed instead of decrypting with stale key
            // material.
            self.clear_connection_application_crypto_confirmation(connection_id);
            println!(
                "[OpenRTC][KEY-AGREEMENT] replacement requires fresh confirmation connection_id={} transport_stable_id={}",
                connection_id, current_transport_stable_id,
            );
        }
        retired
    }

    #[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
    async fn cleanup_native_webrtc_session_after_signal_stream_end(
        &self,
        connection_id: &str,
        owner: NativeControlStreamOwner,
        reason: &str,
    ) {
        if !self
            .evict_native_signal_stream_if_owner(connection_id, owner, reason)
            .await
        {
            return;
        }

        let state = {
            let sessions = self.native_webrtc_sessions.read().await;
            sessions.get(connection_id).map(|session| session.state())
        };
        if let Some(state) = state {
            println!(
                "[SIGNAL-STREAM] session-retained-after-stream-end connection_id={} webrtc_state={:?} reason={}",
                connection_id,
                state,
                reason,
            );
        }
    }

    #[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
    async fn remove_native_webrtc_session_if_current(
        &self,
        connection_id: &str,
        expected_session: &Arc<crate::transport::NativeWebRTCDataChannel>,
        reason: &str,
    ) -> bool {
        let removed = {
            let mut sessions = self.native_webrtc_sessions.write().await;
            match sessions.get(connection_id) {
                Some(current) if Arc::ptr_eq(current, expected_session) => {
                    sessions.remove(connection_id);
                    true
                }
                _ => false,
            }
        };

        if removed {
            self.clear_native_optional_route_generation_if_current(
                connection_id,
                NativeOptionalRouteKind::WebRtc,
                Arc::as_ptr(expected_session) as usize,
            )
            .await;
            println!(
                "[NativeWebRTC] removed session connection_id={} reason={}",
                connection_id, reason,
            );
            expected_session.close();
            // WebRTC is an optional avenue layered over the native-main Iroh
            // control route. Session failure, timeout, or renegotiation must
            // never evict that route; it owns admission, capability exchange,
            // and signaling for this and every other optional avenue.
        }

        removed
    }

    #[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
    pub(crate) async fn close_connecting_native_webrtc_session(
        &self,
        connection_id: &str,
        reason: &str,
    ) -> bool {
        let session = {
            let sessions = self.native_webrtc_sessions.read().await;
            sessions.get(connection_id).cloned()
        };

        let Some(session) = session else {
            return false;
        };

        if session.state() != crate::transport::NativeWebRTCState::Connecting {
            return false;
        }

        if let Some(attempt) = self.active_native_webrtc_attempt(connection_id).await {
            if attempt.preserve_on_direct_path && attempt.negotiation_id == session.negotiation_id()
            {
                println!(
                    "[NativeWebRTC] keeping explicit connecting session through direct path connection_id={} reason={} negotiation_id={}",
                    connection_id,
                    reason,
                    attempt.negotiation_id,
                );
                return false;
            }
        }

        println!(
            "[NativeWebRTC] closing connecting session connection_id={} reason={}",
            connection_id, reason,
        );
        session.close();
        let removed = self
            .remove_native_webrtc_session_if_current(connection_id, &session, reason)
            .await;
        if removed {
            self.finalize_deferred_managed_retirement_if_needed(connection_id)
                .await;
        }
        removed
    }

    #[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
    async fn dedupe_native_webrtc_session_for_remote(
        &self,
        connection_id: &str,
        remote_node_id: &str,
        force_restart: bool,
    ) -> bool {
        let related_records = self.connection_manager.get_by_node_id(remote_node_id).await;
        for record in related_records {
            if record.connection_id == connection_id {
                continue;
            }

            let session = {
                let sessions = self.native_webrtc_sessions.read().await;
                sessions.get(&record.connection_id).cloned()
            };
            let Some(session) = session else {
                continue;
            };

            match session.state() {
                crate::transport::NativeWebRTCState::Connected => {
                    println!(
                        "[NativeWebRTC] upgrade skipped: remote node already has connected session remote_node_id={} connection_id={} existing_connection_id={}",
                        remote_node_id,
                        connection_id,
                        record.connection_id,
                    );
                    return true;
                }
                crate::transport::NativeWebRTCState::Connecting if !force_restart => {
                    println!(
                        "[NativeWebRTC] upgrade skipped: remote node already has connecting session remote_node_id={} connection_id={} existing_connection_id={}",
                        remote_node_id,
                        connection_id,
                        record.connection_id,
                    );
                    return true;
                }
                crate::transport::NativeWebRTCState::Connecting => {
                    println!(
                        "[NativeWebRTC] closing duplicate connecting session remote_node_id={} connection_id={} existing_connection_id={}",
                        remote_node_id,
                        connection_id,
                        record.connection_id,
                    );
                    session.close();
                    if self
                        .remove_native_webrtc_session_if_current(
                            &record.connection_id,
                            &session,
                            "duplicate-connecting-session",
                        )
                        .await
                    {
                        self.finalize_deferred_managed_retirement_if_needed(&record.connection_id)
                            .await;
                    }
                }
                _ => {}
            }
        }

        false
    }

    #[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
    pub(crate) async fn defer_managed_retirement_while_webrtc_active(
        &self,
        connection_id: &str,
        reason: Option<String>,
    ) -> bool {
        if should_immediately_retire_active_webrtc_for_reason(reason.as_deref()) {
            self.deferred_managed_retirements
                .write()
                .await
                .remove(connection_id);
            println!(
                "[pluto-rtc][managed-retire] not deferring cleanup for terminal reason connection_id={} reason={:?}",
                connection_id,
                reason,
            );
            return false;
        }

        let Some(session) = self
            .native_webrtc_session_for_connection(connection_id)
            .await
        else {
            self.deferred_managed_retirements
                .write()
                .await
                .remove(connection_id);
            return false;
        };

        let state = session.state();
        if !matches!(
            state,
            crate::transport::NativeWebRTCState::Connecting
                | crate::transport::NativeWebRTCState::Connected
        ) {
            self.deferred_managed_retirements
                .write()
                .await
                .remove(connection_id);
            return false;
        }

        let mut deferred = self.deferred_managed_retirements.write().await;
        deferred
            .entry(connection_id.to_string())
            .and_modify(|existing| {
                if existing.is_none() {
                    *existing = reason.clone();
                }
            })
            .or_insert(reason.clone());

        println!(
            "[pluto-rtc][managed-retire] deferring cleanup until native webrtc session finishes connection_id={} webrtc_state={:?} reason={:?}",
            connection_id,
            state,
            reason,
        );
        true
    }

    /// Returns true if this connection is already waiting for its native WebRTC
    /// session to finish before the retirement is finalized.  The bridge loop
    /// uses this to avoid calling `retire_managed_connection` again (which
    /// causes the 500 ms storm seen in the desktop logs).
    #[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
    pub async fn is_managed_retirement_deferred(&self, connection_id: &str) -> bool {
        self.deferred_managed_retirements
            .read()
            .await
            .contains_key(connection_id)
    }

    #[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
    pub(crate) async fn finalize_deferred_managed_retirement_if_needed(
        &self,
        connection_id: &str,
    ) -> bool {
        let reason = self
            .deferred_managed_retirements
            .write()
            .await
            .remove(connection_id);
        let Some(reason) = reason else {
            return false;
        };

        println!(
            "[pluto-rtc][managed-retire] finalizing deferred cleanup after native webrtc session ended connection_id={} reason={:?}",
            connection_id,
            reason,
        );
        self.retire_managed_connection_now(connection_id, reason)
            .await;
        true
    }

    #[cfg(all(
        not(target_arch = "wasm32"),
        any(feature = "transport-webrtc", feature = "transport-moq")
    ))]
    async fn resolve_parallel_iroh_transport_label(
        &self,
        connection_id: &str,
        remote_node_id: Option<&str>,
    ) -> String {
        if let Some(record) = self
            .connection_manager
            .get_by_connection_id(connection_id)
            .await
        {
            if let Some(parallel) = record.parallel_transport.as_deref() {
                if crate::transport_label::is_iroh_base(parallel) {
                    return parallel.to_string();
                }
            }

            if crate::transport_label::is_iroh_base(record.active_transport.as_str()) {
                return record.active_transport;
            }
        }

        #[cfg(not(target_arch = "wasm32"))]
        {
            let path_lookup_id = remote_node_id
                .map(str::trim)
                .filter(|value| !value.is_empty())
                .unwrap_or(connection_id);
            return match self.iroh_path_kind(path_lookup_id).await {
                kind => kind.transport_label().to_string(),
            };
        }

        #[cfg(target_arch = "wasm32")]
        {
            let _ = remote_node_id;
            "iroh".to_string()
        }
    }

    pub async fn update_transport_config(
        &self,
        transport_config: TransportConfig,
    ) -> anyhow::Result<()> {
        if let Some(moq) = transport_config.moq.as_ref() {
            anyhow::ensure!(
                !moq.relay_url.trim().is_empty(),
                "transports.moq.relayUrl is required because OpenRTC has no anonymous MoQ relay default"
            );
        }
        #[cfg(not(target_arch = "wasm32"))]
        let ble_available = self
            .native_custom_transport_kinds
            .read()
            .await
            .values()
            .any(|kind| matches!(kind, IrohPathKind::Ble));
        #[cfg(target_arch = "wasm32")]
        let ble_available = false;
        let transport_config =
            transport_config.sanitize_for_runtime_with_ble_available(ble_available);

        #[cfg(not(target_arch = "wasm32"))]
        if self.iroh_endpoint.read().await.is_some() {
            let current = self.transport_config.read().await;
            anyhow::ensure!(
                !current.endpoint_rebind_required(&transport_config),
                "irohRelayOnly, irohRelayTransportPolicy, and irohLan must be configured before Iroh endpoint initialization"
            );
        }

        let mut guard = self.transport_config.write().await;
        *guard = transport_config;
        Ok(())
    }

    pub async fn transport_config(&self) -> TransportConfig {
        self.transport_config.read().await.clone()
    }

    pub async fn is_webrtc_transport_enabled(&self) -> bool {
        self.transport_config.read().await.webrtc.is_some()
    }

    pub async fn is_moq_transport_enabled(&self) -> bool {
        self.transport_config.read().await.moq.is_some()
    }

    #[cfg(not(target_arch = "wasm32"))]
    pub async fn is_ble_transport_enabled(&self) -> bool {
        self.transport_config
            .read()
            .await
            .ble
            .as_ref()
            .is_some_and(|config| config.enabled)
            && self
                .native_transport_upgrade_available(IrohPathKind::Ble)
                .await
    }

    #[cfg(target_arch = "wasm32")]
    pub async fn is_ble_transport_enabled(&self) -> bool {
        false
    }

    #[cfg(all(not(target_arch = "wasm32"), feature = "transport-moq"))]
    async fn peer_has_application_crypto_key(&self, id: &str) -> bool {
        self.resolve_transport_connection_ids(id)
            .await
            .iter()
            .any(|connection_id| {
                self.application_crypto_key_for_connection(Some(connection_id))
                    .is_some()
            })
    }

    #[cfg(all(
        not(target_arch = "wasm32"),
        any(feature = "transport-webrtc", feature = "transport-moq")
    ))]
    pub(crate) async fn resolve_transport_connection_ids(&self, id: &str) -> Vec<String> {
        let mut connection_ids = Vec::new();
        let exact_id = id.trim();
        if !exact_id.is_empty() {
            connection_ids.push(exact_id.to_string());
        }

        connection_ids.extend(self.resolve_peer_connection_ids(id).await);

        let records = self.resolve_peer_connection_records(id).await;
        connection_ids.extend(records.into_iter().map(|record| record.connection_id));

        if connection_ids.is_empty() {
            if let Some(record) = self.best_connection_record_for_peer(id).await {
                connection_ids.push(record.connection_id);
            }
        }

        if connection_ids.len() > 1 {
            let mut unique = std::collections::HashSet::with_capacity(connection_ids.len());
            connection_ids.retain(|connection_id| unique.insert(connection_id.clone()));
        }

        connection_ids
    }

    pub async fn native_webrtc_state_for_peer(
        &self,
        id: &str,
    ) -> Option<(String, crate::transport::NativeWebRTCState)> {
        #[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
        {
            for connection_id in self.resolve_transport_connection_ids(id).await {
                let session = {
                    let sessions = self.native_webrtc_sessions.read().await;
                    sessions.get(&connection_id).cloned()
                };
                if let Some(session) = session {
                    return Some((connection_id, session.state()));
                }
            }

            return None;
        }

        #[cfg(any(target_arch = "wasm32", not(feature = "transport-webrtc")))]
        {
            let _ = id;
            None
        }
    }

    #[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
    pub(crate) async fn report_native_webrtc_candidate_transport(
        &self,
        connection_id: &str,
        remote_node_id: Option<&str>,
    ) {
        let Some(session) = self
            .native_webrtc_session_for_connection(connection_id)
            .await
        else {
            return;
        };
        self.report_native_webrtc_candidate_transport_for_session(
            connection_id,
            remote_node_id,
            &session,
        )
        .await;
    }

    #[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
    async fn report_native_webrtc_candidate_transport_for_session(
        &self,
        connection_id: &str,
        remote_node_id: Option<&str>,
        session: &Arc<crate::transport::NativeWebRTCDataChannel>,
    ) {
        if !self
            .current_native_webrtc_route_matches(connection_id, session)
            .await
        {
            return;
        }
        let fallback_transport = self
            .resolve_parallel_iroh_transport_label(connection_id, remote_node_id)
            .await;
        if !self
            .current_native_webrtc_route_matches(connection_id, session)
            .await
        {
            return;
        }
        let preserve_proven_active = self
            .connection_manager
            .get_by_connection_id(connection_id)
            .await
            .map(|record| crate::transport_label::is_webrtc(record.active_transport.as_str()))
            .unwrap_or(false);

        if preserve_proven_active {
            let _ = self
                .report_transport_status(connection_id, "webrtc", Some(fallback_transport.as_str()))
                .await;
        } else {
            let _ = self
                .report_transport_status(connection_id, fallback_transport.as_str(), Some("webrtc"))
                .await;
        }
    }

    #[cfg(any(target_arch = "wasm32", not(feature = "transport-webrtc")))]
    pub(crate) async fn report_native_webrtc_candidate_transport(
        &self,
        connection_id: &str,
        remote_node_id: Option<&str>,
    ) {
        let _ = (connection_id, remote_node_id);
    }

    pub async fn native_moq_state_for_peer(
        &self,
        id: &str,
    ) -> Option<(String, crate::transport::NativeMoQState)> {
        self.native_moq_state_detail_for_peer(id)
            .await
            .map(|detail| (detail.0, detail.1))
    }

    pub async fn native_moq_state_detail_for_peer(
        &self,
        id: &str,
    ) -> Option<(String, crate::transport::NativeMoQState, bool)> {
        #[cfg(all(not(target_arch = "wasm32"), feature = "transport-moq"))]
        {
            let mut first_seen = None;
            for connection_id in self.resolve_transport_connection_ids(id).await {
                let session = {
                    let sessions = self.native_moq_sessions.read().await;
                    sessions.get(&connection_id).cloned()
                };
                if let Some(session) = session {
                    if !self
                        .current_native_moq_route_matches(&connection_id, &session)
                        .await
                    {
                        continue;
                    }
                    let state = session.state();
                    // Public `dataReady` is an application-route fact, not merely
                    // a transport subscription fact. MoQ payloads fail closed
                    // without the negotiated application key, so reporting ready
                    // before that key exists lets projections and harnesses claim
                    // routability that the send path will immediately reject.
                    let subscribed = session.is_peer_data_subscribed().await;
                    if !self
                        .current_native_moq_route_matches(&connection_id, &session)
                        .await
                    {
                        continue;
                    }
                    let data_ready = self
                        .application_crypto_key_for_connection(Some(&connection_id))
                        .is_some()
                        && native_moq_is_data_ready(state, subscribed);
                    if data_ready {
                        return Some((connection_id, state, true));
                    }
                    if state == crate::transport::NativeMoQState::Connected {
                        first_seen.get_or_insert((connection_id, state, false));
                    } else {
                        first_seen.get_or_insert((connection_id, state, false));
                    }
                }
            }

            return first_seen;
        }

        #[cfg(any(target_arch = "wasm32", not(feature = "transport-moq")))]
        {
            let _ = id;
            None
        }
    }

    pub async fn native_moq_data_ready_for_peer(&self, id: &str) -> bool {
        #[cfg(all(not(target_arch = "wasm32"), feature = "transport-moq"))]
        {
            for connection_id in self.resolve_transport_connection_ids(id).await {
                if self
                    .application_crypto_key_for_connection(Some(&connection_id))
                    .is_none()
                {
                    continue;
                }

                let session = {
                    let sessions = self.native_moq_sessions.read().await;
                    sessions.get(&connection_id).cloned()
                };
                if let Some(session) = session {
                    if !self
                        .current_native_moq_route_matches(&connection_id, &session)
                        .await
                    {
                        continue;
                    }
                    let subscribed = session.is_peer_data_subscribed().await;
                    if self
                        .current_native_moq_route_matches(&connection_id, &session)
                        .await
                        && native_moq_is_data_ready(session.state(), subscribed)
                    {
                        return true;
                    }
                }
            }
            return false;
        }

        #[cfg(any(target_arch = "wasm32", not(feature = "transport-moq")))]
        {
            let _ = id;
            false
        }
    }

    pub async fn try_send_peer_over_webrtc(
        &self,
        id: &str,
        payload: &[u8],
    ) -> anyhow::Result<bool> {
        if payload.is_empty() {
            return Ok(true);
        }

        if !self.is_webrtc_transport_enabled().await {
            return Ok(false);
        }

        #[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
        {
            for connection_id in self.resolve_transport_connection_ids(id).await {
                let session = {
                    let sessions = self.native_webrtc_sessions.read().await;
                    sessions.get(&connection_id).cloned()
                };

                let Some(session) = session else {
                    continue;
                };

                if session.state() != crate::transport::NativeWebRTCState::Connected
                    || !self
                        .current_native_webrtc_route_matches(&connection_id, &session)
                        .await
                {
                    continue;
                }

                if session.send(payload).await.is_ok() {
                    if !self
                        .current_native_webrtc_route_matches(&connection_id, &session)
                        .await
                    {
                        continue;
                    }
                    let parallel_transport = self
                        .resolve_parallel_iroh_transport_label(&connection_id, None)
                        .await;
                    if !self
                        .current_native_webrtc_route_matches(&connection_id, &session)
                        .await
                    {
                        continue;
                    }
                    let _ = self
                        .report_transport_status(
                            &connection_id,
                            "webrtc",
                            Some(parallel_transport.as_str()),
                        )
                        .await;
                    if !self
                        .current_native_webrtc_route_matches(&connection_id, &session)
                        .await
                    {
                        continue;
                    }
                    return Ok(true);
                }
            }

            return Ok(false);
        }

        #[cfg(any(target_arch = "wasm32", not(feature = "transport-webrtc")))]
        {
            let _ = id;
            Ok(false)
        }
    }

    pub async fn get_connected_webrtc_session_for_peer(
        &self,
        id: &str,
    ) -> Option<Arc<crate::transport::NativeWebRTCDataChannel>> {
        #[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
        {
            for connection_id in self.resolve_transport_connection_ids(id).await {
                let session = {
                    let sessions = self.native_webrtc_sessions.read().await;
                    sessions.get(&connection_id).cloned()
                };

                let Some(session) = session else {
                    continue;
                };

                if session.state() == crate::transport::NativeWebRTCState::Connected
                    && self
                        .current_native_webrtc_route_matches(&connection_id, &session)
                        .await
                {
                    return Some(session);
                }
            }

            return None;
        }

        #[cfg(any(target_arch = "wasm32", not(feature = "transport-webrtc")))]
        {
            let _ = id;
            None
        }
    }

    pub async fn get_webrtc_session_for_peer(
        &self,
        id: &str,
    ) -> Option<Arc<crate::transport::NativeWebRTCDataChannel>> {
        #[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
        {
            for connection_id in self.resolve_transport_connection_ids(id).await {
                let session = {
                    let sessions = self.native_webrtc_sessions.read().await;
                    sessions.get(&connection_id).cloned()
                };

                if let Some(session) = session {
                    if native_webrtc_session_is_active(session.state())
                        && self
                            .current_native_webrtc_route_matches(&connection_id, &session)
                            .await
                    {
                        return Some(session);
                    }
                }
            }

            return None;
        }

        #[cfg(any(target_arch = "wasm32", not(feature = "transport-webrtc")))]
        {
            let _ = id;
            None
        }
    }

    pub async fn try_send_peer_over_moq(&self, id: &str, payload: &[u8]) -> anyhow::Result<bool> {
        if payload.is_empty() {
            return Ok(true);
        }

        if !self.is_moq_transport_enabled().await {
            return Ok(false);
        }

        #[cfg(all(not(target_arch = "wasm32"), feature = "transport-moq"))]
        {
            for connection_id in self.resolve_transport_connection_ids(id).await {
                let session = {
                    let sessions = self.native_moq_sessions.read().await;
                    sessions.get(&connection_id).cloned()
                };

                let Some(session) = session else {
                    continue;
                };

                if session.state() != crate::transport::NativeMoQState::Connected
                    || !self
                        .current_native_moq_route_matches(&connection_id, &session)
                        .await
                {
                    continue;
                }

                if !self
                    .wait_for_stable_native_moq_data_ready(&connection_id, &session)
                    .await
                {
                    continue;
                }

                if session.send(payload).await.is_ok() {
                    if !self
                        .current_native_moq_route_matches(&connection_id, &session)
                        .await
                    {
                        continue;
                    }
                    let parallel_transport = self
                        .resolve_parallel_iroh_transport_label(&connection_id, None)
                        .await;
                    if !self
                        .current_native_moq_route_matches(&connection_id, &session)
                        .await
                    {
                        continue;
                    }
                    let report = successful_send_status_report(parallel_transport.as_str());
                    let _ = self
                        .report_transport_status(
                            &connection_id,
                            report.active_transport,
                            report.parallel_transport,
                        )
                        .await;
                    if !self
                        .current_native_moq_route_matches(&connection_id, &session)
                        .await
                    {
                        continue;
                    }
                    return Ok(true);
                }
            }

            return Ok(false);
        }

        #[cfg(any(target_arch = "wasm32", not(feature = "transport-moq")))]
        {
            let _ = id;
            Ok(false)
        }
    }

    #[cfg(all(not(target_arch = "wasm32"), feature = "transport-moq"))]
    async fn wait_for_stable_native_moq_data_ready(
        &self,
        connection_id: &str,
        session: &Arc<crate::transport::NativeMoQSession>,
    ) -> bool {
        let deadline = tokio::time::Instant::now()
            + std::time::Duration::from_millis(Self::NATIVE_MOQ_DATA_READY_TIMEOUT_MS);
        let poll_delay =
            std::time::Duration::from_millis(Self::NATIVE_MOQ_DATA_READY_STABLE_POLL_MS);
        let mut ready_samples = 0u32;

        loop {
            if session.state() != crate::transport::NativeMoQState::Connected
                || !self
                    .current_native_moq_route_matches(connection_id, session)
                    .await
            {
                return false;
            }

            let subscribed = session.is_peer_data_subscribed().await;
            if !self
                .current_native_moq_route_matches(connection_id, session)
                .await
            {
                return false;
            }
            if subscribed {
                ready_samples = ready_samples.saturating_add(1);
                if ready_samples >= Self::NATIVE_MOQ_DATA_READY_STABLE_SAMPLES {
                    return true;
                }
            } else {
                ready_samples = 0;
            }

            let now = tokio::time::Instant::now();
            if now >= deadline {
                return false;
            }

            tokio::time::sleep(std::cmp::min(poll_delay, deadline - now)).await;
        }
    }

    #[cfg(all(not(target_arch = "wasm32"), feature = "transport-moq"))]
    async fn native_moq_has_repairable_session_for_peer(&self, id: &str) -> bool {
        for connection_id in self.resolve_transport_connection_ids(id).await {
            let session = {
                let sessions = self.native_moq_sessions.read().await;
                sessions.get(&connection_id).cloned()
            };
            let Some(session) = session else {
                continue;
            };
            if native_moq_is_repairable_state(session.state()) {
                return true;
            }
        }
        false
    }

    #[cfg(all(not(target_arch = "wasm32"), feature = "transport-moq"))]
    async fn native_moq_state_summary_for_peer(&self, id: &str) -> String {
        let mut parts = Vec::new();
        for connection_id in self.resolve_transport_connection_ids(id).await {
            let session = {
                let sessions = self.native_moq_sessions.read().await;
                sessions.get(&connection_id).cloned()
            };
            if let Some(session) = session {
                let subscribed = session.is_peer_data_subscribed().await;
                parts.push(format!(
                    "connection_id={} state={:?} data_ready={}",
                    connection_id,
                    session.state(),
                    native_moq_is_data_ready(session.state(), subscribed)
                ));
            }
        }
        if parts.is_empty() {
            "no native MoQ session".to_string()
        } else {
            parts.join("; ")
        }
    }

    /// Send `data` to `peer_id` only when a connected native MoQ session can
    /// carry the payload. Unlike `send_peer`, this never falls back to WebRTC or
    /// Iroh, so callers can use it as a deterministic transport capability
    /// probe.
    #[cfg(all(not(target_arch = "wasm32"), feature = "transport-moq"))]
    pub async fn send_peer_over_moq(&self, id: &str, data: &[u8]) -> anyhow::Result<()> {
        if data.is_empty() {
            return Ok(());
        }

        if !self.peer_has_application_crypto_key(id).await {
            anyhow::bail!(
                "MoQ peer payloads require application crypto; no key is installed for peer {id}"
            );
        }

        let protected = self.protect_outbound_peer_payload(id, data).await?;
        if self.try_send_peer_over_moq(id, &protected).await? {
            return Ok(());
        }

        let deadline = tokio::time::Instant::now()
            + std::time::Duration::from_millis(Self::NATIVE_MOQ_SEND_READY_RETRY_WINDOW_MS);
        let mut requested_repair = false;
        let mut repair_requests = 0u32;
        let mut last_repair_requested_at: Option<tokio::time::Instant> = None;
        loop {
            let now = tokio::time::Instant::now();
            if now >= deadline {
                break;
            }

            let repair_retry_elapsed = last_repair_requested_at
                .map(|instant| {
                    now.duration_since(instant)
                        >= std::time::Duration::from_millis(
                            Self::NATIVE_MOQ_SEND_REPAIR_RETRY_MIN_INTERVAL_MS,
                        )
                })
                .unwrap_or(true);
            let repairable_session = if requested_repair && repair_retry_elapsed {
                self.native_moq_has_repairable_session_for_peer(id).await
            } else {
                false
            };
            let should_request_repair = (!requested_repair || repairable_session)
                && repair_requests < Self::NATIVE_MOQ_SEND_MAX_REPAIR_REQUESTS
                && repair_retry_elapsed;
            if should_request_repair {
                repair_requests += 1;
                last_repair_requested_at = Some(now);
                let repair_started = self
                    .request_moq_repair(id, None, Some("explicit-moq-send"))
                    .await?;
                requested_repair = requested_repair || repair_started;
            }

            if !requested_repair {
                break;
            }

            tokio::time::sleep(std::time::Duration::from_millis(
                Self::NATIVE_MOQ_SEND_READY_RETRY_DELAY_MS,
            ))
            .await;
            if self.try_send_peer_over_moq(id, &protected).await? {
                return Ok(());
            }
        }

        let state = self.native_moq_state_summary_for_peer(id).await;
        anyhow::bail!("MoQ transport is not data-ready for peer {id}: {state}");
    }

    #[cfg(any(target_arch = "wasm32", not(feature = "transport-moq")))]
    pub async fn send_peer_over_moq(&self, id: &str, data: &[u8]) -> anyhow::Result<()> {
        let _ = (id, data);
        anyhow::bail!("MoQ transport is not available in this native build");
    }

    pub(crate) async fn maybe_handle_typescript_handshake_capabilities(
        &self,
        connection_id: &str,
        remote_node_id: Option<&str>,
        handshake: &crate::native_protocol::TypeScriptHandshake,
    ) {
        let remote_supports_webrtc = handshake
            .capabilities
            .as_ref()
            .and_then(|caps| caps.webrtc)
            .unwrap_or(false);
        let remote_supports_moq = handshake
            .capabilities
            .as_ref()
            .and_then(|caps| caps.moq)
            .unwrap_or(false);
        let remote_supports_ble = handshake
            .capabilities
            .as_ref()
            .and_then(|caps| caps.ble)
            .unwrap_or(false);
        #[cfg(not(target_arch = "wasm32"))]
        {
            let mut capabilities = self.native_peer_transport_capabilities.write().await;
            let peer_capabilities = capabilities.entry(connection_id.to_string()).or_default();
            if remote_supports_ble {
                peer_capabilities.insert(IrohPathKind::Ble);
            } else {
                peer_capabilities.remove(&IrohPathKind::Ble);
                if peer_capabilities.is_empty() {
                    capabilities.remove(connection_id);
                }
            }
        }
        // Browser TypeScript owns browser-side key agreement and syncs the
        // derived key into WASM. Letting the embedded WASM Client negotiate a
        // second ephemeral key races two public keys under one connection id
        // and can leave the native peer decrypting with a different key.
        #[cfg(not(target_arch = "wasm32"))]
        let application_crypto_was_confirmed =
            self.connection_application_crypto_is_confirmed(connection_id);
        #[cfg(target_arch = "wasm32")]
        let application_crypto_was_confirmed = false;
        #[cfg(not(target_arch = "wasm32"))]
        let local_application_key_agreement_public_key = self
            .maybe_negotiate_typescript_application_crypto(connection_id, remote_node_id, handshake)
            .await;
        #[cfg(target_arch = "wasm32")]
        let local_application_key_agreement_public_key = None;

        let local_webrtc_enabled = self.is_webrtc_transport_enabled().await;
        let local_moq_enabled = self.is_moq_transport_enabled().await;
        let local_ble_enabled = self.is_ble_transport_enabled().await;
        #[cfg(not(target_arch = "wasm32"))]
        let received_application_key = handshake.application_key_agreement_public_key.is_some();
        #[cfg(target_arch = "wasm32")]
        let received_application_key = false;

        if let Some(reply_action) = typescript_application_key_reply_action(
            handshake.action.as_deref(),
            received_application_key,
            local_application_key_agreement_public_key.is_some(),
            application_crypto_was_confirmed,
        ) {
            let public_key = local_application_key_agreement_public_key
                .or_else(|| self.connection_key_agreement_public_key(connection_id));
            if let Err(error) = self
                .send_typescript_handshake_update(
                    connection_id,
                    "application-key-confirmation",
                    public_key,
                    reply_action,
                )
                .await
            {
                println!(
                    "[OpenRTC][capability] key acknowledgement send failed connection_id={} remote_node_id={:?} error={}",
                    connection_id, remote_node_id, error,
                );
            }
        }
        println!(
            "[OpenRTC][transport] TS handshake capabilities connection_id={} remote_node_id={:?} remote_webrtc={} local_webrtc={} remote_moq={} local_moq={} remote_ble={} local_ble={}",
            connection_id,
            remote_node_id,
            remote_supports_webrtc,
            local_webrtc_enabled,
            remote_supports_moq,
            local_moq_enabled,
            remote_supports_ble,
            local_ble_enabled,
        );
        let should_send_capability_update = should_send_typescript_capability_update(
            local_webrtc_enabled,
            local_moq_enabled,
            local_ble_enabled,
            local_application_key_agreement_public_key.is_some() && !received_application_key,
            handshake.action.as_deref(),
        );

        if should_send_capability_update {
            // Capability publication describes this endpoint, so it must not wait
            // for the peer to advertise the same transport first. Remote support
            // gates upgrade initiation below. Application key agreement shares the
            // same handshake because native explicit streams use its ORTCE1 key.
            if let Err(error) = self
                .send_typescript_capability_update(
                    connection_id,
                    if local_webrtc_enabled || local_moq_enabled || local_ble_enabled {
                        "local-capability"
                    } else {
                        "application-key-agreement"
                    },
                    local_application_key_agreement_public_key,
                )
                .await
            {
                println!(
                    "[OpenRTC][capability] update send failed connection_id={} remote_node_id={:?} error={}",
                    connection_id,
                    remote_node_id,
                    error,
                );
            }
        } else if handshake.action.as_deref() == Some("capability-update") {
            println!(
                "[OpenRTC][capability] update received without echo connection_id={} remote_node_id={:?}",
                connection_id, remote_node_id,
            );
        }

        if remote_supports_webrtc && local_webrtc_enabled {
            #[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
            {
                // Only start a native session eagerly when *we* are the initiator.
                // If we're the responder, starting a session here creates a fresh
                // native negotiation_id which will cause the first inbound browser
                // offer (with its own negotiation_id) to be treated as stale.
                let local_node_id = self.current_node_id().await;
                let remote_node_id_str = remote_node_id.map(str::trim).filter(|v| !v.is_empty());
                let should_initiate = local_node_id
                    .as_deref()
                    .zip(remote_node_id_str)
                    .map(|(local, remote)| local > remote)
                    .unwrap_or(false);

                if should_initiate {
                    let _ = self
                        .request_native_webrtc_recovery(
                            connection_id,
                            remote_node_id,
                            NativeWebRTCRecoveryTrigger::BrowserSignal(
                                NativeWebRTCBrowserSignalTrigger::TypeScriptHandshake,
                            ),
                            NativeWebRTCRecoveryOptions {
                                force_restart: false,
                                preferred_negotiation_id: None,
                                role_override: None,
                            },
                        )
                        .await;
                } else {
                    println!(
                        "[NativeWebRTC] handshake-triggered upgrade skipped (responder) connection_id={} local_node_id={:?} remote_node_id={:?}",
                        connection_id,
                        local_node_id,
                        remote_node_id,
                    );
                }
            }
        }

        if remote_supports_moq && local_moq_enabled {
            let _ = self
                .maybe_start_native_moq_upgrade(connection_id, remote_node_id)
                .await;
        }

        #[cfg(not(target_arch = "wasm32"))]
        if remote_supports_ble && local_ble_enabled {
            let _ = self
                .maybe_start_native_ble_upgrade(connection_id, remote_node_id)
                .await;
        }
    }

    pub async fn handle_typescript_webrtc_signal_frame(
        &self,
        connection_id: &str,
        remote_node_id: Option<&str>,
        frame: &serde_json::Value,
    ) -> bool {
        if parse_webrtc_signal_message(frame).is_none() {
            return false;
        }

        self.maybe_handle_typescript_json_frame(connection_id, remote_node_id, frame)
            .await;
        true
    }

    #[cfg(not(target_arch = "wasm32"))]
    async fn send_native_ble_upgrade_control(
        &self,
        connection_id: &str,
        action: &str,
        upgrade_id: &str,
    ) -> anyhow::Result<()> {
        let frame = serde_json::json!({
            "type": "transport-upgrade",
            "transport": "ble",
            "action": action,
            "upgradeId": upgrade_id,
        });
        self.send_typescript_handshake_over_native_main(connection_id, &frame)
            .await
    }

    #[cfg(not(target_arch = "wasm32"))]
    async fn wait_for_native_ble_switch_admission(
        &self,
        connection_id: &str,
        endpoint_id: iroh::EndpointId,
        expected_transport_stable_id: u64,
    ) -> anyhow::Result<()> {
        let deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(15);
        loop {
            let current_transport_stable_id = self
                .get_connection(endpoint_id)
                .await
                .map(|connection| connection.stable_id() as u64);
            if current_transport_stable_id != Some(expected_transport_stable_id) {
                anyhow::bail!(
                    "BLE switch request lost its base transport generation: expected={} current={:?}",
                    expected_transport_stable_id,
                    current_transport_stable_id,
                );
            }

            let block = self.session_admission_block_reason_for_transport(
                connection_id,
                Some(expected_transport_stable_id),
            );
            match classify_native_ble_switch_admission(block.as_ref()) {
                NativeBleSwitchAdmission::Ready => return Ok(()),
                NativeBleSwitchAdmission::Reject => {
                    anyhow::bail!(
                        "BLE switch request admission rejected: {}",
                        block
                            .as_ref()
                            .map(|(_, reason)| reason.as_str())
                            .unwrap_or("unknown"),
                    );
                }
                NativeBleSwitchAdmission::Wait => {}
            }

            if tokio::time::Instant::now() >= deadline {
                anyhow::bail!(
                    "timed out waiting for BLE switch admission: {}",
                    block
                        .as_ref()
                        .map(|(_, reason)| reason.as_str())
                        .unwrap_or("unknown"),
                );
            }
            tokio::time::sleep(std::time::Duration::from_millis(50)).await;
        }
    }

    #[cfg(not(target_arch = "wasm32"))]
    async fn accept_native_ble_switch_request(
        &self,
        connection_id: &str,
        remote_node_id: &str,
        upgrade_id: &str,
    ) -> anyhow::Result<bool> {
        let Some(local_node_id) = self.current_node_id().await else {
            return Ok(false);
        };
        let current_path = self.iroh_path_kind(remote_node_id).await;
        if !should_accept_native_ble_switch_request(&local_node_id, remote_node_id, current_path) {
            return Ok(false);
        }

        let gate_key = (connection_id.to_string(), IrohPathKind::Ble);
        {
            let mut gates = self.native_transport_upgrade_gates.lock().await;
            match gates.get(&gate_key) {
                Some(expected) if expected != upgrade_id => return Ok(false),
                Some(_) => {}
                None => {
                    gates.insert(gate_key.clone(), upgrade_id.to_string());
                }
            }
        }

        if let Err(error) = self
            .send_native_ble_upgrade_control(connection_id, "switch-ready", upgrade_id)
            .await
        {
            remove_native_transport_upgrade_gate(
                &self.native_transport_upgrade_gates,
                &gate_key,
                upgrade_id,
            )
            .await;
            return Err(error);
        }

        println!(
            "[OpenRTC][BLE] switch request accepted connection_id={} remote_node_id={} upgrade_id={}",
            connection_id, remote_node_id, upgrade_id,
        );
        let client = self.clone();
        let remote = remote_node_id.to_string();
        let gates = self.native_transport_upgrade_gates.clone();
        let responder_upgrade_id = upgrade_id.to_string();
        tokio::spawn(async move {
            tokio::time::sleep(std::time::Duration::from_millis(125)).await;
            let Ok(endpoint_id) = remote.parse::<iroh::EndpointId>() else {
                return;
            };
            let node = client.iroh_node.read().await.clone();
            if let Some(node) = node {
                let _ = node
                    .disconnect_with_reason(
                        endpoint_id,
                        crate::lifecycle_reason::REASON_REPLACEMENT_IN_PROGRESS,
                    )
                    .await;
            }
            tokio::time::sleep(std::time::Duration::from_secs(45)).await;
            remove_native_transport_upgrade_gate(&gates, &gate_key, &responder_upgrade_id).await;
        });
        Ok(true)
    }

    #[cfg(not(target_arch = "wasm32"))]
    pub(super) async fn maybe_start_native_ble_upgrade(
        &self,
        connection_id: &str,
        remote_node_id: Option<&str>,
    ) -> anyhow::Result<()> {
        let remote_node_id = remote_node_id
            .map(str::trim)
            .filter(|value| !value.is_empty())
            .ok_or_else(|| anyhow::anyhow!("BLE upgrade requires a remote node id"))?;
        let local_node_id = self
            .current_node_id()
            .await
            .ok_or_else(|| anyhow::anyhow!("BLE upgrade requires an initialized native node"))?;
        let current_path = self.iroh_path_kind(remote_node_id).await;
        if !should_initiate_native_ble_upgrade(local_node_id.as_str(), remote_node_id, current_path)
        {
            println!(
                "[OpenRTC][BLE] upgrade decision connection_id={} local_node_id={} remote_node_id={} current_path={:?} role=responder-or-better-path",
                connection_id, local_node_id, remote_node_id, current_path,
            );
            return Ok(());
        }
        println!(
            "[OpenRTC][BLE] upgrade decision connection_id={} local_node_id={} remote_node_id={} current_path={:?} role=initiator",
            connection_id, local_node_id, remote_node_id, current_path,
        );

        let gate_key = (connection_id.to_string(), IrohPathKind::Ble);
        let upgrade_id = uuid::Uuid::new_v4().to_string();
        {
            let mut gates = self.native_transport_upgrade_gates.lock().await;
            if gates.contains_key(&gate_key) {
                return Ok(());
            }
            gates.insert(gate_key.clone(), upgrade_id.clone());
        }

        let result = async {
            let endpoint_id = remote_node_id.parse::<iroh::EndpointId>()?;
            let provider = self
                .native_transport_upgrade_provider(IrohPathKind::Ble)
                .await
                .ok_or_else(|| anyhow::anyhow!("BLE upgrade provider is unavailable"))?;
            // Targeted discovery happens while the admitted Iroh control route
            // is still live. A failed scan leaves that base route untouched.
            let _ = provider.prepare_endpoint_addr(endpoint_id).await?;
            self.send_native_ble_upgrade_control(connection_id, "switch-request", &upgrade_id)
                .await?;
            println!(
                "[OpenRTC][BLE] prepared upgrade connection_id={} remote_node_id={} upgrade_id={}",
                connection_id, remote_node_id, upgrade_id
            );
            Ok::<(), anyhow::Error>(())
        }
        .await;

        if result.is_err() {
            remove_native_transport_upgrade_gate(
                &self.native_transport_upgrade_gates,
                &gate_key,
                &upgrade_id,
            )
            .await;
        } else {
            let gates = self.native_transport_upgrade_gates.clone();
            tokio::spawn(async move {
                tokio::time::sleep(std::time::Duration::from_secs(45)).await;
                remove_native_transport_upgrade_gate(&gates, &gate_key, &upgrade_id).await;
            });
        }
        result
    }

    #[cfg(not(target_arch = "wasm32"))]
    async fn handle_native_ble_upgrade_control(
        &self,
        connection_id: &str,
        remote_node_id: Option<&str>,
        frame: &serde_json::Value,
    ) -> bool {
        if frame.get("type").and_then(|value| value.as_str()) != Some("transport-upgrade")
            || frame.get("transport").and_then(|value| value.as_str()) != Some("ble")
        {
            return false;
        }

        let Some(remote_node_id) = remote_node_id
            .map(str::trim)
            .filter(|value| !value.is_empty())
        else {
            return true;
        };
        let action = frame
            .get("action")
            .and_then(|value| value.as_str())
            .unwrap_or_default();
        let upgrade_id = frame
            .get("upgradeId")
            .and_then(|value| value.as_str())
            .map(str::trim)
            .filter(|value| !value.is_empty())
            .unwrap_or("legacy");

        if !self.is_ble_transport_enabled().await {
            return true;
        }

        let Some(local_node_id) = self.current_node_id().await else {
            return true;
        };
        let current_path = self.iroh_path_kind(remote_node_id).await;
        println!(
            "[OpenRTC][BLE] control received connection_id={} remote_node_id={} action={} upgrade_id={} current_path={:?}",
            connection_id, remote_node_id, action, upgrade_id, current_path,
        );

        match action {
            "switch-request" => {
                if !should_accept_native_ble_switch_request(
                    &local_node_id,
                    remote_node_id,
                    current_path,
                ) {
                    return true;
                }
                let Ok(endpoint_id) = remote_node_id.parse::<iroh::EndpointId>() else {
                    return true;
                };
                let Some(transport_stable_id) = self
                    .get_connection(endpoint_id)
                    .await
                    .map(|connection| connection.stable_id() as u64)
                else {
                    return true;
                };
                let admission_block = self.session_admission_block_reason_for_transport(
                    connection_id,
                    Some(transport_stable_id),
                );
                match classify_native_ble_switch_admission(admission_block.as_ref()) {
                    NativeBleSwitchAdmission::Reject => {
                        println!(
                            "[OpenRTC][BLE] switch request rejected connection_id={} remote_node_id={} upgrade_id={} reason={}",
                            connection_id,
                            remote_node_id,
                            upgrade_id,
                            admission_block
                                .as_ref()
                                .map(|(_, reason)| reason.as_str())
                                .unwrap_or("unknown"),
                        );
                        return true;
                    }
                    NativeBleSwitchAdmission::Wait => {
                        let gate_key = (connection_id.to_string(), IrohPathKind::Ble);
                        {
                            let mut gates = self.native_transport_upgrade_gates.lock().await;
                            match gates.get(&gate_key) {
                                Some(expected) if expected != upgrade_id => return true,
                                Some(_) => return true,
                                None => {
                                    gates.insert(gate_key.clone(), upgrade_id.to_string());
                                }
                            }
                        }
                        println!(
                            "[OpenRTC][BLE] switch request deferred connection_id={} remote_node_id={} upgrade_id={} transport_stable_id={} reason={}",
                            connection_id,
                            remote_node_id,
                            upgrade_id,
                            transport_stable_id,
                            admission_block
                                .as_ref()
                                .map(|(_, reason)| reason.as_str())
                                .unwrap_or("unknown"),
                        );
                        let client = self.clone();
                        let connection_id = connection_id.to_string();
                        let remote_node_id = remote_node_id.to_string();
                        let upgrade_id = upgrade_id.to_string();
                        tokio::spawn(async move {
                            let result = client
                                .wait_for_native_ble_switch_admission(
                                    &connection_id,
                                    endpoint_id,
                                    transport_stable_id,
                                )
                                .await;
                            if let Err(error) = result {
                                remove_native_transport_upgrade_gate(
                                    &client.native_transport_upgrade_gates,
                                    &(connection_id.clone(), IrohPathKind::Ble),
                                    &upgrade_id,
                                )
                                .await;
                                eprintln!(
                                    "[OpenRTC][BLE] deferred switch request failed connection_id={} remote_node_id={} upgrade_id={} error={}",
                                    connection_id, remote_node_id, upgrade_id, error,
                                );
                                return;
                            }
                            if let Err(error) = client
                                .accept_native_ble_switch_request(
                                    &connection_id,
                                    &remote_node_id,
                                    &upgrade_id,
                                )
                                .await
                            {
                                eprintln!(
                                    "[OpenRTC][BLE] deferred switch request failed connection_id={} remote_node_id={} upgrade_id={} error={}",
                                    connection_id, remote_node_id, upgrade_id, error,
                                );
                            }
                        });
                        return true;
                    }
                    NativeBleSwitchAdmission::Ready => {}
                }
                if let Err(error) = self
                    .accept_native_ble_switch_request(connection_id, remote_node_id, upgrade_id)
                    .await
                {
                    eprintln!(
                        "[OpenRTC][BLE] switch request failed connection_id={} remote_node_id={} upgrade_id={} error={}",
                        connection_id, remote_node_id, upgrade_id, error,
                    );
                }
            }
            "switch-ready" => {
                let gate_key = (connection_id.to_string(), IrohPathKind::Ble);
                let expected = self
                    .native_transport_upgrade_gates
                    .lock()
                    .await
                    .get(&gate_key)
                    .is_some_and(|expected| expected == upgrade_id);
                if !expected {
                    println!(
                        "[OpenRTC][BLE] switch ready ignored connection_id={} remote_node_id={} upgrade_id={} reason=generation-mismatch",
                        connection_id, remote_node_id, upgrade_id,
                    );
                    return true;
                }
                println!(
                    "[OpenRTC][BLE] switch ready accepted connection_id={} remote_node_id={} upgrade_id={}",
                    connection_id, remote_node_id, upgrade_id,
                );
                let Ok(endpoint_id) = remote_node_id.parse::<iroh::EndpointId>() else {
                    return true;
                };
                let Some(transport_stable_id) = self
                    .get_connection(endpoint_id)
                    .await
                    .map(|connection| connection.stable_id() as u64)
                else {
                    return true;
                };
                let client = self.clone();
                let connection_id = connection_id.to_string();
                let remote = remote_node_id.to_string();
                let upgrade_id = upgrade_id.to_string();
                tokio::spawn(async move {
                    let gate_key = (connection_id.clone(), IrohPathKind::Ble);
                    if let Err(error) = client
                        .wait_for_native_ble_switch_admission(
                            &connection_id,
                            endpoint_id,
                            transport_stable_id,
                        )
                        .await
                    {
                        remove_native_transport_upgrade_gate(
                            &client.native_transport_upgrade_gates,
                            &gate_key,
                            &upgrade_id,
                        )
                        .await;
                        eprintln!(
                            "[OpenRTC][BLE] switch ready admission failed connection_id={} remote_node_id={} upgrade_id={} error={}",
                            connection_id, remote, upgrade_id, error,
                        );
                        return;
                    }
                    let result = client
                        .complete_native_ble_upgrade(&connection_id, &remote, &upgrade_id)
                        .await;
                    remove_native_transport_upgrade_gate(
                        &client.native_transport_upgrade_gates,
                        &gate_key,
                        &upgrade_id,
                    )
                    .await;
                    if let Err(error) = result {
                        eprintln!(
                            "[OpenRTC][BLE] upgrade failed connection_id={} remote_node_id={} upgrade_id={} error={}",
                            connection_id, remote, upgrade_id, error
                        );
                    }
                });
            }
            _ => {}
        }
        true
    }

    #[cfg(not(target_arch = "wasm32"))]
    async fn complete_native_ble_upgrade(
        &self,
        connection_id: &str,
        remote_node_id: &str,
        upgrade_id: &str,
    ) -> anyhow::Result<()> {
        use futures::StreamExt as _;

        let endpoint_id = remote_node_id.parse::<iroh::EndpointId>()?;
        let provider = self
            .native_transport_upgrade_provider(IrohPathKind::Ble)
            .await
            .ok_or_else(|| anyhow::anyhow!("BLE upgrade provider is unavailable"))?;
        println!(
            "[OpenRTC][BLE] replacement connect starting connection_id={} remote_node_id={} upgrade_id={}",
            connection_id, remote_node_id, upgrade_id,
        );
        let endpoint_addr = provider.prepare_endpoint_addr(endpoint_id).await?;
        let mut events = {
            let node = self.iroh_node.read().await;
            let node = node
                .as_ref()
                .ok_or_else(|| anyhow::anyhow!("Iroh node is unavailable"))?;
            node.connect_replacement_addr(endpoint_id, endpoint_addr, provider.transport_id())
        };
        let connected = tokio::time::timeout(std::time::Duration::from_secs(30), async {
            while let Some(event) = events.next().await {
                match event {
                    crate::native_node::ConnectEvent::Connected => {
                        println!(
                            "[OpenRTC][BLE] replacement transport connected connection_id={} remote_node_id={} upgrade_id={}",
                            connection_id, remote_node_id, upgrade_id,
                        );
                        return Ok(())
                    }
                    crate::native_node::ConnectEvent::Closed { error } => {
                        anyhow::bail!(error.unwrap_or_else(|| "replacement closed".to_string()));
                    }
                }
            }
            anyhow::bail!("replacement event stream ended")
        })
        .await
        .map_err(|_| anyhow::anyhow!("timed out waiting for BLE replacement"))??;
        let _ = connected;
        self.finalize_transport_dial_record(endpoint_id, Some("ble-upgrade".to_string()))
            .await?;
        self.migrate_remote_admission_proof_for_authorized_ble_upgrade(
            connection_id,
            endpoint_id,
            upgrade_id,
        )
        .await;
        let known_device_id = self
            .known_device_ids_by_node
            .read()
            .ok()
            .and_then(|known| known.get(remote_node_id).cloned());
        if let Some(device_id) = known_device_id {
            let _ = self
                .mark_trusted_user_device_connection_admitted(connection_id, &device_id)
                .await;
        }
        let path = self.iroh_path_kind(remote_node_id).await;
        if path != IrohPathKind::Ble {
            anyhow::bail!("replacement connected without selecting BLE: {path:?}");
        }
        println!(
            "[OpenRTC][BLE] upgrade complete connection_id={} remote_node_id={} upgrade_id={}",
            connection_id, remote_node_id, upgrade_id
        );
        Ok(())
    }

    #[cfg(not(target_arch = "wasm32"))]
    pub(crate) async fn migrate_remote_admission_proof_for_authorized_ble_upgrade(
        &self,
        connection_id: &str,
        endpoint_id: iroh::EndpointId,
        upgrade_id: &str,
    ) -> bool {
        let gate_key = (connection_id.to_string(), IrohPathKind::Ble);
        let authorized = self
            .native_transport_upgrade_gates
            .lock()
            .await
            .get(&gate_key)
            .is_some_and(|expected| expected == upgrade_id);
        if !authorized || self.iroh_path_kind(&endpoint_id.to_string()).await != IrohPathKind::Ble {
            return false;
        }
        if !matches!(
            self.session_admission(connection_id),
            crate::session_token::SessionAdmission::Accepted {
                mechanism: crate::session_token::SessionAdmissionMechanism::SessionToken,
                ..
            }
        ) {
            return false;
        }
        let Some(stable_id) = self
            .get_connection(endpoint_id)
            .await
            .map(|connection| connection.stable_id() as u64)
        else {
            return false;
        };
        let migrated = self
            .remote_session_admission_proofs
            .write()
            .ok()
            .and_then(|mut proofs| {
                let proof = proofs.get_mut(connection_id)?;
                let previous_stable_id = proof.transport_stable_id;
                proof.transport_stable_id = stable_id;
                Some(previous_stable_id)
            });
        if let Some(previous_stable_id) = migrated {
            println!(
                "[OpenRTC][BLE][session-admission] migrated remote proof connection_id={} endpoint_id={} upgrade_id={} previous_stable_id={} stable_id={}",
                connection_id, endpoint_id, upgrade_id, previous_stable_id, stable_id
            );
            true
        } else {
            false
        }
    }

    #[cfg(not(target_arch = "wasm32"))]
    pub(crate) async fn migrate_remote_admission_proof_for_pending_ble_upgrade(
        &self,
        connection_id: &str,
        endpoint_id: iroh::EndpointId,
    ) -> bool {
        let upgrade_id = self
            .native_transport_upgrade_gates
            .lock()
            .await
            .get(&(connection_id.to_string(), IrohPathKind::Ble))
            .cloned();
        let Some(upgrade_id) = upgrade_id else {
            return false;
        };
        self.migrate_remote_admission_proof_for_authorized_ble_upgrade(
            connection_id,
            endpoint_id,
            &upgrade_id,
        )
        .await
    }

    pub(crate) async fn maybe_handle_typescript_json_frame(
        &self,
        connection_id: &str,
        remote_node_id: Option<&str>,
        frame: &serde_json::Value,
    ) {
        #[cfg(not(target_arch = "wasm32"))]
        if self
            .handle_native_ble_upgrade_control(connection_id, remote_node_id, frame)
            .await
        {
            return;
        }
        let frame_type = frame
            .get("type")
            .and_then(|value| value.as_str())
            .unwrap_or("<missing>");
        let content_transport = frame
            .get("content")
            .and_then(|value| value.get("transport"))
            .and_then(|value| value.as_str())
            .unwrap_or("<missing>");
        let signal = match parse_webrtc_signal_message(frame) {
            Some(value) => value,
            None => {
                if frame_type == "#pluto-signal" {
                    println!(
                        "[NativeWebRTC] inbound signal envelope ignored connection_id={} remote_node_id={:?} reason=unsupported-or-malformed transport={} has_content={} has_sdp={} has_candidate={} negotiation_id={}",
                        connection_id,
                        remote_node_id,
                        content_transport,
                        frame.get("content").is_some(),
                        frame
                            .get("content")
                            .and_then(|value| value.get("sdp"))
                            .is_some(),
                        frame
                            .get("content")
                            .and_then(|value| value.get("candidate"))
                            .is_some(),
                        frame
                            .get("content")
                            .and_then(|value| value.get("negotiationId"))
                            .and_then(|value| value.as_str())
                            .unwrap_or("legacy")
                    );
                }
                return;
            }
        };

        println!(
            "[NativeWebRTC] inbound signal envelope detected connection_id={} remote_node_id={:?} signal_type={} negotiation_id={}",
            connection_id,
            remote_node_id,
            signal.signal_type,
            signal.negotiation_id.as_deref().unwrap_or("legacy"),
        );
        println!(
            "[NativeWebRTC] inbound signal dispatch begin connection_id={} remote_node_id={:?} signal_type={} negotiation_id={} has_sdp={} has_candidate={}",
            connection_id,
            remote_node_id,
            signal.signal_type,
            signal.negotiation_id.as_deref().unwrap_or("legacy"),
            signal.sdp.is_some(),
            signal.candidate.is_some(),
        );

        if !self.is_webrtc_transport_enabled().await {
            println!(
                "[NativeWebRTC] inbound signal dispatch skipped connection_id={} remote_node_id={:?} reason=webrtc-disabled signal_type={} negotiation_id={}",
                connection_id,
                remote_node_id,
                signal.signal_type,
                signal.negotiation_id.as_deref().unwrap_or("legacy"),
            );
            return;
        }

        // "renegotiate" is sent by the web peer when it retries after an ICE
        // disconnect (e.g. page refresh or a stuck ICE restart).  Close the
        // existing session (even if Connected or stuck in Connecting) so a
        // fresh SDP offer is generated for the new RTCPeerConnection.
        //
        // Security: the inbound `renegotiate` envelope arrives over the iroh
        // QUIC control stream, whose TLS binds the sender's node_id — the
        // sender cannot forge a peer identity. Before we start a new upgrade
        // session, `maybe_start_native_webrtc_upgrade` re-checks
        // `session_admission(connection_id)` and aborts on Rejected/Pending,
        // so the replacement session is only created if the remote is still
        // an admitted peer. For a true browser refresh (new WASM iroh key),
        // the browser presents its session token on first iroh contact and
        // only becomes Admitted once the token is validated server-side, so
        // the admission gate here stays authoritative.
        if signal.signal_type == "renegotiate" {
            #[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
            {
                if let crate::session_token::SessionAdmission::Rejected { reason } =
                    self.session_admission(connection_id)
                {
                    println!(
                        "[NativeWebRTC] renegotiate ignored: session admission rejected connection_id={} remote_node_id={:?} reason={}",
                        connection_id, remote_node_id, reason,
                    );
                    return;
                }
                let active_attempt = self.active_native_webrtc_attempt(connection_id).await;
                if let Some(session) = self
                    .native_webrtc_session_for_connection(connection_id)
                    .await
                {
                    let existing_state = session.state();
                    let signal_negotiation_id = signal
                        .negotiation_id
                        .as_deref()
                        .map(str::trim)
                        .filter(|value| !value.is_empty());
                    let negotiation_mismatch = signal_negotiation_id
                        .map(|value| value != session.negotiation_id().as_str())
                        .unwrap_or(false);
                    if existing_state == crate::transport::NativeWebRTCState::Connected {
                        println!(
                            "[NativeWebRTC] renegotiate ignored: connected session owns settled generation connection_id={} remote_node_id={:?} active_negotiation_id={} signal_negotiation_id={} negotiation_mismatch={} has_active_attempt={}",
                            connection_id,
                            remote_node_id,
                            session.negotiation_id(),
                            signal_negotiation_id.unwrap_or("none"),
                            negotiation_mismatch,
                            active_attempt.is_some(),
                        );
                        return;
                    }
                    if existing_state == crate::transport::NativeWebRTCState::Connecting {
                        if !negotiation_mismatch {
                            println!(
                                "[NativeWebRTC] renegotiate ignored: same negotiation still connecting connection_id={} remote_node_id={:?} active_negotiation_id={}",
                                connection_id,
                                remote_node_id,
                                session.negotiation_id(),
                            );
                            return;
                        }
                        if let Some(attempt) = active_attempt.as_ref() {
                            let age_ms = now_millis_i64().saturating_sub(attempt.started_at_ms);
                            if age_ms < Self::NATIVE_WEBRTC_ATTEMPT_IN_FLIGHT_GRACE_MS {
                                println!(
                                    "[NativeWebRTC] renegotiate ignored: connection attempt still inside grace window connection_id={} remote_node_id={:?} active_negotiation_id={} signal_negotiation_id={} age_ms={} grace_ms={}",
                                    connection_id,
                                    remote_node_id,
                                    attempt.negotiation_id,
                                    signal_negotiation_id.unwrap_or("none"),
                                    age_ms,
                                    Self::NATIVE_WEBRTC_ATTEMPT_IN_FLIGHT_GRACE_MS,
                                );
                                return;
                            }
                        }
                        println!(
                            "[NativeWebRTC] renegotiate replacing stuck Connecting session connection_id={} remote_node_id={:?}",
                            connection_id,
                            remote_node_id,
                        );
                    }
                }
            }
            println!(
                "[NativeWebRTC] renegotiate requested: restarting session connection_id={} remote_node_id={:?}",
                connection_id,
                remote_node_id,
            );
            // Reset the attempt budget — renegotiate signals a fresh browser
            // session (e.g. page refresh), so old retry history is irrelevant.
            #[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
            self.reset_native_webrtc_attempt_budget(connection_id, "renegotiate")
                .await;
            // Explicit renegotiate supersedes any in-flight attempt — drop
            // the marker so the incoming fresh-session path can start cleanly.
            #[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
            self.clear_native_webrtc_attempt(connection_id).await;
            #[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
            {
                if let Some(session) = self
                    .native_webrtc_session_for_connection(connection_id)
                    .await
                {
                    session.close();
                    if self
                        .remove_native_webrtc_session_if_current(
                            connection_id,
                            &session,
                            "renegotiate-restart",
                        )
                        .await
                    {
                        self.finalize_deferred_managed_retirement_if_needed(connection_id)
                            .await;
                    }
                }
            }
            #[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
            {
                if let Err(error) = self
                    .request_native_webrtc_recovery(
                        connection_id,
                        remote_node_id,
                        NativeWebRTCRecoveryTrigger::BrowserSignal(
                            NativeWebRTCBrowserSignalTrigger::InboundRenegotiate,
                        ),
                        NativeWebRTCRecoveryOptions {
                            force_restart: true,
                            preferred_negotiation_id: signal.negotiation_id.as_deref(),
                            role_override: None,
                        },
                    )
                    .await
                {
                    println!(
                        "[NativeWebRTC] renegotiate: failed to restart session connection_id={} error={}",
                        connection_id,
                        error,
                    );
                }
            }
            return;
        }

        if signal.signal_type == "sdp" {
            #[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
            {
                if let Some(existing) = self
                    .native_webrtc_session_for_connection(connection_id)
                    .await
                {
                    let signal_negotiation_id = signal
                        .negotiation_id
                        .as_deref()
                        .map(str::trim)
                        .filter(|value| !value.is_empty());
                    let negotiation_mismatch = signal_negotiation_id
                        .map(|value| value != existing.negotiation_id().as_str())
                        .unwrap_or(false);
                    let is_offer = signal.sdp.as_ref().is_some_and(|value| {
                        value
                            .get("type")
                            .and_then(serde_json::Value::as_str)
                            .map(|value| value == "offer")
                            .unwrap_or(false)
                    });

                    if negotiation_mismatch
                        && is_offer
                        && native_webrtc_session_is_active(existing.state())
                    {
                        println!(
                            "[NativeWebRTC] mismatched SDP offer ignored: explicit renegotiate required connection_id={} remote_node_id={:?} active_negotiation_id={} signal_negotiation_id={} state={:?}",
                            connection_id,
                            remote_node_id,
                            existing.negotiation_id(),
                            signal_negotiation_id.unwrap_or("none"),
                            existing.state(),
                        );
                        return;
                    }
                }

                if let Err(error) = self
                    .request_native_webrtc_recovery(
                        connection_id,
                        remote_node_id,
                        NativeWebRTCRecoveryTrigger::BrowserSignal(
                            NativeWebRTCBrowserSignalTrigger::InboundSdp,
                        ),
                        NativeWebRTCRecoveryOptions {
                            force_restart: false,
                            preferred_negotiation_id: signal.negotiation_id.as_deref(),
                            role_override: None,
                        },
                    )
                    .await
                {
                    println!(
                        "[NativeWebRTC] signal ignored: failed to initialize upgrade session connection_id={} remote_node_id={:?} error={}",
                        connection_id,
                        remote_node_id,
                        error,
                    );
                    return;
                }
            }
        } else {
            let mut has_session_for_signal = self
                .get_webrtc_session_for_peer(connection_id)
                .await
                .is_some();
            if !has_session_for_signal {
                if let Some(remote_node_id) = remote_node_id
                    .map(str::trim)
                    .filter(|value| !value.is_empty())
                {
                    has_session_for_signal = self
                        .get_webrtc_session_for_peer(remote_node_id)
                        .await
                        .is_some();
                }
            }

            if !has_session_for_signal {
                return;
            }
        }

        if let Err(error) = self
            .forward_native_webrtc_signal(connection_id, remote_node_id, signal)
            .await
        {
            println!(
                "[NativeWebRTC] signal ignored: forward failed connection_id={} remote_node_id={:?} error={}",
                connection_id,
                remote_node_id,
                error,
            );
        } else {
            println!(
                "[NativeWebRTC] inbound signal dispatched to session connection_id={} remote_node_id={:?}",
                connection_id,
                remote_node_id,
            );
        }
    }

    // When `force_restart` is true, a session in `Connecting` state is also replaced.
    // Use this when a *fresh* iroh connection arrives for the same peer (e.g. browser
    // refresh) so that the stale in-flight handshake does not block a new offer from
    // being sent to the new browser session.
    #[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
    pub(crate) async fn maybe_start_native_webrtc_upgrade(
        &self,
        connection_id: &str,
        remote_node_id: Option<&str>,
        force_restart: bool,
        preferred_negotiation_id: Option<&str>,
        role_override: Option<crate::transport::NativeWebRTCRole>,
    ) -> anyhow::Result<()> {
        if connection_id.trim().is_empty() {
            return Ok(());
        }

        let remote_node_id = match remote_node_id
            .map(str::trim)
            .filter(|value| !value.is_empty())
        {
            Some(value) => value.to_string(),
            None => {
                let fallback = self
                    .connection_manager
                    .get_by_connection_id(connection_id)
                    .await
                    .and_then(|record| record.endpoint_id.or(record.node_id))
                    .map(|value| value.trim().to_string())
                    .filter(|value| !value.is_empty());

                match fallback {
                    Some(value) => value,
                    None => {
                        println!("[NativeWebRTC] upgrade skipped: no remote_node_id for connection_id={}", connection_id);
                        return Ok(());
                    }
                }
            }
        };

        let webrtc_config = {
            let guard = self.transport_config.read().await;
            match guard.webrtc.clone() {
                Some(value) => value,
                None => {
                    println!("[NativeWebRTC] upgrade skipped: webrtc config is None for connection_id={}", connection_id);
                    return Ok(());
                }
            }
        };

        let local_node_id = match self.current_node_id().await {
            Some(value) => value,
            None => {
                println!(
                    "[NativeWebRTC] upgrade skipped: no local_node_id for connection_id={}",
                    connection_id
                );
                return Ok(());
            }
        };

        let preferred_negotiation_id = preferred_negotiation_id
            .map(str::trim)
            .filter(|value| !value.is_empty())
            .map(ToOwned::to_owned);

        #[cfg(any(target_arch = "wasm32", not(feature = "transport-webrtc")))]
        let _ = (force_restart, preferred_negotiation_id, role_override);

        #[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
        {
            if let Some(reason) = self
                .native_webrtc_upgrade_fast_path_skip_reason(
                    connection_id,
                    Some(&remote_node_id),
                    force_restart,
                    preferred_negotiation_id.as_deref(),
                )
                .await
            {
                println!(
                    "[NativeWebRTC] upgrade skipped: fast-path guard connection_id={} remote_node_id={} reason={}",
                    connection_id,
                    remote_node_id,
                    reason,
                );
                return Ok(());
            }

            self.cancel_native_webrtc_retry(connection_id).await;

            let start_gate = self
                .acquire_native_webrtc_start_gate(
                    connection_id,
                    &remote_node_id,
                    force_restart,
                    preferred_negotiation_id.as_deref(),
                )
                .await;

            let result = async {
                if let crate::session_token::SessionAdmission::Rejected { reason } =
                    self.session_admission(connection_id)
                {
                    self.suppress_native_webrtc_restarts(
                        connection_id,
                        Self::NATIVE_WEBRTC_ADMISSION_FAILURE_BACKOFF_MS,
                        format!("session-admission-rejected: {}", reason),
                    )
                    .await;
                    println!(
                        "[NativeWebRTC] upgrade skipped: session admission rejected connection_id={} remote_node_id={} reason={}",
                        connection_id,
                        remote_node_id,
                        reason,
                    );
                    return Ok(());
                }

                if self.session_registry_active()
                    && matches!(
                        self.session_admission(connection_id),
                        crate::session_token::SessionAdmission::Pending
                    )
                {
                    println!(
                        "[NativeWebRTC] upgrade deferred: session admission pending connection_id={} remote_node_id={}",
                        connection_id,
                        remote_node_id,
                    );
                    return Ok(());
                }

                if crate::native_webrtc_policy::native_webrtc_suppression_blocks_recovery(
                    preferred_negotiation_id.as_deref(),
                ) {
                    if let Some(suppression) =
                        self.active_native_webrtc_suppression(connection_id).await
                    {
                        println!(
                            "[NativeWebRTC] upgrade skipped: restart suppressed connection_id={} remote_node_id={} reason={} remaining_ms={}",
                            connection_id,
                            remote_node_id,
                            suppression.reason,
                            suppression.until_ms.saturating_sub(now_millis_i64()),
                        );
                        return Ok(());
                    }
                } else if let Some(suppression) =
                    self.active_native_webrtc_suppression(connection_id).await
                {
                    println!(
                        "[NativeWebRTC][lifecycle] start continuing through suppression connection_id={} remote_node_id={} reason={} remaining_ms={} preferred_negotiation_id={}",
                        connection_id,
                        remote_node_id,
                        suppression.reason,
                        suppression.until_ms.saturating_sub(now_millis_i64()),
                        preferred_negotiation_id.as_deref().unwrap_or("none"),
                    );
                }

                // A native WebRTC negotiation uses the current Iroh generation as
                // its control path. A record can be recreated by a concurrent
                // incoming handler after the old physical connection has retired,
                // so record presence alone is not sufficient admission to start.
                let Some(record) = self
                    .connection_manager
                    .get_by_connection_id(connection_id)
                    .await
                else {
                    println!(
                        "[NativeWebRTC][lifecycle] start skipped connection_id={} remote_node_id={} reason=connection-record-retired force_restart={} preferred_negotiation_id={}",
                        connection_id,
                        remote_node_id,
                        force_restart,
                        preferred_negotiation_id.as_deref().unwrap_or("none"),
                    );
                    return Ok(());
                };

                if !matches!(
                    record.state,
                    crate::connection_manager::ConnectionState::Pending
                        | crate::connection_manager::ConnectionState::Connecting
                        | crate::connection_manager::ConnectionState::Connected
                ) {
                    println!(
                        "[NativeWebRTC][lifecycle] start skipped connection_id={} remote_node_id={} reason=connection-record-terminal state={:?} force_restart={} preferred_negotiation_id={}",
                        connection_id,
                        remote_node_id,
                        record.state,
                        force_restart,
                        preferred_negotiation_id.as_deref().unwrap_or("none"),
                    );
                    return Ok(());
                }

                let endpoint_text = record
                    .endpoint_id
                    .as_deref()
                    .or(record.node_id.as_deref())
                    .unwrap_or(remote_node_id.as_str());
                let Ok(endpoint_id) = endpoint_text.parse::<iroh::EndpointId>() else {
                    println!(
                        "[NativeWebRTC][lifecycle] start skipped connection_id={} remote_node_id={} reason=invalid-control-endpoint endpoint={}",
                        connection_id, remote_node_id, endpoint_text,
                    );
                    return Ok(());
                };
                let Some(control_connection) = self.get_connection(endpoint_id).await else {
                    println!(
                        "[NativeWebRTC][lifecycle] start skipped connection_id={} remote_node_id={} reason=missing-live-iroh-control-path endpoint_id={}",
                        connection_id, remote_node_id, endpoint_id,
                    );
                    return Ok(());
                };
                let control_stable_id = control_connection.stable_id() as u64;
                if record
                    .transport_stable_id
                    .is_some_and(|expected| expected != control_stable_id)
                {
                    println!(
                        "[NativeWebRTC][lifecycle] start skipped connection_id={} remote_node_id={} reason=stale-iroh-control-generation expected_stable_id={:?} active_stable_id={}",
                        connection_id,
                        remote_node_id,
                        record.transport_stable_id,
                        control_stable_id,
                    );
                    return Ok(());
                }

                {
                    let sessions = self.native_webrtc_sessions.read().await;
                    if let Some(existing) = sessions.get(connection_id) {
                        let state = existing.state();
                        let negotiation_mismatch = preferred_negotiation_id
                            .as_deref()
                            .map(|value| value != existing.negotiation_id().as_str())
                            .unwrap_or(false);
                        // Always replace Failed/Closed sessions.
                        // With force_restart, also replace Connecting AND Connected
                        // sessions — this happens when the browser refreshes: the
                        // same connection_id (derived from the stable node_id pair)
                        // is reused but the old RTCPeerConnection on the browser
                        // side is gone. The remote has fresh ICE credentials and a
                        // new SCTP association, so the old session is a zombie even
                        // if its state still says Connected. Without force_restart
                        // we leave Connecting/Connected sessions alone so a working
                        // negotiation is not disrupted.
                        let is_stale = state == crate::transport::NativeWebRTCState::Failed
                            || state == crate::transport::NativeWebRTCState::Closed
                            || (force_restart
                                && matches!(
                                    state,
                                    crate::transport::NativeWebRTCState::Connecting
                                        | crate::transport::NativeWebRTCState::Connected
                            ))
                            || negotiation_mismatch;
                        if !is_stale {
                            println!(
                                "[NativeWebRTC][lifecycle] start skipped connection_id={} remote_node_id={} reason=existing-session-active state={:?} force_restart={} negotiation_id={} preferred_negotiation_id={}",
                                connection_id,
                                remote_node_id,
                                state,
                                force_restart,
                                existing.negotiation_id(),
                                preferred_negotiation_id.as_deref().unwrap_or("none"),
                            );
                            return Ok(());
                        }
                        println!(
                            "[NativeWebRTC] replacing stale session connection_id={} prior_state={:?} force_restart={} negotiation_mismatch={}",
                            connection_id,
                            state,
                            force_restart,
                            negotiation_mismatch,
                        );
                        let stale = existing.clone();
                        drop(sessions);
                        stale.close();
                        // Allow 500ms for the ICE agent and its UDP sockets
                        // to fully release before we bind a new PeerConnection.
                        // Without this delay, webrtc-rs reuses the same UDP port
                        // and stale STUN binding requests from the old session
                        // trigger ICE username mismatch errors on the new agent.
                        tokio::time::sleep(std::time::Duration::from_millis(500)).await;
                        let _ = self
                            .remove_native_webrtc_session_if_current(
                                connection_id,
                                &stale,
                                "replace-stale-session",
                            )
                            .await;
                    }
                }

                if self
                    .dedupe_native_webrtc_session_for_remote(
                        connection_id,
                        &remote_node_id,
                        force_restart,
                    )
                    .await
                {
                    return Ok(());
                }

                if preferred_negotiation_id.is_none() {
                    let primary_iroh_path = self.iroh_path_kind(&remote_node_id).await;
                    if matches!(
                        primary_iroh_path,
                        crate::client::IrohPathKind::DirectQuic
                            | crate::client::IrohPathKind::DirectLan
                            | crate::client::IrohPathKind::Ble
                    ) {
                        let primary_label = primary_iroh_path.transport_label();
                        self.suppress_native_webrtc_restarts(
                            connection_id,
                            Self::NATIVE_WEBRTC_DIRECT_QUIC_SUPPRESSION_MS,
                            primary_label,
                        )
                        .await;
                        let _ = self
                            .close_connecting_native_webrtc_session(connection_id, primary_label)
                            .await;
                        println!(
                            "[NativeWebRTC] upgrade skipped: optimal iroh path is primary connection_id={} remote_node_id={} path={}",
                            connection_id,
                            remote_node_id,
                            primary_label,
                        );
                        return Ok(());
                    }
                }

                if preferred_negotiation_id.is_some() {
                    let removed = self
                        .native_webrtc_attempt_counts
                        .write()
                        .await
                        .remove(connection_id);
                    if removed.is_some() {
                        println!(
                            "[NativeWebRTC][lifecycle] reset attempt budget for preferred negotiation connection_id={} preferred_negotiation_id={}",
                            connection_id,
                            preferred_negotiation_id.as_deref().unwrap_or("none"),
                        );
                    }
                }

                // Check upgrade attempt count — do not retry indefinitely.
                let attempt_reservation = {
                    let mut counts = self.native_webrtc_attempt_counts.write().await;
                    let attempt = counts.entry(connection_id.to_string()).or_insert(0);
                    let reservation = Self::reserve_native_webrtc_attempt_slot(*attempt);
                    match reservation {
                        NativeWebRTCAttemptReservation::Start(next) => {
                            *attempt = next;
                        }
                        NativeWebRTCAttemptReservation::JustExhausted(exhausted) => {
                            *attempt = exhausted;
                        }
                        NativeWebRTCAttemptReservation::AlreadyExhausted => {}
                    }
                    reservation
                };
                let current_attempt = match attempt_reservation {
                    NativeWebRTCAttemptReservation::Start(attempt) => attempt,
                    NativeWebRTCAttemptReservation::JustExhausted(attempt) => {
                        self.suppress_native_webrtc_restarts(
                            connection_id,
                            Self::NATIVE_WEBRTC_CONNECT_TIMEOUT_BACKOFF_MS,
                            "upgrade-attempts-exhausted",
                        )
                        .await;
                        println!(
                            "[NativeWebRTC] upgrade exhausted: max attempts reached connection_id={} remote_node_id={} attempts={}",
                            connection_id,
                            remote_node_id,
                            attempt,
                        );
                        return Ok(());
                    }
                    NativeWebRTCAttemptReservation::AlreadyExhausted => {
                        println!(
                            "[NativeWebRTC][lifecycle] start skipped connection_id={} remote_node_id={} reason=attempt-budget-already-exhausted force_restart={} preferred_negotiation_id={}",
                            connection_id,
                            remote_node_id,
                            force_restart,
                            preferred_negotiation_id.as_deref().unwrap_or("none"),
                        );
                        return Ok(());
                    }
                };
                println!(
                    "[NativeWebRTC] starting upgrade attempt={}/{} connection_id={} local_node_id={} remote_node_id={} force_restart={} preferred_negotiation_id={} role_override={:?}",
                    current_attempt,
                    Self::NATIVE_WEBRTC_MAX_UPGRADE_ATTEMPTS,
                    connection_id,
                    local_node_id,
                    remote_node_id,
                    force_restart,
                    preferred_negotiation_id.as_deref().unwrap_or("none"),
                    role_override,
                );

                if self
                    .should_proactively_notify_browser_peer(connection_id, &remote_node_id)
                    .await
                {
                    if let Err(error) = self
                        .send_typescript_capability_update(
                            connection_id,
                            "native-upgrade-start",
                            self.connection_key_agreement_public_key(connection_id),
                        )
                        .await
                    {
                        println!(
                            "[NativeWebRTC] proactive capability-update send failed connection_id={} remote_node_id={} reason=native-upgrade-start error={}",
                            connection_id,
                            remote_node_id,
                            error,
                        );
                    }
                }

                let negotiation_id = preferred_negotiation_id
                    .clone()
                    .unwrap_or_else(|| Self::create_native_webrtc_negotiation_id(connection_id));
                let source_session = Arc::new(tokio::sync::OnceCell::<
                    std::sync::Weak<crate::transport::NativeWebRTCDataChannel>,
                >::new());

                let signal_sender: crate::transport::NativeWebRTCSignalSender = {
                    let client = self.clone();
                    let connection_id = connection_id.to_string();
                    let remote_node_id = remote_node_id.clone();
                    let source_session = source_session.clone();
                    Arc::new(move |frame: serde_json::Value| {
                        let client = client.clone();
                        let connection_id = connection_id.clone();
                        let remote_node_id = remote_node_id.clone();
                        let source_session = source_session.clone();
                        Box::pin(async move {
                        let signal_type = frame
                            .get("content")
                            .and_then(|value| value.get("type"))
                            .and_then(|value| value.as_str())
                            .unwrap_or("unknown");
                        let Ok(serialized) = serde_json::to_vec(&frame) else {
                            println!(
                                "[NativeWebRTC] signal drop: serialization failed connection_id={} signal_type={}",
                                connection_id,
                                signal_type,
                            );
                            return;
                        };
                        let frame_len = (serialized.len() as u32).to_be_bytes();
                        let mut frame_buf = Vec::with_capacity(4 + serialized.len());
                        frame_buf.extend_from_slice(&frame_len);
                        frame_buf.extend_from_slice(&serialized);
                        let started = std::time::Instant::now();
                        let retry_delay = std::time::Duration::from_millis(
                            Self::NATIVE_WEBRTC_SIGNAL_SEND_RETRY_DELAY_MS,
                        );
                        let send_timeout = std::time::Duration::from_millis(
                            Self::NATIVE_WEBRTC_SIGNAL_SEND_TIMEOUT_MS,
                        );
                        let mut attempts = 0usize;
                        let mut endpoint_for_logs: Option<String> = None;

                        let last_failure = loop {
                            attempts += 1;

                            if client
                                .connection_manager
                                .get_by_connection_id(&connection_id)
                                .await
                                .is_none()
                            {
                                println!(
                                    "[NativeWebRTC] signal send aborted: connection retired connection_id={} signal_type={} attempts={}",
                                    connection_id,
                                    signal_type,
                                    attempts,
                                );
                                return;
                            }

                            // Abort immediately if the session token was rejected — the
                            // iroh connection is being closed and retrying is pointless.
                            if client.session_registry_active() {
                                if let crate::session_token::SessionAdmission::Rejected { reason } =
                                    client.session_admission(&connection_id)
                                {
                                    println!(
                                        "[NativeWebRTC] signal send aborted: session admission rejected connection_id={} signal_type={} reason={} attempts={}",
                                        connection_id,
                                        signal_type,
                                        reason,
                                        attempts,
                                    );
                                    return;
                                }
                            }

                            let Some(session) = source_session
                                .get()
                                .and_then(std::sync::Weak::upgrade)
                            else {
                                println!(
                                    "[NativeWebRTC] signal send aborted: source session missing connection_id={} signal_type={} attempts={}",
                                    connection_id,
                                    signal_type,
                                    attempts,
                                );
                                return;
                            };

                            if !client
                                .current_native_webrtc_route_matches(&connection_id, &session)
                                .await
                            {
                                println!(
                                    "[NativeWebRTC] signal send aborted: stale route generation connection_id={} signal_type={} attempts={}",
                                    connection_id, signal_type, attempts,
                                );
                                return;
                            }

                            match session.state() {
                                crate::transport::NativeWebRTCState::Failed
                                | crate::transport::NativeWebRTCState::Closed => {
                                    println!(
                                        "[NativeWebRTC] signal send aborted: session ended connection_id={} signal_type={} state={:?} attempts={}",
                                        connection_id,
                                        signal_type,
                                        session.state(),
                                        attempts,
                                    );
                                    return;
                                }
                                _ => {}
                            }

                            let endpoint_str = client
                                .connection_manager
                                .get_by_connection_id(&connection_id)
                                .await
                                .and_then(|record| {
                                    record
                                        .endpoint_id
                                        .as_ref()
                                        .or(record.node_id.as_ref())
                                        .map(|value| value.trim().to_string())
                                })
                                .filter(|value| !value.is_empty())
                                .or_else(|| {
                                    Some(remote_node_id.trim().to_string())
                                        .filter(|value| !value.is_empty())
                                });

                            let Some(endpoint_str) = endpoint_str else {
                                if started.elapsed() >= send_timeout {
                                    break "missing endpoint mapping";
                                }
                                tokio::time::sleep(retry_delay).await;
                                continue;
                            };

                            endpoint_for_logs = Some(endpoint_str.clone());

                            let Ok(endpoint_id) = endpoint_str.parse::<iroh::EndpointId>() else {
                                println!(
                                    "[NativeWebRTC] signal drop: invalid endpoint_id connection_id={} endpoint={} signal_type={}",
                                    connection_id,
                                    endpoint_str,
                                    signal_type,
                                );
                                return;
                            };

                            match client
                                .send_native_main_frame(
                                    &connection_id,
                                    endpoint_id,
                                    &signal_type,
                                    &frame_buf,
                                )
                                .await
                            {
                                Ok(_) => {
                                    if !client
                                        .current_native_webrtc_route_matches(
                                            &connection_id,
                                            &session,
                                        )
                                        .await
                                    {
                                        println!(
                                            "[NativeWebRTC] signal send completion ignored: route replaced connection_id={} signal_type={} attempts={}",
                                            connection_id, signal_type, attempts,
                                        );
                                        return;
                                    }
                                    if attempts > 1 {
                                        println!(
                                            "[NativeWebRTC] signal send delayed until signaling path recovered connection_id={} endpoint_id={} signal_type={} attempts={} elapsed_ms={}",
                                            connection_id,
                                            endpoint_id,
                                            signal_type,
                                            attempts,
                                            started.elapsed().as_millis(),
                                        );
                                    }
                                    println!(
                                        "[NativeWebRTC] signal sent connection_id={} endpoint_id={} signal_type={}",
                                        connection_id, endpoint_id, signal_type,
                                    );
                                    return;
                                }
                                Err(err) => {
                                    if !client
                                        .current_native_webrtc_route_matches(
                                            &connection_id,
                                            &session,
                                        )
                                        .await
                                    {
                                        println!(
                                            "[NativeWebRTC] signal send failure ignored: route replaced connection_id={} signal_type={} attempts={}",
                                            connection_id, signal_type, attempts,
                                        );
                                        return;
                                    }
                                    if native_signal_send_failure_is_terminal_for_attempt(&err) {
                                        let negotiation_id = session.negotiation_id();
                                        let retired_route = client
                                            .native_optional_route_generation_for_instance(
                                                &connection_id,
                                                NativeOptionalRouteKind::WebRtc,
                                                Arc::as_ptr(&session) as usize,
                                            )
                                            .await;
                                        println!(
                                            "[NativeWebRTC] signal send aborted: control path unavailable connection_id={} endpoint_id={} signal_type={} attempts={} error={}",
                                            connection_id,
                                            endpoint_id,
                                            signal_type,
                                            attempts,
                                            err,
                                        );
                                        session.close();
                                        client
                                            .clear_native_webrtc_attempt_if_current(
                                                &connection_id,
                                                &negotiation_id,
                                            )
                                            .await;
                                        if client
                                            .remove_native_webrtc_session_if_current(
                                                &connection_id,
                                                &session,
                                                "signal-control-path-unavailable",
                                            )
                                            .await
                                        {
                                            client
                                                .finalize_deferred_managed_retirement_if_needed(
                                                    &connection_id,
                                                )
                                                .await;
                                        }
                                        if let Some(retired_route) = retired_route {
                                            client.schedule_native_webrtc_retry(
                                                &connection_id,
                                                &remote_node_id,
                                                retired_route,
                                                Self::NATIVE_WEBRTC_CONNECT_TIMEOUT_RETRY_DELAY_MS,
                                                "signal-control-path-unavailable",
                                            );
                                        }
                                        return;
                                    }
                                    if started.elapsed() >= send_timeout {
                                        println!(
                                            "[SIGNAL-STREAM] send failed (terminal) connection_id={} signal_type={} error={}",
                                            connection_id, signal_type, err,
                                        );
                                        break "signal-stream write failed";
                                    }
                                    println!(
                                        "[SIGNAL-STREAM] send failed (retry) connection_id={} signal_type={} attempts={} error={}",
                                        connection_id, signal_type, attempts, err,
                                    );
                                    tokio::time::sleep(retry_delay).await;
                                    continue;
                                }
                            }
                        };

                        println!(
                            "[NativeWebRTC] signal send timed out waiting for signaling path connection_id={} endpoint={} signal_type={} attempts={} timeout_ms={} last_failure={}",
                            connection_id,
                            endpoint_for_logs.unwrap_or_else(|| "unknown".to_string()),
                            signal_type,
                            attempts,
                            Self::NATIVE_WEBRTC_SIGNAL_SEND_TIMEOUT_MS,
                            last_failure,
                        );
                        })
                    })
                };

                // Record that this attempt is in flight before creating the
                // session. The recovery funnel consults this to reject
                // concurrent triggers for the lifetime of this attempt.
                self.record_native_webrtc_attempt_started(
                    connection_id,
                    &negotiation_id,
                    preferred_negotiation_id.is_some(),
                )
                .await;
                let session = Arc::new(
                    crate::transport::NativeWebRTCDataChannel::new_with_role(
                        &local_node_id,
                        &remote_node_id,
                        &webrtc_config,
                        signal_sender,
                        &negotiation_id,
                        role_override.unwrap_or(crate::transport::NativeWebRTCRole::Auto),
                    )
                    .await?,
                );
                source_session
                    .set(Arc::downgrade(&session))
                    .map_err(|_| anyhow::anyhow!("native WebRTC source session already installed"))?;
                {
                    let client_ref = self.clone();
                    let connection_id_ref = connection_id.to_string();
                    let remote_node_id_ref = remote_node_id.clone();
                    let source_session = Arc::downgrade(&session);
                    session
                        .set_message_handler(Arc::new(move |bytes| {
                            let client_ref = client_ref.clone();
                            let connection_id_ref = connection_id_ref.clone();
                            let remote_node_id_ref = remote_node_id_ref.clone();
                            let source_session = source_session.clone();
                            Box::pin(async move {
                                let Some(source_session) = source_session.upgrade() else {
                                    return;
                                };
                                if let Err(error) = client_ref
                                    .handle_current_native_webrtc_peer_data(
                                        &connection_id_ref,
                                        Some(&remote_node_id_ref),
                                        &source_session,
                                        bytes,
                                    )
                                    .await
                                {
                                    println!(
                                        "[NativeWebRTC] peer-data receive failed connection_id={} remote_node_id={} error={}",
                                        connection_id_ref, remote_node_id_ref, error,
                                    );
                                }
                            })
                        }))
                        .await;
                }
                // Arm the native heartbeat before start() so the session can
                // respond to browser pings as soon as data-channel traffic lands.
                let (heartbeat_health_tx, _heartbeat_health_rx) =
                    tokio::sync::mpsc::channel::<crate::heartbeat::HealthTransition>(8);
                session
                    .with_heartbeat(
                        "webrtc".to_string(),
                        Self::native_webrtc_heartbeat_config(),
                        heartbeat_health_tx,
                    )
                    .await;

                // Register the session before start() so early SDP/candidate sends
                // from the native transport can resolve the current session.
                self.native_webrtc_sessions
                    .write()
                    .await
                    .insert(connection_id.to_string(), session.clone());
                if self
                    .install_native_optional_route_generation(
                        connection_id,
                        NativeOptionalRouteKind::WebRtc,
                        Arc::as_ptr(&session) as usize,
                    )
                    .await
                    .is_none()
                {
                    self.native_webrtc_sessions.write().await.remove(connection_id);
                    self.clear_native_webrtc_attempt_if_current(connection_id, &negotiation_id)
                        .await;
                    println!(
                        "[NativeWebRTC][lifecycle] start skipped connection_id={} remote_node_id={} reason=missing-current-route-generation",
                        connection_id, remote_node_id,
                    );
                    return Ok(());
                }

                if !self
                    .current_native_webrtc_route_identity_matches(connection_id, &session)
                    .await
                {
                    let _ = self
                        .remove_native_webrtc_session_if_current(
                            connection_id,
                            &session,
                            "stale-before-start",
                        )
                        .await;
                    return Ok(());
                }

                if let Err(error) = session.start().await {
                    self.clear_native_webrtc_attempt_if_current(connection_id, &negotiation_id)
                        .await;
                    if self
                        .remove_native_webrtc_session_if_current(
                            connection_id,
                            &session,
                            "start-failed",
                        )
                        .await
                    {
                        self.finalize_deferred_managed_retirement_if_needed(connection_id)
                            .await;
                    }
                    return Err(error);
                }

                if !self
                    .current_native_webrtc_route_matches(connection_id, &session)
                    .await
                {
                    let _ = self
                        .remove_native_webrtc_session_if_current(
                            connection_id,
                            &session,
                            "stale-after-start",
                        )
                        .await;
                    return Ok(());
                }

                println!(
                    "[NativeWebRTC] upgrade started connection_id={} state={:?}",
                    connection_id,
                    session.state(),
                );

                // Spawn a lifecycle watcher for this WebRTC session.
                //
                // Phase 1 — Connecting: poll up to NATIVE_WEBRTC_CONNECT_TIMEOUT_MS
                //   for the datachannel to
                //   open.  If it hasn't by then, close the session and let the
                //   path-watcher's next relay-detected tick (or the auto-connect
                //   health-check) restart it with force_restart=true.
                //
                // Phase 2 — Connected: keep re-reporting the transport status
                //   every 5 s while the session stays Connected.  This ensures
                //   the frontend badge stays current after iroh reconnects without
                //   requiring a full re-negotiation.
                {
                    let session_ref = session.clone();
                    let connection_id_ref = connection_id.to_string();
                    let remote_node_id_ref = remote_node_id.clone();
                    let negotiation_id_ref = negotiation_id.clone();
                    let client_ref = self.clone();
                    tokio::spawn(async move {
                        // Phase 1: wait up to NATIVE_WEBRTC_CONNECT_TIMEOUT_MS for Connected.
                        // Also early-abort if no remote SDP has arrived after
                        // NATIVE_WEBRTC_NO_SIGNAL_ABORT_MS — that indicates
                        // the peer never received our offer (or never sent
                        // theirs) and waiting the full 45s just produces
                        // log spam while other triggers are rejected by the
                        // in-flight gate.
                        let no_signal_deadline_ticks =
                            (Self::NATIVE_WEBRTC_NO_SIGNAL_ABORT_MS / 250) as usize;
                        let mut connected = false;
                        let mut aborted_no_signal = false;
                        for tick in 0..(Self::NATIVE_WEBRTC_CONNECT_TIMEOUT_MS / 250) as usize {
                            if !client_ref
                                .current_native_webrtc_route_matches(
                                    &connection_id_ref,
                                    &session_ref,
                                )
                                .await
                            {
                                client_ref
                                    .clear_native_webrtc_attempt_if_current(
                                        &connection_id_ref,
                                        &negotiation_id_ref,
                                    )
                                    .await;
                                return;
                            }
                            match session_ref.state() {
                                crate::transport::NativeWebRTCState::Connected => {
                                    connected = true;
                                    break;
                                }
                                crate::transport::NativeWebRTCState::Failed
                                | crate::transport::NativeWebRTCState::Closed => {
                                    client_ref
                                        .clear_native_webrtc_attempt_if_current(
                                            &connection_id_ref,
                                            &negotiation_id_ref,
                                        )
                                        .await;
                                    return;
                                }
                                _ => {}
                            }
                            let missing_remote_description = tick == no_signal_deadline_ticks
                                && !session_ref.has_remote_description().await;
                            if !client_ref
                                .current_native_webrtc_route_matches(
                                    &connection_id_ref,
                                    &session_ref,
                                )
                                .await
                            {
                                return;
                            }
                            if missing_remote_description {
                                println!(
                                    "[NativeWebRTC] upgrade aborted: no remote SDP after {}ms connection_id={} negotiation_id={}",
                                    Self::NATIVE_WEBRTC_NO_SIGNAL_ABORT_MS,
                                    connection_id_ref,
                                    negotiation_id_ref,
                                );
                                aborted_no_signal = true;
                                break;
                            }
                            tokio::time::sleep(tokio::time::Duration::from_millis(250)).await;
                        }

                        if aborted_no_signal {
                            let retired_route = client_ref
                                .native_optional_route_generation_for_instance(
                                    &connection_id_ref,
                                    NativeOptionalRouteKind::WebRtc,
                                    Arc::as_ptr(&session_ref) as usize,
                                )
                                .await;
                            session_ref.close();
                            client_ref
                                .clear_native_webrtc_attempt_if_current(
                                    &connection_id_ref,
                                    &negotiation_id_ref,
                                )
                                .await;
                            if client_ref
                                .remove_native_webrtc_session_if_current(
                                    &connection_id_ref,
                                    &session_ref,
                                    "upgrade-no-signal",
                                )
                                .await
                            {
                                client_ref
                                    .finalize_deferred_managed_retirement_if_needed(
                                        &connection_id_ref,
                                    )
                                    .await;
                            }
                            if let Some(retired_route) = retired_route {
                                client_ref.schedule_native_webrtc_retry(
                                    &connection_id_ref,
                                    &remote_node_id_ref,
                                    retired_route,
                                    Self::NATIVE_WEBRTC_CONNECT_TIMEOUT_RETRY_DELAY_MS,
                                    "upgrade-no-signal",
                                );
                            }
                            return;
                        }

                        if !connected {
                            let retired_route = client_ref
                                .native_optional_route_generation_for_instance(
                                    &connection_id_ref,
                                    NativeOptionalRouteKind::WebRtc,
                                    Arc::as_ptr(&session_ref) as usize,
                                )
                                .await;
                            // Clear the in-flight record so the scheduled
                            // retry (or next trigger) can start a fresh
                            // attempt once the suppression window expires.
                            client_ref
                                .clear_native_webrtc_attempt_if_current(
                                    &connection_id_ref,
                                    &negotiation_id_ref,
                                )
                                .await;
                            client_ref
                                .suppress_native_webrtc_restarts(
                                    &connection_id_ref,
                                    Self::NATIVE_WEBRTC_CONNECT_TIMEOUT_BACKOFF_MS,
                                    "upgrade-timeout",
                                )
                                .await;
                            // Timed out in Connecting — close the session so it
                            // is treated as stale (Closed) on the next trigger.
                            // The path-watcher's next relay tick will restart it
                            // with force_restart=false (Closed is always replaced).
                            println!(
                                "[NativeWebRTC] upgrade timeout: closing stuck Connecting session connection_id={}",
                                connection_id_ref,
                            );
                            session_ref.close();
                            if client_ref
                                .remove_native_webrtc_session_if_current(
                                    &connection_id_ref,
                                    &session_ref,
                                    "upgrade-timeout",
                                )
                                .await
                            {
                                client_ref
                                    .finalize_deferred_managed_retirement_if_needed(
                                        &connection_id_ref,
                                    )
                                    .await;
                            }
                            if let Some(retired_route) = retired_route {
                                client_ref.schedule_native_webrtc_retry(
                                    &connection_id_ref,
                                    &remote_node_id_ref,
                                    retired_route,
                                    Self::NATIVE_WEBRTC_CONNECT_TIMEOUT_RETRY_DELAY_MS,
                                    "upgrade-timeout",
                                );
                            }
                            return;
                        }

                        let _ = client_ref
                            .clear_native_webrtc_suppression(&connection_id_ref, None)
                            .await;
                        // Reset attempt counter on successful connection — future
                        // renegotiations (e.g. browser refresh) get fresh attempts.
                        client_ref
                            .native_webrtc_attempt_counts
                            .write()
                            .await
                            .remove(&connection_id_ref);
                        // Attempt succeeded — drop the in-flight record so
                        // subsequent triggers (e.g. renegotiate after browser
                        // refresh) can start a new attempt.
                        client_ref
                            .clear_native_webrtc_attempt_if_current(
                                &connection_id_ref,
                                &negotiation_id_ref,
                            )
                            .await;

                        client_ref
                            .report_native_webrtc_candidate_transport_for_session(
                                &connection_id_ref,
                                Some(remote_node_id_ref.as_str()),
                                &session_ref,
                            )
                            .await;

                        // Phase 2: report Connected and keep refreshing while
                        // the session stays alive so the badge survives iroh
                        // reconnects. Poll at 5s in foreground, 30s in background
                        // (no UI to update when backgrounded, and mobile OS kills
                        // apps that do unnecessary background wakeups).
                        loop {
                            if session_ref.state()
                                != crate::transport::NativeWebRTCState::Connected
                                || !client_ref
                                    .current_native_webrtc_route_matches(
                                        &connection_id_ref,
                                        &session_ref,
                                    )
                                    .await
                            {
                                break;
                            }
                            if !client_ref.is_app_backgrounded() {
                                client_ref
                                    .report_native_webrtc_candidate_transport_for_session(
                                        &connection_id_ref,
                                        None,
                                        &session_ref,
                                    )
                                    .await;
                            }
                            let interval = if client_ref.is_app_backgrounded() {
                                tokio::time::Duration::from_secs(30)
                            } else {
                                tokio::time::Duration::from_secs(5)
                            };
                            tokio::time::sleep(interval).await;
                        }
                        // Session is no longer Connected — remove it so the Arc
                        // and underlying WebRTC resources can be freed.
                        if client_ref
                            .remove_native_webrtc_session_if_current(
                                &connection_id_ref,
                                &session_ref,
                                "session-ended",
                            )
                            .await
                        {
                            client_ref
                                .finalize_deferred_managed_retirement_if_needed(&connection_id_ref)
                                .await;
                        }
                    });
                }

                Ok(())
            }
            .await;

            self.release_native_webrtc_start_gate(connection_id, start_gate)
                .await;
            return result;
        }

        #[cfg(any(target_arch = "wasm32", not(feature = "transport-webrtc")))]
        {
            let _ = (local_node_id, remote_node_id, webrtc_config);
            Ok(())
        }
    }

    #[cfg(all(not(target_arch = "wasm32"), feature = "transport-moq"))]
    pub(crate) async fn maybe_start_native_moq_upgrade(
        &self,
        connection_id: &str,
        remote_node_id: Option<&str>,
    ) -> anyhow::Result<bool> {
        self.maybe_start_native_moq_upgrade_with_options(connection_id, remote_node_id, false)
            .await
    }

    #[cfg(any(target_arch = "wasm32", not(feature = "transport-moq")))]
    pub(crate) async fn maybe_start_native_moq_upgrade(
        &self,
        connection_id: &str,
        remote_node_id: Option<&str>,
    ) -> anyhow::Result<bool> {
        let _ = (connection_id, remote_node_id);
        Ok(false)
    }

    #[cfg(all(not(target_arch = "wasm32"), feature = "transport-moq"))]
    async fn maybe_start_native_moq_upgrade_with_options(
        &self,
        connection_id: &str,
        remote_node_id: Option<&str>,
        force_restart_when_not_ready: bool,
    ) -> anyhow::Result<bool> {
        if connection_id.trim().is_empty() {
            return Ok(false);
        }

        let remote_node_id = match remote_node_id
            .map(str::trim)
            .filter(|value| !value.is_empty())
        {
            Some(value) => value,
            None => return Ok(false),
        };

        let moq_config = {
            let guard = self.transport_config.read().await;
            match guard.moq.clone() {
                Some(value) => value,
                None => return Ok(false),
            }
        };

        let local_node_id = match self.current_node_id().await {
            Some(value) => value,
            None => return Ok(false),
        };

        #[cfg(all(not(target_arch = "wasm32"), feature = "transport-moq"))]
        {
            enum ExistingMoQAction {
                Close(Arc<crate::transport::NativeMoQSession>),
                Reannounce(Arc<crate::transport::NativeMoQSession>),
                Reuse,
            }

            let session_to_close = {
                let existing = {
                    let sessions = self.native_moq_sessions.read().await;
                    sessions.get(connection_id).cloned()
                };
                if let Some(existing) = existing {
                    let state = existing.state();
                    if !self
                        .current_native_moq_route_matches(connection_id, &existing)
                        .await
                    {
                        Some(ExistingMoQAction::Close(existing.clone()))
                    } else {
                        let subscribed = existing.is_peer_data_subscribed().await;
                        if !self
                            .current_native_moq_route_matches(connection_id, &existing)
                            .await
                        {
                            Some(ExistingMoQAction::Close(existing.clone()))
                        } else {
                            let data_ready = native_moq_is_data_ready(state, subscribed);
                            let force_replace = force_restart_when_not_ready && !data_ready;
                            if force_replace || native_moq_should_close_before_restart(state) {
                                Some(ExistingMoQAction::Close(existing.clone()))
                            } else if state == crate::transport::NativeMoQState::Connected
                                && !data_ready
                            {
                                Some(ExistingMoQAction::Reannounce(existing.clone()))
                            } else if native_moq_should_reuse_existing_setup(state) {
                                Some(ExistingMoQAction::Reuse)
                            } else {
                                None
                            }
                        }
                    }
                } else {
                    None
                }
            };
            if let Some(action) = session_to_close {
                match action {
                    ExistingMoQAction::Close(existing) => {
                        existing.close_gracefully().await;
                        let mut sessions = self.native_moq_sessions.write().await;
                        if sessions
                            .get(connection_id)
                            .map(|current| Arc::ptr_eq(current, &existing))
                            .unwrap_or(false)
                        {
                            sessions.remove(connection_id);
                        }
                        drop(sessions);
                        self.clear_native_optional_route_generation_if_current(
                            connection_id,
                            NativeOptionalRouteKind::Moq,
                            Arc::as_ptr(&existing) as usize,
                        )
                        .await;
                    }
                    ExistingMoQAction::Reannounce(existing) => {
                        if !self
                            .current_native_moq_route_matches(connection_id, &existing)
                            .await
                        {
                            return Ok(false);
                        }
                        let parallel_transport = self
                            .resolve_parallel_iroh_transport_label(connection_id, None)
                            .await;
                        if !self
                            .current_native_moq_route_matches(connection_id, &existing)
                            .await
                        {
                            return Ok(false);
                        }
                        let report = setup_status_report(parallel_transport.as_str());
                        let _ = self
                            .report_transport_status(
                                connection_id,
                                report.active_transport,
                                report.parallel_transport,
                            )
                            .await;
                        if !self
                            .current_native_moq_route_matches(connection_id, &existing)
                            .await
                        {
                            return Ok(false);
                        }
                        if let Err(error) = self
                            .send_typescript_capability_update(
                                connection_id,
                                "native-moq-ready",
                                self.connection_key_agreement_public_key(connection_id),
                            )
                            .await
                        {
                            println!(
                                "[NativeMoQ] capability-update resend failed connection_id={} remote_node_id={} reason=native-moq-ready error={}",
                                connection_id, remote_node_id, error,
                            );
                        }
                        return Ok(true);
                    }
                    ExistingMoQAction::Reuse => return Ok(true),
                }
            }

            let session = Arc::new(
                crate::transport::NativeMoQSession::new(
                    &local_node_id,
                    remote_node_id,
                    &moq_config,
                )
                .await?,
            );
            {
                let client_ref = self.clone();
                let connection_id_ref = connection_id.to_string();
                let remote_node_id_ref = remote_node_id.to_string();
                let source_session = Arc::downgrade(&session);
                session
                    .set_message_handler(Arc::new(move |bytes| {
                        let client_ref = client_ref.clone();
                        let connection_id_ref = connection_id_ref.clone();
                        let remote_node_id_ref = remote_node_id_ref.clone();
                        let source_session = source_session.clone();
                        Box::pin(async move {
                            let Some(source_session) = source_session.upgrade() else {
                                return;
                            };
                            if let Err(error) = client_ref
                                .handle_current_native_moq_peer_data(
                                    &connection_id_ref,
                                    Some(&remote_node_id_ref),
                                    &source_session,
                                    bytes,
                                )
                                .await
                            {
                                println!(
                                    "[NativeMoQ] peer-data receive failed connection_id={} remote_node_id={} error={}",
                                    connection_id_ref, remote_node_id_ref, error,
                                );
                            }
                        })
                    }))
                    .await;
            }

            // Register before start() so Connecting/Failed states are observable
            // through the same harness/state surface that WebRTC exposes.
            self.native_moq_sessions
                .write()
                .await
                .insert(connection_id.to_string(), session.clone());
            if self
                .install_native_optional_route_generation(
                    connection_id,
                    NativeOptionalRouteKind::Moq,
                    Arc::as_ptr(&session) as usize,
                )
                .await
                .is_none()
            {
                self.native_moq_sessions.write().await.remove(connection_id);
                println!(
                    "[NativeMoQ][lifecycle] setup skipped connection_id={} remote_node_id={} reason=missing-current-route-generation",
                    connection_id, remote_node_id,
                );
                return Ok(false);
            }

            if !self
                .current_native_moq_route_identity_matches(connection_id, &session)
                .await
            {
                session.close_gracefully().await;
                return Ok(false);
            }

            if let Err(error) = session.start().await {
                println!(
                    "[NativeMoQ] setup failed connection_id={} local_node_id={} remote_node_id={} error={:#}",
                    connection_id, local_node_id, remote_node_id, error,
                );
                let mut sessions = self.native_moq_sessions.write().await;
                if sessions
                    .get(connection_id)
                    .is_some_and(|current| Arc::ptr_eq(current, &session))
                {
                    sessions.remove(connection_id);
                }
                drop(sessions);
                self.clear_native_optional_route_generation_if_current(
                    connection_id,
                    NativeOptionalRouteKind::Moq,
                    Arc::as_ptr(&session) as usize,
                )
                .await;
                // MoQ is optional. Setup failure leaves the settled Iroh route
                // untouched and removes only this failed candidate.
                return Ok(true);
            }

            if !self
                .current_native_moq_route_matches(connection_id, &session)
                .await
            {
                session.close_gracefully().await;
                return Ok(false);
            }

            let parallel_transport = self
                .resolve_parallel_iroh_transport_label(connection_id, None)
                .await;
            if !self
                .current_native_moq_route_matches(connection_id, &session)
                .await
            {
                return Ok(false);
            }
            let report = setup_status_report(parallel_transport.as_str());
            let _ = self
                .report_transport_status(
                    connection_id,
                    report.active_transport,
                    report.parallel_transport,
                )
                .await;
            if !self
                .current_native_moq_route_matches(connection_id, &session)
                .await
            {
                return Ok(false);
            }
            if let Err(error) = self
                .send_typescript_capability_update(
                    connection_id,
                    "native-moq-ready",
                    self.connection_key_agreement_public_key(connection_id),
                )
                .await
            {
                println!(
                    "[NativeMoQ] capability-update send failed connection_id={} remote_node_id={} reason=native-moq-ready error={}",
                    connection_id, remote_node_id, error,
                );
            }
            if !self
                .current_native_moq_route_matches(connection_id, &session)
                .await
            {
                return Ok(false);
            }
            println!(
                "[NativeMoQ] setup connected connection_id={} local_node_id={} remote_node_id={} active={} parallel=moq",
                connection_id, local_node_id, remote_node_id, parallel_transport,
            );
            return Ok(true);
        }

        #[cfg(any(target_arch = "wasm32", not(feature = "transport-moq")))]
        {
            let _ = (local_node_id, remote_node_id, moq_config);
            Ok(false)
        }
    }

    async fn forward_native_webrtc_signal(
        &self,
        connection_id: &str,
        remote_node_id: Option<&str>,
        signal: crate::transport::WebRTCSignalMessage,
    ) -> anyhow::Result<()> {
        #[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
        {
            let mut candidate_connection_ids = vec![connection_id.to_string()];
            if let Some(remote_node_id) = remote_node_id
                .map(str::trim)
                .filter(|value| !value.is_empty())
            {
                candidate_connection_ids
                    .extend(self.resolve_transport_connection_ids(remote_node_id).await);
            }

            if candidate_connection_ids.len() > 1 {
                let mut seen =
                    std::collections::HashSet::with_capacity(candidate_connection_ids.len());
                candidate_connection_ids.retain(|candidate| seen.insert(candidate.clone()));
            }

            let resolved_session = {
                let sessions = self.native_webrtc_sessions.read().await;
                let mut resolved: Option<(String, Arc<crate::transport::NativeWebRTCDataChannel>)> =
                    None;
                for candidate_connection_id in candidate_connection_ids.iter() {
                    if let Some(session) = sessions.get(candidate_connection_id) {
                        resolved = Some((candidate_connection_id.clone(), session.clone()));
                        break;
                    }
                }
                resolved
            };

            if let Some((resolved_connection_id, session)) = resolved_session {
                if resolved_connection_id != connection_id {
                    println!(
                        "[NativeWebRTC] signal routing alias resolved incoming_connection_id={} resolved_connection_id={}",
                        connection_id,
                        resolved_connection_id,
                    );
                }

                let initial_state = session.state();
                if !self
                    .current_native_webrtc_route_matches(&resolved_connection_id, &session)
                    .await
                {
                    anyhow::bail!(
                        "native WebRTC signal rejected for retired session connection_id={} state={:?}",
                        resolved_connection_id,
                        initial_state,
                    );
                }

                let was_connected =
                    session.state() == crate::transport::NativeWebRTCState::Connected;
                let signal_type = signal.signal_type.clone();
                let result = session.handle_signal(signal).await;
                if !self
                    .current_native_webrtc_route_matches(&resolved_connection_id, &session)
                    .await
                {
                    println!(
                        "[NativeWebRTC] signal completion ignored: route replaced connection_id={} signal_type={} negotiation_id={}",
                        resolved_connection_id,
                        signal_type,
                        session.negotiation_id(),
                    );
                    return Ok(());
                }
                result?;

                println!(
                    "[NativeWebRTC] signal received connection_id={} signal_type={} state={:?}",
                    resolved_connection_id,
                    signal_type,
                    session.state(),
                );

                if !was_connected
                    && session.state() == crate::transport::NativeWebRTCState::Connected
                {
                    println!(
                        "[NativeWebRTC] state=Connected connection_id={} route_state=candidate",
                        resolved_connection_id,
                    );
                    self.report_native_webrtc_candidate_transport_for_session(
                        &resolved_connection_id,
                        remote_node_id,
                        &session,
                    )
                    .await;
                }
            } else {
                // Production-parity guard: the TS/WASM peer can legitimately start
                // negotiation by sending an SDP offer before the native side has
                // created a `NativeWebRTCDataChannel` session (e.g. if the native
                // capability-update handshake reply failed, or if upgrade initiation
                // is driven purely by the browser actor).
                //
                // If we have no session yet but the connection is session-admitted,
                // proactively start a responder session and then re-attempt routing.
                #[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
                {
                    let should_bootstrap = inbound_signal_can_bootstrap_responder(
                        signal.signal_type.as_str(),
                        signal
                            .sdp
                            .as_ref()
                            .and_then(|sdp| sdp.get("type").and_then(|value| value.as_str())),
                    );

                    if should_bootstrap {
                        let role_override = if signal.signal_type == "sdp"
                            && signal.sdp.as_ref().is_some_and(|sdp| {
                                sdp.get("type")
                                    .and_then(|value| value.as_str())
                                    .map(|value| value == "offer")
                                    .unwrap_or(false)
                            }) {
                            Some(crate::transport::NativeWebRTCRole::Responder)
                        } else {
                            None
                        };
                        let _ = self
                            .maybe_start_native_webrtc_upgrade(
                                connection_id,
                                remote_node_id,
                                false,
                                signal.negotiation_id.as_deref(),
                                role_override,
                            )
                            .await;

                        // Retry resolution after bootstrapping.
                        let sessions = self.native_webrtc_sessions.read().await;
                        let session = sessions.get(connection_id).cloned();
                        drop(sessions);
                        if let Some(session) = session {
                            if !self
                                .current_native_webrtc_route_matches(connection_id, &session)
                                .await
                            {
                                return Ok(());
                            }
                            let was_connected =
                                session.state() == crate::transport::NativeWebRTCState::Connected;
                            let signal_type = signal.signal_type.clone();
                            let result = session.handle_signal(signal).await;
                            if !self
                                .current_native_webrtc_route_matches(connection_id, &session)
                                .await
                            {
                                return Ok(());
                            }
                            result?;
                            println!(
                                "[NativeWebRTC] signal received connection_id={} signal_type={} state={:?}",
                                connection_id,
                                signal_type,
                                session.state(),
                            );
                            if !was_connected
                                && session.state() == crate::transport::NativeWebRTCState::Connected
                            {
                                println!(
                                    "[NativeWebRTC] state=Connected connection_id={} route_state=candidate",
                                    connection_id,
                                );
                                self.report_native_webrtc_candidate_transport_for_session(
                                    connection_id,
                                    remote_node_id,
                                    &session,
                                )
                                .await;
                            }
                            return Ok(());
                        }
                    }
                }

                println!(
                    "[NativeWebRTC] signal ignored: no native session found incoming_connection_id={} remote_node_id={:?}",
                    connection_id,
                    remote_node_id,
                );
            }
            return Ok(());
        }

        #[cfg(any(target_arch = "wasm32", not(feature = "transport-webrtc")))]
        {
            let _ = (connection_id, remote_node_id, signal);
            Ok(())
        }
    }

    /// Public, dependency-free API for explicitly requesting a native WebRTC upgrade
    /// attempt for an already-tracked managed connection.
    ///
    /// This exists primarily for deterministic test harnesses and host runtimes
    /// that want to trigger an upgrade immediately after connection/admission,
    /// without depending on crate-private trigger enums.
    ///
    /// No-op if WebRTC is disabled, the connection is unknown, or session admission
    /// is not yet accepted.
    #[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
    pub async fn request_webrtc_upgrade(
        &self,
        id: &str,
        remote_node_id: Option<&str>,
        reason: Option<&str>,
        role_override: Option<&str>,
        force_restart: bool,
    ) -> anyhow::Result<bool> {
        if !self.is_webrtc_transport_enabled().await {
            println!(
                "[NativeWebRTC] explicit request ignored id={} remote_node_id={} force_restart={} reason={} cause=transport-disabled",
                id,
                remote_node_id.unwrap_or("none"),
                force_restart,
                reason.unwrap_or("none"),
            );
            return Ok(false);
        }

        let reason_label = reason.unwrap_or("none");
        let role_override = match role_override
            .map(str::trim)
            .filter(|value| !value.is_empty())
            .map(|value| value.to_ascii_lowercase())
            .as_deref()
        {
            Some("auto") => Some(crate::transport::NativeWebRTCRole::Auto),
            Some("initiator") => Some(crate::transport::NativeWebRTCRole::Initiator),
            Some("responder") => Some(crate::transport::NativeWebRTCRole::Responder),
            Some(other) => anyhow::bail!("invalid WebRTC role override: {}", other),
            None => None,
        };
        let explicit_remote_node_id = remote_node_id
            .map(str::trim)
            .filter(|value| !value.is_empty())
            .map(ToOwned::to_owned);
        let mut requested = false;

        let resolved_connection_ids = self.resolve_transport_connection_ids(id).await;
        println!(
            "[NativeWebRTC] explicit request resolving id={} remote_node_id={} force_restart={} role_override={:?} reason={} candidates={}",
            id,
            explicit_remote_node_id.as_deref().unwrap_or("none"),
            force_restart,
            role_override,
            reason_label,
            resolved_connection_ids.join(","),
        );

        for connection_id in resolved_connection_ids {
            let Some(record) = self
                .connection_manager
                .get_by_connection_id(&connection_id)
                .await
            else {
                println!(
                    "[NativeWebRTC] explicit request skipped connection_id={} reason=missing-managed-record",
                    connection_id,
                );
                continue;
            };
            let remote_node_id = explicit_remote_node_id
                .clone()
                .or_else(|| record.endpoint_id.clone())
                .or_else(|| record.node_id.clone());
            let Some(remote_node_id) = remote_node_id
                .map(|value| value.trim().to_string())
                .filter(|value| !value.is_empty())
            else {
                println!(
                    "[NativeWebRTC] explicit request skipped connection_id={} reason=missing-remote-node-id",
                    record.connection_id,
                );
                continue;
            };

            let explicit_negotiation_id =
                Self::create_native_webrtc_negotiation_id(&record.connection_id);

            if let Some(reason) = self
                .native_webrtc_upgrade_fast_path_skip_reason(
                    &record.connection_id,
                    Some(&remote_node_id),
                    force_restart,
                    Some(&explicit_negotiation_id),
                )
                .await
            {
                println!(
                    "[NativeWebRTC] explicit request skipped connection_id={} remote_node_id={} reason={}",
                    record.connection_id,
                    remote_node_id,
                    reason,
                );
                continue;
            }

            println!(
                "[NativeWebRTC] explicit request starting connection_id={} remote_node_id={} force_restart={} role_override={:?} negotiation_id={} reason={}",
                record.connection_id,
                remote_node_id,
                force_restart,
                role_override,
                explicit_negotiation_id,
                reason_label,
            );
            self.request_native_webrtc_recovery(
                &record.connection_id,
                Some(&remote_node_id),
                NativeWebRTCRecoveryTrigger::Native(NativeWebRTCNativeTrigger::ExplicitRequest),
                NativeWebRTCRecoveryOptions {
                    force_restart,
                    preferred_negotiation_id: Some(&explicit_negotiation_id),
                    role_override,
                },
            )
            .await?;
            requested = true;
        }

        println!(
            "[NativeWebRTC] explicit request complete id={} remote_node_id={} requested={} reason={}",
            id,
            explicit_remote_node_id.as_deref().unwrap_or("none"),
            requested,
            reason_label,
        );
        Ok(requested)
    }

    #[cfg(any(target_arch = "wasm32", not(feature = "transport-webrtc")))]
    pub async fn request_webrtc_upgrade(
        &self,
        id: &str,
        remote_node_id: Option<&str>,
        reason: Option<&str>,
        role_override: Option<&str>,
        force_restart: bool,
    ) -> anyhow::Result<bool> {
        let _ = (id, remote_node_id, reason, role_override, force_restart);
        Ok(false)
    }

    /// Public native API for explicitly requesting a MoQ setup attempt for an
    /// already-tracked managed connection.
    ///
    /// Like WebRTC, this only starts the parallel upgrade lifecycle. The route is
    /// not promoted to active `moq` until an application payload is actually sent
    /// over a connected MoQ session.
    #[cfg(all(not(target_arch = "wasm32"), feature = "transport-moq"))]
    pub async fn request_moq_upgrade(
        &self,
        id: &str,
        remote_node_id: Option<&str>,
        reason: Option<&str>,
    ) -> anyhow::Result<bool> {
        if !self.is_moq_transport_enabled().await {
            return Ok(false);
        }

        let _ = reason; // reserved for future logging/metrics
        let explicit_remote_node_id = remote_node_id
            .map(str::trim)
            .filter(|value| !value.is_empty())
            .map(ToOwned::to_owned);
        let mut requested = false;

        for connection_id in self.resolve_transport_connection_ids(id).await {
            let Some(record) = self
                .connection_manager
                .get_by_connection_id(&connection_id)
                .await
            else {
                continue;
            };
            let remote_node_id = explicit_remote_node_id
                .clone()
                .or_else(|| record.endpoint_id.clone())
                .or_else(|| record.node_id.clone());
            let Some(remote_node_id) = remote_node_id
                .map(|value| value.trim().to_string())
                .filter(|value| !value.is_empty())
            else {
                continue;
            };

            requested = self
                .maybe_start_native_moq_upgrade(&record.connection_id, Some(&remote_node_id))
                .await?
                || requested;
        }

        Ok(requested)
    }

    #[cfg(all(not(target_arch = "wasm32"), feature = "transport-moq"))]
    pub async fn request_moq_repair(
        &self,
        id: &str,
        remote_node_id: Option<&str>,
        reason: Option<&str>,
    ) -> anyhow::Result<bool> {
        if !self.is_moq_transport_enabled().await {
            return Ok(false);
        }

        let _ = reason;
        let explicit_remote_node_id = remote_node_id
            .map(str::trim)
            .filter(|value| !value.is_empty())
            .map(ToOwned::to_owned);
        let mut requested = false;

        for connection_id in self.resolve_transport_connection_ids(id).await {
            let Some(record) = self
                .connection_manager
                .get_by_connection_id(&connection_id)
                .await
            else {
                continue;
            };
            let remote_node_id = explicit_remote_node_id
                .clone()
                .or_else(|| record.endpoint_id.clone())
                .or_else(|| record.node_id.clone());
            let Some(remote_node_id) = remote_node_id
                .map(|value| value.trim().to_string())
                .filter(|value| !value.is_empty())
            else {
                continue;
            };

            requested = self
                .maybe_start_native_moq_upgrade_with_options(
                    &record.connection_id,
                    Some(&remote_node_id),
                    true,
                )
                .await?
                || requested;
        }

        Ok(requested)
    }

    #[cfg(any(target_arch = "wasm32", not(feature = "transport-moq")))]
    pub async fn request_moq_upgrade(
        &self,
        id: &str,
        remote_node_id: Option<&str>,
        reason: Option<&str>,
    ) -> anyhow::Result<bool> {
        let _ = (id, remote_node_id, reason);
        Ok(false)
    }
}

/// Transport-agnostic send — available on all targets.
impl Client {
    /// Send `data` to `peer_id` using the best available transport.
    ///
    /// Native route order is edge-aware: proven upgraded routes stay first,
    /// relay/unknown iroh can probe upgraded transports first, and direct
    /// iroh/LAN/BLE paths stay primary with upgraded transports as fallback.
    /// On WASM, only the iroh path is active; use the TypeScript `Connection.sendTyped()`
    /// for full transport priority in browser environments.
    pub async fn send_peer(&self, id: &str, data: &[u8]) -> anyhow::Result<()> {
        if data.is_empty() {
            return Ok(());
        }
        if data.len() > crate::native_protocol::MAX_PEER_MESSAGE_BYTES {
            anyhow::bail!(
                "peer message exceeds {}-byte limit; use a named stream for bulk data",
                crate::native_protocol::MAX_PEER_MESSAGE_BYTES,
            );
        }
        let protected = self.protect_outbound_peer_payload(id, data).await?;
        match self.native_peer_send_order(id).await {
            NativePeerSendOrder::OptionalTransportsFirst => {
                if self.try_send_peer_over_webrtc(id, &protected).await? {
                    return Ok(());
                }
                if self.try_send_peer_over_moq(id, &protected).await? {
                    return Ok(());
                }
                self.send_peer_over_iroh(id, &protected).await
            }
            NativePeerSendOrder::IrohFirstAllowOptionalFallback => {
                match self.send_peer_over_iroh(id, &protected).await {
                    Ok(()) => Ok(()),
                    Err(iroh_error) => {
                        if self.try_send_peer_over_webrtc(id, &protected).await? {
                            return Ok(());
                        }
                        if self.try_send_peer_over_moq(id, &protected).await? {
                            return Ok(());
                        }
                        Err(iroh_error)
                    }
                }
            }
        }
    }

    async fn protect_outbound_peer_payload(
        &self,
        id: &str,
        data: &[u8],
    ) -> anyhow::Result<Vec<u8>> {
        let mut connection_ids = Vec::new();
        if let Some(record) = self.connection_manager.get_by_connection_id(id).await {
            connection_ids.push(record.connection_id);
        } else if let Some(record) = self.connection_manager.best_connection_for_peer(id).await {
            connection_ids.push(record.connection_id);
        }

        for connection_id in &connection_ids {
            if self
                .connection_application_crypto_keys
                .read()
                .ok()
                .and_then(|keys| keys.get(connection_id).copied())
                .is_some()
            {
                return self.protect_outbound_application_payload(connection_id, data);
            }
        }
        if self.connection_ids_requiring_application_crypto(&connection_ids) {
            anyhow::bail!("application crypto required for peer send but no key is installed");
        }
        Ok(data.to_vec())
    }

    #[cfg(not(target_arch = "wasm32"))]
    async fn native_peer_send_order(&self, id: &str) -> NativePeerSendOrder {
        let record = if let Some(record) = self.connection_manager.get_by_connection_id(id).await {
            Some(record)
        } else {
            self.connection_manager.best_connection_for_peer(id).await
        };
        let active_transport = record.as_ref().map(|value| value.active_transport.as_str());
        let iroh_lookup_id = record
            .as_ref()
            .and_then(|value| value.node_id.as_deref())
            .or_else(|| {
                record
                    .as_ref()
                    .and_then(|value| value.endpoint_id.as_deref())
            })
            .unwrap_or(id);
        let iroh_transport = self.iroh_path_kind(iroh_lookup_id).await.transport_label();
        peer_send_order(active_transport, Some(iroh_transport))
    }

    #[cfg(target_arch = "wasm32")]
    async fn native_peer_send_order(&self, _id: &str) -> NativePeerSendOrder {
        NativePeerSendOrder::OptionalTransportsFirst
    }
}

/// Native-only iroh send path.
#[cfg(not(target_arch = "wasm32"))]
fn frame_iroh_peer_application_payload(data: &[u8]) -> anyhow::Result<Vec<u8>> {
    let frame_len = 1usize
        .checked_add(data.len())
        .and_then(|len| u32::try_from(len).ok())
        .ok_or_else(|| anyhow::anyhow!("peer application frame is too large"))?;
    let mut frame = Vec::with_capacity(4 + frame_len as usize);
    frame.extend_from_slice(&frame_len.to_be_bytes());
    frame.push(0x00);
    frame.extend_from_slice(data);
    Ok(frame)
}

#[cfg(not(target_arch = "wasm32"))]
impl Client {
    /// Send a bounded peer message over a short-lived application stream.
    ///
    /// Frame format after stream decryption: `[frame_len u32be][type=0x00][payload]`.
    pub async fn send_peer_over_iroh(&self, id: &str, data: &[u8]) -> anyhow::Result<()> {
        let record = if let Some(r) = self.connection_manager.get_by_connection_id(id).await {
            r
        } else if let Some(r) = self.connection_manager.best_connection_for_peer(id).await {
            r
        } else {
            anyhow::bail!("send_peer_over_iroh: no connection record for id={}", id);
        };

        // Product payloads belong on an application stream, never on the SDK
        // native-main control stream. This distinction matters after a physical
        // Iroh replacement: the new admitted stream is wrapped by the host's
        // application-crypto boundary, while native-main remains reserved for
        // admission and optional-route signaling.
        let (_, _, mut send, recv) = self
            .open_peer_bi(&record.connection_id, Some(5_000))
            .await?;
        let frame = frame_iroh_peer_application_payload(data)?;
        send.write_all(&frame).await?;
        send.flush().await?;
        send.finish()?;
        drop(recv);

        let transport_label = self.iroh_path_kind(id).await.transport_label();
        // The Iroh stream above is the base application/fallback route. It must
        // not demote an already-proven MoQ app route while that MoQ session is
        // still data-ready; otherwise health probes over fallback Iroh look like
        // transport churn to consumers.
        #[cfg(all(not(target_arch = "wasm32"), feature = "transport-moq"))]
        if should_preserve_active_moq_on_iroh_send(
            &record.active_transport,
            self.native_moq_data_ready_for_peer(&record.connection_id)
                .await,
        ) {
            let _ = self
                .report_transport_status(
                    &record.connection_id,
                    crate::transport_label::MOQ,
                    record.parallel_transport.as_deref(),
                )
                .await;
            return Ok(());
        }

        // A connected native WebRTC data channel is visible as a parallel
        // candidate until an application send over WebRTC succeeds; only that
        // proven app route may promote WebRTC to active.
        if self
            .get_connected_webrtc_session_for_peer(&record.connection_id)
            .await
            .is_some()
        {
            self.report_native_webrtc_candidate_transport(&record.connection_id, None)
                .await;
        } else {
            let _ = self
                .report_transport_status(&record.connection_id, transport_label, None)
                .await;
        }

        Ok(())
    }

    #[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
    async fn should_proactively_notify_browser_peer(
        &self,
        connection_id: &str,
        remote_node_id: &str,
    ) -> bool {
        let scopes = self.connection_manager.get_scopes(connection_id).await;
        if scopes.iter().any(|scope| {
            super::admission_impl::session_scope_uses_transient_peer_identity(scope.as_str())
        }) {
            return true;
        }

        self.connection_manager
            .get_by_connection_id(connection_id)
            .await
            .and_then(|record| record.device_id)
            .map(|device_id| device_id.trim().eq(remote_node_id))
            .unwrap_or(false)
    }

    #[cfg(not(target_arch = "wasm32"))]
    async fn maybe_negotiate_typescript_application_crypto(
        &self,
        connection_id: &str,
        remote_node_id: Option<&str>,
        handshake: &crate::native_protocol::TypeScriptHandshake,
    ) -> Option<[u8; crate::key_agreement::KEY_AGREEMENT_PUBLIC_KEY_BYTES]> {
        let remote_public = handshake.application_key_agreement_public_key?;
        let remote_node_id = remote_node_id
            .map(str::trim)
            .filter(|value| !value.is_empty())?;
        let local_node_id = self.current_node_id().await?;
        let key_agreement = self
            .get_or_create_connection_key_agreement(connection_id)
            .ok()?;
        let key = key_agreement
            .derive_application_crypto_key(&remote_public, &local_node_id, remote_node_id)
            .ok()?;
        let key_changed =
            self.application_crypto_key_for_connection(Some(connection_id)) != Some(key);
        self.set_connection_application_crypto_required(connection_id);
        self.set_connection_application_crypto_key(connection_id, key);
        if matches!(handshake.action.as_deref(), Some("response" | "ack")) {
            // A key-bearing response is emitted by the TS reciprocal-route
            // owner only after it has installed our public key; an ack is the
            // equivalent native reply. Either proves the current derived key
            // is reciprocal. Hello and capability advertisements remain local
            // derivation facts and must not make the application route ready.
            self.confirm_connection_application_crypto(connection_id);
        }
        println!(
            "[OpenRTC][KEY-AGREEMENT] connection_id={} action={} key_changed={} confirmed={} key_fingerprint={} local_public_fingerprint={} remote_public_fingerprint={} local_node_id={} remote_node_id={}",
            connection_id,
            handshake.action.as_deref().unwrap_or("hello"),
            key_changed,
            self.connection_application_crypto_is_confirmed(connection_id),
            crate::key_agreement::key_agreement_fingerprint(&key),
            crate::key_agreement::key_agreement_fingerprint(&key_agreement.public_key_bytes()),
            crate::key_agreement::key_agreement_fingerprint(&remote_public),
            local_node_id,
            remote_node_id,
        );
        key_changed.then(|| key_agreement.public_key_bytes())
    }

    pub(crate) async fn send_typescript_capability_update(
        &self,
        connection_id: &str,
        reason: &str,
        application_key_agreement_public_key: Option<
            [u8; crate::key_agreement::KEY_AGREEMENT_PUBLIC_KEY_BYTES],
        >,
    ) -> anyhow::Result<()> {
        self.send_typescript_handshake_update(
            connection_id,
            reason,
            application_key_agreement_public_key,
            "capability-update",
        )
        .await
    }

    async fn send_typescript_handshake_update(
        &self,
        connection_id: &str,
        reason: &str,
        application_key_agreement_public_key: Option<
            [u8; crate::key_agreement::KEY_AGREEMENT_PUBLIC_KEY_BYTES],
        >,
        action: &str,
    ) -> anyhow::Result<()> {
        use base64::{engine::general_purpose::URL_SAFE_NO_PAD, Engine as _};

        let local_webrtc_enabled = self.is_webrtc_transport_enabled().await;
        let local_moq_enabled = self.is_moq_transport_enabled().await;
        let local_ble_enabled = self.is_ble_transport_enabled().await;
        // Native must seed mandatory application key agreement. Waiting for the
        // peer's public key here deadlocks with TS peers, which also wait for a
        // remote advertisement before sending a reciprocal capability update.
        let application_key_agreement_public_key =
            application_key_agreement_public_key.or_else(|| {
                self.get_or_create_connection_key_agreement(connection_id)
                    .ok()
                    .map(|agreement| agreement.public_key_bytes())
            });
        let mut capability_frame = serde_json::json!({
            "type": "handshake",
            "action": action,
            "capabilities": {
                "webrtc": local_webrtc_enabled,
                "moq": local_moq_enabled,
                "ble": local_ble_enabled,
            }
        });
        if let Some(public_key) = application_key_agreement_public_key {
            capability_frame["capabilities"]["applicationKeyAgreement"] =
                serde_json::Value::Bool(true);
            capability_frame["applicationKeyAgreement"] = serde_json::json!({
                "algorithm": crate::key_agreement::KEY_AGREEMENT_ALGORITHM,
                "publicKey": URL_SAFE_NO_PAD.encode(public_key),
            });
        }
        println!(
            "[OpenRTC][capability] sending update connection_id={} reason={} action={} webrtc={} moq={} ble={} application_key_agreement={}",
            connection_id,
            reason,
            action,
            local_webrtc_enabled,
            local_moq_enabled,
            local_ble_enabled,
            application_key_agreement_public_key.is_some(),
        );
        self.send_typescript_handshake_over_native_main(connection_id, &capability_frame)
            .await
    }

    async fn send_typescript_handshake_over_native_main(
        &self,
        id: &str,
        message: &serde_json::Value,
    ) -> anyhow::Result<()> {
        let record = if let Some(r) = self.connection_manager.get_by_connection_id(id).await {
            r
        } else if let Some(r) = self.connection_manager.best_connection_for_peer(id).await {
            r
        } else {
            anyhow::bail!(
                "send_typescript_handshake_over_native_main: no connection record for id={}",
                id
            );
        };

        let endpoint_str = record
            .endpoint_id
            .as_deref()
            .or(record.node_id.as_deref())
            .ok_or_else(|| {
                anyhow::anyhow!(
                    "send_typescript_handshake_over_native_main: missing endpoint for connection_id={}",
                    record.connection_id
                )
            })?;

        let endpoint_id = endpoint_str.parse::<iroh::EndpointId>().map_err(|e| {
            anyhow::anyhow!(
                "send_typescript_handshake_over_native_main: invalid endpoint_id={} error={}",
                endpoint_str,
                e
            )
        })?;

        let serialized = serde_json::to_vec(message)?;
        let frame_len = (serialized.len() as u32 + 1).to_be_bytes();
        let mut frame = Vec::with_capacity(4 + 1 + serialized.len());
        frame.extend_from_slice(&frame_len);
        frame.push(0x00);
        frame.extend_from_slice(&serialized);

        self.send_native_main_frame(
            id,
            endpoint_id,
            message
                .get("type")
                .and_then(|value| value.as_str())
                .unwrap_or("control"),
            &frame,
        )
        .await
        .map_err(anyhow::Error::msg)
    }
}

/// WASM stub — `send_peer_over_iroh` is not meaningful on WASM since the TypeScript
/// layer owns transport selection. Defined here so `send_peer` compiles on all targets.
#[cfg(target_arch = "wasm32")]
impl Client {
    pub async fn send_peer_over_iroh(&self, id: &str, data: &[u8]) -> anyhow::Result<()> {
        let _ = (id, data);
        Err(anyhow::anyhow!(
            "send_peer_over_iroh: not available on WASM — use TypeScript Connection.sendTyped()"
        ))
    }
}

fn parse_webrtc_signal_message(
    frame: &serde_json::Value,
) -> Option<crate::transport::WebRTCSignalMessage> {
    if frame.get("type").and_then(|value| value.as_str()) != Some("#pluto-signal") {
        return None;
    }

    let content = frame.get("content")?;
    let transport = content
        .get("transport")
        .and_then(|value| value.as_str())
        .unwrap_or("webrtc");
    if transport != "webrtc" {
        return None;
    }

    let signal_type = content
        .get("type")
        .and_then(|value| value.as_str())
        .map(ToOwned::to_owned)
        .unwrap_or_else(|| "candidate".to_string());

    Some(crate::transport::WebRTCSignalMessage {
        transport: transport.to_string(),
        signal_type,
        negotiation_id: content
            .get("negotiationId")
            .and_then(|value| value.as_str())
            .map(str::trim)
            .filter(|value| !value.is_empty())
            .map(ToOwned::to_owned),
        sdp: content.get("sdp").cloned(),
        candidate: content.get("candidate").cloned(),
    })
}

fn should_send_typescript_capability_update(
    local_webrtc_enabled: bool,
    local_moq_enabled: bool,
    local_ble_enabled: bool,
    has_application_key_agreement_public_key: bool,
    incoming_action: Option<&str>,
) -> bool {
    match incoming_action {
        // Key-bearing replies already carry the complete local capability set.
        // Never answer one with another publication: doing so amplifies a
        // stable key transcript into an unbounded response/update loop.
        Some("capability-update" | "response" | "ack") => return false,
        _ => {}
    }

    has_application_key_agreement_public_key
        || local_webrtc_enabled
        || local_moq_enabled
        || local_ble_enabled
}

fn typescript_application_key_reply_action(
    incoming_action: Option<&str>,
    has_application_key: bool,
    _application_key_changed: bool,
    _application_crypto_was_confirmed: bool,
) -> Option<&'static str> {
    if !has_application_key {
        return None;
    }
    match incoming_action {
        Some("ack") => None,
        Some("response") => Some("ack"),
        _ => Some("response"),
    }
}

#[cfg(not(target_arch = "wasm32"))]
fn should_initiate_native_ble_upgrade(
    local_node_id: &str,
    remote_node_id: &str,
    current_path: IrohPathKind,
) -> bool {
    local_node_id > remote_node_id
        && !matches!(
            current_path,
            IrohPathKind::DirectQuic | IrohPathKind::DirectLan | IrohPathKind::Ble
        )
}

#[cfg(not(target_arch = "wasm32"))]
fn should_accept_native_ble_switch_request(
    local_node_id: &str,
    remote_node_id: &str,
    current_path: IrohPathKind,
) -> bool {
    local_node_id < remote_node_id
        && matches!(current_path, IrohPathKind::Relay | IrohPathKind::Unknown)
}

#[cfg(not(target_arch = "wasm32"))]
async fn remove_native_transport_upgrade_gate(
    gates: &std::sync::Arc<
        tokio::sync::Mutex<std::collections::HashMap<(String, IrohPathKind), String>>,
    >,
    key: &(String, IrohPathKind),
    upgrade_id: &str,
) {
    let mut gates = gates.lock().await;
    if gates.get(key).is_some_and(|current| current == upgrade_id) {
        gates.remove(key);
    }
}

#[cfg(test)]
mod tests {
    #[cfg(not(target_arch = "wasm32"))]
    use super::{
        classify_native_ble_switch_admission, frame_iroh_peer_application_payload,
        native_control_stream_rank, remove_native_transport_upgrade_gate,
        should_accept_native_ble_switch_request, should_initiate_native_ble_upgrade,
        should_replace_native_control_stream, NativeBleSwitchAdmission,
    };
    use super::{
        should_send_typescript_capability_update, typescript_application_key_reply_action,
    };
    #[cfg(not(target_arch = "wasm32"))]
    use crate::client::NativeControlStreamOwner;

    #[cfg(not(target_arch = "wasm32"))]
    #[test]
    fn optional_route_generation_fails_closed_without_manager_record() {
        let expected = crate::client::NativeOptionalRouteGeneration {
            transport_stable_id: 7,
            transport_generation: 2,
            route_generation: 3,
            route_instance_id: 11,
        };

        assert!(!super::native_optional_route_generation_matches(
            None,
            Some(expected),
            expected.route_instance_id,
        ));
    }

    #[cfg(not(target_arch = "wasm32"))]
    use super::Client;

    #[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn closed_webrtc_session_rejects_delayed_signals_from_competing_generations() {
        let client = Client::new_with_app_tag(
            crate::test_constants::TEST_PROJECT_ID.to_string(),
            "test-app".to_string(),
            Box::new(|| None),
        );
        let connection_id = "conn-closed-signal-storm";
        let session = std::sync::Arc::new(
            crate::transport::NativeWebRTCDataChannel::new(
                "node-z",
                "node-a",
                &crate::client::WebRTCConfig {
                    ice_servers: vec![],
                    privacy_mode: false,
                    lan_mode: false,
                },
                std::sync::Arc::new(|_| Box::pin(async move {})),
                "negotiation-retired",
            )
            .await
            .expect("native webrtc session"),
        );
        session.close();
        client
            .native_webrtc_sessions
            .write()
            .await
            .insert(connection_id.to_string(), session.clone());

        for negotiation_id in ["negotiation-old", "negotiation-racing"] {
            for _ in 0..2 {
                let error = client
                    .forward_native_webrtc_signal(
                        connection_id,
                        Some("node-a"),
                        crate::transport::WebRTCSignalMessage {
                            transport: "webrtc".to_string(),
                            signal_type: "candidate".to_string(),
                            negotiation_id: Some(negotiation_id.to_string()),
                            sdp: None,
                            candidate: Some(serde_json::json!({
                                "candidate": "candidate:delayed",
                            })),
                        },
                    )
                    .await
                    .expect_err("retired sessions must reject delayed ICE");
                assert!(
                    error.to_string().contains("retired session"),
                    "unexpected rejection: {error}",
                );
            }
        }

        assert!(
            client
                .get_webrtc_session_for_peer(connection_id)
                .await
                .is_none(),
            "a delayed signal storm must not expose a retired route to product handlers",
        );
        let retained = client
            .native_webrtc_sessions
            .read()
            .await
            .get(connection_id)
            .cloned()
            .expect("retired generation remains fenced until its owner replaces it");
        assert!(std::sync::Arc::ptr_eq(&retained, &session));
        assert_eq!(
            retained.state(),
            crate::transport::NativeWebRTCState::Closed,
        );
    }

    #[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn optional_route_fence_tracks_post_promotion_route_generation() {
        let client = Client::new_with_app_tag(
            crate::test_constants::TEST_PROJECT_ID.to_string(),
            "test-app".to_string(),
            Box::new(|| None),
        );
        let connection_id = "conn-post-promotion-generation";
        client
            .connection_manager
            .upsert_pending(
                connection_id.to_string(),
                Some("node-a".to_string()),
                None,
                Some("node-a".to_string()),
            )
            .await;
        client
            .connection_manager
            .set_connected_with_transport(
                connection_id,
                None,
                Some(73),
                Some(crate::transport_label::IROH_RELAY.to_string()),
            )
            .await;
        let session = std::sync::Arc::new(
            crate::transport::NativeWebRTCDataChannel::new(
                "node-z",
                "node-a",
                &crate::client::WebRTCConfig {
                    ice_servers: vec![],
                    privacy_mode: false,
                    lan_mode: false,
                },
                std::sync::Arc::new(|_| Box::pin(async move {})),
                "negotiation-post-promotion",
            )
            .await
            .expect("native webrtc session"),
        );
        let route_instance_id = std::sync::Arc::as_ptr(&session) as usize;
        client
            .native_webrtc_sessions
            .write()
            .await
            .insert(connection_id.to_string(), session);
        let installed = client
            .install_native_optional_route_generation(
                connection_id,
                super::NativeOptionalRouteKind::WebRtc,
                route_instance_id,
            )
            .await
            .expect("current transport must install a route fence");
        assert_eq!(installed.route_generation, 0);

        client
            .report_transport_status(
                connection_id,
                crate::transport_label::WEBRTC,
                Some(crate::transport_label::IROH_RELAY),
            )
            .await
            .expect("promotion should retain the peer");

        let record = client
            .connection_manager
            .get_by_connection_id(connection_id)
            .await
            .expect("promoted connection record");
        let refreshed = client
            .native_optional_route_generation_for_instance(
                connection_id,
                super::NativeOptionalRouteKind::WebRtc,
                route_instance_id,
            )
            .await
            .expect("promotion must refresh the exact installed route fence");
        assert_eq!(record.route_generation, 1);
        assert_eq!(refreshed.route_generation, record.route_generation);
        assert!(!super::native_optional_route_generation_matches(
            Some(&record),
            Some(installed),
            route_instance_id,
        ));
        assert!(super::native_optional_route_generation_matches(
            Some(&record),
            Some(refreshed),
            route_instance_id,
        ));
    }

    #[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn idle_webrtc_route_can_be_fenced_before_start_without_being_data_ready() {
        let client = Client::new_with_app_tag(
            crate::test_constants::TEST_PROJECT_ID.to_string(),
            "test-app".to_string(),
            Box::new(|| None),
        );
        let connection_id = "conn-idle-webrtc-fence";
        client
            .connection_manager
            .upsert_pending(
                connection_id.to_string(),
                Some("node-a".to_string()),
                None,
                Some("node-a".to_string()),
            )
            .await;
        client
            .connection_manager
            .set_connected_with_transport(
                connection_id,
                None,
                Some(81),
                Some(crate::transport_label::IROH_RELAY.to_string()),
            )
            .await;
        let session = std::sync::Arc::new(
            crate::transport::NativeWebRTCDataChannel::new(
                "node-z",
                "node-a",
                &crate::client::WebRTCConfig {
                    ice_servers: vec![],
                    privacy_mode: false,
                    lan_mode: false,
                },
                std::sync::Arc::new(|_| Box::pin(async move {})),
                "negotiation-idle-fence",
            )
            .await
            .expect("native webrtc session"),
        );
        let route_instance_id = std::sync::Arc::as_ptr(&session) as usize;
        client
            .native_webrtc_sessions
            .write()
            .await
            .insert(connection_id.to_string(), session.clone());
        client
            .install_native_optional_route_generation(
                connection_id,
                super::NativeOptionalRouteKind::WebRtc,
                route_instance_id,
            )
            .await
            .expect("idle route should install its generation fence");

        assert!(
            client
                .current_native_webrtc_route_identity_matches(connection_id, &session)
                .await
        );
        assert!(
            !client
                .current_native_webrtc_route_matches(connection_id, &session)
                .await
        );
    }

    #[cfg(all(not(target_arch = "wasm32"), feature = "transport-moq"))]
    #[tokio::test]
    async fn idle_moq_route_can_be_fenced_before_start_without_being_data_ready() {
        let client = Client::new_with_app_tag(
            crate::test_constants::TEST_PROJECT_ID.to_string(),
            "test-app".to_string(),
            Box::new(|| None),
        );
        let connection_id = "conn-idle-moq-fence";
        client
            .connection_manager
            .upsert_pending(
                connection_id.to_string(),
                Some("node-a".to_string()),
                None,
                Some("node-a".to_string()),
            )
            .await;
        client
            .connection_manager
            .set_connected_with_transport(
                connection_id,
                None,
                Some(82),
                Some(crate::transport_label::IROH_RELAY.to_string()),
            )
            .await;
        let session = std::sync::Arc::new(
            crate::transport::NativeMoQSession::new(
                "node-z",
                "node-a",
                &crate::client::MoQConfig {
                    relay_url: "https://relay.invalid/moq".to_string(),
                    ..crate::client::MoQConfig::default()
                },
            )
            .await
            .expect("native moq session"),
        );
        let route_instance_id = std::sync::Arc::as_ptr(&session) as usize;
        client
            .native_moq_sessions
            .write()
            .await
            .insert(connection_id.to_string(), session.clone());
        client
            .install_native_optional_route_generation(
                connection_id,
                super::NativeOptionalRouteKind::Moq,
                route_instance_id,
            )
            .await
            .expect("idle route should install its generation fence");

        assert!(
            client
                .current_native_moq_route_identity_matches(connection_id, &session)
                .await
        );
        assert!(
            !client
                .current_native_moq_route_matches(connection_id, &session)
                .await
        );
    }

    #[cfg(not(target_arch = "wasm32"))]
    #[tokio::test]
    async fn repeated_typescript_application_key_is_not_reciprocated_twice() {
        let client = Client::new_with_app_tag(
            crate::test_constants::TEST_PROJECT_ID.to_string(),
            "test-app".to_string(),
            Box::new(|| None),
        );
        *client.node_id.write().await = Some("local-node".to_string());
        let remote_agreement =
            crate::key_agreement::EphemeralKeyAgreement::generate().expect("remote agreement");
        let handshake = crate::native_protocol::TypeScriptHandshake {
            action: Some("capability-update".to_string()),
            session_token: None,
            session_token_payload: None,
            claimed_device_id: None,
            capabilities: Some(crate::native_protocol::TypeScriptHandshakeCapabilities {
                webrtc: Some(true),
                moq: Some(false),
                ble: Some(false),
                application_key_agreement: Some(true),
            }),
            application_key_agreement_public_key: Some(remote_agreement.public_key_bytes()),
        };

        assert!(client
            .maybe_negotiate_typescript_application_crypto(
                "connection-1",
                Some("remote-node"),
                &handshake,
            )
            .await
            .is_some());
        assert!(client
            .maybe_negotiate_typescript_application_crypto(
                "connection-1",
                Some("remote-node"),
                &handshake,
            )
            .await
            .is_none());
        assert!(client.connection_requires_application_crypto_confirmation("connection-1"));
        assert!(!client.connection_application_crypto_is_confirmed("connection-1"));
        assert_eq!(
            client.session_admission_block_reason("connection-1"),
            Some((false, "application-crypto-confirmation-pending".to_string())),
        );

        let mut response = handshake.clone();
        response.action = Some("response".to_string());
        assert!(client
            .maybe_negotiate_typescript_application_crypto(
                "connection-1",
                Some("remote-node"),
                &response,
            )
            .await
            .is_none());
        assert!(client.connection_application_crypto_is_confirmed("connection-1"));
        assert_eq!(client.session_admission_block_reason("connection-1"), None);

        client.set_connection_application_crypto_key("connection-1", [9u8; 32]);
        assert!(
            !client.connection_application_crypto_is_confirmed("connection-1"),
            "rotating the key must retire confirmation for the prior key",
        );

        let mut ack = handshake;
        ack.action = Some("ack".to_string());
        assert!(client
            .maybe_negotiate_typescript_application_crypto(
                "connection-1",
                Some("remote-node"),
                &ack,
            )
            .await
            .is_some());
        assert!(client.connection_application_crypto_is_confirmed("connection-1"));
    }

    #[cfg(all(feature = "transport-webrtc", not(target_arch = "wasm32")))]
    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn native_webrtc_does_not_start_without_live_iroh_control_generation() {
        let client = Client::new_with_app_tag(
            crate::test_constants::TEST_PROJECT_ID.to_string(),
            "test-app".to_string(),
            Box::new(|| None),
        );
        let connection_id = "stale-manager-record";
        let remote_node_id = "5d388fcaaa61efa536d78051e060928f6c1fff0983a32f6b532176b12328c3ac";
        *client.node_id.write().await = Some("local-node".to_string());
        client
            .update_transport_config(crate::client::TransportConfig {
                webrtc: Some(crate::client::WebRTCConfig::default()),
                ..crate::client::TransportConfig::default()
            })
            .await
            .expect("transport config");
        client
            .connection_manager
            .upsert_pending(
                connection_id.to_string(),
                Some(remote_node_id.to_string()),
                Some("remote-device".to_string()),
                Some(remote_node_id.to_string()),
            )
            .await;
        client
            .connection_manager
            .set_connected(connection_id, Some(remote_node_id.to_string()))
            .await;

        client
            .maybe_start_native_webrtc_upgrade(
                connection_id,
                Some(remote_node_id),
                true,
                None,
                None,
            )
            .await
            .expect("stale start is ignored");

        assert!(
            client
                .native_webrtc_session_for_connection(connection_id)
                .await
                .is_none(),
            "a manager record without its live Iroh generation must not create a WebRTC session",
        );
        assert!(
            client
                .active_native_webrtc_attempt(connection_id)
                .await
                .is_none(),
            "a rejected start must not leave an in-flight negotiation owner",
        );
    }

    #[cfg(all(feature = "transport-webrtc", not(target_arch = "wasm32")))]
    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn native_webrtc_fast_path_allows_forced_recovery_of_active_session() {
        let client = Client::new_with_app_tag(
            crate::test_constants::TEST_PROJECT_ID.to_string(),
            "test-app".to_string(),
            Box::new(|| None),
        );
        let connection_id = "local-node-remote-node";
        let remote_node_id = "remote-node";

        client
            .connection_manager
            .upsert_pending(
                connection_id.to_string(),
                Some(remote_node_id.to_string()),
                Some("remote-device".to_string()),
                Some(remote_node_id.to_string()),
            )
            .await;
        client
            .connection_manager
            .set_connected(connection_id, Some(remote_node_id.to_string()))
            .await;

        let (tx, _rx) = tokio::sync::mpsc::unbounded_channel::<serde_json::Value>();
        let sender: crate::transport::NativeWebRTCSignalSender =
            std::sync::Arc::new(move |signal| {
                let tx = tx.clone();
                Box::pin(async move {
                    let _ = tx.send(signal);
                })
            });
        let session = std::sync::Arc::new(
            crate::transport::NativeWebRTCDataChannel::new_with_role(
                "local-node",
                remote_node_id,
                &crate::client::WebRTCConfig {
                    ice_servers: vec![],
                    privacy_mode: false,
                    lan_mode: false,
                },
                sender,
                "native-local-node-remote-node",
                crate::transport::NativeWebRTCRole::Responder,
            )
            .await
            .expect("native WebRTC session"),
        );
        session.start().await.expect("session start");
        assert_eq!(
            session.state(),
            crate::transport::NativeWebRTCState::Connecting
        );

        client
            .native_webrtc_sessions
            .write()
            .await
            .insert(connection_id.to_string(), session.clone());

        let negotiation_id = session.negotiation_id();
        let unforced_skip = client
            .native_webrtc_upgrade_fast_path_skip_reason(
                connection_id,
                Some(remote_node_id),
                false,
                Some(&negotiation_id),
            )
            .await;
        assert!(
            unforced_skip
                .as_deref()
                .unwrap_or("")
                .contains("existing session active"),
            "unforced explicit request should skip an active session, got {unforced_skip:?}",
        );

        let forced_skip = client
            .native_webrtc_upgrade_fast_path_skip_reason(
                connection_id,
                Some(remote_node_id),
                true,
                Some(&negotiation_id),
            )
            .await;
        assert!(
            forced_skip.is_none(),
            "forceRestart must allow explicit recovery to replace an active native WebRTC session, got {forced_skip:?}",
        );

        session.close();
    }

    #[cfg(all(feature = "transport-webrtc", not(target_arch = "wasm32")))]
    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn native_webrtc_force_restart_preempts_matching_in_flight_attempt() {
        let client = Client::new_with_app_tag(
            crate::test_constants::TEST_PROJECT_ID.to_string(),
            "test-app".to_string(),
            Box::new(|| None),
        );
        let connection_id = "local-node-remote-node";
        let remote_node_id = "remote-node";
        let negotiation_id = "native-local-node-remote-node";

        client
            .update_transport_config(crate::client::TransportConfig {
                webrtc: Some(crate::client::WebRTCConfig::default()),
                ..crate::client::TransportConfig::default()
            })
            .await
            .expect("transport config");
        client
            .connection_manager
            .upsert_pending(
                connection_id.to_string(),
                Some(remote_node_id.to_string()),
                Some("remote-device".to_string()),
                Some(remote_node_id.to_string()),
            )
            .await;
        client
            .connection_manager
            .set_connected(connection_id, Some(remote_node_id.to_string()))
            .await;
        client
            .native_webrtc_attempts_in_flight
            .write()
            .await
            .insert(
                connection_id.to_string(),
                crate::client::NativeWebRTCAttemptInFlight {
                    negotiation_id: negotiation_id.to_string(),
                    started_at_ms: 1,
                    expires_at_ms: i64::MAX,
                    preserve_on_direct_path: true,
                    last_skip_log_ms: 0,
                },
            );

        client
            .request_native_webrtc_recovery(
                connection_id,
                Some(remote_node_id),
                crate::native_webrtc_policy::NativeWebRTCRecoveryTrigger::Native(
                    crate::native_webrtc_policy::NativeWebRTCNativeTrigger::ExplicitRequest,
                ),
                crate::native_webrtc_policy::NativeWebRTCRecoveryOptions {
                    force_restart: true,
                    preferred_negotiation_id: Some(negotiation_id),
                    role_override: Some(crate::transport::NativeWebRTCRole::Initiator),
                },
            )
            .await
            .expect("forced recovery request");

        assert!(
            client
                .active_native_webrtc_attempt(connection_id)
                .await
                .is_none(),
            "forceRestart must preempt a stale in-flight native WebRTC attempt even when the deterministic negotiation id matches",
        );
    }

    #[cfg(all(feature = "transport-webrtc", not(target_arch = "wasm32")))]
    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn native_webrtc_force_restart_preserves_recent_matching_attempt() {
        let client = Client::new_with_app_tag(
            crate::test_constants::TEST_PROJECT_ID.to_string(),
            "test-app".to_string(),
            Box::new(|| None),
        );
        let connection_id = "local-node-remote-node";
        let remote_node_id = "remote-node";
        let negotiation_id = "native-local-node-remote-node";

        client
            .update_transport_config(crate::client::TransportConfig {
                webrtc: Some(crate::client::WebRTCConfig::default()),
                ..crate::client::TransportConfig::default()
            })
            .await
            .expect("transport config");
        client
            .connection_manager
            .upsert_pending(
                connection_id.to_string(),
                Some(remote_node_id.to_string()),
                Some("remote-device".to_string()),
                Some(remote_node_id.to_string()),
            )
            .await;
        client
            .connection_manager
            .set_connected(connection_id, Some(remote_node_id.to_string()))
            .await;
        client
            .native_webrtc_attempts_in_flight
            .write()
            .await
            .insert(
                connection_id.to_string(),
                crate::client::NativeWebRTCAttemptInFlight {
                    negotiation_id: negotiation_id.to_string(),
                    started_at_ms: super::now_millis_i64(),
                    expires_at_ms: i64::MAX,
                    preserve_on_direct_path: true,
                    last_skip_log_ms: 0,
                },
            );

        client
            .request_native_webrtc_recovery(
                connection_id,
                Some(remote_node_id),
                crate::native_webrtc_policy::NativeWebRTCRecoveryTrigger::Native(
                    crate::native_webrtc_policy::NativeWebRTCNativeTrigger::ExplicitRequest,
                ),
                crate::native_webrtc_policy::NativeWebRTCRecoveryOptions {
                    force_restart: true,
                    preferred_negotiation_id: Some(negotiation_id),
                    role_override: Some(crate::transport::NativeWebRTCRole::Initiator),
                },
            )
            .await
            .expect("forced recovery request");

        assert!(
            client
                .active_native_webrtc_attempt(connection_id)
                .await
                .is_some(),
            "forceRestart must not preempt a recent in-flight attempt without a replacement generation",
        );
    }

    #[test]
    fn sends_typescript_capability_update_for_application_key_without_webrtc() {
        assert!(should_send_typescript_capability_update(
            false,
            false,
            false,
            true,
            Some("hello")
        ));
    }

    #[test]
    fn application_key_reply_protocol_is_advertise_response_ack() {
        assert_eq!(
            typescript_application_key_reply_action(Some("capability-update"), true, true, false,),
            Some("response"),
        );
        assert_eq!(
            typescript_application_key_reply_action(Some("hello"), true, true, false),
            Some("response"),
        );
        assert_eq!(
            typescript_application_key_reply_action(Some("response"), true, false, false),
            Some("ack"),
        );
        assert_eq!(
            typescript_application_key_reply_action(Some("ack"), true, true, false),
            None,
        );
        assert_eq!(
            typescript_application_key_reply_action(Some("capability-update"), false, false, false,),
            None,
        );
        assert_eq!(
            typescript_application_key_reply_action(Some("capability-update"), true, false, true,),
            Some("response"),
        );
        assert_eq!(
            typescript_application_key_reply_action(Some("response"), true, false, true),
            Some("ack"),
        );
    }

    #[test]
    fn advertises_local_transport_before_remote_capability_is_known() {
        assert!(should_send_typescript_capability_update(
            true,
            false,
            false,
            false,
            Some("hello")
        ));
        assert!(!should_send_typescript_capability_update(
            false,
            true,
            false,
            false,
            Some("response")
        ));
        assert!(should_send_typescript_capability_update(
            false, false, true, false, None
        ));
        assert!(!should_send_typescript_capability_update(
            false,
            false,
            false,
            false,
            Some("hello")
        ));
        assert!(!should_send_typescript_capability_update(
            false,
            false,
            true,
            false,
            Some("capability-update")
        ));
        assert!(!should_send_typescript_capability_update(
            true,
            false,
            false,
            true,
            Some("capability-update")
        ));
        assert!(!should_send_typescript_capability_update(
            true,
            true,
            true,
            true,
            Some("ack")
        ));
    }

    #[cfg(not(target_arch = "wasm32"))]
    #[test]
    fn native_ble_upgrade_has_one_initiator_and_preserves_better_paths() {
        assert!(should_initiate_native_ble_upgrade(
            "node-z",
            "node-a",
            crate::client::IrohPathKind::Relay,
        ));
        assert!(should_send_typescript_capability_update(
            true,
            true,
            true,
            false,
            Some("hello")
        ));
        assert!(!should_initiate_native_ble_upgrade(
            "node-a",
            "node-z",
            crate::client::IrohPathKind::Relay,
        ));
        for path in [
            crate::client::IrohPathKind::DirectQuic,
            crate::client::IrohPathKind::DirectLan,
            crate::client::IrohPathKind::Ble,
        ] {
            assert!(!should_initiate_native_ble_upgrade(
                "node-z", "node-a", path,
            ));
        }
        assert!(should_initiate_native_ble_upgrade(
            "node-z",
            "node-a",
            crate::client::IrohPathKind::Unknown,
        ));
        assert!(should_accept_native_ble_switch_request(
            "node-a",
            "node-z",
            crate::client::IrohPathKind::Relay,
        ));
        assert!(!should_accept_native_ble_switch_request(
            "node-z",
            "node-a",
            crate::client::IrohPathKind::Relay,
        ));
        assert!(!should_accept_native_ble_switch_request(
            "node-a",
            "node-z",
            crate::client::IrohPathKind::Ble,
        ));
    }

    #[cfg(not(target_arch = "wasm32"))]
    #[test]
    fn native_ble_switch_waits_for_pending_admission_without_accepting_rejection() {
        let pending = (false, "native-main-route-pending".to_string());
        let rejected = (true, "session-token-rejected".to_string());

        assert_eq!(
            classify_native_ble_switch_admission(None),
            NativeBleSwitchAdmission::Ready,
        );
        assert_eq!(
            classify_native_ble_switch_admission(Some(&pending)),
            NativeBleSwitchAdmission::Wait,
        );
        assert_eq!(
            classify_native_ble_switch_admission(Some(&rejected)),
            NativeBleSwitchAdmission::Reject,
        );
    }

    #[cfg(not(target_arch = "wasm32"))]
    #[tokio::test]
    async fn native_ble_upgrade_gate_cleanup_is_generation_fenced() {
        let gates =
            std::sync::Arc::new(tokio::sync::Mutex::new(std::collections::HashMap::from([
                (
                    ("connection-1".to_string(), crate::client::IrohPathKind::Ble),
                    "upgrade-new".to_string(),
                ),
            ])));
        let key = ("connection-1".to_string(), crate::client::IrohPathKind::Ble);

        remove_native_transport_upgrade_gate(&gates, &key, "upgrade-old").await;
        assert_eq!(
            gates.lock().await.get(&key).map(String::as_str),
            Some("upgrade-new")
        );

        remove_native_transport_upgrade_gate(&gates, &key, "upgrade-new").await;
        assert!(!gates.lock().await.contains_key(&key));
    }

    #[cfg(not(target_arch = "wasm32"))]
    #[test]
    fn native_control_stream_rank_converges_on_same_quic_stream() {
        let client_stream =
            iroh::endpoint::StreamId::new(iroh::endpoint::Side::Client, iroh::endpoint::Dir::Bi, 0);
        let server_stream =
            iroh::endpoint::StreamId::new(iroh::endpoint::Side::Server, iroh::endpoint::Dir::Bi, 0);

        assert!(
            native_control_stream_rank(client_stream) < native_control_stream_rank(server_stream)
        );
    }

    #[cfg(not(target_arch = "wasm32"))]
    #[test]
    fn native_control_stream_replacement_is_generation_fenced() {
        let generation_one = NativeControlStreamOwner {
            transport_stable_id: 10,
            stream_rank: 4,
        };
        let same_generation_better_stream = NativeControlStreamOwner {
            transport_stable_id: 10,
            stream_rank: 2,
        };
        let same_generation_worse_stream = NativeControlStreamOwner {
            transport_stable_id: 10,
            stream_rank: 6,
        };
        let generation_two = NativeControlStreamOwner {
            transport_stable_id: 11,
            stream_rank: 8,
        };

        assert!(should_replace_native_control_stream(
            generation_one,
            same_generation_better_stream,
        ));
        assert!(!should_replace_native_control_stream(
            generation_one,
            same_generation_worse_stream,
        ));
        assert!(should_replace_native_control_stream(
            generation_one,
            generation_two,
        ));
    }

    #[cfg(not(target_arch = "wasm32"))]
    #[test]
    fn iroh_peer_payload_uses_application_stream_framing_not_native_main() {
        let frame = frame_iroh_peer_application_payload(b"hello").unwrap();

        assert_eq!(u32::from_be_bytes(frame[..4].try_into().unwrap()), 6);
        assert_eq!(frame[4], 0x00);
        assert_eq!(&frame[5..], b"hello");
        assert_ne!(&frame[5..], b"main");
    }
}