openrtc/client/

auto_connect_impl.rs

//! Delegated `Client` auto-connect and transport-health internals.
//!
//! This module owns retry/backoff/tie-break policy and snapshot/event reconciliation.

#[cfg(test)]
use super::state_signaling_impl::should_present_session_token_on_connected_transport;
use super::*;

#[cfg(test)]
pub(crate) fn should_auto_present_session_token(
    has_extracted_token: bool,
    has_current_remote_admission_proof: bool,
) -> bool {
    should_present_session_token_on_connected_transport(
        has_extracted_token,
        has_current_remote_admission_proof,
    )
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[cfg(not(target_arch = "wasm32"))]
pub(crate) struct SessionAdmissionPresentationDecision {
    pub(crate) should_present: bool,
    pub(crate) request_reciprocal: bool,
}

/// Decide how the sole auto-connect actor repairs directional admission.
///
/// A current outbound proof normally suppresses duplicate presentation. The
/// exception is a current-generation route missing its inbound proof: this side
/// sends a valid typed request and asks the peer to present its own token in the
/// same pass. A validated peer request likewise forces one bounded presentation
/// through this same actor.
#[cfg(not(target_arch = "wasm32"))]
pub(crate) fn session_admission_presentation_decision(
    has_extracted_token: bool,
    has_current_remote_admission_proof: bool,
    has_current_inbound_admission_proof: bool,
    peer_requested_reciprocal: bool,
    is_designated_initiator: bool,
) -> SessionAdmissionPresentationDecision {
    if !has_extracted_token {
        return SessionAdmissionPresentationDecision {
            should_present: false,
            request_reciprocal: false,
        };
    }

    // A replacement route commonly starts with neither directional proof. Ask
    // for the reverse presentation on the first valid request instead of
    // waiting for a second health-loop pass after remote approval. A validated
    // peer request already arrived on the peer's presentation stream, so echoing
    // another reciprocal request before its ACK-bound proof becomes visible
    // creates a request ping-pong. Present once to satisfy that request without
    // bouncing it back.
    let bilateral_proof_is_current =
        has_current_remote_admission_proof && has_current_inbound_admission_proof;
    if bilateral_proof_is_current {
        return SessionAdmissionPresentationDecision {
            should_present: false,
            request_reciprocal: false,
        };
    }

    // A fresh physical leg commonly becomes visible to both auto-connect
    // actors before either directional proof exists. Only the deterministic
    // initiator may open the first persistent admission stream; otherwise both
    // peers install different canonical control streams and each closes the
    // route selected by the other. The observing peer answers that one stream
    // inline. If it could not do so, the validated request below authorizes one
    // bounded follow-up presentation without another reciprocal request.
    if !has_current_remote_admission_proof && !has_current_inbound_admission_proof {
        return SessionAdmissionPresentationDecision {
            should_present: is_designated_initiator,
            request_reciprocal: is_designated_initiator,
        };
    }

    if peer_requested_reciprocal {
        return SessionAdmissionPresentationDecision {
            should_present: true,
            request_reciprocal: false,
        };
    }

    // An inbound proof means the peer's presentation and ACK exchange has
    // already completed, so a one-way follow-up cannot collide with that
    // stream. A presenter that is still missing its inbound proof owns the
    // reciprocal request only when it is the deterministic initiator.
    let request_reciprocal = has_current_remote_admission_proof && is_designated_initiator;
    SessionAdmissionPresentationDecision {
        should_present: !has_current_remote_admission_proof || request_reciprocal,
        request_reciprocal,
    }
}

/// Return whether the persistent `user-device` actor may inspect or repair the
/// current physical transport.
///
/// A deterministic node-pair connection can be reused by an explicit
/// capability such as `drive-grant:*`, `room:*`, or `space:*`. That capability
/// owns its admission/security epoch while active. The background user-device
/// actor must yield before probing, presenting its cached ticket, or retiring
/// that transport; otherwise it can replace the explicit capability's
/// semantic security epoch and strand product traffic.
///
/// An in-flight presentation is also exclusive. The registry presentation
/// guard resolves the narrow race where an explicit actor starts admission
/// between discovery snapshots.
#[cfg(not(target_arch = "wasm32"))]
pub(crate) fn persistent_managed_actor_may_own_transport(
    current_scopes: &[String],
    admitted_scope: Option<&str>,
    presentation_in_flight: bool,
) -> bool {
    if presentation_in_flight {
        return false;
    }

    current_scopes
        .iter()
        .map(String::as_str)
        .chain(admitted_scope)
        .map(str::trim)
        .filter(|scope| !scope.is_empty())
        .all(|scope| scope == super::core_impl::PERSISTENT_MANAGED_ADMISSION_SCOPE)
}

#[cfg(not(target_arch = "wasm32"))]
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum ActiveConnectionProbeDecision {
    Healthy,
    PreserveLiveTransport,
    Failed,
}

#[cfg(not(target_arch = "wasm32"))]
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum FailedProbeRetirementOutcome {
    Retired,
    DeferredAdmissionInFlight,
    CancelledCurrentGenerationActivity,
    GenerationChanged,
}

const ACTIVE_CONNECTION_PROBE_TIMEOUT_MS: i64 = 4_000;

/// Retire shared-endpoint connections from the superseded remote process while
/// preserving the exact Iroh connection that owns OpenRTC control traffic.
///
/// OpenRTC's `transport_generation` is a process-lifetime logical fence. Iroh's
/// `Connection::stable_id()` is a separate runtime identifier used by the
/// endpoint's connection registry. Keeping this adapter typed to a concrete
/// connection prevents those two identifiers from being mixed.
#[cfg(all(
    not(target_arch = "wasm32"),
    openrtc_iroh_retire_remote_connections_api
))]
pub(crate) async fn retire_stale_endpoint_connections(
    endpoint: &iroh::Endpoint,
    endpoint_id: iroh::EndpointId,
    current_transport: &iroh::endpoint::Connection,
    reason: &[u8],
) -> usize {
    endpoint
        .retire_remote_connections_except(endpoint_id, current_transport.stable_id(), reason)
        .await
}

pub(crate) fn generation_activity_is_recent(
    last_inbound_activity_age: Option<std::time::Duration>,
    probe_interval_ms: i64,
) -> bool {
    let activity_window_ms = probe_interval_ms
        .max(0)
        .saturating_add(ACTIVE_CONNECTION_PROBE_TIMEOUT_MS) as u64;
    last_inbound_activity_age
        .is_some_and(|age| age <= std::time::Duration::from_millis(activity_window_ms))
}

#[cfg(not(target_arch = "wasm32"))]
impl Client {
    /// Wake whichever native desired-peer adapter currently feeds the one
    /// Rust auto-connect owner. Gateway-integrated hosts use the provider
    /// subscription loop; IPC-only hosts use the external revisioned mailbox.
    pub(crate) async fn wake_native_auto_connect_owner(&self) -> bool {
        if self.external_auto_connect_is_active().await {
            return self.wake_native_external_auto_connect().await;
        }

        let provider_actor_active = self
            .auto_connect_loop_key
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner())
            .is_some();
        if provider_actor_active {
            self.native_auto_connect_wake.notify_one();
        }
        provider_actor_active
    }

    /// Record valid peer protocol activity for the exact physical generation
    /// that carried it. Callers must only invoke this after parsing a complete
    /// OpenRTC control frame or after the Rust admission router has classified
    /// a current, admitted application stream. Unclassified raw bytes and
    /// failed authentication are not liveness evidence.
    pub(crate) fn record_native_transport_protocol_activity(
        &self,
        connection_id: &str,
        transport_stable_id: u64,
    ) {
        if let Ok(mut activity) = self.native_transport_protocol_activity.write() {
            activity.insert(
                connection_id.to_string(),
                super::NativeTransportProtocolActivity {
                    transport_stable_id,
                    observed_at: std::time::Instant::now(),
                },
            );
        }
    }

    pub(crate) fn native_transport_protocol_activity_age(
        &self,
        connection_id: &str,
        expected_transport_stable_id: u64,
    ) -> Option<std::time::Duration> {
        let observed_at = self
            .native_transport_protocol_activity
            .read()
            .ok()
            .and_then(|activity| activity.get(connection_id).copied())
            .filter(|activity| activity.transport_stable_id == expected_transport_stable_id)?
            .observed_at;
        Some(std::time::Instant::now().saturating_duration_since(observed_at))
    }

    pub(crate) fn forget_native_transport_protocol_activity_for_transport(
        &self,
        connection_id: &str,
        expected_transport_stable_id: u64,
    ) -> bool {
        self.native_transport_protocol_activity
            .write()
            .ok()
            .is_some_and(|mut activity| {
                if activity
                    .get(connection_id)
                    .is_some_and(|entry| entry.transport_stable_id == expected_transport_stable_id)
                {
                    activity.remove(connection_id);
                    true
                } else {
                    false
                }
            })
    }

    pub(crate) fn forget_native_transport_protocol_activity(&self, connection_id: &str) {
        if let Ok(mut activity) = self.native_transport_protocol_activity.write() {
            activity.remove(connection_id);
        }
    }

    /// Retire a physical transport after a failed liveness probe without
    /// racing a valid admission exchange on that same generation.
    ///
    /// Every native failed-probe caller must converge here. The retirement
    /// guard serializes teardown with response writers and outbound token
    /// presenters, then the exact generation and its protocol activity are
    /// re-read after the guard is acquired.
    async fn retire_transport_generation_after_failed_probe(
        &self,
        connection_id: &str,
        endpoint_id: iroh::EndpointId,
        expected_transport_stable_id: u64,
        activity_window_ms: i64,
        reason: &str,
    ) -> FailedProbeRetirementOutcome {
        let Some(retirement_guard) = self
            .session_token_registry
            .try_begin_transport_retirement(connection_id)
        else {
            return FailedProbeRetirementOutcome::DeferredAdmissionInFlight;
        };

        let current_transport_stable_id = self
            .get_connection(endpoint_id)
            .await
            .map(|connection| crate::transport_generation::for_connection(&connection));
        if current_transport_stable_id != Some(expected_transport_stable_id) {
            drop(retirement_guard);
            return FailedProbeRetirementOutcome::GenerationChanged;
        }

        if generation_activity_is_recent(
            self.native_transport_protocol_activity_age(
                connection_id,
                expected_transport_stable_id,
            ),
            activity_window_ms,
        ) {
            drop(retirement_guard);
            return FailedProbeRetirementOutcome::CancelledCurrentGenerationActivity;
        }

        let retired = self
            .disconnect_transport_generation_with_reason(
                endpoint_id,
                expected_transport_stable_id,
                reason,
            )
            .await
            .unwrap_or(false);
        drop(retirement_guard);

        if retired {
            FailedProbeRetirementOutcome::Retired
        } else {
            FailedProbeRetirementOutcome::GenerationChanged
        }
    }
}

#[cfg(target_arch = "wasm32")]
impl Client {
    pub(crate) fn native_transport_protocol_activity_age(
        &self,
        _connection_id: &str,
        _expected_transport_stable_id: u64,
    ) -> Option<std::time::Duration> {
        None
    }
}

#[cfg(not(target_arch = "wasm32"))]
pub(crate) fn active_connection_probe_decision(
    active_probe_healthy: bool,
    passive_transport_alive: bool,
) -> ActiveConnectionProbeDecision {
    if active_probe_healthy {
        ActiveConnectionProbeDecision::Healthy
    } else if passive_transport_alive {
        ActiveConnectionProbeDecision::PreserveLiveTransport
    } else {
        ActiveConnectionProbeDecision::Failed
    }
}

#[cfg(not(target_arch = "wasm32"))]
pub(crate) fn active_connection_probe_decision_with_generation_activity(
    active_probe_healthy: bool,
    passive_transport_alive: bool,
    recent_generation_activity: bool,
) -> ActiveConnectionProbeDecision {
    if recent_generation_activity {
        ActiveConnectionProbeDecision::Healthy
    } else {
        active_connection_probe_decision(active_probe_healthy, passive_transport_alive)
    }
}

#[cfg(not(target_arch = "wasm32"))]
pub(crate) fn next_active_connection_probe_failure_count(
    decision: ActiveConnectionProbeDecision,
    current_failures: u8,
) -> u8 {
    if matches!(decision, ActiveConnectionProbeDecision::Healthy) {
        0
    } else {
        current_failures.saturating_add(1).min(10)
    }
}

#[cfg(not(target_arch = "wasm32"))]
pub(crate) fn active_connection_probe_is_healthy(
    decision: ActiveConnectionProbeDecision,
    failure_count: u8,
    failure_threshold: u8,
) -> bool {
    matches!(decision, ActiveConnectionProbeDecision::Healthy)
        || (matches!(
            decision,
            ActiveConnectionProbeDecision::PreserveLiveTransport
        ) && failure_count < failure_threshold)
}

/// A session-token response proves authorization only when it is received and
/// parsed. Losing the physical Iroh leg before that response is transport
/// churn, not an authorization denial. The owner of auto-connect must preserve
/// the logical peer session in that case so the canonical replacement leg can
/// retry admission after backoff.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum SessionTokenPresentationFailureAction {
    RetryOnNextTransport,
    RejectAndDisconnect,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[cfg(not(target_arch = "wasm32"))]
pub(crate) enum RetryableSessionAdmissionFailureDecision {
    PreserveCurrentTransport,
    RetireCurrentTransport,
    ReplacementAlreadyWon,
}

/// A superseded desired-peer pass may retire a connection only when the same
/// actor accepted a newer authoritative roster that actually withdrew that
/// durable device. A revision bump caused by another peer must not close the
/// still-desired peer's shared physical transport.
pub(crate) fn superseded_desired_peer_was_withdrawn(
    actor_is_current: bool,
    current_revision: u64,
    attempt_revision: u64,
    peer_is_still_desired: bool,
) -> bool {
    actor_is_current && current_revision > attempt_revision && !peer_is_still_desired
}

/// Bound admission repair to the physical generation that failed.
///
/// A single response loss is normal during crossed reconnects, so the current
/// transport gets one retry. Repeated failures mean the passive Iroh handle is
/// no longer useful for application admission even when QUIC has not reported
/// closed. Only the generation that produced the failure may be retired; a
/// replacement that won while the request was in flight remains authoritative.
#[cfg(not(target_arch = "wasm32"))]
pub(crate) fn retryable_session_admission_failure_decision(
    failure_count: u8,
    failure_threshold: u8,
    attempted_transport_stable_id: Option<u64>,
    current_transport_stable_id: Option<u64>,
    remote_runtime_instance_changed: bool,
) -> RetryableSessionAdmissionFailureDecision {
    if attempted_transport_stable_id != current_transport_stable_id
        || attempted_transport_stable_id.is_none()
    {
        return RetryableSessionAdmissionFailureDecision::ReplacementAlreadyWon;
    }

    // A new remote runtime instance with the same durable Iroh identity can
    // leave the old process's QUIC handle looking passively alive. If that
    // exact generation cannot obtain an admission response from the newly
    // advertised runtime, preserving it for the ordinary second attempt only
    // sends the retry to the retired process. The runtime-instance observation
    // is coordination evidence, while the failed action-bound response is the
    // lifecycle input that makes this generation unusable.
    if !remote_runtime_instance_changed && failure_count < failure_threshold.max(1) {
        RetryableSessionAdmissionFailureDecision::PreserveCurrentTransport
    } else {
        RetryableSessionAdmissionFailureDecision::RetireCurrentTransport
    }
}

pub(crate) fn session_token_presentation_failure_action(
    error: &str,
) -> SessionTokenPresentationFailureAction {
    let error = error.to_ascii_lowercase();
    if error.contains("connection lost")
        || error.contains("[session-token-response:timeout]")
        || error.contains("stream finished early")
        || error.contains("peer stream ended before")
        || error.contains("connection closed")
        || error.contains("stream reset")
        || error.contains("stream stopped")
        || error.contains("failed to open bi-stream")
        || error.contains("failed to write session-token presentation")
        || error.contains("missing connection for session-token presentation")
        || error.contains("[session-token-presentation:in-flight]")
        || error.contains("authoritative-device-binding-pending")
        || error.contains(crate::session_token::PEER_ADMISSION_LEASE_EXPIRED)
    {
        SessionTokenPresentationFailureAction::RetryOnNextTransport
    } else {
        SessionTokenPresentationFailureAction::RejectAndDisconnect
    }
}

#[cfg(not(target_arch = "wasm32"))]
pub(crate) fn session_token_presentation_was_duplicate(error: &str) -> bool {
    error
        .to_ascii_lowercase()
        .contains("[session-token-presentation:in-flight]")
}

fn remote_excludes_local_device(excluded_peers: &[String], local_device_id: &str) -> bool {
    let local_device_id = local_device_id.trim();
    !local_device_id.is_empty()
        && excluded_peers
            .iter()
            .any(|peer| peer.trim().eq_ignore_ascii_case(local_device_id))
}

#[cfg(not(target_arch = "wasm32"))]
fn observe_remote_runtime_instance(
    last_known: &mut std::collections::HashMap<String, String>,
    device_id: &str,
    runtime_instance_id: Option<&str>,
) -> bool {
    let Some(runtime_instance_id) = runtime_instance_id
        .map(str::trim)
        .filter(|value| !value.is_empty() && *value != device_id)
    else {
        return false;
    };
    let key = format!("{device_id}::runtime-instance");
    last_known
        .insert(key, runtime_instance_id.to_string())
        .is_some_and(|previous| previous != runtime_instance_id)
}

#[cfg(not(target_arch = "wasm32"))]
fn runtime_instance_change_requires_replacement(
    runtime_instance_changed: bool,
    active_transport_responsive: bool,
) -> bool {
    runtime_instance_changed && !active_transport_responsive
}

#[cfg(not(target_arch = "wasm32"))]
fn observe_remote_ticket(
    last_known: &mut std::collections::HashMap<String, String>,
    device_id: &str,
    ticket: &str,
) -> bool {
    let ticket = ticket.trim();
    if ticket.is_empty() {
        return false;
    }
    let key = format!("{device_id}::ticket");
    let fingerprint = crate::session_token::token_fingerprint(ticket);
    last_known
        .insert(key, fingerprint.clone())
        .is_some_and(|previous| previous != fingerprint)
}

#[cfg(not(target_arch = "wasm32"))]
fn reconcile_remote_ticket_retry_state(
    last_known: &mut std::collections::HashMap<String, String>,
    device_id: &str,
    ticket: &str,
    last_attempt_at: &mut std::collections::HashMap<String, i64>,
    failure_count: &mut std::collections::HashMap<String, u8>,
    failure_backoff_until: &mut std::collections::HashMap<String, i64>,
) -> bool {
    if !observe_remote_ticket(last_known, device_id, ticket) {
        return false;
    }
    last_attempt_at.remove(device_id);
    clear_auto_connect_failure_state(device_id, failure_count, failure_backoff_until);
    true
}

#[derive(Debug, Clone, PartialEq, Eq, serde::Deserialize)]
#[serde(rename_all = "camelCase")]
pub(crate) struct BrowserDesiredPeer {
    #[serde(default)]
    pub(crate) connection_intent: DesiredPeerConnectionIntent,
    pub(crate) device_id: String,
    #[serde(default)]
    pub(crate) node_id: Option<String>,
    #[serde(default)]
    pub(crate) ticket: Option<String>,
    #[serde(default)]
    pub(crate) online: bool,
    #[serde(default)]
    pub(crate) session_id: Option<String>,
    #[serde(default)]
    pub(crate) excluded_peers: Vec<String>,
    #[serde(default)]
    pub(crate) topology_role: Option<SparseTopologyRole>,
    #[serde(default)]
    pub(crate) topology_revision: Option<u64>,
}

/// Membership observation does not grant automatic dialing. Explicit commands
/// still pass through the normal scoped connect/admission owner.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, serde::Deserialize)]
#[serde(rename_all = "lowercase")]
pub(crate) enum DesiredPeerConnectionIntent {
    #[default]
    Automatic,
    Observe,
}

/// Fleet observations and legacy dial candidates have independent bounds. One
/// observed record cannot exceed the gateway's 64 KiB page bound. Keep the
/// existing 100-candidate/256 KiB automatic bound, including in mixed inputs.
fn parse_desired_peer_intents(peers_json: &str) -> anyhow::Result<Vec<BrowserDesiredPeer>> {
    const AUTOMATIC_BYTES: usize = 256 * 1024;
    const OBSERVATION_BYTES: usize = 64 * 1024;
    const OBSERVATIONS: usize = 500;
    if peers_json.len() > OBSERVATIONS * OBSERVATION_BYTES + AUTOMATIC_BYTES + 602 {
        return Err(anyhow::anyhow!(
            "desired-peer intent payload exceeds its bounded fleet envelope"
        ));
    }
    let values: Vec<serde_json::Value> = serde_json::from_str(peers_json)
        .map_err(|error| anyhow::anyhow!("invalid desired-peer intent payload: {error}"))?;
    if values.len() > OBSERVATIONS + 100 {
        return Err(anyhow::anyhow!(
            "desired-peer intent payload exceeds 600 references"
        ));
    }
    let mut observations = 0;
    let mut automatic = 0;
    let mut automatic_bytes = 2;
    let mut peers = Vec::with_capacity(values.len());
    for value in values {
        let bytes = serde_json::to_vec(&value)?.len();
        let peer: BrowserDesiredPeer = serde_json::from_value(value)
            .map_err(|error| anyhow::anyhow!("invalid desired-peer intent: {error}"))?;
        match peer.connection_intent {
            DesiredPeerConnectionIntent::Observe => {
                observations += 1;
                if observations > OBSERVATIONS || bytes > OBSERVATION_BYTES {
                    return Err(anyhow::anyhow!(
                        "desired-peer observations exceed 500 records or 64 KiB per record"
                    ));
                }
            }
            DesiredPeerConnectionIntent::Automatic => {
                automatic += 1;
                automatic_bytes += bytes + usize::from(automatic > 1);
                if automatic > 100 || automatic_bytes > AUTOMATIC_BYTES {
                    return Err(anyhow::anyhow!(
                        "automatic desired peers exceed 100 candidates or 256 KiB"
                    ));
                }
            }
        }
        peers.push(peer);
    }
    if observations == 0 && peers_json.len() > AUTOMATIC_BYTES {
        return Err(anyhow::anyhow!(
            "automatic desired-peer payload exceeds 256 KiB"
        ));
    }
    Ok(peers)
}

#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Deserialize)]
#[serde(rename_all = "lowercase")]
pub(crate) enum SparseTopologyRole {
    Active,
    Backup,
}

#[derive(Debug, Default)]
struct BrowserDesiredPeerMailbox {
    revision: u64,
    transfer_demands: super::transfer_peer_demand::TransferPeerDemands,
    peers: std::collections::BTreeMap<String, BrowserDesiredPeer>,
    pending_topology_revision: Option<u64>,
    pending_ready_active_peers: std::collections::HashSet<String>,
    settled_active: std::collections::BTreeMap<String, BrowserDesiredPeer>,
    /// Devices whose authoritative live-route desire was explicitly withdrawn.
    ///
    /// Retiring the current connection is not sufficient: a revoked peer can
    /// still race back in over a cached endpoint ticket before its own gateway
    /// lifecycle observes the fatal close.  Keep the withdrawal beside the
    /// revisioned roster authority so incoming transport admission can reject
    /// that replacement generation.  A later online roster revision is the
    /// only event that clears the fence.
    withdrawn: std::collections::BTreeMap<String, Option<String>>,
}

impl BrowserDesiredPeerMailbox {
    #[cfg(test)]
    fn accept(&mut self, revision: u64, peers: Vec<BrowserDesiredPeer>) -> bool {
        self.accept_with_withdrawn(revision, peers).is_some()
    }

    /// Accepts a full authoritative roster and returns peers whose live-route
    /// desire was withdrawn by this revision. Persistent user avenues retain
    /// offline devices in the roster, so both removal and an `online ->
    /// offline` transition must retire any managed route for the old live
    /// generation.
    fn accept_with_withdrawn(
        &mut self,
        revision: u64,
        peers: Vec<BrowserDesiredPeer>,
    ) -> Option<Vec<BrowserDesiredPeer>> {
        self.accept_update(revision, peers, None)
    }

    /// Apply one candidate-only reachability expiry. The exact missing peer is
    /// removed from dial eligibility without becoming authoritatively
    /// withdrawn; any other removal keeps the normal withdrawal semantics.
    fn accept_with_observation_expired(
        &mut self,
        revision: u64,
        peers: Vec<BrowserDesiredPeer>,
        expired_device_id: &str,
    ) -> Option<Vec<BrowserDesiredPeer>> {
        let expired_device_id = expired_device_id.trim();
        if expired_device_id.is_empty() {
            return None;
        }
        self.accept_update(revision, peers, Some(expired_device_id))
    }

    fn accept_update(
        &mut self,
        revision: u64,
        peers: Vec<BrowserDesiredPeer>,
        observation_expired_device_id: Option<&str>,
    ) -> Option<Vec<BrowserDesiredPeer>> {
        if revision <= self.revision {
            return None;
        }
        let mut peers = peers
            .into_iter()
            .filter_map(|mut peer| {
                peer.device_id = peer.device_id.trim().to_string();
                if peer.device_id.is_empty() {
                    return None;
                }
                peer.node_id = peer
                    .node_id
                    .take()
                    .map(|value| value.trim().to_string())
                    .filter(|value| !value.is_empty());
                peer.ticket = peer
                    .ticket
                    .take()
                    .map(|value| value.trim().to_string())
                    .filter(|value| !value.is_empty());
                peer.session_id = peer
                    .session_id
                    .take()
                    .map(|value| value.trim().to_string())
                    .filter(|value| !value.is_empty());
                peer.excluded_peers = peer
                    .excluded_peers
                    .into_iter()
                    .map(|value| value.trim().to_ascii_lowercase())
                    .filter(|value| !value.is_empty())
                    .collect();
                Some((peer.device_id.clone(), peer))
            })
            .collect::<std::collections::BTreeMap<_, _>>();
        let topology_revision = peers
            .values()
            .filter_map(|peer| peer.topology_revision)
            .max();
        if topology_revision == self.pending_topology_revision {
            self.pending_ready_active_peers.retain(|device_id| {
                let Some(previous) = self.peers.get(device_id) else {
                    return false;
                };
                peers.get(device_id).is_some_and(|current| {
                    current.topology_role == Some(SparseTopologyRole::Active)
                        && browser_peer_evidence(current) == browser_peer_evidence(previous)
                })
            });
        } else {
            self.pending_ready_active_peers.clear();
            if topology_revision.is_none() {
                self.settled_active.clear();
            } else if !self.settled_active.is_empty() {
                // A gateway lease overlaps only its immediately prior active
                // set. Rapid revisions can make that set speculative. Keep
                // exactly one bounded, Rust-proven settled generation instead
                // and never accumulate intermediate generations.
                peers.retain(|_, peer| peer.topology_role != Some(SparseTopologyRole::Backup));
                for (device_id, settled) in self.settled_active.iter().take(4) {
                    let still_authorized =
                        peers.get(device_id).is_some_and(|current| current.online);
                    let is_current_active = peers.get(device_id).is_some_and(|current| {
                        current.topology_role == Some(SparseTopologyRole::Active)
                    });
                    if still_authorized && !is_current_active {
                        let mut overlap = settled.clone();
                        overlap.topology_role = Some(SparseTopologyRole::Backup);
                        overlap.topology_revision = topology_revision;
                        peers.insert(device_id.clone(), overlap);
                    }
                }
            }
        }
        let active_ids = peers
            .values()
            .filter(|peer| peer.topology_role == Some(SparseTopologyRole::Active))
            .map(|peer| peer.device_id.clone())
            .collect::<std::collections::HashSet<_>>();
        if !active_ids.is_empty() && active_ids.is_subset(&self.pending_ready_active_peers) {
            peers.retain(|_, peer| peer.topology_role != Some(SparseTopologyRole::Backup));
        }
        let withdrawn = self
            .peers
            .values()
            .filter(|peer| {
                peer.online
                    && observation_expired_device_id != Some(peer.device_id.as_str())
                    && peers
                        .get(&peer.device_id)
                        .is_none_or(|current| !current.online)
            })
            .cloned()
            .collect::<Vec<_>>();
        for peer in &withdrawn {
            self.withdrawn
                .insert(peer.device_id.clone(), peer.node_id.clone());
        }
        for peer in peers.values() {
            if peer.online {
                self.withdrawn.remove(&peer.device_id);
            }
        }
        self.revision = revision;
        self.pending_topology_revision = topology_revision;
        self.peers = peers;
        Some(withdrawn)
    }

    /// Record the existing desired-peer owner's settled transport+admission
    /// result. Once every active edge for this exact topology is ready, remove
    /// overlap backups from the same authoritative mailbox and return them to
    /// the caller for physical retirement.
    fn acknowledge_topology_peer_ready(
        &mut self,
        device_id: &str,
        topology_revision: Option<u64>,
    ) -> Vec<BrowserDesiredPeer> {
        let Some(topology_revision) = topology_revision else {
            return Vec::new();
        };
        if self.pending_topology_revision != Some(topology_revision)
            || !self.peers.get(device_id).is_some_and(|peer| {
                peer.topology_revision == Some(topology_revision)
                    && peer.topology_role == Some(SparseTopologyRole::Active)
            })
        {
            return Vec::new();
        }
        self.pending_ready_active_peers
            .insert(device_id.to_string());
        let active_ids = self
            .peers
            .values()
            .filter(|peer| peer.topology_role == Some(SparseTopologyRole::Active))
            .map(|peer| peer.device_id.clone())
            .collect::<std::collections::HashSet<_>>();
        if active_ids.is_empty() || !active_ids.is_subset(&self.pending_ready_active_peers) {
            return Vec::new();
        }
        self.settled_active = self
            .peers
            .iter()
            .filter(|(_, peer)| peer.topology_role == Some(SparseTopologyRole::Active))
            .map(|(device_id, peer)| (device_id.clone(), peer.clone()))
            .take(4)
            .collect();
        let backups = self
            .peers
            .values()
            .filter(|peer| peer.topology_role == Some(SparseTopologyRole::Backup))
            .cloned()
            .collect::<Vec<_>>();
        for peer in &backups {
            self.peers.remove(&peer.device_id);
            self.withdrawn
                .insert(peer.device_id.clone(), peer.node_id.clone());
        }
        backups
    }

    fn observe_topology_peer_not_ready(&mut self, device_id: &str, topology_revision: Option<u64>) {
        if self.pending_topology_revision == topology_revision
            && self.peers.get(device_id).is_some_and(|peer| {
                peer.topology_revision == topology_revision
                    && peer.topology_role == Some(SparseTopologyRole::Active)
            })
        {
            self.pending_ready_active_peers.remove(device_id);
        }
    }

