openrtc/

client.rs

use super::coordination::room::{GatewayRequiredRoomBackend, RoomBackend};
use crate::signaling::{GatewayRequiredSignalingBackend, SignalingBackend};
use iroh::Endpoint;
use iroh_tickets::endpoint::EndpointTicket;
use std::collections::{HashMap, HashSet};
use std::str::FromStr;
#[cfg(any(
    not(target_arch = "wasm32"),
    feature = "transport-webrtc",
    feature = "transport-moq"
))]
use std::sync::atomic::Ordering;
use std::sync::atomic::{AtomicBool, AtomicU64};
use std::sync::{Arc, Mutex, RwLock as StdRwLock};
use tokio::sync::RwLock;

#[cfg(not(target_arch = "wasm32"))]
use crate::client::scope_classifier::{default_classifier, ScopeClassifier};

#[cfg(target_arch = "wasm32")]
fn wasm_init_log(stage: &str) {
    let now = js_sys::Date::now();
    web_sys::console::log_1(&wasm_bindgen::JsValue::from_str(&format!(
        "[OPENRTC][WASM-INIT] ts_ms={:.0} stage={}",
        now, stage
    )));
}

#[cfg(target_arch = "wasm32")]
use crate::wasm_node::{AcceptEvent, ConnectEvent, IrohWasmNode};

#[cfg(not(target_arch = "wasm32"))]
use crate::native_node::{AcceptEvent, ConnectEvent, IncomingStream, IrohNativeNode};

#[cfg(not(target_arch = "wasm32"))]
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct EndpointHandle {
    pub node_id: String,
    pub node_addr: String,
}

#[cfg(not(target_arch = "wasm32"))]
pub struct BiStream {
    pub send: crate::application_crypto_streams::PeerSendStream,
    pub recv: crate::application_crypto_streams::PeerRecvStream,
    pub id: String,
}

/// Result of routing a newly accepted native bidirectional stream through the
/// Rust admission authority. Pending connections must complete SDK-owned token
/// admission before an application or webview can observe their streams.
#[cfg(not(target_arch = "wasm32"))]
pub enum BiStreamDisposition {
    Consumed,
    Forward {
        send: iroh::endpoint::SendStream,
        recv: iroh::endpoint::RecvStream,
        /// Bytes consumed only to classify an out-of-order encrypted product
        /// frame while reciprocal key confirmation was in flight. The native
        /// host wrapper must feed them into the application-crypto decoder
        /// before reading the remaining QUIC stream.
        recv_prefix: Vec<u8>,
    },
}

#[cfg(not(target_arch = "wasm32"))]
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct NativePeerDataEvent {
    pub connection_id: String,
    pub remote_node_id: Option<String>,
    pub transport: String,
    pub transport_stable_id: u64,
    pub transport_generation: u64,
    pub route_generation: u64,
    pub payload: Vec<u8>,
}

#[cfg(not(target_arch = "wasm32"))]
#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
#[serde(rename_all = "camelCase")]
pub(crate) struct NativePeerDataGeneration {
    pub transport_stable_id: u64,
    pub transport_generation: u64,
    pub route_generation: u64,
}

#[cfg(target_arch = "wasm32")]
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) struct WasmPeerDataGeneration {
    pub transport_stable_id: u64,
    pub transport_generation: u64,
    pub route_generation: u64,
}

#[cfg(not(target_arch = "wasm32"))]
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) struct NativeBleUpgradeAttemptState {
    pub generation: NativePeerDataGeneration,
    pub attempts: u8,
}

#[cfg(not(target_arch = "wasm32"))]
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) struct NativeTransportUpgradeGate {
    pub upgrade_id: String,
    pub generation: NativePeerDataGeneration,
    pub policy_epoch: u64,
    pub peer_policy_epoch: u64,
}

#[cfg(target_arch = "wasm32")]
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) struct WasmTransportUpgradeGate {
    pub upgrade_id: String,
    /// The logical base generation this browser-carrier attempt is permitted
    /// to replace. A duplicate capability/ready frame must never supersede a
    /// live proof for this same base.
    pub generation: WasmPeerDataGeneration,
    pub policy_epoch: u64,
    pub peer_policy_epoch: u64,
}

/// Outcome of asking to own a browser-carrier attempt fence.
///
/// This is deliberately a pure decision boundary so a replay cannot be
/// mistaken for a new lifecycle owner by the carrier control handlers.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[cfg(any(
    test,
    all(
        target_arch = "wasm32",
        any(feature = "transport-webrtc", feature = "transport-moq")
    )
))]
pub(crate) enum WasmCarrierUpgradeReservation {
    Reserved,
    CurrentReplay,
    Busy,
}

/// State of a previously reserved browser-carrier attempt at a proof boundary.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[cfg(any(
    test,
    all(
        target_arch = "wasm32",
        any(feature = "transport-webrtc", feature = "transport-moq")
    )
))]
pub(crate) enum WasmCarrierUpgradeFence {
    Current(u64),
    Retired,
    Superseded,
    PolicyChanged,
    BaseGenerationChanged,
}

/// Preserve the two recoverable peer responses that carry lifecycle meaning.
/// All other peer failure strings collapse to one bounded public-internal
/// category so adapter diagnostics cannot become transport policy.
#[cfg(any(
    test,
    all(
        target_arch = "wasm32",
        any(feature = "transport-webrtc", feature = "transport-moq")
    )
))]
pub(crate) fn wasm_peer_carrier_failure_code(failure_code: Option<&str>) -> &'static str {
    match failure_code {
        Some("carrier-base-generation-stale") => "carrier-base-generation-stale",
        Some("carrier-base-not-ready") => "carrier-base-not-ready",
        _ => "peer-rejected-carrier",
    }
}

/// A transient peer response means the exact fenced browser attempt should be
/// eligible for the next Rust settlement/capability wake. It must not create a
/// new upgrade ID or schedule an independent retry owner.
#[cfg(any(
    test,
    all(
        target_arch = "wasm32",
        any(feature = "transport-webrtc", feature = "transport-moq")
    )
))]
pub(crate) fn should_rearm_wasm_carrier_event_retry(
    failure_code: &str,
    initiator: bool,
    completion_started: bool,
) -> bool {
    initiator && !completion_started && failure_code == "carrier-base-not-ready"
}

/// Classify a reservation without mutating the gate. The three generation
/// fields are kept as a tuple here so this policy remains native-testable even
/// though browser carrier attempts themselves are WASM-only.
#[cfg(any(
    test,
    all(
        target_arch = "wasm32",
        any(feature = "transport-webrtc", feature = "transport-moq")
    )
))]
pub(crate) fn classify_wasm_carrier_upgrade_reservation(
    current: Option<(&str, (u64, u64, u64), u64)>,
    competing_kind_current: bool,
    upgrade_id: &str,
    generation: (u64, u64, u64),
    policy_epoch: u64,
) -> WasmCarrierUpgradeReservation {
    if competing_kind_current {
        return WasmCarrierUpgradeReservation::Busy;
    }
    match current {
        Some((current_id, current_generation, current_policy_epoch))
            if current_generation == generation
                && current_id == upgrade_id
                && current_policy_epoch == policy_epoch =>
        {
            WasmCarrierUpgradeReservation::CurrentReplay
        }
        Some((_, current_generation, _)) if current_generation == generation => {
            WasmCarrierUpgradeReservation::Busy
        }
        _ => WasmCarrierUpgradeReservation::Reserved,
    }
}

#[cfg(any(
    test,
    all(
        target_arch = "wasm32",
        any(feature = "transport-webrtc", feature = "transport-moq")
    )
))]
pub(crate) fn classify_wasm_carrier_upgrade_fence(
    current: Option<(&str, (u64, u64, u64), u64)>,
    upgrade_id: &str,
    generation: (u64, u64, u64),
    policy_epoch: u64,
) -> WasmCarrierUpgradeFence {
    match current {
        None => WasmCarrierUpgradeFence::Retired,
        Some((_, current_generation, _)) if current_generation != generation => {
            WasmCarrierUpgradeFence::BaseGenerationChanged
        }
        Some((current_id, _, _)) if current_id != upgrade_id => WasmCarrierUpgradeFence::Superseded,
        Some((_, _, current_policy_epoch)) if current_policy_epoch != policy_epoch => {
            WasmCarrierUpgradeFence::PolicyChanged
        }
        Some((_, _, current_policy_epoch)) => {
            WasmCarrierUpgradeFence::Current(current_policy_epoch)
        }
    }
}

/// Remote capability authority is connection-scoped. Preserve a matching
/// authenticated attempt when its first canonical advertisement arrives
/// slightly later, but invalidate every later capability change for that peer.
#[cfg(any(test, feature = "iroh-carrier-core"))]
pub(crate) fn should_bump_remote_carrier_peer_policy_epoch(
    capabilities_were_known: bool,
    capabilities_changed: bool,
    matching_first_attempt: bool,
) -> bool {
    capabilities_changed && (capabilities_were_known || !matching_first_attempt)
}

#[cfg(test)]
mod wasm_carrier_upgrade_gate_tests {
    use super::*;

    const BASE: (u64, u64, u64) = (7, 11, 13);

    #[test]
    fn delayed_proof_duplicate_replay_keeps_the_original_attempt_fenced() {
        let current = Some(("attempt-a", BASE, 23));

        assert_eq!(
            classify_wasm_carrier_upgrade_reservation(current, false, "attempt-a", BASE, 23),
            WasmCarrierUpgradeReservation::CurrentReplay,
        );
        assert_eq!(
            classify_wasm_carrier_upgrade_reservation(current, false, "attempt-b", BASE, 23),
            WasmCarrierUpgradeReservation::Busy,
        );
        assert_eq!(
            classify_wasm_carrier_upgrade_fence(current, "attempt-a", BASE, 23),
            WasmCarrierUpgradeFence::Current(23),
        );
    }

    #[test]
    fn another_carrier_cannot_overtake_a_proving_browser_generation() {
        assert_eq!(
            classify_wasm_carrier_upgrade_reservation(None, true, "webrtc-attempt", BASE, 23,),
            WasmCarrierUpgradeReservation::Busy,
        );
    }

    #[test]
    fn remote_capability_epoch_preserves_only_a_matching_first_observation() {
        assert!(!should_bump_remote_carrier_peer_policy_epoch(
            false, true, true,
        ));
        assert!(should_bump_remote_carrier_peer_policy_epoch(
            false, true, false,
        ));
        assert!(should_bump_remote_carrier_peer_policy_epoch(
            true, true, true,
        ));
        assert!(!should_bump_remote_carrier_peer_policy_epoch(
            true, false, false,
        ));
    }

    #[test]
    fn peer_base_readiness_failure_remains_recoverable() {
        assert_eq!(
            wasm_peer_carrier_failure_code(Some("carrier-base-not-ready")),
            "carrier-base-not-ready",
        );
        assert_eq!(
            wasm_peer_carrier_failure_code(Some("carrier-base-generation-stale")),
            "carrier-base-generation-stale",
        );
        assert_eq!(
            wasm_peer_carrier_failure_code(Some("carrier-authorization-rejected")),
            "peer-rejected-carrier",
        );
        assert!(should_rearm_wasm_carrier_event_retry(
            "carrier-base-not-ready",
            true,
            false,
        ));
        assert!(!should_rearm_wasm_carrier_event_retry(
            "peer-rejected-carrier",
            true,
            false,
        ));
        assert!(!should_rearm_wasm_carrier_event_retry(
            "carrier-base-not-ready",
            false,
            false,
        ));
        assert!(!should_rearm_wasm_carrier_event_retry(
            "carrier-base-not-ready",
            true,
            true,
        ));
    }

    #[test]
    fn carrier_fence_distinguishes_lifecycle_retirement_from_policy_change() {
        assert_eq!(
            classify_wasm_carrier_upgrade_fence(None, "attempt-a", BASE, 23),
            WasmCarrierUpgradeFence::Retired,
        );
        assert_eq!(
            classify_wasm_carrier_upgrade_fence(
                Some(("attempt-a", BASE, 23)),
                "attempt-a",
                BASE,
                24,
            ),
            WasmCarrierUpgradeFence::PolicyChanged,
        );
        assert_eq!(
            classify_wasm_carrier_upgrade_fence(
                Some(("attempt-b", BASE, 23)),
                "attempt-a",
                BASE,
                23,
            ),
            WasmCarrierUpgradeFence::Superseded,
        );
        assert_eq!(
            classify_wasm_carrier_upgrade_fence(
                Some(("attempt-a", (17, 19, 23), 23)),
                "attempt-a",
                BASE,
                23,
            ),
            WasmCarrierUpgradeFence::BaseGenerationChanged,
        );
    }
}

#[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
pub(crate) struct NativeWebRtcCarrierAttempt {
    pub upgrade_id: String,
    pub bootstrap: crate::iroh_carrier_bootstrap::CarrierBootstrapFrame,
    pub generation: NativePeerDataGeneration,
    pub remote_endpoint_id: iroh::EndpointId,
    pub channel: Arc<crate::transport::WebRtcDataChannel>,
    pub pump: tokio::sync::Mutex<Option<crate::native_webrtc_carrier::NativeWebRtcCarrierSession>>,
}

#[cfg(all(not(target_arch = "wasm32"), feature = "transport-moq"))]
pub(crate) struct NativeMoqCarrierAttempt {
    pub upgrade_id: String,
    pub bootstrap: crate::iroh_carrier_bootstrap::CarrierBootstrapFrame,
    pub generation: NativePeerDataGeneration,
    pub remote_endpoint_id: iroh::EndpointId,
    pub session: Arc<crate::transport::NativeMoQSession>,
    pub pump: tokio::sync::Mutex<Option<crate::native_moq_carrier::NativeMoqCarrierSession>>,
}

#[cfg(all(not(target_arch = "wasm32"), feature = "transport-moq"))]
impl std::fmt::Debug for NativeMoqCarrierAttempt {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        formatter
            .debug_struct("NativeMoqCarrierAttempt")
            .field("upgrade_id", &self.upgrade_id)
            .field("generation", &self.generation)
            .field("remote_endpoint_id", &self.remote_endpoint_id)
            .finish_non_exhaustive()
    }
}

#[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
impl std::fmt::Debug for NativeWebRtcCarrierAttempt {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        formatter
            .debug_struct("NativeWebRtcCarrierAttempt")
            .field("upgrade_id", &self.upgrade_id)
            .field("generation", &self.generation)
            .field("remote_endpoint_id", &self.remote_endpoint_id)
            .finish_non_exhaustive()
    }
}

/// Exact physical owner of one SDK native-main stream. The logical connection
/// id intentionally remains the map key; this token prevents stale physical
/// generations and duplicate QUIC streams from mutating the current entry.
#[cfg(not(target_arch = "wasm32"))]
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) struct NativeControlStreamOwner {
    pub transport_stable_id: u64,
    pub stream_rank: u64,
}

#[cfg(not(target_arch = "wasm32"))]
#[derive(Clone)]
pub(crate) struct NativeControlStreamEntry {
    pub owner: NativeControlStreamOwner,
    pub endpoint_id: iroh::EndpointId,
    pub send: Arc<tokio::sync::Mutex<iroh::endpoint::SendStream>>,
}

/// Last valid OpenRTC protocol activity observed from one physical Iroh
/// generation.
///
/// QUIC health probes use a separate uni-stream. A mobile runtime can process
/// admission or control frames on an already-open bi-stream while that
/// diagnostic probe is delayed or dropped. Recording the protocol activity
/// here lets the lifecycle owner use that stronger positive evidence without
/// allowing an old physical generation to keep its replacement alive.
#[cfg(not(target_arch = "wasm32"))]
#[derive(Debug, Clone, Copy)]
pub(crate) struct NativeTransportProtocolActivity {
    pub transport_stable_id: u64,
    pub observed_at: std::time::Instant,
}

#[cfg(not(target_arch = "wasm32"))]
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub(crate) struct NativeAdmissionStreamContracts {
    pub inbound: Option<crate::native_protocol::TokenStreamContract>,
    pub outbound: Option<crate::native_protocol::TokenStreamContract>,
}

#[cfg(not(target_arch = "wasm32"))]
#[derive(Debug, Clone)]
pub(crate) struct CachedManagedScopeTicket {
    pub scope: crate::session_token::GrantScope,
    pub token: String,
    pub max_connections: u32,
    pub compound_ticket: String,
    pub iroh_ticket: String,
}

