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(not(target_arch = "wasm32"))]
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_scope_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 IncomingBiStreamDisposition {
    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)]
pub(crate) struct NativePeerDataGeneration {
    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,
}

/// 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::SessionTokenStreamContract>,
    pub outbound: Option<crate::native_protocol::SessionTokenStreamContract>,
}

#[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,
}

#[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
#[derive(Debug, Clone)]
pub(crate) struct NativeWebRTCSuppression {
    pub until_ms: i64,
    pub reason: String,
}

/// Records that an upgrade attempt is currently in-flight for a connection.
/// Used to serialize the many triggers (inbound-sdp, admission, health-check,
/// relay-path, foreground, handshake, scheduled-retry) onto a single attempt
/// at a time so they stop tearing each other's sessions down.
#[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
#[derive(Debug, Clone)]
pub(crate) struct NativeWebRTCAttemptInFlight {
    /// Unique owner for callbacks spawned by this attempt. Negotiation ids can
    /// be reused across retries, so they are not sufficient lifecycle fences.
    pub attempt_id: String,
    pub negotiation_id: String,
    /// Physical Iroh generation that owns this attempt's signaling path.
    pub transport_stable_id: Option<u64>,
    pub started_at_ms: i64,
    pub expires_at_ms: i64,
    /// True when this attempt was created for an explicit or inbound
    /// negotiation id and should not be closed by direct-Iroh path suppression.
    pub preserve_on_direct_path: bool,
    /// Last time we logged a "skipped: attempt in flight" line for this
    /// attempt. Used to rate-limit the log so concurrent triggers don't
    /// spam it once per tick.
    pub last_skip_log_ms: i64,
}

/// Deferred retirement of the managed Iroh connection underneath an active
/// WebRTC route.
///
/// The logical connection id is stable across physical Iroh replacement, so a
/// deferred cleanup must retain the exact physical generation that requested
/// it. Otherwise the old WebRTC session can finish after a replacement is
/// installed and incorrectly remove the new connection record.
#[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) struct DeferredManagedRetirement {
    pub reason: Option<String>,
    pub transport_stable_id: Option<u64>,
    pub transport_generation: u64,
}

#[cfg(all(not(target_arch = "wasm32"), feature = "transport-moq"))]
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) struct NativeMoQRouteProofState {
    pub route_instance_id: usize,
    pub probe_id: String,
    pub local_node_id: String,
    pub remote_node_id: String,
    pub retry_active: bool,
    pub proven: bool,
}

#[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) struct NativeWebRTCRouteProofState {
    pub route_instance_id: usize,
    pub probe_id: String,
    pub negotiation_id: String,
    pub retry_active: bool,
    pub proven: bool,
}

/// 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.
#[derive(Debug, Clone, PartialEq, Eq)]
#[cfg(feature = "legacy-v1")]
pub struct NativeTrustToken {
    pub token: String,
    pub expires_at_ms: u64,
}

