openrtc 2.5.0

//! Rust-owned custom Iroh carrier configuration, negotiation, and lifecycle.
//!
//! WebRTC and MoQ are subordinate packet carriers beneath one admitted logical
//! Iroh connection. This module does not create independent application routes.

use super::*;

#[cfg(all(not(target_arch = "wasm32"), feature = "iroh-carrier-core"))]
fn native_iroh_carrier_failure_code(error: &anyhow::Error) -> &'static str {
    let message = format!("{error:#}");
    if message.contains("base generation changed before candidate acknowledgement")
        || message.contains("incumbent generation changed")
        || message.contains("replacement incumbent is stale")
    {
        "carrier-base-generation-stale"
    } else {
        "carrier-proof-failed"
    }
}

/// Authorization captured before a bilateral carrier commit starts.
///
/// The responder commits first and may close its incumbent as soon as the
/// initiator receives the committed reply. That close invalidates the live
/// route proofs on the initiator before its local atomic commit can run. The
/// candidate proof is already encrypted and generation-bound, so the commit
/// fence must retain the pre-proof authorization epoch while still rejecting
/// any admission or application-security epoch change.
#[cfg(all(not(target_arch = "wasm32"), feature = "iroh-carrier-core"))]
#[derive(Debug, Clone)]
pub(super) struct NativeIrohCarrierAuthorizationFence {
    inbound_admission_was_current: bool,
    remote_admission_proof: Option<RemoteSessionAdmissionProof>,
    admission_fingerprint: Option<String>,
    application_security_epoch_fingerprint: Option<String>,
    policy_epoch: u64,
}

#[cfg(all(not(target_arch = "wasm32"), feature = "transport-moq"))]
fn moq_retry_attempt(failure_code: &str, attempt: u8) -> Option<u8> {
    (failure_code == "carrier-base-generation-stale" && attempt == 1).then_some(2)
}

#[cfg(all(not(target_arch = "wasm32"), feature = "transport-moq"))]
fn native_moq_peer_failure_code(failure_code: Option<&str>) -> (&str, &'static str) {
    let diagnostic = failure_code.unwrap_or("peer-rejected-carrier");
    let lifecycle = match diagnostic {
        "carrier-base-generation-stale" => "carrier-base-generation-stale",
        "carrier-proof-failed" => "carrier-proof-failed",
        _ => "peer-rejected-carrier",
    };
    (diagnostic, lifecycle)
}

#[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
fn webrtc_retry_attempt(failure_code: &str, attempt: u8) -> Option<u8> {
    (attempt == 1
        && matches!(
            failure_code,
            "carrier-base-generation-stale"
                | "carrier-start-failed"
                | "carrier-timeout"
                | "carrier-proof-failed"
        ))
    .then_some(2)
}

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

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

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

#[cfg(not(target_arch = "wasm32"))]
const MAX_REPLACED_NATIVE_CONTROL_DRAIN_FRAMES: u8 = 16;

#[cfg(not(target_arch = "wasm32"))]
fn should_drain_replaced_native_control_frame(drained_frames: u8) -> bool {
    drained_frames < MAX_REPLACED_NATIVE_CONTROL_DRAIN_FRAMES
}

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

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

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

#[cfg(not(target_arch = "wasm32"))]
const NATIVE_BLE_MAX_UPGRADE_ATTEMPTS: u8 = 3;

#[cfg(not(target_arch = "wasm32"))]
const NATIVE_BLE_RETRY_BASE_DELAY_MS: u64 = 1_000;

#[cfg(not(target_arch = "wasm32"))]
const NATIVE_BLE_SWITCH_RESPONSE_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(15);

#[cfg(not(target_arch = "wasm32"))]
const NATIVE_BLE_COMPLETION_GATE_PREFIX: &str = "completion:";

#[cfg(not(target_arch = "wasm32"))]
fn native_ble_retry_delay(attempt: u8, jitter_seed: u64) -> Option<std::time::Duration> {
    if attempt >= NATIVE_BLE_MAX_UPGRADE_ATTEMPTS {
        return None;
    }
    let exponent = u32::from(attempt.saturating_sub(1));
    let base_ms = NATIVE_BLE_RETRY_BASE_DELAY_MS.saturating_mul(2_u64.saturating_pow(exponent));
    let jitter_percent = i64::try_from(jitter_seed % 41).unwrap_or(0) - 20;
    let jitter_ms = i128::from(base_ms) * i128::from(jitter_percent) / 100;
    let delay_ms = (i128::from(base_ms) + jitter_ms).max(1) as u64;
    Some(std::time::Duration::from_millis(delay_ms))
}

#[cfg(not(target_arch = "wasm32"))]
fn native_ble_attempt_state_after_recovery_wake(
    current: crate::client::NativeBleUpgradeAttemptState,
    gate_active: bool,
) -> Option<crate::client::NativeBleUpgradeAttemptState> {
    gate_active.then_some(current)
}

#[cfg(not(target_arch = "wasm32"))]
fn reserve_native_ble_attempt_state(
    current: Option<crate::client::NativeBleUpgradeAttemptState>,
    generation: crate::client::NativePeerDataGeneration,
) -> (crate::client::NativeBleUpgradeAttemptState, Option<u8>) {
    let mut state = current
        .filter(|state| state.generation == generation)
        .unwrap_or(crate::client::NativeBleUpgradeAttemptState {
            generation,
            attempts: 0,
        });
    if state.attempts >= NATIVE_BLE_MAX_UPGRADE_ATTEMPTS {
        return (state, None);
    }
    state.attempts = state.attempts.saturating_add(1);
    let attempt = state.attempts;
    (state, Some(attempt))
}

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

/// Native advertises the dedicated control route only after both application
/// crypto and the current generation's admission proof are live. Keeping this
/// as one predicate makes it difficult for a capability update to publish a
/// stale crypto-only readiness bit during carrier recovery.
fn dedicated_carrier_control_is_ready(
    dedicated_carrier_control: bool,
    application_crypto_confirmed: bool,
    native_admission_route_ready: bool,
) -> bool {
    dedicated_carrier_control && application_crypto_confirmed && native_admission_route_ready
}