#[cfg(not(target_arch = "wasm32"))]
#[derive(Debug, Clone)]
pub(crate) struct NativeRouteRepairCredential {
    pub token: String,
    pub token_payload: String,
    pub authoritative_device_id: String,
}

/// Last explicit product capability accepted for a known remote device.
///
/// Unlike provider `user-device` credentials, this intent is shared by native
/// and WASM runtimes. It lets the existing physical-transport owner restore
/// the exact product scope after a network-generation replacement without
/// waiting for an application or coordination callback to dial again.
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) struct ScopedRouteRepairCredential {
    pub token: String,
    pub scope: String,
    pub max_connections: u32,
    pub expires_at_ms: Option<u64>,
    pub authoritative_device_id: String,
}

/// Opaque, short-lived native trust evidence supplied by a developer backend.
///
/// OpenRTC does not interpret the token. The active control plane verifies and
/// exchanges it while callers and host bridges remain provider-neutral.
#[allow(dead_code)]
#[derive(Clone)]
pub struct Client {
    app_tag: String,
    broadcasts: crate::broadcast::Broadcasts,
    signaling: Arc<dyn SignalingBackend>,
    pub room: Arc<dyn RoomBackend>,
    node_id: Arc<RwLock<Option<String>>>,
    pub iroh_endpoint: Arc<RwLock<Option<Endpoint>>>,
    #[cfg(target_arch = "wasm32")]
    pub iroh_node: Arc<RwLock<Option<IrohWasmNode>>>,
    #[cfg(not(target_arch = "wasm32"))]
    pub iroh_node: Arc<RwLock<Option<IrohNativeNode>>>,
    #[cfg(not(target_arch = "wasm32"))]
    native_application_streams: async_channel::Sender<IncomingStream>,
    #[cfg(not(target_arch = "wasm32"))]
    native_application_streams_receiver: async_channel::Receiver<IncomingStream>,
    pub connection_manager: Arc<crate::connection_manager::ConnectionManager>,
    #[cfg(not(target_arch = "wasm32"))]
    native_device_identity: Arc<RwLock<Option<crate::native_device::NativeDeviceIdentity>>>,
    #[cfg(not(target_arch = "wasm32"))]
    native_device_base_dir: Arc<RwLock<Option<std::path::PathBuf>>>,
    #[cfg(not(target_arch = "wasm32"))]
    native_device_identity_init_guard: Arc<tokio::sync::Mutex<()>>,
    #[cfg(not(target_arch = "wasm32"))]
    native_device_updates:
        tokio::sync::broadcast::Sender<crate::native_device::NativeDeviceIdentity>,
    #[cfg(not(target_arch = "wasm32"))]
    native_connection_state_updates: tokio::sync::broadcast::Sender<StateSnapshot>,
    #[cfg(not(target_arch = "wasm32"))]
    native_peer_data_updates: tokio::sync::broadcast::Sender<NativePeerDataEvent>,
    auto_connect_loop_key: Arc<Mutex<Option<(String, String)>>>,
    auto_connect_generation: Arc<AtomicU64>,
    /// Generation-local input for the native provider-backed auto-connect
    /// actor. Admission and transport callbacks notify the existing owner;
    /// they never start a second polling loop or perform a local redial.
    #[cfg(not(target_arch = "wasm32"))]
    native_auto_connect_wake: Arc<tokio::sync::Notify>,
    /// Provider-neutral native desired-peer input. Hosted coordination adapters
    /// submit revisioned snapshots; Rust remains the only dial/retry owner.
    #[cfg(not(target_arch = "wasm32"))]
    external_desired_peer_actor:
        Arc<tokio::sync::Mutex<auto_connect_impl::NativeExternalAutoConnectActorState>>,
    /// Device IDs excluded from auto-connect. Session-scoped: cleared on restart.
    /// Serializes every exclusion mutation so an authenticated peer resume
    /// cannot erase a concurrent local disconnect intent.
    auto_connect_exclusion_owner: Arc<Mutex<()>>,
    auto_connect_excluded: Arc<Mutex<HashSet<String>>>,
    /// Subset of exclusions installed because the peer explicitly disconnected.
    ///
    /// A fresh authenticated user-device admission from that peer is an explicit
    /// reconnect request and may clear this subset. Locally requested exclusions
    /// are deliberately not recorded here, so a remote dial cannot override the
    /// local user's disconnect choice.
    auto_connect_peer_requested_excluded: Arc<Mutex<HashSet<String>>>,
    /// Node-id aliases for session-scoped auto-connect exclusions, keyed by
    /// canonical device id. These are local-only and must not be published into
    /// the coordination roster's `excludedPeers`, which is a device-id contract.
    auto_connect_excluded_node_aliases: Arc<Mutex<HashMap<String, HashSet<String>>>>,
    /// Session token registry — gates incoming connections during short-lived
    /// sessions (e.g. share page). When non-empty, incoming handshakes must
    /// carry a valid token. Shared across WASM and native paths.
    pub session_token_registry: Arc<crate::session_token::SessionTokenRegistry>,
    /// Directional proof that this runtime validated the remote token on the
    /// current physical Iroh generation. Logical admission survives reconnect,
    /// but a replacement leg must establish a fresh SDK control route.
    #[cfg(not(target_arch = "wasm32"))]
    inbound_session_admission_transport_ids: Arc<StdRwLock<HashMap<String, u64>>>,
    /// Trust-validated offline candidates that require a fresh proof before
    /// any product stream on their deterministic connection can be routed.
    #[cfg(not(target_arch = "wasm32"))]
    offline_admission_requirements:
        Arc<StdRwLock<HashMap<String, crate::offline::OfflineCandidateHandoff>>>,
    /// Accepted offline proofs fenced to their exact physical generation.
    #[cfg(not(target_arch = "wasm32"))]
    offline_admission_transport_ids: Arc<StdRwLock<HashMap<String, u64>>>,
    /// Host-installed offline trust/signing owner. Discovery reports local
    /// addresses into this state; it never owns dialing or admission.
    #[cfg(not(target_arch = "wasm32"))]
    pub(crate) offline_runtime:
        Arc<tokio::sync::Mutex<Option<crate::offline::InstalledOfflineRuntime>>>,
    /// At most one bounded offline proof request per logical connection and
    /// physical generation. Retry remains owned by the desired-peer actor.
    #[cfg(not(target_arch = "wasm32"))]
    pub(crate) offline_proof_attempts: Arc<tokio::sync::Mutex<HashMap<String, u64>>>,
    /// Declared lifetime of the stream that established admission for this
    /// logical connection. The contract determines which generation-bound
    /// proofs gate readiness; it never owns reconnection or settlement.
    #[cfg(not(target_arch = "wasm32"))]
    native_admission_stream_contracts:
        Arc<StdRwLock<HashMap<String, NativeAdmissionStreamContracts>>>,
    /// Host approval observed by this dialer, fenced to the physical Iroh leg
    /// and token that produced it. Local trusted-device admission is a separate
    /// fact and must never suppress remote token presentation.
    remote_session_admission_proofs: Arc<StdRwLock<HashMap<String, RemoteSessionAdmissionProof>>>,
    /// Rust-owned transcript for a reciprocal admission response that is
    /// awaiting an ACK on one exact stream and physical transport generation.
    /// Host adapters may relay observations, but cannot supply the token or
    /// scope at commit time.
    pending_inline_reciprocal_admissions:
        Arc<StdRwLock<HashMap<String, PendingInlineReciprocalAdmission>>>,
    /// Last capability token approved by the remote host for each logical
    /// connection. This is application-security state, not transport state:
    /// route replacement preserves a same-token epoch, while token rotation
    /// must retire the previous application key before product traffic resumes.
    outbound_application_security_epoch_fingerprints: Arc<StdRwLock<HashMap<String, String>>>,
    /// Valid token presentations may ask this runtime to restore the reverse
    /// directional proof. The existing auto-connect actor consumes this typed
    /// input; it does not introduce a second retry or lifecycle owner.
    #[cfg(not(target_arch = "wasm32"))]
    pending_reciprocal_session_admission_requests: Arc<StdRwLock<HashSet<String>>>,
    /// Latest validated managed credential advertised by each remote native
    /// node. The native admission responder uses this desired-state input to
    /// answer an inline reciprocal request without opening a second stream.
    #[cfg(not(target_arch = "wasm32"))]
    native_route_repair_credentials: Arc<StdRwLock<HashMap<String, NativeRouteRepairCredential>>>,
    /// Explicit product admission intent keyed by authoritative remote node.
    /// Only non-`user-device` scopes are retained here; provider auto-connect
    /// remains independently owned by its desired-peer actor.
    scoped_route_repair_credentials: Arc<StdRwLock<HashMap<String, ScopedRouteRepairCredential>>>,
    connection_application_crypto_keys:
        Arc<StdRwLock<HashMap<String, [u8; crate::application_crypto::APPLICATION_KEY_BYTES]>>>,
    /// Serializes key agreement, key, requirement, sequence, and generation-
    /// bound confirmation as one application-security state. The maps remain
    /// separate for their existing read shapes, but a handshake commit is
    /// observed atomically across all five.
    connection_application_crypto_state: Arc<StdRwLock<()>>,
    connection_application_crypto_required: Arc<StdRwLock<HashSet<String>>>,
    /// Product opt-in requiring every trusted native user-device route to
    /// complete reciprocal application key agreement before product streams
    /// become routable.
    #[cfg(not(target_arch = "wasm32"))]
    trusted_user_device_application_crypto_required: Arc<AtomicBool>,
    /// Automatic key agreement is not routable until the remote peer has
    /// acknowledged the exact connection key. Key presence alone is only a
    /// local derivation fact and can race the reciprocal handshake.
    connection_application_crypto_confirmed: Arc<
        StdRwLock<
            HashMap<
                String,
                crate::client::application_crypto_impl::ConnectionApplicationCryptoConfirmation,
            >,
        >,
    >,
    /// Wakes native ingress tasks when any fact in the current application
    /// route changes. Independent QUIC streams have no cross-stream delivery
    /// order, so an application stream may arrive immediately before either a
    /// directional admission proof or the reciprocal key acknowledgement. The
    /// Rust admission owner uses this typed wake to hold that stream behind the
    /// exact transport and security epoch without polling.
    #[cfg(not(target_arch = "wasm32"))]
    connection_application_route_updates: Arc<tokio::sync::Notify>,
    connection_application_crypto_outbound_sequences: Arc<StdRwLock<HashMap<String, u64>>>,
    connection_application_key_agreements:
        Arc<StdRwLock<HashMap<String, crate::key_agreement::KeyAgreement>>>,
    /// Native-managed admission grants that must survive frontend refreshes
    /// for the lifetime of the desktop app process. The user-device scope is
    /// intentionally backend-owned so auto-connect presence can keep
    /// publishing the same tokenized ticket until explicit revoke/app restart.
    ///
    /// This cache is read-heavy (ticket lookups/refreshes) with infrequent
    /// writes (revoke/rehydrate), so a standard RwLock avoids unnecessary
    /// exclusive locking overhead from Mutex.
    #[cfg(not(target_arch = "wasm32"))]
    managed_scope_tickets: Arc<StdRwLock<HashMap<String, CachedManagedScopeTicket>>>,
    /// Latest ticket-derived Iroh address for each remote endpoint.
    ///
    /// The coordination gateway publishes fresh device tickets. Native stream
    /// recovery uses this cache to redial Iroh directly when a cached
    /// connection/stream has gone stale.
    known_endpoint_addrs: Arc<RwLock<HashMap<String, iroh::EndpointAddr>>>,
    /// Session-scoped, authoritative mapping from the current remote Iroh node
    /// to its durable device id. Physical connection records are deliberately
    /// ephemeral and may be retired before an inbound replacement leg arrives;
    /// logical identity must survive that transport-generation boundary.
    known_device_ids_by_node: Arc<StdRwLock<HashMap<String, String>>>,
    /// Wakes bounded known-device admission calls when either half of the
    /// Rust-owned device-to-endpoint lookup changes. Callers wait on this
    /// revision instead of installing an application-layer roster watcher.
    known_device_endpoint_revision: tokio::sync::watch::Sender<u64>,
    /// IO-less sparse avenue owner shared by native and WASM. It reduces
    /// provider rosters and verifies forwarded broadcast envelopes but never
    /// opens a socket or schedules lifecycle work.
    sparse_fanout: Arc<tokio::sync::Mutex<crate::sparse_fanout::SparseFanoutState>>,
    /// When true, the auto-connect loop runs at reduced frequency and skips
    /// expensive operations (health probing, presence republish, network change
    /// recovery). Event-driven connects still fire immediately.
    app_backgrounded: Arc<AtomicBool>,
    background_execution_allowed: Arc<AtomicBool>,
    #[cfg(target_arch = "wasm32")]
    wasm_accept_bridge_started: Arc<std::sync::atomic::AtomicBool>,
    #[cfg(target_arch = "wasm32")]
    last_emitted_connection_states:
        Arc<Mutex<std::collections::HashMap<String, WasmConnectionStateFingerprint>>>,
    #[cfg(target_arch = "wasm32")]
    last_empty_peer_sessions_warning_ms: Arc<AtomicU64>,
    /// Serializes concurrent calls to `init_iroh_with_router_mode` so that only
    /// one endpoint is ever created. Without this, the background init and a
    /// frontend IPC `start_iroh_node` can race and bind two endpoints with the
    /// same secret key, causing the relay to reject the duplicate.
    iroh_init_guard: Arc<tokio::sync::Mutex<()>>,
    /// Packet carriers installed on the one immutable endpoint before bind.
    /// Session pumps may activate a peer address, but the peer-session actor
    /// remains the only owner allowed to commit a replacement generation.
    #[cfg(any(feature = "transport-webrtc", feature = "transport-moq"))]
    iroh_packet_carriers: Arc<
        RwLock<
            HashMap<
                crate::iroh_carrier_kind::IrohCarrierKind,
                Arc<crate::packet_carrier_transport::PacketCarrierAddressProvider>,
            >,
        >,
    >,
    /// Coalesces concurrent managed dials for the same deterministic connection id.
    ///
    /// Without this, two near-simultaneous `connect_device` callers can both dial
    /// the same endpoint before either observes the other's Pending record. The
    /// native node then replaces one outbound transport with the other, killing the
    /// session-token admission stream before the host can adopt the connection.
    managed_connect_gates: Arc<tokio::sync::Mutex<HashMap<String, Arc<tokio::sync::Mutex<()>>>>>,
    /// Physical Iroh connection stable IDs with an active selected-path watcher.
    /// Every install/adoption path may request a watcher; this registry keeps
    /// ownership singular without assuming another path already spawned one.
    #[cfg(not(target_arch = "wasm32"))]
    iroh_path_watcher_stable_ids: Arc<Mutex<HashSet<u64>>>,
    /// Maps Iroh custom-transport discriminators to OpenRTC path semantics.
    ///
    /// Companion crates register their transport id here. Core owns only path
    /// classification and lifecycle projection; transport-specific sockets,
    /// scanning, and retries remain in the companion.
    #[cfg(not(target_arch = "wasm32"))]
    native_custom_transport_kinds: Arc<RwLock<HashMap<u64, IrohPathKind>>>,
    /// Runtime-installed native transport providers keyed by their OpenRTC path
    /// kind. Providers own hardware discovery; OpenRTC owns peer negotiation,
    /// connection replacement, and lifecycle projection.
    #[cfg(not(target_arch = "wasm32"))]
    native_transport_upgrade_providers:
        Arc<RwLock<HashMap<IrohPathKind, Arc<dyn UpgradeProvider>>>>,
    /// Rust-owned browser carrier attempt gates. JavaScript owns the
    /// `RTCPeerConnection` objects, but it cannot authorize, supersede, retry,
    /// or commit a physical Iroh generation.
    #[cfg(target_arch = "wasm32")]
    wasm_transport_upgrade_gates:
        Arc<tokio::sync::Mutex<HashMap<(String, IrohPathKind), WasmTransportUpgradeGate>>>,
    /// Coalesces capability handshakes and retries into one transport upgrade
    /// attempt per logical peer/path pair.
    #[cfg(not(target_arch = "wasm32"))]
    native_transport_upgrade_gates:
        Arc<tokio::sync::Mutex<HashMap<(String, IrohPathKind), NativeTransportUpgradeGate>>>,
    /// Bounded BLE replacement attempts for the current logical/physical
    /// generation. The Rust transport owner uses this to retry radio or route
    /// failures without creating a second lifecycle loop in TypeScript.
    #[cfg(not(target_arch = "wasm32"))]
    native_ble_upgrade_attempts:
        Arc<tokio::sync::Mutex<HashMap<String, NativeBleUpgradeAttemptState>>>,
    #[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
    native_webrtc_carrier_attempts: Arc<RwLock<HashMap<String, Arc<NativeWebRtcCarrierAttempt>>>>,
    /// Bounded, test-only ownership trail for native Iroh carrier attempts.
    /// Physical-device harnesses cannot rely on WebView or process stdout, so
    /// the Rust owner retains the last decisions for generation-safe failure
    /// evidence without exposing carrier details through the public API.
    #[cfg(all(
        not(target_arch = "wasm32"),
        feature = "iroh-carrier-core",
        any(feature = "test-harness", feature = "testing-endpoints")
    ))]
    native_iroh_carrier_debug_events: Arc<StdRwLock<std::collections::VecDeque<serde_json::Value>>>,
    #[cfg(all(not(target_arch = "wasm32"), feature = "transport-moq"))]
    native_moq_carrier_attempts: Arc<RwLock<HashMap<String, Arc<NativeMoqCarrierAttempt>>>>,
    /// Negotiated optional-transport facts for the current logical peer
    /// session. Path watchers consult this before asking the single upgrade
    /// owner to react to relay demotion.
    #[cfg(not(target_arch = "wasm32"))]
    native_peer_transport_capabilities:
        Arc<RwLock<HashMap<String, HashSet<NativePeerTransportCapability>>>>,
    pub(crate) presence_loop_tx:
        Arc<Mutex<Option<tokio::sync::mpsc::Sender<crate::presence::PresenceCommand>>>>,
    /// Platform project and token source used for authenticated gateway
    /// admission. Logical usage policy is enforced by the control plane, not
    /// cached in the transport runtime.
    #[cfg(not(target_arch = "wasm32"))]
    pub(crate) project_id: String,
    #[cfg(not(target_arch = "wasm32"))]
    pub(crate) token_provider: Arc<dyn Fn() -> Option<String> + Send + Sync>,
    transport_config: Arc<RwLock<TransportConfig>>,
    /// Monotonic fence for carrier selection inputs. Attempts capture this at
    /// reservation and must match it immediately before atomic replacement.
    iroh_carrier_policy_epoch: Arc<AtomicU64>,
    /// Per-connection remote capability revisions. A third peer must never
    /// invalidate an unrelated carrier proof, while a capability change for
    /// the owning peer must still retire its stale reservation.
    iroh_carrier_peer_policy_epochs: Arc<StdRwLock<HashMap<String, u64>>>,
    #[cfg(all(not(target_arch = "wasm32"), feature = "transport-lan"))]
    pub(crate) local_discovery_registry: crate::local_discovery::LocalDiscoveryRegistry,
    #[cfg(all(not(target_arch = "wasm32"), feature = "transport-lan"))]
    mdns_address_lookup: Arc<RwLock<Option<iroh_mdns_address_lookup::MdnsAddressLookup>>>,
    /// Persistent SDK-owned native-main control stream. The admission request
    /// establishes this channel before application streams are exposed; transport
    /// capability, replacement, and signaling frames reuse it for the peer
    /// session lifetime. The map is keyed by deterministic connection id, but
    /// each value is owned by one physical Iroh generation and QUIC stream.
    #[cfg(not(target_arch = "wasm32"))]
    pub(crate) native_control_streams:
        Arc<tokio::sync::Mutex<HashMap<String, NativeControlStreamEntry>>>,
    /// Generation-bound positive liveness evidence produced by successfully
    /// parsed OpenRTC protocol frames. This is an input to the single Rust
    /// lifecycle owner, not a second timer or connection-state authority.
    #[cfg(not(target_arch = "wasm32"))]
    native_transport_protocol_activity:
        Arc<StdRwLock<HashMap<String, NativeTransportProtocolActivity>>>,
    /// Experimental, opt-in scoped actor registry. The default runtime has no
    /// Rust-side actor: the TypeScript scoped actor is the single active
    /// coalescer until a native actor owns real dial and channel routing.
    #[cfg(all(not(target_arch = "wasm32"), feature = "experimental-scoped-actor"))]
    scoped_connection_actor_registry:
        Arc<RwLock<Option<Arc<crate::client::scoped_connection_actor::ConnectionActors>>>>,
    /// Phase 5: shared auth-readiness gate. The TS auth bridge (or any
    /// host that owns auth lifecycle) pushes leg state via `mark_*`;
    /// the drive-grant connection actor and other readiness-aware
    /// callers consult it via `wait_until_ready`. Always present so
    /// `Default`-style construction does not need a feature flag.
    #[cfg(not(target_arch = "wasm32"))]
    auth_readiness: Arc<crate::client::auth_readiness::AuthReadinessStore>,
    /// Scope classifier for mapping admitted scopes → correlation labels.
    /// Default preserves legacy drive-grant/user-device behavior.
    #[cfg(not(target_arch = "wasm32"))]
    scope_classifier: Arc<RwLock<Arc<dyn ScopeClassifier>>>,
}