#[allow(dead_code)]
#[derive(Clone)]
pub struct Client {
    app_tag: String,
    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<ConnectionStateSnapshot>,
    #[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>,
    /// 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.
    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>>>,
    /// 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>>>,
    connection_application_crypto_keys:
        Arc<StdRwLock<HashMap<String, [u8; crate::application_crypto::APPLICATION_KEY_BYTES]>>>,
    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<HashSet<String>>>,
    /// Wakes native ingress tasks when reciprocal key confirmation changes.
    /// Independent QUIC streams have no cross-stream delivery order, so an
    /// application stream may arrive immediately before its control-stream
    /// acknowledgement. The Rust admission owner uses this typed wake to hold
    /// that stream behind the exact key/admission epoch without polling.
    #[cfg(not(target_arch = "wasm32"))]
    connection_application_crypto_confirmation_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::EphemeralKeyAgreement>>>,
    /// 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>>>,
    /// 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>,
    #[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<()>>,
    /// 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 NativeTransportUpgradeProvider>>>>,
    /// 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), String>>>,
    /// 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>>>,
    /// 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>>>>,
    /// Connections for which this native runtime has received a TypeScript
    /// protocol handshake. Recovery uses this protocol fact to retain the
    /// native-to-native persistent-control contract.
    #[cfg(not(target_arch = "wasm32"))]
    native_scoped_webrtc_signal_peer_connections: Arc<RwLock<HashSet<String>>>,
    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>>,
    #[cfg(all(not(target_arch = "wasm32"), feature = "transport-lan"))]
    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>>>,
    #[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
    native_webrtc_sessions:
        Arc<RwLock<HashMap<String, Arc<crate::transport::NativeWebRTCDataChannel>>>>,
    #[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
    native_webrtc_route_proofs: Arc<RwLock<HashMap<String, NativeWebRTCRouteProofState>>>,
    #[cfg(not(target_arch = "wasm32"))]
    native_optional_route_generations: Arc<RwLock<HashMap<String, NativeOptionalRouteGenerations>>>,
    #[cfg(not(target_arch = "wasm32"))]
    native_route_start_gates: Arc<route_adapter::NativeRouteStartGateRegistry>,
    #[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
    native_webrtc_suppressions: Arc<RwLock<HashMap<String, NativeWebRTCSuppression>>>,
    #[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
    native_webrtc_attempt_counts: Arc<RwLock<HashMap<String, u32>>>,
    #[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
    native_webrtc_start_gates: Arc<tokio::sync::Mutex<HashMap<String, Arc<tokio::sync::Notify>>>>,
    #[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
    pub(crate) native_webrtc_attempts_in_flight:
        Arc<RwLock<HashMap<String, NativeWebRTCAttemptInFlight>>>,
    #[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
    native_webrtc_retry_deadlines: Arc<tokio::sync::Mutex<HashMap<String, i64>>>,
    #[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
    deferred_managed_retirements: Arc<RwLock<HashMap<String, DeferredManagedRetirement>>>,
    #[cfg(all(not(target_arch = "wasm32"), feature = "transport-moq"))]
    native_moq_sessions: Arc<RwLock<HashMap<String, Arc<crate::transport::NativeMoQSession>>>>,
    #[cfg(all(not(target_arch = "wasm32"), feature = "transport-moq"))]
    native_moq_route_proofs: Arc<RwLock<HashMap<String, NativeMoQRouteProofState>>>,
    /// 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>>>,
    /// Same-transport deterministic control-stream handoff observed while a
    /// bilateral persistent admission response is still in flight.
    ///
    /// The higher-ranked duplicate can receive EOF immediately before the
    /// lower-ranked stream is installed locally. Keep the generation proofs
    /// intact across that bounded handoff, but remember the exact stable ID so
    /// an unacknowledged candidate can fail closed.
    #[cfg(not(target_arch = "wasm32"))]
    pending_native_control_handoffs: Arc<StdRwLock<HashMap<String, u64>>>,
    /// 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::ScopedConnectionActorRegistry>>>,
    >,
    /// 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,
    /// 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 NativeTransportUpgradeProvider: 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::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 DeviceConnectionStatus {
    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 TransportLatencySnapshot {
    pub webrtc: Option<u64>,
    pub webrtc_lan: Option<u64>,
    pub webrtc_turn: 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 TransportLatencySnapshot {
    fn set(&mut self, transport: &str, latency_ms: u64) {
        match transport.trim().to_ascii_lowercase().as_str() {
            "webrtc-lan" => self.webrtc_lan = Some(latency_ms),
            "webrtc-turn" => self.webrtc_turn = Some(latency_ms),
            "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-lan" => self.webrtc_lan,
            "webrtc-turn" => self.webrtc_turn,
            "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: DeviceConnectionStatus,
    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: TransportLatencySnapshot,
}

#[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,
    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 ManagedConnectionAdoption {
    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 ManagedConnectionHealthStatus {
    Healthy,
    AwaitingReplacement,
    Dead,
}

#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct ManagedConnectionHealthSnapshot {
    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: ManagedConnectionHealthStatus,
    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 ConnectionStateSnapshot {
    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,
    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>,
    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,
    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>,
    error: Option<String>,
}

#[cfg(target_arch = "wasm32")]
impl From<&ConnectionStateSnapshot> for WasmConnectionStateFingerprint {
    fn from(snapshot: &ConnectionStateSnapshot) -> 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,
            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(),
            error: snapshot.error.clone(),
        }
    }
}

#[derive(Debug, Clone, serde::Serialize, serde::Deserialize, PartialEq, Eq)]
#[serde(rename_all = "camelCase", tag = "kind")]
pub enum ManagedConnectionBridgeAction {
    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>,
) -> DeviceConnectionStatus {
    match peer.map(|value| &value.status) {
        Some(crate::connection_manager::ConnectionState::Pending)
        | Some(crate::connection_manager::ConnectionState::Connecting) => {
            DeviceConnectionStatus::Connecting
        }
        Some(crate::connection_manager::ConnectionState::Connected)
            if peer.map(peer_snapshot_settled_ready).unwrap_or(false) =>
        {
            DeviceConnectionStatus::Connected
        }
        Some(crate::connection_manager::ConnectionState::Connected) => {
            DeviceConnectionStatus::Connecting
        }
        Some(crate::connection_manager::ConnectionState::Failed) => DeviceConnectionStatus::Failed,
        Some(crate::connection_manager::ConnectionState::Closed)
        | Some(crate::connection_manager::ConnectionState::Closing) => {
            DeviceConnectionStatus::Closed
        }
        None if device.online => DeviceConnectionStatus::Online,
        None => DeviceConnectionStatus::Disconnected,
    }
}

fn snapshot_settled_ready(
    peer: Option<&crate::connection_manager::PeerSnapshot>,
    connection_status: &DeviceConnectionStatus,
) -> bool {
    matches!(connection_status, DeviceConnectionStatus::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()
            || crate::transport_label::is_independent_transport(peer.active_transport.as_str()))
}

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 optional route 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_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 => "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,
        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,
        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>,
) -> ConnectionStateSnapshot {
    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_reconnect (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_reconnect()
    })
    .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",
    };