impl Client {
    #[cfg(all(
        not(target_arch = "wasm32"),
        feature = "iroh-carrier-core",
        any(feature = "test-harness", feature = "testing-endpoints")
    ))]
    pub fn native_iroh_carrier_debug_events(&self) -> Vec<serde_json::Value> {
        self.native_iroh_carrier_debug_events
            .read()
            .map(|events| events.iter().cloned().collect())
            .unwrap_or_default()
    }

    #[cfg(all(
        not(target_arch = "wasm32"),
        feature = "iroh-carrier-core",
        any(feature = "test-harness", feature = "testing-endpoints")
    ))]
    pub(super) fn record_native_iroh_carrier_debug_event(
        &self,
        connection_id: &str,
        upgrade_id: &str,
        action: &'static str,
        reason: Option<&str>,
    ) {
        // Physical multi-peer runs can emit more than 64 selector/proof events
        // before their failure receipt is captured. Keep the original owner
        // error available without introducing a second diagnostic registry.
        const DEBUG_EVENT_LIMIT: usize = 256;
        let Ok(mut events) = self.native_iroh_carrier_debug_events.write() else {
            return;
        };
        if events.len() == DEBUG_EVENT_LIMIT {
            events.pop_front();
        }
        events.push_back(serde_json::json!({
            "connectionId": connection_id,
            "upgradeId": upgrade_id,
            "action": action,
            "reason": reason,
            "observedAtMs": std::time::SystemTime::now()
                .duration_since(std::time::UNIX_EPOCH)
                .map(|duration| duration.as_millis() as u64)
                .unwrap_or_default(),
        }));
        if Self::admission_trace_enabled() {
            eprintln!(
                "[OpenRTC][Iroh carrier][candidate] connection_id={} upgrade_id={} action={} reason={}",
                connection_id,
                upgrade_id,
                action,
                reason.unwrap_or("none"),
            );
        }
    }

    #[cfg(all(
        not(target_arch = "wasm32"),
        feature = "iroh-carrier-core",
        not(any(feature = "test-harness", feature = "testing-endpoints"))
    ))]
    pub(super) fn record_native_iroh_carrier_debug_event(
        &self,
        _connection_id: &str,
        _upgrade_id: &str,
        _action: &'static str,
        _reason: Option<&str>,
    ) {
    }

    #[cfg(not(target_arch = "wasm32"))]
    pub(crate) fn send_native_main_frame<'a>(
        &'a self,
        connection_id: &'a str,
        endpoint_id: iroh::EndpointId,
        signal_type: &'a str,
        frame_buf: &'a [u8],
    ) -> NativeMainSendFuture<'a> {
        Box::pin(self.send_native_main_frame_inner(
            connection_id,
            endpoint_id,
            signal_type,
            frame_buf,
        ))
    }

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

        let current_transport_stable_id = self
            .get_connection(endpoint_id)
            .await
            .map(|connection| crate::transport_generation::for_connection(&connection))
            .ok_or_else(|| format!("missing Iroh connection for endpoint {}", endpoint_id))?;

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

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

        // Phase 2: open a fresh persistent stream.
        let (transport_stable_id, mut send, recv) = self
            .open_current_bi_internal_with_transport_stable_id(endpoint_id)
            .await
            .map_err(|e| format!("open_bi via iroh failed: {}", e))?;
        if transport_stable_id != current_transport_stable_id {
            return Err(format!(
                "fresh native-main stream opened on a replacement transport generation expected={} actual={}",
                current_transport_stable_id, transport_stable_id,
            ));
        }

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

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

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

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

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

        Ok(())
    }

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

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

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

        let reader_client = self.clone();
        let reader_connection_id = connection_id_owned.clone();
        let reader_endpoint_id = endpoint_id_for_log.clone();
        tokio::spawn(async move {
            reader_client
                .read_signal_stream_loop(
                    reader_connection_id,
                    reader_endpoint_id,
                    owner.transport_stable_id,
                    recv,
                    frame_tx,
                )
                .await;
        });

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

    #[cfg(not(target_arch = "wasm32"))]
    async fn dispatch_signal_stream_frames(
        &self,
        connection_id: String,
        endpoint_id_for_log: String,
        owner: NativeControlStreamOwner,
        mut frame_rx: tokio::sync::mpsc::Receiver<crate::native_protocol::ParsedMainFrame>,
    ) {
        use crate::native_protocol::ParsedMainFrame;
        let mut replaced_same_generation_frames = 0u8;
        while let Some(parsed) = frame_rx.recv().await {
            let owns_current_stream = self
                .native_signal_stream_is_current(&connection_id, owner)
                .await;
            let same_transport_generation = self
                .connection_manager
                .current_transport_matches(&connection_id, Some(owner.transport_stable_id))
                .await;
            if !owns_current_stream && !same_transport_generation {
                println!(
                    "[SIGNAL-STREAM] stale-generation dispatcher stopped connection_id={} transport_stable_id={} stream_rank={}",
                    connection_id, owner.transport_stable_id, owner.stream_rank,
                );
                break;
            }
            if !owns_current_stream
                && !should_drain_replaced_native_control_frame(replaced_same_generation_frames)
            {
                println!(
                    "[SIGNAL-STREAM] same-generation drain cap reached connection_id={} transport_stable_id={} stream_rank={} drained_frames={}",
                    connection_id,
                    owner.transport_stable_id,
                    owner.stream_rank,
                    replaced_same_generation_frames,
                );
                break;
            }
            match parsed {
                ParsedMainFrame::TypeScriptJson(json) => {
                    #[cfg(feature = "iroh-carrier-core")]
                    if json.get("type").and_then(serde_json::Value::as_str)
                        == Some(crate::iroh_carrier_bootstrap::CARRIER_BOOTSTRAP_FRAME_TYPE)
                    {
                        println!(
                            "[OpenRTC][Iroh carrier] dispatching bootstrap connection_id={} endpoint_id={} transport_stable_id={} stream_rank={} owns_current_stream={} same_transport_generation={}",
                            connection_id,
                            endpoint_id_for_log,
                            owner.transport_stable_id,
                            owner.stream_rank,
                            owns_current_stream,
                            same_transport_generation,
                        );
                    }
                    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_admitted_device(&connection_id, device_id).await;
                    }
                    self.maybe_handle_typescript_handshake_capabilities(
                        &connection_id,
                        Some(&endpoint_id_for_log),
                        Some(owner.transport_stable_id),
                        &handshake,
                    )
                    .await;
                }
                ParsedMainFrame::NativeMessage(message) => {
                    let serialized = match serde_json::to_vec(&message) {
                        Ok(serialized) => serialized,
                        Err(error) => {
                            println!(
                                "[SIGNAL-STREAM] native-message rejected connection_id={} endpoint_id={} reason=serialize-failed error={}",
                                connection_id, endpoint_id_for_log, error,
                            );
                            continue;
                        }
                    };
                    if let Err(error) = self
                        .inspect_incoming_native_main_frame_for_transport(
                            &connection_id,
                            Some(&endpoint_id_for_log),
                            None,
                            Some(owner.transport_stable_id),
                            &serialized,
                        )
                        .await
                    {
                        println!(
                            "[SIGNAL-STREAM] native-message rejected connection_id={} endpoint_id={} channel={} error={}",
                            connection_id, endpoint_id_for_log, message.channel, error,
                        );
                    }
                }
                ParsedMainFrame::Opaque => {
                    println!(
                        "[SIGNAL-STREAM] frame skipped connection_id={} endpoint_id={} reason=unsupported-on-signal-stream kind=opaque",
                        connection_id, endpoint_id_for_log,
                    );
                }
            }
            if !owns_current_stream {
                // Canonical-stream convergence can replace a same-generation
                // stream while its reverse half still carries one SDP answer
                // followed by several trickled ICE candidates. Retiring after
                // only the first frame strands native WebRTC in ICE checking.
                // The physical generation remains the security boundary and
                // the replacement owner has already FINed this stream, so drain
                // only a small, hard-bounded tail before retiring. This keeps
                // one persistent control owner without discarding in-flight
                // negotiation frames.
                replaced_same_generation_frames = replaced_same_generation_frames.saturating_add(1);
                println!(
                    "[SIGNAL-STREAM] same-generation buffered frame drained connection_id={} transport_stable_id={} stream_rank={} drained_frames={}",
                    connection_id,
                    owner.transport_stable_id,
                    owner.stream_rank,
                    replaced_same_generation_frames,
                );
            }
        }

        println!(
            "[SIGNAL-STREAM] dispatcher-exited connection_id={}",
            connection_id,
        );
        self.evict_native_signal_stream_if_owner(&connection_id, owner, "control-stream-ended")
            .await;
    }

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

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

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

    #[cfg(not(target_arch = "wasm32"))]
    fn native_main_read_error_proves_transport_loss(error: &str) -> bool {
        error.to_ascii_lowercase().contains("connection lost")
    }

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

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

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

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

    /// Persistent reader on the reverse half of an outbound signaling stream.
    /// Reads [4-byte big-endian len][payload] frames, parses each as JSON, and
    /// forwards them to the dispatcher task through a bounded channel.
    #[cfg(not(target_arch = "wasm32"))]
    async fn read_signal_stream_loop(
        &self,
        connection_id: String,
        endpoint_id_for_log: String,
        transport_stable_id: u64,
        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,
                    );
                    if Self::native_main_read_error_proves_transport_loss(&err) {
                        let demoted = self
                            .observe_native_transport_generation_lost(
                                &connection_id,
                                transport_stable_id,
                                "native-main-frame-read-failed",
                            )
                            .await;
                        println!(
                            "[SIGNAL-STREAM] transport-loss-observed connection_id={} transport_stable_id={} demoted={}",
                            connection_id, transport_stable_id, demoted,
                        );
                    }
                    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 !matches!(parsed, crate::native_protocol::ParsedMainFrame::Opaque) {
                self.record_native_transport_protocol_activity(&connection_id, transport_stable_id);
            }

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

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

    /// Conditional eviction is the control-stream generation barrier. Reader
    /// tasks outlive physical replacement, so only the task that still owns the
    /// registry entry may remove it.
    #[cfg(not(target_arch = "wasm32"))]
    async fn evict_native_signal_stream_if_owner(
        &self,
        connection_id: &str,
        expected_owner: NativeControlStreamOwner,
        reason: &str,
    ) -> bool {
        let removed = {
            let mut map = self.native_control_streams.lock().await;
            match map.get(connection_id) {
                Some(entry) if entry.owner == expected_owner => {
                    let removed = map
                        .remove(connection_id)
                        .expect("current native control owner must still exist");
                    let proofs_invalidated =
                        if self.native_admission_stream_is_persistent(connection_id) {
                            // Keep proof invalidation atomic with exact-owner
                            // removal. Admission installers use this same mutex
                            // when committing their generation-bound proof, so a
                            // duplicate reader cannot erase a newly installed
                            // canonical route.
                            self.invalidate_native_persistent_route_proofs_for_transport(
                                connection_id,
                                expected_owner.transport_stable_id,
                            )
                        } else {
                            false
                        };
                    Some((removed.endpoint_id, proofs_invalidated))
                }
                _ => None,
            }
        };
        if let Some((endpoint_id, proofs_invalidated)) = removed {
            println!(
                "[SIGNAL-STREAM] evicted-current connection_id={} transport_stable_id={} stream_rank={} reason={}",
                connection_id,
                expected_owner.transport_stable_id,
                expected_owner.stream_rank,
                reason,
            );
            if proofs_invalidated {
                // A persistent control-stream close is typed admission-loss
                // input even when the physical Iroh generation remains live.
                // Wake the existing desired-peer actor so it re-presents the
                // cached credential; the resulting proof callback re-enters
                // the single carrier selector. Do not repair this with a
                // control-stream-local retry or an app-owned reconnect.
                let remote_node_id = endpoint_id.to_string();
                let _ = self
                    .wake_native_external_auto_connect_for_replaced_route(remote_node_id.as_str())
                    .await;
            }
            true
        } else {
            println!(
                "[SIGNAL-STREAM] stale-eviction-ignored connection_id={} transport_stable_id={} stream_rank={} reason={}",
                connection_id,
                expected_owner.transport_stable_id,
                expected_owner.stream_rank,
                reason,
            );
            false
        }
    }

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

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

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

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

        #[cfg(feature = "iroh-carrier-core")]
        let carrier_gate_guard = self.native_transport_upgrade_gates.lock().await;
        let mut guard = self.transport_config.write().await;
        #[cfg(feature = "iroh-carrier-core")]
        if *guard != transport_config {
            self.bump_iroh_carrier_policy_epoch();
        }
        *guard = transport_config;
        drop(guard);
        #[cfg(feature = "iroh-carrier-core")]
        drop(carrier_gate_guard);
        #[cfg(not(target_arch = "wasm32"))]
        if !was_ble_enabled && will_enable_ble {
            self.wake_native_ble_recovery("capability-enabled").await;
        }
        #[cfg(all(
            not(target_arch = "wasm32"),
            any(feature = "transport-webrtc", feature = "transport-moq")
        ))]
        self.reconcile_native_iroh_carrier_policy().await?;
        Ok(())
    }

    /// Reconcile global carrier policy through the existing native peer owner.
    /// The caller supplies no peer-specific route decision: Rust ranks mutual
    /// capabilities and prepares a changed custom-carrier winner over the live
    /// incumbent. Atomic candidate commit retires the old generation. A close
    /// without replacement is reserved for the case where no custom carrier
    /// remains eligible.
    #[cfg(all(
        not(target_arch = "wasm32"),
        any(feature = "transport-webrtc", feature = "transport-moq")
    ))]
    async fn reconcile_native_iroh_carrier_policy(&self) -> anyhow::Result<usize> {
        let records = self.connection_manager.list_active().await;
        let capabilities = self.native_peer_transport_capabilities.read().await.clone();
        let config = self.transport_config.read().await.clone();
        #[cfg(feature = "transport-webrtc")]
        let local_webrtc = self.is_webrtc_carrier_enabled().await;
        #[cfg(not(feature = "transport-webrtc"))]
        let local_webrtc = false;
        #[cfg(feature = "transport-moq")]
        let local_moq = self.is_moq_carrier_enabled().await;
        #[cfg(not(feature = "transport-moq"))]
        let local_moq = false;
        let mut reconciled = 0usize;

        for record in records {
            let current = match record.active_transport.as_str() {
                crate::transport_label::WEBRTC => IrohPathKind::WebRtc,
                crate::transport_label::MOQ => IrohPathKind::Moq,
                _ => continue,
            };
            // Missing capability state is unknown, not an explicit withdrawal.
            // Keep a healthy proven carrier until a canonical handshake says
            // otherwise; key-only handshakes and replacement-loop timing must
            // never manufacture a policy close.
            let Some(remote) = capabilities.get(&record.connection_id).cloned() else {
                continue;
            };
            let ranked = config.ranked_iroh_carriers(
                local_webrtc && remote.contains(&NativePeerTransportCapability::WebRtc),
                local_moq && remote.contains(&NativePeerTransportCapability::Moq),
            );
            let desired = ranked.first().copied();
            if desired.is_some_and(|route| route.as_str() == current.transport_label()) {
                continue;
            }
            let Some(remote_node_id) = record.endpoint_id.as_deref().or(record.node_id.as_deref())
            else {
                continue;
            };
            match desired {
                #[cfg(feature = "transport-webrtc")]
                Some(crate::route_policy::KnownRoute::WebRtc) => {
                    self.maybe_start_native_webrtc_iroh_carrier(
                        &record.connection_id,
                        Some(remote_node_id),
                    )
                    .await?;
                    reconciled = reconciled.saturating_add(1);
                }
                #[cfg(feature = "transport-moq")]
                Some(crate::route_policy::KnownRoute::Moq) => {
                    self.maybe_start_native_moq_iroh_carrier(
                        &record.connection_id,
                        Some(remote_node_id),
                    )
                    .await?;
                    reconciled = reconciled.saturating_add(1);
                }
                _ => {
                    let Ok(remote_endpoint_id) = remote_node_id.parse::<iroh::EndpointId>() else {
                        continue;
                    };
                    if self
                        .close_current_iroh_carrier_generation_with_reason(
                            &record.connection_id,
                            remote_endpoint_id,
                            current,
                            record.transport_generation,
                            crate::lifecycle_reason::REASON_IROH_CARRIER_POLICY_CHANGED,
                        )
                        .await
                    {
                        reconciled = reconciled.saturating_add(1);
                    }
                }
            }
        }
        Ok(reconciled)
    }

    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_webrtc_external_transport_enabled(&self) -> bool {
        false
    }

    #[cfg(not(target_arch = "wasm32"))]
    pub async fn is_webrtc_carrier_enabled(&self) -> bool {
        self.transport_config.read().await.webrtc.as_ref().is_some()
            && self
                .native_transport_upgrade_available(IrohPathKind::WebRtc)
                .await
    }

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

    pub async fn is_moq_external_transport_enabled(&self) -> bool {
        false
    }

    /// Capability used by the shared TypeScript/native-main handshake.
    ///
    /// Native advertises only a live compiled mechanism. Browser JavaScript
    /// prepares its configured carrier actor before auto-connect, so WASM can
    /// advertise stable configured intent without depending on which peer wins
    /// deterministic initiation.
    async fn local_typescript_webrtc_capability(&self) -> bool {
        #[cfg(all(feature = "transport-webrtc", target_arch = "wasm32"))]
        let enabled = self.is_webrtc_transport_enabled().await;
        #[cfg(all(feature = "transport-webrtc", not(target_arch = "wasm32")))]
        let enabled = self.is_webrtc_carrier_enabled().await;
        #[cfg(not(feature = "transport-webrtc"))]
        let enabled = false;
        enabled
    }

    async fn local_typescript_moq_capability(&self) -> bool {
        #[cfg(all(feature = "transport-moq", target_arch = "wasm32"))]
        let enabled = self.is_moq_transport_enabled().await;
        #[cfg(all(feature = "transport-moq", not(target_arch = "wasm32")))]
        let enabled = self.is_moq_carrier_enabled().await;
        #[cfg(not(feature = "transport-moq"))]
        let enabled = false;
        enabled
    }

    #[cfg(not(target_arch = "wasm32"))]
    pub async fn is_moq_carrier_enabled(&self) -> bool {
        self.transport_config.read().await.moq.as_ref().is_some()
            && self
                .native_transport_upgrade_available(IrohPathKind::Moq)
                .await
    }

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

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

    pub(crate) async fn maybe_handle_typescript_handshake_capabilities(
        &self,
        connection_id: &str,
        remote_node_id: Option<&str>,
        transport_stable_id: Option<u64>,
        handshake: &crate::native_protocol::TypeScriptHandshake,
    ) {
        let uses_canonical_carrier_schema = handshake
            .capabilities
            .as_ref()
            .is_some_and(|caps| caps.carrier_schema == Some(2));
        let remote_supports_webrtc = uses_canonical_carrier_schema
            && handshake
                .capabilities
                .as_ref()
                .and_then(|caps| caps.webrtc)
                .unwrap_or(false);
        let remote_supports_moq = uses_canonical_carrier_schema
            && handshake
                .capabilities
                .as_ref()
                .and_then(|caps| caps.moq)
                .unwrap_or(false);
        let remote_supports_ble = handshake
            .capabilities
            .as_ref()
            .and_then(|caps| caps.ble)
            .unwrap_or(false);
        #[cfg(all(not(target_arch = "wasm32"), feature = "iroh-carrier-core"))]
        let mut carrier_capabilities_changed = false;
        #[cfg(all(not(target_arch = "wasm32"), feature = "iroh-carrier-core"))]
        let mut carrier_capabilities_were_known = false;
        #[cfg(all(not(target_arch = "wasm32"), feature = "iroh-carrier-core"))]
        let carrier_gate_guard = if uses_canonical_carrier_schema {
            // Capability facts and their per-peer invalidation revision are a
            // single lifecycle mutation. Proof/commit checks use this same
            // gate lock, so a withdrawal cannot become visible while a stale
            // attempt still passes its old revision.
            Some(self.native_transport_upgrade_gates.lock().await)
        } else {
            None
        };
        #[cfg(not(target_arch = "wasm32"))]
        {
            // Key-only transcript frames intentionally omit carrierSchema.
            // They must not erase transport capabilities learned from the
            // canonical carrier advertisement on the same connection.
            if uses_canonical_carrier_schema {
                let mut capabilities = self.native_peer_transport_capabilities.write().await;
                #[cfg(feature = "iroh-carrier-core")]
                {
                    carrier_capabilities_were_known = capabilities.contains_key(connection_id);
                }
                let peer_capabilities = capabilities.entry(connection_id.to_string()).or_default();
                #[cfg(feature = "iroh-carrier-core")]
                let previous = peer_capabilities.clone();
                if remote_supports_webrtc {
                    peer_capabilities.insert(NativePeerTransportCapability::WebRtc);
                } else {
                    peer_capabilities.remove(&NativePeerTransportCapability::WebRtc);
                }
                if remote_supports_moq {
                    peer_capabilities.insert(NativePeerTransportCapability::Moq);
                } else {
                    peer_capabilities.remove(&NativePeerTransportCapability::Moq);
                }
                if remote_supports_ble {
                    peer_capabilities.insert(NativePeerTransportCapability::Ble);
                } else {
                    peer_capabilities.remove(&NativePeerTransportCapability::Ble);
                }
                #[cfg(feature = "iroh-carrier-core")]
                {
                    carrier_capabilities_changed = *peer_capabilities != previous;
                }
                // Retain an explicit empty set. Absence means capability state
                // is not known for this logical connection; an empty set means
                // a canonical carrier advertisement explicitly withdrew all
                // custom carriers. Logical retirement removes the entry.
            }
        }
        #[cfg(not(target_arch = "wasm32"))]
        let (remote_supports_webrtc, remote_supports_moq, remote_supports_ble) = {
            // Response/ACK key transcript frames intentionally omit the
            // carrier schema. Retaining the canonical facts only in the map
            // is insufficient: application-key confirmation is itself the
            // event that re-enters the selector, so that call must consume the
            // retained facts instead of manufacturing false withdrawals.
            let known = self
                .native_peer_transport_capabilities
                .read()
                .await
                .get(connection_id)
                .cloned()
                .unwrap_or_default();
            (
                effective_typescript_transport_capability(
                    uses_canonical_carrier_schema,
                    remote_supports_webrtc,
                    known.contains(&NativePeerTransportCapability::WebRtc),
                ),
                effective_typescript_transport_capability(
                    uses_canonical_carrier_schema,
                    remote_supports_moq,
                    known.contains(&NativePeerTransportCapability::Moq),
                ),
                effective_typescript_transport_capability(
                    uses_canonical_carrier_schema,
                    remote_supports_ble,
                    known.contains(&NativePeerTransportCapability::Ble),
                ),
            )
        };
        #[cfg(all(not(target_arch = "wasm32"), feature = "iroh-carrier-core"))]
        let matching_first_attempt = carrier_capabilities_changed
            && !carrier_capabilities_were_known
            && carrier_gate_guard.as_ref().is_some_and(|gates| {
                let global_policy_epoch = self.current_iroh_carrier_policy_epoch();
                let peer_policy_epoch = self.current_iroh_carrier_peer_policy_epoch(connection_id);
                gates.iter().any(|((candidate_connection_id, kind), gate)| {
                    candidate_connection_id == connection_id
                        && gate.policy_epoch == global_policy_epoch
                        && gate.peer_policy_epoch == peer_policy_epoch
                        && match kind {
                            IrohPathKind::WebRtc => remote_supports_webrtc,
                            IrohPathKind::Moq => remote_supports_moq,
                            IrohPathKind::Ble => remote_supports_ble,
                            _ => false,
                        }
                })
            });
        #[cfg(all(not(target_arch = "wasm32"), feature = "iroh-carrier-core"))]
        if crate::client::should_bump_remote_carrier_peer_policy_epoch(
            carrier_capabilities_were_known,
            carrier_capabilities_changed,
            matching_first_attempt,
        ) {
            self.bump_iroh_carrier_peer_policy_epoch(connection_id);
        }
        #[cfg(all(not(target_arch = "wasm32"), feature = "iroh-carrier-core"))]
        drop(carrier_gate_guard);
        #[cfg(all(
            not(target_arch = "wasm32"),
            any(feature = "transport-webrtc", feature = "transport-moq")
        ))]
        if carrier_capabilities_changed {
            // Remote capabilities are scoped to this logical connection. The
            // policy epoch is global local configuration authority; advancing
            // it here would invalidate in-flight proofs for unrelated peers.
            // A first canonical advertisement can race behind its already-
            // authenticated inbound carrier request; preserve that matching
            // attempt, but invalidate every later change for this peer.
            if let Err(error) = self.reconcile_native_iroh_carrier_policy().await {
                eprintln!(
                    "[OpenRTC][Iroh carrier] capability reconciliation failed connection_id={} error={}",
                    connection_id, error,
                );
            }
        }
        let application_crypto_was_confirmed =
            self.connection_application_crypto_is_confirmed(connection_id, transport_stable_id);
        let local_application_key_agreement_public_key = self
            .maybe_negotiate_typescript_application_crypto(
                connection_id,
                remote_node_id,
                transport_stable_id,
                handshake,
            )
            .await;

        let local_webrtc_enabled = self.local_typescript_webrtc_capability().await;
        let local_moq_enabled = self.local_typescript_moq_capability().await;
        let local_ble_enabled = self.is_ble_transport_enabled().await;
        let received_application_key = handshake.application_key_agreement_public_key.is_some();

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

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

        #[cfg(not(target_arch = "wasm32"))]
        if let Err(error) = self
            .maybe_start_preferred_native_iroh_carrier(
                connection_id,
                remote_node_id,
                remote_supports_webrtc,
                remote_supports_moq,
            )
            .await
        {
            eprintln!(
                "[OpenRTC][Iroh carrier] preferred start failed connection_id={} remote_node_id={:?} error={}",
                connection_id, remote_node_id, error,
            );
        }

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

        #[cfg(not(target_arch = "wasm32"))]
        if self.connection_application_crypto_is_confirmed(connection_id, transport_stable_id) {
            let _ = self
                .confirm_managed_connection_readiness(connection_id)
                .await;
        }
    }

    /// Select one carrier after privacy/capability filtering. Iroh direct/LAN
    /// path gating remains inside each mechanism attempt and no second carrier
    /// is started from this decision.
    #[cfg(all(
        not(target_arch = "wasm32"),
        any(feature = "transport-webrtc", feature = "transport-moq")
    ))]
    pub(super) async fn maybe_start_preferred_native_iroh_carrier(
        &self,
        connection_id: &str,
        remote_node_id: Option<&str>,
        remote_supports_webrtc: bool,
        remote_supports_moq: bool,
    ) -> anyhow::Result<Option<crate::route_policy::KnownRoute>> {
        let Some(base_generation) = self
            .current_native_peer_data_generation(connection_id, None)
            .await
        else {
            self.record_native_iroh_carrier_debug_event(
                connection_id,
                "",
                "selector-deferred",
                Some("base-generation-unavailable"),
            );
            return Ok(None);
        };
        if let Some((_rejected, reason)) = self.session_admission_block_reason_for_transport(
            connection_id,
            Some(base_generation.transport_stable_id),
        ) {
            // Path observation can precede reciprocal token presentation on a
            // replacement Iroh leg. Starting a carrier against that provisional
            // generation lets the admission commit supersede the attempt while
            // the peer still owns its old upgrade fence. The admission/key
            // callbacks already revisit this selector, so wait for that logical
            // event instead of adding a timer or another recovery owner.
            println!(
                "[OpenRTC][Iroh carrier] preferred start deferred connection_id={} transport_stable_id={} reason={}",
                connection_id, base_generation.transport_stable_id, reason,
            );
            self.record_native_iroh_carrier_debug_event(
                connection_id,
                "",
                "selector-deferred",
                Some(&reason),
            );
            return Ok(None);
        }
        #[cfg(feature = "transport-webrtc")]
        let local_webrtc = self.is_webrtc_carrier_enabled().await;
        #[cfg(not(feature = "transport-webrtc"))]
        let local_webrtc = false;
        #[cfg(feature = "transport-moq")]
        let local_moq = self.is_moq_carrier_enabled().await;
        #[cfg(not(feature = "transport-moq"))]
        let local_moq = false;
        let config = self.transport_config.read().await.clone();
        let ranked = config.ranked_iroh_carriers(
            remote_supports_webrtc && local_webrtc,
            remote_supports_moq && local_moq,
        );
        let Some(selected) = ranked.first().copied() else {
            let reason = format!(
                "no-eligible-carrier:remote-webrtc={remote_supports_webrtc}:local-webrtc={local_webrtc}:remote-moq={remote_supports_moq}:local-moq={local_moq}:privacy-mode={}:webrtc-relay-only={}",
                config.privacy_mode,
                config.webrtc.as_ref().is_some_and(|webrtc| webrtc.privacy_mode),
            );
            self.record_native_iroh_carrier_debug_event(
                connection_id,
                "",
                "selector-deferred",
                Some(&reason),
            );
            return Ok(None);
        };
        self.record_native_iroh_carrier_debug_event(
            connection_id,
            "",
            "selector-selected",
            Some(selected.as_str()),
        );
        match selected {
            #[cfg(feature = "transport-webrtc")]
            crate::route_policy::KnownRoute::WebRtc => {
                self.maybe_start_native_webrtc_iroh_carrier(connection_id, remote_node_id)
                    .await?;
            }
            #[cfg(feature = "transport-moq")]
            crate::route_policy::KnownRoute::Moq => {
                self.maybe_start_native_moq_iroh_carrier(connection_id, remote_node_id)
                    .await?;
            }
            _ => return Ok(None),
        }
        Ok(Some(selected))
    }

    /// Re-enter the single carrier selector after an admission proof or local
    /// admission transition. These are logical lifecycle events, so recovery
    /// remains event-driven and cannot create gateway polling traffic.
    #[cfg(all(
        not(target_arch = "wasm32"),
        any(feature = "transport-webrtc", feature = "transport-moq")
    ))]
    pub(super) async fn revisit_iroh_route_after_admission(&self, connection_id: &str) {
        let Some(remote_node_id) = self
            .connection_manager
            .get_by_connection_id(connection_id)
            .await
            .and_then(|record| record.endpoint_id.or(record.node_id))
        else {
            return;
        };
        let remote_capabilities = self
            .native_peer_transport_capabilities
            .read()
            .await
            .get(connection_id)
            .cloned()
            .unwrap_or_default();
        let local_is_responder = self
            .current_node_id()
            .await
            .is_some_and(|local| local.as_str() <= remote_node_id.as_str());
        let selected = match self
            .maybe_start_preferred_native_iroh_carrier(
                connection_id,
                Some(&remote_node_id),
                remote_capabilities.contains(&NativePeerTransportCapability::WebRtc),
                remote_capabilities.contains(&NativePeerTransportCapability::Moq),
            )
            .await
        {
            Ok(selected) => selected,
            Err(error) => {
                eprintln!(
                    "[OpenRTC][Iroh carrier] admission-event start failed connection_id={} remote_node_id={} error={}",
                    connection_id, remote_node_id, error,
                );
                None
            }
        };
        if selected.is_some() && local_is_responder {
            // The deterministic initiator can be a browser. Publish the newly
            // actionable control capability once so its existing handshake
            // observer re-enters the Rust-owned selector without a timer.
            if let Err(error) = self
                .send_typescript_capability_update(connection_id, "native-admission-ready", None)
                .await
            {
                eprintln!(
                    "[OpenRTC][Iroh carrier] admission-ready advertisement failed connection_id={} remote_node_id={} error={}",
                    connection_id, remote_node_id, error,
                );
            }
        }
    }

    #[cfg(all(
        not(target_arch = "wasm32"),
        not(any(feature = "transport-webrtc", feature = "transport-moq"))
    ))]
    pub(super) async fn maybe_start_preferred_native_iroh_carrier(
        &self,
        _connection_id: &str,
        _remote_node_id: Option<&str>,
        _remote_supports_webrtc: bool,
        _remote_supports_moq: bool,
    ) -> anyhow::Result<Option<crate::route_policy::KnownRoute>> {
        Ok(None)
    }

    #[cfg(all(
        not(target_arch = "wasm32"),
        any(feature = "transport-webrtc", feature = "transport-moq")
    ))]
    fn maybe_start_next_native_iroh_carrier<'a>(
        &'a self,
        connection_id: &'a str,
        remote_node_id: &'a str,
        failed: crate::route_policy::KnownRoute,
    ) -> std::pin::Pin<Box<dyn std::future::Future<Output = bool> + Send + 'a>> {
        Box::pin(async move {
            let should_initiate = self
                .current_node_id()
                .await
                .is_some_and(|local| local.as_str() > remote_node_id);
            if !should_initiate {
                return false;
            }
            let remote_capabilities = self
                .native_peer_transport_capabilities
                .read()
                .await
                .get(connection_id)
                .cloned()
                .unwrap_or_default();
            #[cfg(feature = "transport-webrtc")]
            let local_webrtc = self.is_webrtc_carrier_enabled().await;
            #[cfg(not(feature = "transport-webrtc"))]
            let local_webrtc = false;
            #[cfg(feature = "transport-moq")]
            let local_moq = self.is_moq_carrier_enabled().await;
            #[cfg(not(feature = "transport-moq"))]
            let local_moq = false;
            let config = self.transport_config.read().await.clone();
            let ranked = config.ranked_iroh_carriers(
                local_webrtc
                    && remote_capabilities.contains(&NativePeerTransportCapability::WebRtc),
                local_moq && remote_capabilities.contains(&NativePeerTransportCapability::Moq),
            );
            let Some(next) = ranked
                .iter()
                .position(|route| *route == failed)
                .and_then(|index| ranked.get(index + 1))
                .copied()
            else {
                return false;
            };
            let result = match next {
                #[cfg(feature = "transport-webrtc")]
                crate::route_policy::KnownRoute::WebRtc => {
                    self.maybe_start_native_webrtc_iroh_carrier(connection_id, Some(remote_node_id))
                        .await
                }
                #[cfg(feature = "transport-moq")]
                crate::route_policy::KnownRoute::Moq => {
                    self.maybe_start_native_moq_iroh_carrier(connection_id, Some(remote_node_id))
                        .await
                }
                _ => return false,
            };
            if let Err(error) = result {
                eprintln!(
                "[OpenRTC][Iroh carrier] fallback start failed connection_id={} route={} error={}",
                connection_id,
                next.as_str(),
                error,
            );
            }
            true
        })
    }

    #[cfg(not(target_arch = "wasm32"))]
    async fn send_native_ble_upgrade_control(
        &self,
        connection_id: &str,
        action: &str,
        upgrade_id: &str,
        proof_generation: crate::client::NativePeerDataGeneration,
    ) -> anyhow::Result<()> {
        let frame = serde_json::json!({
            "type": "transport-upgrade",
            "transport": "ble",
            "action": action,
            "upgradeId": upgrade_id,
            // This tuple is process-local authority owned by the initiator.
            // The responder only echoes it and binds the encrypted candidate
            // proof to it; it never installs these numbers as local state.
            "base": proof_generation,
        });
        self.send_typescript_handshake_over_native_main(connection_id, &frame)
            .await
    }

    #[cfg(all(
        not(target_arch = "wasm32"),
        any(feature = "transport-webrtc", feature = "transport-moq")
    ))]
    async fn send_iroh_carrier_control(
        &self,
        connection_id: &str,
        frame: &crate::iroh_carrier_bootstrap::CarrierBootstrapFrame,
    ) -> anyhow::Result<()> {
        self.send_typescript_handshake_over_native_main(
            connection_id,
            &serde_json::to_value(frame)?,
        )
        .await
    }

    #[cfg(all(
        not(target_arch = "wasm32"),
        any(feature = "transport-webrtc", feature = "transport-moq")
    ))]
    async fn reject_native_iroh_carrier_request_if_base_blocked(
        &self,
        connection_id: &str,
        bootstrap: &crate::iroh_carrier_bootstrap::CarrierBootstrapFrame,
        transport_stable_id: u64,
        carrier_name: &str,
    ) -> bool {
        let Some((terminal, reason)) = self
            .session_admission_block_reason_for_transport(connection_id, Some(transport_stable_id))
        else {
            return false;
        };
        let failure_code = if terminal {
            "carrier-authorization-rejected"
        } else {
            "carrier-base-not-ready"
        };
        if let Ok(failed) = crate::iroh_carrier_bootstrap::CarrierBootstrapFrame::failed_from(
            bootstrap,
            failure_code,
        ) {
            let _ = self.send_iroh_carrier_control(connection_id, &failed).await;
        }
        eprintln!(
            "[OpenRTC][{carrier_name} carrier] responder rejected connection_id={connection_id} transport_stable_id={transport_stable_id} failure_code={failure_code} reason={reason}",
        );
        true
    }

    #[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
    async fn create_native_webrtc_carrier_attempt(
        &self,
        connection_id: &str,
        remote_node_id: &str,
        bootstrap: crate::iroh_carrier_bootstrap::CarrierBootstrapFrame,
        generation: crate::client::NativePeerDataGeneration,
        role: crate::transport::NativeWebRTCRole,
        start: bool,
    ) -> anyhow::Result<Arc<crate::client::NativeWebRtcCarrierAttempt>> {
        let config = self
            .transport_config
            .read()
            .await
            .webrtc
            .clone()
            .ok_or_else(|| anyhow::anyhow!("WebRTC carrier config is unavailable"))?;
        let local_node_id = self
            .current_node_id()
            .await
            .ok_or_else(|| anyhow::anyhow!("local Iroh node id is unavailable"))?;
        let remote_endpoint_id = remote_node_id.parse::<iroh::EndpointId>()?;
        // A fresh admitted base generation may negotiate the same carrier
        // before the old DataChannel reports terminal. Retire that exact
        // packet session first so the mechanism-neutral carrier can attach
        // the replacement instead of rejecting it as a duplicate.
        if let Some(previous) = self
            .native_webrtc_carrier_attempts
            .write()
            .await
            .remove(connection_id)
        {
            self.record_native_iroh_carrier_debug_event(
                connection_id,
                &previous.upgrade_id,
                "close-requested",
                Some("superseded-before-carrier-activation"),
            );
            previous.channel.close();
            let _ = previous.pump.lock().await.take();
            let _ = self
                .deactivate_native_iroh_packet_carrier(
                    IrohPathKind::WebRtc,
                    previous.remote_endpoint_id,
                )
                .await;
        }
        let packet_session = self
            .activate_native_iroh_packet_carrier(IrohPathKind::WebRtc, remote_endpoint_id)
            .await?;
        let expected = bootstrap.frame_expectation()?;
        let application_key = self
            .application_crypto_key_for_connection(Some(connection_id))
            .ok_or_else(|| {
                anyhow::anyhow!("WebRTC carrier requires the admitted application crypto key")
            })?;
        let client = self.clone();
        let signal_connection_id = connection_id.to_string();
        let signal_upgrade_id = bootstrap.upgrade_id.clone();
        let signal_sender: crate::transport::WebRtcSignalSender =
            Arc::new(move |frame: serde_json::Value| {
                let client = client.clone();
                let connection_id = signal_connection_id.clone();
                let upgrade_id = signal_upgrade_id.clone();
                Box::pin(async move {
                    let current = client
                        .native_webrtc_carrier_attempts
                        .read()
                        .await
                        .get(&connection_id)
                        .is_some_and(|attempt| attempt.upgrade_id == upgrade_id);
                    if current {
                        let _ = client
                            .send_typescript_handshake_over_native_main(&connection_id, &frame)
                            .await;
                    }
                })
            });
        let channel = Arc::new(
            crate::transport::WebRtcDataChannel::new_packet_carrier(
                &local_node_id,
                remote_node_id,
                &config,
                signal_sender,
                &bootstrap.upgrade_id,
                role,
            )
            .await?,
        );
        let pump = crate::native_webrtc_carrier::NativeWebRtcCarrierSession::attach(
            channel.clone(),
            packet_session,
            expected,
            application_key,
        )
        .await?;
        let attempt = Arc::new(crate::client::NativeWebRtcCarrierAttempt {
            upgrade_id: bootstrap.upgrade_id.clone(),
            bootstrap,
            generation,
            remote_endpoint_id,
            channel: channel.clone(),
            pump: tokio::sync::Mutex::new(Some(pump)),
        });
        if let Some(previous) = self
            .native_webrtc_carrier_attempts
            .write()
            .await
            .insert(connection_id.to_string(), attempt.clone())
        {
            self.record_native_iroh_carrier_debug_event(
                connection_id,
                &previous.upgrade_id,
                "close-requested",
                Some("superseded-during-carrier-install"),
            );
            previous.channel.close();
        }
        self.record_native_iroh_carrier_debug_event(
            connection_id,
            &attempt.upgrade_id,
            "attempt-created",
            Some(match role {
                crate::transport::NativeWebRTCRole::Initiator => "initiator",
                crate::transport::NativeWebRTCRole::Responder => "responder",
                crate::transport::NativeWebRTCRole::Auto => "auto",
            }),
        );
        if start {
            if let Err(error) = channel.start().await {
                self.native_webrtc_carrier_attempts
                    .write()
                    .await
                    .remove(connection_id);
                return Err(error);
            }
        }
        Ok(attempt)
    }

    #[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
    async fn wait_native_webrtc_carrier_connected(
        &self,
        connection_id: &str,
        upgrade_id: &str,
    ) -> anyhow::Result<Arc<crate::client::NativeWebRtcCarrierAttempt>> {
        let deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(25);
        loop {
            let attempt = self
                .native_webrtc_carrier_attempts
                .read()
                .await
                .get(connection_id)
                .filter(|attempt| attempt.upgrade_id == upgrade_id)
                .cloned()
                .ok_or_else(|| anyhow::anyhow!("WebRTC carrier attempt was retired"))?;
            match attempt.channel.state() {
                crate::transport::NativeWebRTCState::Connected => return Ok(attempt),
                crate::transport::NativeWebRTCState::Failed
                | crate::transport::NativeWebRTCState::Closed => {
                    anyhow::bail!("WebRTC carrier negotiation failed")
                }
                _ if tokio::time::Instant::now() >= deadline => {
                    anyhow::bail!("WebRTC carrier negotiation timed out")
                }
                _ => tokio::time::sleep(std::time::Duration::from_millis(25)).await,
            }
        }
    }

    #[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
    fn spawn_native_webrtc_carrier_terminal_watcher(
        &self,
        connection_id: String,
        remote_endpoint_id: iroh::EndpointId,
        attempt: Arc<crate::client::NativeWebRtcCarrierAttempt>,
    ) {
        let client = self.clone();
        tokio::spawn(async move {
            loop {
                let current = client
                    .native_webrtc_carrier_attempts
                    .read()
                    .await
                    .get(&connection_id)
                    .is_some_and(|value| Arc::ptr_eq(value, &attempt));
                if !current {
                    return;
                }
                let terminal_reason = attempt
                    .pump
                    .lock()
                    .await
                    .as_ref()
                    .and_then(|pump| pump.take_terminal_reason());
                let terminal_reason = terminal_reason.or_else(|| {
                    matches!(
                        attempt.channel.state(),
                        crate::transport::NativeWebRTCState::Failed
                            | crate::transport::NativeWebRTCState::Closed
                    )
                    .then_some("datachannel-terminal")
                });
                if let Some(reason) = terminal_reason {
                    client.record_native_iroh_carrier_debug_event(
                        &connection_id,
                        &attempt.upgrade_id,
                        "terminal-observed",
                        Some(reason),
                    );
                    // A packet-carrier terminal event belongs to this exact
                    // logical upgrade.  Retire its attempt and reservation
                    // before releasing the selected Iroh generation, so an
                    // admitted replacement cannot mistake the retired
                    // bootstrap for a live duplicate.
                    let retired = client
                        .retire_native_webrtc_carrier_attempt_if_current(
                            &connection_id,
                            &attempt,
                            reason,
                        )
                        .await;
                    if retired {
                        remove_native_transport_upgrade_gate_for_generation(
                            &client.native_transport_upgrade_gates,
                            &(connection_id.clone(), IrohPathKind::WebRtc),
                            &attempt.upgrade_id,
                            attempt.generation,
                        )
                        .await;
                    }
                    let _ = client
                        .close_current_iroh_carrier_generation_with_reason(
                            &connection_id,
                            remote_endpoint_id,
                            IrohPathKind::WebRtc,
                            attempt.generation.transport_generation.saturating_add(1),
                            reason,
                        )
                        .await;
                    return;
                }
                tokio::time::sleep(std::time::Duration::from_millis(250)).await;
            }
        });
    }

    #[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
    async fn retire_native_webrtc_carrier_attempt(
        &self,
        connection_id: &str,
        upgrade_id: &str,
        reason: &str,
    ) -> bool {
        let removed = {
            let mut attempts = self.native_webrtc_carrier_attempts.write().await;
            if attempts
                .get(connection_id)
                .is_some_and(|attempt| attempt.upgrade_id == upgrade_id)
            {
                attempts.remove(connection_id)
            } else {
                None
            }
        };
        if let Some(attempt) = removed {
            self.record_native_iroh_carrier_debug_event(
                connection_id,
                &attempt.upgrade_id,
                "close-requested",
                Some(reason),
            );
            attempt.channel.close();
            let _ = attempt.pump.lock().await.take();
            let _ = self
                .deactivate_native_iroh_packet_carrier(
                    IrohPathKind::WebRtc,
                    attempt.remote_endpoint_id,
                )
                .await;
            true
        } else {
            false
        }
    }

    #[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
    async fn retire_native_webrtc_carrier_attempt_if_current(
        &self,
        connection_id: &str,
        expected: &Arc<crate::client::NativeWebRtcCarrierAttempt>,
        reason: &str,
    ) -> bool {
        let removed = {
            let mut attempts = self.native_webrtc_carrier_attempts.write().await;
            if attempts
                .get(connection_id)
                .is_some_and(|attempt| Arc::ptr_eq(attempt, expected))
            {
                attempts.remove(connection_id)
            } else {
                None
            }
        };
        if let Some(attempt) = removed {
            self.record_native_iroh_carrier_debug_event(
                connection_id,
                &attempt.upgrade_id,
                "close-requested",
                Some(reason),
            );
            attempt.channel.close();
            let _ = attempt.pump.lock().await.take();
            let _ = self
                .deactivate_native_iroh_packet_carrier(
                    IrohPathKind::WebRtc,
                    attempt.remote_endpoint_id,
                )
                .await;
            true
        } else {
            false
        }
    }

    #[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
    async fn fail_native_webrtc_carrier_attempt(
        &self,
        connection_id: &str,
        remote_node_id: &str,
        request: &crate::iroh_carrier_bootstrap::CarrierBootstrapFrame,
        failure_code: &'static str,
        notify_peer: bool,
    ) -> bool {
        let retry_pending = webrtc_retry_attempt(failure_code, request.attempt).is_some();
        let retired = self
            .retire_native_webrtc_carrier_attempt(connection_id, &request.upgrade_id, failure_code)
            .await;
        let gate_removed = remove_native_transport_upgrade_gate(
            &self.native_transport_upgrade_gates,
            &(connection_id.to_string(), IrohPathKind::WebRtc),
            &request.upgrade_id,
        )
        .await;
        if !retired && !gate_removed {
            return false;
        }
        if notify_peer {
            if let Ok(failed) = crate::iroh_carrier_bootstrap::CarrierBootstrapFrame::failed_from(
                request,
                failure_code,
            ) {
                let _ = self.send_iroh_carrier_control(connection_id, &failed).await;
            }
        }
        eprintln!(
            "[OpenRTC][WebRTC carrier] attempt failed connection_id={} upgrade_id={} failure_code={}",
            connection_id, request.upgrade_id, failure_code,
        );
        if !retry_pending {
            let _ = self
                .maybe_start_next_native_iroh_carrier(
                    connection_id,
                    remote_node_id,
                    crate::route_policy::KnownRoute::WebRtc,
                )
                .await;
        }
        true
    }

    #[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
    fn schedule_failed_webrtc_retry(
        &self,
        connection_id: &str,
        remote_node_id: &str,
        request: &crate::iroh_carrier_bootstrap::CarrierBootstrapFrame,
        failure_code: &str,
    ) -> bool {
        let Some(attempt) = webrtc_retry_attempt(failure_code, request.attempt) else {
            return false;
        };
        // Retry once through the same Rust lifecycle owner. A new bootstrap
        // and generation fence prevent delayed failure callbacks from either
        // reviving the retired mechanism or starting a competing attempt.
        let client = self.clone();
        let connection_id = connection_id.to_string();
        let remote_node_id = remote_node_id.to_string();
        tokio::spawn(async move {
            tokio::time::sleep(std::time::Duration::from_millis(500)).await;
            if let Err(error) = client
                .maybe_start_native_webrtc_iroh_carrier_attempt(
                    &connection_id,
                    Some(&remote_node_id),
                    attempt.saturating_sub(1),
                )
                .await
            {
                eprintln!(
                    "[OpenRTC][WebRTC carrier] bounded retry failed connection_id={} attempt={} error={error:#}",
                    connection_id, attempt,
                );
            }
        });
        true
    }

    #[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
    pub(super) async fn maybe_start_native_webrtc_iroh_carrier(
        &self,
        connection_id: &str,
        remote_node_id: Option<&str>,
    ) -> anyhow::Result<()> {
        self.maybe_start_native_webrtc_iroh_carrier_attempt(connection_id, remote_node_id, 0)
            .await
    }

    #[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
    async fn maybe_start_native_webrtc_iroh_carrier_attempt(
        &self,
        connection_id: &str,
        remote_node_id: Option<&str>,
        retry_count: u8,
    ) -> anyhow::Result<()> {
        if !self.is_webrtc_carrier_enabled().await {
            self.record_native_iroh_carrier_debug_event(
                connection_id,
                "",
                "start-deferred",
                Some("carrier-disabled"),
            );
            return Ok(());
        }
        let remote_node_id = remote_node_id
            .map(str::trim)
            .filter(|value| !value.is_empty())
            .ok_or_else(|| anyhow::anyhow!("WebRTC carrier requires remote node id"))?;
        let local_node_id = self
            .current_node_id()
            .await
            .ok_or_else(|| anyhow::anyhow!("WebRTC carrier requires local node id"))?;
        if local_node_id.as_str() <= remote_node_id {
            self.record_native_iroh_carrier_debug_event(
                connection_id,
                "",
                "start-deferred",
                Some("deterministic-responder"),
            );
            return Ok(());
        }
        let endpoint_id = remote_node_id.parse::<iroh::EndpointId>()?;
        let generation = self
            .current_native_peer_data_generation(connection_id, None)
            .await
            .ok_or_else(|| anyhow::anyhow!("WebRTC carrier generation is unavailable"))?;
        let selected_attempt = self
            .native_webrtc_carrier_attempts
            .read()
            .await
            .get(connection_id)
            .filter(|attempt| {
                attempt.remote_endpoint_id == endpoint_id
                    && attempt.channel.state() == crate::transport::NativeWebRTCState::Connected
                    && attempt.generation.transport_generation.saturating_add(1)
                        == generation.transport_generation
            })
            .cloned();
        let selected_record = self
            .connection_manager
            .get_by_connection_id(connection_id)
            .await
            .is_some_and(|record| {
                record.transport_stable_id == Some(generation.transport_stable_id)
                    && record.active_transport == crate::transport_label::WEBRTC
            });
        if selected_attempt.is_some() && selected_record {
            // The path watcher can briefly observe Relay while Iroh converges
            // on an already-proven custom path. The selected carrier owns the
            // current physical generation; preparing another attempt would
            // close that healthy DataChannel and manufacture LocallyClosed.
            self.record_native_iroh_carrier_debug_event(
                connection_id,
                "",
                "start-deferred",
                Some("carrier-already-selected"),
            );
            return Ok(());
        }
        let base_path = self.iroh_path_kind(remote_node_id).await;
        if !crate::iroh_connection_policy::custom_carrier_base_allows(
            base_path,
            IrohPathKind::WebRtc,
            false,
        ) {
            let reason = format!("base-path-not-relay:{base_path:?}");
            self.record_native_iroh_carrier_debug_event(
                connection_id,
                "",
                "start-deferred",
                Some(&reason),
            );
            return Ok(());
        }
        self.get_connection(endpoint_id)
            .await
            .ok_or_else(|| anyhow::anyhow!("WebRTC carrier base is unavailable"))?;
        let kind = IrohPathKind::WebRtc;
        let gate_key = (connection_id.to_string(), kind);
        let bootstrap = crate::iroh_carrier_bootstrap::CarrierBootstrapFrame::request(
            crate::iroh_carrier_bootstrap::CarrierBootstrapKind::WebRtc,
            generation.into(),
            retry_count.saturating_add(1),
        )?;
        {
            let mut gates = self.native_transport_upgrade_gates.lock().await;
            if !reserve_native_transport_upgrade_gate(
                &mut gates,
                gate_key.clone(),
                &bootstrap.upgrade_id,
                generation,
                self.current_iroh_carrier_policy_epoch(),
                self.current_iroh_carrier_peer_policy_epoch(connection_id),
                false,
            ) {
                self.record_native_iroh_carrier_debug_event(
                    connection_id,
                    &bootstrap.upgrade_id,
                    "start-deferred",
                    Some("upgrade-gate-active"),
                );
                return Ok(());
            }
        }
        let result = async {
            let attempt = self
                .create_native_webrtc_carrier_attempt(
                    connection_id,
                    remote_node_id,
                    bootstrap.clone(),
                    generation,
                    crate::transport::NativeWebRTCRole::Initiator,
                    false,
                )
                .await?;
            self.send_iroh_carrier_control(connection_id, &bootstrap)
                .await?;
            attempt.channel.start().await?;
            Ok::<(), anyhow::Error>(())
        }
        .await;
        if let Err(error) = result {
            eprintln!(
                "[OpenRTC][WebRTC carrier] start failed connection_id={} upgrade_id={} retry_count={} error={error:#}",
                connection_id, bootstrap.upgrade_id, retry_count,
            );
            let retired = self
                .fail_native_webrtc_carrier_attempt(
                    connection_id,
                    remote_node_id,
                    &bootstrap,
                    "carrier-start-failed",
                    true,
                )
                .await;
            if retired
                && self.schedule_failed_webrtc_retry(
                    connection_id,
                    remote_node_id,
                    &bootstrap,
                    "carrier-start-failed",
                )
            {
                return Ok(());
            }
            return Err(error);
        }
        let client = self.clone();
        let connection_id = connection_id.to_string();
        let remote_node_id = remote_node_id.to_string();
        tokio::spawn(async move {
            tokio::time::sleep(std::time::Duration::from_secs(30)).await;
            let still_current = client
                .native_transport_upgrade_gates
                .lock()
                .await
                .get(&(connection_id.clone(), kind))
                .is_some_and(|gate| {
                    native_transport_upgrade_gate_matches(gate, &bootstrap.upgrade_id)
                });
            if still_current {
                let retired = client
                    .fail_native_webrtc_carrier_attempt(
                        &connection_id,
                        &remote_node_id,
                        &bootstrap,
                        "carrier-timeout",
                        true,
                    )
                    .await;
                if retired {
                    let _ = client.schedule_failed_webrtc_retry(
                        &connection_id,
                        &remote_node_id,
                        &bootstrap,
                        "carrier-timeout",
                    );
                }
            }
        });
        Ok(())
    }

    #[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
    async fn handle_native_webrtc_iroh_carrier_control(
        &self,
        connection_id: &str,
        remote_node_id: Option<&str>,
        frame: &serde_json::Value,
    ) -> bool {
        if let Some(signal) = parse_webrtc_signal_message_for_transport(frame, "webrtc") {
            let attempt = self
                .native_webrtc_carrier_attempts
                .read()
                .await
                .get(connection_id)
                .cloned();
            if let Some(attempt) = attempt {
                if iroh_webrtc_signal_matches_upgrade(&signal, &attempt.upgrade_id) {
                    let _ = attempt.channel.handle_signal(signal).await;
                } else {
                    eprintln!(
                        "[OpenRTC][WebRTC carrier] ignored stale signal connection_id={} current_upgrade_id={} signal_upgrade_id={:?}",
                        connection_id, attempt.upgrade_id, signal.negotiation_id,
                    );
                }
            }
            return true;
        }
        let Some(bootstrap) =
            crate::iroh_carrier_bootstrap::CarrierBootstrapFrame::from_json(frame)
        else {
            if frame.get("type").and_then(serde_json::Value::as_str)
                == Some(crate::iroh_carrier_bootstrap::CARRIER_BOOTSTRAP_FRAME_TYPE)
            {
                eprintln!("[OpenRTC][WebRTC carrier] rejected invalid bootstrap envelope={frame}");
            }
            return false;
        };
        if bootstrap.carrier != crate::iroh_carrier_bootstrap::CarrierBootstrapKind::WebRtc {
            return false;
        }
        let Some(remote_node_id) = remote_node_id
            .map(str::trim)
            .filter(|value| !value.is_empty())
        else {
            return true;
        };
        let kind = IrohPathKind::WebRtc;
        let endpoint_id = match remote_node_id.parse::<iroh::EndpointId>() {
            Ok(value) => value,
            Err(_) => return true,
        };
        match bootstrap.action {
            crate::iroh_carrier_bootstrap::CarrierBootstrapAction::Request => {
                let local_node_id = match self.current_node_id().await {
                    Some(value) => value,
                    None => {
                        eprintln!("[OpenRTC][WebRTC carrier] responder skipped connection_id={connection_id} reason=local-endpoint-unavailable");
                        return true;
                    }
                };
                if local_node_id.as_str() >= remote_node_id {
                    eprintln!("[OpenRTC][WebRTC carrier] responder skipped connection_id={connection_id} reason=deterministic-initiator");
                    return true;
                }
                if !self.is_webrtc_carrier_enabled().await {
                    eprintln!("[OpenRTC][WebRTC carrier] responder skipped connection_id={connection_id} reason=carrier-disabled");
                    return true;
                }
                let duplicate_attempt = self
                    .native_webrtc_carrier_attempts
                    .read()
                    .await
                    .get(connection_id)
                    .filter(|attempt| {
                        native_carrier_duplicate_request_is_live(
                            &bootstrap,
                            &attempt.bootstrap,
                            matches!(
                                attempt.channel.state(),
                                crate::transport::NativeWebRTCState::Failed
                                    | crate::transport::NativeWebRTCState::Closed
                            ),
                        )
                    })
                    .cloned();
                if let Some(attempt) = duplicate_attempt {
                    if attempt.channel.state() == crate::transport::NativeWebRTCState::Connected {
                        if let Ok(ready) =
                            crate::iroh_carrier_bootstrap::CarrierBootstrapFrame::ready_from(
                                &bootstrap,
                            )
                        {
                            let _ = self.send_iroh_carrier_control(connection_id, &ready).await;
                        }
                    }
                    eprintln!("[OpenRTC][WebRTC carrier] responder reused connection_id={connection_id} reason=duplicate-live-bootstrap");
                    return true;
                }
                if !crate::iroh_connection_policy::custom_carrier_base_allows(
                    self.iroh_path_kind(remote_node_id).await,
                    IrohPathKind::WebRtc,
                    false,
                ) {
                    eprintln!("[OpenRTC][WebRTC carrier] responder skipped connection_id={connection_id} reason=carrier-already-selected");
                    return true;
                }
                if self.get_connection(endpoint_id).await.is_none() {
                    eprintln!("[OpenRTC][WebRTC carrier] responder skipped connection_id={connection_id} reason=base-unavailable");
                    return true;
                }
                // `bootstrap.base` is the initiator's process-local fence. The
                // responder owns a distinct lifecycle and fences this attempt
                // with its own current generation; Ready echoes the initiator's
                // fence so the initiator can validate it on return.
                let generation = match self
                    .current_native_peer_data_generation(connection_id, None)
                    .await
                {
                    Some(value) => value,
                    None => {
                        eprintln!("[OpenRTC][WebRTC carrier] responder skipped connection_id={connection_id} reason=generation-unavailable");
                        return true;
                    }
                };
                if self
                    .reject_native_iroh_carrier_request_if_base_blocked(
                        connection_id,
                        &bootstrap,
                        generation.transport_stable_id,
                        "WebRTC",
                    )
                    .await
                {
                    return true;
                }
                let node = match self.iroh_node.read().await.as_ref().cloned() {
                    Some(value) => value,
                    None => {
                        eprintln!("[OpenRTC][WebRTC carrier] responder skipped connection_id={connection_id} reason=node-unavailable");
                        return true;
                    }
                };
                let provider = match self.native_transport_upgrade_provider(kind).await {
                    Some(value) => value,
                    None => {
                        eprintln!("[OpenRTC][WebRTC carrier] responder skipped connection_id={connection_id} reason=provider-unavailable");
                        return true;
                    }
                };
                let gate_key = (connection_id.to_string(), kind);
                let current_attempt = self
                    .native_webrtc_carrier_attempts
                    .read()
                    .await
                    .get(connection_id)
                    .cloned();
                {
                    let mut gates = self.native_transport_upgrade_gates.lock().await;
                    if let Some(current) = gates.get(&gate_key).filter(|current| {
                        self.iroh_carrier_policy_epoch_is_current(current.policy_epoch)
                            && current.peer_policy_epoch
                                == self.current_iroh_carrier_peer_policy_epoch(connection_id)
                    }) {
                        if native_webrtc_duplicate_bootstrap_is_reserved_or_live(
                            current,
                            &bootstrap.upgrade_id,
                            generation,
                            current_attempt.as_deref(),
                        ) {
                            eprintln!("[OpenRTC][WebRTC carrier] responder reused connection_id={connection_id} reason=duplicate-bootstrap");
                            return true;
                        } else if native_transport_upgrade_gate_matches(
                            current,
                            &bootstrap.upgrade_id,
                        ) {
                            // The same bootstrap ID can be replayed after its
                            // prior DataChannel or base generation retired.
                            // It is no longer a duplicate of a live attempt.
                            gates.remove(&gate_key);
                        } else {
                            eprintln!("[OpenRTC][WebRTC carrier] responder skipped connection_id={connection_id} reason=competing-upgrade");
                            return true;
                        }
                    }
                    if !reserve_native_transport_upgrade_gate(
                        &mut gates,
                        gate_key.clone(),
                        &bootstrap.upgrade_id,
                        generation,
                        self.current_iroh_carrier_policy_epoch(),
                        self.current_iroh_carrier_peer_policy_epoch(connection_id),
                        true,
                    ) {
                        eprintln!("[OpenRTC][WebRTC carrier] responder skipped connection_id={connection_id} reason=reservation-rejected");
                        return true;
                    }
                }
                let authorization_expiry = node
                    .authorize_inbound_replacement(
                        endpoint_id,
                        provider.transport_id(),
                        std::time::Duration::from_secs(45),
                    )
                    .await;

                // The native-main signal dispatcher is ordered, but the
                // initiator can enqueue its SDP offer and ICE candidates
                // immediately after the bootstrap request. Install and start
                // the responder attempt before returning to that dispatcher;
                // spawning attempt creation here would let those already-
                // buffered frames run first and be discarded because no
                // attempt existed yet. The long-lived proof/commit wait stays
                // asynchronous below.
                let attempt = match self
                    .create_native_webrtc_carrier_attempt(
                        connection_id,
                        remote_node_id,
                        bootstrap.clone(),
                        generation,
                        crate::transport::NativeWebRTCRole::Responder,
                        true,
                    )
                    .await
                {
                    Ok(attempt) => attempt,
                    Err(error) => {
                        node.revoke_inbound_replacement_if_current(
                            endpoint_id,
                            provider.transport_id(),
                            authorization_expiry,
                        )
                        .await;
                        let failure_code = native_iroh_carrier_failure_code(&error);
                        eprintln!(
                            "[OpenRTC][WebRTC carrier] responder start failed connection_id={} upgrade_id={} failure_code={} error={:#}",
                            connection_id, bootstrap.upgrade_id, failure_code, error,
                        );
                        self.fail_native_webrtc_carrier_attempt(
                            connection_id,
                            remote_node_id,
                            &bootstrap,
                            failure_code,
                            true,
                        )
                        .await;
                        return true;
                    }
                };
                let client = self.clone();
                let connection_id = connection_id.to_string();
                let remote_node_id = remote_node_id.to_string();
                tokio::spawn(async move {
                    let result = async {
                        client
                            .wait_native_webrtc_carrier_connected(
                                &connection_id,
                                &bootstrap.upgrade_id,
                            )
                            .await?;
                        let ready =
                            crate::iroh_carrier_bootstrap::CarrierBootstrapFrame::ready_from(
                                &bootstrap,
                            )?;
                        // Ready means this endpoint's relay session is prepared.
                        // The browser opens its directed subscription in response;
                        // waiting for that subscription before Ready deadlocks the
                        // browser-initiated role.
                        client
                            .send_iroh_carrier_control(&connection_id, &ready)
                            .await?;
                        client
                            .complete_inbound_native_iroh_carrier_upgrade(
                                &connection_id,
                                &remote_node_id,
                                &bootstrap.upgrade_id,
                                generation,
                                bootstrap.base.transport_generation,
                                bootstrap.base.route_generation,
                                kind,
                                bootstrap.carrier.transport_id(),
                                authorization_expiry,
                            )
                            .await?;
                        let _ = attempt;
                        Ok::<(), anyhow::Error>(())
                    }
                    .await;
                    node.revoke_inbound_replacement_if_current(
                        endpoint_id,
                        provider.transport_id(),
                        authorization_expiry,
                    )
                    .await;
                    if let Err(error) = result {
                        let failure_code = native_iroh_carrier_failure_code(&error);
                        eprintln!(
                            "[OpenRTC][WebRTC carrier] responder completion failed connection_id={} upgrade_id={} failure_code={} error={:#}",
                            connection_id, bootstrap.upgrade_id, failure_code, error,
                        );
                        client
                            .fail_native_webrtc_carrier_attempt(
                                &connection_id,
                                &remote_node_id,
                                &bootstrap,
                                failure_code,
                                true,
                            )
                            .await;
                    } else {
                        remove_native_transport_upgrade_gate(
                            &client.native_transport_upgrade_gates,
                            &(connection_id.clone(), kind),
                            &bootstrap.upgrade_id,
                        )
                        .await;
                        if let Some(attempt) = client
                            .native_webrtc_carrier_attempts
                            .read()
                            .await
                            .get(&connection_id)
                            .cloned()
                        {
                            client.spawn_native_webrtc_carrier_terminal_watcher(
                                connection_id.clone(),
                                endpoint_id,
                                attempt,
                            );
                        }
                    }
                });
            }
            crate::iroh_carrier_bootstrap::CarrierBootstrapAction::Ready => {
                let attempt = self
                    .native_webrtc_carrier_attempts
                    .read()
                    .await
                    .get(connection_id)
                    .filter(|attempt| bootstrap.is_response_to(&attempt.bootstrap))
                    .cloned();
                let Some(attempt) = attempt else {
                    return true;
                };
                let client = self.clone();
                let connection_id = connection_id.to_string();
                let remote_node_id = remote_node_id.to_string();
                tokio::spawn(async move {
                    let result = async {
                        client
                            .wait_native_webrtc_carrier_connected(
                                &connection_id,
                                &bootstrap.upgrade_id,
                            )
                            .await?;
                        client
                            .complete_native_iroh_carrier_upgrade(
                                &connection_id,
                                &remote_node_id,
                                &bootstrap.upgrade_id,
                                attempt.generation,
                                kind,
                            )
                            .await
                    }
                    .await;
                    if let Err(error) = result {
                        let failure_code = native_iroh_carrier_failure_code(&error);
                        let error_detail = format!("{error:#}");
                        client.record_native_iroh_carrier_debug_event(
                            &connection_id,
                            &bootstrap.upgrade_id,
                            "completion-failed",
                            Some(&error_detail),
                        );
                        eprintln!(
                            "[OpenRTC][WebRTC carrier] initiator completion failed connection_id={} upgrade_id={} failure_code={} error={error:#}",
                            connection_id, bootstrap.upgrade_id, failure_code,
                        );
                        let retired = client
                            .fail_native_webrtc_carrier_attempt(
                                &connection_id,
                                &remote_node_id,
                                &attempt.bootstrap,
                                failure_code,
                                true,
                            )
                            .await;
                        if retired {
                            let _ = client.schedule_failed_webrtc_retry(
                                &connection_id,
                                &remote_node_id,
                                &attempt.bootstrap,
                                failure_code,
                            );
                        }
                    } else {
                        remove_native_transport_upgrade_gate(
                            &client.native_transport_upgrade_gates,
                            &(connection_id.clone(), kind),
                            &bootstrap.upgrade_id,
                        )
                        .await;
                        if let Some(current) = client
                            .native_webrtc_carrier_attempts
                            .read()
                            .await
                            .get(&connection_id)
                            .cloned()
                        {
                            client.spawn_native_webrtc_carrier_terminal_watcher(
                                connection_id.clone(),
                                endpoint_id,
                                current,
                            );
                        }
                    }
                });
            }
            crate::iroh_carrier_bootstrap::CarrierBootstrapAction::Failed => {
                let attempt = self
                    .native_webrtc_carrier_attempts
                    .read()
                    .await
                    .get(connection_id)
                    .filter(|attempt| bootstrap.is_response_to(&attempt.bootstrap))
                    .cloned();
                if let Some(attempt) = attempt {
                    let failure_code = match bootstrap.failure_code.as_deref() {
                        Some("carrier-base-generation-stale") => "carrier-base-generation-stale",
                        Some("carrier-start-failed") => "carrier-start-failed",
                        Some("carrier-timeout") => "carrier-timeout",
                        Some("carrier-proof-failed") => "carrier-proof-failed",
                        _ => "peer-rejected-carrier",
                    };
                    let retired = self
                        .fail_native_webrtc_carrier_attempt(
                            connection_id,
                            remote_node_id,
                            &attempt.bootstrap,
                            failure_code,
                            false,
                        )
                        .await;
                    if retired {
                        let _ = self.schedule_failed_webrtc_retry(
                            connection_id,
                            remote_node_id,
                            &attempt.bootstrap,
                            failure_code,
                        );
                    }
                }
            }
        }
        true
    }

    #[cfg(all(not(target_arch = "wasm32"), feature = "transport-moq"))]
    async fn create_native_moq_carrier_attempt(
        &self,
        connection_id: &str,
        remote_node_id: &str,
        bootstrap: crate::iroh_carrier_bootstrap::CarrierBootstrapFrame,
        generation: crate::client::NativePeerDataGeneration,
    ) -> anyhow::Result<Arc<crate::client::NativeMoqCarrierAttempt>> {
        let config = self
            .transport_config
            .read()
            .await
            .moq
            .clone()
            .ok_or_else(|| anyhow::anyhow!("MoQ carrier config is unavailable"))?;
        let local_node_id = self
            .current_node_id()
            .await
            .ok_or_else(|| anyhow::anyhow!("local Iroh node id is unavailable"))?;
        let remote_endpoint_id = remote_node_id.parse::<iroh::EndpointId>()?;
        // Retire the previous physical attempt before activating its
        // replacement. Packet-carrier activation is keyed by carrier kind and
        // endpoint, so deactivating the old attempt after activation would
        // accidentally remove the newly installed session.
        if let Some(previous) = self
            .native_moq_carrier_attempts
            .write()
            .await
            .remove(connection_id)
        {
            previous.session.close();
            let _ = previous.pump.lock().await.take();
            let _ = self
                .deactivate_native_iroh_packet_carrier(
                    IrohPathKind::Moq,
                    previous.remote_endpoint_id,
                )
                .await;
        }
        let packet_session = self
            .activate_native_iroh_packet_carrier(IrohPathKind::Moq, remote_endpoint_id)
            .await?;
        let session = Arc::new(
            crate::transport::NativeMoQSession::new_packet_carrier(
                &local_node_id,
                remote_node_id,
                &config,
                &bootstrap.carrier_session_id,
            )
            .await?,
        );
        // Start the relay session before attaching Iroh's outbound packet
        // consumer. Recovery and re-share can produce a QUIC Initial as soon
        // as the custom transport is activated; attaching first lets that
        // packet observe an Idle MoQ session and permanently exits the pump.
        if let Err(error) = session.start().await {
            let _ = self
                .deactivate_native_iroh_packet_carrier(IrohPathKind::Moq, remote_endpoint_id)
                .await;
            return Err(error);
        }
        let application_key = self
            .application_crypto_key_for_connection(Some(connection_id))
            .ok_or_else(|| anyhow::anyhow!("MoQ carrier requires an installed application key"))?;
        let pump = match crate::native_moq_carrier::NativeMoqCarrierSession::attach(
            session.clone(),
            packet_session,
            bootstrap.frame_expectation()?,
            application_key,
        )
        .await
        {
            Ok(pump) => pump,
            Err(error) => {
                session.close_gracefully().await;
                let _ = self
                    .deactivate_native_iroh_packet_carrier(IrohPathKind::Moq, remote_endpoint_id)
                    .await;
                return Err(error);
            }
        };
        let attempt = Arc::new(crate::client::NativeMoqCarrierAttempt {
            upgrade_id: bootstrap.upgrade_id.clone(),
            bootstrap,
            generation,
            remote_endpoint_id,
            session: session.clone(),
            pump: tokio::sync::Mutex::new(Some(pump)),
        });
        if let Some(previous) = self
            .native_moq_carrier_attempts
            .write()
            .await
            .insert(connection_id.to_string(), attempt.clone())
        {
            // A competing creator should be fenced by the upgrade gate. If
            // one still wins the map race, close only its mechanism; the
            // provider now belongs to the attempt inserted above.
            previous.session.close();
            let _ = previous.pump.lock().await.take();
        }
        Ok(attempt)
    }

    #[cfg(all(not(target_arch = "wasm32"), feature = "transport-moq"))]
    async fn wait_native_moq_carrier_ready(
        &self,
        connection_id: &str,
        upgrade_id: &str,
    ) -> anyhow::Result<Arc<crate::client::NativeMoqCarrierAttempt>> {
        let deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(25);
        loop {
            let attempt = self
                .native_moq_carrier_attempts
                .read()
                .await
                .get(connection_id)
                .filter(|attempt| attempt.upgrade_id == upgrade_id)
                .cloned()
                .ok_or_else(|| anyhow::anyhow!("MoQ carrier attempt was retired"))?;
            match attempt.session.state() {
                crate::transport::NativeMoQState::Connected
                    if attempt.session.is_peer_data_bidirectionally_ready().await =>
                {
                    // The protected Ready exchange proves that the peer has
                    // attached its packet pumps. Native still waits for the
                    // relay to acknowledge both directed tracks before dialing
                    // Iroh so the first request and response cannot race a
                    // one-sided MoQ publication.
                    return Ok(attempt);
                }
                crate::transport::NativeMoQState::Failed
                | crate::transport::NativeMoQState::Closed => {
                    anyhow::bail!(
                        "MoQ Draft 14 reliable object-stream carrier negotiation failed: {}",
                        attempt.session.peer_data_readiness_diagnostic().await,
                    )
                }
                _ if tokio::time::Instant::now() >= deadline => {
                    anyhow::bail!(
                        "MoQ Draft 14 reliable object-stream carrier negotiation timed out: {}",
                        attempt.session.peer_data_readiness_diagnostic().await,
                    )
                }
                _ => tokio::time::sleep(std::time::Duration::from_millis(25)).await,
            }
        }
    }

    #[cfg(all(not(target_arch = "wasm32"), feature = "transport-moq"))]
    fn spawn_native_moq_carrier_terminal_watcher(
        &self,
        connection_id: String,
        attempt: Arc<crate::client::NativeMoqCarrierAttempt>,
    ) {
        let client = self.clone();
        tokio::spawn(async move {
            loop {
                let current = client
                    .native_moq_carrier_attempts
                    .read()
                    .await
                    .get(&connection_id)
                    .is_some_and(|value| Arc::ptr_eq(value, &attempt));
                if !current {
                    return;
                }
                if matches!(
                    attempt.session.state(),
                    crate::transport::NativeMoQState::Failed
                        | crate::transport::NativeMoQState::Closed
                ) {
                    let _ = client
                        .close_current_iroh_carrier_generation_with_reason(
                            &connection_id,
                            attempt.remote_endpoint_id,
                            IrohPathKind::Moq,
                            attempt.generation.transport_generation.saturating_add(1),
                            "iroh-carrier-ended",
                        )
                        .await;
                    return;
                }
                let terminal_reason = attempt
                    .pump
                    .lock()
                    .await
                    .as_ref()
                    .and_then(|pump| pump.take_terminal_reason());
                if let Some(reason) = terminal_reason {
                    let _ = client
                        .close_current_iroh_carrier_generation_with_reason(
                            &connection_id,
                            attempt.remote_endpoint_id,
                            IrohPathKind::Moq,
                            attempt.generation.transport_generation.saturating_add(1),
                            reason,
                        )
                        .await;
                    return;
                }
                tokio::time::sleep(std::time::Duration::from_millis(250)).await;
            }
        });
    }

    #[cfg(all(not(target_arch = "wasm32"), feature = "transport-moq"))]
    async fn retire_native_moq_carrier_attempt(&self, connection_id: &str, upgrade_id: &str) {
        let removed = {
            let mut attempts = self.native_moq_carrier_attempts.write().await;
            if attempts
                .get(connection_id)
                .is_some_and(|attempt| attempt.upgrade_id == upgrade_id)
            {
                attempts.remove(connection_id)
            } else {
                None
            }
        };
        if let Some(attempt) = removed {
            attempt.session.close();
            let _ = attempt.pump.lock().await.take();
            let _ = self
                .deactivate_native_iroh_packet_carrier(
                    IrohPathKind::Moq,
                    attempt.remote_endpoint_id,
                )
                .await;
        }
    }

    #[cfg(all(not(target_arch = "wasm32"), feature = "transport-moq"))]
    async fn fail_native_moq_carrier_attempt(
        &self,
        connection_id: &str,
        remote_node_id: &str,
        request: &crate::iroh_carrier_bootstrap::CarrierBootstrapFrame,
        failure_code: &'static str,
        notify_peer: bool,
    ) {
        let retry_pending = moq_retry_attempt(failure_code, request.attempt).is_some();
        self.retire_native_moq_carrier_attempt(connection_id, &request.upgrade_id)
            .await;
        remove_native_transport_upgrade_gate(
            &self.native_transport_upgrade_gates,
            &(connection_id.to_string(), IrohPathKind::Moq),
            &request.upgrade_id,
        )
        .await;
        if notify_peer {
            if let Ok(failed) = crate::iroh_carrier_bootstrap::CarrierBootstrapFrame::failed_from(
                request,
                failure_code,
            ) {
                let _ = self.send_iroh_carrier_control(connection_id, &failed).await;
            }
        }
        eprintln!(
            "[OpenRTC][MoQ carrier] attempt failed connection_id={} upgrade_id={} failure_code={}",
            connection_id, request.upgrade_id, failure_code,
        );
        if !retry_pending {
            let _ = self
                .maybe_start_next_native_iroh_carrier(
                    connection_id,
                    remote_node_id,
                    crate::route_policy::KnownRoute::Moq,
                )
                .await;
        }
    }

    #[cfg(all(not(target_arch = "wasm32"), feature = "transport-moq"))]
    async fn retry_stale_moq(
        &self,
        connection_id: &str,
        remote_node_id: &str,
        request: &crate::iroh_carrier_bootstrap::CarrierBootstrapFrame,
        failure_code: &str,
    ) {
        let Some(attempt) = moq_retry_attempt(failure_code, request.attempt) else {
            return;
        };
        // An admitted replacement can supersede the base while the carrier
        // proves itself. Retry once on that new generation.
        tokio::time::sleep(std::time::Duration::from_millis(500)).await;
        if let Err(error) = self
            .maybe_start_native_moq_iroh_carrier_attempt(
                connection_id,
                Some(remote_node_id),
                attempt,
            )
            .await
        {
            eprintln!(
                "[OpenRTC][MoQ carrier] stale-base retry failed connection_id={} attempt={} error={error:#}",
                connection_id, attempt,
            );
        }
    }

    #[cfg(all(not(target_arch = "wasm32"), feature = "transport-moq"))]
    pub(super) async fn maybe_start_native_moq_iroh_carrier(
        &self,
        connection_id: &str,
        remote_node_id: Option<&str>,
    ) -> anyhow::Result<()> {
        self.maybe_start_native_moq_iroh_carrier_attempt(connection_id, remote_node_id, 1)
            .await
    }

    #[cfg(all(not(target_arch = "wasm32"), feature = "transport-moq"))]
    async fn maybe_start_native_moq_iroh_carrier_attempt(
        &self,
        connection_id: &str,
        remote_node_id: Option<&str>,
        attempt_number: u8,
    ) -> anyhow::Result<()> {
        if !self.is_moq_carrier_enabled().await {
            return Ok(());
        }
        let remote_node_id = remote_node_id
            .map(str::trim)
            .filter(|value| !value.is_empty())
            .ok_or_else(|| anyhow::anyhow!("MoQ carrier requires remote node id"))?;
        let local_node_id = self
            .current_node_id()
            .await
            .ok_or_else(|| anyhow::anyhow!("MoQ carrier requires local node id"))?;
        if local_node_id.as_str() <= remote_node_id {
            return Ok(());
        }
        let endpoint_id = remote_node_id.parse::<iroh::EndpointId>()?;
        let generation = self
            .current_native_peer_data_generation(connection_id, None)
            .await
            .ok_or_else(|| anyhow::anyhow!("MoQ carrier generation is unavailable"))?;
        let selected_attempt = self
            .native_moq_carrier_attempts
            .read()
            .await
            .get(connection_id)
            .filter(|attempt| {
                attempt.remote_endpoint_id == endpoint_id
                    && attempt.session.state() == crate::transport::NativeMoQState::Connected
                    && attempt.generation.transport_generation.saturating_add(1)
                        == generation.transport_generation
            })
            .cloned();
        let selected_record = self
            .connection_manager
            .get_by_connection_id(connection_id)
            .await
            .is_some_and(|record| {
                record.transport_stable_id == Some(generation.transport_stable_id)
                    && record.active_transport == crate::transport_label::MOQ
            });
        if selected_attempt.is_some() && selected_record {
            return Ok(());
        }
        // The path observation can briefly remain `Moq` after attempt 1
        // is retired for a replaced base generation. Only the bounded stale-
        // base retry may cross that transient observation; ordinary starts
        // still require a relay or unknown Iroh base path.
        if !crate::iroh_connection_policy::custom_carrier_base_allows(
            self.iroh_path_kind(remote_node_id).await,
            IrohPathKind::Moq,
            attempt_number == 2,
        ) {
            return Ok(());
        }
        self.get_connection(endpoint_id)
            .await
            .ok_or_else(|| anyhow::anyhow!("MoQ carrier base is unavailable"))?;
        let gate_key = (connection_id.to_string(), IrohPathKind::Moq);
        let bootstrap = crate::iroh_carrier_bootstrap::CarrierBootstrapFrame::request(
            crate::iroh_carrier_bootstrap::CarrierBootstrapKind::MoqDraft14,
            generation.into(),
            attempt_number,
        )?;
        {
            let mut gates = self.native_transport_upgrade_gates.lock().await;
            if !reserve_native_transport_upgrade_gate(
                &mut gates,
                gate_key.clone(),
                &bootstrap.upgrade_id,
                generation,
                self.current_iroh_carrier_policy_epoch(),
                self.current_iroh_carrier_peer_policy_epoch(connection_id),
                false,
            ) {
                return Ok(());
            }
        }
        let result = async {
            self.create_native_moq_carrier_attempt(
                connection_id,
                remote_node_id,
                bootstrap.clone(),
                generation,
            )
            .await?;
            self.send_iroh_carrier_control(connection_id, &bootstrap)
                .await?;
            Ok::<(), anyhow::Error>(())
        }
        .await;
        if let Err(error) = result {
            self.fail_native_moq_carrier_attempt(
                connection_id,
                remote_node_id,
                &bootstrap,
                "carrier-start-failed",
                true,
            )
            .await;
            return Err(error);
        }
        let client = self.clone();
        let connection_id = connection_id.to_string();
        let remote_node_id = remote_node_id.to_string();
        tokio::spawn(async move {
            tokio::time::sleep(std::time::Duration::from_secs(30)).await;
            let still_current = client
                .native_transport_upgrade_gates
                .lock()
                .await
                .get(&(connection_id.clone(), IrohPathKind::Moq))
                .is_some_and(|gate| {
                    native_transport_upgrade_gate_matches(gate, &bootstrap.upgrade_id)
                });
            if still_current {
                client
                    .fail_native_moq_carrier_attempt(
                        &connection_id,
                        &remote_node_id,
                        &bootstrap,
                        "carrier-timeout",
                        true,
                    )
                    .await;
            }
        });
        Ok(())
    }

    #[cfg(all(not(target_arch = "wasm32"), feature = "transport-moq"))]
    async fn handle_native_moq_iroh_carrier_control(
        &self,
        connection_id: &str,
        remote_node_id: Option<&str>,
        frame: &serde_json::Value,
    ) -> bool {
        let Some(bootstrap) =
            crate::iroh_carrier_bootstrap::CarrierBootstrapFrame::from_json(frame)
        else {
            if frame.get("type").and_then(serde_json::Value::as_str)
                == Some(crate::iroh_carrier_bootstrap::CARRIER_BOOTSTRAP_FRAME_TYPE)
            {
                eprintln!("[OpenRTC][MoQ carrier] rejected invalid bootstrap envelope={frame}");
            }
            return false;
        };
        if bootstrap.carrier != crate::iroh_carrier_bootstrap::CarrierBootstrapKind::MoqDraft14 {
            return false;
        }
        let Some(remote_node_id) = remote_node_id
            .map(str::trim)
            .filter(|value| !value.is_empty())
        else {
            return true;
        };
        let endpoint_id = match remote_node_id.parse::<iroh::EndpointId>() {
            Ok(value) => value,
            Err(_) => return true,
        };
        match bootstrap.action {
            crate::iroh_carrier_bootstrap::CarrierBootstrapAction::Request => {
                let local_node_id = match self.current_node_id().await {
                    Some(value) => value,
                    None => return true,
                };
                if local_node_id.as_str() >= remote_node_id || !self.is_moq_carrier_enabled().await
                {
                    return true;
                }
                let duplicate_attempt = self
                    .native_moq_carrier_attempts
                    .read()
                    .await
                    .get(connection_id)
                    .filter(|attempt| {
                        native_carrier_duplicate_request_is_live(
                            &bootstrap,
                            &attempt.bootstrap,
                            matches!(
                                attempt.session.state(),
                                crate::transport::NativeMoQState::Failed
                                    | crate::transport::NativeMoQState::Closed
                            ),
                        )
                    })
                    .cloned();
                if let Some(attempt) = duplicate_attempt {
                    if attempt.session.state() == crate::transport::NativeMoQState::Connected
                        && attempt.session.is_peer_data_subscribed().await
                    {
                        if let Ok(ready) =
                            crate::iroh_carrier_bootstrap::CarrierBootstrapFrame::ready_from(
                                &bootstrap,
                            )
                        {
                            let _ = self.send_iroh_carrier_control(connection_id, &ready).await;
                        }
                    }
                    eprintln!("[OpenRTC][MoQ carrier] responder reused connection_id={connection_id} reason=duplicate-live-bootstrap");
                    return true;
                }
                if !crate::iroh_connection_policy::custom_carrier_base_allows(
                    self.iroh_path_kind(remote_node_id).await,
                    IrohPathKind::Moq,
                    false,
                ) {
                    return true;
                }
                if self.get_connection(endpoint_id).await.is_none() {
                    return true;
                }
                let generation = match self
                    .current_native_peer_data_generation(connection_id, None)
                    .await
                {
                    Some(value) => value,
                    None => return true,
                };
                if self
                    .reject_native_iroh_carrier_request_if_base_blocked(
                        connection_id,
                        &bootstrap,
                        generation.transport_stable_id,
                        "MoQ",
                    )
                    .await
                {
                    return true;
                }
                let node = match self.iroh_node.read().await.as_ref().cloned() {
                    Some(value) => value,
                    None => return true,
                };
                let provider = match self
                    .native_transport_upgrade_provider(IrohPathKind::Moq)
                    .await
                {
                    Some(value) => value,
                    None => return true,
                };
                let gate_key = (connection_id.to_string(), IrohPathKind::Moq);
                {
                    let mut gates = self.native_transport_upgrade_gates.lock().await;
                    // Gateway/control replay is at-least-once. A matching gate
                    // means this exact physical attempt already owns creation;
                    // accepting it again would attach the same packet session
                    // twice and reject the valid attempt as DuplicateSession.
                    if gates.get(&gate_key).is_some_and(|current| {
                        current.generation == generation
                            && self.iroh_carrier_policy_epoch_is_current(current.policy_epoch)
                            && current.peer_policy_epoch
                                == self.current_iroh_carrier_peer_policy_epoch(connection_id)
                    }) {
                        return true;
                    }
                    if !reserve_native_transport_upgrade_gate(
                        &mut gates,
                        gate_key.clone(),
                        &bootstrap.upgrade_id,
                        generation,
                        self.current_iroh_carrier_policy_epoch(),
                        self.current_iroh_carrier_peer_policy_epoch(connection_id),
                        true,
                    ) {
                        return true;
                    }
                }
                let authorization_expiry = node
                    .authorize_inbound_replacement(
                        endpoint_id,
                        provider.transport_id(),
                        std::time::Duration::from_secs(45),
                    )
                    .await;
                let client = self.clone();
                let connection_id = connection_id.to_string();
                let remote_node_id = remote_node_id.to_string();
                tokio::spawn(async move {
                    let result = async {
                        client
                            .create_native_moq_carrier_attempt(
                                &connection_id,
                                &remote_node_id,
                                bootstrap.clone(),
                                generation,
                            )
                            .await?;
                        let ready =
                            crate::iroh_carrier_bootstrap::CarrierBootstrapFrame::ready_from(
                                &bootstrap,
                            )?;
                        client
                            .send_iroh_carrier_control(&connection_id, &ready)
                            .await?;
                        client
                            .wait_native_moq_carrier_ready(&connection_id, &bootstrap.upgrade_id)
                            .await?;
                        client
                            .complete_inbound_native_iroh_carrier_upgrade(
                                &connection_id,
                                &remote_node_id,
                                &bootstrap.upgrade_id,
                                generation,
                                bootstrap.base.transport_generation,
                                bootstrap.base.route_generation,
                                IrohPathKind::Moq,
                                bootstrap.carrier.transport_id(),
                                authorization_expiry,
                            )
                            .await
                    }
                    .await;
                    node.revoke_inbound_replacement_if_current(
                        endpoint_id,
                        provider.transport_id(),
                        authorization_expiry,
                    )
                    .await;
                    if let Err(error) = result {
                        let failure_code = native_iroh_carrier_failure_code(&error);
                        let error_detail = format!("{error:#}");
                        client.record_native_iroh_carrier_debug_event(
                            &connection_id,
                            &bootstrap.upgrade_id,
                            "completion-failed",
                            Some(&error_detail),
                        );
                        eprintln!(
                            "[OpenRTC][MoQ carrier] responder completion failed connection_id={} upgrade_id={} failure_code={} error={error:#}",
                            connection_id, bootstrap.upgrade_id, failure_code,
                        );
                        client
                            .fail_native_moq_carrier_attempt(
                                &connection_id,
                                &remote_node_id,
                                &bootstrap,
                                failure_code,
                                true,
                            )
                            .await;
                        client
                            .retry_stale_moq(
                                &connection_id,
                                &remote_node_id,
                                &bootstrap,
                                failure_code,
                            )
                            .await;
                    } else {
                        remove_native_transport_upgrade_gate(
                            &client.native_transport_upgrade_gates,
                            &(connection_id.clone(), IrohPathKind::Moq),
                            &bootstrap.upgrade_id,
                        )
                        .await;
                        if let Some(attempt) = client
                            .native_moq_carrier_attempts
                            .read()
                            .await
                            .get(&connection_id)
                            .cloned()
                        {
                            client.spawn_native_moq_carrier_terminal_watcher(
                                connection_id.clone(),
                                attempt,
                            );
                        }
                    }
                });
            }
            crate::iroh_carrier_bootstrap::CarrierBootstrapAction::Ready => {
                let attempt = self
                    .native_moq_carrier_attempts
                    .read()
                    .await
                    .get(connection_id)
                    .filter(|attempt| bootstrap.is_response_to(&attempt.bootstrap))
                    .cloned();
                let Some(attempt) = attempt else {
                    return true;
                };
                let client = self.clone();
                let connection_id = connection_id.to_string();
                let remote_node_id = remote_node_id.to_string();
                tokio::spawn(async move {
                    let result = async {
                        client
                            .wait_native_moq_carrier_ready(&connection_id, &bootstrap.upgrade_id)
                            .await?;
                        client
                            .complete_native_iroh_carrier_upgrade(
                                &connection_id,
                                &remote_node_id,
                                &bootstrap.upgrade_id,
                                attempt.generation,
                                IrohPathKind::Moq,
                            )
                            .await
                    }
                    .await;
                    if let Err(error) = result {
                        let failure_code = native_iroh_carrier_failure_code(&error);
                        let error_detail = format!("{error:#}");
                        client.record_native_iroh_carrier_debug_event(
                            &connection_id,
                            &bootstrap.upgrade_id,
                            "completion-failed",
                            Some(&error_detail),
                        );
                        eprintln!(
                            "[OpenRTC][MoQ carrier] initiator completion failed connection_id={} upgrade_id={} failure_code={} error={error:#}",
                            connection_id, bootstrap.upgrade_id, failure_code,
                        );
                        client
                            .fail_native_moq_carrier_attempt(
                                &connection_id,
                                &remote_node_id,
                                &attempt.bootstrap,
                                failure_code,
                                true,
                            )
                            .await;
                        client
                            .retry_stale_moq(
                                &connection_id,
                                &remote_node_id,
                                &attempt.bootstrap,
                                failure_code,
                            )
                            .await;
                    } else {
                        remove_native_transport_upgrade_gate(
                            &client.native_transport_upgrade_gates,
                            &(connection_id.clone(), IrohPathKind::Moq),
                            &bootstrap.upgrade_id,
                        )
                        .await;
                        if let Some(current) = client
                            .native_moq_carrier_attempts
                            .read()
                            .await
                            .get(&connection_id)
                            .cloned()
                        {
                            client.spawn_native_moq_carrier_terminal_watcher(
                                connection_id.clone(),
                                current,
                            );
                        }
                    }
                });
            }
            crate::iroh_carrier_bootstrap::CarrierBootstrapAction::Failed => {
                let request = self
                    .native_moq_carrier_attempts
                    .read()
                    .await
                    .get(connection_id)
                    .filter(|attempt| bootstrap.is_response_to(&attempt.bootstrap))
                    .map(|attempt| attempt.bootstrap.clone());
                if let Some(request) = request {
                    let (peer_failure_code, failure_code) =
                        native_moq_peer_failure_code(bootstrap.failure_code.as_deref());
                    self.record_native_iroh_carrier_debug_event(
                        connection_id,
                        &bootstrap.upgrade_id,
                        "peer-failed",
                        Some(peer_failure_code),
                    );
                    self.fail_native_moq_carrier_attempt(
                        connection_id,
                        remote_node_id,
                        &request,
                        failure_code,
                        false,
                    )
                    .await;
                    self.retry_stale_moq(connection_id, remote_node_id, &request, failure_code)
                        .await;
                }
            }
        }
        true
    }

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

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

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

    #[cfg(not(target_arch = "wasm32"))]
    async fn reserve_native_ble_upgrade_attempt(
        &self,
        connection_id: &str,
        generation: crate::client::NativePeerDataGeneration,
    ) -> Option<u8> {
        let mut attempts = self.native_ble_upgrade_attempts.lock().await;
        let (state, attempt) =
            reserve_native_ble_attempt_state(attempts.get(connection_id).copied(), generation);
        attempts.insert(connection_id.to_string(), state);
        attempt
    }

    #[cfg(not(target_arch = "wasm32"))]
    async fn current_native_ble_upgrade_attempt(
        &self,
        connection_id: &str,
        generation: crate::client::NativePeerDataGeneration,
    ) -> Option<u8> {
        self.native_ble_upgrade_attempts
            .lock()
            .await
            .get(connection_id)
            .filter(|state| state.generation == generation)
            .map(|state| state.attempts)
    }

    #[cfg(not(target_arch = "wasm32"))]
    async fn clear_native_ble_upgrade_attempts(
        &self,
        connection_id: &str,
        generation: crate::client::NativePeerDataGeneration,
    ) {
        let mut attempts = self.native_ble_upgrade_attempts.lock().await;
        if attempts
            .get(connection_id)
            .is_some_and(|state| state.generation == generation)
        {
            attempts.remove(connection_id);
        }
    }

    #[cfg(not(target_arch = "wasm32"))]
    pub(crate) async fn wake_native_ble_recovery(&self, reason: &'static str) -> usize {
        if !self.is_ble_transport_enabled().await {
            return 0;
        }

        let active_gates = self
            .native_transport_upgrade_gates
            .lock()
            .await
            .keys()
            .filter(|(_, kind)| *kind == IrohPathKind::Ble)
            .map(|(connection_id, _)| connection_id.clone())
            .collect::<std::collections::HashSet<_>>();
        let reset_count = {
            let mut attempts = self.native_ble_upgrade_attempts.lock().await;
            let before = attempts.len();
            attempts.retain(|connection_id, state| {
                native_ble_attempt_state_after_recovery_wake(
                    *state,
                    active_gates.contains(connection_id),
                )
                .is_some()
            });
            before.saturating_sub(attempts.len())
        };

        let ble_capable_connections = self
            .native_peer_transport_capabilities
            .read()
            .await
            .iter()
            .filter(|(_, capabilities)| capabilities.contains(&NativePeerTransportCapability::Ble))
            .map(|(connection_id, _)| connection_id.clone())
            .collect::<std::collections::HashSet<_>>();
        let mut eligible_records = 0usize;
        for record in self.connection_manager.list_all().await {
            if !ble_capable_connections.contains(&record.connection_id)
                || !matches!(
                    record.state,
                    crate::connection_manager::ConnectionState::Connecting
                        | crate::connection_manager::ConnectionState::Connected
                )
            {
                continue;
            }
            let Some(remote_node_id) = record
                .endpoint_id
                .as_deref()
                .or(record.node_id.as_deref())
                .map(str::trim)
                .filter(|value| !value.is_empty())
            else {
                continue;
            };
            eligible_records = eligible_records.saturating_add(1);
            if let Err(error) = self
                .maybe_start_native_ble_upgrade(&record.connection_id, Some(remote_node_id))
                .await
            {
                eprintln!(
                    "[OpenRTC][BLE] recovery wake failed connection_id={} remote_node_id={} reason={} error={:#}",
                    record.connection_id, remote_node_id, reason, error,
                );
            }
        }

        println!(
            "[OpenRTC][BLE] recovery wake reason={} reset_attempt_budgets={} eligible_records={}",
            reason, reset_count, eligible_records,
        );
        eligible_records
    }

    #[cfg(not(target_arch = "wasm32"))]
    fn schedule_native_ble_upgrade_retry(
        &self,
        connection_id: String,
        remote_node_id: String,
        expected_generation: crate::client::NativePeerDataGeneration,
        failed_attempt: u8,
        reason: &'static str,
    ) {
        let Some(delay) =
            native_ble_retry_delay(failed_attempt, uuid::Uuid::new_v4().as_u128() as u64)
        else {
            println!(
                "[OpenRTC][BLE] retry exhausted connection_id={} remote_node_id={} attempts={} reason={}",
                connection_id, remote_node_id, failed_attempt, reason,
            );
            return;
        };
        let client = self.clone();
        tokio::spawn(async move {
            tokio::time::sleep(delay).await;
            let attempt_is_current = client
                .current_native_ble_upgrade_attempt(&connection_id, expected_generation)
                .await
                == Some(failed_attempt);
            let generation_is_current = client
                .current_native_peer_data_generation(
                    &connection_id,
                    Some(expected_generation.transport_stable_id),
                )
                .await
                == Some(expected_generation);
            if !attempt_is_current || !generation_is_current {
                return;
            }
            println!(
                "[OpenRTC][BLE] retrying upgrade connection_id={} remote_node_id={} next_attempt={} reason={}",
                connection_id,
                remote_node_id,
                failed_attempt.saturating_add(1),
                reason,
            );
            if let Err(error) = client
                .maybe_start_native_ble_upgrade(&connection_id, Some(&remote_node_id))
                .await
            {
                eprintln!(
                    "[OpenRTC][BLE] retry setup failed connection_id={} remote_node_id={} error={}",
                    connection_id, remote_node_id, error,
                );
            }
        });
    }

    #[cfg(all(not(target_arch = "wasm32"), feature = "iroh-carrier-core"))]
    async fn complete_inbound_native_ble_upgrade(
        &self,
        connection_id: &str,
        remote_node_id: &str,
        upgrade_id: &str,
        expected_generation: crate::client::NativePeerDataGeneration,
        proof_generation: crate::client::NativePeerDataGeneration,
        replacement_transport_id: u64,
        authorization_expires_at: std::time::Instant,
    ) -> anyhow::Result<()> {
        self.complete_inbound_native_iroh_carrier_upgrade(
            connection_id,
            remote_node_id,
            upgrade_id,
            expected_generation,
            proof_generation.transport_generation,
            proof_generation.route_generation,
            IrohPathKind::Ble,
            replacement_transport_id,
            authorization_expires_at,
        )
        .await
    }

    #[cfg(all(not(target_arch = "wasm32"), not(feature = "iroh-carrier-core")))]
    async fn complete_inbound_native_ble_upgrade(
        &self,
        _connection_id: &str,
        _remote_node_id: &str,
        _upgrade_id: &str,
        _expected_generation: crate::client::NativePeerDataGeneration,
        _proof_generation: crate::client::NativePeerDataGeneration,
        _replacement_transport_id: u64,
        _authorization_expires_at: std::time::Instant,
    ) -> anyhow::Result<()> {
        anyhow::bail!("BLE replacement requires the Iroh custom-carrier core")
    }

    #[cfg(not(target_arch = "wasm32"))]
    async fn accept_native_ble_switch_request(
        &self,
        connection_id: &str,
        remote_node_id: &str,
        upgrade_id: &str,
        expected_generation: crate::client::NativePeerDataGeneration,
        proof_generation: crate::client::NativePeerDataGeneration,
    ) -> anyhow::Result<bool> {
        let Some(local_node_id) = self.current_node_id().await else {
            return Ok(false);
        };
        let current_path = self.iroh_path_kind(remote_node_id).await;
        if !should_accept_native_ble_switch_request(&local_node_id, remote_node_id, current_path) {
            return Ok(false);
        }
        let remote_endpoint_id = remote_node_id.parse::<iroh::EndpointId>()?;
        let provider = self
            .native_transport_upgrade_provider(IrohPathKind::Ble)
            .await
            .ok_or_else(|| anyhow::anyhow!("BLE upgrade provider is unavailable"))?;
        let replacement_transport_id = provider.transport_id();
        let native_node = self
            .iroh_node
            .read()
            .await
            .as_ref()
            .cloned()
            .ok_or_else(|| anyhow::anyhow!("Iroh node is unavailable"))?;

        let gate_key = (connection_id.to_string(), IrohPathKind::Ble);
        {
            let mut gates = self.native_transport_upgrade_gates.lock().await;
            if !reserve_native_transport_upgrade_gate(
                &mut gates,
                gate_key.clone(),
                upgrade_id,
                expected_generation,
                self.current_iroh_carrier_policy_epoch(),
                self.current_iroh_carrier_peer_policy_epoch(connection_id),
                true,
            ) {
                return Ok(false);
            }
        }
        let authorization_expires_at = native_node
            .authorize_inbound_replacement(
                remote_endpoint_id,
                replacement_transport_id,
                std::time::Duration::from_secs(45),
            )
            .await;

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

        println!(
            "[OpenRTC][BLE] switch request accepted connection_id={} remote_node_id={} upgrade_id={}",
            connection_id, remote_node_id, upgrade_id,
        );
        let client = self.clone();
        let responder_upgrade_id = upgrade_id.to_string();
        let responder_connection_id = connection_id.to_string();
        let responder_remote_node_id = remote_node_id.to_string();
        tokio::spawn(async move {
            let result = client
                .complete_inbound_native_ble_upgrade(
                    &responder_connection_id,
                    &responder_remote_node_id,
                    &responder_upgrade_id,
                    expected_generation,
                    proof_generation,
                    replacement_transport_id,
                    authorization_expires_at,
                )
                .await;
            native_node
                .revoke_inbound_replacement_if_current(
                    remote_endpoint_id,
                    replacement_transport_id,
                    authorization_expires_at,
                )
                .await;
            remove_native_transport_upgrade_gate(
                &client.native_transport_upgrade_gates,
                &gate_key,
                &responder_upgrade_id,
            )
            .await;
            if let Err(error) = result {
                eprintln!(
                    "[OpenRTC][BLE] inbound replacement completion failed connection_id={} remote_node_id={} upgrade_id={} error={:#}",
                    responder_connection_id,
                    responder_remote_node_id,
                    responder_upgrade_id,
                    error,
                );
            }
        });
        Ok(true)
    }

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

        let endpoint_id = remote_node_id.parse::<iroh::EndpointId>()?;
        let transport_stable_id = self
            .get_connection(endpoint_id)
            .await
            .map(|connection| crate::transport_generation::for_connection(&connection))
            .ok_or_else(|| anyhow::anyhow!("BLE upgrade requires a live base transport"))?;
        let expected_generation = self
            .current_native_peer_data_generation(connection_id, Some(transport_stable_id))
            .await
            .ok_or_else(|| anyhow::anyhow!("BLE upgrade requires a current peer generation"))?;

        let gate_key = (connection_id.to_string(), IrohPathKind::Ble);
        let upgrade_id = uuid::Uuid::new_v4().to_string();
        {
            let mut gates = self.native_transport_upgrade_gates.lock().await;
            if !reserve_native_transport_upgrade_gate(
                &mut gates,
                gate_key.clone(),
                &upgrade_id,
                expected_generation,
                self.current_iroh_carrier_policy_epoch(),
                self.current_iroh_carrier_peer_policy_epoch(connection_id),
                false,
            ) {
                return Ok(());
            }
        }
        let Some(attempt) = self
            .reserve_native_ble_upgrade_attempt(connection_id, expected_generation)
            .await
        else {
            remove_native_transport_upgrade_gate(
                &self.native_transport_upgrade_gates,
                &gate_key,
                &upgrade_id,
            )
            .await;
            println!(
                "[OpenRTC][BLE] upgrade skipped connection_id={} remote_node_id={} reason=attempts-exhausted",
                connection_id, remote_node_id,
            );
            return Ok(());
        };

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

        if result.is_err() {
            let removed = remove_native_transport_upgrade_gate(
                &self.native_transport_upgrade_gates,
                &gate_key,
                &upgrade_id,
            )
            .await;
            if removed {
                self.schedule_native_ble_upgrade_retry(
                    connection_id.to_string(),
                    remote_node_id.to_string(),
                    expected_generation,
                    attempt,
                    "setup-failed",
                );
            }
        } else {
            let client = self.clone();
            let gates = self.native_transport_upgrade_gates.clone();
            let connection_id = connection_id.to_string();
            let remote_node_id = remote_node_id.to_string();
            tokio::spawn(async move {
                tokio::time::sleep(NATIVE_BLE_SWITCH_RESPONSE_TIMEOUT).await;
                if remove_native_transport_upgrade_gate(&gates, &gate_key, &upgrade_id).await {
                    client.schedule_native_ble_upgrade_retry(
                        connection_id,
                        remote_node_id,
                        expected_generation,
                        attempt,
                        "switch-response-timeout",
                    );
                }
            });
        }
        result
    }

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

        let Some(remote_node_id) = remote_node_id
            .map(str::trim)
            .filter(|value| !value.is_empty())
        else {
            return true;
        };
        let action = frame
            .get("action")
            .and_then(|value| value.as_str())
            .unwrap_or_default();
        let Some(upgrade_id) = frame
            .get("upgradeId")
            .and_then(|value| value.as_str())
            .map(str::trim)
            .filter(|value| !value.is_empty())
            .filter(|value| {
                (8..=96).contains(&value.len())
                    && value
                        .bytes()
                        .all(|byte| byte.is_ascii_alphanumeric() || byte == b'-')
            })
        else {
            eprintln!("[OpenRTC][BLE] ignored control frame with invalid upgrade id");
            return true;
        };
        let Some(proof_generation) = frame
            .get("base")
            .cloned()
            .and_then(|value| serde_json::from_value(value).ok())
            .filter(|generation: &crate::client::NativePeerDataGeneration| {
                generation.transport_stable_id != 0 && generation.transport_generation != 0
            })
        else {
            eprintln!(
                "[OpenRTC][BLE] ignored control frame without a valid base-generation fence upgrade_id={}",
                upgrade_id,
            );
            return true;
        };

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

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

        match action {
            "switch-request" => {
                if !should_accept_native_ble_switch_request(
                    &local_node_id,
                    remote_node_id,
                    current_path,
                ) {
                    return true;
                }
                let Ok(endpoint_id) = remote_node_id.parse::<iroh::EndpointId>() else {
                    return true;
                };
                let Some(transport_stable_id) = self
                    .get_connection(endpoint_id)
                    .await
                    .map(|connection| crate::transport_generation::for_connection(&connection))
                else {
                    return true;
                };
                let Some(expected_generation) = self
                    .current_native_peer_data_generation(connection_id, Some(transport_stable_id))
                    .await
                else {
                    return true;
                };
                let admission_block = self.session_admission_block_reason_for_transport(
                    connection_id,
                    Some(transport_stable_id),
                );
                match classify_native_ble_switch_admission(admission_block.as_ref()) {
                    NativeBleSwitchAdmission::Reject => {
                        println!(
                            "[OpenRTC][BLE] switch request rejected connection_id={} remote_node_id={} upgrade_id={} reason={}",
                            connection_id,
                            remote_node_id,
                            upgrade_id,
                            admission_block
                                .as_ref()
                                .map(|(_, reason)| reason.as_str())
                                .unwrap_or("unknown"),
                        );
                        return true;
                    }
                    NativeBleSwitchAdmission::Wait => {
                        let gate_key = (connection_id.to_string(), IrohPathKind::Ble);
                        {
                            let mut gates = self.native_transport_upgrade_gates.lock().await;
                            if !reserve_native_transport_upgrade_gate(
                                &mut gates,
                                gate_key.clone(),
                                upgrade_id,
                                expected_generation,
                                self.current_iroh_carrier_policy_epoch(),
                                self.current_iroh_carrier_peer_policy_epoch(connection_id),
                                false,
                            ) {
                                return true;
                            }
                        }
                        println!(
                            "[OpenRTC][BLE] switch request deferred connection_id={} remote_node_id={} upgrade_id={} transport_stable_id={} reason={}",
                            connection_id,
                            remote_node_id,
                            upgrade_id,
                            transport_stable_id,
                            admission_block
                                .as_ref()
                                .map(|(_, reason)| reason.as_str())
                                .unwrap_or("unknown"),
                        );
                        let client = self.clone();
                        let connection_id = connection_id.to_string();
                        let remote_node_id = remote_node_id.to_string();
                        let upgrade_id = upgrade_id.to_string();
                        tokio::spawn(async move {
                            let result = client
                                .wait_for_native_ble_switch_admission(
                                    &connection_id,
                                    endpoint_id,
                                    expected_generation,
                                )
                                .await;
                            if let Err(error) = result {
                                remove_native_transport_upgrade_gate(
                                    &client.native_transport_upgrade_gates,
                                    &(connection_id.clone(), IrohPathKind::Ble),
                                    &upgrade_id,
                                )
                                .await;
                                eprintln!(
                                    "[OpenRTC][BLE] deferred switch request failed connection_id={} remote_node_id={} upgrade_id={} error={}",
                                    connection_id, remote_node_id, upgrade_id, error,
                                );
                                return;
                            }
                            if let Err(error) = client
                                .accept_native_ble_switch_request(
                                    &connection_id,
                                    &remote_node_id,
                                    &upgrade_id,
                                    expected_generation,
                                    proof_generation,
                                )
                                .await
                            {
                                eprintln!(
                                    "[OpenRTC][BLE] deferred switch request failed connection_id={} remote_node_id={} upgrade_id={} error={}",
                                    connection_id, remote_node_id, upgrade_id, error,
                                );
                            }
                        });
                        return true;
                    }
                    NativeBleSwitchAdmission::Ready => {}
                }
                if let Err(error) = self
                    .accept_native_ble_switch_request(
                        connection_id,
                        remote_node_id,
                        upgrade_id,
                        expected_generation,
                        proof_generation,
                    )
                    .await
                {
                    eprintln!(
                        "[OpenRTC][BLE] switch request failed connection_id={} remote_node_id={} upgrade_id={} error={}",
                        connection_id, remote_node_id, upgrade_id, error,
                    );
                }
            }
            "switch-ready" => {
                let gate_key = (connection_id.to_string(), IrohPathKind::Ble);
                let completion_gate_id = format!("{NATIVE_BLE_COMPLETION_GATE_PREFIX}{upgrade_id}");
                let expected_generation = {
                    let mut gates = self.native_transport_upgrade_gates.lock().await;
                    gates.get_mut(&gate_key).and_then(|current| {
                        transition_native_ble_upgrade_gate_to_completion(
                            current,
                            upgrade_id,
                            proof_generation,
                        )
                    })
                };
                let Some(expected_generation) = expected_generation else {
                    println!(
                        "[OpenRTC][BLE] switch ready ignored connection_id={} remote_node_id={} upgrade_id={} reason=generation-mismatch",
                        connection_id, remote_node_id, upgrade_id,
                    );
                    return true;
                };
                println!(
                    "[OpenRTC][BLE] switch ready accepted connection_id={} remote_node_id={} upgrade_id={}",
                    connection_id, remote_node_id, upgrade_id,
                );
                let Ok(endpoint_id) = remote_node_id.parse::<iroh::EndpointId>() else {
                    return true;
                };
                let client = self.clone();
                let connection_id = connection_id.to_string();
                let remote = remote_node_id.to_string();
                let upgrade_id = upgrade_id.to_string();
                let attempt = self
                    .current_native_ble_upgrade_attempt(connection_id.as_str(), expected_generation)
                    .await;
                tokio::spawn(async move {
                    let gate_key = (connection_id.clone(), IrohPathKind::Ble);
                    if let Err(error) = client
                        .wait_for_native_ble_switch_admission(
                            &connection_id,
                            endpoint_id,
                            expected_generation,
                        )
                        .await
                    {
                        let removed = remove_native_transport_upgrade_gate(
                            &client.native_transport_upgrade_gates,
                            &gate_key,
                            &completion_gate_id,
                        )
                        .await;
                        if removed {
                            if let Some(attempt) = attempt {
                                client.schedule_native_ble_upgrade_retry(
                                    connection_id.clone(),
                                    remote.clone(),
                                    expected_generation,
                                    attempt,
                                    "admission-wait-failed",
                                );
                            }
                        }
                        eprintln!(
                            "[OpenRTC][BLE] switch ready admission failed connection_id={} remote_node_id={} upgrade_id={} error={}",
                            connection_id, remote, upgrade_id, error,
                        );
                        return;
                    }
                    let result = client
                        .complete_native_ble_upgrade(
                            &connection_id,
                            &remote,
                            &upgrade_id,
                            expected_generation,
                        )
                        .await;
                    let removed = remove_native_transport_upgrade_gate(
                        &client.native_transport_upgrade_gates,
                        &gate_key,
                        &completion_gate_id,
                    )
                    .await;
                    match result {
                        Ok(()) => {
                            client
                                .clear_native_ble_upgrade_attempts(
                                    &connection_id,
                                    expected_generation,
                                )
                                .await;
                        }
                        Err(error) => {
                            if removed {
                                if let Some(attempt) = attempt {
                                    client.schedule_native_ble_upgrade_retry(
                                        connection_id.clone(),
                                        remote.clone(),
                                        expected_generation,
                                        attempt,
                                        "replacement-failed",
                                    );
                                }
                            }
                            eprintln!(
                                "[OpenRTC][BLE] upgrade failed connection_id={} remote_node_id={} upgrade_id={} error={}",
                                connection_id, remote, upgrade_id, error
                            );
                        }
                    }
                });
            }
            _ => {}
        }
        true
    }

    #[cfg(all(not(target_arch = "wasm32"), feature = "iroh-carrier-core"))]
    pub(crate) async fn receive_native_iroh_carrier_candidate_proof(
        &self,
        connection_id: &str,
        send: iroh::endpoint::SendStream,
        recv: iroh::endpoint::RecvStream,
        upgrade_id: &str,
        proof_base_transport_generation: u64,
        proof_base_route_generation: u64,
        kind: IrohPathKind,
        transport_id: u64,
    ) -> anyhow::Result<(
        crate::application_crypto_streams::PeerSendStream,
        crate::iroh_carrier_proof::CarrierCandidateProofFrame,
        crate::iroh_carrier_proof::CarrierCandidateProofFrame,
    )> {
        let (send, mut recv) =
            self.wrap_peer_streams_for_connection(Some(connection_id), send, recv)?;
        let payload = crate::iroh_carrier_proof::read_candidate_proof_stream(&mut recv).await?;
        let probe: crate::iroh_carrier_proof::CarrierCandidateProofFrame =
            serde_json::from_slice(&payload)?;
        let expected_carrier = match kind {
            IrohPathKind::Ble => crate::iroh_carrier_bootstrap::CarrierBootstrapKind::Ble,
            IrohPathKind::WebRtc => crate::iroh_carrier_bootstrap::CarrierBootstrapKind::WebRtc,
            IrohPathKind::Moq => crate::iroh_carrier_bootstrap::CarrierBootstrapKind::MoqDraft14,
            _ => anyhow::bail!("unsupported inbound Iroh carrier proof kind {kind:?}"),
        };
        anyhow::ensure!(
            probe.frame_type == crate::iroh_carrier_proof::CARRIER_CANDIDATE_PROOF_TYPE
                && probe.role == crate::iroh_carrier_proof::CarrierCandidateProofRole::Probe
                && probe.carrier == expected_carrier
                && probe.transport_id == transport_id
                && probe.upgrade_id == upgrade_id
                && probe.base_transport_generation == proof_base_transport_generation
                && probe.base_route_generation == proof_base_route_generation,
            "carrier candidate proof does not match the authorized native attempt: expected_carrier={:?} actual_carrier={:?} expected_transport_id={} actual_transport_id={} expected_upgrade_id={} actual_upgrade_id={} expected_base_transport_generation={} actual_base_transport_generation={} expected_base_route_generation={} actual_base_route_generation={}",
            expected_carrier,
            probe.carrier,
            transport_id,
            probe.transport_id,
            upgrade_id,
            probe.upgrade_id,
            proof_base_transport_generation,
            probe.base_transport_generation,
            proof_base_route_generation,
            probe.base_route_generation,
        );
        probe.validate()?;
        let ack = crate::iroh_carrier_proof::CarrierCandidateProofFrame::ack_from(&probe)?;
        Ok((send, probe, ack))
    }

    #[cfg(all(not(target_arch = "wasm32"), feature = "iroh-carrier-core"))]
    pub(crate) async fn send_native_iroh_carrier_candidate_ack(
        &self,
        mut send: crate::application_crypto_streams::PeerSendStream,
        ack: &crate::iroh_carrier_proof::CarrierCandidateProofFrame,
    ) -> anyhow::Result<()> {
        send.write_all(&serde_json::to_vec(&ack)?).await?;
        send.finish_and_wait_for_peer(
            crate::iroh_carrier_proof::CARRIER_CANDIDATE_PROOF_DELIVERY_TIMEOUT,
        )
        .await?;
        Ok(())
    }

    #[cfg(all(not(target_arch = "wasm32"), feature = "iroh-carrier-core"))]
    async fn receive_native_iroh_carrier_commit(
        &self,
        connection_id: &str,
        connection: &iroh::endpoint::Connection,
        probe: &crate::iroh_carrier_proof::CarrierCandidateProofFrame,
    ) -> anyhow::Result<(
        crate::application_crypto_streams::PeerSendStream,
        crate::iroh_carrier_proof::CarrierCandidateProofFrame,
    )> {
        let (send, recv) = connection.accept_bi().await?;
        let (send, mut recv) =
            self.wrap_peer_streams_for_connection(Some(connection_id), send, recv)?;
        let payload = crate::iroh_carrier_proof::read_candidate_proof_stream(&mut recv).await?;
        let commit: crate::iroh_carrier_proof::CarrierCandidateProofFrame =
            serde_json::from_slice(&payload)?;
        anyhow::ensure!(
            commit.matches_commit_for_probe(probe),
            "carrier commit does not match the proven candidate"
        );
        let committed =
            crate::iroh_carrier_proof::CarrierCandidateProofFrame::committed_from(&commit)?;
        Ok((send, committed))
    }

    #[cfg(all(not(target_arch = "wasm32"), feature = "iroh-carrier-core"))]
    async fn confirm_outbound_native_iroh_carrier_commit(
        &self,
        connection_id: &str,
        connection: &iroh::endpoint::Connection,
        probe: &crate::iroh_carrier_proof::CarrierCandidateProofFrame,
    ) -> anyhow::Result<()> {
        let commit = crate::iroh_carrier_proof::CarrierCandidateProofFrame::commit_from(probe)?;
        let (send, recv) = connection.open_bi().await?;
        let (mut send, mut recv) =
            self.wrap_peer_streams_for_connection(Some(connection_id), send, recv)?;
        send.write_all(&serde_json::to_vec(&commit)?).await?;
        send.finish()?;
        let payload = crate::iroh_carrier_proof::read_candidate_proof_stream(&mut recv).await?;
        let committed: crate::iroh_carrier_proof::CarrierCandidateProofFrame =
            serde_json::from_slice(&payload)?;
        anyhow::ensure!(
            committed.matches_committed_for_commit(&commit),
            "carrier commit acknowledgement does not match the current candidate"
        );
        Ok(())
    }

    #[cfg(all(not(target_arch = "wasm32"), feature = "iroh-carrier-core"))]
    async fn prove_outbound_native_iroh_carrier_candidate(
        &self,
        connection_id: &str,
        connection: &iroh::endpoint::Connection,
        upgrade_id: &str,
        expected_generation: crate::client::NativePeerDataGeneration,
        kind: IrohPathKind,
        transport_id: u64,
    ) -> anyhow::Result<crate::iroh_carrier_proof::CarrierCandidateProofFrame> {
        anyhow::ensure!(
            self.application_crypto_key_for_connection(Some(connection_id))
                .is_some(),
            "carrier candidate proof requires an installed application-crypto key"
        );
        let carrier = match kind {
            IrohPathKind::Ble => crate::iroh_carrier_bootstrap::CarrierBootstrapKind::Ble,
            IrohPathKind::WebRtc => crate::iroh_carrier_bootstrap::CarrierBootstrapKind::WebRtc,
            IrohPathKind::Moq => crate::iroh_carrier_bootstrap::CarrierBootstrapKind::MoqDraft14,
            _ => anyhow::bail!("unsupported outbound Iroh carrier proof kind {kind:?}"),
        };
        let probe = crate::iroh_carrier_proof::CarrierCandidateProofFrame::probe(
            carrier,
            transport_id,
            upgrade_id,
            expected_generation.transport_generation,
            expected_generation.route_generation,
        )?;
        let (send, recv) = connection.open_bi().await?;
        let (mut send, mut recv) =
            self.wrap_peer_streams_for_connection(Some(connection_id), send, recv)?;
        send.write_all(&serde_json::to_vec(&probe)?).await?;
        send.finish()?;
        let payload = crate::iroh_carrier_proof::read_candidate_proof_stream(&mut recv).await?;
        let ack: crate::iroh_carrier_proof::CarrierCandidateProofFrame =
            serde_json::from_slice(&payload)?;
        anyhow::ensure!(
            ack.matches_probe(&probe),
            "carrier candidate proof acknowledgement did not match the current probe"
        );
        Ok(probe)
    }

    #[cfg(all(not(target_arch = "wasm32"), feature = "iroh-carrier-core"))]
    async fn complete_native_iroh_carrier_upgrade(
        &self,
        connection_id: &str,
        remote_node_id: &str,
        upgrade_id: &str,
        expected_generation: crate::client::NativePeerDataGeneration,
        kind: IrohPathKind,
    ) -> anyhow::Result<()> {
        anyhow::ensure!(
            matches!(
                kind,
                IrohPathKind::Ble | IrohPathKind::WebRtc | IrohPathKind::Moq
            ),
            "unsupported Iroh packet carrier: {kind:?}"
        );
        let endpoint_id = remote_node_id.parse::<iroh::EndpointId>()?;
        self.record_native_iroh_carrier_debug_event(
            connection_id,
            upgrade_id,
            "candidate-outbound-provider-start",
            Some(kind.transport_label()),
        );
        let provider = self
            .native_transport_upgrade_provider(kind)
            .await
            .ok_or_else(|| anyhow::anyhow!("{kind:?} carrier provider is unavailable"))?;
        self.record_native_iroh_carrier_debug_event(
            connection_id,
            upgrade_id,
            "candidate-outbound-provider-ready",
            Some(kind.transport_label()),
        );
        let endpoint_addr = provider.prepare_endpoint_addr(endpoint_id).await?;
        self.record_native_iroh_carrier_debug_event(
            connection_id,
            upgrade_id,
            "candidate-outbound-address-ready",
            Some(kind.transport_label()),
        );
        let gate_key = (connection_id.to_string(), kind);
        let policy_epoch = self.current_iroh_carrier_policy_epoch();
        let peer_policy_epoch = self.current_iroh_carrier_peer_policy_epoch(connection_id);
        anyhow::ensure!(
            self.native_transport_upgrade_gates
                .lock()
                .await
                .get(&gate_key)
                .is_some_and(|gate| {
                    native_transport_upgrade_gate_matches(gate, upgrade_id)
                        && gate.policy_epoch == policy_epoch
                        && gate.peer_policy_epoch == peer_policy_epoch
                }),
            "{kind:?} replacement belongs to a retired upgrade"
        );
        let node = self
            .iroh_node
            .read()
            .await
            .as_ref()
            .cloned()
            .ok_or_else(|| anyhow::anyhow!("Iroh node is unavailable"))?;
        self.record_native_iroh_carrier_debug_event(
            connection_id,
            upgrade_id,
            "candidate-outbound-dial-start",
            Some(kind.transport_label()),
        );
        let candidate = node
            .dial_replacement_candidate(endpoint_id, endpoint_addr, provider.transport_id())
            .await?;
        self.record_native_iroh_carrier_debug_event(
            connection_id,
            upgrade_id,
            "candidate-outbound-dial-ready",
            Some(kind.transport_label()),
        );
        let authorization_fence = self.capture_native_iroh_carrier_authorization_fence(
            connection_id,
            expected_generation,
            kind,
            policy_epoch,
        )?;
        self.record_native_iroh_carrier_debug_event(
            connection_id,
            upgrade_id,
            "candidate-outbound-proof-start",
            Some(kind.transport_label()),
        );
        let probe = self
            .prove_outbound_native_iroh_carrier_candidate(
                connection_id,
                candidate.connection(),
                upgrade_id,
                expected_generation,
                kind,
                provider.transport_id(),
            )
            .await?;
        self.record_native_iroh_carrier_debug_event(
            connection_id,
            upgrade_id,
            "candidate-outbound-proof-ready",
            Some(kind.transport_label()),
        );
        // Ask the responder to commit first. If its incumbent changed while
        // the carrier was being prepared, it rejects the commit while this
        // side still owns its healthy base route instead of leaving one peer
        // stranded on the custom carrier.
        self.confirm_outbound_native_iroh_carrier_commit(
            connection_id,
            candidate.connection(),
            &probe,
        )
        .await?;
        self.record_native_iroh_carrier_debug_event(
            connection_id,
            upgrade_id,
            "candidate-outbound-commit-confirmed",
            Some(kind.transport_label()),
        );
        let committed_replacement = self
            .commit_proven_native_iroh_carrier_candidate(
                connection_id,
                remote_node_id,
                upgrade_id,
                expected_generation,
                kind,
                authorization_fence,
                node,
                candidate,
            )
            .await?;
        self.record_native_iroh_carrier_debug_event(
            connection_id,
            upgrade_id,
            "candidate-outbound-installed",
            Some(kind.transport_label()),
        );
        let replacement_transport_stable_id = committed_replacement
            .logical_result()
            .transport_stable_id
            .ok_or_else(|| anyhow::anyhow!("{kind:?} replacement has no stable ID"))?;
        committed_replacement
            .finish(crate::lifecycle_reason::REASON_NATIVE_CUSTOM_TRANSPORT_UPGRADE.as_bytes());
        self.settle_committed_native_iroh_carrier_application_route(
            connection_id,
            upgrade_id,
            kind,
            replacement_transport_stable_id,
        )
        .await;
        Ok(())
    }

    #[cfg(all(not(target_arch = "wasm32"), feature = "iroh-carrier-core"))]
    async fn complete_inbound_native_iroh_carrier_upgrade(
        &self,
        connection_id: &str,
        remote_node_id: &str,
        upgrade_id: &str,
        expected_generation: crate::client::NativePeerDataGeneration,
        proof_base_transport_generation: u64,
        proof_base_route_generation: u64,
        kind: IrohPathKind,
        transport_id: u64,
        authorization_expires_at: std::time::Instant,
    ) -> anyhow::Result<()> {
        let endpoint_id = remote_node_id.parse::<iroh::EndpointId>()?;
        let policy_epoch = self.current_iroh_carrier_policy_epoch();
        let peer_policy_epoch = self.current_iroh_carrier_peer_policy_epoch(connection_id);
        anyhow::ensure!(
            self.native_transport_upgrade_gates
                .lock()
                .await
                .get(&(connection_id.to_string(), kind))
                .is_some_and(|gate| {
                    native_transport_upgrade_gate_matches(gate, upgrade_id)
                        && gate.policy_epoch == policy_epoch
                        && gate.peer_policy_epoch == peer_policy_epoch
                }),
            "{kind:?} inbound replacement belongs to a retired upgrade"
        );
        let node = self
            .iroh_node
            .read()
            .await
            .as_ref()
            .cloned()
            .ok_or_else(|| anyhow::anyhow!("Iroh node is unavailable"))?;
        self.record_native_iroh_carrier_debug_event(
            connection_id,
            upgrade_id,
            "candidate-inbound-wait-start",
            Some(kind.transport_label()),
        );
        let candidate = node
            .wait_for_inbound_replacement_candidate(
                endpoint_id,
                transport_id,
                authorization_expires_at,
                std::time::Duration::from_secs(40),
            )
            .await?;
        self.record_native_iroh_carrier_debug_event(
            connection_id,
            upgrade_id,
            "candidate-inbound-arrived",
            Some(kind.transport_label()),
        );
        let authorization_fence = self.capture_native_iroh_carrier_authorization_fence(
            connection_id,
            expected_generation,
            kind,
            policy_epoch,
        )?;
        let (send, recv) = candidate.connection().accept_bi().await?;
        self.record_native_iroh_carrier_debug_event(
            connection_id,
            upgrade_id,
            "candidate-inbound-proof-start",
            Some(kind.transport_label()),
        );
        let (send, probe, ack) = self
            .receive_native_iroh_carrier_candidate_proof(
                connection_id,
                send,
                recv,
                upgrade_id,
                proof_base_transport_generation,
                proof_base_route_generation,
                kind,
                transport_id,
            )
            .await?;
        self.record_native_iroh_carrier_debug_event(
            connection_id,
            upgrade_id,
            "candidate-inbound-proof-ready",
            Some(kind.transport_label()),
        );
        anyhow::ensure!(
            self.native_iroh_carrier_pre_ack_authority_is_current(
                connection_id,
                kind,
                upgrade_id,
                &authorization_fence,
            )
            .await,
            "{kind:?} inbound carrier authority changed before candidate acknowledgement"
        );
        self.send_native_iroh_carrier_candidate_ack(send, &ack)
            .await?;
        self.record_native_iroh_carrier_debug_event(
            connection_id,
            upgrade_id,
            "candidate-inbound-proof-acknowledged",
            Some(kind.transport_label()),
        );
        let (commit_send, committed) = self
            .receive_native_iroh_carrier_commit(connection_id, candidate.connection(), &probe)
            .await?;
        self.record_native_iroh_carrier_debug_event(
            connection_id,
            upgrade_id,
            "candidate-inbound-commit-received",
            Some(kind.transport_label()),
        );
        let committed_replacement = self
            .commit_proven_native_iroh_carrier_candidate(
                connection_id,
                remote_node_id,
                upgrade_id,
                expected_generation,
                kind,
                authorization_fence,
                node,
                candidate,
            )
            .await?;
        self.record_native_iroh_carrier_debug_event(
            connection_id,
            upgrade_id,
            "candidate-inbound-installed",
            Some(kind.transport_label()),
        );
        self.send_native_iroh_carrier_candidate_ack(commit_send, &committed)
            .await?;
        let replacement_transport_stable_id = committed_replacement
            .logical_result()
            .transport_stable_id
            .ok_or_else(|| anyhow::anyhow!("{kind:?} replacement has no stable ID"))?;
        committed_replacement
            .finish(crate::lifecycle_reason::REASON_NATIVE_CUSTOM_TRANSPORT_UPGRADE.as_bytes());
        self.settle_committed_native_iroh_carrier_application_route(
            connection_id,
            upgrade_id,
            kind,
            replacement_transport_stable_id,
        )
        .await;
        Ok(())
    }

    /// Re-run the one application-key owner only after both peers have
    /// committed the physical carrier and the displaced incumbent has been
    /// retired. The carrier challenge proves possession of the existing key;
    /// it does not replace the stable-ID-bound capability/update/response/ACK
    /// transcript required for product traffic.
    #[cfg(all(not(target_arch = "wasm32"), feature = "iroh-carrier-core"))]
    async fn settle_committed_native_iroh_carrier_application_route(
        &self,
        connection_id: &str,
        upgrade_id: &str,
        kind: IrohPathKind,
        replacement_transport_stable_id: u64,
    ) {
        self.record_native_iroh_carrier_debug_event(
            connection_id,
            upgrade_id,
            "application-crypto-confirmation-start",
            Some(kind.transport_label()),
        );
        let result = async {
            self.ensure_connection_application_crypto(connection_id, 20_000)
                .await?;
            anyhow::ensure!(
                self.connection_application_crypto_is_confirmed(
                    connection_id,
                    Some(replacement_transport_stable_id),
                ),
                "{kind:?} replacement key transcript did not confirm the committed generation"
            );
            anyhow::ensure!(
                self.native_application_stream_admitted_for_transport(
                    connection_id,
                    replacement_transport_stable_id,
                ),
                "{kind:?} replacement key transcript did not settle admission: {}",
                self.native_application_stream_pending_diagnostic(
                    connection_id,
                    replacement_transport_stable_id,
                ),
            );
            anyhow::ensure!(
                self.confirm_managed_connection_readiness_from_transport_proof(
                    connection_id,
                    replacement_transport_stable_id,
                )
                .await,
                "{kind:?} replacement key transcript belongs to a retired generation"
            );
            Ok::<(), anyhow::Error>(())
        }
        .await;

        match result {
            Ok(()) => {
                self.record_native_iroh_carrier_debug_event(
                    connection_id,
                    upgrade_id,
                    "application-crypto-confirmed",
                    Some(kind.transport_label()),
                );
                println!(
                    "[OpenRTC][Iroh carrier] generation settled connection_id={} carrier={} transport_stable_id={}",
                    connection_id,
                    kind.transport_label(),
                    replacement_transport_stable_id,
                );
            }
            Err(error) => {
                let error_detail = format!("{error:#}");
                self.record_native_iroh_carrier_debug_event(
                    connection_id,
                    upgrade_id,
                    "application-crypto-confirmation-failed",
                    Some(&error_detail),
                );
                eprintln!(
                    "[OpenRTC][Iroh carrier] committed generation remains application-pending connection_id={} carrier={} transport_stable_id={} error={error:#}",
                    connection_id,
                    kind.transport_label(),
                    replacement_transport_stable_id,
                );
            }
        }
    }

    #[cfg(all(not(target_arch = "wasm32"), feature = "iroh-carrier-core"))]
    pub(super) fn capture_native_iroh_carrier_authorization_fence(
        &self,
        connection_id: &str,
        expected_generation: crate::client::NativePeerDataGeneration,
        kind: IrohPathKind,
        policy_epoch: u64,
    ) -> anyhow::Result<NativeIrohCarrierAuthorizationFence> {
        anyhow::ensure!(
            self.native_application_stream_admitted_for_transport(
                connection_id,
                expected_generation.transport_stable_id,
            ),
            "{kind:?} replacement base is not fully admitted: {}",
            self.native_application_stream_pending_diagnostic(
                connection_id,
                expected_generation.transport_stable_id,
            ),
        );
        Ok(NativeIrohCarrierAuthorizationFence {
            inbound_admission_was_current: self.inbound_session_token_admitted_for_transport(
                connection_id,
                expected_generation.transport_stable_id,
            ),
            remote_admission_proof: self
                .remote_session_admission_proofs
                .read()
                .ok()
                .and_then(|proofs| proofs.get(connection_id).cloned())
                .filter(|proof| {
                    proof.transport_stable_id == expected_generation.transport_stable_id
                }),
            admission_fingerprint: self
                .session_token_registry
                .admission_fingerprint(connection_id),
            application_security_epoch_fingerprint: self
                .session_token_registry
                .application_security_epoch_fingerprint(connection_id),
            policy_epoch,
        })
    }

    #[cfg(all(not(target_arch = "wasm32"), feature = "iroh-carrier-core"))]
    pub(super) fn native_iroh_carrier_authorization_epoch_is_current(
        &self,
        connection_id: &str,
        authorization_fence: &NativeIrohCarrierAuthorizationFence,
    ) -> bool {
        matches!(
            self.session_admission(connection_id),
            crate::session_token::SessionAdmission::Accepted { .. }
        ) && self
            .session_token_registry
            .admission_fingerprint(connection_id)
            == authorization_fence.admission_fingerprint
            && self
                .session_token_registry
                .application_security_epoch_fingerprint(connection_id)
                == authorization_fence.application_security_epoch_fingerprint
            && self.iroh_carrier_policy_epoch_is_current(authorization_fence.policy_epoch)
    }

    #[cfg(all(not(target_arch = "wasm32"), feature = "iroh-carrier-core"))]
    pub(super) async fn native_iroh_carrier_pre_ack_authority_is_current(
        &self,
        connection_id: &str,
        kind: IrohPathKind,
        upgrade_id: &str,
        authorization_fence: &NativeIrohCarrierAuthorizationFence,
    ) -> bool {
        // Route observations and a locally projected successor may advance the
        // connection record while the protected candidate proves itself. The
        // upgrade gate and captured authorization epoch remain the control
        // authority here; atomic commit below still rejects a competing
        // physical successor against the exact incumbent stable ID.
        self.native_iroh_carrier_upgrade_is_current(
            connection_id,
            kind,
            upgrade_id,
            authorization_fence.policy_epoch,
        )
        .await
            && self.native_iroh_carrier_authorization_epoch_is_current(
                connection_id,
                authorization_fence,
            )
    }

    #[cfg(all(not(target_arch = "wasm32"), feature = "iroh-carrier-core"))]
    async fn native_iroh_carrier_upgrade_is_current(
        &self,
        connection_id: &str,
        kind: IrohPathKind,
        upgrade_id: &str,
        policy_epoch: u64,
    ) -> bool {
        let gates = self.native_transport_upgrade_gates.lock().await;
        let peer_policy_epoch = self.current_iroh_carrier_peer_policy_epoch(connection_id);
        gates
            .get(&(connection_id.to_string(), kind))
            .is_some_and(|gate| {
                native_transport_upgrade_gate_matches(gate, upgrade_id)
                    && gate.policy_epoch == policy_epoch
                    && gate.peer_policy_epoch == peer_policy_epoch
                    && self.iroh_carrier_policy_epoch_is_current(policy_epoch)
            })
    }

    #[cfg(all(not(target_arch = "wasm32"), feature = "iroh-carrier-core"))]
    async fn commit_proven_native_iroh_carrier_candidate(
        &self,
        connection_id: &str,
        _remote_node_id: &str,
        upgrade_id: &str,
        expected_generation: crate::client::NativePeerDataGeneration,
        kind: IrohPathKind,
        authorization_fence: NativeIrohCarrierAuthorizationFence,
        node: crate::native_node::IrohNativeNode,
        candidate: crate::native_node::PendingReplacementConnection,
    ) -> anyhow::Result<
        crate::native_node::CommittedReplacement<crate::connection_manager::ConnectionRecord>,
    > {
        anyhow::ensure!(
            self.native_iroh_carrier_upgrade_is_current(
                connection_id,
                kind,
                upgrade_id,
                authorization_fence.policy_epoch,
            )
            .await,
            "{kind:?} replacement belongs to a retired policy or upgrade"
        );
        let authorization_fence_for_commit = authorization_fence.clone();
        let NativeIrohCarrierAuthorizationFence {
            inbound_admission_was_current,
            remote_admission_proof,
            ..
        } = authorization_fence;
        let client = self.clone();
        let connection_id_for_commit = connection_id.to_string();
        let upgrade_id_for_commit = upgrade_id.to_string();
        let policy_epoch_for_commit = authorization_fence_for_commit.policy_epoch;
        let transport_source = Some(format!("{}-upgrade", kind.transport_label()));
        let committed_replacement = node
            .commit_replacement_candidate(
                candidate,
                expected_generation.transport_stable_id,
                move |replacement_transport_stable_id| async move {
                    let gates = client.native_transport_upgrade_gates.lock().await;
                    let peer_policy_epoch =
                        client.current_iroh_carrier_peer_policy_epoch(&connection_id_for_commit);
                    anyhow::ensure!(
                        gates
                            .get(&(connection_id_for_commit.clone(), kind))
                            .is_some_and(|gate| {
                                native_transport_upgrade_gate_matches(gate, &upgrade_id_for_commit)
                                    && gate.policy_epoch == policy_epoch_for_commit
                                    && gate.peer_policy_epoch == peer_policy_epoch
                                    && client.iroh_carrier_policy_epoch_is_current(
                                        policy_epoch_for_commit,
                                    )
                            }),
                        "{kind:?} replacement upgrade changed before atomic commit"
                    );
                    anyhow::ensure!(
                        client.native_iroh_carrier_authorization_epoch_is_current(
                            &connection_id_for_commit,
                            &authorization_fence_for_commit,
                        ),
                        "{kind:?} replacement authorization epoch changed before atomic commit"
                    );
                    let installed = client
                        .begin_crypto_bound_transport_replacement_commit(
                            &connection_id_for_commit,
                            expected_generation.transport_stable_id,
                            expected_generation.transport_generation,
                            expected_generation.route_generation,
                            replacement_transport_stable_id,
                            transport_source,
                        )
                        .await;
                    drop(gates);
                    installed
                        .ok_or_else(|| anyhow::anyhow!("{kind:?} replacement generation is stale"))
                },
            )
            .await?;
        let committed = committed_replacement.logical_result();
        let replacement_transport_stable_id = committed
            .transport_stable_id
            .ok_or_else(|| anyhow::anyhow!("{kind:?} replacement has no stable ID"))?;
        self.retire_stale_native_main_route(connection_id, replacement_transport_stable_id)
            .await;
        anyhow::ensure!(
            committed.transport_stable_id == Some(replacement_transport_stable_id)
                && committed.transport_generation
                    == expected_generation.transport_generation.saturating_add(1),
            "{kind:?} replacement failed atomic generation commit"
        );
        if inbound_admission_was_current {
            self.bind_inbound_session_token_transport(
                connection_id,
                replacement_transport_stable_id,
            );
        }
        if let Some(mut proof) = remote_admission_proof {
            proof.transport_stable_id = replacement_transport_stable_id;
            if let Ok(mut proofs) = self.remote_session_admission_proofs.write() {
                proofs.insert(connection_id.to_string(), proof);
                self.notify_connection_application_route_update();
            }
        }
        println!(
            "[OpenRTC][Iroh carrier] admission proof rebound connection_id={} carrier={} transport_stable_id={} route=[{}]",
            connection_id,
            kind.transport_label(),
            replacement_transport_stable_id,
            self.native_application_stream_pending_diagnostic(
                connection_id,
                replacement_transport_stable_id,
            ),
        );
        anyhow::ensure!(
            self.report_transport_status_for_current_generation(
                connection_id,
                kind.transport_label(),
                None,
            )
            .await
            .is_some(),
            "{kind:?} replacement could not publish its proven route"
        );
        // Candidate proof is transport authorization only. The atomic commit
        // above deliberately leaves application-crypto confirmation false;
        // the caller retires the incumbent and then re-drives the existing
        // stable-ID-bound key transcript.
        // Exact carrier selection is already proven by the pending candidate:
        // the native node admits/dials only `kind`'s transport ID, and the
        // bilateral encrypted probe binds that ID, carrier kind, upgrade ID,
        // challenge, and base generations before this commit. Do not veto the
        // committed candidate by querying `iroh_path_kind` here. That lookup
        // resolves through the shared current-connection map, which may still
        // expose the incumbent during the two-sided swap; treating that
        // point-in-time projection as authority closes the healthy carrier and
        // creates an endless custom-carrier/relay replacement loop.
        Ok(committed_replacement)
    }

    #[cfg(all(not(target_arch = "wasm32"), feature = "iroh-carrier-core"))]
    async fn complete_native_ble_upgrade(
        &self,
        connection_id: &str,
        remote_node_id: &str,
        upgrade_id: &str,
        expected_generation: crate::client::NativePeerDataGeneration,
    ) -> anyhow::Result<()> {
        self.complete_native_iroh_carrier_upgrade(
            connection_id,
            remote_node_id,
            upgrade_id,
            expected_generation,
            IrohPathKind::Ble,
        )
        .await
    }

    #[cfg(all(not(target_arch = "wasm32"), not(feature = "iroh-carrier-core")))]
    async fn complete_native_ble_upgrade(
        &self,
        _connection_id: &str,
        _remote_node_id: &str,
        _upgrade_id: &str,
        _expected_generation: crate::client::NativePeerDataGeneration,
    ) -> anyhow::Result<()> {
        anyhow::bail!("BLE replacement requires the Iroh custom-carrier core")
    }

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

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

    pub(crate) async fn maybe_handle_typescript_json_frame(
        &self,
        connection_id: &str,
        remote_node_id: Option<&str>,
        frame: &serde_json::Value,
    ) {
        #[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
        if self
            .handle_native_webrtc_iroh_carrier_control(connection_id, remote_node_id, frame)
            .await
        {
            return;
        }
        #[cfg(all(not(target_arch = "wasm32"), feature = "transport-moq"))]
        if self
            .handle_native_moq_iroh_carrier_control(connection_id, remote_node_id, frame)
            .await
        {
            return;
        }
        #[cfg(not(target_arch = "wasm32"))]
        if self
            .handle_native_ble_upgrade_control(connection_id, remote_node_id, frame)
            .await
        {
            return;
        }
    }

    /// Send a bounded protected message on the admitted logical Iroh connection.
    ///
    /// WebRTC and MoQ are packet carriers beneath this logical connection. They
    /// are selected by Iroh and never appear as independent application send
    /// fallbacks here.
    pub async fn send_peer(&self, id: &str, data: &[u8]) -> anyhow::Result<()> {
        if data.is_empty() {
            return Ok(());
        }
        if data.len() > crate::native_protocol::MAX_PEER_MESSAGE_BYTES {
            anyhow::bail!(
                "peer message exceeds {}-byte limit; use a named stream for bulk data",
                crate::native_protocol::MAX_PEER_MESSAGE_BYTES,
            );
        }
        let protected = self.protect_outbound_peer_payload(id, data).await?;
        self.send_peer_over_iroh(id, &protected).await
    }

    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
                .application_crypto_key_for_connection(Some(connection_id))
                .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())
    }
}

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

