openrtc 0.2.0

//! 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,
    should_ignore_stale_inbound_signal_recovery,
    should_immediately_retire_active_webrtc_for_reason, NativeWebRTCAttemptReservation,
    NativeWebRTCBrowserSignalTrigger, NativeWebRTCNativeTrigger, NativeWebRTCRecoveryOptions,
    NativeWebRTCRecoveryTrigger,
};

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

impl Client {
    #[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"),
        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,
        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(),
            payload,
        });
        Ok(())
    }

    #[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;
        if removed {
            self.finalize_deferred_managed_retirement_if_needed(connection_id)
                .await;
        } else {
            self.evict_native_signal_stream(connection_id, reason).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);
            // 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".
            let is_preempting_trigger =
                trigger.can_preempt_in_flight_attempt(preferred_negotiation_id, matches_in_flight);
            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"),
                        now_millis_i64().saturating_sub(in_flight.started_at_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"),
                    now_millis_i64().saturating_sub(in_flight.started_at_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,
        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;
            }

            if client
                .connection_manager
                .get_by_connection_id(&connection_id)
                .await
                .is_none()
            {
                return;
            }
            let Ok(endpoint_id) = remote_node_id.parse::<iroh::EndpointId>() else {
                return;
            };
            if client.get_connection(endpoint_id).await.is_none() {
                println!(
                    "[NativeWebRTC] scheduled retry skipped: no live iroh control path connection_id={} remote_node_id={} reason={}",
                    connection_id,
                    remote_node_id,
                    reason_prefix,
                );
                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 signal over the persistent outbound bi-stream to
    /// `endpoint_id`, opening it on first use and caching the send half in
    /// `native_signal_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(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
    pub(crate) fn send_native_signal_frame<'a>(
        &'a self,
        connection_id: &'a str,
        endpoint_id: iroh::EndpointId,
        signal_type: &'a str,
        frame_buf: &'a [u8],
    ) -> NativeSignalSendFuture<'a> {
        Box::pin(self.send_native_signal_frame_inner(
            connection_id,
            endpoint_id,
            signal_type,
            frame_buf,
        ))
    }

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

        // Phase 1: try the cached stream.
        let cached = {
            let map = self.native_signal_streams.lock().await;
            map.get(connection_id).cloned()
        };

        if let Some(stream_arc) = cached {
            let mut send_guard = stream_arc.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.native_signal_streams
                            .lock()
                            .await
                            .remove(connection_id);
                    } 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.native_signal_streams
                        .lock()
                        .await
                        .remove(connection_id);
                }
            }
        }

        // Phase 2: open a fresh persistent stream.
        let Some(connection) = self.get_connection(endpoint_id).await else {
            return Err(format!(
                "no live iroh connection for endpoint {}",
                endpoint_id
            ));
        };

        let (mut send, recv) = connection
            .open_bi()
            .await
            .map_err(|e| format!("open_bi 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(),
        );

        self.native_signal_streams.lock().await.insert(
            connection_id.to_string(),
            Arc::new(tokio::sync::Mutex::new(send)),
        );

        self.spawn_native_signal_stream_tasks(connection_id, endpoint_id, recv);

        Ok(())
    }

    #[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
    fn spawn_native_signal_stream_tasks(
        &self,
        connection_id: &str,
        endpoint_id: iroh::EndpointId,
        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, frame_rx)
                .await;
        });
    }

    #[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
    async fn dispatch_signal_stream_frames(
        &self,
        connection_id: String,
        endpoint_id_for_log: String,
        mut frame_rx: tokio::sync::mpsc::Receiver<crate::native_protocol::ParsedMainFrame>,
    ) {
        use crate::native_protocol::ParsedMainFrame;
        while let Some(parsed) = frame_rx.recv().await {
            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) => {
                    println!(
                        "[SIGNAL-STREAM] frame skipped connection_id={} endpoint_id={} reason=unsupported-on-signal-stream kind=native-message channel={}",
                        connection_id, endpoint_id_for_log, message.channel,
                    );
                }
                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,
        );
        self.cleanup_native_webrtc_session_after_signal_stream_end(
            &connection_id,
            "signal-stream-ended",
        )
        .await;
    }

    #[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
    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(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
    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(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
    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(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
    pub(crate) async fn evict_native_signal_stream(&self, connection_id: &str, reason: &str) {
        let removed = {
            let mut map = self.native_signal_streams.lock().await;
            map.remove(connection_id)
        };
        if removed.is_some() {
            println!(
                "[SIGNAL-STREAM] evicted connection_id={} reason={}",
                connection_id, reason,
            );
        }
    }

    #[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
    async fn cleanup_native_webrtc_session_after_signal_stream_end(
        &self,
        connection_id: &str,
        reason: &str,
    ) {
        self.evict_native_signal_stream(connection_id, reason).await;

        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 {
            println!(
                "[NativeWebRTC] removed session connection_id={} reason={}",
                connection_id, reason,
            );
            expected_session.close();
            self.evict_native_signal_stream(connection_id, reason).await;
        }

        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 transport_config
            .ble
            .as_ref()
            .map(|ble| ble.enabled)
            .unwrap_or(false)
            && !cfg!(all(not(target_arch = "wasm32"), openrtc_unpublished_ble))
        {
            anyhow::bail!(
                "BLE transport is not compiled into this host; build with openrtc_unpublished_ble to enable it"
            );
        }

        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(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 fallback_transport = self
            .resolve_parallel_iroh_transport_label(connection_id, remote_node_id)
            .await;
        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 {
                    let state = session.state();
                    let data_ready =
                        native_moq_is_data_ready(state, session.is_peer_data_subscribed().await);
                    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 native_moq_is_data_ready(
                        session.state(),
                        session.is_peer_data_subscribed().await,
                    ) {
                        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 {
                    continue;
                }

                if session.send(payload).await.is_ok() {
                    let parallel_transport = self
                        .resolve_parallel_iroh_transport_label(&connection_id, None)
                        .await;
                    let _ = self
                        .report_transport_status(
                            &connection_id,
                            "webrtc",
                            Some(parallel_transport.as_str()),
                        )
                        .await;
                    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 {
                    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 session.is_some() {
                    return 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 {
                    continue;
                }

                if !Self::wait_for_stable_native_moq_data_ready(&session).await {
                    continue;
                }

                if session.send(payload).await.is_ok() {
                    let parallel_transport = self
                        .resolve_parallel_iroh_transport_label(&connection_id, None)
                        .await;
                    let report = successful_send_status_report(parallel_transport.as_str());
                    let _ = self
                        .report_transport_status(
                            &connection_id,
                            report.active_transport,
                            report.parallel_transport,
                        )
                        .await;
                    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(
        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 {
                return false;
            }

            if session.is_peer_data_subscribed().await {
                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 local_application_key_agreement_public_key = self
            .maybe_negotiate_typescript_application_crypto(connection_id, remote_node_id, handshake)
            .await;

        println!(
            "[NativeWebRTC] TS handshake capabilities connection_id={} remote_node_id={:?} remote_supports_webrtc={} local_webrtc_enabled={}",
            connection_id,
            remote_node_id,
            remote_supports_webrtc,
            self.is_webrtc_transport_enabled().await,
        );

        let local_webrtc_enabled = self.is_webrtc_transport_enabled().await;
        let should_send_capability_update = should_send_typescript_capability_update(
            remote_supports_webrtc,
            local_webrtc_enabled,
            local_application_key_agreement_public_key.is_some(),
        );

        if should_send_capability_update {
            // Announce our own WebRTC capability back to the browser peer so it
            // can trigger its initiator-side upgrade. The WASM browser only calls
            // requestWebRTCUpgrade() upon receiving a TS handshake with
            // capabilities.webrtc=true; without this reply it never starts, leaving
            // the native host stuck in responder mode waiting for an offer that
            // never arrives. We also send this when application key agreement
            // completes without WebRTC, because native iroh explicit streams use
            // the same negotiated ORTCE1 envelope.
            if let Err(error) = self
                .send_typescript_capability_update(
                    connection_id,
                    if remote_supports_webrtc && local_webrtc_enabled {
                        "ts-handshake"
                    } else {
                        "application-key-agreement"
                    },
                    local_application_key_agreement_public_key,
                )
                .await
            {
                println!(
                    "[NativeWebRTC] capability-update send failed connection_id={} remote_node_id={:?} reason=ts-handshake error={}",
                    connection_id,
                    remote_node_id,
                    error,
                );
            }
        }

        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 && self.is_moq_transport_enabled().await {
            let _ = self
                .maybe_start_native_moq_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
    }

    pub(crate) async fn maybe_handle_typescript_json_frame(
        &self,
        connection_id: &str,
        remote_node_id: Option<&str>,
        frame: &serde_json::Value,
    ) {
        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
                        && negotiation_mismatch
                        && active_attempt.is_some()
                    {
                        println!(
                            "[NativeWebRTC] renegotiate ignored: connected session has in-flight attempt connection_id={} remote_node_id={:?} active_negotiation_id={} signal_negotiation_id={}",
                            connection_id,
                            remote_node_id,
                            session.negotiation_id(),
                            signal_negotiation_id.unwrap_or("none"),
                        );
                        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"))]
            {
                let mut needs_recovery = true;
                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
                        && !existing.is_initiator()
                        && existing.state() == crate::transport::NativeWebRTCState::Connecting
                        && is_offer
                    {
                        if existing.has_remote_description().await {
                            println!(
                                "[NativeWebRTC] replacing stale Connecting responder session for fresh SDP offer connection_id={} remote_node_id={:?} active_negotiation_id={} signal_negotiation_id={}",
                                connection_id,
                                remote_node_id,
                                existing.negotiation_id(),
                                signal_negotiation_id.unwrap_or("none"),
                            );
                            existing.close();
                            self.clear_native_webrtc_attempt(connection_id).await;
                            if self
                                .remove_native_webrtc_session_if_current(
                                    connection_id,
                                    &existing,
                                    "fresh-sdp-replaces-stale-connecting",
                                )
                                .await
                            {
                                self.finalize_deferred_managed_retirement_if_needed(connection_id)
                                    .await;
                            }
                        }

                        println!(
                            "[NativeWebRTC] routing first mismatched SDP offer to existing responder session connection_id={} remote_node_id={:?} active_negotiation_id={} signal_negotiation_id={}",
                            connection_id,
                            remote_node_id,
                            existing.negotiation_id(),
                            signal_negotiation_id.unwrap_or("none"),
                        );
                        needs_recovery = false;
                    }
                }

                if needs_recovery {
                    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"),
                    );
                }

                // Guard: don't create a new session for a retired connection.
                // retire_managed_connection removes the connection record from
                // connection_manager, so if there's no record here, the connection
                // was cleaned up and we must not restart the WebRTC session.
                if self
                    .connection_manager
                    .get_by_connection_id(connection_id)
                    .await
                    .is_none()
                {
                    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(());
                }

                {
                    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 signal_sender: crate::transport::NativeWebRTCSignalSender = {
                    let client = self.clone();
                    let connection_id = connection_id.to_string();
                    let remote_node_id = remote_node_id.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();
                        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) = client
                                .native_webrtc_session_for_connection(&connection_id)
                                .await
                            else {
                                println!(
                                    "[NativeWebRTC] signal send aborted: session missing 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_signal_frame(
                                    &connection_id,
                                    endpoint_id,
                                    &signal_type,
                                    &frame_buf,
                                )
                                .await
                            {
                                Ok(_) => {
                                    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 native_signal_send_failure_is_terminal_for_attempt(&err) {
                                        let negotiation_id = session.negotiation_id();
                                        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;
                                        }
                                        client.schedule_native_webrtc_retry(
                                            &connection_id,
                                            &remote_node_id,
                                            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,
                        );
                        })
                    })
                };

                let negotiation_id = preferred_negotiation_id
                    .clone()
                    .unwrap_or_else(|| Self::create_native_webrtc_negotiation_id(connection_id));
                // 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?,
                );
                {
                    let client_ref = self.clone();
                    let connection_id_ref = connection_id.to_string();
                    let remote_node_id_ref = remote_node_id.clone();
                    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();
                            Box::pin(async move {
                                if let Err(error) = client_ref
                                    .handle_native_transport_peer_data(
                                        &connection_id_ref,
                                        Some(&remote_node_id_ref),
                                        "webrtc",
                                        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 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);
                }

                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 {
                            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;
                                }
                                _ => {}
                            }
                            if tick == no_signal_deadline_ticks
                                && !session_ref.has_remote_description().await
                            {
                                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 {
                            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;
                            }
                            client_ref.schedule_native_webrtc_retry(
                                &connection_id_ref,
                                &remote_node_id_ref,
                                Self::NATIVE_WEBRTC_CONNECT_TIMEOUT_RETRY_DELAY_MS,
                                "upgrade-no-signal",
                            );
                            return;
                        }

                        if !connected {
                            // 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;
                            }
                            client_ref.schedule_native_webrtc_retry(
                                &connection_id_ref,
                                &remote_node_id_ref,
                                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(
                                &connection_id_ref,
                                Some(remote_node_id_ref.as_str()),
                            )
                            .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 {
                                break;
                            }
                            if !client_ref.is_app_backgrounded() {
                                client_ref
                                    .report_native_webrtc_candidate_transport(
                                        &connection_id_ref,
                                        None,
                                    )
                                    .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,
                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();
                    let data_ready =
                        native_moq_is_data_ready(state, existing.is_peer_data_subscribed().await);
                    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)
                    } 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);
                        }
                    }
                    ExistingMoQAction::Reannounce => {
                        let parallel_transport = self
                            .resolve_parallel_iroh_transport_label(connection_id, None)
                            .await;
                        let report = setup_status_report(parallel_transport.as_str());
                        let _ = self
                            .report_transport_status(
                                connection_id,
                                report.active_transport,
                                report.parallel_transport,
                            )
                            .await;
                        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();
                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();
                        Box::pin(async move {
                            if let Err(error) = client_ref
                                .handle_native_transport_peer_data(
                                    &connection_id_ref,
                                    Some(&remote_node_id_ref),
                                    "moq",
                                    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 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,
                );
                return Ok(true);
            }

            let parallel_transport = self
                .resolve_parallel_iroh_transport_label(connection_id, None)
                .await;
            let report = setup_status_report(parallel_transport.as_str());
            let _ = self
                .report_transport_status(
                    connection_id,
                    report.active_transport,
                    report.parallel_transport,
                )
                .await;
            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,
                );
            }
            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 was_connected =
                    session.state() == crate::transport::NativeWebRTCState::Connected;
                let signal_type = signal.signal_type.clone();
                session.handle_signal(signal).await?;

                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(
                        &resolved_connection_id,
                        remote_node_id,
                    )
                    .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 {
                            let was_connected =
                                session.state() == crate::transport::NativeWebRTCState::Connected;
                            let signal_type = signal.signal_type.clone();
                            session.handle_signal(signal).await?;
                            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(
                                    connection_id,
                                    remote_node_id,
                                )
                                .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>,
    ) -> anyhow::Result<bool> {
        if !self.is_webrtc_transport_enabled().await {
            return Ok(false);
        }

        let _ = reason; // reserved for future logging/metrics
        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;

        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;
            };

            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),
                    false,
                    Some(&explicit_negotiation_id),
                )
                .await
            {
                println!(
                    "[NativeWebRTC] explicit request skipped connection_id={} remote_node_id={} reason={}",
                    record.connection_id,
                    remote_node_id,
                    reason,
                );
                continue;
            }

            self.request_native_webrtc_recovery(
                &record.connection_id,
                Some(&remote_node_id),
                NativeWebRTCRecoveryTrigger::Native(NativeWebRTCNativeTrigger::ExplicitRequest),
                NativeWebRTCRecoveryOptions {
                    force_restart: false,
                    preferred_negotiation_id: Some(&explicit_negotiation_id),
                    role_override,
                },
            )
            .await?;
            requested = true;
        }

        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>,
    ) -> anyhow::Result<bool> {
        let _ = (id, remote_node_id, reason, role_override);
        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(());
        }
        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"))]
impl Client {
    /// Send `data` to `peer_id` over an iroh unidirectional stream.
    ///
    /// Frame format: `[0x00][label_len u32be][b"main"][frame_len u32be][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);
        };

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

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

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

        let mut send = self.open_uni_internal(endpoint_id).await?;
        send.write_all(&buf).await?;
        let _ = send.finish();

        let transport_label = self.iroh_path_kind(id).await.transport_label();
        // The Iroh stream above is the durable control/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| {
            matches!(scope.as_str(), "share" | "magic-link") || scope.starts_with("share:")
        }) {
            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)
    }

    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()?;
        self.set_connection_application_crypto_required(connection_id);
        self.set_connection_application_crypto_key(connection_id, key);
        Some(key_agreement.public_key_bytes())
    }

    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<()> {
        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 mut capability_frame = serde_json::json!({
            "type": "handshake",
            "action": "capability-update",
            "capabilities": {
                "webrtc": local_webrtc_enabled,
                "moq": local_moq_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!(
            "[NativeWebRTC] sending capability-update to browser peer connection_id={} reason={} webrtc={} moq={}",
            connection_id,
            reason,
            local_webrtc_enabled,
            local_moq_enabled,
        );
        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 label = b"main";
        let label_len = (label.len() as u32).to_be_bytes();
        let frame_len = (serialized.len() as u32 + 1).to_be_bytes();
        let mut buf = Vec::with_capacity(1 + 4 + label.len() + 4 + 1 + serialized.len());
        buf.push(0x00);
        buf.extend_from_slice(&label_len);
        buf.extend_from_slice(label);
        buf.extend_from_slice(&frame_len);
        buf.push(0x00);
        buf.extend_from_slice(&serialized);

        let connection = self.get_connection(endpoint_id).await.ok_or_else(|| {
            anyhow::anyhow!(
                "send_typescript_handshake_over_native_main: no live connection for endpoint_id={}",
                endpoint_id
            )
        })?;
        let (mut send, _recv) = connection.open_bi().await?;
        tokio::io::AsyncWriteExt::write_all(&mut send, &buf).await?;
        let _ = send.finish();
        Ok(())
    }
}

/// 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(
    remote_supports_webrtc: bool,
    local_webrtc_enabled: bool,
    has_application_key_agreement_public_key: bool,
) -> bool {
    has_application_key_agreement_public_key || (remote_supports_webrtc && local_webrtc_enabled)
}

#[cfg(test)]
mod tests {
    use super::should_send_typescript_capability_update;

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

    #[test]
    fn sends_typescript_capability_update_for_webrtc_capability() {
        assert!(should_send_typescript_capability_update(true, true, false));
        assert!(!should_send_typescript_capability_update(
            true, false, false
        ));
        assert!(!should_send_typescript_capability_update(
            false, true, false
        ));
    }
}