    ConnectionStateSnapshot {
        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,
        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_transport_change_at_ms
            .max(record.updated_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(),
        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: TransportLatencySnapshot::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 base transport or an
    // independent upgraded 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: IrohRelayTransportPolicy,
) -> anyhow::Result<iroh::endpoint::Builder> {
    #[cfg(openrtc_iroh_relay_transport_policy_api)]
    {
        let policy = match relay_transport_policy {
            IrohRelayTransportPolicy::Auto => iroh::RelayTransportPolicy::Auto,
            IrohRelayTransportPolicy::QuicRequired => iroh::RelayTransportPolicy::QuicRequired,
            IrohRelayTransportPolicy::WebsocketRequired => {
                iroh::RelayTransportPolicy::WebsocketRequired
            }
        };
        builder = builder.relay_transport_policy(policy);
    }
    #[cfg(not(openrtc_iroh_relay_transport_policy_api))]
    {
        match relay_transport_policy {
            IrohRelayTransportPolicy::QuicRequired => {
                anyhow::bail!(
                    "requested relay carriage quicRequired is unsupported by this upstream-Iroh build; refusing to initialize"
                );
            }
            IrohRelayTransportPolicy::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"
                );
            }
            IrohRelayTransportPolicy::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::native_iroh_transport_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: &ConnectionStateSnapshot) {
    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("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: RuntimeTransportStatus,
}

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

impl RuntimeTransportFeatureStatus {
    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 RuntimeTransportStatus {
    pub iroh_quic: RuntimeTransportFeatureStatus,
    pub iroh_lan: RuntimeTransportFeatureStatus,
    pub web_rtc: RuntimeTransportFeatureStatus,
    pub web_rtc_lan: RuntimeTransportFeatureStatus,
    pub moq: RuntimeTransportFeatureStatus,
    pub ble: RuntimeTransportFeatureStatus,
    pub iroh_relay_only: bool,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub iroh_relay_transport_policy: Option<IrohRelayTransportPolicy>,
    pub iroh_relay: IrohRelayDiagnosticStatus,
}

#[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 IrohRelayEffectiveCarriage {
    Quic,
    Websocket,
    QuicWithWebsocketFallback,
    WebsocketFallback,
    Unsupported,
}

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

impl IrohRelayDiagnosticStatus {
    pub(crate) fn from_policy(
        requested_carriage: IrohRelayTransportPolicy,
        provider: IrohRelayProvider,
    ) -> Self {
        let effective_carriage = match requested_carriage {
            IrohRelayTransportPolicy::WebsocketRequired => IrohRelayEffectiveCarriage::Websocket,
            IrohRelayTransportPolicy::Auto => {
                #[cfg(target_arch = "wasm32")]
                {
                    IrohRelayEffectiveCarriage::Websocket
                }
                #[cfg(all(not(target_arch = "wasm32"), openrtc_iroh_relay_transport_policy_api))]
                {
                    IrohRelayEffectiveCarriage::QuicWithWebsocketFallback
                }
                #[cfg(all(
                    not(target_arch = "wasm32"),
                    not(openrtc_iroh_relay_transport_policy_api)
                ))]
                {
                    IrohRelayEffectiveCarriage::WebsocketFallback
                }
            }
            IrohRelayTransportPolicy::QuicRequired => {
                #[cfg(target_arch = "wasm32")]
                {
                    IrohRelayEffectiveCarriage::Unsupported
                }
                #[cfg(all(not(target_arch = "wasm32"), openrtc_iroh_relay_transport_policy_api))]
                {
                    IrohRelayEffectiveCarriage::Quic
                }
                #[cfg(all(
                    not(target_arch = "wasm32"),
                    not(openrtc_iroh_relay_transport_policy_api)
                ))]
                {
                    IrohRelayEffectiveCarriage::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 RuntimeTransportStatus {
    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 webrtc_lan_enabled = config
            .webrtc
            .as_ref()
            .map(|webrtc| webrtc.lan_mode)
            .unwrap_or(false);
        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: RuntimeTransportFeatureStatus::new(true, true),
            iroh_lan: RuntimeTransportFeatureStatus::new(lan_compiled, lan_enabled),
            web_rtc: RuntimeTransportFeatureStatus::new(webrtc_compiled, webrtc_enabled),
            web_rtc_lan: RuntimeTransportFeatureStatus::new(
                webrtc_compiled,
                webrtc_enabled && webrtc_lan_enabled,
            ),
            moq: RuntimeTransportFeatureStatus::new(moq_compiled, moq_enabled),
            ble: RuntimeTransportFeatureStatus::new(ble_compiled, ble_enabled),
            iroh_relay_only: config.iroh_relay_only,
            iroh_relay_transport_policy: config.iroh_relay_transport_policy.clone(),
            iroh_relay: IrohRelayDiagnosticStatus::from_policy(
                config
                    .iroh_relay_transport_policy
                    .clone()
                    .unwrap_or(IrohRelayTransportPolicy::WebsocketRequired),
                relay_provider,
            ),
        }
    }
}