/// Frame an SDK-owned TypeScript JSON control message for native-main.
///
/// The `0x00` discriminator is part of the native-main wire contract. Keeping
/// capability, WebRTC signaling, and future carrier-control messages on
/// one encoder prevents a valid JSON envelope from being mistaken for opaque
/// application data by the receiving Rust runtime.
#[cfg(not(target_arch = "wasm32"))]
fn frame_typescript_native_main_json(message: &serde_json::Value) -> anyhow::Result<Vec<u8>> {
    let serialized = serde_json::to_vec(message)?;
    let frame_len = serialized
        .len()
        .checked_add(1)
        .and_then(|len| u32::try_from(len).ok())
        .ok_or_else(|| anyhow::anyhow!("TypeScript native-main frame is too large"))?;
    let mut frame = Vec::with_capacity(4 + frame_len as usize);
    frame.extend_from_slice(&frame_len.to_be_bytes());
    frame.push(0x00);
    frame.extend_from_slice(&serialized);
    Ok(frame)
}

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

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

        let transport_label = self.iroh_path_kind(id).await.transport_label();
        if let Some(transport_stable_id) = record.transport_stable_id {
            let _ = self
                .report_transport_status_for_generation(
                    &record.connection_id,
                    transport_label,
                    None,
                    transport_stable_id,
                    record.transport_generation,
                    record.route_generation,
                )
                .await;
        }

        Ok(())
    }

    async fn maybe_negotiate_typescript_application_crypto(
        &self,
        connection_id: &str,
        remote_node_id: Option<&str>,
        transport_stable_id: Option<u64>,
        handshake: &crate::native_protocol::TypeScriptHandshake,
    ) -> Option<[u8; crate::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 outcome = self
            .accept_application_key_handshake(
                connection_id,
                remote_node_id,
                transport_stable_id,
                handshake.action.as_deref(),
                remote_public,
            )
            .await
            .ok()?;
        outcome.key_changed.then_some(outcome.local_public_key)
    }

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

    /// Complete reciprocal application-key agreement for one exact logical
    /// connection generation.
    ///
    /// This is the single native/WASM recovery owner. Adapters may await the
    /// result and project readiness, but must not generate keys or retry
    /// handshake frames independently.
    pub async fn ensure_connection_application_crypto(
        &self,
        connection_id: &str,
        timeout_ms: u64,
    ) -> anyhow::Result<()> {
        let connection_id = connection_id.trim();
        anyhow::ensure!(!connection_id.is_empty(), "connection id is required");
        let timeout_ms = timeout_ms.max(1);
        let initial = self
            .connection_manager
            .get_by_connection_id(connection_id)
            .await
            .ok_or_else(|| anyhow::anyhow!("connection {connection_id} is not registered"))?;
        let initial_endpoint = initial
            .endpoint_id
            .as_deref()
            .or(initial.node_id.as_deref())
            .map(ToOwned::to_owned)
            .ok_or_else(|| anyhow::anyhow!("connection {connection_id} has no endpoint"))?;
        let initial_transport_stable_id = initial.transport_stable_id;
        let initial_transport_generation = initial.transport_generation;
        let initial_route_generation = initial.route_generation;
        self.set_connection_application_crypto_required(connection_id);

        let started = std::time::Instant::now();
        let retry_delays_ms = [0_u64, 250, 750];
        for (attempt, delay_ms) in retry_delays_ms.into_iter().enumerate() {
            if delay_ms > 0 {
                let remaining = timeout_ms.saturating_sub(started.elapsed().as_millis() as u64);
                if remaining == 0 {
                    break;
                }
                tokio::time::sleep(std::time::Duration::from_millis(delay_ms.min(remaining))).await;
            }
            self.assert_application_crypto_generation(
                connection_id,
                &initial_endpoint,
                initial_transport_stable_id,
                initial_transport_generation,
                initial_route_generation,
            )
            .await?;
            if self
                .application_crypto_key_for_connection(Some(connection_id))
                .is_some()
                && self.connection_application_crypto_is_confirmed(
                    connection_id,
                    initial_transport_stable_id,
                )
            {
                return Ok(());
            }
            let public_key = self
                .get_or_create_connection_key_agreement(connection_id)
                .map_err(|error| {
                    anyhow::anyhow!("application key agreement initialization failed: {error:?}")
                })?
                .public_key_bytes();
            self.send_typescript_capability_update(
                connection_id,
                if attempt == 0 {
                    "application-crypto-required"
                } else {
                    "application-crypto-retry"
                },
                Some(public_key),
            )
            .await?;

            let attempt_deadline_ms = ((attempt + 1) as u64 * timeout_ms / 3).max(1);
            loop {
                if self
                    .application_crypto_key_for_connection(Some(connection_id))
                    .is_some()
                    && self.connection_application_crypto_is_confirmed(
                        connection_id,
                        initial_transport_stable_id,
                    )
                {
                    return Ok(());
                }
                if started.elapsed().as_millis() as u64 >= attempt_deadline_ms
                    || started.elapsed().as_millis() as u64 >= timeout_ms
                {
                    break;
                }
                self.assert_application_crypto_generation(
                    connection_id,
                    &initial_endpoint,
                    initial_transport_stable_id,
                    initial_transport_generation,
                    initial_route_generation,
                )
                .await?;
                tokio::time::sleep(std::time::Duration::from_millis(25)).await;
            }
        }

        anyhow::bail!(
            "connection {connection_id} reciprocal application key agreement timed out after {timeout_ms}ms"
        )
    }

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

        let local_webrtc_enabled = self.local_typescript_webrtc_capability().await;
        let local_moq_enabled = self.local_typescript_moq_capability().await;
        let local_ble_enabled = self.is_ble_transport_enabled().await;
        let dedicated_carrier_control = local_webrtc_enabled || local_moq_enabled;
        let native_admission_route_ready = {
            #[cfg(not(target_arch = "wasm32"))]
            {
                self.current_native_peer_data_generation(connection_id, None)
                    .await
                    .is_some_and(|generation| {
                        self.native_admission_route_is_ready_for_transport(
                            connection_id,
                            Some(generation.transport_stable_id),
                        )
                    })
            }
            #[cfg(target_arch = "wasm32")]
            {
                true
            }
        };
        let dedicated_carrier_control_ready = dedicated_carrier_control_is_ready(
            dedicated_carrier_control,
            self.connection_application_crypto_is_confirmed(
                connection_id,
                self.connection_manager
                    .get_by_connection_id(connection_id)
                    .await
                    .and_then(|record| record.transport_stable_id),
            ),
            native_admission_route_ready,
        );
        // Native must seed mandatory application key agreement. Waiting for the
        // peer's public key here deadlocks with TS peers, which also wait for a
        // remote advertisement before sending a reciprocal capability update.
        let application_key_agreement_public_key =
            application_key_agreement_public_key.or_else(|| {
                self.get_or_create_connection_key_agreement(connection_id)
                    .ok()
                    .map(|agreement| agreement.public_key_bytes())
            });
        let mut capability_frame = serde_json::json!({
            "type": "handshake",
            "action": action,
            "capabilities": {
                "carrierSchema": 2,
                "webrtc": local_webrtc_enabled,
                "moq": local_moq_enabled,
                "dedicatedCarrierControl": dedicated_carrier_control,
                "dedicatedCarrierControlReady": dedicated_carrier_control_ready,
                "ble": local_ble_enabled,
            }
        });
        if let Some(public_key) = application_key_agreement_public_key {
            capability_frame["capabilities"]["applicationKeyAgreement"] =
                serde_json::Value::Bool(true);
            capability_frame["applicationKeyAgreement"] = serde_json::json!({
                "algorithm": crate::key_agreement::KEY_ALGORITHM,
                "publicKey": URL_SAFE_NO_PAD.encode(public_key),
            });
        }
        println!(
            "[OpenRTC][capability] sending update connection_id={} reason={} action={} webrtc={} moq={} ble={} application_key_agreement={}",
            connection_id,
            reason,
            action,
            local_webrtc_enabled,
            local_moq_enabled,
            local_ble_enabled,
            application_key_agreement_public_key.is_some(),
        );
        #[cfg(not(target_arch = "wasm32"))]
        let persistent_native_control = self.native_admission_stream_is_persistent(connection_id);
        #[cfg(target_arch = "wasm32")]
        let persistent_native_control = false;
        if application_key_reply_uses_fresh_signal(action, persistent_native_control) {
            self.send_typescript_handshake_over_fresh_signal(connection_id, &capability_frame)
                .await
        } else {
            self.send_typescript_handshake_over_native_main(connection_id, &capability_frame)
                .await
        }
    }

    async fn send_typescript_handshake_over_fresh_signal(
        &self,
        connection_id: &str,
        message: &serde_json::Value,
    ) -> anyhow::Result<()> {
        let record = self
            .connection_manager
            .get_by_connection_id(connection_id)
            .await
            .ok_or_else(|| anyhow::anyhow!("connection {connection_id} is not registered"))?;
        let endpoint_id = record
            .endpoint_id
            .as_deref()
            .or(record.node_id.as_deref())
            .ok_or_else(|| anyhow::anyhow!("connection {connection_id} has no endpoint"))?
            .parse::<iroh::EndpointId>()?;
        let frame = frame_typescript_native_main_json(message)?;
        let (mut send, _recv) = self.open_bi_internal(endpoint_id).await?;
        let label = b"signal";
        send.write_all(&[0x00]).await?;
        send.write_all(&(label.len() as u32).to_be_bytes()).await?;
        send.write_all(label).await?;
        send.write_all(&frame).await?;
        crate::application_crypto_streams::PeerSendStream::plain(send)
            .finish_and_wait_for_peer(std::time::Duration::from_secs(2))
            .await
            .map_err(anyhow::Error::msg)
    }

    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 frame = frame_typescript_native_main_json(message)?;

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

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