    fn device_is_withdrawn(&self, device_id: &str) -> bool {
        self.withdrawn.contains_key(device_id.trim())
    }

    fn withdrawn_device_for_node(&self, node_id: &str) -> Option<String> {
        let node_id = node_id.trim();
        if node_id.is_empty() {
            return None;
        }
        self.withdrawn
            .iter()
            .find_map(|(device_id, withdrawn_node_id)| {
                (withdrawn_node_id.as_deref() == Some(node_id)).then(|| device_id.clone())
            })
    }

    fn current_online_device_for_node(&self, node_id: &str) -> Option<String> {
        let node_id = node_id.trim();
        if node_id.is_empty() {
            return None;
        }
        self.peers.values().find_map(|peer| {
            if !peer.online {
                return None;
            }
            let ticket_node_id = peer.ticket.as_deref().and_then(|ticket| {
                let (iroh_ticket, token_suffix) = crate::session_token::split_ticket(ticket);
                let payload = token_suffix
                    .and_then(|suffix| crate::session_token::decode_payload(iroh_ticket, suffix))?;
                if payload.scope.as_str() != "user-device" {
                    return None;
                }
                parse_endpoint_ticket(iroh_ticket)
                    .ok()
                    .map(|address| address.id.to_string())
            });
            let ticket_node_id = ticket_node_id.as_deref()?;
            let resolved_node_id = match peer.node_id.as_deref() {
                Some(directory) if directory != ticket_node_id => return None,
                Some(directory) => directory,
                None => ticket_node_id,
            };
            (resolved_node_id == node_id).then(|| peer.device_id.clone())
        })
    }

    fn current_online_assignment_matches(
        &self,
        device_id: &str,
        node_id: &str,
        ticket: &str,
        evidence: &str,
    ) -> bool {
        self.peers.get(device_id.trim()).is_some_and(|peer| {
            peer.online
                && (peer.connection_intent != DesiredPeerConnectionIntent::Observe
                    || self.transfer_demands.active(&peer.device_id))
                && peer.ticket.as_deref() == Some(ticket.trim())
                && browser_peer_evidence(peer) == evidence
                && self.current_online_device_for_node(node_id).as_deref()
                    == Some(peer.device_id.as_str())
        })
    }
}

#[cfg(not(target_arch = "wasm32"))]
#[derive(Debug, Default)]
pub(crate) struct NativeExternalAutoConnectActorState {
    active: bool,
    generation: u64,
    local_device_id: String,
    mailbox: BrowserDesiredPeerMailbox,
    peer_evidence: std::collections::HashMap<String, String>,
    failure_count: std::collections::HashMap<String, u8>,
    retry_not_before_ms: std::collections::HashMap<String, i64>,
    non_initiator_not_before_ms: std::collections::HashMap<String, i64>,
    scheduled_deadline_ms: Option<i64>,
    schedule_epoch: u64,
    reconcile_running: bool,
    wake_pending: bool,
    /// Recoverable transport closes reconcile only the affected desired peer.
    /// A provider revision clears this scope and performs the normal full pass.
    recovery_device_ids: std::collections::HashSet<String>,
    /// Deduplicates close/accept/path-watcher observations for one physical
    /// loss. A new provider revision clears this fence immediately.
    recovery_suppressed_until_ms: std::collections::HashMap<String, i64>,
}

#[cfg(target_arch = "wasm32")]
struct BrowserAutoConnectActorState {
    /// Serializes demand mutations and idle retirement without borrowing the
    /// browser mailbox across an await. Roster/lifecycle callbacks stay live.
    demand_control: std::rc::Rc<tokio::sync::Mutex<()>>,
    active: bool,
    client: std::sync::Weak<Client>,
    generation: u64,
    local_device_id: String,
    mailbox: BrowserDesiredPeerMailbox,
    peer_evidence: std::collections::HashMap<String, String>,
    failure_count: std::collections::HashMap<String, u8>,
    retry_not_before_ms: std::collections::HashMap<String, i64>,
    non_initiator_not_before_ms: std::collections::HashMap<String, i64>,
    scheduled_deadline_ms: Option<i64>,
    schedule_epoch: u64,
    reconcile_running: bool,
    wake_pending: bool,
}

#[cfg(target_arch = "wasm32")]
thread_local! {
    static BROWSER_AUTO_CONNECT_ACTORS: std::cell::RefCell<
        std::collections::HashMap<usize, std::rc::Rc<std::cell::RefCell<BrowserAutoConnectActorState>>>
    > = std::cell::RefCell::new(std::collections::HashMap::new());
}

fn browser_auto_connect_failure_backoff_ms(failure_count: u8) -> i64 {
    500_i64
        .saturating_mul(1_i64 << (failure_count as u32).min(4))
        .min(8_000)
}

fn browser_peer_evidence(peer: &BrowserDesiredPeer) -> String {
    format!(
        "{}|{}|{}{}",
        peer.node_id.as_deref().unwrap_or_default(),
        peer.ticket
            .as_deref()
            .map(crate::session_token::token_fingerprint)
            .unwrap_or_default(),
        peer.session_id.as_deref().unwrap_or_default(),
        if peer.connection_intent == DesiredPeerConnectionIntent::Observe {
            "|observe"
        } else {
            ""
        },
    )
}

fn desired_peer_evidence_is_current(
    mailbox: &BrowserDesiredPeerMailbox,
    peer: &BrowserDesiredPeer,
) -> bool {
    // Mailbox revisions describe the complete desired-peer set. An unrelated
    // peer update must not cancel an in-flight connection for this peer after
    // it has already established transport or admission. The peer's own
    // evidence is the typed cancellation input.
    mailbox
        .peers
        .get(&peer.device_id)
        .is_some_and(|current| browser_peer_evidence(current) == browser_peer_evidence(peer))
}

#[cfg(target_arch = "wasm32")]
fn browser_actor_key(client: &Arc<Client>) -> usize {
    Arc::as_ptr(client) as usize
}

impl Client {
    async fn with_transfer_peer_demands<T>(
        self: &Arc<Self>,
        expected_generation: Option<u64>,
        mutate: bool,
        operation: impl FnOnce(
            &mut super::transfer_peer_demand::TransferPeerDemands,
            u64,
        ) -> anyhow::Result<T>,
    ) -> anyhow::Result<T> {
        #[cfg(not(target_arch = "wasm32"))]
        let result = {
            let mut state = self.external_desired_peer_actor.lock().await;
            if !state.active
                || self
                    .auto_connect_generation
                    .load(std::sync::atomic::Ordering::SeqCst)
                    != state.generation
                || expected_generation.is_some_and(|generation| generation != state.generation)
            {
                anyhow::bail!("execution_suspended: transfer peer demand requires its current desired-peer session");
            }
            let generation = state.generation;
            let previously_requested = state.mailbox.transfer_demands.devices_with_requests();
            let result = operation(&mut state.mailbox.transfer_demands, generation)?;
            if mutate {
                let cooldown = state.retry_not_before_ms.clone();
                let local = state.local_device_id.clone();
                let mailbox = &mut state.mailbox;
                let newly_admitted = mailbox.transfer_demands.reconcile(
                    &mailbox.peers,
                    &cooldown,
                    &local,
                    now_millis_i64(),
                    |device| self.is_auto_connect_excluded(device),
                );
                if !state.recovery_device_ids.is_empty() {
                    state.recovery_device_ids.extend(newly_admitted);
                }
            }
            let retired: Vec<_> = previously_requested
                .into_iter()
                .filter(|device| !state.mailbox.transfer_demands.has_requests_for(device))
                .collect();
            (result, generation, retired)
        };
        #[cfg(not(target_arch = "wasm32"))]
        let result = {
            let (result, generation, retired) = result;
            for device in retired {
                self.retire_unused_native_transfer_peer(&device, generation)
                    .await;
            }
            result
        };
        #[cfg(target_arch = "wasm32")]
        let result = {
            let actor = BROWSER_AUTO_CONNECT_ACTORS
                .with(|actors| actors.borrow().get(&browser_actor_key(self)).cloned())
                .ok_or_else(|| {
                    anyhow::anyhow!(
                        "execution_suspended: transfer peer demand requires a desired-peer session"
                    )
                })?;
            let control = actor.borrow().demand_control.clone();
            let _control = if mutate {
                Some(control.lock().await)
            } else {
                None
            };
            let (result, generation, retired) = {
                let mut state = actor.borrow_mut();
                if !browser_actor_is_current(&state)
                    || expected_generation.is_some_and(|generation| generation != state.generation)
                {
                    anyhow::bail!("execution_suspended: transfer peer demand belongs to a retired desired-peer session");
                }
                let generation = state.generation;
                let previously_requested = state.mailbox.transfer_demands.devices_with_requests();
                let result = operation(&mut state.mailbox.transfer_demands, generation)?;
                if mutate {
                    let cooldown = state.retry_not_before_ms.clone();
                    let local = state.local_device_id.clone();
                    let mailbox = &mut state.mailbox;
                    let _ = mailbox.transfer_demands.reconcile(
                        &mailbox.peers,
                        &cooldown,
                        &local,
                        now_millis_i64(),
                        |device| self.is_auto_connect_excluded(device),
                    );
                }
                let retired: Vec<_> = previously_requested
                    .into_iter()
                    .filter(|device| !state.mailbox.transfer_demands.has_requests_for(device))
                    .collect();
                (result, generation, retired)
            };
            for device in retired {
                self.retire_unused_browser_transfer_peer(&actor, &device, generation)
                    .await;
            }
            result
        };
        if mutate {
            #[cfg(not(target_arch = "wasm32"))]
            {
                wake_native_external_auto_connect_boxed(self.clone()).await;
            }
            #[cfg(target_arch = "wasm32")]
            {
                self.wake_browser_auto_connect();
            }
        }
        Ok(result)
    }

    pub async fn request_transfer_peer(
        self: &Arc<Self>,
        request: super::TransferPeerDemandRequest,
    ) -> anyhow::Result<super::TransferPeerDemandStatus> {
        let id = request.id.clone();
        let generation = self
            .with_transfer_peer_demands(None, true, |demands, generation| {
                demands.request(request)?;
                Ok(generation)
            })
            .await?;
        self.transfer_peer_demand_status(&id, generation)
            .await?
            .ok_or_else(|| {
                anyhow::anyhow!(
                    "execution_suspended: transfer peer demand was released during acceptance"
                )
            })
    }

    pub async fn transfer_peer_demand_status(
        self: &Arc<Self>,
        id: &str,
        generation: u64,
    ) -> anyhow::Result<Option<super::TransferPeerDemandStatus>> {
        super::TransferPeerDemandRequest::validate_identity(id)?;
        self.with_transfer_peer_demands(Some(generation), false, |demands, generation| {
            Ok(demands.status(id, generation))
        })
        .await
    }

    pub async fn release_transfer_peer(
        self: &Arc<Self>,
        id: &str,
        generation: u64,
    ) -> anyhow::Result<bool> {
        super::TransferPeerDemandRequest::validate_identity(id)?;
        self.with_transfer_peer_demands(
            Some(generation),
            true,
            |demands, _| Ok(demands.release(id)),
        )
        .await
    }

    pub async fn set_transfer_peer_budget(
        self: &Arc<Self>,
        budget: usize,
        generation: u64,
    ) -> anyhow::Result<()> {
        self.with_transfer_peer_demands(Some(generation), true, |demands, _| {
            demands.set_budget(budget)
        })
        .await
    }

    #[cfg(not(target_arch = "wasm32"))]
    async fn retire_unused_native_transfer_peer(self: &Arc<Self>, device: &str, generation: u64) {
        let records = self.resolve_peer_connection_records(device).await;
        for record in &records {
            let gate = self.managed_connect_gate(&record.connection_id).await;
            let _connect_guard = gate.lock().await;
            // Match the existing connect ordering: connect gate, then desired
            // owner. A new demand cannot be accepted during physical retirement.
            let state = self.external_desired_peer_actor.lock().await;
            if !state.active
                || state.generation != generation
                || self
                    .auto_connect_generation
                    .load(std::sync::atomic::Ordering::SeqCst)
                    != generation
                || state.mailbox.transfer_demands.has_requests_for(device)
                || !state.mailbox.peers.get(device).is_some_and(|peer| {
                    peer.connection_intent == DesiredPeerConnectionIntent::Observe
                })
            {
                return;
            }
            self.retire_unreferenced_transfer_physical_route(record, device, || {
                self.auto_connect_generation
                    .load(std::sync::atomic::Ordering::SeqCst)
                    == generation
            })
            .await;
        }
    }

    #[cfg(target_arch = "wasm32")]
    async fn retire_unused_browser_transfer_peer(
        self: &Arc<Self>,
        actor: &std::rc::Rc<std::cell::RefCell<BrowserAutoConnectActorState>>,
        device: &str,
        generation: u64,
    ) {
        let allowed = || {
            actor.try_borrow().is_ok_and(|state| {
                browser_actor_is_current(&state)
                    && state.generation == generation
                    && !state.mailbox.transfer_demands.has_requests_for(device)
                    && state.mailbox.peers.get(device).is_some_and(|peer| {
                        peer.connection_intent == DesiredPeerConnectionIntent::Observe
                    })
            })
        };
        for record in self.resolve_peer_connection_records(device).await {
            let gate = self.managed_connect_gate(&record.connection_id).await;
            let _connect_guard = gate.lock().await;
            if !allowed() {
                return;
            }
            self.retire_unreferenced_transfer_physical_route(&record, device, allowed)
                .await;
        }
    }

    async fn retire_unreferenced_transfer_physical_route(
        &self,
        record: &crate::connection_manager::ConnectionRecord,
        device: &str,
        allowed: impl Fn() -> bool,
    ) {
        let Some(_references) = self
            .connection_manager
            .guard_transfer_retirement(device)
            .await
        else {
            return;
        };
        let Some(endpoint) = record
            .endpoint_id
            .as_deref()
            .or(record.node_id.as_deref())
            .and_then(|id| id.parse::<iroh::EndpointId>().ok())
        else {
            return;
        };
        let Some(transport) = self.get_connection(endpoint).await else {
            return;
        };
        let stable_id = crate::transport_generation::for_connection(&transport);
        let node = self.iroh_node.read().await.as_ref().cloned();
        if let Some(node) = node {
            // The owner verdict is synchronous under the physical registry
            // lock, after every await. The existing close bridge owns cleanup.
            let _ = node
                .disconnect_with_reason_if_current_when(
                    endpoint,
                    stable_id,
                    crate::lifecycle_reason::REASON_TRANSFER_IDLE,
                    allowed,
                )
                .await;
        }
    }

    async fn transfer_peer_demand_is_admitted(self: &Arc<Self>, device: &str) -> bool {
        self.with_transfer_peer_demands(None, false, |demands, _| Ok(demands.active(device)))
            .await
            .unwrap_or(false)
    }
}

impl Client {
    #[cfg(not(target_arch = "wasm32"))]
    pub(crate) async fn native_external_auto_connect_owner_is_current(
        &self,
        user_id: &str,
        local_device_id: &str,
    ) -> bool {
        let expected = (
            user_id.trim().to_string(),
            local_device_id.trim().to_string(),
        );
        let state = self.external_desired_peer_actor.lock().await;
        let key = self
            .auto_connect_loop_key
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        key.as_ref() == Some(&expected)
            && state.active
            && state.local_device_id == expected.1
            && self
                .auto_connect_generation
                .load(std::sync::atomic::Ordering::SeqCst)
                == state.generation
    }

    /// Returns the explicitly withdrawn device bound to `remote_node_id`, if
    /// the active managed-session roster has retired that route.
    ///
    /// This is a read of the existing desired-peer owner, not a second
    /// admission registry.  Accepted transport handlers use it before
    /// projecting a replacement so cached peer tickets cannot resurrect a
    /// device after gateway revocation.
    pub(crate) async fn withdrawn_desired_device_for_node(
        &self,
        remote_node_id: &str,
    ) -> Option<String> {
        #[cfg(not(target_arch = "wasm32"))]
        {
            let state = self.external_desired_peer_actor.lock().await;
            if !state.active
                || self
                    .auto_connect_generation
                    .load(std::sync::atomic::Ordering::SeqCst)
                    != state.generation
            {
                return None;
            }
            state.mailbox.withdrawn_device_for_node(remote_node_id)
        }
        #[cfg(target_arch = "wasm32")]
        {
            let actor_key = self as *const Client as usize;
            BROWSER_AUTO_CONNECT_ACTORS.with(|actors| {
                let actor = actors.borrow().get(&actor_key).cloned()?;
                let state = actor.borrow();
                browser_actor_is_current(&state)
                    .then(|| state.mailbox.withdrawn_device_for_node(remote_node_id))
                    .flatten()
            })
        }
    }

    /// Fences a claimed same-user device identity after its authoritative
    /// roster entry has been withdrawn.  Room/ticket peers are unaffected
    /// unless that exact device previously belonged to this managed device
    /// roster and was explicitly retired.
    pub(crate) async fn desired_device_is_withdrawn(&self, device_id: &str) -> bool {
        #[cfg(not(target_arch = "wasm32"))]
        {
            let state = self.external_desired_peer_actor.lock().await;
            state.active
                && self
                    .auto_connect_generation
                    .load(std::sync::atomic::Ordering::SeqCst)
                    == state.generation
                && state.mailbox.device_is_withdrawn(device_id)
        }
        #[cfg(target_arch = "wasm32")]
        {
            let actor_key = self as *const Client as usize;
            BROWSER_AUTO_CONNECT_ACTORS.with(|actors| {
                actors
                    .borrow()
                    .get(&actor_key)
                    .cloned()
                    .is_some_and(|actor| {
                        let state = actor.borrow();
                        browser_actor_is_current(&state)
                            && state.mailbox.device_is_withdrawn(device_id)
                    })
            })
        }
    }

    #[cfg(target_arch = "wasm32")]
    pub(crate) fn current_desired_user_device_for_node_now(
        &self,
        remote_node_id: &str,
    ) -> Option<String> {
        let actor_key = self as *const Client as usize;
        BROWSER_AUTO_CONNECT_ACTORS.with(|actors| {
            let actor = actors.borrow().get(&actor_key).cloned()?;
            let state = actor.borrow();
            browser_actor_is_current(&state)
                .then(|| state.mailbox.current_online_device_for_node(remote_node_id))
                .flatten()
        })
    }

    #[cfg(target_arch = "wasm32")]
    pub(crate) async fn current_desired_user_device_for_node(
        &self,
        remote_node_id: &str,
    ) -> Option<String> {
        self.current_desired_user_device_for_node_now(remote_node_id)
    }

    #[cfg(not(target_arch = "wasm32"))]
    pub(super) async fn current_directory_reciprocal_credential(
        &self,
        remote_node_id: &str,
    ) -> Option<super::NativeRouteRepairCredential> {
        // Passive observed peers do not dial and therefore have no dial cache.
        // Read their reciprocal credential from the existing committed mailbox;
        // an offline, withdrawn or replaced directory entry cannot supply it.
        let state = self.external_desired_peer_actor.lock().await;
        if !state.active
            || self
                .auto_connect_generation
                .load(std::sync::atomic::Ordering::SeqCst)
                != state.generation
        {
            return None;
        }
        let device_id = state
            .mailbox
            .current_online_device_for_node(remote_node_id)?;
        let peer = state.mailbox.peers.get(&device_id)?;
        let (ticket, suffix) = crate::session_token::split_ticket(peer.ticket.as_deref()?);
        if parse_endpoint_ticket(ticket).ok()?.id.to_string() != remote_node_id {
            return None;
        }
        let suffix = suffix?;
        let payload = crate::session_token::decode_payload(ticket, suffix)?;
        if payload.scope.as_str() != "user-device"
            || payload.token.trim().is_empty()
            || payload
                .expires_at_ms
                .is_some_and(|expiry| expiry <= crate::session_token::now_unix_ms())
        {
            return None;
        }
        Some(super::NativeRouteRepairCredential {
            token: payload.token,
            token_payload: suffix.to_string(),
            scope: payload.scope,
            authoritative_device_id: device_id,
        })
    }

    #[cfg(not(target_arch = "wasm32"))]
    pub(super) async fn current_demanded_user_device_assignment_matches(
        &self,
        device_id: &str,
        remote_node_id: &str,
        ticket: &str,
        evidence: &str,
    ) -> bool {
        // Read intent and directory under the same existing owner lock. A
        // paused/released demand or retired generation cannot elect a presenter.
        let state = self.external_desired_peer_actor.lock().await;
        state.active
            && self
                .auto_connect_generation
                .load(std::sync::atomic::Ordering::SeqCst)
                == state.generation
            && state.mailbox.transfer_demands.active(device_id)
            && state.mailbox.current_online_assignment_matches(
                device_id,
                remote_node_id,
                ticket,
                evidence,
            )
    }