#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
#[serde(rename_all = "camelCase")]
#[cfg(feature = "legacy-v1")]
pub enum AuthMode {
    External,
    Anonymous,
    Required,
}

#[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, Default)]
#[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,
}

#[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 IrohRelayTransportPolicy {
    Auto,
    QuicRequired,
    WebsocketRequired,
}

#[derive(Debug, Clone, serde::Serialize, serde::Deserialize, PartialEq, Eq)]
#[serde(rename_all = "camelCase")]
pub struct TransportConfig {
    #[serde(default)]
    pub iroh_relay_only: bool,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub iroh_relay_transport_policy: Option<IrohRelayTransportPolicy>,
    #[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>,
}

impl Default for TransportConfig {
    fn default() -> Self {
        Self {
            iroh_relay_only: false,
            iroh_relay_transport_policy: None,
            iroh_lan: Some(IrohLanConfig::default()),
            webrtc: None,
            moq: None,
            ble: None,
        }
    }
}

impl TransportConfig {
    /// 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.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)
    }

    pub(crate) fn sanitize_for_runtime_with_ble_available(mut self, ble_available: bool) -> Self {
        if self.ble.as_ref().map(|ble| ble.enabled).unwrap_or(false) && !ble_available {
            self.ble = None;
        }

        self
    }
}

pub struct ClientBuilder {
    project_id: String,
    app_tag: String,
    transport_config: TransportConfig,
    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 {
        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.
            project_id: String::new(),
            app_tag,
            transport_config: TransportConfig::default(),
            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(any(test, feature = "legacy-v1"))]
    pub fn new(
        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(any(test, feature = "legacy-v1"))]
    pub fn new_with_app_tag(
        project_id: String,
        app_tag: String,
        token_provider: Box<dyn Fn() -> Option<String> + Send + Sync>,
    ) -> Self {
        Self {
            project_id,
            app_tag,
            transport_config: TransportConfig::default(),
            token_provider: Arc::from(token_provider),
            signaling: None,
            room: None,
        }
    }

    #[cfg(feature = "legacy-v1")]
    pub fn auth_mode(self, auth_mode: AuthMode) -> Self {
        let _ = auth_mode;
        self
    }