#[derive(Debug, Clone, PartialEq, Eq)]
struct RemoteSessionAdmissionProof {
    transport_stable_id: u64,
    token_fingerprint: String,
    approval_scope: String,
}

#[derive(Debug, Clone)]
struct PendingInlineReciprocalAdmission {
    connection_id: String,
    remote_node_id: String,
    local_device_id: String,
    remote_device_id: String,
    transport_stable_id: u64,
    stream_instance_id: String,
    presentation_id: String,
    token: String,
    token_fingerprint: String,
    expected_scope: String,
    inbound_admission_fingerprint: String,
    inbound_admission_epoch: u64,
}

fn now_millis_i64() -> i64 {
    crate::coordination::now_millis_u64().min(i64::MAX as u64) as i64
}

/// The current path kind of an iroh connection, used to gate transport upgrades.
///
/// Priority order for native data sending is edge-aware:
///   proven WebRTC/MoQ routes stay first; `DirectQuic` / `DirectLan` / `Ble`
///   remain primary base paths; relay or unknown paths can probe optional
///   upgrades first.
///
/// WebRTC and MoQ upgrades are started when the path is `Relay` and suspended
/// (but not torn down) when it transitions back to `DirectQuic`.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub enum IrohPathKind {
    /// Direct UDP/QUIC hole-punched path — low latency, no intermediary.
    DirectQuic,
    /// Direct QUIC over a private/link-local LAN address.
    DirectLan,
    /// Traffic routed through an iroh relay server — higher latency, rate-limited for web.
    Relay,
    /// Selected path uses a BLE custom transport.
    Ble,
    /// Selected path carries Iroh packets through an unreliable WebRTC DataChannel.
    WebRtc,
    /// Selected path carries Iroh packets through a FIFO MoQ object track.
    Moq,
    /// No live connection or path information is not yet available.
    Unknown,
}

#[cfg(not(target_arch = "wasm32"))]
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub(crate) enum NativePeerTransportCapability {
    WebRtc,
    Moq,
    Ble,
}

/// Hardware-specific preparation for a native Iroh transport upgrade.
///
/// Implementations may scan radios or resolve a platform route, but they do
/// not mutate OpenRTC connection state. The returned address must contain only
/// the prepared transport path so OpenRTC can prove the replacement did not
/// silently fall back to relay or IP.
#[cfg(not(target_arch = "wasm32"))]
#[async_trait::async_trait]
pub trait UpgradeProvider: std::fmt::Debug + Send + Sync {
    fn kind(&self) -> IrohPathKind;
    fn transport_id(&self) -> u64;
    async fn prepare_endpoint_addr(
        &self,
        endpoint_id: iroh::EndpointId,
    ) -> anyhow::Result<iroh::EndpointAddr>;
}

impl IrohPathKind {
    pub fn is_relay_path(self) -> bool {
        matches!(self, Self::Relay)
    }

    pub fn transport_label(self) -> &'static str {
        match self {
            Self::DirectQuic => crate::transport_label::IROH_QUIC,
            Self::DirectLan => crate::transport_label::IROH_LAN,
            Self::Relay => crate::transport_label::IROH_RELAY,
            Self::Ble => crate::transport_label::BLE,
            Self::WebRtc => crate::transport_label::WEBRTC,
            Self::Moq => crate::transport_label::MOQ,
            Self::Unknown => crate::transport_label::IROH,
        }
    }
}

/// Native iroh LAN discovery configuration.
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize, PartialEq, Eq)]
#[serde(rename_all = "camelCase")]
pub struct IrohLanConfig {
    #[serde(default = "default_lan_enabled")]
    pub enabled: bool,
    /// When false, listen for LAN peers without advertising this endpoint.
    #[serde(default = "default_lan_advertise")]
    pub advertise: bool,
}

fn default_lan_enabled() -> bool {
    true
}

fn default_lan_advertise() -> bool {
    true
}

impl Default for IrohLanConfig {
    fn default() -> Self {
        Self {
            enabled: true,
            advertise: true,
        }
    }
}

/// Native BLE discovery and iroh custom-transport configuration.
///
/// BLE is native-only and is intended as a nearby-device path for poor or
/// unavailable internet conditions. Browser runtimes ignore this setting.
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize, PartialEq, Eq, Default)]
#[serde(rename_all = "camelCase")]
pub struct BleConfig {
    #[serde(default)]
    pub enabled: bool,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub connect_timeout_ms: Option<u64>,
}

/// How long a newly-bound transport has to reach settled-ready state before the
/// connection is considered dead and retired.  Must be longer than the
/// duplicate-close grace window (8 s) so that auto-connect suppression covers
/// the entire hold period.
pub(crate) const MANAGED_SETTLE_DEADLINE_MS: i64 =
    crate::runtime_policy::MANAGED_SETTLE_DEADLINE_MS;

/// How long an incoming connection has to present a valid session token before
/// it is closed with session-admission-timeout.  This is intentionally longer
/// than MANAGED_SETTLE_DEADLINE_MS because share-ticket peers must complete
/// full WASM initialization and iroh connection setup before they can send the
/// token stream, which can take 20-40 s on a cold web load.  Matches
/// NATIVE_WEBRTC_CONNECT_TIMEOUT_MS so WebRTC negotiation can complete in the
/// same window.
#[cfg(not(target_arch = "wasm32"))]
pub(crate) const SESSION_ADMISSION_TIMEOUT_MS: u64 =
    crate::runtime_policy::SESSION_ADMISSION_TIMEOUT_MS;

#[derive(Debug, Clone, serde::Serialize, serde::Deserialize, PartialEq, Eq)]
#[serde(rename_all = "camelCase")]
pub enum ConnectionStatus {
    Disconnected,
    Connecting,
    Connected,
    Failed,
    Closed,
    Online,
}

#[derive(Debug, Clone, serde::Serialize, serde::Deserialize, PartialEq, Eq)]
#[serde(rename_all = "camelCase")]
pub enum DevicePresenceStatus {
    Online,
    Idle,
    Offline,
}

#[derive(Debug, Clone, serde::Serialize, serde::Deserialize, PartialEq, Eq)]
#[serde(rename_all = "camelCase")]
pub enum ReadinessState {
    Connecting,
    TransportOnly,
    Settling,
    Routable,
    AwaitingReplacement,
    Closed,
    Failed,
}

#[derive(Debug, Clone, Default, serde::Serialize, serde::Deserialize, PartialEq, Eq)]
#[serde(rename_all = "camelCase")]
pub struct LatencySnapshot {
    pub webrtc: Option<u64>,
    pub iroh: Option<u64>,
    pub iroh_lan: Option<u64>,
    pub iroh_relay: Option<u64>,
    pub ble: Option<u64>,
    pub moq: Option<u64>,
}

impl LatencySnapshot {
    fn set(&mut self, transport: &str, latency_ms: u64) {
        match transport.trim().to_ascii_lowercase().as_str() {
            "webrtc" => self.webrtc = Some(latency_ms),
            "iroh-lan" => self.iroh_lan = Some(latency_ms),
            "iroh-relay" => self.iroh_relay = Some(latency_ms),
            "iroh" | "iroh-quic" => self.iroh = Some(latency_ms),
            "ble" => self.ble = Some(latency_ms),
            "moq" => self.moq = Some(latency_ms),
            _ => {}
        }
    }

    fn get(&self, transport: &str) -> Option<u64> {
        match transport.trim().to_ascii_lowercase().as_str() {
            "webrtc" => self.webrtc,
            "iroh-lan" => self.iroh_lan,
            "iroh-relay" => self.iroh_relay,
            "iroh" | "iroh-quic" => self.iroh,
            "ble" => self.ble,
            "moq" => self.moq,
            _ => None,
        }
    }
}

#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct DeviceStatusSnapshot {
    #[serde(flatten)]
    pub device: crate::signaling::Device,
    pub presence_status: DevicePresenceStatus,
    pub presence_updated_at: Option<i64>,
    pub presence_expires_at: Option<i64>,
    pub connectable: bool,
    pub connection_status: ConnectionStatus,
    pub settled_ready: bool,
    pub readiness_state: ReadinessState,
    pub readiness_reason: String,
    pub peer_health: crate::connection_manager::ConnectionHealth,
    pub peer_id: Option<String>,
    pub scopes: Vec<String>,
    pub connection_id: Option<String>,
    pub device_id_hint: Option<String>,
    pub active_transport_stable_id: Option<u64>,
    pub transport_generation: u64,
    pub route_generation: u64,
    pub active_transport: String,
    pub parallel_transport: Option<String>,
    #[serde(default)]
    pub latency_ms: Option<u64>,
    #[serde(default)]
    pub latency_by_transport: LatencySnapshot,
}

#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct PeerSessionSnapshot {
    pub peer_id: String,
    pub device_id: Option<String>,
    pub device_id_hint: Option<String>,
    pub node_id: Option<String>,
    pub active_connection_id: Option<String>,
    pub candidate_connection_ids: Vec<String>,
    pub status: crate::connection_manager::ConnectionState,
    pub health: crate::connection_manager::ConnectionHealth,
    pub settled_ready: bool,
    #[serde(default)]
    pub logical_session_terminal: bool,
    pub readiness_state: ReadinessState,
    pub active_transport_stable_id: Option<u64>,
    pub transport_generation: u64,
    pub route_generation: u64,
    pub active_transport: String,
    pub parallel_transport: Option<String>,
    pub replacement_pending: bool,
    pub last_lifecycle_transition_at_ms: i64,
    pub readiness_reason: String,
    pub transition_count: u64,
    pub connecting_transition_count: u64,
    pub replacement_count: u64,
    pub retire_count: u64,
    pub last_disconnect_reason: Option<String>,
    pub last_reconnect_reason: Option<String>,
    pub scopes: Vec<String>,
    pub last_seen_at_ms: i64,
    pub error: Option<String>,
}

#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct ManagedConnectResult {
    pub connection_id: String,
    pub device_id: Option<String>,
    pub device_id_hint: Option<String>,
    pub remote_node_id: String,
    pub state: String,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub approved_scope: Option<String>,
}

#[cfg(not(target_arch = "wasm32"))]
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct ConnectionAdoption {
    pub connection_id: String,
    pub node_id: String,
    pub device_id: Option<String>,
    pub transport_generation: u64,
    pub status_reason: Option<String>,
    pub main_stream_ready: bool,
}

#[derive(Debug, Clone, serde::Serialize, serde::Deserialize, PartialEq, Eq)]
#[serde(rename_all = "camelCase")]
pub enum ManagedHealth {
    Healthy,
    AwaitingReplacement,
    Dead,
}

#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct ConnectionHealthView {
    pub connection_id: String,
    pub device_id: Option<String>,
    pub device_id_hint: Option<String>,
    pub node_id: Option<String>,
    pub active_transport_stable_id: Option<u64>,
    pub transport_generation: u64,
    pub route_generation: u64,
    pub status: ManagedHealth,
    pub settled_ready: bool,
    pub readiness_state: ReadinessState,
    pub replacement_pending: bool,
    pub last_lifecycle_transition_at_ms: i64,
    pub readiness_reason: String,
    pub transition_count: u64,
    pub connecting_transition_count: u64,
    pub replacement_count: u64,
    pub retire_count: u64,
    pub last_disconnect_reason: Option<String>,
    pub last_reconnect_reason: Option<String>,
}