    pub(crate) async fn current_desired_user_device_assignment_matches(
        &self,
        device_id: &str,
        remote_node_id: &str,
        ticket: &str,
        evidence: &str,
    ) -> bool {
        #[cfg(not(target_arch = "wasm32"))]
        {
            let state = self.external_desired_peer_actor.lock().await;
            state.active
                && self
                    .auto_connect_generation
                    .load(std::sync::atomic::Ordering::SeqCst)
                    == state.generation
                && state.mailbox.current_online_assignment_matches(
                    device_id,
                    remote_node_id,
                    ticket,
                    evidence,
                )
        }
        #[cfg(target_arch = "wasm32")]
        {
            let actor_key = self as *const Client as usize;
            BROWSER_AUTO_CONNECT_ACTORS.with(|actors| {
                actors
                    .borrow()
                    .get(&actor_key)
                    .cloned()
                    .is_some_and(|actor| {
                        let state = actor.borrow();
                        browser_actor_is_current(&state)
                            && state.mailbox.current_online_assignment_matches(
                                device_id,
                                remote_node_id,
                                ticket,
                                evidence,
                            )
                    })
            })
        }
    }
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum BrowserDesiredPeerDecision {
    ObserveExisting,
    WaitUntil(i64),
    Dial,
}

fn browser_desired_peer_decision(
    local_node_id: &str,
    remote_node_id: &str,
    transport_alive: bool,
    now_ms: i64,
    non_initiator_not_before_ms: Option<i64>,
) -> BrowserDesiredPeerDecision {
    if transport_alive {
        return BrowserDesiredPeerDecision::ObserveExisting;
    }
    if local_node_id > remote_node_id {
        return BrowserDesiredPeerDecision::Dial;
    }

    let not_before = non_initiator_not_before_ms.unwrap_or_else(|| now_ms.saturating_add(3_000));
    if now_ms >= not_before {
        BrowserDesiredPeerDecision::Dial
    } else {
        BrowserDesiredPeerDecision::WaitUntil(not_before)
    }
}

fn should_replace_browser_retry_deadline(current: Option<i64>, next: Option<i64>) -> bool {
    match (current, next) {
        (None, None) => false,
        (Some(_), None) | (None, Some(_)) => true,
        (Some(current), Some(next)) => next < current,
    }
}

fn browser_reconcile_deadlines(
    outcome: BrowserPeerReconcileOutcome,
    failure_retry_deadline: Option<i64>,
) -> (Option<i64>, Option<i64>) {
    match outcome {
        BrowserPeerReconcileOutcome::Retry => (failure_retry_deadline, None),
        BrowserPeerReconcileOutcome::WakeAt(deadline) => (Some(deadline), None),
        BrowserPeerReconcileOutcome::WaitUntil(deadline) => (None, Some(deadline)),
        BrowserPeerReconcileOutcome::Stable | BrowserPeerReconcileOutcome::Ineligible => {
            (None, None)
        }
    }
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum BrowserPeerReconcileOutcome {
    Stable,
    Ineligible,
    Retry,
    WakeAt(i64),
    WaitUntil(i64),
}

#[cfg(target_arch = "wasm32")]
fn browser_actor_is_current(actor: &BrowserAutoConnectActorState) -> bool {
    actor.active
        && actor.client.upgrade().is_some_and(|client| {
            client
                .auto_connect_generation
                .load(std::sync::atomic::Ordering::SeqCst)
                == actor.generation
        })
}

#[cfg(target_arch = "wasm32")]
fn browser_attempt_is_current(
    actor: &BrowserAutoConnectActorState,
    _revision: u64,
    peer: &BrowserDesiredPeer,
) -> bool {
    browser_actor_is_current(actor) && desired_peer_evidence_is_current(&actor.mailbox, peer)
}

#[cfg(target_arch = "wasm32")]
fn deactivate_browser_actor(actor: &mut BrowserAutoConnectActorState) {
    actor.active = false;
    actor.mailbox.peers.clear();
    actor.mailbox.transfer_demands = Default::default();
    actor.peer_evidence.clear();
    actor.failure_count.clear();
    actor.retry_not_before_ms.clear();
    actor.non_initiator_not_before_ms.clear();
    actor.scheduled_deadline_ms = None;
    actor.schedule_epoch = actor.schedule_epoch.saturating_add(1);
    actor.wake_pending = false;
}

#[cfg(target_arch = "wasm32")]
fn schedule_browser_auto_connect_retry(
    actor: &std::rc::Rc<std::cell::RefCell<BrowserAutoConnectActorState>>,
) {
    let (next_deadline, generation) = {
        let state = actor.borrow();
        if !browser_actor_is_current(&state) {
            return;
        }
        let next = state
            .retry_not_before_ms
            .values()
            .chain(state.non_initiator_not_before_ms.values())
            .copied()
            .min();
        (next, state.generation)
    };

    let schedule = {
        let mut state = actor.borrow_mut();
        if !should_replace_browser_retry_deadline(state.scheduled_deadline_ms, next_deadline) {
            return;
        }
        state.schedule_epoch = state.schedule_epoch.saturating_add(1);
        state.scheduled_deadline_ms = next_deadline;
        next_deadline.map(|deadline| (deadline, state.schedule_epoch))
    };

    let Some((deadline, epoch)) = schedule else {
        return;
    };
    let actor = actor.clone();
    wasm_bindgen_futures::spawn_local(async move {
        let delay_ms = deadline.saturating_sub(now_millis_i64()).max(0) as u64;
        gloo_timers::future::sleep(std::time::Duration::from_millis(delay_ms)).await;
        let should_wake = {
            let mut state = actor.borrow_mut();
            if !state.active
                || state.generation != generation
                || state.schedule_epoch != epoch
                || !browser_actor_is_current(&state)
            {
                false
            } else {
                state.scheduled_deadline_ms = None;
                true
            }
        };
        if should_wake {
            wake_browser_auto_connect_actor(actor);
        }
    });
}

#[cfg(target_arch = "wasm32")]
fn wake_browser_auto_connect_actor(
    actor: std::rc::Rc<std::cell::RefCell<BrowserAutoConnectActorState>>,
) {
    let should_spawn = {
        let mut state = actor.borrow_mut();
        if !browser_actor_is_current(&state) {
            return;
        }
        if state.reconcile_running {
            state.wake_pending = true;
            false
        } else {
            state.reconcile_running = true;
            true
        }
    };
    if !should_spawn {
        return;
    }

    wasm_bindgen_futures::spawn_local(async move {
        loop {
            run_browser_auto_connect_pass(actor.clone()).await;
            let run_again = {
                let mut state = actor.borrow_mut();
                if !browser_actor_is_current(&state) {
                    state.reconcile_running = false;
                    false
                } else if state.wake_pending {
                    state.wake_pending = false;
                    true
                } else {
                    state.reconcile_running = false;
                    false
                }
            };
            if !run_again {
                schedule_browser_auto_connect_retry(&actor);
                break;
            }
        }
    });
}

#[cfg(target_arch = "wasm32")]
async fn reconcile_browser_desired_peer(
    actor: &std::rc::Rc<std::cell::RefCell<BrowserAutoConnectActorState>>,
    client: &Arc<Client>,
    generation: u64,
    revision: u64,
    local_device_id: &str,
    peer: &BrowserDesiredPeer,
    non_initiator_not_before_ms: Option<i64>,
) -> BrowserPeerReconcileOutcome {
    if client
        .auto_connect_generation
        .load(std::sync::atomic::Ordering::SeqCst)
        != generation
    {
        return BrowserPeerReconcileOutcome::Ineligible;
    }
    if client.is_app_execution_suspended() {
        return BrowserPeerReconcileOutcome::Ineligible;
    }

    let parsed_ticket = peer.ticket.as_deref().and_then(|ticket| {
        let (iroh_ticket, _) = crate::session_token::split_ticket(ticket);
        parse_endpoint_ticket(iroh_ticket).ok()
    });
    let parsed_node_id = parsed_ticket.as_ref().map(|address| address.id.to_string());
    let binding_node_id = peer.node_id.as_deref().or(parsed_node_id.as_deref());
    if let Some(node_id) = binding_node_id {
        client.bind_node_device_id(node_id, &peer.device_id).await;
    }

    if !actor
        .try_borrow()
        .is_ok_and(|state| browser_attempt_is_current(&state, revision, peer))
    {
        return BrowserPeerReconcileOutcome::Ineligible;
    }
    if !peer.online || peer.device_id == local_device_id {
        return BrowserPeerReconcileOutcome::Ineligible;
    }
    if remote_excludes_local_device(&peer.excluded_peers, local_device_id) {
        // `excludedPeers` is a durable auto-connect policy, not a transport
        // teardown command. A product may keep auto-connect disabled while the
        // user explicitly opens a session. Only the authenticated native
        // manual-disconnect notice (or an actual roster withdrawal) may retire
        // an already-admitted generation.
        return BrowserPeerReconcileOutcome::Ineligible;
    }
    if client.is_auto_connect_excluded(&peer.device_id) {
        return BrowserPeerReconcileOutcome::Ineligible;
    }

    let Some(ticket) = peer.ticket.as_deref() else {
        return BrowserPeerReconcileOutcome::Ineligible;
    };
    let Some(endpoint_addr) = parsed_ticket else {
        return BrowserPeerReconcileOutcome::Ineligible;
    };
    let endpoint_id = endpoint_addr.id;
    let remote_node_id = endpoint_id.to_string();
    if peer
        .node_id
        .as_deref()
        .is_some_and(|node_id| node_id != remote_node_id)
    {
        return BrowserPeerReconcileOutcome::Ineligible;
    }

    let Some(local_node_id) = client.current_node_id().await else {
        return BrowserPeerReconcileOutcome::WakeAt(now_millis_i64().saturating_add(250));
    };
    if local_node_id == remote_node_id {
        return BrowserPeerReconcileOutcome::Ineligible;
    }

    client
        .reconcile_authoritative_device_node(&peer.device_id, &remote_node_id)
        .await;
    if !actor
        .try_borrow()
        .is_ok_and(|state| browser_attempt_is_current(&state, revision, peer))
    {
        return BrowserPeerReconcileOutcome::Ineligible;
    }

    if peer.connection_intent == DesiredPeerConnectionIntent::Observe
        && !client
            .transfer_peer_demand_is_admitted(&peer.device_id)
            .await
    {
        return BrowserPeerReconcileOutcome::Ineligible;
    }

    let transport_alive = client.is_connection_transport_alive(endpoint_id).await;
    match browser_desired_peer_decision(
        &local_node_id,
        &remote_node_id,
        transport_alive,
        now_millis_i64(),
        non_initiator_not_before_ms,
    ) {
        BrowserDesiredPeerDecision::WaitUntil(deadline) => {
            return BrowserPeerReconcileOutcome::WaitUntil(deadline);
        }
        BrowserDesiredPeerDecision::ObserveExisting | BrowserDesiredPeerDecision::Dial => {}
    }

    let desired_peer_evidence = browser_peer_evidence(peer);
    let result = client
        .connect_desired_device(&peer.device_id, ticket, &desired_peer_evidence)
        .await;
    let (attempt_is_current, peer_was_withdrawn) = actor
        .try_borrow()
        .map(|state| {
            let actor_is_current = browser_actor_is_current(&state);
            (
                actor_is_current && browser_attempt_is_current(&state, revision, peer),
                superseded_desired_peer_was_withdrawn(
                    actor_is_current,
                    state.mailbox.revision,
                    revision,
                    state.mailbox.peers.contains_key(&peer.device_id),
                ),
            )
        })
        .unwrap_or((false, false));
    if !attempt_is_current {
        if peer_was_withdrawn {
            client
                .retire_withdrawn_desired_peer_connections(&peer.device_id, peer.node_id.as_deref())
                .await;
        }
        return BrowserPeerReconcileOutcome::Ineligible;
    }
    if client.is_auto_connect_excluded(&peer.device_id) {
        // This is a dialing preference, not a manual-disconnect command.
        // A concurrent explicit connect may share the completed transport.
        // Stop this automatic attempt without retiring that shared generation.
        return BrowserPeerReconcileOutcome::Ineligible;
    }

    let Ok(result) = result else {
        return BrowserPeerReconcileOutcome::Retry;
    };
    if matches!(result.state.as_str(), "failed" | "closed")
        || !client.is_connection_transport_alive(endpoint_id).await
    {
        return BrowserPeerReconcileOutcome::Retry;
    }

    let (iroh_ticket, token_suffix) = crate::session_token::split_ticket(ticket);
    if let Some(token) = token_suffix.and_then(|suffix| {
        crate::session_token::decode_payload(iroh_ticket, suffix).map(|payload| payload.token)
    }) {
        let has_proof = client
            .has_current_remote_session_admission_proof(&result.connection_id, endpoint_id, &token)
            .await;
        // An explicit product scope supersedes the provider user-device token
        // on this shared physical route. `connect_desired_device` substitutes
        // Rust's cached scoped intent and reports the host-approved scope.
        if !has_proof && result.approved_scope.as_deref().is_none_or(str::is_empty) {
            return BrowserPeerReconcileOutcome::Retry;
        }
    }

    BrowserPeerReconcileOutcome::Stable
}

#[cfg(target_arch = "wasm32")]
async fn run_browser_auto_connect_pass(
    actor: std::rc::Rc<std::cell::RefCell<BrowserAutoConnectActorState>>,
) {
    let (client, generation, revision, local_device_id, peers) = {
        let mut state = actor.borrow_mut();
        if !browser_actor_is_current(&state) {
            return;
        }
        let Some(client) = state.client.upgrade() else {
            return;
        };
        let cooldown = state.retry_not_before_ms.clone();
        let local = state.local_device_id.clone();
        let mailbox = &mut state.mailbox;
        mailbox.transfer_demands.reconcile(
            &mailbox.peers,
            &cooldown,
            &local,
            now_millis_i64(),
            |device| client.is_auto_connect_excluded(device),
        );
        (
            client,
            state.generation,
            state.mailbox.revision,
            state.local_device_id.clone(),
            state.mailbox.peers.values().cloned().collect::<Vec<_>>(),
        )
    };

    for peer in peers {
        let (retry_not_before, non_initiator_not_before) = {
            let state = actor.borrow();
            if !browser_attempt_is_current(&state, revision, &peer) {
                continue;
            }
            (
                state.retry_not_before_ms.get(&peer.device_id).copied(),
                state
                    .non_initiator_not_before_ms
                    .get(&peer.device_id)
                    .copied(),
            )
        };
        let now = now_millis_i64();
        if retry_not_before.is_some_and(|deadline| now < deadline) {
            continue;
        }

        let outcome = reconcile_browser_desired_peer(
            &actor,
            &client,
            generation,
            revision,
            &local_device_id,
            &peer,
            non_initiator_not_before,
        )
        .await;
        let backups_to_retire = {
            let mut state = actor.borrow_mut();
            if !browser_attempt_is_current(&state, revision, &peer) {
                continue;
            }
            let backups = if outcome == BrowserPeerReconcileOutcome::Stable {
                state
                    .mailbox
                    .acknowledge_topology_peer_ready(&peer.device_id, peer.topology_revision)
            } else {
                state
                    .mailbox
                    .observe_topology_peer_not_ready(&peer.device_id, peer.topology_revision);
                Vec::new()
            };
            match outcome {
                BrowserPeerReconcileOutcome::Stable | BrowserPeerReconcileOutcome::Ineligible => {
                    state.failure_count.remove(&peer.device_id);
                    state.retry_not_before_ms.remove(&peer.device_id);
                    state.non_initiator_not_before_ms.remove(&peer.device_id);
                }
                BrowserPeerReconcileOutcome::Retry => {
                    if state.mailbox.transfer_demands.active(&peer.device_id) {
                        state.mailbox.transfer_demands.yield_failed(&peer.device_id);
                        state.wake_pending = true;
                    }
                    let count = state
                        .failure_count
                        .entry(peer.device_id.clone())
                        .or_insert(0);
                    *count = count.saturating_add(1).min(10);
                    let deadline = now_millis_i64()
                        .saturating_add(browser_auto_connect_failure_backoff_ms(*count));
                    let (retry_deadline, non_initiator_deadline) =
                        browser_reconcile_deadlines(outcome, Some(deadline));
                    match retry_deadline {
                        Some(deadline) => {
                            state
                                .retry_not_before_ms
                                .insert(peer.device_id.clone(), deadline);
                        }
                        None => {
                            state.retry_not_before_ms.remove(&peer.device_id);
                        }
                    }
                    match non_initiator_deadline {
                        Some(deadline) => {
                            state
                                .non_initiator_not_before_ms
                                .insert(peer.device_id.clone(), deadline);
                        }
                        None => {
                            state.non_initiator_not_before_ms.remove(&peer.device_id);
                        }
                    }
                }
                BrowserPeerReconcileOutcome::WakeAt(deadline) => {
                    state
                        .retry_not_before_ms
                        .insert(peer.device_id.clone(), deadline);
                    state.non_initiator_not_before_ms.remove(&peer.device_id);
                }
                BrowserPeerReconcileOutcome::WaitUntil(deadline) => {
                    state.retry_not_before_ms.remove(&peer.device_id);
                    state
                        .non_initiator_not_before_ms
                        .insert(peer.device_id.clone(), deadline);
                }
            }
            backups
        };
        for backup in backups_to_retire {
            client
                .retire_withdrawn_desired_peer_connections(
                    &backup.device_id,
                    backup.node_id.as_deref(),
                )
                .await;
        }
    }
}

#[cfg(not(target_arch = "wasm32"))]
async fn native_external_attempt_is_current(
    client: &Arc<Client>,
    generation: u64,
    _revision: u64,
    peer: &BrowserDesiredPeer,
) -> bool {
    if client
        .auto_connect_generation
        .load(std::sync::atomic::Ordering::SeqCst)
        != generation
    {
        return false;
    }
    let state = client.external_desired_peer_actor.lock().await;
    state.active
        && state.generation == generation
        && desired_peer_evidence_is_current(&state.mailbox, peer)
}

#[cfg(not(target_arch = "wasm32"))]
async fn reconcile_native_external_desired_peer(
    client: &Arc<Client>,
    generation: u64,
    revision: u64,
    local_device_id: &str,
    peer: &BrowserDesiredPeer,
    non_initiator_not_before_ms: Option<i64>,
) -> BrowserPeerReconcileOutcome {
    if !native_external_attempt_is_current(client, generation, revision, peer).await
        || client.is_app_execution_suspended()
    {
        return BrowserPeerReconcileOutcome::Ineligible;
    }

    let parsed_ticket = peer.ticket.as_deref().and_then(|ticket| {
        let (iroh_ticket, _) = crate::session_token::split_ticket(ticket);
        parse_endpoint_ticket(iroh_ticket).ok()
    });
    let parsed_node_id = parsed_ticket.as_ref().map(|address| address.id.to_string());
    let binding_node_id = peer.node_id.as_deref().or(parsed_node_id.as_deref());
    if let Some(node_id) = binding_node_id {
        client.bind_node_device_id(node_id, &peer.device_id).await;
    }

    if !peer.online || peer.device_id == local_device_id {
        return BrowserPeerReconcileOutcome::Ineligible;
    }
    if remote_excludes_local_device(&peer.excluded_peers, local_device_id) {
        // Roster exclusion suppresses automatic dialing only. It cannot
        // invalidate an explicit, authenticated session that shares the same
        // physical generation.
        return BrowserPeerReconcileOutcome::Ineligible;
    }
    if client.is_auto_connect_excluded(&peer.device_id) {
        return BrowserPeerReconcileOutcome::Ineligible;
    }

    let Some(ticket) = peer.ticket.as_deref() else {
        return BrowserPeerReconcileOutcome::Ineligible;
    };
    let Some(endpoint_addr) = parsed_ticket else {
        return BrowserPeerReconcileOutcome::Ineligible;
    };
    let endpoint_id = endpoint_addr.id;
    let remote_node_id = endpoint_id.to_string();
    if peer
        .node_id
        .as_deref()
        .is_some_and(|node_id| node_id != remote_node_id)
    {
        return BrowserPeerReconcileOutcome::Ineligible;
    }

    let Some(local_node_id) = client.current_node_id().await else {
        return BrowserPeerReconcileOutcome::WakeAt(now_millis_i64().saturating_add(250));
    };
    if local_node_id == remote_node_id {
        return BrowserPeerReconcileOutcome::Ineligible;
    }

    client
        .reconcile_authoritative_device_node(&peer.device_id, &remote_node_id)
        .await;
    if !native_external_attempt_is_current(client, generation, revision, peer).await {
        return BrowserPeerReconcileOutcome::Ineligible;
    }

    if peer.connection_intent == DesiredPeerConnectionIntent::Observe
        && !client
            .transfer_peer_demand_is_admitted(&peer.device_id)
            .await
    {
        return BrowserPeerReconcileOutcome::Ineligible;
    }

    let transport_alive = client.is_connection_transport_alive(endpoint_id).await;
    match browser_desired_peer_decision(
        &local_node_id,
        &remote_node_id,
        transport_alive,
        now_millis_i64(),
        non_initiator_not_before_ms,
    ) {
        BrowserDesiredPeerDecision::WaitUntil(deadline) => {
            return BrowserPeerReconcileOutcome::WaitUntil(deadline);
        }
        BrowserDesiredPeerDecision::ObserveExisting | BrowserDesiredPeerDecision::Dial => {}
    }

    let desired_peer_evidence = browser_peer_evidence(peer);
    let result = client
        .connect_desired_device(&peer.device_id, ticket, &desired_peer_evidence)
        .await;
    let transport_stable_id = client
        .get_connection(endpoint_id)
        .await
        .map(|connection| crate::transport_generation::for_connection(&connection));
    if let Some(transport_stable_id) = transport_stable_id {
        let connection_id = Client::deterministic_connection_id(&local_node_id, &remote_node_id);
        let _ = client
            .maybe_start_offline_proof_for_transport(
                &connection_id,
                endpoint_id,
                transport_stable_id,
            )
            .await;
    }
    if !native_external_attempt_is_current(client, generation, revision, peer).await {
        let peer_was_withdrawn = {
            let state = client.external_desired_peer_actor.lock().await;
            let actor_is_current = state.active && state.generation == generation;
            superseded_desired_peer_was_withdrawn(
                actor_is_current,
                state.mailbox.revision,
                revision,
                state.mailbox.peers.contains_key(&peer.device_id),
            )
        };
        if peer_was_withdrawn {
            client
                .retire_withdrawn_desired_peer_connections(&peer.device_id, peer.node_id.as_deref())
                .await;
        }
        return BrowserPeerReconcileOutcome::Ineligible;
    }
    if client.is_auto_connect_excluded(&peer.device_id) {
        // This is a dialing preference, not a manual-disconnect command.
        // A concurrent explicit connect may share the completed transport.
        // Stop this automatic attempt without retiring that shared generation.
        return BrowserPeerReconcileOutcome::Ineligible;
    }

    let Ok(result) = result else {
        return BrowserPeerReconcileOutcome::Retry;
    };
    if matches!(result.state.as_str(), "failed" | "closed")
        || !client.is_connection_transport_alive(endpoint_id).await
    {
        return BrowserPeerReconcileOutcome::Retry;
    }

    let (iroh_ticket, token_suffix) = crate::session_token::split_ticket(ticket);
    if let Some(token) = token_suffix.and_then(|suffix| {
        crate::session_token::decode_payload(iroh_ticket, suffix).map(|payload| payload.token)
    }) {
        let has_proof = client
            .has_current_remote_session_admission_proof(&result.connection_id, endpoint_id, &token)
            .await;
        if !has_proof && result.approved_scope.as_deref().is_none_or(str::is_empty) {
            return BrowserPeerReconcileOutcome::Retry;
        }
    }

    BrowserPeerReconcileOutcome::Stable
}

#[cfg(not(target_arch = "wasm32"))]
async fn run_native_external_auto_connect_pass(client: Arc<Client>) {
    let (generation, revision, local_device_id, peers, recovery_scoped) = {
        let mut state = client.external_desired_peer_actor.lock().await;
        if !state.active
            || client
                .auto_connect_generation
                .load(std::sync::atomic::Ordering::SeqCst)
                != state.generation
        {
            return;
        }
        let cooldown = state.retry_not_before_ms.clone();
        let local = state.local_device_id.clone();
        let mailbox = &mut state.mailbox;
        let newly_admitted = mailbox.transfer_demands.reconcile(
            &mailbox.peers,
            &cooldown,
            &local,
            now_millis_i64(),
            |device| client.is_auto_connect_excluded(device),
        );
        let recovery_device_ids = std::mem::take(&mut state.recovery_device_ids);
        let recovery_scoped = !recovery_device_ids.is_empty();
        let peers = state
            .mailbox
            .peers
            .values()
            .filter(|peer| {
                recovery_device_ids.is_empty()
                    || recovery_device_ids.contains(&peer.device_id)
                    || newly_admitted.contains(&peer.device_id)
            })
            .cloned()
            .collect::<Vec<_>>();
        (
            state.generation,
            state.mailbox.revision,
            state.local_device_id.clone(),
            peers,
            recovery_scoped,
        )
    };

    for peer in peers {
        let (retry_not_before, non_initiator_not_before) = {
            let state = client.external_desired_peer_actor.lock().await;
            if !state.active
                || state.generation != generation
                || state.mailbox.revision != revision
                || !state
                    .mailbox
                    .peers
                    .get(&peer.device_id)
                    .is_some_and(|current| {
                        browser_peer_evidence(current) == browser_peer_evidence(&peer)
                    })
            {
                continue;
            }
            (
                state.retry_not_before_ms.get(&peer.device_id).copied(),
                state
                    .non_initiator_not_before_ms
                    .get(&peer.device_id)
                    .copied(),
            )
        };
        let now = now_millis_i64();
        if retry_not_before.is_some_and(|deadline| now < deadline) {
            continue;
        }

        let outcome = reconcile_native_external_desired_peer(
            &client,
            generation,
            revision,
            &local_device_id,
            &peer,
            non_initiator_not_before,
        )
        .await;
        if recovery_scoped {
            eprintln!(
                "[pluto-rtc][auto-connect][recovery-pass] device_id={} outcome={:?}",
                peer.device_id, outcome,
            );
        }
        let backups_to_retire = {
            let mut state = client.external_desired_peer_actor.lock().await;
            if !state.active
                || state.generation != generation
                || state.mailbox.revision != revision
                || !state
                    .mailbox
                    .peers
                    .get(&peer.device_id)
                    .is_some_and(|current| {
                        browser_peer_evidence(current) == browser_peer_evidence(&peer)
                    })
            {
                continue;
            }
            let backups = if outcome == BrowserPeerReconcileOutcome::Stable {
                state
                    .mailbox
                    .acknowledge_topology_peer_ready(&peer.device_id, peer.topology_revision)
            } else {
                state
                    .mailbox
                    .observe_topology_peer_not_ready(&peer.device_id, peer.topology_revision);
                Vec::new()
            };
            match outcome {
                BrowserPeerReconcileOutcome::Stable | BrowserPeerReconcileOutcome::Ineligible => {
                    state.failure_count.remove(&peer.device_id);
                    state.retry_not_before_ms.remove(&peer.device_id);
                    state.non_initiator_not_before_ms.remove(&peer.device_id);
                }
                BrowserPeerReconcileOutcome::Retry => {
                    if state.mailbox.transfer_demands.active(&peer.device_id) {
                        state.mailbox.transfer_demands.yield_failed(&peer.device_id);
                        state.wake_pending = true;
                    }
                    let count = state
                        .failure_count
                        .entry(peer.device_id.clone())
                        .or_insert(0);
                    *count = count.saturating_add(1).min(10);
                    let deadline = now_millis_i64()
                        .saturating_add(browser_auto_connect_failure_backoff_ms(*count));
                    let (retry_deadline, non_initiator_deadline) =
                        browser_reconcile_deadlines(outcome, Some(deadline));
                    match retry_deadline {
                        Some(deadline) => {
                            state
                                .retry_not_before_ms
                                .insert(peer.device_id.clone(), deadline);
                        }
                        None => {
                            state.retry_not_before_ms.remove(&peer.device_id);
                        }
                    }
                    match non_initiator_deadline {
                        Some(deadline) => {
                            state
                                .non_initiator_not_before_ms
                                .insert(peer.device_id.clone(), deadline);
                        }
                        None => {
                            state.non_initiator_not_before_ms.remove(&peer.device_id);
                        }
                    }
                    // A close-triggered pass is deliberately scoped to the
                    // affected durable device. Preserve that scope across its
                    // bounded retry: otherwise the timer below wakes an empty
                    // recovery set and re-runs every unrelated desired peer,
                    // turning one physical loss into mesh-wide transport churn.
                    if recovery_scoped {
                        state.recovery_device_ids.insert(peer.device_id.clone());
                    }
                }
                BrowserPeerReconcileOutcome::WakeAt(deadline) => {
                    state
                        .retry_not_before_ms
                        .insert(peer.device_id.clone(), deadline);
                    state.non_initiator_not_before_ms.remove(&peer.device_id);
                    if recovery_scoped {
                        state.recovery_device_ids.insert(peer.device_id.clone());
                    }
                }
                BrowserPeerReconcileOutcome::WaitUntil(deadline) => {
                    state.retry_not_before_ms.remove(&peer.device_id);
                    state
                        .non_initiator_not_before_ms
                        .insert(peer.device_id.clone(), deadline);
                    if recovery_scoped {
                        state.recovery_device_ids.insert(peer.device_id.clone());
                    }
                }
            }
            backups
        };
        for backup in backups_to_retire {
            client
                .retire_withdrawn_desired_peer_connections(
                    &backup.device_id,
                    backup.node_id.as_deref(),
                )
                .await;
        }
    }
}

#[cfg(not(target_arch = "wasm32"))]
async fn schedule_native_external_auto_connect_retry(client: Arc<Client>) {
    let schedule = {
        let mut state = client.external_desired_peer_actor.lock().await;
        if !state.active
            || client
                .auto_connect_generation
                .load(std::sync::atomic::Ordering::SeqCst)
                != state.generation
        {
            return;
        }
        let next = state
            .retry_not_before_ms
            .values()
            .chain(state.non_initiator_not_before_ms.values())
            .copied()
            .min();
        if !should_replace_browser_retry_deadline(state.scheduled_deadline_ms, next) {
            return;
        }
        state.schedule_epoch = state.schedule_epoch.saturating_add(1);
        state.scheduled_deadline_ms = next;
        next.map(|deadline| (deadline, state.schedule_epoch, state.generation))
    };
    let Some((deadline, epoch, generation)) = schedule else {
        return;
    };
    tokio::spawn(async move {
        let delay_ms = deadline.saturating_sub(now_millis_i64()).max(0) as u64;
        tokio::time::sleep(std::time::Duration::from_millis(delay_ms)).await;
        let should_wake = {
            let mut state = client.external_desired_peer_actor.lock().await;
            if !state.active
                || state.generation != generation
                || state.schedule_epoch != epoch
                || client
                    .auto_connect_generation
                    .load(std::sync::atomic::Ordering::SeqCst)
                    != generation
            {
                false
            } else {
                state.scheduled_deadline_ms = None;
                true
            }
        };
        if should_wake {
            let _ = wake_native_external_auto_connect_boxed(client).await;
        }
    });
}

#[cfg(not(target_arch = "wasm32"))]
fn wake_native_external_auto_connect_boxed(
    client: Arc<Client>,
) -> std::pin::Pin<Box<dyn std::future::Future<Output = bool> + Send>> {
    Box::pin(async move {
        let should_spawn = {
            let mut state = client.external_desired_peer_actor.lock().await;
            if !state.active
                || client
                    .auto_connect_generation
                    .load(std::sync::atomic::Ordering::SeqCst)
                    != state.generation
            {
                return false;
            }
            if state.reconcile_running {
                state.wake_pending = true;
                false
            } else {
                state.reconcile_running = true;
                true
            }
        };
        if !should_spawn {
            return true;
        }
        tokio::spawn(async move {
            loop {
                run_native_external_auto_connect_pass(client.clone()).await;
                let run_again = {
                    let mut state = client.external_desired_peer_actor.lock().await;
                    if !state.active
                        || client
                            .auto_connect_generation
                            .load(std::sync::atomic::Ordering::SeqCst)
                            != state.generation
                    {
                        state.reconcile_running = false;
                        false
                    } else if state.wake_pending {
                        state.wake_pending = false;
                        true
                    } else {
                        state.reconcile_running = false;
                        false
                    }
                };
                if !run_again {
                    schedule_native_external_auto_connect_retry(client.clone()).await;
                    break;
                }
            }
        });
        true
    })
}

#[cfg(not(target_arch = "wasm32"))]
impl Client {
    /// Test-harness diagnostic for the provider-neutral desired-peer mailbox.
    ///
    /// Tickets are never returned. Only their endpoint identity and a one-way
    /// fingerprint are exposed so a production-shaped local test can prove
    /// that a gateway replacement revision crossed the host boundary.
    #[doc(hidden)]
    pub async fn external_desired_peer_debug_snapshot(&self) -> serde_json::Value {
        let state = self.external_desired_peer_actor.lock().await;
        let mut ready_active_peers = state
            .mailbox
            .pending_ready_active_peers
            .iter()
            .cloned()
            .collect::<Vec<_>>();
        ready_active_peers.sort();
        let peers = state
            .mailbox
            .peers
            .values()
            .map(|peer| {
                let ticket_node_id = peer.ticket.as_deref().and_then(|ticket| {
                    let (iroh_ticket, _) = crate::session_token::split_ticket(ticket);
                    parse_endpoint_ticket(iroh_ticket)
                        .ok()
                        .map(|address| address.id.to_string())
                });
                serde_json::json!({
                    "deviceId": peer.device_id,
                    "nodeId": peer.node_id,
                    "ticketNodeId": ticket_node_id,
                    "ticketFingerprint": peer.ticket.as_deref()
                        .map(crate::session_token::token_fingerprint),
                    "online": peer.online,
                    "sessionId": peer.session_id,
                })
            })
            .collect::<Vec<_>>();
        serde_json::json!({
            "active": state.active,
            "generation": state.generation,
            "revision": state.mailbox.revision,
            "reconcileRunning": state.reconcile_running,
            "wakePending": state.wake_pending,
            "scheduledDeadlineMs": state.scheduled_deadline_ms,
            "lastFailureCounts": state.failure_count,
            "retryNotBeforeMs": state.retry_not_before_ms,
            "topologyRevision": state.mailbox.pending_topology_revision,
            "readyActivePeers": ready_active_peers,
            "backupPeerCount": state.mailbox.peers.values()
                .filter(|peer| peer.topology_role == Some(SparseTopologyRole::Backup))
                .count(),
            "peers": peers,
        })
    }

    pub async fn start_external_auto_connect(
        self: &Arc<Self>,
        user_id: String,
        local_device_id: String,
    ) -> anyhow::Result<()> {
        let user_id = user_id.trim().to_string();
        let local_device_id = local_device_id.trim().to_string();
        if user_id.is_empty() || local_device_id.is_empty() {
            return Err(anyhow::anyhow!(
                "external auto-connect requires non-empty user and local device ids"
            ));
        }
        let key = (user_id.clone(), local_device_id.clone());
        let duplicate_start = {
            let state = self.external_desired_peer_actor.lock().await;
            let guard = self
                .auto_connect_loop_key
                .lock()
                .unwrap_or_else(|poisoned| poisoned.into_inner());
            guard.as_ref() == Some(&key)
                && state.active
                && state.local_device_id == local_device_id
                && self
                    .auto_connect_generation
                    .load(std::sync::atomic::Ordering::SeqCst)
                    == state.generation
        };
        if duplicate_start {
            self.wake_native_external_auto_connect().await;
            return Ok(());
        }
        let generation = {
            let mut guard = self
                .auto_connect_loop_key
                .lock()
                .unwrap_or_else(|poisoned| poisoned.into_inner());
            self.known_device_ids_by_node
                .write()
                .unwrap_or_else(|poisoned| poisoned.into_inner())
                .clear();
            self.native_route_repair_credentials
                .write()
                .unwrap_or_else(|poisoned| poisoned.into_inner())
                .clear();
            self.scoped_route_repair_credentials
                .write()
                .unwrap_or_else(|poisoned| poisoned.into_inner())
                .clear();
            *guard = Some(key);
            self.auto_connect_generation
                .fetch_add(1, std::sync::atomic::Ordering::SeqCst)
                + 1
        };
        let mut state = self.external_desired_peer_actor.lock().await;
        *state = NativeExternalAutoConnectActorState {
            active: true,
            generation,
            local_device_id,
            ..NativeExternalAutoConnectActorState::default()
        };
        Ok(())
    }

    pub(crate) async fn start_external_auto_connect_with_peer_budget(
        self: &Arc<Self>,
        user_id: String,
        local_device_id: String,
        budget: usize,
    ) -> anyhow::Result<()> {
        self.start_external_auto_connect(user_id, local_device_id)
            .await?;
        let generation = self.external_desired_peer_actor.lock().await.generation;
        self.set_transfer_peer_budget(budget, generation).await
    }

    /// Additive intent-aware host ABI. Hosts using observed membership must
    /// compile against this API rather than send ignored fields to an older core.
    pub async fn submit_external_peer_intents(
        self: &Arc<Self>,
        revision: u64,
        peers_json: &str,
    ) -> anyhow::Result<bool> {
        self.submit_external_desired_peers(revision, peers_json)
            .await
    }

    pub async fn submit_external_desired_peers(
        self: &Arc<Self>,
        revision: u64,
        peers_json: &str,
    ) -> anyhow::Result<bool> {
        self.submit_external_desired_peer_update(revision, peers_json, None, false)
            .await
    }

    /// Publish a capability withdrawal only if the root actor is still running.
    /// A stopped actor has already discarded its desired peers; retiring an old
    /// identity must not restart it or fail a subsequent identity startup.
    /// Ordinary admission/startup submissions remain strict.
    pub async fn withdraw_external_desired_peers(
        self: &Arc<Self>,
        revision: u64,
        peers_json: &str,
    ) -> anyhow::Result<bool> {
        self.submit_external_desired_peer_update(revision, peers_json, None, true)
            .await
    }

    #[cfg(feature = "transport-lan")]
    pub(crate) async fn submit_external_desired_peer_observation_expired(
        self: &Arc<Self>,
        revision: u64,
        peers_json: &str,
        expired_device_id: &str,
    ) -> anyhow::Result<bool> {
        let expired_device_id = expired_device_id.trim();
        if expired_device_id.is_empty() {
            return Err(anyhow::anyhow!("expired offline device id is required"));
        }
        self.submit_external_desired_peer_update(
            revision,
            peers_json,
            Some(expired_device_id),
            false,
        )
        .await
    }

    async fn submit_external_desired_peer_update(
        self: &Arc<Self>,
        revision: u64,
        peers_json: &str,
        observation_expired_device_id: Option<&str>,
        withdrawing: bool,
    ) -> anyhow::Result<bool> {
        let peers = parse_desired_peer_intents(peers_json)?;
        if observation_expired_device_id
            .is_some_and(|expired| peers.iter().any(|peer| peer.device_id.trim() == expired))
        {
            return Err(anyhow::anyhow!(
                "expired offline device remains in desired-peer input"
            ));
        }

        let (bindings, withdrawn) = {
            let mut state = self.external_desired_peer_actor.lock().await;
            if !state.active {
                if withdrawing {
                    return Ok(false);
                }
                return Err(anyhow::anyhow!("external auto-connect is not started"));
            }
            let accepted = match observation_expired_device_id {
                Some(expired) => state
                    .mailbox
                    .accept_with_observation_expired(revision, peers, expired),
                None => state.mailbox.accept_with_withdrawn(revision, peers),
            };
            let Some(withdrawn) = accepted else {
                return Ok(false);
            };
            let desired_ids = state
                .mailbox
                .peers
                .keys()
                .cloned()
                .collect::<std::collections::HashSet<_>>();
            state.peer_evidence.retain(|id, _| desired_ids.contains(id));
            state.failure_count.retain(|id, _| desired_ids.contains(id));
            state
                .retry_not_before_ms
                .retain(|id, _| desired_ids.contains(id));
            state
                .non_initiator_not_before_ms
                .retain(|id, _| desired_ids.contains(id));

            let evidence = state
                .mailbox
                .peers
                .iter()
                .map(|(id, peer)| (id.clone(), browser_peer_evidence(peer)))
                .collect::<Vec<_>>();
            for (id, evidence) in evidence {
                let changed = state
                    .peer_evidence
                    .insert(id.clone(), evidence.clone())
                    .is_some_and(|previous| previous != evidence);
                if changed {
                    state.failure_count.remove(&id);
                    state.retry_not_before_ms.remove(&id);
                    state.non_initiator_not_before_ms.remove(&id);
                }
            }
            state.schedule_epoch = state.schedule_epoch.saturating_add(1);
            state.scheduled_deadline_ms = None;
            let bindings = state
                .mailbox
                .peers
                .values()
                .filter(|peer| peer.online)
                .filter_map(|peer| {
                    let ticket_node_id = peer.ticket.as_deref().and_then(|ticket| {
                        let (iroh_ticket, _) = crate::session_token::split_ticket(ticket);
                        parse_endpoint_ticket(iroh_ticket)
                            .ok()
                            .map(|address| address.id.to_string())
                    });
                    let node_id = match (peer.node_id.as_deref(), ticket_node_id.as_deref()) {
                        (Some(directory), Some(ticket)) if directory != ticket => return None,
                        (Some(directory), _) => directory.to_string(),
                        (None, Some(ticket)) => ticket.to_string(),
                        (None, None) => return None,
                    };
                    Some((peer.device_id.clone(), node_id))
                })
                .collect::<Vec<_>>();
            (bindings, withdrawn)
        };
        for peer in withdrawn {
            let still_withdrawn = {
                let state = self.external_desired_peer_actor.lock().await;
                state.active
                    && state.mailbox.revision == revision
                    && state
                        .mailbox
                        .peers
                        .get(&peer.device_id)
                        .is_none_or(|current| !current.online)
            };
            if still_withdrawn {
                self.retire_withdrawn_desired_peer_connections(
                    &peer.device_id,
                    peer.node_id.as_deref(),
                )
                .await;
            }
        }
        for (device_id, node_id) in bindings {
            self.observe_authoritative_device_node(&device_id, &node_id);
        }
        self.wake_native_external_auto_connect().await;
        Ok(true)
    }

    pub async fn wake_native_external_auto_connect(&self) -> bool {
        // Explicit desired-peer input is authoritative and may affect the full
        // roster. It supersedes any narrow close-recovery timer/pass.
        {
            let mut state = self.external_desired_peer_actor.lock().await;
            state.recovery_device_ids.clear();
            state.recovery_suppressed_until_ms.clear();
        }
        wake_native_external_auto_connect_boxed(Arc::new(self.clone())).await
    }

    /// Reconcile one already-installed replacement generation immediately.
    ///
    /// A close observer can schedule (and temporarily suppress duplicate)
    /// recovery before Iroh finishes selecting the successor connection. Once
    /// the incoming owner has atomically installed that successor and retired
    /// the preceding generation's native-main proofs, the old suppression no
    /// longer describes duplicate work: this is new typed lifecycle evidence
    /// that the current route itself needs admission repair. Feed that evidence
    /// to the existing desired-peer actor instead of waiting for another
    /// provider revision or starting a second retry owner.
    pub(crate) async fn wake_native_external_auto_connect_for_replaced_route(
        &self,
        remote_node_id: &str,
    ) -> bool {
        let remote_node_id = remote_node_id.trim();
        let Some(remote_device_id) = self
            .known_device_ids_by_node
            .read()
            .unwrap_or_else(|poisoned| poisoned.into_inner())
            .get(remote_node_id)
            .cloned()
        else {
            eprintln!(
                "[pluto-rtc][auto-connect][replacement-route] remote_node_id={} action=skip reason=device-binding-missing",
                remote_node_id,
            );
            return false;
        };

        {
            let mut state = self.external_desired_peer_actor.lock().await;
            if !state.active
                || self
                    .auto_connect_generation
                    .load(std::sync::atomic::Ordering::SeqCst)
                    != state.generation
                || !state.mailbox.peers.contains_key(&remote_device_id)
            {
                eprintln!(
                    "[pluto-rtc][auto-connect][replacement-route] remote_node_id={} device_id={} action=skip reason=actor-or-peer-inactive",
                    remote_node_id, remote_device_id,
                );
                return false;
            }

            state.recovery_device_ids.insert(remote_device_id.clone());
            state.recovery_suppressed_until_ms.remove(&remote_device_id);
            state.retry_not_before_ms.remove(&remote_device_id);
            state.non_initiator_not_before_ms.remove(&remote_device_id);
            state.schedule_epoch = state.schedule_epoch.saturating_add(1);
            state.scheduled_deadline_ms = None;
        }

        eprintln!(
            "[pluto-rtc][auto-connect][replacement-route] remote_node_id={} device_id={} action=wake",
            remote_node_id, remote_device_id,
        );
        wake_native_external_auto_connect_boxed(Arc::new(self.clone())).await
    }

    /// Coalesce a burst of recoverable native transport-loss notices before
    /// waking the one external desired-peer actor. A local interface can close
    /// several routes in one network transition; eagerly reconciling after
    /// each close would redial routes that the same transition is still
    /// retiring. A newer provider revision still wakes immediately and fences
    /// this scheduled notice through `schedule_epoch`.
    pub(crate) async fn schedule_native_external_auto_connect_recovery(
        &self,
        remote_node_id: &str,
    ) -> bool {
        // One physical QUIC loss produces close, accept, path-watcher, and
        // admission-stream notifications on independent tasks. Give the
        // successor transport time to win before the sole desired-peer actor
        // re-examines it; 75ms allowed those observers to redial against an
        // already-arriving replacement and created generation churn.
        const RECOVERY_COALESCE_MS: i64 = 1_000;
        // A single close is reported by the Iroh connection, the accept loop,
        // and sometimes the path watcher. Once the actor has been asked to
        // repair that loss, suppress duplicate reports long enough for the
        // tie-break winner and session admission to settle. This is a
        // per-device fence, not a second retry owner.
        const RECOVERY_DUPLICATE_SUPPRESSION_MS: i64 = MANAGED_SETTLE_DEADLINE_MS;
        let remote_node_id = remote_node_id.trim();
        let Some(remote_device_id) = self
            .known_device_ids_by_node
            .read()
            .unwrap_or_else(|poisoned| poisoned.into_inner())
            .get(remote_node_id)
            .cloned()
        else {
            // Do not fan a route-level loss into a full desired-peer scan when
            // the route has not been bound to a durable device yet. The next
            // provider revision remains the authority for that startup race.
            return false;
        };
        let schedule = {
            let mut state = self.external_desired_peer_actor.lock().await;
            if !state.active
                || self
                    .auto_connect_generation
                    .load(std::sync::atomic::Ordering::SeqCst)
                    != state.generation
            {
                return false;
            }

            let now = now_millis_i64();
            if state
                .recovery_suppressed_until_ms
                .get(&remote_device_id)
                .is_some_and(|deadline| *deadline > now)
            {
                return true;
            }
            state.recovery_device_ids.insert(remote_device_id.clone());
            state.recovery_suppressed_until_ms.insert(
                remote_device_id,
                now.saturating_add(RECOVERY_DUPLICATE_SUPPRESSION_MS),
            );

            let deadline = now.saturating_add(RECOVERY_COALESCE_MS);
            if state
                .scheduled_deadline_ms
                .is_some_and(|scheduled| scheduled <= deadline)
            {
                return true;
            }
            state.schedule_epoch = state.schedule_epoch.saturating_add(1);
            state.scheduled_deadline_ms = Some(deadline);
            (deadline, state.schedule_epoch, state.generation)
        };

        let client = Arc::new(self.clone());
        tokio::spawn(async move {
            let (deadline, epoch, generation) = schedule;
            let delay_ms = deadline.saturating_sub(now_millis_i64()).max(0) as u64;
            tokio::time::sleep(std::time::Duration::from_millis(delay_ms)).await;
            let should_wake = {
                let mut state = client.external_desired_peer_actor.lock().await;
                if !state.active
                    || state.generation != generation
                    || state.schedule_epoch != epoch
                    || client
                        .auto_connect_generation
                        .load(std::sync::atomic::Ordering::SeqCst)
                        != generation
                {
                    false
                } else {
                    state.scheduled_deadline_ms = None;
                    true
                }
            };
            if should_wake {
                let _ = wake_native_external_auto_connect_boxed(client).await;
            }
        });
        true
    }

    pub(crate) async fn external_auto_connect_is_active(&self) -> bool {
        self.external_desired_peer_actor.lock().await.active
    }

    pub async fn stop_external_auto_connect(&self) {
        let mut state = self.external_desired_peer_actor.lock().await;
        state.active = false;
        state.mailbox.peers.clear();
        state.mailbox.transfer_demands = Default::default();
        state.peer_evidence.clear();
        state.failure_count.clear();
        state.retry_not_before_ms.clear();
        state.non_initiator_not_before_ms.clear();
        state.scheduled_deadline_ms = None;
        state.schedule_epoch = state.schedule_epoch.saturating_add(1);
        state.wake_pending = false;
        state.recovery_device_ids.clear();
        state.recovery_suppressed_until_ms.clear();
    }
}

#[cfg(target_arch = "wasm32")]
impl Client {
    pub(crate) fn start_browser_auto_connect(
        self: &Arc<Self>,
        user_id: String,
        local_device_id: String,
    ) -> anyhow::Result<()> {
        let user_id = user_id.trim().to_string();
        let local_device_id = local_device_id.trim().to_string();
        if user_id.is_empty() || local_device_id.is_empty() {
            return Err(anyhow::anyhow!(
                "browser auto-connect requires non-empty user and local device ids"
            ));
        }

        let key = (user_id, local_device_id.clone());
        let actor_key = browser_actor_key(self);
        let duplicate_actor = BROWSER_AUTO_CONNECT_ACTORS.with(|actors| {
            actors.borrow().get(&actor_key).cloned().filter(|actor| {
                let state = actor.borrow();
                browser_actor_is_current(&state) && state.local_device_id == local_device_id
            })
        });
        {
            let guard = self
                .auto_connect_loop_key
                .lock()
                .unwrap_or_else(|poisoned| poisoned.into_inner());
            if guard.as_ref() == Some(&key) {
                if let Some(actor) = duplicate_actor {
                    wake_browser_auto_connect_actor(actor);
                    return Ok(());
                }
            }
        }

        let generation = {
            let mut guard = self
                .auto_connect_loop_key
                .lock()
                .unwrap_or_else(|poisoned| poisoned.into_inner());
            self.known_device_ids_by_node
                .write()
                .unwrap_or_else(|poisoned| poisoned.into_inner())
                .clear();
            *guard = Some(key);
            self.auto_connect_generation
                .fetch_add(1, std::sync::atomic::Ordering::SeqCst)
                + 1
        };

        let actor = std::rc::Rc::new(std::cell::RefCell::new(BrowserAutoConnectActorState {
            demand_control: std::rc::Rc::new(tokio::sync::Mutex::new(())),
            active: true,
            client: Arc::downgrade(self),
            generation,
            local_device_id,
            mailbox: BrowserDesiredPeerMailbox::default(),
            peer_evidence: std::collections::HashMap::new(),
            failure_count: std::collections::HashMap::new(),
            retry_not_before_ms: std::collections::HashMap::new(),
            non_initiator_not_before_ms: std::collections::HashMap::new(),
            scheduled_deadline_ms: None,
            schedule_epoch: 0,
            reconcile_running: false,
            wake_pending: false,
        }));
        BROWSER_AUTO_CONNECT_ACTORS.with(|actors| {
            if let Some(previous) = actors.borrow_mut().insert(actor_key, actor) {
                deactivate_browser_actor(&mut previous.borrow_mut());
            }
        });
        Ok(())
    }

    pub(crate) fn submit_browser_peer_intents(
        self: &Arc<Self>,
        revision: u64,
        peers_json: &str,
    ) -> anyhow::Result<bool> {
        let peers = parse_desired_peer_intents(peers_json)?;
        self.submit_browser_desired_peer_update(revision, peers)
    }

    pub(crate) fn submit_browser_desired_peers(
        self: &Arc<Self>,
        revision: u64,
        peers_json: &str,
    ) -> anyhow::Result<bool> {
        let peers: Vec<BrowserDesiredPeer> = serde_json::from_str(peers_json)
            .map_err(|error| anyhow::anyhow!("invalid browser desired-peer payload: {error}"))?;
        self.submit_browser_desired_peer_update(revision, peers)
    }

    fn submit_browser_desired_peer_update(
        self: &Arc<Self>,
        revision: u64,
        peers: Vec<BrowserDesiredPeer>,
    ) -> anyhow::Result<bool> {
        let actor = BROWSER_AUTO_CONNECT_ACTORS
            .with(|actors| actors.borrow().get(&browser_actor_key(self)).cloned());
        let Some(actor) = actor else {
            return Err(anyhow::anyhow!("browser auto-connect is not started"));
        };

        let (accepted, current_bindings, withdrawn) = {
            let mut state = actor.borrow_mut();
            if !browser_actor_is_current(&state) {
                (false, Vec::new(), Vec::new())
            } else if let Some(withdrawn) = state.mailbox.accept_with_withdrawn(revision, peers) {
                let desired_ids = state
                    .mailbox
                    .peers
                    .keys()
                    .cloned()
                    .collect::<std::collections::HashSet<_>>();
                state
                    .peer_evidence
                    .retain(|device_id, _| desired_ids.contains(device_id));
                state
                    .failure_count
                    .retain(|device_id, _| desired_ids.contains(device_id));
                state
                    .retry_not_before_ms
                    .retain(|device_id, _| desired_ids.contains(device_id));
                state
                    .non_initiator_not_before_ms
                    .retain(|device_id, _| desired_ids.contains(device_id));

                let evidence = state
                    .mailbox
                    .peers
                    .iter()
                    .map(|(device_id, peer)| (device_id.clone(), browser_peer_evidence(peer)))
                    .collect::<Vec<_>>();
                for (device_id, evidence) in evidence {
                    let changed = state
                        .peer_evidence
                        .insert(device_id.clone(), evidence.clone())
                        .is_some_and(|previous| previous != evidence);
                    if changed {
                        state.failure_count.remove(&device_id);
                        state.retry_not_before_ms.remove(&device_id);
                        state.non_initiator_not_before_ms.remove(&device_id);
                    }
                }
                state.schedule_epoch = state.schedule_epoch.saturating_add(1);
                state.scheduled_deadline_ms = None;
                let bindings = state
                    .mailbox
                    .peers
                    .values()
                    .filter(|peer| peer.online)
                    .filter_map(|peer| {
                        let ticket_node_id = peer.ticket.as_deref().and_then(|ticket| {
                            let (iroh_ticket, _) = crate::session_token::split_ticket(ticket);
                            parse_endpoint_ticket(iroh_ticket)
                                .ok()
                                .map(|address| address.id.to_string())
                        });
                        let node_id = match (peer.node_id.as_deref(), ticket_node_id.as_deref()) {
                            (Some(directory_node_id), Some(ticket_node_id))
                                if directory_node_id != ticket_node_id =>
                            {
                                return None;
                            }
                            (Some(directory_node_id), _) => directory_node_id.to_string(),
                            (None, Some(ticket_node_id)) => ticket_node_id.to_string(),
                            (None, None) => return None,
                        };
                        Some((peer.device_id.clone(), node_id))
                    })
                    .collect::<Vec<_>>();
                (true, bindings, withdrawn)
            } else {
                (false, Vec::new(), Vec::new())
            }
        };
        if accepted {
            // This is the cancellation token for any older desired-peer pass.
            // It must be visible before waking the serial actor, because an old
            // connect may currently be suspended inside Iroh or admission.
            for (device_id, node_id) in current_bindings {
                self.observe_authoritative_device_node(&device_id, &node_id);
            }
            if !withdrawn.is_empty() {
                let client = Arc::clone(self);
                let actor_for_retirement = actor.clone();
                wasm_bindgen_futures::spawn_local(async move {
                    for peer in withdrawn {
                        let still_withdrawn = {
                            let state = actor_for_retirement.borrow();
                            browser_actor_is_current(&state)
                                && state.mailbox.revision == revision
                                && state
                                    .mailbox
                                    .peers
                                    .get(&peer.device_id)
                                    .is_none_or(|current| !current.online)
                        };
                        if still_withdrawn {
                            client
                                .retire_withdrawn_desired_peer_connections(
                                    &peer.device_id,
                                    peer.node_id.as_deref(),
                                )
                                .await;
                        }
                    }
                });
            }
            wake_browser_auto_connect_actor(actor);
        }
        Ok(accepted)
    }

    pub(crate) fn wake_browser_auto_connect(self: &Arc<Self>) -> bool {
        let actor = BROWSER_AUTO_CONNECT_ACTORS
            .with(|actors| actors.borrow().get(&browser_actor_key(self)).cloned());
        let Some(actor) = actor else {
            return false;
        };
        if !browser_actor_is_current(&actor.borrow()) {
            return false;
        }
        wake_browser_auto_connect_actor(actor);
        true
    }

    pub(crate) fn stop_browser_auto_connect(self: &Arc<Self>) {
        let actor = BROWSER_AUTO_CONNECT_ACTORS
            .with(|actors| actors.borrow_mut().remove(&browser_actor_key(self)));
        if let Some(actor) = actor {
            deactivate_browser_actor(&mut actor.borrow_mut());
        }
    }
}

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

    #[cfg(not(target_arch = "wasm32"))]
    #[tokio::test]
    async fn auto_connect_disabled_during_reconcile_preserves_manual_transport() {
        let client = Arc::new(Client::new_with_app_tag(
            crate::test_constants::TEST_PROJECT_ID.to_string(),
            "exclusion-race".to_string(),
            Box::new(|| None),
        ));
        let remote = Client::new_with_app_tag(
            crate::test_constants::TEST_PROJECT_ID.to_string(),
            "exclusion-race".to_string(),
            Box::new(|| None),
        );
        for owner in [&*client, &remote] {
            let endpoint = iroh::Endpoint::builder(iroh::endpoint::presets::N0)
                .alpns(vec![crate::native_node::PlutoniumProtocol::ALPN.to_vec()])
                .relay_mode(iroh::RelayMode::Disabled)
                .bind_addr("127.0.0.1:0".parse::<std::net::SocketAddr>().unwrap())
                .unwrap()
                .bind()
                .await
                .unwrap();
            owner.adopt_endpoint(endpoint).await;
        }
        let ticket = remote.endpoint_ticket().await.unwrap();
        let remote_node_id = remote.current_node_id().await.unwrap();
        let endpoint_id = remote_node_id.parse().unwrap();
        let connected = client
            .connect_device(Some("remote"), &ticket)
            .await
            .unwrap();
        let original = client.get_connection(endpoint_id).await.unwrap();
        let original_stable_id = crate::transport_generation::for_connection(&original);
        let peer = BrowserDesiredPeer {
            connection_intent: DesiredPeerConnectionIntent::Automatic,
            device_id: "remote".to_string(),
            node_id: Some(remote_node_id),
            ticket: Some(ticket),
            online: true,
            session_id: Some("runtime-1".to_string()),
            excluded_peers: Vec::new(),
            topology_role: None,
            topology_revision: None,
        };
        let generation = client
            .auto_connect_generation
            .load(std::sync::atomic::Ordering::SeqCst);
        {
            let mut actor = client.external_desired_peer_actor.lock().await;
            actor.active = true;
            actor.generation = generation;
            actor.local_device_id = "local".to_string();
            actor
                .mailbox
                .accept_with_withdrawn(1, vec![peer.clone()])
                .unwrap();
        }
        // Pause the production actor at the same connect gate shared by a
        // manual request. Restore the durable preference while it is awaiting
        // that gate, as a relaunched WebView can do during admission.
        let gate = client.managed_connect_gate(&connected.connection_id).await;
        let guard = gate.lock().await;
        let reconcile =
            reconcile_native_external_desired_peer(&client, generation, 1, "local", &peer, None);
        tokio::pin!(reconcile);
        assert!(
            tokio::time::timeout(std::time::Duration::from_millis(100), &mut reconcile)
                .await
                .is_err()
        );
        client.set_auto_connect_excluded("remote", true);
        drop(guard);
        assert_eq!(
            tokio::time::timeout(std::time::Duration::from_secs(5), reconcile)
                .await
                .unwrap(),
            BrowserPeerReconcileOutcome::Ineligible,
        );
        assert!(client.is_auto_connect_excluded("remote"));
        assert!(
            original.close_reason().is_none(),
            "auto-connect policy must not close a manual transport"
        );
        let current = client
            .get_connection(endpoint_id)
            .await
            .expect("manual connection survives");
        assert_eq!(
            crate::transport_generation::for_connection(&current),
            original_stable_id
        );
        assert!(client.is_connection_transport_alive(endpoint_id).await);
    }

    fn replacement_snapshot(last_reconnect_reason: Option<&str>) -> ConnectionHealthView {
        ConnectionHealthView {
            connection_id: "conn-1".to_string(),
            device_id: Some("device-1".to_string()),
            device_id_hint: Some("device-1".to_string()),
            node_id: Some("node-1".to_string()),
            active_transport_stable_id: None,
            transport_generation: 2,
            route_generation: 0,
            status: ManagedHealth::AwaitingReplacement,
            settled_ready: false,
            readiness_state: ReadinessState::AwaitingReplacement,
            replacement_pending: true,
            last_lifecycle_transition_at_ms: 1_000,
            readiness_reason: "missing-active-transport".to_string(),
            transition_count: 1,
            connecting_transition_count: 0,
            replacement_count: 1,
            retire_count: 0,
            last_disconnect_reason: None,
            last_reconnect_reason: last_reconnect_reason.map(str::to_string),
        }
    }

    #[test]
    fn runtime_instance_change_is_observed_only_after_the_initial_generation() {
        let mut last_known = std::collections::HashMap::new();

        assert!(!observe_remote_runtime_instance(
            &mut last_known,
            "device-1",
            Some("runtime-1"),
        ));
        assert!(!observe_remote_runtime_instance(
            &mut last_known,
            "device-1",
            Some("runtime-1"),
        ));
        assert!(observe_remote_runtime_instance(
            &mut last_known,
            "device-1",
            Some("runtime-2"),
        ));
    }

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

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

    #[test]
    fn ticket_change_clears_transient_retry_without_retaining_ticket_material() {
        let mut last_known = std::collections::HashMap::new();
        let mut last_attempt_at = std::collections::HashMap::new();
        let mut failure_count = std::collections::HashMap::new();
        let mut failure_backoff_until = std::collections::HashMap::new();

        assert!(!reconcile_remote_ticket_retry_state(
            &mut last_known,
            "device-1",
            "ticket-1.token-1",
            &mut last_attempt_at,
            &mut failure_count,
            &mut failure_backoff_until,
        ));
        last_attempt_at.insert("device-1".to_string(), 100);
        failure_count.insert("device-1".to_string(), 4);
        failure_backoff_until.insert("device-1".to_string(), 8_100);

        assert!(!reconcile_remote_ticket_retry_state(
            &mut last_known,
            "device-1",
            "ticket-1.token-1",
            &mut last_attempt_at,
            &mut failure_count,
            &mut failure_backoff_until,
        ));
        assert_eq!(last_attempt_at.get("device-1"), Some(&100));
        assert_eq!(failure_count.get("device-1"), Some(&4));
        assert_eq!(failure_backoff_until.get("device-1"), Some(&8_100));

        assert!(reconcile_remote_ticket_retry_state(
            &mut last_known,
            "device-1",
            "ticket-1.token-2",
            &mut last_attempt_at,
            &mut failure_count,
            &mut failure_backoff_until,
        ));
        assert!(!last_attempt_at.contains_key("device-1"));
        assert!(!failure_count.contains_key("device-1"));
        assert!(!failure_backoff_until.contains_key("device-1"));

        let stored = last_known
            .get("device-1::ticket")
            .expect("ticket fingerprint");
        assert_eq!(stored.len(), 16);
        assert!(!stored.contains("token-2"));
    }

    #[test]
    fn replacement_window_suppresses_auto_connect_for_active_replacement_churn() {
        let snapshot = replacement_snapshot(Some("incoming-replacement-churn"));

        assert!(should_suppress_auto_connect_for_replacement_window(
            &snapshot,
            snapshot.last_lifecycle_transition_at_ms + 1_000,
            false,
        ));
        assert!(
            !should_suppress_auto_connect_for_replacement_window(
                &snapshot,
                snapshot.last_lifecycle_transition_at_ms + 1_000,
                true,
            ),
            "replacement suppression must not bypass positive liveness checks for an owned transport"
        );
    }

    #[test]
    fn transient_disconnect_replacement_window_allows_auto_connect_redial() {
        let snapshot = replacement_snapshot(Some("transient-disconnect"));

        assert!(!should_suppress_auto_connect_for_replacement_window(
            &snapshot,
            snapshot.last_lifecycle_transition_at_ms + 1_000,
            false,
        ));
    }

    #[test]
    fn expired_replacement_window_allows_auto_connect_redial() {
        let snapshot = replacement_snapshot(Some("incoming-replacement-churn"));

        assert!(!should_suppress_auto_connect_for_replacement_window(
            &snapshot,
            snapshot.last_lifecycle_transition_at_ms + MANAGED_SETTLE_DEADLINE_MS,
            false,
        ));
    }

    #[test]
    fn remote_exclusion_matches_canonical_device_id_case_insensitively() {
        let excluded_peers = vec!["  DEVICE-LOCAL  ".to_string()];

        assert!(remote_excludes_local_device(
            &excluded_peers,
            "device-local"
        ));
        assert!(!remote_excludes_local_device(
            &excluded_peers,
            "device-other"
        ));
        assert!(!remote_excludes_local_device(&excluded_peers, ""));
    }

    #[test]
    fn browser_desired_peer_mailbox_fences_stale_revisions_and_replaces_state() {
        let mut mailbox = BrowserDesiredPeerMailbox::default();
        let peer = BrowserDesiredPeer {
            connection_intent: DesiredPeerConnectionIntent::Automatic,
            device_id: " device-1 ".to_string(),
            node_id: Some(" node-1 ".to_string()),
            ticket: Some(" ticket-1 ".to_string()),
            online: true,
            session_id: Some(" runtime-1 ".to_string()),
            excluded_peers: vec![" DEVICE-LOCAL ".to_string()],
            topology_role: None,
            topology_revision: None,
        };

        assert!(mailbox.accept_with_withdrawn(2, vec![peer]).is_some());
        assert!(mailbox.accept_with_withdrawn(1, Vec::new()).is_none());
        assert!(mailbox.accept_with_withdrawn(2, Vec::new()).is_none());
        assert_eq!(mailbox.revision, 2);
        let stored = mailbox.peers.get("device-1").expect("desired peer");
        assert_eq!(stored.node_id.as_deref(), Some("node-1"));
        assert_eq!(stored.excluded_peers, vec!["device-local"]);

        let withdrawn = mailbox
            .accept_with_withdrawn(3, Vec::new())
            .expect("newer revision is accepted");
        assert_eq!(withdrawn.len(), 1);
        assert_eq!(withdrawn[0].device_id, "device-1");
        assert!(mailbox.peers.is_empty());
        assert!(mailbox.device_is_withdrawn("device-1"));
        assert_eq!(
            mailbox.withdrawn_device_for_node("node-1").as_deref(),
            Some("device-1")
        );

        let mut restored = withdrawn[0].clone();
        restored.online = true;
        assert!(mailbox
            .accept_with_withdrawn(4, vec![restored])
            .expect("restored revision is accepted")
            .is_empty());
        assert!(!mailbox.device_is_withdrawn("device-1"));
        assert_eq!(mailbox.withdrawn_device_for_node("node-1"), None);
    }

    #[test]
    fn current_user_device_assignment_requires_exact_online_runtime_evidence() {
        let endpoint_id = iroh::SecretKey::generate().public();
        let node_id = endpoint_id.to_string();
        let iroh_ticket =
            iroh_tickets::endpoint::EndpointTicket::new(iroh::EndpointAddr::new(endpoint_id))
                .to_string();
        let ticket =
            crate::session_token::build_ticket(&iroh_ticket, "user-device-token", "user-device", 0);
        let peer = BrowserDesiredPeer {
            connection_intent: DesiredPeerConnectionIntent::Automatic,
            device_id: "device-1".to_string(),
            node_id: Some(node_id.clone()),
            ticket: Some(ticket.clone()),
            online: true,
            session_id: Some("runtime-1".to_string()),
            excluded_peers: Vec::new(),
            topology_role: None,
            topology_revision: None,
        };
        let evidence = browser_peer_evidence(&peer);
        let mut mailbox = BrowserDesiredPeerMailbox::default();
        assert!(mailbox.accept(1, vec![peer.clone()]));
        assert_eq!(
            mailbox.current_online_device_for_node(&node_id).as_deref(),
            Some("device-1")
        );
        assert!(
            mailbox.current_online_assignment_matches("device-1", &node_id, &ticket, &evidence,)
        );
        assert!(!mailbox.current_online_assignment_matches(
            "device-1",
            &node_id,
            "stale-ticket",
            &evidence,
        ));

        let mut replacement = peer.clone();
        replacement.session_id = Some("runtime-2".to_string());
        assert!(mailbox.accept(2, vec![replacement.clone()]));
        assert!(
            !mailbox.current_online_assignment_matches("device-1", &node_id, &ticket, &evidence,)
        );
        let replacement_evidence = browser_peer_evidence(&replacement);
        assert!(mailbox.current_online_assignment_matches(
            "device-1",
            &node_id,
            &ticket,
            &replacement_evidence,
        ));

        replacement.online = false;
        assert!(mailbox.accept(3, vec![replacement]));
        assert_eq!(mailbox.current_online_device_for_node(&node_id), None);
        assert!(!mailbox.current_online_assignment_matches(
            "device-1",
            &node_id,
            &ticket,
            &replacement_evidence,
        ));
    }

    #[test]
    fn browser_desired_peer_mailbox_withdraws_live_route_when_device_goes_offline() {
        let mut mailbox = BrowserDesiredPeerMailbox::default();
        let online = BrowserDesiredPeer {
            connection_intent: DesiredPeerConnectionIntent::Automatic,
            device_id: "device-1".to_string(),
            node_id: Some("node-1".to_string()),
            ticket: Some("ticket-1".to_string()),
            online: true,
            session_id: Some("runtime-1".to_string()),
            excluded_peers: Vec::new(),
            topology_role: None,
            topology_revision: None,
        };
        assert!(mailbox
            .accept_with_withdrawn(1, vec![online.clone()])
            .expect("initial revision is accepted")
            .is_empty());

        let mut offline = online.clone();
        offline.online = false;
        let withdrawn = mailbox
            .accept_with_withdrawn(2, vec![offline])
            .expect("offline revision is accepted");
        assert_eq!(withdrawn, vec![online]);
        assert!(mailbox.peers.contains_key("device-1"));
        assert!(!mailbox.peers["device-1"].online);
        assert!(mailbox.device_is_withdrawn("device-1"));
        assert_eq!(
            mailbox.withdrawn_device_for_node("node-1").as_deref(),
            Some("device-1")
        );

        assert!(mailbox
            .accept_with_withdrawn(3, vec![mailbox.peers["device-1"].clone()])
            .expect("repeated offline revision is accepted")
            .is_empty());
    }

    #[test]
    fn offline_boot_is_open() {
        let mut mailbox = BrowserDesiredPeerMailbox::default();
        let offline = BrowserDesiredPeer {
            connection_intent: DesiredPeerConnectionIntent::Automatic,
            device_id: "device-1".to_string(),
            node_id: Some("node-1".to_string()),
            ticket: Some("ticket-1".to_string()),
            online: false,
            session_id: None,
            excluded_peers: Vec::new(),
            topology_role: None,
            topology_revision: None,
        };

        assert!(mailbox
            .accept_with_withdrawn(1, vec![offline])
            .expect("initial offline roster is accepted")
            .is_empty());
        assert!(!mailbox.device_is_withdrawn("device-1"));
        assert_eq!(mailbox.withdrawn_device_for_node("node-1"), None);
    }

    #[test]
    fn candidate_observation_expiry_removes_dial_intent_without_withdrawal() {
        let mut mailbox = BrowserDesiredPeerMailbox::default();
        let peer = BrowserDesiredPeer {
            connection_intent: DesiredPeerConnectionIntent::Automatic,
            device_id: "device-1".to_string(),
            node_id: Some("node-1".to_string()),
            ticket: Some("ticket-1".to_string()),
            online: true,
            session_id: None,
            excluded_peers: Vec::new(),
            topology_role: None,
            topology_revision: None,
        };
        assert!(mailbox
            .accept_with_withdrawn(1, vec![peer.clone()])
            .unwrap()
            .is_empty());
        assert!(mailbox
            .accept_with_observation_expired(2, Vec::new(), "device-1")
            .expect("newer observation revision")
            .is_empty());
        assert!(mailbox.peers.is_empty());
        assert!(!mailbox.device_is_withdrawn("device-1"));
        assert_eq!(mailbox.withdrawn_device_for_node("node-1"), None);
        assert!(mailbox
            .accept_with_observation_expired(2, Vec::new(), "device-1")
            .is_none());

        assert!(mailbox
            .accept_with_withdrawn(3, vec![peer])
            .expect("rediscovery revision")
            .is_empty());
        assert!(mailbox.peers.contains_key("device-1"));
        assert!(!mailbox.device_is_withdrawn("device-1"));
    }

    #[test]
    fn topology_backups_retire_only_after_current_active_edges_are_ready() {
        let peer = |device_id: &str, role: SparseTopologyRole, topology_revision: u64| {
            BrowserDesiredPeer {
                connection_intent: DesiredPeerConnectionIntent::Automatic,
                device_id: device_id.to_string(),
                node_id: Some(format!("node-{device_id}")),
                ticket: Some(format!("ticket-{device_id}")),
                online: true,
                session_id: Some(format!("runtime-{device_id}")),
                excluded_peers: Vec::new(),
                topology_role: Some(role),
                topology_revision: Some(topology_revision),
            }
        };
        let active_a = peer("active-a", SparseTopologyRole::Active, 7);
        let active_b = peer("active-b", SparseTopologyRole::Active, 7);
        let backup = peer("backup", SparseTopologyRole::Backup, 7);
        let membership_only = |mut peer: BrowserDesiredPeer| {
            peer.node_id = None;
            peer.ticket = None;
            peer.topology_role = None;
            peer.topology_revision = None;
            peer
        };
        let mut mailbox = BrowserDesiredPeerMailbox::default();
        assert!(mailbox
            .accept_with_withdrawn(10, vec![active_a.clone(), active_b.clone(), backup.clone()],)
            .expect("initial topology")
            .is_empty());

        assert!(mailbox
            .acknowledge_topology_peer_ready("active-a", Some(7))
            .is_empty());
        assert!(mailbox.peers.contains_key("backup"));
        mailbox.observe_topology_peer_not_ready("active-a", Some(7));
        assert!(mailbox
            .acknowledge_topology_peer_ready("active-b", Some(7))
            .is_empty());
        assert!(mailbox.peers.contains_key("backup"));
        assert!(mailbox
            .acknowledge_topology_peer_ready("active-b", Some(6))
            .is_empty());
        assert!(mailbox.peers.contains_key("backup"));

        assert_eq!(
            mailbox.acknowledge_topology_peer_ready("active-a", Some(7)),
            vec![backup.clone()],
        );
        assert!(!mailbox.peers.contains_key("backup"));
        assert!(mailbox.device_is_withdrawn("backup"));

        // An unrelated public-membership revision cannot revive a retired
        // backup while the gateway topology generation is unchanged.
        assert!(mailbox
            .accept_with_withdrawn(11, vec![active_a.clone(), active_b.clone(), backup.clone()],)
            .expect("same topology revision")
            .is_empty());
        assert!(!mailbox.peers.contains_key("backup"));

        // A new topology clears readiness. A late acknowledgement from the
        // prior generation cannot retire its replacement overlap.
        let next_active = peer("active-next", SparseTopologyRole::Active, 8);
        let next_backup = peer("backup-next", SparseTopologyRole::Backup, 8);
        mailbox
            .accept_with_withdrawn(
                12,
                vec![
                    next_active.clone(),
                    next_backup.clone(),
                    membership_only(active_a.clone()),
                    membership_only(active_b.clone()),
                ],
            )
            .expect("next topology revision");
        assert!(mailbox
            .acknowledge_topology_peer_ready("active-a", Some(7))
            .is_empty());
        assert!(!mailbox.peers.contains_key("backup-next"));
        assert_eq!(
            mailbox.peers["active-a"].topology_role,
            Some(SparseTopologyRole::Backup),
        );

        // A third lease arriving before revision 8 settles retains only the
        // last Rust-proven revision 7 active set. It does not accumulate the
        // speculative revision 8 active or backup set.
        let latest_active = peer("active-latest", SparseTopologyRole::Active, 9);
        let speculative_overlap = peer("active-next", SparseTopologyRole::Backup, 9);
        mailbox
            .accept_with_withdrawn(
                13,
                vec![
                    latest_active.clone(),
                    speculative_overlap,
                    membership_only(active_a.clone()),
                    membership_only(active_b.clone()),
                    membership_only(next_active.clone()),
                ],
            )
            .expect("rapid topology revision");
        assert_eq!(
            mailbox
                .peers
                .values()
                .filter(|peer| peer.topology_role == Some(SparseTopologyRole::Backup))
                .map(|peer| peer.device_id.as_str())
                .collect::<std::collections::BTreeSet<_>>(),
            std::collections::BTreeSet::from(["active-a", "active-b"]),
        );
        assert_eq!(
            mailbox.acknowledge_topology_peer_ready("active-latest", Some(9)),
            vec![active_a.clone(), active_b.clone()]
                .into_iter()
                .map(|mut peer| {
                    peer.topology_role = Some(SparseTopologyRole::Backup);
                    peer.topology_revision = Some(9);
                    peer
                })
                .collect::<Vec<_>>(),
        );

        // Restarting the actor starts with no inherited readiness proof.
        let mut restarted = BrowserDesiredPeerMailbox::default();
        restarted
            .accept_with_withdrawn(1, vec![next_active, next_backup])
            .expect("restarted topology");
        assert!(restarted.peers.contains_key("backup-next"));
    }

    #[test]
    fn browser_peer_evidence_fingerprints_ticket_instead_of_retaining_it() {
        let peer = BrowserDesiredPeer {
            connection_intent: DesiredPeerConnectionIntent::Automatic,
            device_id: "device-1".to_string(),
            node_id: Some("node-1".to_string()),
            ticket: Some("iroh-ticket.sensitive-token-material".to_string()),
            online: true,
            session_id: Some("runtime-1".to_string()),
            excluded_peers: Vec::new(),
            topology_role: None,
            topology_revision: None,
        };

        let evidence = browser_peer_evidence(&peer);
        assert!(evidence.starts_with("node-1|"));
        assert!(!evidence.contains("sensitive-token-material"));
    }

    #[test]
    fn desired_peer_attempt_survives_unrelated_mailbox_revision() {
        let mut mailbox = BrowserDesiredPeerMailbox::default();
        let tracked = BrowserDesiredPeer {
            connection_intent: DesiredPeerConnectionIntent::Automatic,
            device_id: "device-tracked".to_string(),
            node_id: Some("node-tracked".to_string()),
            ticket: Some("ticket-tracked".to_string()),
            online: true,
            session_id: Some("runtime-tracked".to_string()),
            excluded_peers: Vec::new(),
            topology_role: None,
            topology_revision: None,
        };
        let unrelated = BrowserDesiredPeer {
            connection_intent: DesiredPeerConnectionIntent::Automatic,
            device_id: "device-unrelated".to_string(),
            node_id: Some("node-unrelated".to_string()),
            ticket: Some("ticket-unrelated".to_string()),
            online: true,
            session_id: Some("runtime-unrelated".to_string()),
            excluded_peers: Vec::new(),
            topology_role: None,
            topology_revision: None,
        };

        assert!(mailbox.accept(1, vec![tracked.clone()]));
        assert!(desired_peer_evidence_is_current(&mailbox, &tracked));

        assert!(mailbox.accept(2, vec![tracked.clone(), unrelated.clone()]));
        assert!(desired_peer_evidence_is_current(&mailbox, &tracked));

        let mut changed = tracked.clone();
        changed.session_id = Some("runtime-replacement".to_string());
        assert!(mailbox.accept(3, vec![changed, unrelated.clone()]));
        assert!(!desired_peer_evidence_is_current(&mailbox, &tracked));

        assert!(mailbox.accept(4, vec![unrelated]));
        assert!(!desired_peer_evidence_is_current(&mailbox, &tracked));
    }

    #[test]
    fn browser_desired_peer_tie_break_has_one_initial_dialer_and_bounded_takeover() {
        assert_eq!(
            browser_desired_peer_decision("node-z", "node-a", false, 1_000, None),
            BrowserDesiredPeerDecision::Dial
        );
        assert_eq!(
            browser_desired_peer_decision("node-a", "node-z", false, 1_000, None),
            BrowserDesiredPeerDecision::WaitUntil(4_000)
        );
        assert_eq!(
            browser_desired_peer_decision("node-a", "node-z", false, 4_000, Some(4_000)),
            BrowserDesiredPeerDecision::Dial
        );
        assert_eq!(
            browser_desired_peer_decision("node-a", "node-z", true, 1_000, None),
            BrowserDesiredPeerDecision::ObserveExisting
        );
    }

    #[test]
    fn browser_retry_deadline_only_replaces_with_earlier_work_or_cancellation() {
        assert!(!should_replace_browser_retry_deadline(None, None));
        assert!(should_replace_browser_retry_deadline(None, Some(5_000)));
        assert!(!should_replace_browser_retry_deadline(
            Some(5_000),
            Some(6_000)
        ));
        assert!(should_replace_browser_retry_deadline(
            Some(5_000),
            Some(4_000)
        ));
        assert!(should_replace_browser_retry_deadline(Some(5_000), None));
    }

    #[test]
    fn browser_failure_backoff_is_bounded_for_recovery() {
        assert_eq!(browser_auto_connect_failure_backoff_ms(1), 1_000);
        assert_eq!(browser_auto_connect_failure_backoff_ms(2), 2_000);
        assert_eq!(browser_auto_connect_failure_backoff_ms(10), 8_000);
    }

    #[test]
    fn reconcile_deadlines_replace_expired_non_initiator_wait_with_failure_backoff() {
        let (retry_deadline, non_initiator_deadline) =
            browser_reconcile_deadlines(BrowserPeerReconcileOutcome::Retry, Some(9_000));
        assert_eq!(retry_deadline, Some(9_000));
        assert_eq!(non_initiator_deadline, None);

        let (retry_deadline, non_initiator_deadline) =
            browser_reconcile_deadlines(BrowserPeerReconcileOutcome::WaitUntil(4_000), None);
        assert_eq!(retry_deadline, None);
        assert_eq!(non_initiator_deadline, Some(4_000));
    }

    #[cfg(not(target_arch = "wasm32"))]
    #[tokio::test]
    async fn observed_fleet_directory_is_retained_without_expanding_automatic_peer_admission() {
        let client = Arc::new(Client::new_with_app_tag(
            crate::test_constants::TEST_PROJECT_ID.to_string(),
            "observed-fleet".to_string(),
            Box::new(|| None),
        ));
        client
            .start_external_auto_connect("user".into(), "local".into())
            .await
            .unwrap();
        let peers = (0..500)
            .map(|index| {
                serde_json::json!({
                    "deviceId": format!("device-{index}"), "online": index % 2 == 0,
                    "connectionIntent": "observe",
                })
            })
            .collect::<Vec<_>>();
        assert!(client
            .submit_external_desired_peers(1, &serde_json::to_string(&peers).unwrap())
            .await
            .expect("500 observed devices must be retained"));
        let snapshot = client.external_desired_peer_debug_snapshot().await;
        assert_eq!(snapshot["peers"].as_array().unwrap().len(), 500);
        assert_eq!(
            snapshot["peers"]
                .as_array()
                .unwrap()
                .iter()
                .filter(|peer| peer["online"] == true)
                .count(),
            250
        );
        let mut overflow = peers.clone();
        overflow.push(serde_json::json!({"deviceId":"overflow", "connectionIntent":"observe"}));
        assert!(client
            .submit_external_desired_peers(2, &serde_json::to_string(&overflow).unwrap())
            .await
            .is_err());
        let automatic = (0..101)
            .map(|index| serde_json::json!({"deviceId":format!("automatic-{index}")}))
            .collect::<Vec<_>>();
        assert!(client
            .submit_external_desired_peers(3, &serde_json::to_string(&automatic).unwrap())
            .await
            .is_err());
        assert_eq!(
            client.external_desired_peer_debug_snapshot().await["revision"],
            1
        );
        client.stop_external_auto_connect().await;
    }

    #[cfg(not(target_arch = "wasm32"))]
    #[tokio::test]
    async fn observed_online_peer_does_not_dial_and_preserves_an_explicit_transport() {
        let client = Arc::new(Client::new_with_app_tag(
            crate::test_constants::TEST_PROJECT_ID.to_string(),
            "observe-route".into(),
            Box::new(|| None),
        ));
        let remote = Client::new_with_app_tag(
            crate::test_constants::TEST_PROJECT_ID.to_string(),
            "observe-route".into(),
            Box::new(|| None),
        );
        for owner in [&*client, &remote] {
            let endpoint = iroh::Endpoint::builder(iroh::endpoint::presets::N0)
                .alpns(vec![crate::native_node::PlutoniumProtocol::ALPN.to_vec()])
                .relay_mode(iroh::RelayMode::Disabled)
                .bind_addr("127.0.0.1:0".parse::<std::net::SocketAddr>().unwrap())
                .unwrap()
                .bind()
                .await
                .unwrap();
            owner.adopt_endpoint(endpoint).await;
        }
        let ticket = remote.endpoint_ticket().await.unwrap();
        let node = remote.current_node_id().await.unwrap();
        let endpoint_id = node.parse().unwrap();
        client
            .start_external_auto_connect("user".into(), "local".into())
            .await
            .unwrap();
        let json = serde_json::to_string(&serde_json::json!([{
            "deviceId":"remote", "nodeId":node, "ticket":ticket, "online":true,
            "sessionId":"runtime-1", "connectionIntent":"observe",
        }]))
        .unwrap();
        assert!(client
            .submit_external_desired_peers(1, &json)
            .await
            .unwrap());
        tokio::time::timeout(std::time::Duration::from_secs(5), async {
            while client.external_desired_peer_debug_snapshot().await["reconcileRunning"] == true {
                tokio::time::sleep(std::time::Duration::from_millis(10)).await;
            }
        })
        .await
        .unwrap();
        assert!(
            client.get_connection(endpoint_id).await.is_none(),
            "observing an online routable peer must not dial it"
        );
        client
            .connect_device(Some("remote"), &ticket)
            .await
            .unwrap();
        let transport = client.get_connection(endpoint_id).await.unwrap();
        let stable_id = crate::transport_generation::for_connection(&transport);
        assert!(client
            .submit_external_desired_peers(2, &json)
            .await
            .unwrap());
        tokio::time::timeout(std::time::Duration::from_secs(5), async {
            while client.external_desired_peer_debug_snapshot().await["reconcileRunning"] == true {
                tokio::time::sleep(std::time::Duration::from_millis(10)).await;
            }
        })
        .await
        .unwrap();
        assert!(transport.close_reason().is_none());
        assert_eq!(
            crate::transport_generation::for_connection(
                &client.get_connection(endpoint_id).await.unwrap()
            ),
            stable_id
        );
        // Observation does not pin a superseded endpoint. Directory authority
        // still retires the old route, without dialing the new observed endpoint.
        let replacement = iroh::SecretKey::generate().public();
        let replacement_ticket =
            iroh_tickets::endpoint::EndpointTicket::new(iroh::EndpointAddr::new(replacement))
                .to_string();
        let replacement_json = serde_json::to_string(&serde_json::json!([{
            "deviceId":"remote", "nodeId":replacement.to_string(), "ticket":replacement_ticket,
            "online":true, "sessionId":"runtime-2", "connectionIntent":"observe",
        }]))
        .unwrap();
        assert!(client
            .submit_external_desired_peers(3, &replacement_json)
            .await
            .unwrap());
        tokio::time::timeout(std::time::Duration::from_secs(5), async {
            while client.external_desired_peer_debug_snapshot().await["reconcileRunning"] == true {
                tokio::time::sleep(std::time::Duration::from_millis(10)).await;
            }
        })
        .await
        .unwrap();
        assert!(
            transport.close_reason().is_some(),
            "a current observed endpoint must retire its superseded route"
        );
        assert!(
            client.get_connection(replacement).await.is_none(),
            "replacement observation must not dial"
        );
        client.stop_external_auto_connect().await;
    }

    #[cfg(not(target_arch = "wasm32"))]
    #[tokio::test]
    async fn transfer_peer_demand_drives_only_the_admitted_observed_peer_and_fences_restart() {
        let mut owners = Vec::new();
        for _ in 0..3 {
            let owner = Arc::new(Client::new_with_app_tag(
                crate::test_constants::TEST_PROJECT_ID.into(),
                "transfer-demand".into(),
                Box::new(|| None),
            ));
            let endpoint = iroh::Endpoint::builder(iroh::endpoint::presets::N0)
                .alpns(vec![crate::native_node::PlutoniumProtocol::ALPN.to_vec()])
                .relay_mode(iroh::RelayMode::Disabled)
                .bind_addr("127.0.0.1:0".parse::<std::net::SocketAddr>().unwrap())
                .unwrap()
                .bind()
                .await
                .unwrap();
            owner.adopt_endpoint(endpoint).await;
            owners.push(owner);
        }
        let client = &owners[0];
        client
            .start_external_auto_connect("user".into(), "local".into())
            .await
            .unwrap();
        let bootstrap = client
            .request_transfer_peer(super::super::TransferPeerDemandRequest {
                id: "offline-job".into(),
                device_id: "offline".into(),
                purpose: super::super::TransferPeerDemandPurpose::FileTransfer,
            })
            .await
            .unwrap();
        assert_eq!(
            bootstrap.state,
            super::super::TransferPeerDemandState::WaitingForPeer
        );
        client
            .set_transfer_peer_budget(1, bootstrap.generation)
            .await
            .unwrap();
        let mut peers = Vec::new();
        let mut nodes = Vec::new();
        for (index, remote) in owners.iter().enumerate().skip(1) {
            let node = remote.current_node_id().await.unwrap();
            nodes.push(node.parse::<iroh::EndpointId>().unwrap());
            peers.push(serde_json::json!({"deviceId":format!("remote-{index}"), "nodeId":node,
                "ticket":remote.endpoint_ticket().await.unwrap(), "online":true, "connectionIntent":"observe"}));
        }
        client
            .submit_external_peer_intents(1, &serde_json::to_string(&peers).unwrap())
            .await
            .unwrap();
        let first = client
            .request_transfer_peer(super::super::TransferPeerDemandRequest {
                id: "first-job".into(),
                device_id: "remote-1".into(),
                purpose: super::super::TransferPeerDemandPurpose::FileTransfer,
            })
            .await
            .unwrap();
        assert_eq!(first.state, super::super::TransferPeerDemandState::Admitted);
        let second = client
            .request_transfer_peer(super::super::TransferPeerDemandRequest {
                id: "second-job".into(),
                device_id: "remote-2".into(),
                purpose: super::super::TransferPeerDemandPurpose::DriveAccess,
            })
            .await
            .unwrap();
        assert_eq!(second.state, super::super::TransferPeerDemandState::Queued);
        tokio::time::timeout(std::time::Duration::from_secs(10), async {
            while client.get_connection(nodes[0]).await.is_none() {
                tokio::time::sleep(std::time::Duration::from_millis(10)).await;
            }
        })
        .await
        .expect("the admitted demand must reach the existing Iroh dial owner");
        assert!(
            client.get_connection(nodes[1]).await.is_none(),
            "queued work must not dial its observed peer"
        );
        let first_transport = client.get_connection(nodes[0]).await.unwrap();
        let shared = client
            .request_transfer_peer(super::super::TransferPeerDemandRequest {
                id: "same-peer-job".into(),
                device_id: "remote-1".into(),
                purpose: super::super::TransferPeerDemandPurpose::ContentSync,
            })
            .await
            .unwrap();
        client
            .release_transfer_peer(&shared.id, shared.generation)
            .await
            .unwrap();
        assert!(
            first_transport.close_reason().is_none(),
            "another job must retain the shared physical route"
        );
        let connection_id = Client::deterministic_connection_id(
            &client.current_node_id().await.unwrap(),
            &nodes[0].to_string(),
        );
        let gate = client.managed_connect_gate(&connection_id).await;
        let held = gate.lock().await;
        let releasing_client = client.clone();
        let first_id = first.id.clone();
        let releasing = tokio::spawn(async move {
            releasing_client
                .release_transfer_peer(&first_id, first.generation)
                .await
                .unwrap()
        });
        tokio::time::timeout(std::time::Duration::from_secs(5), async {
            while client
                .transfer_peer_demand_status(&first.id, first.generation)
                .await
                .unwrap()
                .is_some()
            {
                tokio::task::yield_now().await;
            }
        })
        .await
        .expect("the old reference must be removed before waiting at the connect gate");
        let requesting_client = client.clone();
        let requesting = tokio::spawn(async move {
            requesting_client
                .request_transfer_peer(super::super::TransferPeerDemandRequest {
                    id: "replacement-job".into(),
                    device_id: "remote-1".into(),
                    purpose: super::super::TransferPeerDemandPurpose::FileTransfer,
                })
                .await
                .unwrap()
        });
        tokio::time::timeout(std::time::Duration::from_secs(5), async {
            while client
                .transfer_peer_demand_status("replacement-job", first.generation)
                .await
                .unwrap()
                .is_none()
            {
                tokio::task::yield_now().await;
            }
        })
        .await
        .expect("a new reference must enter the existing owner while retirement waits");
        drop(held);
        assert!(releasing.await.unwrap());
        let replacement = requesting.await.unwrap();
        assert!(
            first_transport.close_reason().is_none(),
            "retirement must recheck a new reference after waiting at the connect gate"
        );
        client
            .release_transfer_peer(&replacement.id, replacement.generation)
            .await
            .unwrap();
        assert!(
            first_transport.close_reason().is_some(),
            "releasing the last demand must retire its unreferenced physical route"
        );
        tokio::time::timeout(std::time::Duration::from_secs(10), async {
            while client.get_connection(nodes[1]).await.is_none() {
                tokio::time::sleep(std::time::Duration::from_millis(10)).await;
            }
        })
        .await
        .expect("releasing the last demand must admit the next peer");
        let second_transport = client.get_connection(nodes[1]).await.unwrap();
        let mut automatic = peers.clone();
        automatic[1]["connectionIntent"] = serde_json::json!("automatic");
        client
            .submit_external_peer_intents(2, &serde_json::to_string(&automatic).unwrap())
            .await
            .unwrap();
        client
            .release_transfer_peer(&second.id, second.generation)
            .await
            .unwrap();
        assert!(
            second_transport.close_reason().is_none(),
            "an automatic avenue must retain its shared route"
        );
        client
            .connect_device(
                Some("remote-2"),
                &owners[2].endpoint_ticket().await.unwrap(),
            )
            .await
            .unwrap();
        client
            .submit_external_peer_intents(3, &serde_json::to_string(&peers).unwrap())
            .await
            .unwrap();
        let manual_shared = client
            .request_transfer_peer(super::super::TransferPeerDemandRequest {
                id: "manual-shared-job".into(),
                device_id: "remote-2".into(),
                purpose: super::super::TransferPeerDemandPurpose::DriveAccess,
            })
            .await
            .unwrap();
        client
            .release_transfer_peer(&manual_shared.id, manual_shared.generation)
            .await
            .unwrap();
        assert!(
            second_transport.close_reason().is_none(),
            "an explicit connection must survive final bulk release"
        );
        client.stop_external_auto_connect().await;
        client
            .start_external_auto_connect("user".into(), "local".into())
            .await
            .unwrap();
        assert!(
            client
                .release_transfer_peer(&second.id, second.generation)
                .await
                .is_err(),
            "an old job release cannot mutate a restarted same-identity owner"
        );
        client.stop_external_auto_connect().await;
    }
    #[cfg(not(target_arch = "wasm32"))]
    #[tokio::test]
    async fn duplicate_external_auto_connect_preserves_gateway_mailbox_generation() {
        let client = std::sync::Arc::new(Client::new_with_app_tag(
            crate::test_constants::TEST_PROJECT_ID.to_string(),
            "test-app".to_string(),
            Box::new(|| None),
        ));
        client
            .start_external_auto_connect("test-user".to_string(), "local-device".to_string())
            .await
            .expect("external actor starts");
        assert!(client
            .submit_external_desired_peers(7, "[]")
            .await
            .expect("gateway revision is accepted"));

        let before = client.external_desired_peer_debug_snapshot().await;
        client
            .start_external_auto_connect("test-user".to_string(), "local-device".to_string())
            .await
            .expect("duplicate start is idempotent");
        let after = client.external_desired_peer_debug_snapshot().await;

        assert_eq!(after["generation"], before["generation"]);
        assert_eq!(after["revision"], serde_json::json!(7));
        assert_eq!(after["active"], serde_json::json!(true));
    }
}

#[cfg(not(target_arch = "wasm32"))]
fn should_suppress_auto_connect_for_replacement_window(
    snapshot: &ConnectionHealthView,
    now_ms: i64,
    transport_connected: bool,
) -> bool {
    if !snapshot.replacement_pending || transport_connected {
        return false;
    }

    if now_ms.saturating_sub(snapshot.last_lifecycle_transition_at_ms) >= MANAGED_SETTLE_DEADLINE_MS
    {
        return false;
    }

    // A transient network drop clears the active transport but does not itself
    // create a successor transport. Auto-connect must redial immediately;
    // otherwise native->browser edges remain stuck in transport-only readiness.
    !matches!(
        snapshot.last_reconnect_reason.as_deref(),
        Some("transient-disconnect")
    )
}

impl Client {
    async fn retire_desired_peer_connections(
        &self,
        remote_device_id: &str,
        remote_node_id: Option<&str>,
        reason: &str,
    ) {
        let mut records = self.resolve_peer_connection_records(remote_device_id).await;
        if records.is_empty() {
            if let Some(node_id) = remote_node_id
                .map(str::trim)
                .filter(|value| !value.is_empty())
            {
                records = self.resolve_peer_connection_records(node_id).await;
            }
        }

        for record in records {
            let endpoint_id = record
                .endpoint_id
                .as_deref()
                .or(record.node_id.as_deref())
                .and_then(|value| value.trim().parse::<iroh::EndpointId>().ok());
            let disconnected = if let Some(endpoint_id) = endpoint_id {
                self.disconnect_with_reason(endpoint_id, reason)
                    .await
                    .is_ok()
            } else {
                false
            };
            if !disconnected {
                self.retire_managed_connection(&record.connection_id, Some(reason.to_string()))
                    .await;
            }
            eprintln!(
                "[pluto-rtc][auto-connect] retired undesired connection remote_device_id={} connection_id={} reason={} transport_closed={}",
                remote_device_id, record.connection_id, reason, disconnected
            );
        }
    }