#[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
fn parse_webrtc_signal_message_for_transport(
    frame: &serde_json::Value,
    expected_transport: &str,
) -> 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 != expected_transport {
        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(),
    })
}

#[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
fn iroh_webrtc_signal_matches_upgrade(
    signal: &crate::transport::WebRTCSignalMessage,
    upgrade_id: &str,
) -> bool {
    signal.negotiation_id.as_deref() == Some(upgrade_id)
}

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

    has_application_key_agreement_public_key
        || local_webrtc_enabled
        || local_moq_enabled
        || local_ble_enabled
}

fn application_key_reply_uses_fresh_signal(action: &str, persistent_native_control: bool) -> bool {
    matches!(action, "response" | "ack") && !persistent_native_control
}

#[cfg(not(target_arch = "wasm32"))]
fn effective_typescript_transport_capability(
    uses_canonical_carrier_schema: bool,
    advertised: bool,
    known: bool,
) -> bool {
    if uses_canonical_carrier_schema {
        advertised
    } else {
        known
    }
}

fn typescript_application_key_reply_action(
    incoming_action: Option<&str>,
    has_application_key: bool,
    _application_key_changed: bool,
    application_crypto_was_confirmed: bool,
) -> Option<&'static str> {
    if !has_application_key {
        return None;
    }
    match incoming_action {
        // The ACK is the first frame that confirms application crypto when
        // this endpoint sent the response. Publish the now-actionable carrier
        // readiness once so the peer can move carrier control off the
        // bootstrap main stream. A repeated ACK for an already-confirmed
        // security epoch remains a no-op.
        Some("ack") if !application_crypto_was_confirmed => Some("capability-update"),
        Some("ack") => None,
        Some("response") => Some("ack"),
        _ => Some("response"),
    }
}

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

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