    #[cfg(all(not(target_arch = "wasm32"), feature = "legacy-v1"))]
    pub fn native_auth_state(mut self, auth_state: crate::native_auth::NativeAuthState) -> Self {
        self.token_provider = Arc::from(auth_state.token_provider());
        self
    }

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

    #[cfg(feature = "legacy-v1")]
    pub fn native_trust_token_provider(
        self,
        provider: Box<dyn Fn() -> Option<NativeTrustToken> + Send + Sync>,
    ) -> Self {
        let _ = provider;
        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,
            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"))]
            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())),
            connection_application_crypto_keys:
                crate::client::application_crypto_impl::new_connection_application_crypto_key_map(),
            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_set(),
            #[cfg(not(target_arch = "wasm32"))]
            connection_application_crypto_confirmation_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())),
            #[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"))]
            external_desired_peer_actor: Arc::new(tokio::sync::Mutex::new(
                auto_connect_impl::NativeExternalAutoConnectActorState::default(),
            )),
            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,
            #[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(())),
            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(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())),
            #[cfg(not(target_arch = "wasm32"))]
            native_scoped_webrtc_signal_peer_connections: Arc::new(RwLock::new(HashSet::new())),
            presence_loop_tx: Arc::new(Mutex::new(None)),
            #[cfg(not(target_arch = "wasm32"))]
            #[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_runtime())),
            #[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_sessions: Arc::new(RwLock::new(HashMap::new())),
            #[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
            native_webrtc_route_proofs: Arc::new(RwLock::new(HashMap::new())),
            #[cfg(not(target_arch = "wasm32"))]
            native_optional_route_generations: Arc::new(RwLock::new(HashMap::new())),
            #[cfg(not(target_arch = "wasm32"))]
            native_route_start_gates: Arc::new(
                route_adapter::NativeRouteStartGateRegistry::default(),
            ),
            #[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
            native_webrtc_suppressions: Arc::new(RwLock::new(HashMap::new())),
            #[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
            native_webrtc_attempt_counts: Arc::new(RwLock::new(HashMap::new())),
            #[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
            native_webrtc_start_gates: Arc::new(tokio::sync::Mutex::new(HashMap::new())),
            #[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
            native_webrtc_attempts_in_flight: Arc::new(RwLock::new(HashMap::new())),
            #[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
            native_webrtc_retry_deadlines: Arc::new(tokio::sync::Mutex::new(HashMap::new())),
            #[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
            deferred_managed_retirements: Arc::new(RwLock::new(HashMap::new())),
            #[cfg(all(not(target_arch = "wasm32"), feature = "transport-moq"))]
            native_moq_sessions: Arc::new(RwLock::new(HashMap::new())),
            #[cfg(all(not(target_arch = "wasm32"), feature = "transport-moq"))]
            native_moq_route_proofs: 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(not(target_arch = "wasm32"))]
            pending_native_control_handoffs: 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_scope_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"))]
mod route_adapter;
#[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 state_signaling_impl;
#[cfg(all(not(target_arch = "wasm32"), feature = "transport-moq"))]
pub(crate) use route_adapter::NativeOptionalRouteTerminalDisposition;
#[cfg(not(target_arch = "wasm32"))]
pub(crate) use route_adapter::{NativeOptionalRouteGeneration, NativeOptionalRouteGenerations};
#[cfg(not(target_arch = "wasm32"))]
mod transport_upgrade_impl;
#[cfg(all(not(target_arch = "wasm32"), test))]
pub(crate) use transport_upgrade_impl::{
    retire_losing_native_control_candidate, retire_replaced_native_control_send,
};

/// 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 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 relay_diagnostic_truth_tests {
    use super::*;

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

        assert_eq!(
            diagnostic.requested_carriage,
            IrohRelayTransportPolicy::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,
            IrohRelayEffectiveCarriage::QuicWithWebsocketFallback
        );
        #[cfg(not(openrtc_iroh_relay_transport_policy_api))]
        assert_eq!(
            diagnostic.effective_carriage,
            IrohRelayEffectiveCarriage::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,
            IrohRelayTransportPolicy::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 = IrohRelayDiagnosticStatus::from_policy(
            IrohRelayTransportPolicy::QuicRequired,
            current_iroh_relay_provider(),
        );

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