#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct StateSnapshot {
    pub connection_id: String,
    pub device_id: Option<String>,
    pub device_id_hint: Option<String>,
    pub remote_node_id: Option<String>,
    pub state: String,
    pub transport_state: String,
    pub protocol_state: String,
    pub routable: bool,
    #[serde(default)]
    pub logical_session_terminal: bool,
    pub readiness_state: ReadinessState,
    pub readiness_reason: String,
    pub transport_generation: u64,
    pub route_generation: u64,
    pub active_transport_stable_id: Option<u64>,
    pub active_transport: String,
    pub parallel_transport: Option<String>,
    pub replacement_in_progress: bool,
    pub last_lifecycle_transition_at_ms: i64,
    pub transition_count: u64,
    pub connecting_transition_count: u64,
    pub replacement_count: u64,
    pub retire_count: u64,
    pub last_disconnect_reason: Option<String>,
    pub last_reconnect_reason: Option<String>,
    /// Authenticated application scopes owned by this logical peer.
    ///
    /// Native projection uses these exact Rust-owned scopes to route a
    /// connection into its registered capability without inventing a second
    /// roster or relying on gateway desired-peer identity.
    #[serde(default)]
    pub scopes: Vec<String>,
    pub error: Option<String>,
    pub created_at: i64,
    pub updated_at: i64,
}

#[cfg(target_arch = "wasm32")]
#[derive(Debug, Clone, PartialEq, Eq)]
struct WasmConnectionStateFingerprint {
    connection_id: String,
    device_id: Option<String>,
    device_id_hint: Option<String>,
    remote_node_id: Option<String>,
    state: String,
    transport_state: String,
    protocol_state: String,
    routable: bool,
    logical_session_terminal: bool,
    transport_generation: u64,
    route_generation: u64,
    active_transport_stable_id: Option<u64>,
    active_transport: String,
    parallel_transport: Option<String>,
    replacement_in_progress: bool,
    transition_count: u64,
    connecting_transition_count: u64,
    replacement_count: u64,
    retire_count: u64,
    last_disconnect_reason: Option<String>,
    last_reconnect_reason: Option<String>,
    scopes: Vec<String>,
    error: Option<String>,
}

#[cfg(target_arch = "wasm32")]
impl From<&StateSnapshot> for WasmConnectionStateFingerprint {
    fn from(snapshot: &StateSnapshot) -> Self {
        Self {
            connection_id: snapshot.connection_id.clone(),
            device_id: snapshot.device_id.clone(),
            device_id_hint: snapshot.device_id_hint.clone(),
            remote_node_id: snapshot.remote_node_id.clone(),
            state: snapshot.state.clone(),
            transport_state: snapshot.transport_state.clone(),
            protocol_state: snapshot.protocol_state.clone(),
            routable: snapshot.routable,
            logical_session_terminal: snapshot.logical_session_terminal,
            transport_generation: snapshot.transport_generation,
            route_generation: snapshot.route_generation,
            active_transport_stable_id: snapshot.active_transport_stable_id,
            active_transport: snapshot.active_transport.clone(),
            parallel_transport: snapshot.parallel_transport.clone(),
            replacement_in_progress: snapshot.replacement_in_progress,
            transition_count: snapshot.transition_count,
            connecting_transition_count: snapshot.connecting_transition_count,
            replacement_count: snapshot.replacement_count,
            retire_count: snapshot.retire_count,
            last_disconnect_reason: snapshot.last_disconnect_reason.clone(),
            last_reconnect_reason: snapshot.last_reconnect_reason.clone(),
            scopes: snapshot.scopes.clone(),
            error: snapshot.error.clone(),
        }
    }
}

#[derive(Debug, Clone, serde::Serialize, serde::Deserialize, PartialEq, Eq)]
#[serde(rename_all = "camelCase", tag = "kind")]
pub enum BridgeAction {
    Healthy,
    AwaitReplacement {
        reason: String,
    },
    Rebind {
        remote_node_id: Option<String>,
        transport_generation: u64,
    },
    Retire {
        reason: String,
    },
}

#[cfg(not(target_arch = "wasm32"))]
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum DuplicateClosedHandling {
    RebindKeptTransport,
    RetireClosedTransport,
}

#[cfg(not(target_arch = "wasm32"))]
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum IncomingTransportCloseResolution {
    PreservedKeptTransport,
    RetiredClosedTransport,
}

#[cfg(not(target_arch = "wasm32"))]
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum AutoConnectTieBreakDecision {
    ObserveConnectedTransport,
    WaitForInitiator,
    ActAsInitiator,
}

#[cfg(not(target_arch = "wasm32"))]
fn duplicate_closed_handling(
    close_reason_debug: &str,
    kept_transport_alive: bool,
    kept_transport_healthy: bool,
    kept_stable_id: u64,
    closed_stable_id: u64,
) -> DuplicateClosedHandling {
    let _ = close_reason_debug;
    let _ = kept_transport_healthy;
    if kept_stable_id != closed_stable_id && kept_transport_alive {
        DuplicateClosedHandling::RebindKeptTransport
    } else {
        DuplicateClosedHandling::RetireClosedTransport
    }
}

#[cfg(not(target_arch = "wasm32"))]
fn auto_connect_tie_break_decision(
    local_node_id: &str,
    remote_node_id: &str,
    transport_connected: bool,
    waited_ms: i64,
    initiator_grace_ms: i64,
) -> AutoConnectTieBreakDecision {
    if transport_connected {
        return AutoConnectTieBreakDecision::ObserveConnectedTransport;
    }
    if local_node_id <= remote_node_id {
        // Non-initiator waits for the grace period, then acts as initiator
        // to avoid getting stuck when the true initiator is unavailable.
        if waited_ms >= initiator_grace_ms {
            AutoConnectTieBreakDecision::ActAsInitiator
        } else {
            AutoConnectTieBreakDecision::WaitForInitiator
        }
    } else {
        AutoConnectTieBreakDecision::ActAsInitiator
    }
}

fn normalize_lookup_id(value: Option<&str>) -> Option<String> {
    let trimmed = value?.trim();
    if trimmed.is_empty() {
        None
    } else {
        Some(trimmed.to_ascii_lowercase())
    }
}

fn peer_snapshot_lookup_aliases(peer: &crate::connection_manager::PeerSnapshot) -> Vec<String> {
    [
        normalize_lookup_id(peer.device_id.as_deref()),
        normalize_lookup_id(peer.device_id_hint.as_deref()),
        normalize_lookup_id(peer.node_id.as_deref()),
        normalize_lookup_id(Some(peer.peer_id.as_str())),
    ]
    .into_iter()
    .flatten()
    .collect()
}

fn peer_snapshot_matches_any_alias(
    peer: &crate::connection_manager::PeerSnapshot,
    aliases: &std::collections::HashSet<String>,
) -> bool {
    peer_snapshot_lookup_aliases(peer)
        .into_iter()
        .any(|alias| aliases.contains(&alias))
}

fn preferred_peer_snapshot(
    current: Option<crate::connection_manager::PeerSnapshot>,
    candidate: crate::connection_manager::PeerSnapshot,
) -> crate::connection_manager::PeerSnapshot {
    match current {
        Some(existing) if !prefer_device_status_peer_snapshot(&candidate, &existing) => existing,
        _ => candidate,
    }
}

fn snapshot_connection_status(
    device: &crate::signaling::Device,
    peer: Option<&crate::connection_manager::PeerSnapshot>,
) -> ConnectionStatus {
    match peer.map(|value| &value.status) {
        Some(crate::connection_manager::ConnectionState::Pending)
        | Some(crate::connection_manager::ConnectionState::Connecting) => {
            ConnectionStatus::Connecting
        }
        Some(crate::connection_manager::ConnectionState::Connected)
            if peer.map(peer_snapshot_settled_ready).unwrap_or(false) =>
        {
            ConnectionStatus::Connected
        }
        Some(crate::connection_manager::ConnectionState::Connected) => ConnectionStatus::Connecting,
        Some(crate::connection_manager::ConnectionState::Failed) => ConnectionStatus::Failed,
        Some(crate::connection_manager::ConnectionState::Closed)
        | Some(crate::connection_manager::ConnectionState::Closing) => ConnectionStatus::Closed,
        None if device.online => ConnectionStatus::Online,
        None => ConnectionStatus::Disconnected,
    }
}

fn snapshot_settled_ready(
    peer: Option<&crate::connection_manager::PeerSnapshot>,
    connection_status: &ConnectionStatus,
) -> bool {
    matches!(connection_status, ConnectionStatus::Connected)
        && peer.map(peer_snapshot_settled_ready).unwrap_or(false)
}

fn device_presence_updated_at(device: &crate::signaling::Device) -> Option<i64> {
    [
        device.last_seen_at.as_ref(),
        device.updated_at.as_ref(),
        device.created_at.as_ref(),
    ]
    .into_iter()
    .flatten()
    .filter_map(|value| crate::presence_policy::parse_millis(Some(value)))
    .max()
}

fn device_presence_expires_at(device: &crate::signaling::Device) -> Option<i64> {
    crate::presence_policy::parse_millis(device.expires_at.as_ref())
}

fn device_presence_status(
    device: &crate::signaling::Device,
    presence_updated_at: Option<i64>,
    now_ms: i64,
) -> DevicePresenceStatus {
    if !device.online {
        return DevicePresenceStatus::Offline;
    }

    if presence_updated_at
        .map(|updated_at| {
            updated_at.saturating_add(crate::presence_policy::DEVICE_STALE_HEARTBEAT_MS) <= now_ms
        })
        .unwrap_or(false)
    {
        DevicePresenceStatus::Idle
    } else {
        DevicePresenceStatus::Online
    }
}

fn device_connectable(
    device: &crate::signaling::Device,
    presence_status: &DevicePresenceStatus,
) -> bool {
    matches!(presence_status, DevicePresenceStatus::Online)
        && device
            .ticket
            .as_deref()
            .map(str::trim)
            .is_some_and(|value| !value.is_empty())
}

fn peer_snapshot_settled_ready(peer: &crate::connection_manager::PeerSnapshot) -> bool {
    matches!(peer_readiness_state(peer), ReadinessState::Routable)
}

fn peer_has_live_transport(peer: &crate::connection_manager::PeerSnapshot) -> bool {
    !peer.connection_ids.is_empty() && peer.active_transport_stable_id.is_some()
}

fn peer_is_replacement_handoff(peer: &crate::connection_manager::PeerSnapshot) -> bool {
    peer.active_transport_stable_id.is_none()
        && peer.error.as_deref() == Some(crate::lifecycle_reason::REASON_REPLACEMENT_IN_PROGRESS)
}

fn peer_readiness_state(peer: &crate::connection_manager::PeerSnapshot) -> ReadinessState {
    match peer.status {
        crate::connection_manager::ConnectionState::Pending
        | crate::connection_manager::ConnectionState::Connecting => ReadinessState::Connecting,
        crate::connection_manager::ConnectionState::Connected => {
            // A transport object or socket-ready carrier is not readiness
            // proof. Only a current-generation pong or generation-bound
            // application-route observation may set Healthy.
            let connection_ids_present = !peer.connection_ids.is_empty();
            let healthy_with_connections = connection_ids_present
                && matches!(
                    peer.health,
                    crate::connection_manager::ConnectionHealth::Healthy
                );
            if healthy_with_connections {
                ReadinessState::Routable
            } else if peer_is_replacement_handoff(peer) {
                ReadinessState::AwaitingReplacement
            } else if !peer_has_live_transport(peer) {
                ReadinessState::TransportOnly
            } else if now_millis_i64().saturating_sub(peer.last_lifecycle_transition_at_ms)
                >= MANAGED_SETTLE_DEADLINE_MS
            {
                ReadinessState::AwaitingReplacement
            } else {
                ReadinessState::Settling
            }
        }
        crate::connection_manager::ConnectionState::Closing
        | crate::connection_manager::ConnectionState::Closed => ReadinessState::Closed,
        crate::connection_manager::ConnectionState::Failed => ReadinessState::Failed,
    }
}

fn peer_readiness_reason(peer: &crate::connection_manager::PeerSnapshot) -> String {
    match peer_readiness_state(peer) {
        ReadinessState::Connecting => "waiting-for-transport".to_string(),
        ReadinessState::TransportOnly => {
            "transport-attached-awaiting-readiness-confirmation".to_string()
        }
        ReadinessState::Settling => "transport-connected-health-check-pending".to_string(),
        ReadinessState::Routable => "transport-healthy".to_string(),
        ReadinessState::AwaitingReplacement => {
            if peer.active_transport_stable_id.is_none() {
                "missing-active-transport".to_string()
            } else {
                "readiness-confirmation-timed-out".to_string()
            }
        }
        ReadinessState::Closed => "transport-closed".to_string(),
        ReadinessState::Failed => "connection-failed".to_string(),
    }
}

fn connection_readiness_state(
    record: &crate::connection_manager::ConnectionRecord,
    peer_snapshot: Option<&crate::connection_manager::PeerSnapshot>,
) -> ReadinessState {
    match record.state {
        crate::connection_manager::ConnectionState::Pending
        | crate::connection_manager::ConnectionState::Connecting => ReadinessState::Connecting,
        crate::connection_manager::ConnectionState::Connected => {
            if peer_snapshot
                .map(peer_snapshot_settled_ready)
                .unwrap_or(false)
            {
                ReadinessState::Routable
            } else if record.transport_stable_id.is_none() {
                ReadinessState::AwaitingReplacement
            } else if now_millis_i64().saturating_sub(record.last_transport_change_at_ms)
                >= MANAGED_SETTLE_DEADLINE_MS
            {
                ReadinessState::AwaitingReplacement
            } else {
                ReadinessState::Settling
            }
        }
        crate::connection_manager::ConnectionState::Closing
        | crate::connection_manager::ConnectionState::Closed => ReadinessState::Closed,
        crate::connection_manager::ConnectionState::Failed => ReadinessState::Failed,
    }
}

fn connection_readiness_reason(
    record: &crate::connection_manager::ConnectionRecord,
    peer_snapshot: Option<&crate::connection_manager::PeerSnapshot>,
) -> String {
    match connection_readiness_state(record, peer_snapshot) {
        ReadinessState::Connecting => "waiting-for-transport".to_string(),
        ReadinessState::TransportOnly => {
            "transport-attached-awaiting-readiness-confirmation".to_string()
        }
        ReadinessState::Settling => "bounded-health-confirmation-pending".to_string(),
        ReadinessState::Routable => "transport-healthy".to_string(),
        ReadinessState::AwaitingReplacement => {
            if record.transport_stable_id.is_none() {
                "missing-active-transport".to_string()
            } else {
                "readiness-confirmation-timed-out".to_string()
            }
        }
        ReadinessState::Closed => "transport-closed".to_string(),
        ReadinessState::Failed => "connection-failed".to_string(),
    }
}

fn public_peer_state(
    peer: &crate::connection_manager::PeerSnapshot,
) -> crate::connection_manager::ConnectionState {
    if matches!(
        peer.status,
        crate::connection_manager::ConnectionState::Connected
    ) && !peer_snapshot_settled_ready(peer)
    {
        crate::connection_manager::ConnectionState::Connecting
    } else {
        peer.status.clone()
    }
}

fn peer_session_snapshot_from_peer(
    peer: crate::connection_manager::PeerSnapshot,
) -> PeerSessionSnapshot {
    let status = public_peer_state(&peer);
    let active_connection_id = peer.connection_ids.first().cloned();
    let candidate_connection_ids = if peer.connection_ids.len() > 1 {
        peer.connection_ids[1..].to_vec()
    } else {
        Vec::new()
    };
    let settled_ready = peer_snapshot_settled_ready(&peer);
    let readiness_state = peer_readiness_state(&peer);
    let readiness_reason = peer_readiness_reason(&peer);
    let crate::connection_manager::PeerSnapshot {
        peer_id,
        device_id,
        device_id_hint,
        node_id,
        active_transport_stable_id,
        active_transport_generation,
        active_route_generation,
        active_transport,
        parallel_transport,
        last_lifecycle_transition_at_ms,
        health,
        transition_count,
        connecting_transition_count,
        replacement_count,
        retire_count,
        last_disconnect_reason,
        last_reconnect_reason,
        logical_session_terminal,
        scopes,
        last_seen_at_ms,
        error,
        ..
    } = peer;

    PeerSessionSnapshot {
        peer_id,
        device_id,
        device_id_hint,
        node_id,
        active_connection_id,
        candidate_connection_ids,
        status,
        health,
        settled_ready,
        logical_session_terminal,
        readiness_state: readiness_state.clone(),
        active_transport_stable_id,
        transport_generation: active_transport_generation,
        route_generation: active_route_generation,
        active_transport,
        parallel_transport,
        replacement_pending: matches!(readiness_state, ReadinessState::AwaitingReplacement),
        last_lifecycle_transition_at_ms,
        readiness_reason,
        transition_count,
        connecting_transition_count,
        replacement_count,
        retire_count,
        last_disconnect_reason,
        last_reconnect_reason,
        scopes,
        last_seen_at_ms,
        error,
    }
}