    async fn retire_withdrawn_desired_peer_connections(
        &self,
        remote_device_id: &str,
        remote_node_id: Option<&str>,
    ) {
        self.retire_desired_peer_connections(
            remote_device_id,
            remote_node_id,
            crate::lifecycle_reason::REASON_DESIRED_PEER_WITHDRAWN,
        )
        .await;
    }

    #[cfg(not(target_arch = "wasm32"))]
    pub(crate) async fn retire_offline_desired_route(
        &self,
        remote_device_id: &str,
        remote_node_id: Option<&str>,
    ) {
        self.retire_withdrawn_desired_peer_connections(remote_device_id, remote_node_id)
            .await;
    }

    pub async fn is_connection_transport_alive(&self, endpoint_id: iroh::EndpointId) -> bool {
        if !self.is_connected(endpoint_id).await {
            return false;
        }

        #[cfg(not(target_arch = "wasm32"))]
        {
            let Some(connection) = self.get_connection(endpoint_id).await else {
                return false;
            };

            return connection.close_reason().is_none();
        }

        #[cfg(target_arch = "wasm32")]
        {
            let Some(connection) = self.get_connection(endpoint_id).await else {
                return false;
            };

            connection.close_reason().is_none()
        }
    }

    #[cfg(not(target_arch = "wasm32"))]
    pub(crate) async fn validate_active_connection(
        &self,
        endpoint_id: iroh::EndpointId,
    ) -> Option<crate::heartbeat::IrohConnectionProbe> {
        if !self.is_connection_transport_alive(endpoint_id).await {
            return None;
        }
        let node = self.iroh_node.read().await.as_ref().cloned()?;
        node.probe_connection(endpoint_id, std::time::Duration::from_secs(4))
            .await
    }