#[cfg(not(target_arch = "wasm32"))]
fn native_transport_upgrade_gate_matches(
    current: &crate::client::NativeTransportUpgradeGate,
    upgrade_id: &str,
) -> bool {
    current.upgrade_id == upgrade_id
        || current
            .upgrade_id
            .strip_prefix(NATIVE_BLE_COMPLETION_GATE_PREFIX)
            .is_some_and(|completing_upgrade_id| completing_upgrade_id == upgrade_id)
}

#[cfg(not(target_arch = "wasm32"))]
fn transition_native_ble_upgrade_gate_to_completion(
    current: &mut crate::client::NativeTransportUpgradeGate,
    upgrade_id: &str,
    proof_generation: crate::client::NativePeerDataGeneration,
) -> Option<crate::client::NativePeerDataGeneration> {
    if current.upgrade_id != upgrade_id || current.generation != proof_generation {
        return None;
    }

    current.upgrade_id = format!("{NATIVE_BLE_COMPLETION_GATE_PREFIX}{upgrade_id}");
    Some(current.generation)
}

/// An at-least-once control replay must reuse the exact live responder
/// attempt even after that attempt has installed a custom generation. Looking
/// only at the now-current base generation would misclassify this replay as a
/// new request and can retire the winning carrier before Ready is delivered.
#[cfg(all(
    not(target_arch = "wasm32"),
    any(feature = "transport-webrtc", feature = "transport-moq")
))]
fn native_carrier_duplicate_request_is_live(
    request: &crate::iroh_carrier_bootstrap::CarrierBootstrapFrame,
    attempt: &crate::iroh_carrier_bootstrap::CarrierBootstrapFrame,
    terminal: bool,
) -> bool {
    !terminal && request == attempt
}