fn connection_state_snapshot_from_parts(
    record: &crate::connection_manager::ConnectionRecord,
    peer_snapshot: Option<&crate::connection_manager::PeerSnapshot>,
) -> StateSnapshot {
    let settled_ready = peer_snapshot
        .map(peer_snapshot_settled_ready)
        .unwrap_or(false);
    let readiness_state = connection_readiness_state(record, peer_snapshot);
    // An in-place transport reconnect (the base transport closes with a
    // replacement/transient reason, then re-binds on the SAME connection — e.g.
    // physical reconnect, bumping transport_generation) is not a
    // disconnect. The record momentarily transitions Closing/Closed before the
    // successor transport lands; surfacing that intermediate frame as "closed" makes
    // every consumer flicker connected -> disconnected -> connected on each
    // generation bump. When the close reason is an in-place reconnect, report the
    // reconnecting family ("connecting") instead.
    //
    // Scoped to ReplacementInProgress + is_transient (same-connection
    // reconnects). It deliberately excludes is_replacement_churn (replaced-by-new-*),
    // which marks a DIFFERENT connection that was genuinely superseded — that record
    // is correctly closed, and the peer is live on the successor connection. A
    // genuine close (manual / graceful / failed / revoked) keeps a non-reconnect
    // reason and still surfaces as closed; each transition overwrites status_reason,
    // so a later terminal close clears this.
    let in_place_reconnect_close = matches!(
        record.state,
        crate::connection_manager::ConnectionState::Closing
            | crate::connection_manager::ConnectionState::Closed
    ) && crate::lifecycle_reason::LifecycleReasonCode::from_text(
        record.status_reason.as_deref(),
    )
    .map(|code| {
        matches!(
            code,
            crate::lifecycle_reason::LifecycleReasonCode::ReplacementInProgress
        ) || code.is_transient()
    })
    .unwrap_or(false);
    let state = match record.state {
        crate::connection_manager::ConnectionState::Pending => "connecting",
        crate::connection_manager::ConnectionState::Connecting => "connecting",
        crate::connection_manager::ConnectionState::Connected if settled_ready => "connected",
        crate::connection_manager::ConnectionState::Connected => "connecting",
        crate::connection_manager::ConnectionState::Failed => "failed",
        crate::connection_manager::ConnectionState::Closing
        | crate::connection_manager::ConnectionState::Closed
            if in_place_reconnect_close =>
        {
            "connecting"
        }
        crate::connection_manager::ConnectionState::Closing
        | crate::connection_manager::ConnectionState::Closed => "closed",
    };
    let transport_state = match record.state {
        crate::connection_manager::ConnectionState::Pending
        | crate::connection_manager::ConnectionState::Connecting => "connecting",
        crate::connection_manager::ConnectionState::Connected => "connected",
        crate::connection_manager::ConnectionState::Closing
        | crate::connection_manager::ConnectionState::Closed
            if in_place_reconnect_close =>
        {
            "connecting"
        }
        crate::connection_manager::ConnectionState::Failed
        | crate::connection_manager::ConnectionState::Closing
        | crate::connection_manager::ConnectionState::Closed => "closed",
    };
    let protocol_state = match record.state {
        crate::connection_manager::ConnectionState::Connected if settled_ready => "routable",
        crate::connection_manager::ConnectionState::Connected => "transport-only",
        crate::connection_manager::ConnectionState::Pending
        | crate::connection_manager::ConnectionState::Connecting => "connecting",
        crate::connection_manager::ConnectionState::Closing
        | crate::connection_manager::ConnectionState::Closed
            if in_place_reconnect_close =>
        {
            "connecting"
        }
        crate::connection_manager::ConnectionState::Failed
        | crate::connection_manager::ConnectionState::Closing
        | crate::connection_manager::ConnectionState::Closed => "closed",
    };

    StateSnapshot {
        connection_id: record.connection_id.clone(),
        device_id: record
            .device_id
            .clone()
            .or_else(|| peer_snapshot.and_then(|snapshot| snapshot.device_id.clone())),
        device_id_hint: record
            .device_id_hint
            .clone()
            .or_else(|| peer_snapshot.and_then(|snapshot| snapshot.device_id_hint.clone())),
        remote_node_id: record.node_id.clone(),
        state: state.to_string(),
        transport_state: transport_state.to_string(),
        protocol_state: protocol_state.to_string(),
        routable: settled_ready,
        // A different physical record may already own the logical peer. A late
        // close of its predecessor is not a terminal verdict for that successor.
        logical_session_terminal: record.logical_session_terminal()
            && peer_snapshot
                .map(|peer| {
                    peer.logical_session_terminal
                        && peer.connection_ids.first() == Some(&record.connection_id)
                        && peer.active_transport_generation == record.transport_generation
                        && peer.active_route_generation == record.route_generation
                        && peer.active_transport_stable_id == record.transport_stable_id
                })
                .unwrap_or(false),
        readiness_state: readiness_state.clone(),
        readiness_reason: connection_readiness_reason(record, peer_snapshot),
        transport_generation: record.transport_generation,
        route_generation: record.route_generation,
        active_transport_stable_id: record.transport_stable_id,
        active_transport: record.active_transport.clone(),
        parallel_transport: record.parallel_transport.clone(),
        replacement_in_progress: matches!(readiness_state, ReadinessState::AwaitingReplacement),
        last_lifecycle_transition_at_ms: record
            .last_state_change_at_ms
            .max(record.last_transport_change_at_ms)
            .max(record.last_route_change_at_ms),
        transition_count: record.transition_count,
        connecting_transition_count: record.connecting_transition_count,
        replacement_count: record.replacement_count,
        retire_count: record.retire_count,
        last_disconnect_reason: record.last_disconnect_reason.clone(),
        last_reconnect_reason: record.last_reconnect_reason.clone(),
        scopes: peer_snapshot
            .map(|snapshot| snapshot.scopes.clone())
            .unwrap_or_default(),
        error: record.status_reason.clone(),
        created_at: record.created_at_ms,
        updated_at: record.updated_at_ms,
    }
}

fn merge_device_status_snapshots(
    devices: Vec<crate::signaling::Device>,
    peers: Vec<crate::connection_manager::PeerSnapshot>,
) -> Vec<DeviceStatusSnapshot> {
    let mut peer_by_device: std::collections::HashMap<
        String,
        crate::connection_manager::PeerSnapshot,
    > = std::collections::HashMap::new();

    for peer in peers {
        for alias in peer_snapshot_lookup_aliases(&peer) {
            let replace = match peer_by_device.get(&alias) {
                Some(existing) => prefer_device_status_peer_snapshot(&peer, existing),
                None => true,
            };
            if replace {
                peer_by_device.insert(alias, peer.clone());
            }
        }
    }

    devices
        .into_iter()
        .map(|device| {
            let authoritative_node_id = normalize_lookup_id(device.node_id.as_deref());
            let lookup_aliases: Vec<String> = [
                normalize_lookup_id(Some(device.device_id.as_str())),
                authoritative_node_id.clone(),
            ]
            .into_iter()
            .flatten()
            .collect();
            let peer = lookup_aliases
                .iter()
                .filter_map(|key| peer_by_device.get(key).cloned())
                .filter(|candidate| {
                    let candidate_node_id = normalize_lookup_id(candidate.node_id.as_deref());
                    match (&authoritative_node_id, candidate_node_id) {
                        (Some(expected), Some(actual)) => expected == &actual,
                        _ => true,
                    }
                })
                .fold(None, |current, candidate| {
                    Some(preferred_peer_snapshot(current, candidate))
                });
            let connection_status = snapshot_connection_status(&device, peer.as_ref());
            let presence_updated_at = device_presence_updated_at(&device);
            let presence_expires_at = device_presence_expires_at(&device);
            let presence_status =
                device_presence_status(&device, presence_updated_at, now_millis_i64());
            let connectable = device_connectable(&device, &presence_status);
            let readiness_state = peer.as_ref().map(peer_readiness_state).unwrap_or_else(|| {
                if device.online {
                    ReadinessState::Connecting
                } else {
                    ReadinessState::Closed
                }
            });
            DeviceStatusSnapshot {
                presence_status,
                presence_updated_at,
                presence_expires_at,
                connectable,
                settled_ready: snapshot_settled_ready(peer.as_ref(), &connection_status),
                readiness_state: readiness_state.clone(),
                readiness_reason: peer.as_ref().map(peer_readiness_reason).unwrap_or_else(|| {
                    if device.online {
                        "device-online-awaiting-runtime-session".to_string()
                    } else {
                        "device-offline".to_string()
                    }
                }),
                connection_status,
                peer_health: peer
                    .as_ref()
                    .map(|value| value.health.clone())
                    .unwrap_or_else(|| {
                        if device.online {
                            crate::connection_manager::ConnectionHealth::Healthy
                        } else {
                            crate::connection_manager::ConnectionHealth::Unknown
                        }
                    }),
                peer_id: peer.as_ref().map(|value| value.peer_id.clone()),
                scopes: peer
                    .as_ref()
                    .map(|value| value.scopes.clone())
                    .unwrap_or_default(),
                connection_id: peer
                    .as_ref()
                    .and_then(|value| value.connection_ids.first().cloned()),
                device_id_hint: peer.as_ref().and_then(|value| value.device_id_hint.clone()),
                active_transport_stable_id: peer
                    .as_ref()
                    .and_then(|value| value.active_transport_stable_id),
                transport_generation: peer
                    .as_ref()
                    .map(|value| value.active_transport_generation)
                    .unwrap_or(0),
                route_generation: peer
                    .as_ref()
                    .map(|value| value.active_route_generation)
                    .unwrap_or(0),
                active_transport: peer
                    .as_ref()
                    .map(|value| value.active_transport.clone())
                    .unwrap_or_else(|| "iroh".to_string()),
                parallel_transport: peer
                    .as_ref()
                    .and_then(|value| value.parallel_transport.clone()),
                latency_ms: None,
                latency_by_transport: LatencySnapshot::default(),
                device,
            }
        })
        .collect()
}

fn peer_snapshot_recency_ms(peer: &crate::connection_manager::PeerSnapshot) -> i64 {
    peer.last_seen_at_ms
        .max(peer.last_lifecycle_transition_at_ms)
}

/// Picks the winning [`PeerSnapshot`] when the same device/node alias maps to more than one
/// aggregate (different `peer_id` keys / backfill + dial rows). `device_status_peer_priority`
/// ranks *readiness* (Routable) ahead of *status* (Failed), which lets a stale
/// "connected + routable" snapshot replace a newer failed dial; mirror `strongest_state` by
/// comparing `Connected` vs `Failed` with recency first.
fn prefer_device_status_peer_snapshot(
    candidate: &crate::connection_manager::PeerSnapshot,
    existing: &crate::connection_manager::PeerSnapshot,
) -> bool {
    use crate::connection_manager::ConnectionState;
    let c = &candidate.status;
    let e = &existing.status;
    let c_fail = matches!(c, ConnectionState::Failed);
    let c_conn = matches!(c, ConnectionState::Connected);
    let e_fail = matches!(e, ConnectionState::Failed);
    let e_conn = matches!(e, ConnectionState::Connected);

    // Data-plane truth: a Connected snapshot with a live physical transport
    // reflects an actual working data path and beats a Failed snapshot
    // regardless of recency. Recency only matters when
    // neither side has a live transport, in which case we fall back to the
    // previous tie-break.
    let c_live = c_conn && peer_has_live_transport(candidate);
    let e_live = e_conn && peer_has_live_transport(existing);
    if c_live && !e_live {
        return true;
    }
    if !c_live && e_live {
        return false;
    }

    if c_fail && e_conn {
        return peer_snapshot_recency_ms(candidate) >= peer_snapshot_recency_ms(existing);
    }
    if c_conn && e_fail {
        return peer_snapshot_recency_ms(candidate) > peer_snapshot_recency_ms(existing);
    }
    device_status_peer_priority(candidate) > device_status_peer_priority(existing)
}

fn device_status_peer_priority(
    peer: &crate::connection_manager::PeerSnapshot,
) -> (u8, u8, u8, u8, i64, u64, u64) {
    let readiness_rank = match peer_readiness_state(peer) {
        ReadinessState::Routable => 5,
        ReadinessState::Settling => 4,
        ReadinessState::TransportOnly => 3,
        ReadinessState::Connecting => 2,
        ReadinessState::AwaitingReplacement => 1,
        ReadinessState::Closed | ReadinessState::Failed => 0,
    };

    let status_rank = match peer.status {
        crate::connection_manager::ConnectionState::Connected => 3,
        crate::connection_manager::ConnectionState::Connecting
        | crate::connection_manager::ConnectionState::Pending => 2,
        crate::connection_manager::ConnectionState::Closing
        | crate::connection_manager::ConnectionState::Closed => 1,
        crate::connection_manager::ConnectionState::Failed => 0,
    };

    let health_rank = match peer.health {
        crate::connection_manager::ConnectionHealth::Healthy => 3,
        crate::connection_manager::ConnectionHealth::Suspect => 2,
        crate::connection_manager::ConnectionHealth::Unknown => 1,
        crate::connection_manager::ConnectionHealth::Stale => 0,
    };

    (
        readiness_rank,
        status_rank,
        health_rank,
        u8::from(peer_has_live_transport(peer)),
        peer.last_seen_at_ms,
        peer.transition_count,
        peer.replacement_count,
    )
}

#[cfg(not(target_arch = "wasm32"))]
fn auto_connect_verbose() -> bool {
    std::env::var("OPENRTC_AUTOCONNECT_VERBOSE")
        .map(|value| {
            let normalized = value.trim().to_ascii_lowercase();
            normalized == "1" || normalized == "true" || normalized == "yes"
        })
        .unwrap_or(false)
}

#[cfg(not(target_arch = "wasm32"))]
fn parse_env_bool(value: &str) -> Option<bool> {
    match value.trim().to_ascii_lowercase().as_str() {
        "1" | "true" | "yes" | "on" => Some(true),
        "0" | "false" | "no" | "off" => Some(false),
        _ => None,
    }
}

#[cfg(not(target_arch = "wasm32"))]
fn env_bool(name: &str) -> Option<bool> {
    std::env::var(name)
        .ok()
        .and_then(|value| parse_env_bool(&value))
}

#[cfg(not(target_arch = "wasm32"))]
fn should_use_relay_only_mode() -> bool {
    env_bool("PLUTO_IROH_RELAY_ONLY").unwrap_or(false)
}

#[cfg(not(target_arch = "wasm32"))]
fn should_disable_ipv6() -> bool {
    resolve_disable_ipv6(
        env_bool("PLUTO_IROH_DISABLE_IPV6"),
        env_bool("PLUTO_IROH_ENABLE_IPV6"),
    )
}

#[cfg(not(target_arch = "wasm32"))]
fn resolve_disable_ipv6(disable: Option<bool>, enable: Option<bool>) -> bool {
    disable
        .or_else(|| enable.map(|value| !value))
        .unwrap_or(false)
}