    #[cfg(not(target_arch = "wasm32"))]
    pub(crate) async fn should_force_network_change_after_connect_failures(
        &self,
        remote_device_id: &str,
    ) -> bool {
        for record in self.connection_manager.list_active().await {
            if !matches!(
                record.state,
                crate::connection_manager::ConnectionState::Connected
            ) {
                continue;
            }

            let Some(record_device_id) = record
                .device_id
                .as_deref()
                .or(record.device_id_hint.as_deref())
                .map(str::trim)
                .filter(|value| !value.is_empty())
            else {
                continue;
            };

            if record_device_id == remote_device_id {
                continue;
            }

            eprintln!(
                "[pluto-rtc][auto-connect] suppress network-change recovery remote_device_id={} because healthy peer remains connected peer_device_id={} connection_id={}",
                remote_device_id,
                record_device_id,
                record.connection_id
            );
            return false;
        }

        true
    }

    #[cfg(not(target_arch = "wasm32"))]
    pub(crate) async fn try_auto_connect_device(
        self: &Arc<Self>,
        user_id: &str,
        device: crate::signaling::Device,
        local_device_id: &str,
        last_attempt_at: &mut std::collections::HashMap<String, i64>,
        failure_count: &mut std::collections::HashMap<String, u8>,
        failure_backoff_until: &mut std::collections::HashMap<String, i64>,
        retry_throttle_ms: i64,
        non_initiator_wait_started_at: &mut std::collections::HashMap<String, i64>,
        non_initiator_escalation_count: &mut std::collections::HashMap<String, u8>,
        initiator_grace_ms: i64,
        skip_log_at: &mut std::collections::HashMap<String, i64>,
        connected_health_checked_at: &mut std::collections::HashMap<String, i64>,
        connected_health_state: &mut std::collections::HashMap<String, bool>,
        connected_health_failures: &mut std::collections::HashMap<String, u8>,
        connected_health_probe_ms: i64,
        connected_health_failure_threshold: u8,
        last_presence_republish_at_ms: &mut i64,
        presence_republish_interval_ms: i64,
        last_network_change_at_ms: &mut i64,
        network_change_recovery_interval_ms: i64,
        network_change_failure_threshold: u8,
        last_known_node_id: &mut std::collections::HashMap<String, String>,
    ) {
        if self.is_app_execution_suspended() {
            return;
        }

        #[cfg(target_arch = "wasm32")]
        let _ = (
            user_id,
            last_presence_republish_at_ms,
            presence_republish_interval_ms,
            last_network_change_at_ms,
            network_change_recovery_interval_ms,
            network_change_failure_threshold,
        );

        let remote_device_id = device.device_id.clone();
        let now = now_millis_i64();
        let skip_log_interval_ms = 2_000;

        let mut log_skip = |reason: &str| {
            let is_noise_reason = matches!(
                reason,
                "already-connected"
                    | "tie-break-not-initiator-waiting"
                    | "retry-throttle"
                    | "self-ticket"
                    | "auto-connect-excluded"
                    | "remote-excluded-local-device"
            );
            if is_noise_reason && !auto_connect_verbose() {
                return;
            }

            let skip_key = format!("{}::{}", remote_device_id, reason);
            let should_log = match skip_log_at.get(&skip_key) {
                Some(last_at) => now.saturating_sub(*last_at) >= skip_log_interval_ms,
                None => true,
            };
            if !should_log {
                return;
            }
            skip_log_at.insert(skip_key, now);

            #[cfg(not(target_arch = "wasm32"))]
            eprintln!(
                "[pluto-rtc][auto-connect] skip remote_device_id={} local_device_id={} reason={}",
                remote_device_id, local_device_id, reason
            );
            #[cfg(target_arch = "wasm32")]
            web_sys::console::log_1(&wasm_bindgen::JsValue::from_str(&format!(
                "[pluto-rtc][auto-connect] skip remote_device_id={} local_device_id={} reason={}",
                remote_device_id, local_device_id, reason
            )));
        };

        if !device.online {
            last_attempt_at.remove(&remote_device_id);
            non_initiator_wait_started_at.remove(&remote_device_id);
            non_initiator_escalation_count.remove(&remote_device_id);
            connected_health_checked_at.remove(&remote_device_id);
            connected_health_state.remove(&remote_device_id);
            connected_health_failures.remove(&remote_device_id);
            clear_auto_connect_failure_state(
                &remote_device_id,
                failure_count,
                failure_backoff_until,
            );
            return;
        }

        if remote_device_id == local_device_id {
            last_attempt_at.remove(&remote_device_id);
            non_initiator_wait_started_at.remove(&remote_device_id);
            non_initiator_escalation_count.remove(&remote_device_id);
            connected_health_checked_at.remove(&remote_device_id);
            connected_health_state.remove(&remote_device_id);
            connected_health_failures.remove(&remote_device_id);
            clear_auto_connect_failure_state(
                &remote_device_id,
                failure_count,
                failure_backoff_until,
            );
            return;
        }

        if remote_excludes_local_device(&device.excluded_peers, local_device_id) {
            log_skip("remote-excluded-local-device");
            return;
        }

        if self.is_auto_connect_peer_excluded(&remote_device_id, device.node_id.as_deref()) {
            log_skip("auto-connect-excluded");
            return;
        }

        let provider_ticket = match device.ticket.as_deref() {
            Some(ticket) if !ticket.trim().is_empty() => ticket,
            _ => {
                log_skip("missing-ticket");
                return;
            }
        };

        let (provider_iroh_ticket, _) = crate::session_token::split_ticket(provider_ticket.trim());
        let provider_endpoint_addr = match parse_endpoint_ticket(provider_iroh_ticket) {
            Ok(addr) => addr,
            Err(error) => {
                log_skip("invalid-ticket");
                eprintln!(
                    "[pluto-rtc][auto-connect] invalid ticket remote_device_id={} error={}",
                    remote_device_id, error
                );
                return;
            }
        };
        let preferred_ticket = self.preferred_scoped_route_ticket(
            provider_endpoint_addr.id.to_string().as_str(),
            provider_iroh_ticket,
        );
        let ticket_str = preferred_ticket.as_deref().unwrap_or(provider_ticket);

        // Strip compound ticket suffix for iroh parsing.  The token is
        // extracted and will be auto-presented to the host after connection.
        let (iroh_ticket, token_suffix) = crate::session_token::split_ticket(ticket_str.trim());
        let extracted_token_payload = token_suffix
            .and_then(|suffix| crate::session_token::decode_payload(iroh_ticket, suffix));
        let extracted_token = extracted_token_payload
            .as_ref()
            .map(|payload| payload.token.clone());
        let extracted_token_suffix = token_suffix
            .filter(|_| extracted_token_payload.is_some())
            .map(ToOwned::to_owned);

        let endpoint_addr = provider_endpoint_addr;
        #[cfg(feature = "transport-lan")]
        if self.transport_config.read().await.network_policy
            == crate::client::NetworkPolicy::LocalOnly
            && !crate::local_discovery::endpoint_addr_is_local_only(&endpoint_addr, &[])
        {
            log_skip("local-only-address-rejected");
            return;
        }
        let endpoint_id = endpoint_addr.id;
        let node_id_str = endpoint_id.to_string();
        #[cfg(not(target_arch = "wasm32"))]
        if let Some(payload) = extracted_token_payload
            .as_ref()
            .filter(|payload| payload.scope.as_str() == "user-device")
        {
            self.update_native_route_repair_credential(
                &node_id_str,
                Some(&remote_device_id),
                extracted_token.as_deref(),
                extracted_token_suffix.as_deref(),
                &payload.scope,
            );
        }
        let runtime_instance_changed = observe_remote_runtime_instance(
            last_known_node_id,
            &remote_device_id,
            device.session_id.as_deref(),
        );
        let ticket_changed = reconcile_remote_ticket_retry_state(
            last_known_node_id,
            &remote_device_id,
            provider_ticket,
            last_attempt_at,
            failure_count,
            failure_backoff_until,
        );

        if ticket_changed {
            eprintln!(
                "[pluto-rtc][auto-connect] peer ticket changed remote_device_id={} node_id={} - clearing stale retry deadline",
                remote_device_id, node_id_str
            );
        }

        // Detect peer identity change (e.g. web refresh → new node_id).
        // When detected, clear all retry/health/backoff state so reconnection
        // proceeds immediately without being blocked by stale throttles.
        {
            let identity_changed = last_known_node_id
                .get(&remote_device_id)
                .map(|prev| prev.as_str() != node_id_str)
                .unwrap_or(false);
            last_known_node_id.insert(remote_device_id.clone(), node_id_str.clone());
            if identity_changed {
                #[cfg(not(target_arch = "wasm32"))]
                eprintln!(
                    "[pluto-rtc][auto-connect] peer identity changed remote_device_id={} new_node_id={} — resetting state for fast reconnect",
                    remote_device_id, node_id_str
                );
                #[cfg(target_arch = "wasm32")]
                web_sys::console::log_1(&wasm_bindgen::JsValue::from_str(&format!(
                    "[pluto-rtc][auto-connect] peer identity changed remote_device_id={} new_node_id={} — resetting state for fast reconnect",
                    remote_device_id, node_id_str
                )));
                last_attempt_at.remove(&remote_device_id);
                non_initiator_wait_started_at.remove(&remote_device_id);
                non_initiator_escalation_count.remove(&remote_device_id);
                connected_health_checked_at.remove(&remote_device_id);
                connected_health_state.remove(&remote_device_id);
                connected_health_failures.remove(&remote_device_id);
                clear_auto_connect_failure_state(
                    &remote_device_id,
                    failure_count,
                    failure_backoff_until,
                );
            }
        }

        let local_node_id = match self.current_node_id().await {
            Some(id) => id,
            None => {
                log_skip("local-node-not-ready");
                return;
            }
        };

        if local_node_id == node_id_str {
            log_skip("self-ticket");
            return;
        }

        if runtime_instance_changed {
            // Presence is coordination evidence, not physical-lifecycle
            // authority. A newly launched peer can accept a replacement route
            // before its new gateway lease wins aggregation. Retiring that
            // already-responsive route when the delayed instance arrives
            // creates needless generation churn. Probe the exact current
            // generation and retire it only when the peer no longer responds.
            let current_transport = self.get_connection(endpoint_id).await;
            let current_transport_stable_id = current_transport
                .as_ref()
                .map(crate::transport_generation::for_connection);
            #[cfg(openrtc_iroh_retire_remote_connections_api)]
            if let Some(current_transport) = current_transport.as_ref() {
                let transport_stable_id =
                    crate::transport_generation::for_connection(current_transport);
                // One Iroh endpoint can carry OpenRTC control traffic and
                // independent application protocols. A durable node identity
                // survives process restart, so those other protocol
                // connections can otherwise remain attached to the retired
                // process generation even after the new OpenRTC connection is
                // healthy. Runtime-instance replacement is the typed lifecycle
                // evidence: fence every other connection for this peer while
                // preserving the exact admitted OpenRTC generation.
                let endpoint = { self.iroh_endpoint.read().await.clone() };
                if let Some(endpoint) = endpoint {
                    let retired = retire_stale_endpoint_connections(
                        &endpoint,
                        endpoint_id,
                        current_transport,
                        crate::lifecycle_reason::REASON_STALE_ACTIVE_CONNECTION_RECONNECT
                            .as_bytes(),
                    )
                    .await;
                    if retired > 0 {
                        eprintln!(
                            "[pluto-rtc][auto-connect] peer runtime instance changed remote_device_id={} node_id={} - retired {} stale shared-endpoint protocol connection(s), preserved transport_stable_id={}",
                            remote_device_id, node_id_str, retired, transport_stable_id
                        );
                    }
                }
            }
            let active_transport_responsive = if current_transport_stable_id.is_some() {
                self.validate_active_connection(endpoint_id)
                    .await
                    .is_some_and(|probe| probe.responsive)
            } else {
                false
            };
            if runtime_instance_change_requires_replacement(
                runtime_instance_changed,
                active_transport_responsive,
            ) {
                if let Some(transport_stable_id) = current_transport_stable_id {
                    match self
                        .retire_transport_generation_after_failed_probe(
                            &Self::deterministic_connection_id(&local_node_id, &node_id_str),
                            endpoint_id,
                            transport_stable_id,
                            connected_health_probe_ms,
                            crate::lifecycle_reason::REASON_STALE_ACTIVE_CONNECTION_RECONNECT,
                        )
                        .await
                    {
                        FailedProbeRetirementOutcome::Retired => {
                            eprintln!(
                                "[pluto-rtc][auto-connect] peer runtime instance changed remote_device_id={} node_id={} - replaced unresponsive physical transport",
                                remote_device_id, node_id_str
                            );
                        }
                        FailedProbeRetirementOutcome::DeferredAdmissionInFlight => {
                            eprintln!(
                                "[pluto-rtc][auto-connect] peer runtime instance changed remote_device_id={} node_id={} - deferred replacement while admission is in flight",
                                remote_device_id, node_id_str
                            );
                        }
                        FailedProbeRetirementOutcome::CancelledCurrentGenerationActivity => {
                            eprintln!(
                                "[pluto-rtc][auto-connect] peer runtime instance changed remote_device_id={} node_id={} - preserved generation with recent protocol activity",
                                remote_device_id, node_id_str
                            );
                        }
                        FailedProbeRetirementOutcome::GenerationChanged => {
                            eprintln!(
                                "[pluto-rtc][auto-connect] peer runtime instance changed remote_device_id={} node_id={} - ignored stale generation",
                                remote_device_id, node_id_str
                            );
                        }
                    }
                }
            } else {
                eprintln!(
                    "[pluto-rtc][auto-connect] peer runtime instance changed remote_device_id={} node_id={} - preserving responsive physical transport",
                    remote_device_id, node_id_str
                );
            }
            last_attempt_at.remove(&remote_device_id);
            non_initiator_wait_started_at.remove(&remote_device_id);
            non_initiator_escalation_count.remove(&remote_device_id);
            connected_health_checked_at.remove(&remote_device_id);
            connected_health_state.remove(&remote_device_id);
            connected_health_failures.remove(&remote_device_id);
            clear_auto_connect_failure_state(
                &remote_device_id,
                failure_count,
                failure_backoff_until,
            );
        }

        if let Some(discovered_node_id) = device.node_id.as_deref() {
            if discovered_node_id != node_id_str {
                log_skip("ticket-node-mismatch-discovery");
                eprintln!(
                    "[pluto-rtc][auto-connect] rejected inconsistent discovery identity remote_device_id={} discovered_node_id={} ticket_node_id={}",
                    remote_device_id, discovered_node_id, node_id_str
                );
                return;
            }
        }

        // The discovery ticket/node pair is the current coordination authority
        // for this durable device. A transport to an older node can remain open
        // until QUIC idle timeout after a browser/app restart; it must not keep
        // owning the logical device session or receive application payloads.
        // Run this even when the loop did not observe the previous node itself
        // (for example, subscription replacement between snapshots).
        self.reconcile_authoritative_device_node(&remote_device_id, &node_id_str)
            .await;

        #[cfg(feature = "transport-lan")]
        if self.is_peer_locally_reachable(&node_id_str).await {
            eprintln!(
                "[pluto-rtc][auto-connect] LAN peer reachable remote_device_id={} node_id={}",
                remote_device_id, node_id_str
            );
        }

        let connection_id = Self::deterministic_connection_id(&local_node_id, &node_id_str);

        let managed_health_snapshot = self.managed_connection_health(&connection_id).await;

        #[cfg(not(target_arch = "wasm32"))]
        {
            if let Some(existing) = self
                .connection_manager
                .get_by_connection_id(&connection_id)
                .await
            {
                let existing_device_id = existing.device_id.as_deref().map(str::trim);
                if existing_device_id != Some(remote_device_id.as_str()) {
                    self.connection_manager
                        .set_device_id(&connection_id, remote_device_id.clone())
                        .await;
                    eprintln!(
                        "[pluto-rtc][auto-connect] backfilled device_id hint connection_id={} node_id={} device_id={}",
                        connection_id, node_id_str, remote_device_id
                    );
                }
            }
        }

        // The native managed actor owns the durable logical Iroh connection.
        // A carrier mechanism cannot repair a missing committed connection on
        // its own, so only generation-current Iroh liveness may suppress the
        // bounded redial after a network drop.
        #[cfg(not(target_arch = "wasm32"))]
        let transport_connected = self.is_connection_transport_alive(endpoint_id).await;
        #[cfg(target_arch = "wasm32")]
        let transport_connected = self.is_connected(endpoint_id).await;
        if let Some(snapshot) = managed_health_snapshot.as_ref() {
            // The Rust core owns replacement recovery. Suppress a competing
            // dial while no transport owns the replacement window, but never
            // let that suppression bypass positive liveness probes for an
            // already-owned generation.
            if should_suppress_auto_connect_for_replacement_window(
                &snapshot,
                now,
                transport_connected,
            ) {
                log_skip("replacement-pending");
                return;
            }
        }
        #[cfg(target_arch = "wasm32")]
        if !transport_connected {
            if let Some(existing) = self
                .connection_manager
                .get_by_connection_id(&connection_id)
                .await
            {
                if matches!(
                    existing.state,
                    crate::connection_manager::ConnectionState::Connected
                ) {
                    web_sys::console::warn_1(&wasm_bindgen::JsValue::from_str(&format!(
                        "[pluto-rtc][auto-connect][wasm] transport mismatch remote_device_id={} remote_node_id={} connection_id={} record_stable_id={:?} record_generation={} state={:?} status_reason={:?}",
                        remote_device_id,
                        node_id_str,
                        connection_id,
                        existing.transport_stable_id,
                        existing.transport_generation,
                        existing.state,
                        existing.status_reason
                    )));
                }
            }
        }

        // Tie breaker only controls who should dial.
        // If an incoming transport is already connected, the non-initiator must
        // observe and promote that transport instead of waiting forever for the
        // initiator path to "win".
        //
        // Use exponential backoff with jitter so repeated escalations don't
        // produce duplicate-storm collisions (2s→4s→8s→16s cap, ±20% jitter).
        let esc_count = non_initiator_escalation_count
            .get(&remote_device_id)
            .copied()
            .unwrap_or(0);
        let effective_grace_ms = if local_node_id <= node_id_str {
            non_initiator_escalation_grace_ms(esc_count)
        } else {
            initiator_grace_ms
        };
        let waited_ms = if local_node_id <= node_id_str {
            let wait_started_at = non_initiator_wait_started_at
                .entry(remote_device_id.clone())
                .or_insert(now);
            now.saturating_sub(*wait_started_at)
        } else {
            0
        };
        match auto_connect_tie_break_decision(
            &local_node_id,
            &node_id_str,
            transport_connected,
            waited_ms,
            effective_grace_ms,
        ) {
            AutoConnectTieBreakDecision::ObserveConnectedTransport => {
                non_initiator_wait_started_at.remove(&remote_device_id);
                non_initiator_escalation_count.remove(&remote_device_id);
            }
            AutoConnectTieBreakDecision::WaitForInitiator => {
                if waited_ms < 1_000 {
                    log_skip("tie-break-not-initiator-waiting");
                }
                return;
            }
            AutoConnectTieBreakDecision::ActAsInitiator => {
                if local_node_id <= node_id_str {
                    // Non-initiator escalated after grace period expired.
                    let new_esc_count = esc_count.saturating_add(1);
                    non_initiator_escalation_count.insert(remote_device_id.clone(), new_esc_count);
                    #[cfg(not(target_arch = "wasm32"))]
                    eprintln!(
                        "[pluto-rtc][auto-connect] non-initiator escalated after {}ms grace remote_device_id={} local_device_id={} escalation_count={}",
                        waited_ms, remote_device_id, local_device_id, new_esc_count
                    );
                    #[cfg(target_arch = "wasm32")]
                    web_sys::console::log_1(&wasm_bindgen::JsValue::from_str(&format!(
                        "[pluto-rtc][auto-connect] non-initiator escalated after {}ms grace remote_device_id={} local_device_id={} escalation_count={}",
                        waited_ms, remote_device_id, local_device_id, new_esc_count
                    )));
                }
                non_initiator_wait_started_at.remove(&remote_device_id);
            }
        }

        self.reconcile_authoritative_device_node(&remote_device_id, &node_id_str)
            .await;

        if transport_connected {
            let current_scopes = self.connection_manager.get_scopes(&connection_id).await;
            let admitted_scope = match self.session_admission(&connection_id) {
                crate::session_token::SessionAdmission::Accepted {
                    scope: Some(scope), ..
                } => Some(scope.into_inner()),
                _ => None,
            };
            let presentation_in_flight = self
                .session_token_registry
                .is_session_token_presentation_in_flight(&connection_id);
            if !persistent_managed_actor_may_own_transport(
                &current_scopes,
                admitted_scope.as_deref(),
                presentation_in_flight,
            ) {
                log_skip(if presentation_in_flight {
                    "explicit-admission-presentation-owns-current-transport"
                } else {
                    "non-managed-scope-owns-current-transport"
                });
                return;
            }

            connected_health_checked_at
                .entry(remote_device_id.clone())
                .or_insert(now);
            connected_health_state
                .entry(remote_device_id.clone())
                .or_insert(true);

            let should_probe = match connected_health_checked_at.get(&remote_device_id) {
                Some(last_probe_at) => {
                    now.saturating_sub(*last_probe_at) >= connected_health_probe_ms
                }
                None => false,
            };
            let mut healthy = connected_health_state
                .get(&remote_device_id)
                .copied()
                .unwrap_or(true);
            let mut probed_transport_stable_id = None;

            if should_probe {
                let active_probe = self.validate_active_connection(endpoint_id).await;
                probed_transport_stable_id =
                    active_probe.as_ref().map(|probe| probe.transport_stable_id);
                let active_probe_healthy =
                    active_probe.as_ref().is_some_and(|probe| probe.responsive);
                let passive_transport_alive = self.is_connection_transport_alive(endpoint_id).await;
                let recent_generation_activity = active_probe.as_ref().is_some_and(|probe| {
                    generation_activity_is_recent(
                        probe.last_inbound_activity_age,
                        connected_health_probe_ms,
                    ) || generation_activity_is_recent(
                        self.native_transport_protocol_activity_age(
                            &connection_id,
                            probe.transport_stable_id,
                        ),
                        connected_health_probe_ms,
                    )
                });
                let probe_decision = active_connection_probe_decision_with_generation_activity(
                    active_probe_healthy,
                    passive_transport_alive,
                    recent_generation_activity,
                );
                let current_failures = connected_health_failures
                    .get(&remote_device_id)
                    .copied()
                    .unwrap_or(0);
                let failures =
                    next_active_connection_probe_failure_count(probe_decision, current_failures);
                healthy = active_connection_probe_is_healthy(
                    probe_decision,
                    failures,
                    connected_health_failure_threshold,
                );
                connected_health_checked_at.insert(remote_device_id.clone(), now);
                connected_health_state.insert(remote_device_id.clone(), healthy);
                if matches!(probe_decision, ActiveConnectionProbeDecision::Healthy) {
                    connected_health_failures.remove(&remote_device_id);
                    // Keep the canonical peer snapshot aligned with the transport
                    // liveness decision. This is especially important after an
                    // in-place app relaunch: the old generation may have marked the
                    // peer stale while the replacement transport is already live.
                    // A matching pong is current-generation readiness evidence.
                    let _ = self
                        .confirm_managed_connection_readiness(&connection_id)
                        .await;
                } else {
                    connected_health_failures.insert(remote_device_id.clone(), failures);
                    if matches!(
                        probe_decision,
                        ActiveConnectionProbeDecision::PreserveLiveTransport
                    ) && healthy
                    {
                        log_skip("active-probe-missed-preserving-live-transport");
                    }
                }
            }

            if !healthy {
                if !should_probe {
                    log_skip("awaiting-current-generation-health-probe");
                    return;
                }
                if self.is_app_execution_suspended() {
                    log_skip("backgrounded-after-health-probe");
                    return;
                }
                #[cfg(not(target_arch = "wasm32"))]
                {
                    let failures = connected_health_failures
                        .get(&remote_device_id)
                        .copied()
                        .unwrap_or(1);
                    if failures < connected_health_failure_threshold {
                        log_skip("active-connection-health-probe-failed");
                        return;
                    }
                    let Some(transport_stable_id) = probed_transport_stable_id else {
                        log_skip("active-probe-generation-disappeared");
                        connected_health_checked_at.remove(&remote_device_id);
                        connected_health_state.remove(&remote_device_id);
                        connected_health_failures.remove(&remote_device_id);
                        return;
                    };
                    match self
                        .retire_transport_generation_after_failed_probe(
                            &connection_id,
                            endpoint_id,
                            transport_stable_id,
                            connected_health_probe_ms,
                            crate::lifecycle_reason::REASON_STALE_ACTIVE_CONNECTION_RECONNECT,
                        )
                        .await
                    {
                        FailedProbeRetirementOutcome::Retired => {
                            log_skip("stale-active-connection-reconnect");
                        }
                        FailedProbeRetirementOutcome::DeferredAdmissionInFlight => {
                            log_skip("active-probe-retirement-deferred-admission-in-flight");
                            connected_health_checked_at.insert(remote_device_id.clone(), now);
                            connected_health_state.insert(remote_device_id.clone(), true);
                            connected_health_failures.remove(&remote_device_id);
                            return;
                        }
                        FailedProbeRetirementOutcome::CancelledCurrentGenerationActivity => {
                            log_skip(
                                "active-probe-retirement-cancelled-by-current-generation-activity",
                            );
                            connected_health_checked_at.insert(remote_device_id.clone(), now);
                            connected_health_state.insert(remote_device_id.clone(), true);
                            connected_health_failures.remove(&remote_device_id);
                            let _ = self
                                .confirm_managed_connection_readiness(&connection_id)
                                .await;
                            return;
                        }
                        FailedProbeRetirementOutcome::GenerationChanged => {
                            log_skip("active-probe-generation-disappeared");
                            connected_health_checked_at.remove(&remote_device_id);
                            connected_health_state.remove(&remote_device_id);
                            connected_health_failures.remove(&remote_device_id);
                            return;
                        }
                    }
                    self.maybe_refresh_gateway_presence_for_auto_connect(
                        user_id,
                        local_device_id,
                        "stale-active-connection-reconnect",
                        last_presence_republish_at_ms,
                        presence_republish_interval_ms,
                    )
                    .await;
                    self.reconcile_authoritative_device_node(&remote_device_id, &node_id_str)
                        .await;
                    connected_health_checked_at.remove(&remote_device_id);
                    connected_health_state.remove(&remote_device_id);
                    connected_health_failures.remove(&remote_device_id);
                    if failures >= network_change_failure_threshold {
                        self.maybe_force_network_change_for_auto_connect(
                            user_id,
                            local_device_id,
                            "stale-active-connection",
                            last_network_change_at_ms,
                            network_change_recovery_interval_ms,
                            last_presence_republish_at_ms,
                            presence_republish_interval_ms,
                        )
                        .await;
                    }
                }
                #[cfg(target_arch = "wasm32")]
                {
                    let failures = connected_health_failures
                        .get(&remote_device_id)
                        .copied()
                        .unwrap_or(1);
                    if failures < connected_health_failure_threshold {
                        eprintln!(
                            "[pluto-rtc][auto-connect][health] remote_device_id={} local_device_id={} remote_node_id={} healthy=false failures={} threshold={} action=wait",
                            remote_device_id,
                            local_device_id,
                            node_id_str,
                            failures,
                            connected_health_failure_threshold
                        );
                        log_skip("active-connection-health-probe-failed");
                        return;
                    }
                    eprintln!(
                        "[pluto-rtc][auto-connect][health] remote_device_id={} local_device_id={} remote_node_id={} healthy=false failures={} threshold={} action=disconnect-and-reconnect",
                        remote_device_id,
                        local_device_id,
                        node_id_str,
                        failures,
                        connected_health_failure_threshold
                    );
                    log_skip("stale-active-connection-reconnect");
                    let _ = self
                        .disconnect_with_reason(
                            endpoint_id,
                            crate::lifecycle_reason::REASON_STALE_ACTIVE_CONNECTION_RECONNECT,
                        )
                        .await;
                    connected_health_checked_at.remove(&remote_device_id);
                    connected_health_state.remove(&remote_device_id);
                    connected_health_failures.remove(&remote_device_id);
                }
            } else {
                // Do not clear a bounded passive-open miss here. A transport
                // handle can remain locally open after the remote process has
                // disappeared, so only the matching positive probe above may
                // reset `connected_health_failures`. Clearing on the
                // temporarily-healthy PreserveLiveTransport decision would
                // keep every pass at failure 1 and make the retirement
                // threshold unreachable.
                if self.is_auto_connect_peer_excluded(&remote_device_id, Some(&node_id_str)) {
                    log_skip("auto-connect-excluded-after-health-probe");
                    let _ = self
                        .disconnect_with_reason(
                            endpoint_id,
                            crate::lifecycle_reason::REASON_MANUAL_DISCONNECT,
                        )
                        .await;
                    return;
                }
                self.connection_manager
                    .upsert_pending(
                        connection_id.clone(),
                        Some(node_id_str.clone()),
                        Some(remote_device_id.clone()),
                        Some(node_id_str.clone()),
                    )
                    .await;
                let current_transport_stable_id = self
                    .get_connection(endpoint_id)
                    .await
                    .map(|connection| crate::transport_generation::for_connection(&connection));
                self.commit_current_native_transport_record(
                    &connection_id,
                    &node_id_str,
                    Some(remote_device_id.clone()),
                    current_transport_stable_id,
                    Some("health-check".to_string()),
                )
                .await;
                #[cfg(target_arch = "wasm32")]
                {
                    let active_transport =
                        self.iroh_path_kind(&node_id_str).await.transport_label();
                    let _ = self
                        .report_transport_status_for_current_generation(
                            &connection_id,
                            active_transport,
                            None,
                        )
                        .await;
                }

                // Admission is bound to the physical transport generation, not
                // to whichever fresh ticket happens to be in the latest presence
                // row. Native peers routinely rotate advertised ticket tokens
                // while an already admitted connection remains live. Re-presenting
                // the replacement token creates a needless control stream and can
                // disturb an otherwise settled carrier generation.
                #[cfg(not(target_arch = "wasm32"))]
                let has_remote_admission_proof =
                    current_transport_stable_id.is_some_and(|transport_stable_id| {
                        self.remote_session_token_admitted_for_transport(
                            &connection_id,
                            transport_stable_id,
                        )
                    });
                #[cfg(target_arch = "wasm32")]
                let has_remote_admission_proof = if let Some(token) = extracted_token.as_deref() {
                    self.has_current_remote_session_admission_proof(
                        &connection_id,
                        endpoint_id,
                        token,
                    )
                    .await
                } else {
                    false
                };
                #[cfg(not(target_arch = "wasm32"))]
                let has_current_inbound_admission_proof =
                    current_transport_stable_id.is_some_and(|transport_stable_id| {
                        self.inbound_session_token_admitted_for_transport(
                            &connection_id,
                            transport_stable_id,
                        )
                    });
                #[cfg(not(target_arch = "wasm32"))]
                let peer_requested_reciprocal =
                    self.has_pending_reciprocal_session_admission_request(&connection_id);
                #[cfg(not(target_arch = "wasm32"))]
                if has_remote_admission_proof
                    && has_current_inbound_admission_proof
                    && peer_requested_reciprocal
                {
                    self.clear_reciprocal_session_admission_request(&connection_id);
                }
                #[cfg(not(target_arch = "wasm32"))]
                let presentation_decision = session_admission_presentation_decision(
                    extracted_token.is_some(),
                    has_remote_admission_proof,
                    has_current_inbound_admission_proof,
                    peer_requested_reciprocal,
                    local_node_id > node_id_str,
                );
                #[cfg(target_arch = "wasm32")]
                let presentation_decision = session_admission_presentation_decision(
                    extracted_token.is_some(),
                    has_remote_admission_proof,
                    true,
                    false,
                    true,
                );
                if presentation_decision.should_present {
                    if let Some(backoff_until) =
                        failure_backoff_until.get(&remote_device_id).copied()
                    {
                        if now < backoff_until {
                            log_skip("session-admission-backoff");
                            return;
                        }
                        failure_backoff_until.remove(&remote_device_id);
                    }

                    let token = extracted_token
                        .as_ref()
                        .expect("representation requires an extracted token");
                    eprintln!(
                        "[pluto-rtc][auto-connect][session-admission] re-presenting token on live transport connection_id={} remote_device_id={} remote_node_id={} token_fp={}",
                        connection_id,
                        remote_device_id,
                        node_id_str,
                        super::core_impl::log_fingerprint(token.as_str())
                    );
                    #[cfg(not(target_arch = "wasm32"))]
                    let presentation_result = self
                        .present_and_accept_session_token_for_route_repair(
                            endpoint_id,
                            &connection_id,
                            token,
                            extracted_token_suffix.as_deref(),
                            Some(remote_device_id.clone()),
                            Some(local_device_id.to_string()),
                            has_remote_admission_proof,
                            presentation_decision.request_reciprocal,
                        )
                        .await;
                    #[cfg(target_arch = "wasm32")]
                    let presentation_result = self
                        .present_and_accept_session_token_with_local_claim(
                            endpoint_id,
                            &connection_id,
                            token,
                            extracted_token_suffix.as_deref(),
                            Some(remote_device_id.clone()),
                            Some(local_device_id.to_string()),
                        )
                        .await;
                    match presentation_result {
                        Ok(approval_scope) => {
                            clear_auto_connect_failure_state(
                                &remote_device_id,
                                failure_count,
                                failure_backoff_until,
                            );
                            eprintln!(
                                "[pluto-rtc][auto-connect][session-admission] live transport admitted connection_id={} remote_device_id={} remote_node_id={} scope={} token_fp={}",
                                connection_id,
                                remote_device_id,
                                node_id_str,
                                approval_scope,
                                super::core_impl::log_fingerprint(token.as_str())
                            );
                        }
                        Err(error) => {
                            let active_transport_stable_id =
                                self.get_connection(endpoint_id).await.map(|connection| {
                                    crate::transport_generation::for_connection(&connection)
                                });
                            if current_transport_stable_id != active_transport_stable_id {
                                eprintln!(
                                    "[pluto-rtc][auto-connect][session-admission] stale presentation failure ignored because replacement already owns endpoint connection_id={} remote_device_id={} attempted_transport_stable_id={:?} active_transport_stable_id={:?}",
                                    connection_id,
                                    remote_device_id,
                                    current_transport_stable_id,
                                    active_transport_stable_id
                                );
                                return;
                            }

                            #[cfg(not(target_arch = "wasm32"))]
                            let current_generation_is_routable = active_transport_stable_id
                                .is_some_and(|transport_stable_id| {
                                    self.native_application_stream_admitted_for_transport(
                                        &connection_id,
                                        transport_stable_id,
                                    )
                                });
                            #[cfg(target_arch = "wasm32")]
                            let current_generation_is_routable = false;

                            if current_generation_is_routable {
                                clear_auto_connect_failure_state(
                                    &remote_device_id,
                                    failure_count,
                                    failure_backoff_until,
                                );
                                connected_health_failures.remove(&remote_device_id);
                                let _ = self
                                    .confirm_managed_connection_readiness(&connection_id)
                                    .await;
                                eprintln!(
                                    "[pluto-rtc][auto-connect][session-admission] ignored losing presentation failure because another exchange made the generation routable connection_id={} remote_device_id={} transport_stable_id={:?} error={}",
                                    connection_id,
                                    remote_device_id,
                                    active_transport_stable_id,
                                    error,
                                );
                                return;
                            }

                            // A duplicate never owns admission outcome. Ignore
                            // it even if the canonical exchange completed
                            // between returning this tagged error and handling
                            // it here; only the canonical presenter may count
                            // or retire its physical generation.
                            if session_token_presentation_was_duplicate(&error) {
                                eprintln!(
                                    "[pluto-rtc][auto-connect][session-admission] ignored duplicate presentation owned by canonical exchange connection_id={} remote_device_id={} transport_stable_id={:?}",
                                    connection_id,
                                    remote_device_id,
                                    active_transport_stable_id,
                                );
                                return;
                            }

                            if self
                                .session_token_registry
                                .is_session_token_presentation_in_flight(&connection_id)
                            {
                                eprintln!(
                                    "[pluto-rtc][auto-connect][session-admission] ignored duplicate presentation while canonical exchange remains in flight connection_id={} remote_device_id={} transport_stable_id={:?}",
                                    connection_id,
                                    remote_device_id,
                                    active_transport_stable_id,
                                );
                                return;
                            }

                            if session_token_presentation_failure_action(&error)
                                == SessionTokenPresentationFailureAction::RejectAndDisconnect
                            {
                                let rejection_reason =
                                    format!("session-token-presentation-failed: {}", error);
                                self.reject_session_connection(
                                    &connection_id,
                                    rejection_reason.as_str(),
                                );
                                self.connection_manager
                                    .set_failed(&connection_id, Some(rejection_reason.clone()))
                                    .await;
                                let _ = self
                                    .disconnect_with_reason(
                                        endpoint_id,
                                        crate::lifecycle_reason::REASON_SESSION_ADMISSION_REJECTED,
                                    )
                                    .await;
                                self.forget_session_connection(&connection_id);
                                register_auto_connect_admission_rejection(
                                    &remote_device_id,
                                    now_millis_i64(),
                                    failure_count,
                                    failure_backoff_until,
                                );
                                eprintln!(
                                    "[pluto-rtc][auto-connect][session-admission] live transport admission rejected connection_id={} remote_device_id={} remote_node_id={} token_fp={} error={}",
                                    connection_id,
                                    remote_device_id,
                                    node_id_str,
                                    super::core_impl::log_fingerprint(token.as_str()),
                                    error
                                );
                                return;
                            }

                            register_auto_connect_failure(
                                &remote_device_id,
                                now_millis_i64(),
                                failure_count,
                                failure_backoff_until,
                            );
                            let failures =
                                failure_count.get(&remote_device_id).copied().unwrap_or(1);
                            match retryable_session_admission_failure_decision(
                                failures,
                                connected_health_failure_threshold,
                                current_transport_stable_id,
                                active_transport_stable_id,
                                runtime_instance_changed,
                            ) {
                                RetryableSessionAdmissionFailureDecision::PreserveCurrentTransport => {
                                    eprintln!(
                                        "[pluto-rtc][auto-connect][session-admission] transient presentation failure; preserving current generation for one bounded retry connection_id={} remote_device_id={} remote_node_id={} transport_stable_id={:?} failures={} threshold={} token_fp={} error={}",
                                        connection_id,
                                        remote_device_id,
                                        node_id_str,
                                        current_transport_stable_id,
                                        failures,
                                        connected_health_failure_threshold,
                                        super::core_impl::log_fingerprint(token.as_str()),
                                        error
                                    );
                                }
                                RetryableSessionAdmissionFailureDecision::RetireCurrentTransport => {
                                    eprintln!(
                                        "[pluto-rtc][auto-connect][session-admission] repeated transient presentation failure; retiring unresponsive generation for canonical reconnect connection_id={} remote_device_id={} remote_node_id={} transport_stable_id={:?} failures={} threshold={} token_fp={} error={}",
                                        connection_id,
                                        remote_device_id,
                                        node_id_str,
                                        current_transport_stable_id,
                                        failures,
                                        connected_health_failure_threshold,
                                        super::core_impl::log_fingerprint(token.as_str()),
                                        error
                                    );
                                    let _ = self
                                        .disconnect_with_reason(
                                            endpoint_id,
                                            crate::lifecycle_reason::REASON_STALE_ACTIVE_CONNECTION_RECONNECT,
                                        )
                                        .await;
                                    connected_health_checked_at.remove(&remote_device_id);
                                    connected_health_state.remove(&remote_device_id);
                                    connected_health_failures.remove(&remote_device_id);
                                }
                                RetryableSessionAdmissionFailureDecision::ReplacementAlreadyWon => {
                                    eprintln!(
                                        "[pluto-rtc][auto-connect][session-admission] transient failure belongs to retired generation; replacement preserved connection_id={} remote_device_id={} attempted_transport_stable_id={:?} active_transport_stable_id={:?}",
                                        connection_id,
                                        remote_device_id,
                                        current_transport_stable_id,
                                        active_transport_stable_id
                                    );
                                }
                            }
                            return;
                        }
                    }
                }

                #[cfg(not(target_arch = "wasm32"))]
                {
                    if self
                        .promote_known_native_user_device_connection(&connection_id, &node_id_str)
                        .await
                        .is_none()
                    {
                        let _ = self
                            .confirm_managed_connection_readiness(&connection_id)
                            .await;
                    }
                    // Health polling observes the current committed connection;
                    // it does not own carrier recovery. Initial connect, path
                    // changes, foreground resume, inbound signaling, and the
                    // Rust retry scheduler already feed the connection owner.
                    // Starting WebRTC here can transiently clear settlement on
                    // an otherwise healthy direct Iroh route.
                }
                #[cfg(target_arch = "wasm32")]
                {
                    let _ = self
                        .confirm_managed_connection_readiness(&connection_id)
                        .await;
                    self.emit_current_wasm_connection_state(&connection_id)
                        .await;
                }
                clear_auto_connect_failure_state(
                    &remote_device_id,
                    failure_count,
                    failure_backoff_until,
                );
                non_initiator_escalation_count.remove(&remote_device_id);
                log_skip("already-connected");
                return;
            }
        } else {
            connected_health_checked_at.remove(&remote_device_id);
            connected_health_state.remove(&remote_device_id);
            connected_health_failures.remove(&remote_device_id);
        }

        if let Some(backoff_until) = failure_backoff_until.get(&remote_device_id).copied() {
            if now < backoff_until {
                log_skip("failure-backoff");
                return;
            }
            failure_backoff_until.remove(&remote_device_id);
        }

        let last = last_attempt_at.get(&remote_device_id).cloned().unwrap_or(0);
        if now - last < retry_throttle_ms {
            log_skip("retry-throttle");
            return;
        }
        last_attempt_at.insert(remote_device_id.clone(), now);
        let connect_mode = "ticket-endpoint-addr";

        #[cfg(not(target_arch = "wasm32"))]
        {
            self.connection_manager
                .upsert_pending(
                    connection_id.clone(),
                    Some(node_id_str.clone()),
                    Some(remote_device_id.clone()),
                    Some(node_id_str.clone()),
                )
                .await;
            self.connection_manager.set_connecting(&connection_id).await;
            let dial_attempt_fence = self
                .connection_manager
                .get_by_connection_id(&connection_id)
                .await
                .map(|record| (record.transport_generation, record.transport_stable_id));

            eprintln!(
                "[pluto-rtc][auto-connect] attempt remote_device_id={} local_device_id={} local_node_id={} remote_node_id={} mode={}",
                remote_device_id,
                local_device_id,
                local_node_id,
                node_id_str,
                connect_mode
            );

            eprintln!("[pluto-rtc] auto-connecting to {}", remote_device_id);
            let result = self.ensure_connected_addr(endpoint_id, endpoint_addr).await;

            match result {
                Ok(()) => {
                    if self.is_auto_connect_peer_excluded(&remote_device_id, Some(&node_id_str)) {
                        log_skip("auto-connect-excluded-after-dial");
                        let _ = self
                            .disconnect_with_reason(
                                endpoint_id,
                                crate::lifecycle_reason::REASON_MANUAL_DISCONNECT,
                            )
                            .await;
                        self.connection_manager
                            .set_closed(
                                &connection_id,
                                Some(crate::lifecycle_reason::REASON_MANUAL_DISCONNECT.to_string()),
                            )
                            .await;
                        return;
                    }
                    let active_probe = self.validate_active_connection(endpoint_id).await;
                    let active_probe_stable =
                        active_probe.as_ref().is_some_and(|probe| probe.responsive);
                    let transport_alive = self.is_connection_transport_alive(endpoint_id).await;
                    let probe_decision =
                        active_connection_probe_decision(active_probe_stable, transport_alive);
                    let stable = matches!(
                        probe_decision,
                        ActiveConnectionProbeDecision::Healthy
                            | ActiveConnectionProbeDecision::PreserveLiveTransport
                    );
                    if !stable {
                        if self.is_app_execution_suspended() {
                            log_skip("backgrounded-after-stability-probe");
                            return;
                        }
                        // A probe miss is physical-generation evidence, not
                        // authority to fail the whole logical peer. An inbound
                        // winner may be admitting while this dial loses arbitration.
                        let Some(transport_stable_id) =
                            active_probe.as_ref().map(|probe| probe.transport_stable_id)
                        else {
                            log_skip("stability-probe-generation-disappeared");
                            return;
                        };
                        if !matches!(
                            self.retire_transport_generation_after_failed_probe(
                                &connection_id,
                                endpoint_id,
                                transport_stable_id,
                                connected_health_probe_ms,
                                crate::lifecycle_reason::REASON_CONNECTION_FAILED_STABILITY,
                            )
                            .await,
                            FailedProbeRetirementOutcome::Retired
                        ) {
                            log_skip("stability-probe-retirement-preserved-current-generation");
                            return;
                        }
                        self.reconcile_authoritative_device_node(&remote_device_id, &node_id_str)
                            .await;
                        register_auto_connect_failure(
                            &remote_device_id,
                            now_millis_i64(),
                            failure_count,
                            failure_backoff_until,
                        );
                        let current_failures =
                            failure_count.get(&remote_device_id).copied().unwrap_or(0);
                        if current_failures >= network_change_failure_threshold
                            && self
                                .should_force_network_change_after_connect_failures(
                                    &remote_device_id,
                                )
                                .await
                        {
                            self.maybe_force_network_change_for_auto_connect(
                                user_id,
                                local_device_id,
                                "unstable-connection",
                                last_network_change_at_ms,
                                network_change_recovery_interval_ms,
                                last_presence_republish_at_ms,
                                presence_republish_interval_ms,
                            )
                            .await;
                        }
                        eprintln!(
                            "[pluto-rtc][auto-connect] unstable connection suppressed remote_device_id={} local_device_id={} local_node_id={} remote_node_id={} connection_id={}",
                            remote_device_id,
                            local_device_id,
                            local_node_id,
                            node_id_str,
                            connection_id
                        );
                        return;
                    }
                    if matches!(
                        probe_decision,
                        ActiveConnectionProbeDecision::PreserveLiveTransport
                    ) {
                        log_skip("stability-probe-missed-preserving-live-transport");
                    }
                    let current_transport_stable_id = self
                        .get_connection(endpoint_id)
                        .await
                        .map(|connection| crate::transport_generation::for_connection(&connection));
                    self.commit_current_native_transport_record(
                        &connection_id,
                        &node_id_str,
                        Some(remote_device_id.clone()),
                        current_transport_stable_id,
                        Some(connect_mode.to_string()),
                    )
                    .await;

                    if let Some(ref token) = extracted_token {
                        let bilateral_user_device = extracted_token_payload
                            .as_ref()
                            .is_some_and(|payload| payload.scope.as_str() == "user-device");
                        let has_remote_admission_proof =
                            current_transport_stable_id.is_some_and(|transport_stable_id| {
                                self.remote_session_token_admitted_for_transport(
                                    &connection_id,
                                    transport_stable_id,
                                )
                            });
                        let has_current_inbound_admission_proof = current_transport_stable_id
                            .is_some_and(|transport_stable_id| {
                                self.inbound_session_token_admitted_for_transport(
                                    &connection_id,
                                    transport_stable_id,
                                )
                            });
                        let peer_requested_reciprocal =
                            self.has_pending_reciprocal_session_admission_request(&connection_id);
                        let presentation_decision = if bilateral_user_device {
                            session_admission_presentation_decision(
                                true,
                                has_remote_admission_proof,
                                has_current_inbound_admission_proof,
                                peer_requested_reciprocal,
                                local_node_id > node_id_str,
                            )
                        } else {
                            SessionAdmissionPresentationDecision {
                                should_present: !has_remote_admission_proof,
                                request_reciprocal: false,
                            }
                        };
                        if !presentation_decision.should_present {
                            eprintln!(
                                "[pluto-rtc][auto-connect][session-admission] awaiting elected presenter on fresh transport connection_id={} remote_device_id={} remote_node_id={} local_node_id={}",
                                connection_id,
                                remote_device_id,
                                node_id_str,
                                local_node_id,
                            );
                            return;
                        }
                        match self
                            .present_and_accept_session_token_for_route_repair(
                                endpoint_id,
                                &connection_id,
                                token,
                                extracted_token_suffix.as_deref(),
                                Some(remote_device_id.clone()),
                                Some(local_device_id.to_string()),
                                has_remote_admission_proof,
                                presentation_decision.request_reciprocal,
                            )
                            .await
                        {
                            Ok(approval_scope) => {
                                eprintln!(
                                    "[pluto-rtc][auto-connect] session-token presented and approved connection_id={} scope={} device_id={} token_fp={} local_admission=accepted",
                                    connection_id,
                                    approval_scope,
                                    remote_device_id,
                                    super::core_impl::log_fingerprint(token.as_str())
                                );
                            }
                            Err(error) => {
                                if session_token_presentation_failure_action(&error)
                                    == SessionTokenPresentationFailureAction::RetryOnNextTransport
                                {
                                    register_auto_connect_failure(
                                        &remote_device_id,
                                        now_millis_i64(),
                                        failure_count,
                                        failure_backoff_until,
                                    );
                                    eprintln!(
                                        "[pluto-rtc][auto-connect] transient session-token presentation failure; preserving peer session for canonical transport retry connection_id={} device_id={} token_fp={} error={}",
                                        connection_id,
                                        remote_device_id,
                                        super::core_impl::log_fingerprint(token.as_str()),
                                        error
                                    );
                                    return;
                                }
                                let rejection_reason =
                                    format!("session-token-presentation-failed: {}", error);
                                self.reject_session_connection(
                                    &connection_id,
                                    rejection_reason.as_str(),
                                );
                                self.connection_manager
                                    .set_failed(&connection_id, Some(rejection_reason.clone()))
                                    .await;
                                let _ = self
                                    .disconnect_with_reason(
                                        endpoint_id,
                                        crate::lifecycle_reason::REASON_SESSION_ADMISSION_REJECTED,
                                    )
                                    .await;
                                self.reconcile_authoritative_device_node(
                                    &remote_device_id,
                                    &node_id_str,
                                )
                                .await;
                                // Clear the rejection state so the next auto-connect
                                // attempt can re-present a (potentially refreshed) token.
                                // Without this the session stays in Rejected and no
                                // further token presentation occurs for this peer.
                                self.forget_session_connection(&connection_id);
                                register_auto_connect_admission_rejection(
                                    &remote_device_id,
                                    now_millis_i64(),
                                    failure_count,
                                    failure_backoff_until,
                                );
                                eprintln!(
                                    "[pluto-rtc][auto-connect] session-token presentation failed connection_id={} device_id={} token_fp={} error={}",
                                    connection_id,
                                    remote_device_id,
                                    super::core_impl::log_fingerprint(token.as_str()),
                                    error
                                );
                                return;
                            }
                        }
                    }

                    let _ = self
                        .confirm_managed_connection_readiness(&connection_id)
                        .await;
                    // Spawn a path-watcher for this outgoing connection so that
                    // relay → direct-QUIC (and back) transitions retrigger or
                    // suspend transport upgrades reactively — matching the
                    // behaviour we already have for incoming connections.
                    if let Some(outgoing_conn) = self.get_connection(endpoint_id).await {
                        // force_restart=false: the path-watcher fires on relay↔direct
                        // transitions after connection. A Connecting session started just
                        // above should not be replaced by path-change retriggers.
                        self.spawn_iroh_path_watcher(
                            &connection_id,
                            &node_id_str,
                            &outgoing_conn,
                            false,
                        );
                    }
                    clear_auto_connect_failure_state(
                        &remote_device_id,
                        failure_count,
                        failure_backoff_until,
                    );
                    non_initiator_escalation_count.remove(&remote_device_id);
                    eprintln!(
                        "[pluto-rtc][auto-connect] connected remote_device_id={} local_device_id={} local_node_id={} remote_node_id={} connection_id={}",
                        remote_device_id,
                        local_device_id,
                        local_node_id,
                        node_id_str,
                        connection_id
                    );
                }
                Err(e) => {
                    if let Some((transport_generation, transport_stable_id)) = dial_attempt_fence {
                        let failed = self
                            .connection_manager
                            .set_failed_if_dial_attempt_current(
                                &connection_id,
                                transport_generation,
                                transport_stable_id,
                                Some(
                                    crate::lifecycle_reason::REASON_AUTO_CONNECT_FAILURE
                                        .to_string(),
                                ),
                            )
                            .await;
                        if failed.is_none() {
                            eprintln!(
                                "[pluto-rtc][auto-connect] ignored stale dial failure connection_id={} remote_device_id={} remote_node_id={} expected_transport_generation={} expected_transport_stable_id={:?}",
                                connection_id,
                                remote_device_id,
                                node_id_str,
                                transport_generation,
                                transport_stable_id,
                            );
                            // The successor's accept/admission path owns all
                            // remaining settlement. Do not charge this stale
                            // attempt to backoff or let its failure threshold
                            // disconnect the current physical generation.
                            return;
                        }
                    }
                    register_auto_connect_failure(
                        &remote_device_id,
                        now_millis_i64(),
                        failure_count,
                        failure_backoff_until,
                    );
                    let current_failures =
                        failure_count.get(&remote_device_id).copied().unwrap_or(0);
                    if current_failures >= network_change_failure_threshold
                        && self
                            .should_force_network_change_after_connect_failures(&remote_device_id)
                            .await
                    {
                        let _ = self
                            .disconnect_with_reason(
                                endpoint_id,
                                crate::lifecycle_reason::REASON_NETWORK_CHANGE_RECONNECT,
                            )
                            .await;
                        self.reconcile_authoritative_device_node(&remote_device_id, &node_id_str)
                            .await;
                        self.maybe_force_network_change_for_auto_connect(
                            user_id,
                            local_device_id,
                            "repeated-connect-failures",
                            last_network_change_at_ms,
                            network_change_recovery_interval_ms,
                            last_presence_republish_at_ms,
                            presence_republish_interval_ms,
                        )
                        .await;
                    }
                    self.maybe_refresh_gateway_presence_for_auto_connect(
                        user_id,
                        local_device_id,
                        "connect-failed",
                        last_presence_republish_at_ms,
                        presence_republish_interval_ms,
                    )
                    .await;
                    eprintln!(
                        "[pluto-rtc] auto-connect failed to {}: {}",
                        remote_device_id, e
                    );
                }
            }
        }

        #[cfg(target_arch = "wasm32")]
        {
            self.connection_manager
                .upsert_pending(
                    connection_id.clone(),
                    Some(node_id_str.clone()),
                    Some(remote_device_id.clone()),
                    Some(node_id_str.clone()),
                )
                .await;
            // Pre-populate device_id from the signaling record so that peer
            // lookups by device_id succeed immediately instead of waiting
            // for the accept probe to backfill it (~18 s in worst case).
            self.connection_manager
                .set_device_id(&connection_id, remote_device_id.clone())
                .await;
            self.connection_manager.set_connecting(&connection_id).await;
            self.emit_current_wasm_connection_state(&connection_id)
                .await;
            web_sys::console::log_1(&wasm_bindgen::JsValue::from_str(&format!(
                "[pluto-rtc][auto-connect] attempt remote_device_id={} local_device_id={} local_node_id={} remote_node_id={} mode={}",
                remote_device_id,
                local_device_id,
                local_node_id,
                node_id_str,
                connect_mode
            )));
            web_sys::console::log_1(&wasm_bindgen::JsValue::from_str(&format!(
                "[pluto-rtc] auto-connecting to {}",
                remote_device_id
            )));
            let result = self.ensure_connected_addr(endpoint_id, endpoint_addr).await;

            match result {
                Ok(()) => {
                    let active_probe_stable = self.validate_active_connection(endpoint_id).await;
                    let transport_alive = self.is_connection_transport_alive(endpoint_id).await;
                    let probe_decision =
                        active_connection_probe_decision(active_probe_stable, transport_alive);
                    let stable = matches!(
                        probe_decision,
                        ActiveConnectionProbeDecision::Healthy
                            | ActiveConnectionProbeDecision::PreserveLiveTransport
                    );
                    if !stable {
                        if self.is_app_execution_suspended() {
                            log_skip("backgrounded-after-stability-probe");
                            return;
                        }
                        self.connection_manager
                            .upsert_pending(
                                connection_id.clone(),
                                Some(node_id_str.clone()),
                                Some(remote_device_id.clone()),
                                Some(node_id_str.clone()),
                            )
                            .await;
                        self.connection_manager
                            .set_failed(
                                &connection_id,
                                Some("connection failed stability verification".to_string()),
                            )
                            .await;
                        self.emit_current_wasm_connection_state(&connection_id)
                            .await;
                        register_auto_connect_failure(
                            &remote_device_id,
                            now_millis_i64(),
                            failure_count,
                            failure_backoff_until,
                        );
                        let _ = self
                            .disconnect_with_reason(
                                endpoint_id,
                                crate::lifecycle_reason::REASON_AUTO_CONNECT_FAILURE,
                            )
                            .await;
                        web_sys::console::warn_1(&wasm_bindgen::JsValue::from_str(&format!(
                            "[pluto-rtc][auto-connect] unstable connection suppressed remote_device_id={} local_device_id={} local_node_id={} remote_node_id={}",
                            remote_device_id,
                            local_device_id,
                            local_node_id,
                            node_id_str
                        )));
                        return;
                    }
                    if matches!(
                        probe_decision,
                        ActiveConnectionProbeDecision::PreserveLiveTransport
                    ) {
                        web_sys::console::log_1(&wasm_bindgen::JsValue::from_str(&format!(
                            "[pluto-rtc][auto-connect] stability probe missed but transport remains alive; preserving connection remote_device_id={} local_device_id={} local_node_id={} remote_node_id={} connection_id={}",
                            remote_device_id,
                            local_device_id,
                            local_node_id,
                            node_id_str,
                            connection_id
                        )));
                    }

                    self.connection_manager
                        .upsert_pending(
                            connection_id.clone(),
                            Some(node_id_str.clone()),
                            Some(remote_device_id.clone()),
                            Some(node_id_str.clone()),
                        )
                        .await;
                    let pending_record = self
                        .connection_manager
                        .get_by_connection_id(&connection_id)
                        .await;
                    web_sys::console::log_1(&wasm_bindgen::JsValue::from_str(&format!(
                        "[pluto-rtc][auto-connect][wasm] upsert_pending remote_device_id={} remote_node_id={} connection_id={} pending_record_exists={} record={:?}",
                        remote_device_id,
                        node_id_str,
                        connection_id,
                        pending_record.is_some(),
                        pending_record,
                    )));
                    let transport_stable_id = self
                        .get_connection(endpoint_id)
                        .await
                        .map(|connection| crate::transport_generation::for_connection(&connection));
                    self.commit_current_native_transport_record(
                        &connection_id,
                        &node_id_str,
                        Some(remote_device_id.clone()),
                        transport_stable_id,
                        Some(connect_mode.to_string()),
                    )
                    .await;
                    let active_transport =
                        self.iroh_path_kind(&node_id_str).await.transport_label();
                    let _ = self
                        .report_transport_status_for_current_generation(
                            &connection_id,
                            active_transport,
                            None,
                        )
                        .await;

                    if let Some(ref token) = extracted_token {
                        match self
                            .present_and_accept_session_token(
                                endpoint_id,
                                &connection_id,
                                token,
                                extracted_token_suffix.as_deref(),
                                Some(remote_device_id.clone()),
                            )
                            .await
                        {
                            Ok(approval_scope) => {
                                web_sys::console::log_1(&wasm_bindgen::JsValue::from_str(&format!(
                                    "[pluto-rtc][auto-connect][wasm] session-token presented and approved connection_id={} scope={} device_id={} token_fp={} local_admission=accepted",
                                    connection_id,
                                    approval_scope,
                                    remote_device_id,
                                    super::core_impl::log_fingerprint(token.as_str())
                                )));
                            }
                            Err(error) => {
                                if session_token_presentation_failure_action(&error)
                                    == SessionTokenPresentationFailureAction::RetryOnNextTransport
                                {
                                    register_auto_connect_failure(
                                        &remote_device_id,
                                        now_millis_i64(),
                                        failure_count,
                                        failure_backoff_until,
                                    );
                                    web_sys::console::warn_1(&wasm_bindgen::JsValue::from_str(&format!(
                                        "[pluto-rtc][auto-connect][wasm] transient session-token presentation failure; preserving peer session for canonical transport retry connection_id={} device_id={} token_fp={} error={}",
                                        connection_id,
                                        remote_device_id,
                                        super::core_impl::log_fingerprint(token.as_str()),
                                        error
                                    )));
                                    return;
                                }
                                let rejection_reason =
                                    format!("session-token-presentation-failed: {}", error);
                                self.reject_session_connection(
                                    &connection_id,
                                    rejection_reason.as_str(),
                                );
                                self.connection_manager
                                    .set_failed(&connection_id, Some(rejection_reason))
                                    .await;
                                self.emit_current_wasm_connection_state(&connection_id)
                                    .await;
                                let _ = self
                                    .disconnect_with_reason(
                                        endpoint_id,
                                        crate::lifecycle_reason::REASON_SESSION_ADMISSION_REJECTED,
                                    )
                                    .await;
                                // Token presentation failure belongs to this
                                // transport attempt. Let a later deterministic
                                // reconnect authenticate from a clean window.
                                self.forget_session_connection(&connection_id);
                                register_auto_connect_admission_rejection(
                                    &remote_device_id,
                                    now_millis_i64(),
                                    failure_count,
                                    failure_backoff_until,
                                );
                                web_sys::console::warn_1(&wasm_bindgen::JsValue::from_str(&format!(
                                    "[pluto-rtc][auto-connect][wasm] session-token presentation failed connection_id={} device_id={} token_fp={} error={}",
                                    connection_id,
                                    remote_device_id,
                                    super::core_impl::log_fingerprint(token.as_str()),
                                    error
                                )));
                                return;
                            }
                        }
                    }

                    let _ = self
                        .confirm_managed_connection_readiness(&connection_id)
                        .await;
                    let connected_record = self
                        .connection_manager
                        .get_by_connection_id(&connection_id)
                        .await;
                    let record_count = self.connection_manager.list_all().await.len();
                    web_sys::console::log_1(&wasm_bindgen::JsValue::from_str(&format!(
                        "[pluto-rtc][auto-connect][wasm] set_connected remote_device_id={} remote_node_id={} connection_id={} connected_record_exists={} total_records={} record={:?}",
                        remote_device_id,
                        node_id_str,
                        connection_id,
                        connected_record.is_some(),
                        record_count,
                        connected_record,
                    )));
                    self.emit_current_wasm_connection_state(&connection_id)
                        .await;
                    clear_auto_connect_failure_state(
                        &remote_device_id,
                        failure_count,
                        failure_backoff_until,
                    );
                    non_initiator_escalation_count.remove(&remote_device_id);
                    web_sys::console::log_1(&wasm_bindgen::JsValue::from_str(&format!(
                        "[pluto-rtc][auto-connect] connected remote_device_id={} local_device_id={} local_node_id={} remote_node_id={}",
                        remote_device_id,
                        local_device_id,
                        local_node_id,
                        node_id_str
                    )));
                }
                Err(e) => {
                    let err = e.to_string();
                    self.connection_manager
                        .upsert_pending(
                            connection_id.clone(),
                            Some(node_id_str.clone()),
                            Some(remote_device_id.clone()),
                            Some(node_id_str.clone()),
                        )
                        .await;
                    self.connection_manager
                        .set_failed(&connection_id, Some(err.clone()))
                        .await;
                    self.emit_current_wasm_connection_state(&connection_id)
                        .await;
                    register_auto_connect_failure(
                        &remote_device_id,
                        now_millis_i64(),
                        failure_count,
                        failure_backoff_until,
                    );
                    web_sys::console::error_1(&wasm_bindgen::JsValue::from_str(&format!(
                        "[pluto-rtc] auto-connect failed to {}: {}",
                        remote_device_id, err
                    )));
                }
            }
        }
    }