/// A repeated bootstrap is safe to reuse only while it still names the
/// current physical base generation and a non-terminal WebRTC attempt.
#[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
fn native_webrtc_duplicate_bootstrap_is_reserved_or_live(
    gate: &crate::client::NativeTransportUpgradeGate,
    upgrade_id: &str,
    generation: crate::client::NativePeerDataGeneration,
    attempt: Option<&crate::client::NativeWebRtcCarrierAttempt>,
) -> bool {
    if !native_transport_upgrade_gate_matches(gate, upgrade_id) || gate.generation != generation {
        return false;
    }

    // The gate is inserted before packet-session activation and attempt
    // publication. A replay in that window must reuse the reservation; starting
    // a second responder would attach the same packet session twice and can
    // retire the winning carrier with DuplicateSession. Once an attempt is
    // published, only that exact non-terminal attempt keeps the reservation
    // live. Failure and terminal paths remove the gate before a later replay is
    // allowed to create a replacement.
    attempt.is_none_or(|attempt| {
        native_webrtc_duplicate_bootstrap_matches_live_attempt(
            gate,
            upgrade_id,
            generation,
            &attempt.upgrade_id,
            attempt.generation,
            attempt.channel.state(),
        )
    })
}

#[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
fn native_webrtc_duplicate_bootstrap_matches_live_attempt(
    gate: &crate::client::NativeTransportUpgradeGate,
    upgrade_id: &str,
    generation: crate::client::NativePeerDataGeneration,
    attempt_upgrade_id: &str,
    attempt_generation: crate::client::NativePeerDataGeneration,
    attempt_state: crate::transport::NativeWebRTCState,
) -> bool {
    native_transport_upgrade_gate_matches(gate, upgrade_id)
        && gate.generation == generation
        && attempt_upgrade_id == upgrade_id
        && attempt_generation == generation
        && !matches!(
            attempt_state,
            crate::transport::NativeWebRTCState::Failed
                | crate::transport::NativeWebRTCState::Closed
        )
}