#[cfg(not(target_arch = "wasm32"))]
fn apply_native_network_preferences(
    mut builder: iroh::endpoint::Builder,
    relay_only: bool,
    relay_transport_policy: RelayPolicy,
) -> anyhow::Result<iroh::endpoint::Builder> {
    #[cfg(openrtc_iroh_relay_transport_policy_api)]
    {
        let policy = match relay_transport_policy {
            RelayPolicy::Auto => iroh::RelayTransportPolicy::Auto,
            RelayPolicy::QuicRequired => iroh::RelayTransportPolicy::QuicRequired,
            RelayPolicy::WebsocketRequired => iroh::RelayTransportPolicy::WebsocketRequired,
        };
        builder = builder.relay_transport_policy(policy);
    }
    #[cfg(not(openrtc_iroh_relay_transport_policy_api))]
    {
        match relay_transport_policy {
            RelayPolicy::QuicRequired => {
                anyhow::bail!(
                    "requested relay carriage quicRequired is unsupported by this upstream-Iroh build; refusing to initialize"
                );
            }
            RelayPolicy::Auto => {
                eprintln!(
                    "[pluto-rtc][native] requested relay carriage auto; effective carriage is websocket fallback because this upstream-Iroh build has no QUIC relay policy API"
                );
            }
            RelayPolicy::WebsocketRequired => {}
        }
    }

    // Increase QUIC idle timeout to survive iOS background suspension.
    // The default is 6.5s (5s heartbeat + 1.5s grace), which is too aggressive:
    // when iOS suspends the process, PING frames stop and the connection is torn
    // down before the OS resumes us. 120s allows the app to be backgrounded,
    // complete its background task window (~30s), and still reconnect gracefully
    // without the connection being declared dead mid-transfer.
    // Keep-alive interval defaults to 25s on desktop. On mobile targets use a
    // slower cadence to reduce background wake pressure.
    //
    // Flow-control window tuning for relay throughput:
    // When using the iroh relay (WebSocket-based for WASM↔native), the round-trip
    // time is ~100–300 ms. Quinn's defaults (stream_receive_window=256KB,
    // receive_window=1MB, send_window=2MB) cap throughput at roughly 1–10 MB/s on
    // these high-latency paths because the sender stalls waiting for WINDOW_UPDATE
    // frames. Setting windows large enough to cover the bandwidth-delay product
    // (target 40 MB/s × 300 ms RTT = 12 MB) avoids flow-control stalls entirely.
    builder = builder.transport_config(crate::native_node::iroh_config());

    if relay_only {
        // Relay-only mode avoids NAT traversal + local candidate churn on constrained networks.
        return Ok(builder.clear_ip_transports());
    }

    if should_disable_ipv6() {
        // IPv4-only is an explicit compatibility override. Native endpoints are
        // dual-stack by default so iOS remains routable on IPv6-only/NAT64 networks.
        builder = builder.clear_ip_transports();
        builder = builder
            .bind_addr("0.0.0.0:0")
            .map_err(|e| anyhow::anyhow!("failed to bind IPv4 transport: {}", e))?;
    }

    Ok(builder)
}

#[cfg(not(target_arch = "wasm32"))]
fn auto_connect_failure_backoff_ms(failure_count: u8) -> i64 {
    match failure_count {
        0 => 0,
        _ => {
            // Foreground transport failures should recover quickly. Slow or
            // offline peers are still bounded, but one startup race must not
            // turn an already-discovered peer into a minute-long wait.
            let exp = 500_i64.saturating_mul(1_i64 << (failure_count as u32).min(4));
            exp.min(8_000)
        }
    }
}

#[cfg(not(target_arch = "wasm32"))]
fn auto_connect_admission_rejection_backoff_ms(failure_count: u8) -> i64 {
    match failure_count {
        0 => 0,
        _ => {
            // Invalid or unauthorized credentials are not a reachability
            // failure. Keep the conservative schedule until coordination
            // publishes materially new ticket data.
            let exp = 1_000_i64.saturating_mul(1_i64 << (failure_count as u32).min(5));
            exp.min(30_000)
        }
    }
}

#[cfg(not(target_arch = "wasm32"))]
fn auto_connect_network_change_failure_threshold() -> u8 {
    std::env::var("OPENRTC_NETWORK_CHANGE_FAILURE_THRESHOLD")
        .ok()
        .and_then(|value| value.parse::<u8>().ok())
        .map(|value| value.clamp(1, 6))
        .unwrap_or(2)
}

/// Grace period for the non-initiator before it escalates to dial.
/// Uses exponential backoff with 20 % jitter to avoid duplicate-storm collisions.
/// Progression: 2 s → 4 s → 8 s → 16 s (cap) with ±20 % jitter.
#[cfg(not(target_arch = "wasm32"))]
fn non_initiator_escalation_grace_ms(escalation_count: u8) -> i64 {
    let base_ms = 2_000_i64;
    let exp_ms = base_ms.saturating_mul(1_i64 << (escalation_count as u32).min(3));
    let capped_ms = exp_ms.min(16_000);
    // Add ±20 % jitter using the escalation count as a cheap deterministic seed.
    // A real random source would be nicer but we avoid pulling in a dependency.
    let jitter_pct = ((escalation_count as i64 * 37 + 7) % 41) - 20; // -20..+20
    let jitter_ms = capped_ms * jitter_pct / 100;
    (capped_ms + jitter_ms).max(1_000)
}

#[cfg(not(target_arch = "wasm32"))]
fn next_non_initiator_wakeup_delay_ms(
    wait_started_at: &std::collections::HashMap<String, i64>,
    escalation_count: &std::collections::HashMap<String, u8>,
    now_ms: i64,
) -> Option<u64> {
    wait_started_at
        .iter()
        .map(|(device_id, started_at)| {
            let grace_ms = non_initiator_escalation_grace_ms(
                escalation_count.get(device_id).copied().unwrap_or(0),
            );
            started_at
                .saturating_add(grace_ms)
                .saturating_sub(now_ms)
                .max(1) as u64
        })
        .min()
}

#[cfg(not(target_arch = "wasm32"))]
fn register_auto_connect_failure(
    remote_device_id: &str,
    now_ms: i64,
    failure_count: &mut std::collections::HashMap<String, u8>,
    failure_backoff_until: &mut std::collections::HashMap<String, i64>,
) {
    let count = failure_count
        .entry(remote_device_id.to_string())
        .or_insert(0);
    *count = count.saturating_add(1).min(10);
    let backoff_ms = auto_connect_failure_backoff_ms(*count);
    failure_backoff_until.insert(
        remote_device_id.to_string(),
        now_ms.saturating_add(backoff_ms),
    );
}

#[cfg(not(target_arch = "wasm32"))]
fn register_auto_connect_admission_rejection(
    remote_device_id: &str,
    now_ms: i64,
    failure_count: &mut std::collections::HashMap<String, u8>,
    failure_backoff_until: &mut std::collections::HashMap<String, i64>,
) {
    let count = failure_count
        .entry(remote_device_id.to_string())
        .or_insert(0);
    *count = count.saturating_add(1).min(10);
    let backoff_ms = auto_connect_admission_rejection_backoff_ms(*count);
    failure_backoff_until.insert(
        remote_device_id.to_string(),
        now_ms.saturating_add(backoff_ms),
    );
}

#[cfg(not(target_arch = "wasm32"))]
fn clear_auto_connect_failure_state(
    remote_device_id: &str,
    failure_count: &mut std::collections::HashMap<String, u8>,
    failure_backoff_until: &mut std::collections::HashMap<String, i64>,
) {
    failure_count.remove(remote_device_id);
    failure_backoff_until.remove(remote_device_id);
}

#[cfg(target_arch = "wasm32")]
fn emit_wasm_connection_state_event(snapshot: &StateSnapshot) {
    use wasm_bindgen::JsValue;

    let Some(window) = web_sys::window() else {
        return;
    };

    let detail = js_sys::Object::new();
    let _ = js_sys::Reflect::set(
        &detail,
        &JsValue::from_str("connectionId"),
        &JsValue::from_str(snapshot.connection_id.as_str()),
    );
    let _ = js_sys::Reflect::set(
        &detail,
        &JsValue::from_str("state"),
        &JsValue::from_str(snapshot.state.as_str()),
    );
    let _ = js_sys::Reflect::set(
        &detail,
        &JsValue::from_str("createdAt"),
        &JsValue::from_f64(snapshot.created_at as f64),
    );
    let _ = js_sys::Reflect::set(
        &detail,
        &JsValue::from_str("updatedAt"),
        &JsValue::from_f64(snapshot.updated_at as f64),
    );
    let _ = js_sys::Reflect::set(
        &detail,
        &JsValue::from_str("transportState"),
        &JsValue::from_str(snapshot.transport_state.as_str()),
    );
    let _ = js_sys::Reflect::set(
        &detail,
        &JsValue::from_str("protocolState"),
        &JsValue::from_str(snapshot.protocol_state.as_str()),
    );
    let _ = js_sys::Reflect::set(
        &detail,
        &JsValue::from_str("routable"),
        &JsValue::from_bool(snapshot.routable),
    );
    let _ = js_sys::Reflect::set(
        &detail,
        &JsValue::from_str("logicalSessionTerminal"),
        &JsValue::from_bool(snapshot.logical_session_terminal),
    );
    let _ = js_sys::Reflect::set(
        &detail,
        &JsValue::from_str("transportGeneration"),
        &JsValue::from_f64(snapshot.transport_generation as f64),
    );
    let _ = js_sys::Reflect::set(
        &detail,
        &JsValue::from_str("routeGeneration"),
        &JsValue::from_f64(snapshot.route_generation as f64),
    );
    let _ = js_sys::Reflect::set(
        &detail,
        &JsValue::from_str("activeTransportStableId"),
        &snapshot
            .active_transport_stable_id
            .map(|value| JsValue::from_f64(value as f64))
            .unwrap_or(JsValue::NULL),
    );
    let _ = js_sys::Reflect::set(
        &detail,
        &JsValue::from_str("activeTransport"),
        &JsValue::from_str(snapshot.active_transport.as_str()),
    );
    let _ = js_sys::Reflect::set(
        &detail,
        &JsValue::from_str("parallelTransport"),
        &snapshot
            .parallel_transport
            .as_deref()
            .map(JsValue::from_str)
            .unwrap_or(JsValue::NULL),
    );
    let _ = js_sys::Reflect::set(
        &detail,
        &JsValue::from_str("replacementInProgress"),
        &JsValue::from_bool(snapshot.replacement_in_progress),
    );
    let _ = js_sys::Reflect::set(
        &detail,
        &JsValue::from_str("transitionCount"),
        &JsValue::from_f64(snapshot.transition_count as f64),
    );
    let _ = js_sys::Reflect::set(
        &detail,
        &JsValue::from_str("connectingTransitionCount"),
        &JsValue::from_f64(snapshot.connecting_transition_count as f64),
    );
    let _ = js_sys::Reflect::set(
        &detail,
        &JsValue::from_str("replacementCount"),
        &JsValue::from_f64(snapshot.replacement_count as f64),
    );
    let _ = js_sys::Reflect::set(
        &detail,
        &JsValue::from_str("retireCount"),
        &JsValue::from_f64(snapshot.retire_count as f64),
    );

    if let Some(device_id) = snapshot.device_id.as_deref() {
        let _ = js_sys::Reflect::set(
            &detail,
            &JsValue::from_str("deviceId"),
            &JsValue::from_str(device_id),
        );
    }
    if let Some(device_id_hint) = snapshot.device_id_hint.as_deref() {
        let _ = js_sys::Reflect::set(
            &detail,
            &JsValue::from_str("deviceIdHint"),
            &JsValue::from_str(device_id_hint),
        );
    }
    if let Some(remote_node_id) = snapshot.remote_node_id.as_deref() {
        let _ = js_sys::Reflect::set(
            &detail,
            &JsValue::from_str("remoteNodeId"),
            &JsValue::from_str(remote_node_id),
        );
    }
    if let Some(error) = snapshot.error.as_deref() {
        let _ = js_sys::Reflect::set(
            &detail,
            &JsValue::from_str("error"),
            &JsValue::from_str(error),
        );
    }
    if let Some(reason) = snapshot.last_disconnect_reason.as_deref() {
        let _ = js_sys::Reflect::set(
            &detail,
            &JsValue::from_str("lastDisconnectReason"),
            &JsValue::from_str(reason),
        );
    }
    if let Some(reason) = snapshot.last_reconnect_reason.as_deref() {
        let _ = js_sys::Reflect::set(
            &detail,
            &JsValue::from_str("lastReconnectReason"),
            &JsValue::from_str(reason),
        );
    }

    let init = web_sys::CustomEventInit::new();
    init.set_detail(&detail.into());
    if let Ok(event) =
        web_sys::CustomEvent::new_with_event_init_dict("connection-state-changed", &init)
    {
        let _ = window.dispatch_event(&event);
    }
}

fn parse_endpoint_ticket(ticket: &str) -> anyhow::Result<iroh::EndpointAddr> {
    let trimmed = ticket.trim();
    if trimmed.is_empty() {
        return Err(anyhow::anyhow!("endpoint ticket is required"));
    }

    let parsed = EndpointTicket::from_str(trimmed)
        .map_err(|e| anyhow::anyhow!("invalid endpoint ticket: {}", e))?;
    Ok(parsed.endpoint_addr().clone())
}

#[derive(Debug, Clone, serde::Serialize)]
#[serde(rename_all = "camelCase")]
pub struct RuntimeStatus {
    pub ready: bool,
    pub node_id: Option<String>,
    pub transport: TransportStatus,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize)]
#[serde(rename_all = "kebab-case")]
pub enum CapabilityMaturity {
    Stable,
    Preview,
    SupportOnly,
    Unavailable,
}

#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize)]
#[serde(rename_all = "camelCase")]
pub struct ProductCapabilityMaturity {
    pub bounded_rooms: CapabilityMaturity,
    pub service_nodes: CapabilityMaturity,
    pub offline_edge: CapabilityMaturity,
    pub broadcast: CapabilityMaturity,
}

impl ProductCapabilityMaturity {
    pub(crate) fn for_current_target() -> Self {
        Self {
            bounded_rooms: CapabilityMaturity::Stable,
            service_nodes: CapabilityMaturity::Preview,
            offline_edge: if cfg!(all(not(target_arch = "wasm32"), feature = "transport-lan")) {
                CapabilityMaturity::Preview
            } else {
                CapabilityMaturity::SupportOnly
            },
            broadcast: CapabilityMaturity::Preview,
        }
    }
}

#[derive(Debug, Clone, serde::Serialize)]
#[serde(rename_all = "camelCase")]
pub struct FeatureStatus {
    pub compiled: bool,
    pub enabled: bool,
}

impl FeatureStatus {
    pub(crate) fn new(compiled: bool, enabled: bool) -> Self {
        Self {
            compiled,
            enabled: compiled && enabled,
        }
    }
}

#[derive(Debug, Clone, serde::Serialize)]
#[serde(rename_all = "camelCase")]
pub struct TransportStatus {
    pub iroh_quic: FeatureStatus,
    pub iroh_lan: FeatureStatus,
    pub web_rtc: FeatureStatus,
    pub moq: FeatureStatus,
    pub ble: FeatureStatus,
    pub iroh_relay_only: bool,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub iroh_relay_transport_policy: Option<RelayPolicy>,
    pub iroh_relay: RelayStatus,
    pub route_priority: Vec<crate::route_policy::KnownRoute>,
    pub optimize_for: crate::route_policy::RouteOptimization,
    pub network_policy: NetworkPolicy,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize)]
#[serde(rename_all = "camelCase")]
pub enum IrohRelayProvider {
    /// The vendored OpenRTC Iroh fork exposes QUIC relay carriage selection.
    VendoredIroh,
    /// An upstream Iroh build exposes the WebSocket relay carriage only.
    UpstreamIroh,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize)]
#[serde(rename_all = "camelCase")]
pub enum RelayCarriage {
    Quic,
    Websocket,
    QuicWithWebsocketFallback,
    WebsocketFallback,
    Unsupported,
}

#[derive(Debug, Clone, serde::Serialize)]
#[serde(rename_all = "camelCase")]
pub struct RelayStatus {
    pub provider: IrohRelayProvider,
    pub requested_carriage: RelayPolicy,
    pub effective_carriage: RelayCarriage,
}

impl RelayStatus {
    pub(crate) fn from_policy(
        requested_carriage: RelayPolicy,
        provider: IrohRelayProvider,
    ) -> Self {
        let effective_carriage = match requested_carriage {
            RelayPolicy::WebsocketRequired => RelayCarriage::Websocket,
            RelayPolicy::Auto => {
                #[cfg(target_arch = "wasm32")]
                {
                    RelayCarriage::Websocket
                }
                #[cfg(all(not(target_arch = "wasm32"), openrtc_iroh_relay_transport_policy_api))]
                {
                    RelayCarriage::QuicWithWebsocketFallback
                }
                #[cfg(all(
                    not(target_arch = "wasm32"),
                    not(openrtc_iroh_relay_transport_policy_api)
                ))]
                {
                    RelayCarriage::WebsocketFallback
                }
            }
            RelayPolicy::QuicRequired => {
                #[cfg(target_arch = "wasm32")]
                {
                    RelayCarriage::Unsupported
                }
                #[cfg(all(not(target_arch = "wasm32"), openrtc_iroh_relay_transport_policy_api))]
                {
                    RelayCarriage::Quic
                }
                #[cfg(all(
                    not(target_arch = "wasm32"),
                    not(openrtc_iroh_relay_transport_policy_api)
                ))]
                {
                    RelayCarriage::Unsupported
                }
            }
        };

        Self {
            provider,
            requested_carriage,
            effective_carriage,
        }
    }
}