    #[cfg(not(target_arch = "wasm32"))]
    async fn run_auto_connect_snapshot(
        self: &Arc<Self>,
        user_id: &str,
        local_device_id: &str,
        last_attempt_at: &mut std::collections::HashMap<String, i64>,
        failure_count: &mut std::collections::HashMap<String, u8>,
        failure_backoff_until: &mut std::collections::HashMap<String, i64>,
        retry_throttle_ms: i64,
        non_initiator_wait_started_at: &mut std::collections::HashMap<String, i64>,
        non_initiator_escalation_count: &mut std::collections::HashMap<String, u8>,
        initiator_grace_ms: i64,
        skip_log_at: &mut std::collections::HashMap<String, i64>,
        connected_health_checked_at: &mut std::collections::HashMap<String, i64>,
        connected_health_state: &mut std::collections::HashMap<String, bool>,
        connected_health_failures: &mut std::collections::HashMap<String, u8>,
        connected_health_probe_ms: i64,
        connected_health_failure_threshold: u8,
        last_presence_republish_at_ms: &mut i64,
        presence_republish_interval_ms: i64,
        last_network_change_at_ms: &mut i64,
        network_change_recovery_interval_ms: i64,
        network_change_failure_threshold: u8,
        last_snapshot_signature: &mut Option<(usize, usize, usize)>,
        last_snapshot_log_at: &mut i64,
        last_known_node_id: &mut std::collections::HashMap<String, String>,
        known_devices: &mut std::collections::HashMap<String, crate::signaling::Device>,
    ) {
        if self.is_app_execution_suspended() {
            return;
        }

        let now = now_millis_i64();
        match self.search_devices(user_id).await {
            Ok(devices) => {
                let total = devices.len();
                let online = devices.iter().filter(|d| d.online).count();
                let remote_online = devices
                    .iter()
                    .filter(|d| d.online && d.device_id != local_device_id)
                    .count();
                let signature = (total, online, remote_online);
                let changed = last_snapshot_signature
                    .map(|prev| prev != signature)
                    .unwrap_or(true);
                let should_log = if auto_connect_verbose() {
                    now.saturating_sub(*last_snapshot_log_at) >= 2_000
                } else {
                    changed || now.saturating_sub(*last_snapshot_log_at) >= 30_000
                };
                if should_log {
                    *last_snapshot_log_at = now;
                    *last_snapshot_signature = Some(signature);
                    #[cfg(not(target_arch = "wasm32"))]
                    eprintln!(
                        "[pluto-rtc][auto-connect][snapshot] user_id={} local_device_id={} total_devices={} online_devices={} remote_online_devices={}",
                        user_id, local_device_id, total, online, remote_online
                    );
                    #[cfg(target_arch = "wasm32")]
                    web_sys::console::log_1(&wasm_bindgen::JsValue::from_str(&format!(
                        "[pluto-rtc][auto-connect][snapshot] user_id={} local_device_id={} total_devices={} online_devices={} remote_online_devices={}",
                        user_id, local_device_id, total, online, remote_online
                    )));
                }

                let known_device_ids: std::collections::HashSet<String> =
                    devices.iter().map(|d| d.device_id.clone()).collect();

                for device in devices {
                    known_devices.insert(device.device_id.clone(), device.clone());
                    self.try_auto_connect_device(
                        user_id,
                        device,
                        local_device_id,
                        last_attempt_at,
                        failure_count,
                        failure_backoff_until,
                        retry_throttle_ms,
                        non_initiator_wait_started_at,
                        non_initiator_escalation_count,
                        initiator_grace_ms,
                        skip_log_at,
                        connected_health_checked_at,
                        connected_health_state,
                        connected_health_failures,
                        connected_health_probe_ms,
                        connected_health_failure_threshold,
                        last_presence_republish_at_ms,
                        presence_republish_interval_ms,
                        last_network_change_at_ms,
                        network_change_recovery_interval_ms,
                        network_change_failure_threshold,
                        last_known_node_id,
                    )
                    .await;
                }

                // Prune tracking state for device IDs no longer returned by
                // the snapshot. Prevents unbounded growth as peers come and go
                // without always firing a DeviceEvent::Removed.
                last_attempt_at.retain(|k, _| known_device_ids.contains(k));
                failure_count.retain(|k, _| known_device_ids.contains(k));
                failure_backoff_until.retain(|k, _| known_device_ids.contains(k));
                non_initiator_wait_started_at.retain(|k, _| known_device_ids.contains(k));
                non_initiator_escalation_count.retain(|k, _| known_device_ids.contains(k));
                connected_health_checked_at.retain(|k, _| known_device_ids.contains(k));
                connected_health_state.retain(|k, _| known_device_ids.contains(k));
                connected_health_failures.retain(|k, _| known_device_ids.contains(k));
                last_known_node_id.retain(|key, _| {
                    let device_id = key
                        .strip_suffix("::runtime-instance")
                        .or_else(|| key.strip_suffix("::ticket"))
                        .unwrap_or(key.as_str());
                    known_device_ids.contains(device_id)
                });
                // skip_log_at keys are "device_id::reason" — prune by prefix
                skip_log_at.retain(|k, _| {
                    k.split_once("::")
                        .map_or(false, |(id, _)| known_device_ids.contains(id))
                });
            }
            Err(error) => {
                let min_interval = if auto_connect_verbose() {
                    2_000
                } else {
                    30_000
                };
                if now.saturating_sub(*last_snapshot_log_at) >= min_interval {
                    *last_snapshot_log_at = now;
                    #[cfg(not(target_arch = "wasm32"))]
                    eprintln!(
                        "[pluto-rtc][auto-connect][snapshot] user_id={} local_device_id={} search_devices_error={}",
                        user_id, local_device_id, error
                    );
                    #[cfg(target_arch = "wasm32")]
                    web_sys::console::error_1(&wasm_bindgen::JsValue::from_str(&format!(
                        "[pluto-rtc][auto-connect][snapshot] user_id={} local_device_id={} search_devices_error={}",
                        user_id, local_device_id, error
                    )));
                }
            }
        }
    }