#[cfg(not(target_arch = "wasm32"))]
fn reserve_native_transport_upgrade_gate(
    gates: &mut std::collections::HashMap<
        (String, IrohPathKind),
        crate::client::NativeTransportUpgradeGate,
    >,
    key: (String, IrohPathKind),
    upgrade_id: &str,
    generation: crate::client::NativePeerDataGeneration,
    policy_epoch: u64,
    peer_policy_epoch: u64,
    replace_existing: bool,
) -> bool {
    // WebRTC and MoQ are alternative physical carriers for one logical Iroh
    // generation. A capability replay can temporarily rank the fallback while
    // the preferred carrier is already proving; allowing both kind-specific
    // gates would let the faster candidate replace the same incumbent and
    // invalidate the preferred proof. Policy changes use a new epoch and may
    // deliberately prepare a different carrier over the selected incumbent.
    if gates.iter().any(|((connection_id, kind), current)| {
        connection_id == &key.0
            && *kind != key.1
            && current.generation == generation
            && current.policy_epoch == policy_epoch
            && current.peer_policy_epoch == peer_policy_epoch
    }) {
        return false;
    }
    if let Some(current) = gates.get(&key) {
        if current.policy_epoch == policy_epoch && current.peer_policy_epoch == peer_policy_epoch {
            if !replace_existing && current.generation == generation {
                return false;
            }
            if current.upgrade_id == upgrade_id {
                return replace_existing;
            }
        }
    }
    gates.insert(
        key,
        crate::client::NativeTransportUpgradeGate {
            upgrade_id: upgrade_id.to_string(),
            generation,
            policy_epoch,
            peer_policy_epoch,
        },
    );
    true
}

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

#[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
async fn remove_native_transport_upgrade_gate_for_generation(
    gates: &std::sync::Arc<
        tokio::sync::Mutex<
            std::collections::HashMap<
                (String, IrohPathKind),
                crate::client::NativeTransportUpgradeGate,
            >,
        >,
    >,
    key: &(String, IrohPathKind),
    upgrade_id: &str,
    generation: crate::client::NativePeerDataGeneration,
) -> bool {
    let mut gates = gates.lock().await;
    if gates
        .get(key)
        .is_some_and(|current| current.upgrade_id == upgrade_id && current.generation == generation)
    {
        gates.remove(key);
        true
    } else {
        false
    }
}

#[cfg(test)]
mod tests {
    #[cfg(not(target_arch = "wasm32"))]
    use super::effective_typescript_transport_capability;
    #[cfg(all(not(target_arch = "wasm32"), feature = "iroh-carrier-core"))]
    use crate::Client;