#[cfg(openrtc_iroh_relay_transport_policy_api)]
pub(crate) const fn current_iroh_relay_provider() -> IrohRelayProvider {
    IrohRelayProvider::VendoredIroh
}

#[cfg(not(openrtc_iroh_relay_transport_policy_api))]
pub(crate) const fn current_iroh_relay_provider() -> IrohRelayProvider {
    IrohRelayProvider::UpstreamIroh
}

impl TransportStatus {
    pub(crate) fn from_config_with_ble_available(
        config: &TransportConfig,
        ble_available: bool,
        relay_provider: IrohRelayProvider,
    ) -> Self {
        let lan_compiled = cfg!(feature = "transport-lan");
        let lan_enabled = config
            .iroh_lan
            .as_ref()
            .map(|lan| lan.enabled)
            .unwrap_or(false);
        let webrtc_compiled = cfg!(feature = "transport-webrtc");
        let webrtc_enabled = config.webrtc.is_some();
        let moq_compiled = cfg!(feature = "transport-moq");
        let moq_enabled = config.moq.is_some();
        let ble_compiled = ble_available;
        let ble_enabled = config.ble.as_ref().map(|ble| ble.enabled).unwrap_or(false);

        Self {
            iroh_quic: FeatureStatus::new(true, true),
            iroh_lan: FeatureStatus::new(lan_compiled, lan_enabled),
            web_rtc: FeatureStatus::new(webrtc_compiled, webrtc_enabled),
            moq: FeatureStatus::new(moq_compiled, moq_enabled),
            ble: FeatureStatus::new(ble_compiled, ble_enabled),
            iroh_relay_only: config.iroh_relay_only,
            iroh_relay_transport_policy: config.iroh_relay_transport_policy.clone(),
            iroh_relay: RelayStatus::from_policy(
                config
                    .iroh_relay_transport_policy
                    .clone()
                    .unwrap_or(RelayPolicy::WebsocketRequired),
                relay_provider,
            ),
            route_priority: config.route_priority.clone(),
            optimize_for: config.optimize_for,
            network_policy: config.network_policy,
        }
    }
}

#[derive(Debug, Clone, serde::Serialize, serde::Deserialize, PartialEq, Eq)]
#[serde(rename_all = "camelCase")]
pub struct IceServerConfig {
    #[serde(default)]
    pub urls: Vec<String>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub username: Option<String>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub credential: Option<String>,
}

#[derive(Debug, Clone, serde::Serialize, serde::Deserialize, PartialEq, Eq)]
#[serde(rename_all = "camelCase")]
pub struct WebRTCConfig {
    #[serde(default)]
    pub ice_servers: Vec<IceServerConfig>,
    #[serde(default)]
    pub privacy_mode: bool,
    #[serde(default)]
    pub lan_mode: bool,
}

impl Default for WebRTCConfig {
    fn default() -> Self {
        Self {
            ice_servers: Vec::new(),
            privacy_mode: false,
            lan_mode: false,
        }
    }
}

#[derive(Clone, serde::Serialize, serde::Deserialize, PartialEq, Eq)]
#[serde(rename_all = "camelCase")]
pub struct MoQConfig {
    #[serde(default)]
    pub relay_url: String,
    #[serde(default, skip_serializing)]
    pub access_token: Option<String>,
}

impl std::fmt::Debug for MoQConfig {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        let relay_label = self
            .relay_url
            .split(['?', '#'])
            .next()
            .unwrap_or("invalid-moq-relay-url");
        formatter
            .debug_struct("MoQConfig")
            .field("relay_url", &relay_label)
            .field(
                "access_token",
                &self.access_token.as_ref().map(|_| "[redacted]"),
            )
            .finish()
    }
}

impl Default for MoQConfig {
    fn default() -> Self {
        Self {
            relay_url: String::new(),
            access_token: None,
        }
    }
}

#[derive(Debug, Clone, Copy, serde::Serialize, serde::Deserialize, PartialEq, Eq)]
#[serde(rename_all = "camelCase")]
pub enum RelayPolicy {
    Auto,
    QuicRequired,
    WebsocketRequired,
}

/// Endpoint-wide network boundary.
///
/// `LocalOnly` is a hard construction-time policy: the endpoint omits public
/// DNS discovery and relays, and only locally scoped IP candidates (plus an
/// explicitly installed native BLE mechanism) are eligible. It is not a
/// best-effort route preference.
#[derive(Debug, Clone, Copy, Default, serde::Serialize, serde::Deserialize, PartialEq, Eq)]
#[serde(rename_all = "kebab-case")]
pub enum NetworkPolicy {
    #[default]
    Managed,
    LocalOnly,
}

#[derive(Debug, Clone, serde::Serialize, serde::Deserialize, PartialEq, Eq)]
#[serde(rename_all = "camelCase")]
pub struct TransportConfig {
    #[serde(default)]
    pub network_policy: NetworkPolicy,
    /// Allow relay mechanisms selected by their independently configured
    /// runtime capabilities. Defaults to true.
    #[serde(default = "default_true")]
    pub relay: bool,
    /// Hard peer-address privacy policy across every configured route.
    #[serde(default)]
    pub privacy_mode: bool,
    /// Fine-grained Iroh endpoint policy. Unlike `privacy_mode`, this does not
    /// constrain WebRTC, BLE, MoQ, or exact route priority.
    #[serde(default)]
    pub iroh_relay_only: bool,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub iroh_relay_transport_policy: Option<RelayPolicy>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub iroh_lan: Option<IrohLanConfig>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub webrtc: Option<WebRTCConfig>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub moq: Option<MoQConfig>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub ble: Option<BleConfig>,
    /// Exact-route preference used by the Rust peer-session selector. Routes
    /// unavailable on the current runtime or peer are skipped.
    #[serde(default = "default_route_priority")]
    pub route_priority: Vec<crate::route_policy::KnownRoute>,
    #[serde(default)]
    pub optimize_for: crate::route_policy::RouteOptimization,
}

fn default_route_priority() -> Vec<crate::route_policy::KnownRoute> {
    crate::route_policy::DEFAULT_ROUTE_PRIORITY.to_vec()
}

fn default_true() -> bool {
    true
}

impl Default for TransportConfig {
    fn default() -> Self {
        Self {
            network_policy: NetworkPolicy::Managed,
            relay: true,
            privacy_mode: false,
            iroh_relay_only: false,
            iroh_relay_transport_policy: None,
            iroh_lan: Some(IrohLanConfig::default()),
            webrtc: None,
            moq: None,
            ble: None,
            route_priority: default_route_priority(),
            optimize_for: crate::route_policy::RouteOptimization::Balanced,
        }
    }
}

impl TransportConfig {
    /// Rank mutually supported packet carriers under the global privacy
    /// policy. `iroh_relay_only` constrains only Iroh path selection and must
    /// not make an otherwise eligible direct WebRTC carrier disappear.
    #[cfg(any(test, feature = "transport-webrtc", feature = "transport-moq"))]
    pub(crate) fn ranked_iroh_carriers(
        &self,
        supports_webrtc: bool,
        supports_moq: bool,
    ) -> Vec<crate::route_policy::KnownRoute> {
        crate::route_policy::rank_iroh_carriers(
            &self.route_priority,
            supports_webrtc,
            supports_moq,
            self.privacy_mode,
            self.webrtc
                .as_ref()
                .is_some_and(|webrtc| webrtc.privacy_mode),
        )
    }

    /// Returns true when applying `next` requires constructing a new Iroh
    /// endpoint. These options control socket binding or address discovery and
    /// cannot be truthfully changed on an endpoint that is already running.
    #[cfg(not(target_arch = "wasm32"))]
    pub(crate) fn endpoint_rebind_required(&self, next: &Self) -> bool {
        self.network_policy != next.network_policy
            || self.relay != next.relay
            || self.iroh_relay_only != next.iroh_relay_only
            || self.iroh_relay_transport_policy != next.iroh_relay_transport_policy
            || self.iroh_lan != next.iroh_lan
    }

    pub fn sanitize_for_runtime(self) -> Self {
        self.sanitize_for_runtime_with_ble_available(false)
    }

    /// Apply static transport policy before a native host has had a chance to
    /// register its deferred hardware providers. Keeping requested BLE intent
    /// here does not advertise capability: runtime status and handshakes still
    /// require the provider to be registered. WASM has no native provider
    /// installation phase, so it continues to remove BLE immediately.
    fn sanitize_for_client_construction(self) -> Self {
        #[cfg(not(target_arch = "wasm32"))]
        let deferred_ble_requested = self.ble.as_ref().is_some_and(|config| config.enabled);
        #[cfg(target_arch = "wasm32")]
        let deferred_ble_requested = false;
        self.sanitize_for_runtime_with_ble_available(deferred_ble_requested)
    }

    pub(crate) fn sanitize_for_runtime_with_ble_available(mut self, ble_available: bool) -> Self {
        let mut seen_routes = std::collections::HashSet::new();
        self.route_priority
            .retain(|route| seen_routes.insert(*route));
        if self.route_priority.is_empty() {
            self.route_priority = default_route_priority();
        }

        if self.network_policy == NetworkPolicy::LocalOnly {
            self.relay = false;
            self.privacy_mode = false;
            self.iroh_relay_only = false;
            self.iroh_relay_transport_policy = None;
            self.iroh_lan = Some(IrohLanConfig {
                enabled: true,
                advertise: self
                    .iroh_lan
                    .as_ref()
                    .map(|lan| lan.advertise)
                    .unwrap_or(true),
            });
            self.webrtc = None;
            self.moq = None;
            self.route_priority.retain(|route| {
                matches!(
                    route,
                    crate::route_policy::KnownRoute::IrohLan
                        | crate::route_policy::KnownRoute::IrohQuic
                        | crate::route_policy::KnownRoute::Ble
                )
            });
            if self.route_priority.is_empty() {
                self.route_priority = vec![
                    crate::route_policy::KnownRoute::IrohLan,
                    crate::route_policy::KnownRoute::IrohQuic,
                    crate::route_policy::KnownRoute::Ble,
                ];
            }
        }

        if self.privacy_mode && self.relay {
            self.iroh_relay_only = true;
            self.iroh_lan = None;
            self.ble = None;
            self.route_priority.retain(|route| {
                !matches!(
                    route,
                    crate::route_policy::KnownRoute::IrohLan
                        | crate::route_policy::KnownRoute::IrohQuic
                        | crate::route_policy::KnownRoute::Ble
                )
            });
            if let Some(webrtc) = self.webrtc.as_mut() {
                webrtc.privacy_mode = true;
                webrtc.lan_mode = false;
                for server in &mut webrtc.ice_servers {
                    server.urls.retain(|url| {
                        let url = url.trim().to_ascii_lowercase();
                        url.starts_with("turn:") || url.starts_with("turns:")
                    });
                }
                webrtc.ice_servers.retain(|server| !server.urls.is_empty());
            }
        }

        if !self.relay {
            self.privacy_mode = false;
            self.iroh_relay_only = false;
            self.iroh_relay_transport_policy = None;
            self.moq = None;
            self.route_priority.retain(|route| {
                !matches!(
                    route,
                    crate::route_policy::KnownRoute::IrohRelay
                        | crate::route_policy::KnownRoute::Moq
                )
            });
            if self.route_priority.is_empty() {
                self.route_priority = vec![
                    crate::route_policy::KnownRoute::IrohLan,
                    crate::route_policy::KnownRoute::IrohQuic,
                    crate::route_policy::KnownRoute::WebRtc,
                    crate::route_policy::KnownRoute::Ble,
                ];
            }
            if let Some(webrtc) = self.webrtc.as_mut() {
                webrtc.privacy_mode = false;
                for server in &mut webrtc.ice_servers {
                    server.urls.retain(|url| {
                        let url = url.trim().to_ascii_lowercase();
                        !url.starts_with("turn:") && !url.starts_with("turns:")
                    });
                }
                webrtc.ice_servers.retain(|server| !server.urls.is_empty());
            }
        }

        if self.ble.as_ref().map(|ble| ble.enabled).unwrap_or(false) && !ble_available {
            self.ble = None;
        }

        self
    }
}

pub struct ClientBuilder {
    #[cfg(not(target_arch = "wasm32"))]
    project_id: String,
    app_tag: String,
    transport_config: TransportConfig,
    #[cfg(not(target_arch = "wasm32"))]
    token_provider: Arc<dyn Fn() -> Option<String> + Send + Sync>,
    signaling: Option<Arc<dyn SignalingBackend>>,
    room: Option<Arc<dyn RoomBackend>>,
}

impl ClientBuilder {
    pub fn new_provider_neutral(
        app_tag: String,
        identity_credential_provider: Box<dyn Fn() -> Option<String> + Send + Sync>,
    ) -> Self {
        #[cfg(target_arch = "wasm32")]
        let _ = identity_credential_provider;
        Self {
            // OpenRTC 2.0 live coordination is installed explicitly through a
            // provider-neutral SignalingBackend. An empty legacy project ID is
            // deliberate and prevents an accidental Firebase control path.
            #[cfg(not(target_arch = "wasm32"))]
            project_id: String::new(),
            app_tag,
            transport_config: TransportConfig::default(),
            #[cfg(not(target_arch = "wasm32"))]
            token_provider: Arc::from(identity_credential_provider),
            signaling: None,
            room: None,
        }
    }

    // Test-only compatibility for the existing transport corpus. Production
    // crates expose this only when the rollback feature is explicitly enabled.
    #[cfg(test)]
    pub(crate) fn new_for_test(
        project_id: String,
        api_key: String,
        token_provider: Box<dyn Fn() -> Option<String> + Send + Sync>,
    ) -> Self {
        Self::new_with_app_tag(
            project_id,
            crate::app_tag_from_api_key(&api_key),
            token_provider,
        )
    }

    #[cfg(test)]
    pub fn new_with_app_tag(
        project_id: String,
        app_tag: String,
        token_provider: Box<dyn Fn() -> Option<String> + Send + Sync>,
    ) -> Self {
        Self {
            #[cfg(not(target_arch = "wasm32"))]
            project_id,
            app_tag,
            transport_config: TransportConfig::default(),
            #[cfg(not(target_arch = "wasm32"))]
            token_provider: Arc::from(token_provider),
            signaling: None,
            room: None,
        }
    }

    pub fn transport_config(mut self, transport_config: TransportConfig) -> Self {
        self.transport_config = transport_config.sanitize_for_client_construction();
        self
    }

    pub fn signaling_backend(mut self, signaling: Arc<dyn SignalingBackend>) -> Self {
        self.signaling = Some(signaling);
        self
    }

    pub fn room_backend(mut self, room: Arc<dyn RoomBackend>) -> Self {
        self.room = Some(room);
        self
    }