    #[cfg(not(target_arch = "wasm32"))]
    pub(crate) async fn auto_connect_loop(
        self: Arc<Self>,
        user_id: String,
        local_device_id: String,
        generation: u64,
    ) {
        use futures::StreamExt;
        let mut last_attempt_at: std::collections::HashMap<String, i64> =
            std::collections::HashMap::new();
        let mut non_initiator_wait_started_at: std::collections::HashMap<String, i64> =
            std::collections::HashMap::new();
        let mut non_initiator_escalation_count: std::collections::HashMap<String, u8> =
            std::collections::HashMap::new();
        let mut skip_log_at: std::collections::HashMap<String, i64> =
            std::collections::HashMap::new();
        let mut connected_health_checked_at: std::collections::HashMap<String, i64> =
            std::collections::HashMap::new();
        let mut connected_health_state: std::collections::HashMap<String, bool> =
            std::collections::HashMap::new();
        let mut connected_health_failures: std::collections::HashMap<String, u8> =
            std::collections::HashMap::new();
        let mut failure_count: std::collections::HashMap<String, u8> =
            std::collections::HashMap::new();
        let mut failure_backoff_until: std::collections::HashMap<String, i64> =
            std::collections::HashMap::new();
        let mut last_known_node_id: std::collections::HashMap<String, String> =
            std::collections::HashMap::new();
        // The provider subscription owns discovery updates. Keep its latest
        // device values in memory so transport health can run on its own
        // cadence without turning every ping into a Firestore, RTDB, gateway,
        // or future-provider roster read.
        let mut known_devices: std::collections::HashMap<String, crate::signaling::Device> =
            std::collections::HashMap::new();
        let mut last_snapshot_signature: Option<(usize, usize, usize)> = None;
        let mut last_snapshot_log_at = 0_i64;
        let mut last_presence_republish_at_ms = 0_i64;
        let retry_throttle_ms = 1_000;
        // Gateway events deliver roster and live-presence changes.
        // This is only a low-frequency safety reconciliation; it must never be
        // used as a five-second full-scope polling loop.
        let rescan_interval_ms_fg: u64 = 300_000;
        let rescan_interval_ms_bg: u64 = 300_000;
        let initiator_grace_ms = 3_000;
        // Foreground: probe every 5s, fail after 2 misses (~10s total).
        // Background: skip health probing entirely (i64::MAX disables the probe).
        let connected_health_probe_ms_fg = 5_000_i64;
        let connected_health_probe_ms_bg = i64::MAX;
        let connected_health_failure_threshold = 2;
        // Failure-triggered presence repair is throttled to once per 10s.
        // The native presence actor owns periodic gateway lease refreshes.
        let presence_republish_interval_ms_fg = 10_000_i64;
        let presence_republish_interval_ms_bg = i64::MAX;
        let mut last_network_change_at_ms = 0_i64;
        // Foreground: network change recovery every 15s. Background: skip.
        let network_change_recovery_interval_ms_fg = 15_000_i64;
        let network_change_recovery_interval_ms_bg = i64::MAX;
        let network_change_failure_threshold = auto_connect_network_change_failure_threshold();

        loop {
            if !self.is_auto_connect_generation_current(generation) {
                #[cfg(not(target_arch = "wasm32"))]
                eprintln!(
                    "[pluto-rtc][auto-connect] stopping stale loop user_id={} local_device_id={} generation={}",
                    user_id, local_device_id, generation
                );
                #[cfg(target_arch = "wasm32")]
                web_sys::console::log_1(&wasm_bindgen::JsValue::from_str(&format!(
                    "[pluto-rtc][auto-connect] stopping stale loop user_id={} local_device_id={} generation={}",
                    user_id, local_device_id, generation
                )));
                break;
            }

            // Snapshot pass so auto-connect does not depend on receiving a specific
            // event type from the subscription backend.
            if !self.is_app_execution_suspended() {
                self.run_auto_connect_snapshot(
                    &user_id,
                    &local_device_id,
                    &mut last_attempt_at,
                    &mut failure_count,
                    &mut failure_backoff_until,
                    retry_throttle_ms,
                    &mut non_initiator_wait_started_at,
                    &mut non_initiator_escalation_count,
                    initiator_grace_ms,
                    &mut skip_log_at,
                    &mut connected_health_checked_at,
                    &mut connected_health_state,
                    &mut connected_health_failures,
                    connected_health_probe_ms_fg,
                    connected_health_failure_threshold,
                    &mut last_presence_republish_at_ms,
                    presence_republish_interval_ms_fg,
                    &mut last_network_change_at_ms,
                    network_change_recovery_interval_ms_fg,
                    network_change_failure_threshold,
                    &mut last_snapshot_signature,
                    &mut last_snapshot_log_at,
                    &mut last_known_node_id,
                    &mut known_devices,
                )
                .await;
            }

            #[allow(unused_mut)]
            let mut stream = match self.subscribe_devices(&user_id).await {
                Ok(s) => {
                    #[cfg(not(target_arch = "wasm32"))]
                    eprintln!(
                        "[pluto-rtc][auto-connect] subscribed user_id={} local_device_id={}",
                        user_id, local_device_id
                    );
                    #[cfg(target_arch = "wasm32")]
                    web_sys::console::log_1(&wasm_bindgen::JsValue::from_str(&format!(
                        "[pluto-rtc][auto-connect] subscribed user_id={} local_device_id={}",
                        user_id, local_device_id
                    )));
                    s
                }
                Err(e) => {
                    #[cfg(not(target_arch = "wasm32"))]
                    eprintln!(
                        "[pluto-rtc][auto-connect] subscribe failed user_id={} local_device_id={} error={}",
                        user_id, local_device_id, e
                    );
                    #[cfg(target_arch = "wasm32")]
                    web_sys::console::error_1(&wasm_bindgen::JsValue::from_str(&format!(
                        "[pluto-rtc][auto-connect] subscribe failed user_id={} local_device_id={} error={}",
                        user_id, local_device_id, e
                    )));

                    #[cfg(not(target_arch = "wasm32"))]
                    tokio::time::sleep(std::time::Duration::from_millis(1500)).await;
                    #[cfg(target_arch = "wasm32")]
                    gloo_timers::future::sleep(std::time::Duration::from_millis(1500)).await;
                    continue;
                }
            };

            #[cfg(not(target_arch = "wasm32"))]
            {
                let mut last_bg_state = self.is_app_execution_suspended();
                let mut rescan_interval =
                    tokio::time::interval(std::time::Duration::from_millis(if last_bg_state {
                        rescan_interval_ms_bg
                    } else {
                        rescan_interval_ms_fg
                    }));
                let mut health_interval = tokio::time::interval(std::time::Duration::from_millis(
                    connected_health_probe_ms_fg as u64,
                ));
                health_interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
                loop {
                    if !self.is_auto_connect_generation_current(generation) {
                        break;
                    }
                    // Recreate the interval when background state changes so the
                    // new period takes effect immediately rather than after the
                    // current (possibly 30s) tick expires.
                    let bg = self.is_app_execution_suspended();
                    if bg != last_bg_state {
                        last_bg_state = bg;
                        rescan_interval =
                            tokio::time::interval(std::time::Duration::from_millis(if bg {
                                rescan_interval_ms_bg
                            } else {
                                rescan_interval_ms_fg
                            }));
                    }
                    let connected_health_probe_ms = if bg {
                        connected_health_probe_ms_bg
                    } else {
                        connected_health_probe_ms_fg
                    };
                    let presence_republish_interval_ms = if bg {
                        presence_republish_interval_ms_bg
                    } else {
                        presence_republish_interval_ms_fg
                    };
                    let network_change_recovery_interval_ms = if bg {
                        network_change_recovery_interval_ms_bg
                    } else {
                        network_change_recovery_interval_ms_fg
                    };
                    // A discovery event can put this peer into the deterministic
                    // non-initiator grace window before the incoming transport
                    // arrives. Wake at that exact policy deadline so reciprocal
                    // admission is not accidentally coupled to a presence lease
                    // refresh or the low-frequency safety rescan.
                    let grace_wakeup_delay_ms = next_non_initiator_wakeup_delay_ms(
                        &non_initiator_wait_started_at,
                        &non_initiator_escalation_count,
                        now_millis_i64(),
                    );
                    let grace_wakeup = tokio::time::sleep(std::time::Duration::from_millis(
                        grace_wakeup_delay_ms.unwrap_or(rescan_interval_ms_fg),
                    ));
                    tokio::pin!(grace_wakeup);
                    tokio::select! {
                        _ = self.native_auto_connect_wake.notified() => {
                            if self.is_app_execution_suspended() {
                                continue;
                            }
                            // Admission proof changes are typed lifecycle
                            // input even when the provider roster and physical
                            // transport are unchanged. Reconcile through this
                            // actor's normal admission/tie-break path now.
                            self.run_auto_connect_snapshot(
                                &user_id,
                                &local_device_id,
                                &mut last_attempt_at,
                                &mut failure_count,
                                &mut failure_backoff_until,
                                retry_throttle_ms,
                                &mut non_initiator_wait_started_at,
                                &mut non_initiator_escalation_count,
                                initiator_grace_ms,
                                &mut skip_log_at,
                                &mut connected_health_checked_at,
                                &mut connected_health_state,
                                &mut connected_health_failures,
                                connected_health_probe_ms,
                                connected_health_failure_threshold,
                                &mut last_presence_republish_at_ms,
                                presence_republish_interval_ms,
                                &mut last_network_change_at_ms,
                                network_change_recovery_interval_ms,
                                network_change_failure_threshold,
                                &mut last_snapshot_signature,
                                &mut last_snapshot_log_at,
                                &mut last_known_node_id,
                                &mut known_devices,
                            ).await;
                        }
                        _ = rescan_interval.tick() => {
                            if self.is_app_execution_suspended() {
                                continue;
                            }
                            self.run_auto_connect_snapshot(
                                &user_id,
                                &local_device_id,
                                &mut last_attempt_at,
                                &mut failure_count,
                                &mut failure_backoff_until,
                                retry_throttle_ms,
                                &mut non_initiator_wait_started_at,
                                &mut non_initiator_escalation_count,
                                initiator_grace_ms,
                                &mut skip_log_at,
                                &mut connected_health_checked_at,
                                &mut connected_health_state,
                                &mut connected_health_failures,
                                connected_health_probe_ms,
                                connected_health_failure_threshold,
                                &mut last_presence_republish_at_ms,
                                presence_republish_interval_ms,
                                &mut last_network_change_at_ms,
                                network_change_recovery_interval_ms,
                                network_change_failure_threshold,
                                &mut last_snapshot_signature,
                                &mut last_snapshot_log_at,
                                &mut last_known_node_id,
                                &mut known_devices,
                            ).await;
                        }
                        _ = health_interval.tick() => {
                            if self.is_app_execution_suspended() {
                                continue;
                            }

                            // Probe cached provider records, not the provider
                            // itself. Presence is coordination evidence; when a
                            // cached record has gone offline, an already-owned
                            // physical transport still receives positive
                            // liveness probes until that exact generation is
                            // retired or proves responsive.
                            let cached_devices: Vec<crate::signaling::Device> =
                                known_devices.values().cloned().collect();
                            for mut device in cached_devices {
                                if !device.online {
                                    let Some(ticket) = device.ticket.as_deref() else {
                                        continue;
                                    };
                                    let (iroh_ticket, _) =
                                        crate::session_token::split_ticket(ticket.trim());
                                    let Ok(endpoint_addr) = parse_endpoint_ticket(iroh_ticket) else {
                                        continue;
                                    };
                                    if !self.is_connected(endpoint_addr.id).await {
                                        continue;
                                    }
                                    device.online = true;
                                }
                                self.try_auto_connect_device(
                                    &user_id,
                                    device,
                                    &local_device_id,
                                    &mut last_attempt_at,
                                    &mut failure_count,
                                    &mut failure_backoff_until,
                                    retry_throttle_ms,
                                    &mut non_initiator_wait_started_at,
                                    &mut non_initiator_escalation_count,
                                    initiator_grace_ms,
                                    &mut skip_log_at,
                                    &mut connected_health_checked_at,
                                    &mut connected_health_state,
                                    &mut connected_health_failures,
                                    connected_health_probe_ms_fg,
                                    connected_health_failure_threshold,
                                    &mut last_presence_republish_at_ms,
                                    presence_republish_interval_ms_fg,
                                    &mut last_network_change_at_ms,
                                    network_change_recovery_interval_ms_fg,
                                    network_change_failure_threshold,
                                    &mut last_known_node_id,
                                )
                                .await;
                            }
                        }
                        _ = &mut grace_wakeup, if grace_wakeup_delay_ms.is_some() => {
                            if self.is_app_execution_suspended() {
                                continue;
                            }
                            self.run_auto_connect_snapshot(
                                &user_id,
                                &local_device_id,
                                &mut last_attempt_at,
                                &mut failure_count,
                                &mut failure_backoff_until,
                                retry_throttle_ms,
                                &mut non_initiator_wait_started_at,
                                &mut non_initiator_escalation_count,
                                initiator_grace_ms,
                                &mut skip_log_at,
                                &mut connected_health_checked_at,
                                &mut connected_health_state,
                                &mut connected_health_failures,
                                connected_health_probe_ms,
                                connected_health_failure_threshold,
                                &mut last_presence_republish_at_ms,
                                presence_republish_interval_ms,
                                &mut last_network_change_at_ms,
                                network_change_recovery_interval_ms,
                                network_change_failure_threshold,
                                &mut last_snapshot_signature,
                                &mut last_snapshot_log_at,
                                &mut last_known_node_id,
                                &mut known_devices,
                            ).await;
                        }
                        result = stream.next() => {
                            let Some(result) = result else { break; };
                            let events = match result {
                                Ok(e) => e,
                                Err(error) => {
                                    eprintln!(
                                        "[pluto-rtc][auto-connect] discovery subscription ended; recreating owner-managed stream: {}",
                                        error
                                    );
                                    break;
                                }
                            };

                            for event in events {
                                match event {
                                    crate::signaling::DeviceEvent::Added { device }
                                    | crate::signaling::DeviceEvent::Modified { device } => {
                                        known_devices
                                            .insert(device.device_id.clone(), device.clone());
                                        self.try_auto_connect_device(
                                            &user_id,
                                            device,
                                            &local_device_id,
                                            &mut last_attempt_at,
                                            &mut failure_count,
                                            &mut failure_backoff_until,
                                            retry_throttle_ms,
                                            &mut non_initiator_wait_started_at,
                                            &mut non_initiator_escalation_count,
                                            initiator_grace_ms,
                                            &mut skip_log_at,
                                            &mut connected_health_checked_at,
                                            &mut connected_health_state,
                                            &mut connected_health_failures,
                                            connected_health_probe_ms_fg,
                                            connected_health_failure_threshold,
                                            &mut last_presence_republish_at_ms,
                                            presence_republish_interval_ms_fg,
                                            &mut last_network_change_at_ms,
                                            network_change_recovery_interval_ms_fg,
                                            network_change_failure_threshold,
                                            &mut last_known_node_id,
                                        )
                                        .await;
                                    }
                                    crate::signaling::DeviceEvent::Removed { device_id } => {
                                        known_devices.remove(&device_id);
                                        last_attempt_at.remove(&device_id);
                                        non_initiator_wait_started_at.remove(&device_id);
                                        non_initiator_escalation_count.remove(&device_id);
                                        skip_log_at.retain(|key, _| !key.starts_with(&format!("{}::", device_id)));
                                        connected_health_checked_at.remove(&device_id);
                                        connected_health_state.remove(&device_id);
                                        connected_health_failures.remove(&device_id);
                                        last_known_node_id
                                            .remove(&format!("{device_id}::runtime-instance"));
                                        last_known_node_id
                                            .remove(&format!("{device_id}::ticket"));
                                        clear_auto_connect_failure_state(
                                            &device_id,
                                            &mut failure_count,
                                            &mut failure_backoff_until,
                                        );

                                        for record in self.connection_manager.get_by_device_id(&device_id).await {
                                            if matches!(
                                                record.state,
                                                crate::connection_manager::ConnectionState::Pending
                                                    | crate::connection_manager::ConnectionState::Connecting
                                            ) {
                                                self.connection_manager
                                                    .set_failed(
                                                        &record.connection_id,
                                                        Some("device removed from signaling snapshot".to_string()),
                                                    )
                                                    .await;
                                            }
                                        }
                                    }
                                }
                            }
                        }
                    }
                }
            }

            #[cfg(target_arch = "wasm32")]
            {
                let mut stream = stream.fuse();

                loop {
                    if !self.is_auto_connect_generation_current(generation) {
                        break;
                    }
                    let bg = self.is_app_execution_suspended();
                    let connected_health_probe_ms = if bg {
                        connected_health_probe_ms_bg
                    } else {
                        connected_health_probe_ms_fg
                    };
                    let presence_republish_interval_ms = if bg {
                        presence_republish_interval_ms_bg
                    } else {
                        presence_republish_interval_ms_fg
                    };
                    let network_change_recovery_interval_ms = if bg {
                        network_change_recovery_interval_ms_bg
                    } else {
                        network_change_recovery_interval_ms_fg
                    };
                    let rescan_ms = if bg {
                        rescan_interval_ms_bg
                    } else {
                        rescan_interval_ms_fg
                    };
                    let sleep =
                        gloo_timers::future::sleep(std::time::Duration::from_millis(rescan_ms))
                            .fuse();
                    futures::pin_mut!(sleep);
                    futures::select! {
                        _ = sleep => {
                            if self.is_app_execution_suspended() {
                                continue;
                            }
                            self.run_auto_connect_snapshot(
                                &user_id,
                                &local_device_id,
                                &mut last_attempt_at,
                                &mut failure_count,
                                &mut failure_backoff_until,
                                retry_throttle_ms,
                                &mut non_initiator_wait_started_at,
                                &mut non_initiator_escalation_count,
                                initiator_grace_ms,
                                &mut skip_log_at,
                                &mut connected_health_checked_at,
                                &mut connected_health_state,
                                &mut connected_health_failures,
                                connected_health_probe_ms,
                                connected_health_failure_threshold,
                                &mut last_presence_republish_at_ms,
                                presence_republish_interval_ms,
                                &mut last_network_change_at_ms,
                                network_change_recovery_interval_ms,
                                network_change_failure_threshold,
                                &mut last_snapshot_signature,
                                &mut last_snapshot_log_at,
                                &mut last_known_node_id,
                                &mut known_devices,
                            ).await;
                        }
                        result = stream.next() => {
                            let Some(result) = result else { break; };
                            let events = match result {
                                Ok(e) => e,
                                Err(error) => {
                                    web_sys::console::error_1(&wasm_bindgen::JsValue::from_str(
                                        &format!(
                                            "[pluto-rtc][auto-connect] discovery subscription ended; recreating owner-managed stream: {}",
                                            error
                                        ),
                                    ));
                                    break;
                                }
                            };

                            for event in events {
                                match event {
                                    crate::signaling::DeviceEvent::Added { device }
                                    | crate::signaling::DeviceEvent::Modified { device } => {
                                        // Event-driven connects always use foreground parameters.
                                        self.try_auto_connect_device(
                                            &user_id,
                                            device,
                                            &local_device_id,
                                            &mut last_attempt_at,
                                            &mut failure_count,
                                            &mut failure_backoff_until,
                                            retry_throttle_ms,
                                            &mut non_initiator_wait_started_at,
                                            &mut non_initiator_escalation_count,
                                            initiator_grace_ms,
                                            &mut skip_log_at,
                                            &mut connected_health_checked_at,
                                            &mut connected_health_state,
                                            &mut connected_health_failures,
                                            connected_health_probe_ms_fg,
                                            connected_health_failure_threshold,
                                            &mut last_presence_republish_at_ms,
                                            presence_republish_interval_ms_fg,
                                            &mut last_network_change_at_ms,
                                            network_change_recovery_interval_ms_fg,
                                            network_change_failure_threshold,
                                            &mut last_known_node_id,
                                        )
                                        .await;
                                    }
                                    crate::signaling::DeviceEvent::Removed { device_id } => {
                                        last_attempt_at.remove(&device_id);
                                        non_initiator_wait_started_at.remove(&device_id);
                                        non_initiator_escalation_count.remove(&device_id);
                                        skip_log_at.retain(|key, _| !key.starts_with(&format!("{}::", device_id)));
                                        connected_health_checked_at.remove(&device_id);
                                        connected_health_state.remove(&device_id);
                                        connected_health_failures.remove(&device_id);
                                        last_known_node_id
                                            .remove(&format!("{device_id}::runtime-instance"));
                                        last_known_node_id
                                            .remove(&format!("{device_id}::ticket"));
                                        clear_auto_connect_failure_state(
                                            &device_id,
                                            &mut failure_count,
                                            &mut failure_backoff_until,
                                        );

                                        for record in self.connection_manager.get_by_device_id(&device_id).await {
                                            if matches!(
                                                record.state,
                                                crate::connection_manager::ConnectionState::Pending
                                                    | crate::connection_manager::ConnectionState::Connecting
                                            ) {
                                                self.connection_manager
                                                    .set_failed(
                                                        &record.connection_id,
                                                        Some("device removed from signaling snapshot".to_string()),
                                                    )
                                                    .await;
                                            }
                                        }
                                    }
                                }
                            }
                        }
                    }
                }
            }

            #[cfg(not(target_arch = "wasm32"))]
            eprintln!(
                "[pluto-rtc][auto-connect] device stream ended user_id={} local_device_id={}, resubscribing",
                user_id, local_device_id
            );
            #[cfg(target_arch = "wasm32")]
            web_sys::console::log_1(&wasm_bindgen::JsValue::from_str(&format!(
                "[pluto-rtc][auto-connect] device stream ended user_id={} local_device_id={}, resubscribing",
                user_id, local_device_id
            )));

            #[cfg(not(target_arch = "wasm32"))]
            tokio::time::sleep(std::time::Duration::from_millis(400)).await;
            #[cfg(target_arch = "wasm32")]
            gloo_timers::future::sleep(std::time::Duration::from_millis(400)).await;
        }
    }
}