    #[cfg(all(
        not(target_arch = "wasm32"),
        any(feature = "transport-webrtc", feature = "transport-moq")
    ))]
    #[tokio::test]
    async fn carrier_policy_epoch_changes_only_with_effective_config() {
        let client = Client::new_with_app_tag(
            crate::test_constants::TEST_PROJECT_ID.to_string(),
            "test-app".to_string(),
            Box::new(|| None),
        );
        let initial = client.current_iroh_carrier_policy_epoch();
        let mut config = client.transport_config().await;

        client
            .update_transport_config(config.clone())
            .await
            .expect("idempotent carrier policy update");
        assert_eq!(client.current_iroh_carrier_policy_epoch(), initial);

        config.optimize_for = crate::route_policy::RouteOptimization::LowestLatency;
        client
            .update_transport_config(config)
            .await
            .expect("changed carrier policy update");
        assert_eq!(
            client.current_iroh_carrier_policy_epoch(),
            initial.saturating_add(1),
        );
    }

    #[cfg(all(not(target_arch = "wasm32"), feature = "iroh-carrier-core"))]
    #[tokio::test]
    async fn capability_withdrawal_is_serialized_with_the_native_proof_fence() {
        let client = Client::new_with_app_tag(
            crate::test_constants::TEST_PROJECT_ID.to_string(),
            "test-app".to_string(),
            Box::new(|| None),
        );
        let generation = crate::client::NativePeerDataGeneration {
            transport_stable_id: 7,
            transport_generation: 11,
            route_generation: 13,
        };
        let policy_epoch = client.current_iroh_carrier_policy_epoch();
        let peer_epoch = client.current_iroh_carrier_peer_policy_epoch("peer-a");
        {
            let mut capabilities = client.native_peer_transport_capabilities.write().await;
            capabilities.insert(
                "peer-a".to_string(),
                [crate::client::NativePeerTransportCapability::Moq]
                    .into_iter()
                    .collect(),
            );
            capabilities.insert(
                "peer-b".to_string(),
                [crate::client::NativePeerTransportCapability::Moq]
                    .into_iter()
                    .collect(),
            );
        }
        {
            let mut gates = client.native_transport_upgrade_gates.lock().await;
            for (connection_id, upgrade_id) in [("peer-a", "attempt-a"), ("peer-b", "attempt-b")] {
                gates.insert(
                    (connection_id.to_string(), crate::client::IrohPathKind::Moq),
                    crate::client::NativeTransportUpgradeGate {
                        upgrade_id: upgrade_id.to_string(),
                        generation,
                        policy_epoch,
                        peer_policy_epoch: peer_epoch,
                    },
                );
            }
        }

        let held_gate = client.native_transport_upgrade_gates.lock().await;
        let updating = client.clone();
        let update = tokio::spawn(async move {
            updating
                .maybe_handle_typescript_handshake_capabilities(
                    "peer-a",
                    None,
                    None,
                    &crate::native_protocol::TypeScriptHandshake {
                        action: Some("capability-update".to_string()),
                        session_token: None,
                        session_token_payload: None,
                        claimed_device_id: None,
                        capabilities: Some(crate::native_protocol::HandshakeFeatures {
                            carrier_schema: Some(2),
                            webrtc: Some(false),
                            moq: Some(false),
                            dedicated_carrier_control: Some(false),
                            dedicated_carrier_control_ready: Some(false),
                            ble: Some(false),
                            application_key_agreement: Some(false),
                        }),
                        application_key_agreement_public_key: None,
                    },
                )
                .await;
        });
        tokio::task::yield_now().await;
        assert!(
            client
                .native_peer_transport_capabilities
                .read()
                .await
                .get("peer-a")
                .is_some_and(|capabilities| capabilities
                    .contains(&crate::client::NativePeerTransportCapability::Moq)),
            "capability mutation must wait behind the proof/commit fence",
        );
        drop(held_gate);
        update.await.expect("capability update task");

        assert!(
            !client
                .native_iroh_carrier_upgrade_is_current(
                    "peer-a",
                    crate::client::IrohPathKind::Moq,
                    "attempt-a",
                    policy_epoch,
                )
                .await
        );
        assert!(
            client
                .native_iroh_carrier_upgrade_is_current(
                    "peer-b",
                    crate::client::IrohPathKind::Moq,
                    "attempt-b",
                    policy_epoch,
                )
                .await
        );
    }

    #[cfg(not(target_arch = "wasm32"))]
    use super::{
        application_key_reply_uses_fresh_signal, classify_native_ble_switch_admission,
        frame_iroh_peer_application_payload, native_ble_attempt_state_after_recovery_wake,
        native_ble_retry_delay, native_control_stream_rank, native_transport_upgrade_gate_matches,
        remove_native_transport_upgrade_gate, reserve_native_ble_attempt_state,
        reserve_native_transport_upgrade_gate, should_accept_native_ble_switch_request,
        should_drain_replaced_native_control_frame, should_initiate_native_ble_upgrade,
        should_replace_native_control_stream, should_send_typescript_capability_update,
        transition_native_ble_upgrade_gate_to_completion, typescript_application_key_reply_action,
        NativeBleSwitchAdmission, NativeControlStreamOwner,
        MAX_REPLACED_NATIVE_CONTROL_DRAIN_FRAMES, NATIVE_BLE_COMPLETION_GATE_PREFIX,
        NATIVE_BLE_MAX_UPGRADE_ATTEMPTS,
    };

    #[cfg(all(not(target_arch = "wasm32"), feature = "transport-moq"))]
    use super::{moq_retry_attempt, native_moq_peer_failure_code};

    #[cfg(all(
        not(target_arch = "wasm32"),
        any(feature = "transport-webrtc", feature = "transport-moq")
    ))]
    use super::native_carrier_duplicate_request_is_live;

    #[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
    use super::{
        dedicated_carrier_control_is_ready, native_webrtc_duplicate_bootstrap_is_reserved_or_live,
        native_webrtc_duplicate_bootstrap_matches_live_attempt,
        remove_native_transport_upgrade_gate_for_generation, webrtc_retry_attempt,
    };

    #[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
    #[test]
    fn native_dedicated_control_readiness_requires_current_admission_route() {
        assert!(!dedicated_carrier_control_is_ready(true, true, false));
        assert!(!dedicated_carrier_control_is_ready(true, false, true));
        assert!(dedicated_carrier_control_is_ready(true, true, true));
        assert!(!dedicated_carrier_control_is_ready(false, true, true));
    }

    #[cfg(all(
        not(target_arch = "wasm32"),
        any(feature = "transport-webrtc", feature = "transport-moq")
    ))]
    #[test]
    fn exact_live_native_carrier_replay_survives_base_generation_replacement() {
        let request = crate::iroh_carrier_bootstrap::CarrierBootstrapFrame::request_with_ids(
            crate::iroh_carrier_bootstrap::CarrierBootstrapKind::MoqDraft14,
            crate::iroh_carrier_bootstrap::CarrierGenerationFence {
                transport_stable_id: 3,
                transport_generation: 2,
                route_generation: 0,
            },
            1,
            [7; 16],
            [8; 16],
        );
        let mut competing = request.clone();
        competing.upgrade_id = "09".repeat(16);

        assert!(native_carrier_duplicate_request_is_live(
            &request, &request, false,
        ));
        assert!(!native_carrier_duplicate_request_is_live(
            &request, &request, true,
        ));
        assert!(!native_carrier_duplicate_request_is_live(
            &competing, &request, false,
        ));
    }

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

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

        // Confirmation frames are generation-bound. This unit intentionally
        // has no live Iroh/manager generation, so the response must fail
        // closed instead of manufacturing stable id 10. Live-generation ACK
        // and replacement fencing are covered by the native integration tests.
        let mut response = handshake;
        response.action = Some("response".to_string());
        assert!(client
            .maybe_negotiate_typescript_application_crypto(
                "connection-1",
                Some("remote-node"),
                None,
                &response,
            )
            .await
            .is_none());
        assert!(!client.connection_application_crypto_is_confirmed("connection-1", None));
        assert_eq!(
            client.session_admission_block_reason("connection-1"),
            Some((false, "application-crypto-confirmation-pending".to_string())),
        );
    }

    #[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
    #[tokio::test]
    async fn native_webrtc_terminal_cleanup_retires_only_its_exact_upgrade_gate() {
        let generation = crate::client::NativePeerDataGeneration {
            transport_stable_id: 10,
            transport_generation: 2,
            route_generation: 1,
        };
        let key = (
            "connection-1".to_string(),
            crate::client::IrohPathKind::WebRtc,
        );
        let gates =
            std::sync::Arc::new(tokio::sync::Mutex::new(std::collections::HashMap::from([
                (
                    key.clone(),
                    crate::client::NativeTransportUpgradeGate {
                        upgrade_id: "terminal-attempt".to_string(),
                        generation,
                        policy_epoch: 0,
                        peer_policy_epoch: 0,
                    },
                ),
            ])));

        assert!(remove_native_transport_upgrade_gate(&gates, &key, "terminal-attempt").await);
        assert!(gates.lock().await.is_empty());

        gates.lock().await.insert(
            key.clone(),
            crate::client::NativeTransportUpgradeGate {
                upgrade_id: "replacement-attempt".to_string(),
                generation,
                policy_epoch: 0,
                peer_policy_epoch: 0,
            },
        );
        assert!(
            !remove_native_transport_upgrade_gate(&gates, &key, "terminal-attempt").await,
            "a delayed terminal watcher must not retire a replacement attempt's gate",
        );
        assert_eq!(
            gates
                .lock()
                .await
                .get(&key)
                .map(|gate| gate.upgrade_id.as_str()),
            Some("replacement-attempt"),
        );

        let next_generation = crate::client::NativePeerDataGeneration {
            transport_stable_id: 11,
            transport_generation: 3,
            route_generation: 2,
        };
        assert!(
            !remove_native_transport_upgrade_gate_for_generation(
                &gates,
                &key,
                "replacement-attempt",
                next_generation,
            )
            .await
        );
        assert!(
            remove_native_transport_upgrade_gate_for_generation(
                &gates,
                &key,
                "replacement-attempt",
                generation,
            )
            .await
        );
        assert!(gates.lock().await.is_empty());
    }

    #[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
    #[test]
    fn native_webrtc_duplicate_bootstrap_requires_live_same_base_generation() {
        let base = crate::client::NativePeerDataGeneration {
            transport_stable_id: 10,
            transport_generation: 2,
            route_generation: 1,
        };
        let replacement = crate::client::NativePeerDataGeneration {
            transport_stable_id: 11,
            transport_generation: 3,
            route_generation: 0,
        };
        let gate = crate::client::NativeTransportUpgradeGate {
            upgrade_id: "upgrade-1".to_string(),
            generation: base,
            policy_epoch: 0,
            peer_policy_epoch: 0,
        };

        assert!(native_webrtc_duplicate_bootstrap_is_reserved_or_live(
            &gate,
            "upgrade-1",
            base,
            None,
        ));

        assert!(native_webrtc_duplicate_bootstrap_matches_live_attempt(
            &gate,
            "upgrade-1",
            base,
            "upgrade-1",
            base,
            crate::transport::NativeWebRTCState::Connecting,
        ));
        assert!(!native_webrtc_duplicate_bootstrap_matches_live_attempt(
            &gate,
            "upgrade-1",
            replacement,
            "upgrade-1",
            base,
            crate::transport::NativeWebRTCState::Connected,
        ));
        assert!(!native_webrtc_duplicate_bootstrap_matches_live_attempt(
            &gate,
            "upgrade-1",
            base,
            "upgrade-1",
            base,
            crate::transport::NativeWebRTCState::Closed,
        ));
        assert!(!native_webrtc_duplicate_bootstrap_matches_live_attempt(
            &gate,
            "upgrade-1",
            base,
            "other-upgrade",
            base,
            crate::transport::NativeWebRTCState::Connected,
        ));
    }

    #[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
    #[test]
    fn webrtc_carrier_retries_one_transient_candidate_failure() {
        for failure_code in [
            "carrier-base-generation-stale",
            "carrier-start-failed",
            "carrier-timeout",
            "carrier-proof-failed",
        ] {
            assert_eq!(webrtc_retry_attempt(failure_code, 1), Some(2));
            assert_eq!(webrtc_retry_attempt(failure_code, 2), None);
        }
        assert_eq!(webrtc_retry_attempt("peer-rejected-carrier", 1), None);
        assert_eq!(
            webrtc_retry_attempt("carrier-authorization-rejected", 1),
            None,
        );
    }

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

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

    #[test]
    fn application_key_reply_preserves_persistent_native_control_route() {
        assert!(!application_key_reply_uses_fresh_signal("response", true));
        assert!(!application_key_reply_uses_fresh_signal("ack", true));
        assert!(application_key_reply_uses_fresh_signal("response", false));
        assert!(application_key_reply_uses_fresh_signal("ack", false));
        assert!(!application_key_reply_uses_fresh_signal(
            "capability-update",
            false,
        ));
    }

    #[cfg(not(target_arch = "wasm32"))]
    #[test]
    fn key_only_transcript_preserves_last_canonical_transport_capability() {
        assert!(effective_typescript_transport_capability(
            false, false, true
        ));
        assert!(!effective_typescript_transport_capability(
            false, true, false
        ));
        assert!(!effective_typescript_transport_capability(
            true, false, true
        ));
        assert!(effective_typescript_transport_capability(true, true, false));
    }

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

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

    #[cfg(all(not(target_arch = "wasm32"), feature = "transport-moq"))]
    #[test]
    fn moq_carrier_retries_one_stale_base_generation() {
        assert_eq!(
            moq_retry_attempt("carrier-base-generation-stale", 1),
            Some(2),
        );
        assert_eq!(moq_retry_attempt("carrier-base-generation-stale", 2), None,);
        assert_eq!(moq_retry_attempt("carrier-proof-failed", 1), None,);
        assert!(crate::iroh_connection_policy::custom_carrier_base_allows(
            crate::client::IrohPathKind::Moq,
            crate::client::IrohPathKind::Moq,
            true,
        ));
        assert!(!crate::iroh_connection_policy::custom_carrier_base_allows(
            crate::client::IrohPathKind::Moq,
            crate::client::IrohPathKind::Moq,
            false,
        ));
    }

    #[cfg(all(not(target_arch = "wasm32"), feature = "transport-moq"))]
    #[test]
    fn moq_peer_failure_preserves_proof_code_without_broad_retry() {
        let (diagnostic, lifecycle) = native_moq_peer_failure_code(Some("carrier-proof-failed"));
        assert_eq!(diagnostic, "carrier-proof-failed");
        assert_eq!(lifecycle, "carrier-proof-failed");
        assert_eq!(moq_retry_attempt(lifecycle, 1), None);

        let (diagnostic, lifecycle) =
            native_moq_peer_failure_code(Some("carrier-adapter-specific"));
        assert_eq!(diagnostic, "carrier-adapter-specific");
        assert_eq!(lifecycle, "peer-rejected-carrier");
    }

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

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

    #[cfg(not(target_arch = "wasm32"))]
    #[test]
    fn native_ble_retry_is_bounded_jittered_and_reset_by_typed_recovery_wakes() {
        let generation_one = crate::client::NativePeerDataGeneration {
            transport_stable_id: 10,
            transport_generation: 2,
            route_generation: 1,
        };
        let generation_two = crate::client::NativePeerDataGeneration {
            transport_stable_id: 11,
            transport_generation: 3,
            route_generation: 0,
        };

        let (state, attempt) = reserve_native_ble_attempt_state(None, generation_one);
        assert_eq!(attempt, Some(1));
        assert_eq!(native_ble_retry_delay(1, 0).unwrap().as_millis(), 800);
        assert_eq!(native_ble_retry_delay(1, 40).unwrap().as_millis(), 1_200);
        let (state, attempt) = reserve_native_ble_attempt_state(Some(state), generation_one);
        assert_eq!(attempt, Some(2));
        assert_eq!(native_ble_retry_delay(2, 0).unwrap().as_millis(), 1_600);
        assert_eq!(native_ble_retry_delay(2, 40).unwrap().as_millis(), 2_400);
        let (state, attempt) = reserve_native_ble_attempt_state(Some(state), generation_one);
        assert_eq!(attempt, Some(3));
        assert_eq!(native_ble_retry_delay(3, 20), None);
        let (exhausted, attempt) = reserve_native_ble_attempt_state(Some(state), generation_one);
        assert_eq!(attempt, None);
        assert_eq!(exhausted.attempts, NATIVE_BLE_MAX_UPGRADE_ATTEMPTS);

        assert_eq!(
            native_ble_attempt_state_after_recovery_wake(exhausted, true),
            Some(exhausted),
        );
        let reset_after_wake = native_ble_attempt_state_after_recovery_wake(exhausted, false);
        let (reset, attempt) = reserve_native_ble_attempt_state(reset_after_wake, generation_one);
        assert_eq!(attempt, Some(1));
        assert_eq!(reset.generation, generation_one);

        let (reset, attempt) = reserve_native_ble_attempt_state(Some(exhausted), generation_two);
        assert_eq!(attempt, Some(1));
        assert_eq!(reset.generation, generation_two);
    }

    #[cfg(not(target_arch = "wasm32"))]
    #[test]
    fn native_carriers_cannot_race_the_same_logical_generation() {
        let generation = crate::client::NativePeerDataGeneration {
            transport_stable_id: 10,
            transport_generation: 2,
            route_generation: 1,
        };
        let connection_id = "connection-1".to_string();
        let mut gates = std::collections::HashMap::new();

        assert!(reserve_native_transport_upgrade_gate(
            &mut gates,
            (connection_id.clone(), crate::client::IrohPathKind::Moq),
            "moq-attempt",
            generation,
            7,
            0,
            false,
        ));
        assert!(!reserve_native_transport_upgrade_gate(
            &mut gates,
            (connection_id.clone(), crate::client::IrohPathKind::WebRtc),
            "webrtc-attempt",
            generation,
            7,
            0,
            false,
        ));
        assert_eq!(gates.len(), 1);

        assert!(reserve_native_transport_upgrade_gate(
            &mut gates,
            (
                "connection-2".to_string(),
                crate::client::IrohPathKind::WebRtc,
            ),
            "unrelated-peer-policy-change",
            generation,
            7,
            1,
            false,
        ));
        assert!(!reserve_native_transport_upgrade_gate(
            &mut gates,
            (connection_id.clone(), crate::client::IrohPathKind::WebRtc),
            "still-fenced-for-peer-one",
            generation,
            7,
            0,
            false,
        ));

        assert!(reserve_native_transport_upgrade_gate(
            &mut gates,
            (connection_id.clone(), crate::client::IrohPathKind::WebRtc),
            "peer-one-capability-changed",
            generation,
            7,
            1,
            false,
        ));

        assert!(reserve_native_transport_upgrade_gate(
            &mut gates,
            (connection_id, crate::client::IrohPathKind::WebRtc),
            "policy-switched-webrtc-attempt",
            generation,
            8,
            0,
            false,
        ));
        assert_eq!(gates.len(), 3);
    }

    #[cfg(not(target_arch = "wasm32"))]
    #[tokio::test]
    async fn native_ble_upgrade_gate_cleanup_is_generation_fenced() {
        let generation_one = crate::client::NativePeerDataGeneration {
            transport_stable_id: 10,
            transport_generation: 2,
            route_generation: 1,
        };
        let generation_two = crate::client::NativePeerDataGeneration {
            transport_stable_id: 11,
            transport_generation: 3,
            route_generation: 0,
        };
        let gates =
            std::sync::Arc::new(tokio::sync::Mutex::new(std::collections::HashMap::from([
                (
                    ("connection-1".to_string(), crate::client::IrohPathKind::Ble),
                    crate::client::NativeTransportUpgradeGate {
                        upgrade_id: "upgrade-new".to_string(),
                        generation: generation_one,
                        policy_epoch: 0,
                        peer_policy_epoch: 0,
                    },
                ),
            ])));
        let key = ("connection-1".to_string(), crate::client::IrohPathKind::Ble);

        assert!(!remove_native_transport_upgrade_gate(&gates, &key, "upgrade-old").await);
        assert_eq!(
            gates
                .lock()
                .await
                .get(&key)
                .map(|gate| gate.upgrade_id.as_str()),
            Some("upgrade-new")
        );

        assert!(!reserve_native_transport_upgrade_gate(
            &mut *gates.lock().await,
            key.clone(),
            "same-generation-retry",
            generation_one,
            0,
            0,
            false,
        ));
        assert!(reserve_native_transport_upgrade_gate(
            &mut *gates.lock().await,
            key.clone(),
            "same-generation-remote-retry",
            generation_one,
            0,
            0,
            true,
        ));
        assert!(reserve_native_transport_upgrade_gate(
            &mut *gates.lock().await,
            key.clone(),
            "replacement-generation",
            generation_two,
            0,
            0,
            false,
        ));
        assert_eq!(
            gates
                .lock()
                .await
                .get(&key)
                .map(|gate| (gate.upgrade_id.as_str(), gate.generation)),
            Some(("replacement-generation", generation_two)),
        );

        assert!(reserve_native_transport_upgrade_gate(
            &mut *gates.lock().await,
            key.clone(),
            "policy-epoch-replacement",
            generation_two,
            1,
            0,
            false,
        ));
        assert_eq!(
            gates
                .lock()
                .await
                .get(&key)
                .map(|gate| (gate.upgrade_id.as_str(), gate.policy_epoch)),
            Some(("policy-epoch-replacement", 1)),
        );

        assert!(!remove_native_transport_upgrade_gate(&gates, &key, "upgrade-new").await);
        assert!(
            !remove_native_transport_upgrade_gate(&gates, &key, "replacement-generation").await
        );
        assert!(
            remove_native_transport_upgrade_gate(&gates, &key, "policy-epoch-replacement").await
        );
        assert!(!gates.lock().await.contains_key(&key));

        let completion_gate = format!("{NATIVE_BLE_COMPLETION_GATE_PREFIX}upgrade-completing");
        gates.lock().await.insert(
            key.clone(),
            crate::client::NativeTransportUpgradeGate {
                upgrade_id: completion_gate.clone(),
                generation: generation_two,
                policy_epoch: 0,
                peer_policy_epoch: 0,
            },
        );
        let completion = gates.lock().await.get(&key).cloned().unwrap();
        assert!(native_transport_upgrade_gate_matches(
            &completion,
            "upgrade-completing",
        ));
        assert!(
            !remove_native_transport_upgrade_gate(&gates, &key, "upgrade-completing").await,
            "the response timer must not retire an upgrade after switch-ready",
        );
        assert!(
            remove_native_transport_upgrade_gate(&gates, &key, completion_gate.as_str()).await,
            "the completion owner must retire its phase-specific gate",
        );
    }

    #[cfg(not(target_arch = "wasm32"))]
    #[test]
    fn native_ble_switch_ready_requires_the_exact_echoed_generation() {
        let generation = crate::client::NativePeerDataGeneration {
            transport_stable_id: 10,
            transport_generation: 2,
            route_generation: 1,
        };
        let mut gate = crate::client::NativeTransportUpgradeGate {
            upgrade_id: "upgrade-exact".to_string(),
            generation,
            policy_epoch: 7,
            peer_policy_epoch: 0,
        };
        let mismatched_generation = crate::client::NativePeerDataGeneration {
            route_generation: generation.route_generation + 1,
            ..generation
        };

        assert_eq!(
            transition_native_ble_upgrade_gate_to_completion(
                &mut gate,
                "upgrade-exact",
                mismatched_generation,
            ),
            None,
        );
        assert_eq!(gate.upgrade_id, "upgrade-exact");
        assert_eq!(gate.generation, generation);
        assert_eq!(gate.policy_epoch, 7);

        assert_eq!(
            transition_native_ble_upgrade_gate_to_completion(
                &mut gate,
                "upgrade-exact",
                generation,
            ),
            Some(generation),
        );
        assert_eq!(
            gate.upgrade_id,
            format!("{NATIVE_BLE_COMPLETION_GATE_PREFIX}upgrade-exact"),
        );
        assert_eq!(gate.generation, generation);
        assert_eq!(gate.policy_epoch, 7);
    }

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

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

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

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

    #[cfg(not(target_arch = "wasm32"))]
    #[test]
    fn replaced_native_control_stream_drain_is_strictly_bounded() {
        assert!(should_drain_replaced_native_control_frame(0));
        assert!(should_drain_replaced_native_control_frame(
            MAX_REPLACED_NATIVE_CONTROL_DRAIN_FRAMES - 1
        ));
        assert!(!should_drain_replaced_native_control_frame(
            MAX_REPLACED_NATIVE_CONTROL_DRAIN_FRAMES
        ));
    }

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

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