    pub fn build(self) -> Client {
        let app_backgrounded = Arc::new(AtomicBool::new(false));

        let room = self
            .room
            .unwrap_or_else(|| Arc::new(GatewayRequiredRoomBackend) as Arc<dyn RoomBackend>);

        let signaling = self.signaling.unwrap_or_else(|| {
            Arc::new(GatewayRequiredSignalingBackend) as Arc<dyn SignalingBackend>
        });

        #[cfg(not(target_arch = "wasm32"))]
        let (native_device_updates, _) = tokio::sync::broadcast::channel(32);
        #[cfg(not(target_arch = "wasm32"))]
        let (native_connection_state_updates, _) = tokio::sync::broadcast::channel(64);
        #[cfg(not(target_arch = "wasm32"))]
        let (native_peer_data_updates, _) = tokio::sync::broadcast::channel(256);
        #[cfg(not(target_arch = "wasm32"))]
        let (native_application_streams, native_application_streams_receiver) =
            async_channel::bounded(256);

        Client {
            app_tag: self.app_tag,
            broadcasts: crate::broadcast::Broadcasts::default(),
            signaling,
            room,
            node_id: Arc::new(RwLock::new(None)),
            iroh_endpoint: Arc::new(RwLock::new(None)),
            #[cfg(target_arch = "wasm32")]
            iroh_node: Arc::new(RwLock::new(None)),
            #[cfg(not(target_arch = "wasm32"))]
            iroh_node: Arc::new(RwLock::new(None)),
            #[cfg(not(target_arch = "wasm32"))]
            native_application_streams,
            #[cfg(not(target_arch = "wasm32"))]
            native_application_streams_receiver,
            connection_manager: Arc::new(crate::connection_manager::ConnectionManager::new()),
            session_token_registry: Arc::new(crate::session_token::SessionTokenRegistry::new()),
            #[cfg(not(target_arch = "wasm32"))]
            inbound_session_admission_transport_ids: Arc::new(StdRwLock::new(HashMap::new())),
            #[cfg(not(target_arch = "wasm32"))]
            offline_admission_requirements: Arc::new(StdRwLock::new(HashMap::new())),
            #[cfg(not(target_arch = "wasm32"))]
            offline_admission_transport_ids: Arc::new(StdRwLock::new(HashMap::new())),
            #[cfg(not(target_arch = "wasm32"))]
            offline_runtime: Arc::new(tokio::sync::Mutex::new(None)),
            #[cfg(not(target_arch = "wasm32"))]
            offline_proof_attempts: Arc::new(tokio::sync::Mutex::new(HashMap::new())),
            #[cfg(not(target_arch = "wasm32"))]
            native_admission_stream_contracts: Arc::new(StdRwLock::new(HashMap::new())),
            remote_session_admission_proofs: Arc::new(StdRwLock::new(HashMap::new())),
            pending_inline_reciprocal_admissions: Arc::new(StdRwLock::new(HashMap::new())),
            outbound_application_security_epoch_fingerprints: Arc::new(StdRwLock::new(
                HashMap::new(),
            )),
            #[cfg(not(target_arch = "wasm32"))]
            pending_reciprocal_session_admission_requests: Arc::new(StdRwLock::new(HashSet::new())),
            #[cfg(not(target_arch = "wasm32"))]
            native_route_repair_credentials: Arc::new(StdRwLock::new(HashMap::new())),
            scoped_route_repair_credentials: Arc::new(StdRwLock::new(HashMap::new())),
            connection_application_crypto_keys:
                crate::client::application_crypto_impl::new_connection_application_crypto_key_map(),
            connection_application_crypto_state: Arc::new(StdRwLock::new(())),
            connection_application_crypto_required:
                crate::client::application_crypto_impl::new_connection_application_crypto_required_set(),
            #[cfg(not(target_arch = "wasm32"))]
            trusted_user_device_application_crypto_required: Arc::new(AtomicBool::new(false)),
            connection_application_crypto_confirmed:
                crate::client::application_crypto_impl::new_connection_application_crypto_confirmed_map(),
            #[cfg(not(target_arch = "wasm32"))]
            connection_application_route_updates: Arc::new(tokio::sync::Notify::new()),
            connection_application_crypto_outbound_sequences:
                crate::client::application_crypto_impl::new_connection_application_crypto_outbound_sequences(),
            connection_application_key_agreements:
                crate::client::application_crypto_impl::new_connection_application_key_agreement_map(),
            #[cfg(not(target_arch = "wasm32"))]
            managed_scope_tickets: Arc::new(StdRwLock::new(HashMap::new())),
            known_endpoint_addrs: Arc::new(RwLock::new(HashMap::new())),
            known_device_ids_by_node: Arc::new(StdRwLock::new(HashMap::new())),
            known_device_endpoint_revision: tokio::sync::watch::channel(0).0,
            sparse_fanout: Arc::new(tokio::sync::Mutex::new(
                crate::sparse_fanout::SparseFanoutState::default(),
            )),
            #[cfg(not(target_arch = "wasm32"))]
            native_device_identity: Arc::new(RwLock::new(None)),
            #[cfg(not(target_arch = "wasm32"))]
            native_device_base_dir: Arc::new(RwLock::new(None)),
            #[cfg(not(target_arch = "wasm32"))]
            native_device_identity_init_guard: Arc::new(tokio::sync::Mutex::new(())),
            #[cfg(not(target_arch = "wasm32"))]
            native_device_updates,
            #[cfg(not(target_arch = "wasm32"))]
            native_connection_state_updates,
            #[cfg(not(target_arch = "wasm32"))]
            native_peer_data_updates,
            auto_connect_loop_key: Arc::new(Mutex::new(None)),
            auto_connect_generation: Arc::new(AtomicU64::new(0)),
            #[cfg(not(target_arch = "wasm32"))]
            native_auto_connect_wake: Arc::new(tokio::sync::Notify::new()),
            #[cfg(not(target_arch = "wasm32"))]
            external_desired_peer_actor: Arc::new(tokio::sync::Mutex::new(
                auto_connect_impl::NativeExternalAutoConnectActorState::default(),
            )),
            auto_connect_exclusion_owner: Arc::new(Mutex::new(())),
            auto_connect_excluded: Arc::new(Mutex::new(HashSet::new())),
            auto_connect_peer_requested_excluded: Arc::new(Mutex::new(HashSet::new())),
            auto_connect_excluded_node_aliases: Arc::new(Mutex::new(HashMap::new())),
            app_backgrounded,
            background_execution_allowed: Arc::new(AtomicBool::new(false)),
            #[cfg(target_arch = "wasm32")]
            wasm_accept_bridge_started: Arc::new(std::sync::atomic::AtomicBool::new(false)),
            #[cfg(target_arch = "wasm32")]
            last_emitted_connection_states: Arc::new(Mutex::new(std::collections::HashMap::new())),
            #[cfg(target_arch = "wasm32")]
            last_empty_peer_sessions_warning_ms: Arc::new(AtomicU64::new(0)),
            iroh_init_guard: Arc::new(tokio::sync::Mutex::new(())),
            #[cfg(any(feature = "transport-webrtc", feature = "transport-moq"))]
            iroh_packet_carriers: Arc::new(RwLock::new(HashMap::new())),
            managed_connect_gates: Arc::new(tokio::sync::Mutex::new(HashMap::new())),
            #[cfg(not(target_arch = "wasm32"))]
            iroh_path_watcher_stable_ids: Arc::new(Mutex::new(HashSet::new())),
            #[cfg(not(target_arch = "wasm32"))]
            native_custom_transport_kinds: Arc::new(RwLock::new(HashMap::new())),
            #[cfg(not(target_arch = "wasm32"))]
            native_transport_upgrade_providers: Arc::new(RwLock::new(HashMap::new())),
            #[cfg(target_arch = "wasm32")]
            wasm_transport_upgrade_gates: Arc::new(tokio::sync::Mutex::new(HashMap::new())),
            #[cfg(not(target_arch = "wasm32"))]
            native_transport_upgrade_gates: Arc::new(tokio::sync::Mutex::new(HashMap::new())),
            #[cfg(not(target_arch = "wasm32"))]
            native_ble_upgrade_attempts: Arc::new(tokio::sync::Mutex::new(HashMap::new())),
            #[cfg(not(target_arch = "wasm32"))]
            native_peer_transport_capabilities: Arc::new(RwLock::new(HashMap::new())),
            presence_loop_tx: Arc::new(Mutex::new(None)),
            #[cfg(not(target_arch = "wasm32"))]
            project_id: self.project_id,
            #[cfg(not(target_arch = "wasm32"))]
            token_provider: self.token_provider,
            transport_config: Arc::new(RwLock::new(
                self.transport_config.sanitize_for_client_construction(),
            )),
            iroh_carrier_policy_epoch: Arc::new(AtomicU64::new(0)),
            iroh_carrier_peer_policy_epochs: Arc::new(StdRwLock::new(HashMap::new())),
            #[cfg(all(not(target_arch = "wasm32"), feature = "transport-lan"))]
            local_discovery_registry: crate::local_discovery::LocalDiscoveryRegistry::new(),
            #[cfg(all(not(target_arch = "wasm32"), feature = "transport-lan"))]
            mdns_address_lookup: Arc::new(RwLock::new(None)),
            #[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
            native_webrtc_carrier_attempts: Arc::new(RwLock::new(HashMap::new())),
            #[cfg(all(
                not(target_arch = "wasm32"),
                feature = "iroh-carrier-core",
                any(feature = "test-harness", feature = "testing-endpoints")
            ))]
            native_iroh_carrier_debug_events: Arc::new(StdRwLock::new(
                std::collections::VecDeque::new(),
            )),
            #[cfg(all(not(target_arch = "wasm32"), feature = "transport-moq"))]
            native_moq_carrier_attempts: Arc::new(RwLock::new(HashMap::new())),
            #[cfg(not(target_arch = "wasm32"))]
            native_control_streams: Arc::new(tokio::sync::Mutex::new(HashMap::new())),
            #[cfg(not(target_arch = "wasm32"))]
            native_transport_protocol_activity: Arc::new(StdRwLock::new(HashMap::new())),
            #[cfg(all(not(target_arch = "wasm32"), feature = "experimental-scoped-actor"))]
            scoped_connection_actor_registry: Arc::new(RwLock::new(None)),
            #[cfg(not(target_arch = "wasm32"))]
            auth_readiness: Arc::new(crate::client::auth_readiness::AuthReadinessStore::new()),
            #[cfg(not(target_arch = "wasm32"))]
            scope_classifier: Arc::new(RwLock::new(default_classifier())),
        }
    }
}

// Delegated implementation modules for readability and ownership boundaries.
// - `core_impl`: constructors, endpoint lifecycle, transport primitives, accept bridges.
// - `admission_impl`: session-token admission and managed-scope persistence.
// - `state_signaling_impl`: peer state APIs, managed connection snapshots, signaling loops.
// - `auto_connect_impl`: transport liveness checks and auto-connect policy loop internals.
// - `scoped_connection_actor`: optional native experiment, never a default
//   lifecycle authority until it owns real dial and channel routing end-to-end.
mod admission_impl;
mod application_crypto_impl;
#[cfg(not(target_arch = "wasm32"))]
pub mod auth_readiness;
mod auto_connect_impl;
mod core_impl;
#[cfg(not(target_arch = "wasm32"))]
pub mod correlation;
#[cfg(all(not(target_arch = "wasm32"), feature = "experimental-scoped-actor"))]
mod drive_grant_actor;
#[cfg(all(
    test,
    not(target_arch = "wasm32"),
    feature = "experimental-scoped-actor"
))]
mod drive_grant_actor_tests;
#[cfg(not(target_arch = "wasm32"))]
pub mod scope_classifier;
#[cfg(all(not(target_arch = "wasm32"), feature = "experimental-scoped-actor"))]
pub mod scoped_connection_actor;
mod sparse_fanout_impl;
mod state_signaling_impl;
#[cfg(not(target_arch = "wasm32"))]
mod transport_upgrade_impl;
#[cfg(all(
    target_arch = "wasm32",
    any(feature = "transport-webrtc", feature = "transport-moq")
))]
mod wasm_carrier_impl;
#[cfg(all(not(target_arch = "wasm32"), test))]
pub(crate) use transport_upgrade_impl::{
    retire_losing_native_control_candidate, retire_replaced_native_control_send,
};

impl Client {
    /// Opens non-peer live broadcasts through the shared Rust lifecycle owner.
    /// Native applications use this directly; WASM/Tauri only adapt commands.
    pub fn broadcasts(&self) -> crate::broadcast::Broadcasts {
        self.broadcasts.clone()
    }
}

#[cfg(any(
    not(target_arch = "wasm32"),
    feature = "transport-webrtc",
    feature = "transport-moq"
))]
impl Client {
    pub(crate) fn current_iroh_carrier_policy_epoch(&self) -> u64 {
        self.iroh_carrier_policy_epoch.load(Ordering::Acquire)
    }

    #[cfg(feature = "iroh-carrier-core")]
    pub(crate) fn bump_iroh_carrier_policy_epoch(&self) -> u64 {
        self.iroh_carrier_policy_epoch
            .fetch_add(1, Ordering::AcqRel)
            .saturating_add(1)
    }

    #[cfg(feature = "iroh-carrier-core")]
    pub(crate) fn iroh_carrier_policy_epoch_is_current(&self, policy_epoch: u64) -> bool {
        self.current_iroh_carrier_policy_epoch() == policy_epoch
    }

    pub(crate) fn current_iroh_carrier_peer_policy_epoch(&self, connection_id: &str) -> u64 {
        self.iroh_carrier_peer_policy_epochs
            .read()
            .ok()
            .and_then(|epochs| epochs.get(connection_id).copied())
            .unwrap_or_default()
    }

    #[cfg(feature = "iroh-carrier-core")]
    pub(crate) fn bump_iroh_carrier_peer_policy_epoch(&self, connection_id: &str) -> u64 {
        let Ok(mut epochs) = self.iroh_carrier_peer_policy_epochs.write() else {
            return self.current_iroh_carrier_peer_policy_epoch(connection_id);
        };
        let next = epochs
            .get(connection_id)
            .copied()
            .unwrap_or_default()
            .saturating_add(1);
        epochs.insert(connection_id.to_string(), next);
        next
    }

    #[cfg(feature = "iroh-carrier-core")]
    pub(crate) fn forget_iroh_carrier_peer_policy_epoch(&self, connection_id: &str) {
        if let Ok(mut epochs) = self.iroh_carrier_peer_policy_epochs.write() {
            epochs.remove(connection_id);
        }
    }
}

/// WASM stub for `send_peer`.
///
/// On native this is provided by `transport_upgrade_impl`. On WASM the TypeScript
/// layer owns transport selection; this stub exists for API symmetry.
#[cfg(target_arch = "wasm32")]
impl Client {
    pub(crate) async fn set_relay(&self, enabled: bool) -> anyhow::Result<()> {
        anyhow::ensure!(
            self.iroh_endpoint.read().await.is_none(),
            "relay must be configured before Iroh endpoint initialization",
        );
        let mut config = self.transport_config.read().await.clone();
        config.relay = enabled;
        *self.transport_config.write().await = config.sanitize_for_runtime();
        Ok(())
    }

    pub async fn send_peer(&self, _id: &str, _data: &[u8]) -> anyhow::Result<()> {
        Err(anyhow::anyhow!(
            "send_peer: use TypeScript Connection.sendTyped() for transport-aware sending in browser environments"
        ))
    }
}

#[cfg(not(target_arch = "wasm32"))]
impl Client {
    pub fn subscribe_native_peer_data(
        &self,
    ) -> tokio::sync::broadcast::Receiver<NativePeerDataEvent> {
        self.native_peer_data_updates.subscribe()
    }

    #[cfg_attr(
        not(any(feature = "transport-webrtc", feature = "transport-moq")),
        allow(dead_code)
    )]
    pub(crate) fn emit_native_peer_data(&self, event: NativePeerDataEvent) {
        let _ = self.native_peer_data_updates.send(event);
    }
}

#[cfg(test)]
mod tests;

#[cfg(all(test, not(target_arch = "wasm32")))]
mod sparse_iroh_lane_tests;

#[cfg(all(test, not(target_arch = "wasm32")))]
mod relay_diagnostic_truth_tests {
    use super::*;

    #[test]
    fn auto_reports_requested_and_effective_relay_carriage() {
        let diagnostic = RelayStatus::from_policy(RelayPolicy::Auto, current_iroh_relay_provider());

        assert_eq!(diagnostic.requested_carriage, RelayPolicy::Auto);
        let serialized = serde_json::to_value(&diagnostic).expect("relay diagnostic serializes");
        assert_eq!(serialized["requestedCarriage"], "auto");
        #[cfg(openrtc_iroh_relay_transport_policy_api)]
        assert_eq!(
            diagnostic.effective_carriage,
            RelayCarriage::QuicWithWebsocketFallback
        );
        #[cfg(not(openrtc_iroh_relay_transport_policy_api))]
        assert_eq!(
            diagnostic.effective_carriage,
            RelayCarriage::WebsocketFallback
        );
    }

    #[cfg(not(openrtc_iroh_relay_transport_policy_api))]
    #[test]
    fn upstream_quic_required_fails_closed_before_endpoint_bind() {
        let result = apply_native_network_preferences(
            iroh::Endpoint::builder(iroh::endpoint::presets::N0),
            false,
            RelayPolicy::QuicRequired,
        );

        assert!(result.is_err());
        assert!(result
            .err()
            .expect("unsupported QUIC policy should return an error")
            .to_string()
            .contains("unsupported by this upstream-Iroh build"));
    }

    #[cfg(openrtc_iroh_relay_transport_policy_api)]
    #[test]
    fn vendored_quic_required_reports_quic_and_keeps_endpoint_policy_available() {
        let diagnostic =
            RelayStatus::from_policy(RelayPolicy::QuicRequired, current_iroh_relay_provider());

        assert_eq!(diagnostic.effective_carriage, RelayCarriage::Quic);
        assert_eq!(diagnostic.provider, IrohRelayProvider::VendoredIroh);
    }
}