Skip to main content

openrtc/client/
auto_connect_impl.rs

1//! Delegated `Client` auto-connect and transport-health internals.
2//!
3//! This module owns retry/backoff/tie-break policy and snapshot/event reconciliation.
4
5#[cfg(test)]
6use super::state_signaling_impl::should_present_session_token_on_connected_transport;
7use super::*;
8
9#[cfg(test)]
10pub(crate) fn should_auto_present_session_token(
11    has_extracted_token: bool,
12    has_current_remote_admission_proof: bool,
13) -> bool {
14    should_present_session_token_on_connected_transport(
15        has_extracted_token,
16        has_current_remote_admission_proof,
17    )
18}
19
20#[derive(Debug, Clone, Copy, PartialEq, Eq)]
21#[cfg(not(target_arch = "wasm32"))]
22pub(crate) struct SessionAdmissionPresentationDecision {
23    pub(crate) should_present: bool,
24    pub(crate) request_reciprocal: bool,
25}
26
27/// Decide how the sole auto-connect actor repairs directional admission.
28///
29/// A current outbound proof normally suppresses duplicate presentation. The
30/// exception is a current-generation route missing its inbound proof: this side
31/// sends a valid typed request and asks the peer to present its own token in the
32/// same pass. A validated peer request likewise forces one bounded presentation
33/// through this same actor.
34#[cfg(not(target_arch = "wasm32"))]
35pub(crate) fn session_admission_presentation_decision(
36    has_extracted_token: bool,
37    has_current_remote_admission_proof: bool,
38    has_current_inbound_admission_proof: bool,
39    peer_requested_reciprocal: bool,
40    is_designated_initiator: bool,
41) -> SessionAdmissionPresentationDecision {
42    if !has_extracted_token {
43        return SessionAdmissionPresentationDecision {
44            should_present: false,
45            request_reciprocal: false,
46        };
47    }
48
49    // A replacement route commonly starts with neither directional proof. Ask
50    // for the reverse presentation on the first valid request instead of
51    // waiting for a second health-loop pass after remote approval. A validated
52    // peer request already arrived on the peer's presentation stream, so echoing
53    // another reciprocal request before its ACK-bound proof becomes visible
54    // creates a request ping-pong. Present once to satisfy that request without
55    // bouncing it back.
56    let bilateral_proof_is_current =
57        has_current_remote_admission_proof && has_current_inbound_admission_proof;
58    if bilateral_proof_is_current {
59        return SessionAdmissionPresentationDecision {
60            should_present: false,
61            request_reciprocal: false,
62        };
63    }
64
65    // A fresh physical leg commonly becomes visible to both auto-connect
66    // actors before either directional proof exists. Only the deterministic
67    // initiator may open the first persistent admission stream; otherwise both
68    // peers install different canonical control streams and each closes the
69    // route selected by the other. The observing peer answers that one stream
70    // inline. If it could not do so, the validated request below authorizes one
71    // bounded follow-up presentation without another reciprocal request.
72    if !has_current_remote_admission_proof && !has_current_inbound_admission_proof {
73        return SessionAdmissionPresentationDecision {
74            should_present: is_designated_initiator,
75            request_reciprocal: is_designated_initiator,
76        };
77    }
78
79    if peer_requested_reciprocal {
80        return SessionAdmissionPresentationDecision {
81            should_present: true,
82            request_reciprocal: false,
83        };
84    }
85
86    // An inbound proof means the peer's presentation and ACK exchange has
87    // already completed, so a one-way follow-up cannot collide with that
88    // stream. A presenter that is still missing its inbound proof owns the
89    // reciprocal request only when it is the deterministic initiator.
90    let request_reciprocal = has_current_remote_admission_proof && is_designated_initiator;
91    SessionAdmissionPresentationDecision {
92        should_present: !has_current_remote_admission_proof || request_reciprocal,
93        request_reciprocal,
94    }
95}
96
97/// Return whether the persistent `user-device` actor may inspect or repair the
98/// current physical transport.
99///
100/// A deterministic node-pair connection can be reused by an explicit
101/// capability such as `drive-grant:*`, `room:*`, or `space:*`. That capability
102/// owns its admission/security epoch while active. The background user-device
103/// actor must yield before probing, presenting its cached ticket, or retiring
104/// that transport; otherwise it can replace the explicit capability's
105/// semantic security epoch and strand product traffic.
106///
107/// An in-flight presentation is also exclusive. The registry presentation
108/// guard resolves the narrow race where an explicit actor starts admission
109/// between discovery snapshots.
110#[cfg(not(target_arch = "wasm32"))]
111pub(crate) fn persistent_managed_actor_may_own_transport(
112    current_scopes: &[String],
113    admitted_scope: Option<&str>,
114    presentation_in_flight: bool,
115) -> bool {
116    if presentation_in_flight {
117        return false;
118    }
119
120    current_scopes
121        .iter()
122        .map(String::as_str)
123        .chain(admitted_scope)
124        .map(str::trim)
125        .filter(|scope| !scope.is_empty())
126        .all(|scope| scope == super::core_impl::PERSISTENT_MANAGED_ADMISSION_SCOPE)
127}
128
129#[cfg(not(target_arch = "wasm32"))]
130#[derive(Debug, Clone, Copy, PartialEq, Eq)]
131pub(crate) enum ActiveConnectionProbeDecision {
132    Healthy,
133    PreserveLiveTransport,
134    Failed,
135}
136
137#[cfg(not(target_arch = "wasm32"))]
138#[derive(Debug, Clone, Copy, PartialEq, Eq)]
139enum FailedProbeRetirementOutcome {
140    Retired,
141    DeferredAdmissionInFlight,
142    CancelledCurrentGenerationActivity,
143    GenerationChanged,
144}
145
146const ACTIVE_CONNECTION_PROBE_TIMEOUT_MS: i64 = 4_000;
147
148/// Retire shared-endpoint connections from the superseded remote process while
149/// preserving the exact Iroh connection that owns OpenRTC control traffic.
150///
151/// OpenRTC's `transport_generation` is a process-lifetime logical fence. Iroh's
152/// `Connection::stable_id()` is a separate runtime identifier used by the
153/// endpoint's connection registry. Keeping this adapter typed to a concrete
154/// connection prevents those two identifiers from being mixed.
155#[cfg(all(
156    not(target_arch = "wasm32"),
157    openrtc_iroh_retire_remote_connections_api
158))]
159pub(crate) async fn retire_stale_endpoint_connections(
160    endpoint: &iroh::Endpoint,
161    endpoint_id: iroh::EndpointId,
162    current_transport: &iroh::endpoint::Connection,
163    reason: &[u8],
164) -> usize {
165    endpoint
166        .retire_remote_connections_except(endpoint_id, current_transport.stable_id(), reason)
167        .await
168}
169
170pub(crate) fn generation_activity_is_recent(
171    last_inbound_activity_age: Option<std::time::Duration>,
172    probe_interval_ms: i64,
173) -> bool {
174    let activity_window_ms = probe_interval_ms
175        .max(0)
176        .saturating_add(ACTIVE_CONNECTION_PROBE_TIMEOUT_MS) as u64;
177    last_inbound_activity_age
178        .is_some_and(|age| age <= std::time::Duration::from_millis(activity_window_ms))
179}
180
181#[cfg(not(target_arch = "wasm32"))]
182impl Client {
183    /// Wake whichever native desired-peer adapter currently feeds the one
184    /// Rust auto-connect owner. Gateway-integrated hosts use the provider
185    /// subscription loop; IPC-only hosts use the external revisioned mailbox.
186    pub(crate) async fn wake_native_auto_connect_owner(&self) -> bool {
187        if self.external_auto_connect_is_active().await {
188            return self.wake_native_external_auto_connect().await;
189        }
190
191        let provider_actor_active = self
192            .auto_connect_loop_key
193            .lock()
194            .unwrap_or_else(|poisoned| poisoned.into_inner())
195            .is_some();
196        if provider_actor_active {
197            self.native_auto_connect_wake.notify_one();
198        }
199        provider_actor_active
200    }
201
202    /// Record valid peer protocol activity for the exact physical generation
203    /// that carried it. Callers must only invoke this after parsing a complete
204    /// OpenRTC control frame or after the Rust admission router has classified
205    /// a current, admitted application stream. Unclassified raw bytes and
206    /// failed authentication are not liveness evidence.
207    pub(crate) fn record_native_transport_protocol_activity(
208        &self,
209        connection_id: &str,
210        transport_stable_id: u64,
211    ) {
212        if let Ok(mut activity) = self.native_transport_protocol_activity.write() {
213            activity.insert(
214                connection_id.to_string(),
215                super::NativeTransportProtocolActivity {
216                    transport_stable_id,
217                    observed_at: std::time::Instant::now(),
218                },
219            );
220        }
221    }
222
223    pub(crate) fn native_transport_protocol_activity_age(
224        &self,
225        connection_id: &str,
226        expected_transport_stable_id: u64,
227    ) -> Option<std::time::Duration> {
228        let observed_at = self
229            .native_transport_protocol_activity
230            .read()
231            .ok()
232            .and_then(|activity| activity.get(connection_id).copied())
233            .filter(|activity| activity.transport_stable_id == expected_transport_stable_id)?
234            .observed_at;
235        Some(std::time::Instant::now().saturating_duration_since(observed_at))
236    }
237
238    pub(crate) fn forget_native_transport_protocol_activity_for_transport(
239        &self,
240        connection_id: &str,
241        expected_transport_stable_id: u64,
242    ) -> bool {
243        self.native_transport_protocol_activity
244            .write()
245            .ok()
246            .is_some_and(|mut activity| {
247                if activity
248                    .get(connection_id)
249                    .is_some_and(|entry| entry.transport_stable_id == expected_transport_stable_id)
250                {
251                    activity.remove(connection_id);
252                    true
253                } else {
254                    false
255                }
256            })
257    }
258
259    pub(crate) fn forget_native_transport_protocol_activity(&self, connection_id: &str) {
260        if let Ok(mut activity) = self.native_transport_protocol_activity.write() {
261            activity.remove(connection_id);
262        }
263    }
264
265    /// Retire a physical transport after a failed liveness probe without
266    /// racing a valid admission exchange on that same generation.
267    ///
268    /// Every native failed-probe caller must converge here. The retirement
269    /// guard serializes teardown with response writers and outbound token
270    /// presenters, then the exact generation and its protocol activity are
271    /// re-read after the guard is acquired.
272    async fn retire_transport_generation_after_failed_probe(
273        &self,
274        connection_id: &str,
275        endpoint_id: iroh::EndpointId,
276        expected_transport_stable_id: u64,
277        activity_window_ms: i64,
278        reason: &str,
279    ) -> FailedProbeRetirementOutcome {
280        let Some(retirement_guard) = self
281            .session_token_registry
282            .try_begin_transport_retirement(connection_id)
283        else {
284            return FailedProbeRetirementOutcome::DeferredAdmissionInFlight;
285        };
286
287        let current_transport_stable_id = self
288            .get_connection(endpoint_id)
289            .await
290            .map(|connection| crate::transport_generation::for_connection(&connection));
291        if current_transport_stable_id != Some(expected_transport_stable_id) {
292            drop(retirement_guard);
293            return FailedProbeRetirementOutcome::GenerationChanged;
294        }
295
296        if generation_activity_is_recent(
297            self.native_transport_protocol_activity_age(
298                connection_id,
299                expected_transport_stable_id,
300            ),
301            activity_window_ms,
302        ) {
303            drop(retirement_guard);
304            return FailedProbeRetirementOutcome::CancelledCurrentGenerationActivity;
305        }
306
307        let retired = self
308            .disconnect_transport_generation_with_reason(
309                endpoint_id,
310                expected_transport_stable_id,
311                reason,
312            )
313            .await
314            .unwrap_or(false);
315        drop(retirement_guard);
316
317        if retired {
318            FailedProbeRetirementOutcome::Retired
319        } else {
320            FailedProbeRetirementOutcome::GenerationChanged
321        }
322    }
323}
324
325#[cfg(target_arch = "wasm32")]
326impl Client {
327    pub(crate) fn native_transport_protocol_activity_age(
328        &self,
329        _connection_id: &str,
330        _expected_transport_stable_id: u64,
331    ) -> Option<std::time::Duration> {
332        None
333    }
334}
335
336#[cfg(not(target_arch = "wasm32"))]
337pub(crate) fn active_connection_probe_decision(
338    active_probe_healthy: bool,
339    passive_transport_alive: bool,
340) -> ActiveConnectionProbeDecision {
341    if active_probe_healthy {
342        ActiveConnectionProbeDecision::Healthy
343    } else if passive_transport_alive {
344        ActiveConnectionProbeDecision::PreserveLiveTransport
345    } else {
346        ActiveConnectionProbeDecision::Failed
347    }
348}
349
350#[cfg(not(target_arch = "wasm32"))]
351pub(crate) fn active_connection_probe_decision_with_generation_activity(
352    active_probe_healthy: bool,
353    passive_transport_alive: bool,
354    recent_generation_activity: bool,
355) -> ActiveConnectionProbeDecision {
356    if recent_generation_activity {
357        ActiveConnectionProbeDecision::Healthy
358    } else {
359        active_connection_probe_decision(active_probe_healthy, passive_transport_alive)
360    }
361}
362
363#[cfg(not(target_arch = "wasm32"))]
364pub(crate) fn next_active_connection_probe_failure_count(
365    decision: ActiveConnectionProbeDecision,
366    current_failures: u8,
367) -> u8 {
368    if matches!(decision, ActiveConnectionProbeDecision::Healthy) {
369        0
370    } else {
371        current_failures.saturating_add(1).min(10)
372    }
373}
374
375#[cfg(not(target_arch = "wasm32"))]
376pub(crate) fn active_connection_probe_is_healthy(
377    decision: ActiveConnectionProbeDecision,
378    failure_count: u8,
379    failure_threshold: u8,
380) -> bool {
381    matches!(decision, ActiveConnectionProbeDecision::Healthy)
382        || (matches!(
383            decision,
384            ActiveConnectionProbeDecision::PreserveLiveTransport
385        ) && failure_count < failure_threshold)
386}
387
388/// A session-token response proves authorization only when it is received and
389/// parsed. Losing the physical Iroh leg before that response is transport
390/// churn, not an authorization denial. The owner of auto-connect must preserve
391/// the logical peer session in that case so the canonical replacement leg can
392/// retry admission after backoff.
393#[derive(Debug, Clone, Copy, PartialEq, Eq)]
394pub(crate) enum SessionTokenPresentationFailureAction {
395    RetryOnNextTransport,
396    RejectAndDisconnect,
397}
398
399#[derive(Debug, Clone, Copy, PartialEq, Eq)]
400#[cfg(not(target_arch = "wasm32"))]
401pub(crate) enum RetryableSessionAdmissionFailureDecision {
402    PreserveCurrentTransport,
403    RetireCurrentTransport,
404    ReplacementAlreadyWon,
405}
406
407/// A superseded desired-peer pass may retire a connection only when the same
408/// actor accepted a newer authoritative roster that actually withdrew that
409/// durable device. A revision bump caused by another peer must not close the
410/// still-desired peer's shared physical transport.
411pub(crate) fn superseded_desired_peer_was_withdrawn(
412    actor_is_current: bool,
413    current_revision: u64,
414    attempt_revision: u64,
415    peer_is_still_desired: bool,
416) -> bool {
417    actor_is_current && current_revision > attempt_revision && !peer_is_still_desired
418}
419
420/// Bound admission repair to the physical generation that failed.
421///
422/// A single response loss is normal during crossed reconnects, so the current
423/// transport gets one retry. Repeated failures mean the passive Iroh handle is
424/// no longer useful for application admission even when QUIC has not reported
425/// closed. Only the generation that produced the failure may be retired; a
426/// replacement that won while the request was in flight remains authoritative.
427#[cfg(not(target_arch = "wasm32"))]
428pub(crate) fn retryable_session_admission_failure_decision(
429    failure_count: u8,
430    failure_threshold: u8,
431    attempted_transport_stable_id: Option<u64>,
432    current_transport_stable_id: Option<u64>,
433    remote_runtime_instance_changed: bool,
434) -> RetryableSessionAdmissionFailureDecision {
435    if attempted_transport_stable_id != current_transport_stable_id
436        || attempted_transport_stable_id.is_none()
437    {
438        return RetryableSessionAdmissionFailureDecision::ReplacementAlreadyWon;
439    }
440
441    // A new remote runtime instance with the same durable Iroh identity can
442    // leave the old process's QUIC handle looking passively alive. If that
443    // exact generation cannot obtain an admission response from the newly
444    // advertised runtime, preserving it for the ordinary second attempt only
445    // sends the retry to the retired process. The runtime-instance observation
446    // is coordination evidence, while the failed action-bound response is the
447    // lifecycle input that makes this generation unusable.
448    if !remote_runtime_instance_changed && failure_count < failure_threshold.max(1) {
449        RetryableSessionAdmissionFailureDecision::PreserveCurrentTransport
450    } else {
451        RetryableSessionAdmissionFailureDecision::RetireCurrentTransport
452    }
453}
454
455pub(crate) fn session_token_presentation_failure_action(
456    error: &str,
457) -> SessionTokenPresentationFailureAction {
458    let error = error.to_ascii_lowercase();
459    if error.contains("connection lost")
460        || error.contains("[session-token-response:timeout]")
461        || error.contains("stream finished early")
462        || error.contains("peer stream ended before")
463        || error.contains("connection closed")
464        || error.contains("stream reset")
465        || error.contains("stream stopped")
466        || error.contains("failed to open bi-stream")
467        || error.contains("failed to write session-token presentation")
468        || error.contains("missing connection for session-token presentation")
469        || error.contains("[session-token-presentation:in-flight]")
470        || error.contains("authoritative-device-binding-pending")
471        || error.contains(crate::session_token::PEER_ADMISSION_LEASE_EXPIRED)
472    {
473        SessionTokenPresentationFailureAction::RetryOnNextTransport
474    } else {
475        SessionTokenPresentationFailureAction::RejectAndDisconnect
476    }
477}
478
479#[cfg(not(target_arch = "wasm32"))]
480pub(crate) fn session_token_presentation_was_duplicate(error: &str) -> bool {
481    error
482        .to_ascii_lowercase()
483        .contains("[session-token-presentation:in-flight]")
484}
485
486fn remote_excludes_local_device(excluded_peers: &[String], local_device_id: &str) -> bool {
487    let local_device_id = local_device_id.trim();
488    !local_device_id.is_empty()
489        && excluded_peers
490            .iter()
491            .any(|peer| peer.trim().eq_ignore_ascii_case(local_device_id))
492}
493
494#[cfg(not(target_arch = "wasm32"))]
495fn observe_remote_runtime_instance(
496    last_known: &mut std::collections::HashMap<String, String>,
497    device_id: &str,
498    runtime_instance_id: Option<&str>,
499) -> bool {
500    let Some(runtime_instance_id) = runtime_instance_id
501        .map(str::trim)
502        .filter(|value| !value.is_empty() && *value != device_id)
503    else {
504        return false;
505    };
506    let key = format!("{device_id}::runtime-instance");
507    last_known
508        .insert(key, runtime_instance_id.to_string())
509        .is_some_and(|previous| previous != runtime_instance_id)
510}
511
512#[cfg(not(target_arch = "wasm32"))]
513fn runtime_instance_change_requires_replacement(
514    runtime_instance_changed: bool,
515    active_transport_responsive: bool,
516) -> bool {
517    runtime_instance_changed && !active_transport_responsive
518}
519
520#[cfg(not(target_arch = "wasm32"))]
521fn observe_remote_ticket(
522    last_known: &mut std::collections::HashMap<String, String>,
523    device_id: &str,
524    ticket: &str,
525) -> bool {
526    let ticket = ticket.trim();
527    if ticket.is_empty() {
528        return false;
529    }
530    let key = format!("{device_id}::ticket");
531    let fingerprint = crate::session_token::token_fingerprint(ticket);
532    last_known
533        .insert(key, fingerprint.clone())
534        .is_some_and(|previous| previous != fingerprint)
535}
536
537#[cfg(not(target_arch = "wasm32"))]
538fn reconcile_remote_ticket_retry_state(
539    last_known: &mut std::collections::HashMap<String, String>,
540    device_id: &str,
541    ticket: &str,
542    last_attempt_at: &mut std::collections::HashMap<String, i64>,
543    failure_count: &mut std::collections::HashMap<String, u8>,
544    failure_backoff_until: &mut std::collections::HashMap<String, i64>,
545) -> bool {
546    if !observe_remote_ticket(last_known, device_id, ticket) {
547        return false;
548    }
549    last_attempt_at.remove(device_id);
550    clear_auto_connect_failure_state(device_id, failure_count, failure_backoff_until);
551    true
552}
553
554#[derive(Debug, Clone, PartialEq, Eq, serde::Deserialize)]
555#[serde(rename_all = "camelCase")]
556pub(crate) struct BrowserDesiredPeer {
557    pub(crate) device_id: String,
558    #[serde(default)]
559    pub(crate) node_id: Option<String>,
560    #[serde(default)]
561    pub(crate) ticket: Option<String>,
562    #[serde(default)]
563    pub(crate) online: bool,
564    #[serde(default)]
565    pub(crate) session_id: Option<String>,
566    #[serde(default)]
567    pub(crate) excluded_peers: Vec<String>,
568    #[serde(default)]
569    pub(crate) topology_role: Option<SparseTopologyRole>,
570    #[serde(default)]
571    pub(crate) topology_revision: Option<u64>,
572}
573
574#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Deserialize)]
575#[serde(rename_all = "lowercase")]
576pub(crate) enum SparseTopologyRole {
577    Active,
578    Backup,
579}
580
581#[derive(Debug, Default)]
582struct BrowserDesiredPeerMailbox {
583    revision: u64,
584    peers: std::collections::BTreeMap<String, BrowserDesiredPeer>,
585    pending_topology_revision: Option<u64>,
586    pending_ready_active_peers: std::collections::HashSet<String>,
587    settled_active: std::collections::BTreeMap<String, BrowserDesiredPeer>,
588    /// Devices whose authoritative live-route desire was explicitly withdrawn.
589    ///
590    /// Retiring the current connection is not sufficient: a revoked peer can
591    /// still race back in over a cached endpoint ticket before its own gateway
592    /// lifecycle observes the fatal close.  Keep the withdrawal beside the
593    /// revisioned roster authority so incoming transport admission can reject
594    /// that replacement generation.  A later online roster revision is the
595    /// only event that clears the fence.
596    withdrawn: std::collections::BTreeMap<String, Option<String>>,
597}
598
599impl BrowserDesiredPeerMailbox {
600    #[cfg(test)]
601    fn accept(&mut self, revision: u64, peers: Vec<BrowserDesiredPeer>) -> bool {
602        self.accept_with_withdrawn(revision, peers).is_some()
603    }
604
605    /// Accepts a full authoritative roster and returns peers whose live-route
606    /// desire was withdrawn by this revision. Persistent user avenues retain
607    /// offline devices in the roster, so both removal and an `online ->
608    /// offline` transition must retire any managed route for the old live
609    /// generation.
610    fn accept_with_withdrawn(
611        &mut self,
612        revision: u64,
613        peers: Vec<BrowserDesiredPeer>,
614    ) -> Option<Vec<BrowserDesiredPeer>> {
615        self.accept_update(revision, peers, None)
616    }
617
618    /// Apply one candidate-only reachability expiry. The exact missing peer is
619    /// removed from dial eligibility without becoming authoritatively
620    /// withdrawn; any other removal keeps the normal withdrawal semantics.
621    fn accept_with_observation_expired(
622        &mut self,
623        revision: u64,
624        peers: Vec<BrowserDesiredPeer>,
625        expired_device_id: &str,
626    ) -> Option<Vec<BrowserDesiredPeer>> {
627        let expired_device_id = expired_device_id.trim();
628        if expired_device_id.is_empty() {
629            return None;
630        }
631        self.accept_update(revision, peers, Some(expired_device_id))
632    }
633
634    fn accept_update(
635        &mut self,
636        revision: u64,
637        peers: Vec<BrowserDesiredPeer>,
638        observation_expired_device_id: Option<&str>,
639    ) -> Option<Vec<BrowserDesiredPeer>> {
640        if revision <= self.revision {
641            return None;
642        }
643        let mut peers = peers
644            .into_iter()
645            .filter_map(|mut peer| {
646                peer.device_id = peer.device_id.trim().to_string();
647                if peer.device_id.is_empty() {
648                    return None;
649                }
650                peer.node_id = peer
651                    .node_id
652                    .take()
653                    .map(|value| value.trim().to_string())
654                    .filter(|value| !value.is_empty());
655                peer.ticket = peer
656                    .ticket
657                    .take()
658                    .map(|value| value.trim().to_string())
659                    .filter(|value| !value.is_empty());
660                peer.session_id = peer
661                    .session_id
662                    .take()
663                    .map(|value| value.trim().to_string())
664                    .filter(|value| !value.is_empty());
665                peer.excluded_peers = peer
666                    .excluded_peers
667                    .into_iter()
668                    .map(|value| value.trim().to_ascii_lowercase())
669                    .filter(|value| !value.is_empty())
670                    .collect();
671                Some((peer.device_id.clone(), peer))
672            })
673            .collect::<std::collections::BTreeMap<_, _>>();
674        let topology_revision = peers
675            .values()
676            .filter_map(|peer| peer.topology_revision)
677            .max();
678        if topology_revision == self.pending_topology_revision {
679            self.pending_ready_active_peers.retain(|device_id| {
680                let Some(previous) = self.peers.get(device_id) else {
681                    return false;
682                };
683                peers.get(device_id).is_some_and(|current| {
684                    current.topology_role == Some(SparseTopologyRole::Active)
685                        && browser_peer_evidence(current) == browser_peer_evidence(previous)
686                })
687            });
688        } else {
689            self.pending_ready_active_peers.clear();
690            if topology_revision.is_none() {
691                self.settled_active.clear();
692            } else if !self.settled_active.is_empty() {
693                // A gateway lease overlaps only its immediately prior active
694                // set. Rapid revisions can make that set speculative. Keep
695                // exactly one bounded, Rust-proven settled generation instead
696                // and never accumulate intermediate generations.
697                peers.retain(|_, peer| peer.topology_role != Some(SparseTopologyRole::Backup));
698                for (device_id, settled) in self.settled_active.iter().take(4) {
699                    let still_authorized =
700                        peers.get(device_id).is_some_and(|current| current.online);
701                    let is_current_active = peers.get(device_id).is_some_and(|current| {
702                        current.topology_role == Some(SparseTopologyRole::Active)
703                    });
704                    if still_authorized && !is_current_active {
705                        let mut overlap = settled.clone();
706                        overlap.topology_role = Some(SparseTopologyRole::Backup);
707                        overlap.topology_revision = topology_revision;
708                        peers.insert(device_id.clone(), overlap);
709                    }
710                }
711            }
712        }
713        let active_ids = peers
714            .values()
715            .filter(|peer| peer.topology_role == Some(SparseTopologyRole::Active))
716            .map(|peer| peer.device_id.clone())
717            .collect::<std::collections::HashSet<_>>();
718        if !active_ids.is_empty() && active_ids.is_subset(&self.pending_ready_active_peers) {
719            peers.retain(|_, peer| peer.topology_role != Some(SparseTopologyRole::Backup));
720        }
721        let withdrawn = self
722            .peers
723            .values()
724            .filter(|peer| {
725                peer.online
726                    && observation_expired_device_id != Some(peer.device_id.as_str())
727                    && peers
728                        .get(&peer.device_id)
729                        .is_none_or(|current| !current.online)
730            })
731            .cloned()
732            .collect::<Vec<_>>();
733        for peer in &withdrawn {
734            self.withdrawn
735                .insert(peer.device_id.clone(), peer.node_id.clone());
736        }
737        for peer in peers.values() {
738            if peer.online {
739                self.withdrawn.remove(&peer.device_id);
740            }
741        }
742        self.revision = revision;
743        self.pending_topology_revision = topology_revision;
744        self.peers = peers;
745        Some(withdrawn)
746    }
747
748    /// Record the existing desired-peer owner's settled transport+admission
749    /// result. Once every active edge for this exact topology is ready, remove
750    /// overlap backups from the same authoritative mailbox and return them to
751    /// the caller for physical retirement.
752    fn acknowledge_topology_peer_ready(
753        &mut self,
754        device_id: &str,
755        topology_revision: Option<u64>,
756    ) -> Vec<BrowserDesiredPeer> {
757        let Some(topology_revision) = topology_revision else {
758            return Vec::new();
759        };
760        if self.pending_topology_revision != Some(topology_revision)
761            || !self.peers.get(device_id).is_some_and(|peer| {
762                peer.topology_revision == Some(topology_revision)
763                    && peer.topology_role == Some(SparseTopologyRole::Active)
764            })
765        {
766            return Vec::new();
767        }
768        self.pending_ready_active_peers
769            .insert(device_id.to_string());
770        let active_ids = self
771            .peers
772            .values()
773            .filter(|peer| peer.topology_role == Some(SparseTopologyRole::Active))
774            .map(|peer| peer.device_id.clone())
775            .collect::<std::collections::HashSet<_>>();
776        if active_ids.is_empty() || !active_ids.is_subset(&self.pending_ready_active_peers) {
777            return Vec::new();
778        }
779        self.settled_active = self
780            .peers
781            .iter()
782            .filter(|(_, peer)| peer.topology_role == Some(SparseTopologyRole::Active))
783            .map(|(device_id, peer)| (device_id.clone(), peer.clone()))
784            .take(4)
785            .collect();
786        let backups = self
787            .peers
788            .values()
789            .filter(|peer| peer.topology_role == Some(SparseTopologyRole::Backup))
790            .cloned()
791            .collect::<Vec<_>>();
792        for peer in &backups {
793            self.peers.remove(&peer.device_id);
794            self.withdrawn
795                .insert(peer.device_id.clone(), peer.node_id.clone());
796        }
797        backups
798    }
799
800    fn observe_topology_peer_not_ready(&mut self, device_id: &str, topology_revision: Option<u64>) {
801        if self.pending_topology_revision == topology_revision
802            && self.peers.get(device_id).is_some_and(|peer| {
803                peer.topology_revision == topology_revision
804                    && peer.topology_role == Some(SparseTopologyRole::Active)
805            })
806        {
807            self.pending_ready_active_peers.remove(device_id);
808        }
809    }
810
811    fn device_is_withdrawn(&self, device_id: &str) -> bool {
812        self.withdrawn.contains_key(device_id.trim())
813    }
814
815    fn withdrawn_device_for_node(&self, node_id: &str) -> Option<String> {
816        let node_id = node_id.trim();
817        if node_id.is_empty() {
818            return None;
819        }
820        self.withdrawn
821            .iter()
822            .find_map(|(device_id, withdrawn_node_id)| {
823                (withdrawn_node_id.as_deref() == Some(node_id)).then(|| device_id.clone())
824            })
825    }
826
827    fn current_online_device_for_node(&self, node_id: &str) -> Option<String> {
828        let node_id = node_id.trim();
829        if node_id.is_empty() {
830            return None;
831        }
832        self.peers.values().find_map(|peer| {
833            if !peer.online {
834                return None;
835            }
836            let ticket_node_id = peer.ticket.as_deref().and_then(|ticket| {
837                let (iroh_ticket, token_suffix) = crate::session_token::split_ticket(ticket);
838                let payload = token_suffix
839                    .and_then(|suffix| crate::session_token::decode_payload(iroh_ticket, suffix))?;
840                if payload.scope.as_str() != "user-device" {
841                    return None;
842                }
843                parse_endpoint_ticket(iroh_ticket)
844                    .ok()
845                    .map(|address| address.id.to_string())
846            });
847            let ticket_node_id = ticket_node_id.as_deref()?;
848            let resolved_node_id = match peer.node_id.as_deref() {
849                Some(directory) if directory != ticket_node_id => return None,
850                Some(directory) => directory,
851                None => ticket_node_id,
852            };
853            (resolved_node_id == node_id).then(|| peer.device_id.clone())
854        })
855    }
856
857    fn current_online_assignment_matches(
858        &self,
859        device_id: &str,
860        node_id: &str,
861        ticket: &str,
862        evidence: &str,
863    ) -> bool {
864        self.peers.get(device_id.trim()).is_some_and(|peer| {
865            peer.online
866                && peer.ticket.as_deref() == Some(ticket.trim())
867                && browser_peer_evidence(peer) == evidence
868                && self.current_online_device_for_node(node_id).as_deref()
869                    == Some(peer.device_id.as_str())
870        })
871    }
872}
873
874#[cfg(not(target_arch = "wasm32"))]
875#[derive(Debug, Default)]
876pub(crate) struct NativeExternalAutoConnectActorState {
877    active: bool,
878    generation: u64,
879    local_device_id: String,
880    mailbox: BrowserDesiredPeerMailbox,
881    peer_evidence: std::collections::HashMap<String, String>,
882    failure_count: std::collections::HashMap<String, u8>,
883    retry_not_before_ms: std::collections::HashMap<String, i64>,
884    non_initiator_not_before_ms: std::collections::HashMap<String, i64>,
885    scheduled_deadline_ms: Option<i64>,
886    schedule_epoch: u64,
887    reconcile_running: bool,
888    wake_pending: bool,
889    /// Recoverable transport closes reconcile only the affected desired peer.
890    /// A provider revision clears this scope and performs the normal full pass.
891    recovery_device_ids: std::collections::HashSet<String>,
892    /// Deduplicates close/accept/path-watcher observations for one physical
893    /// loss. A new provider revision clears this fence immediately.
894    recovery_suppressed_until_ms: std::collections::HashMap<String, i64>,
895}
896
897#[cfg(target_arch = "wasm32")]
898struct BrowserAutoConnectActorState {
899    active: bool,
900    client: std::sync::Weak<Client>,
901    generation: u64,
902    local_device_id: String,
903    mailbox: BrowserDesiredPeerMailbox,
904    peer_evidence: std::collections::HashMap<String, String>,
905    failure_count: std::collections::HashMap<String, u8>,
906    retry_not_before_ms: std::collections::HashMap<String, i64>,
907    non_initiator_not_before_ms: std::collections::HashMap<String, i64>,
908    scheduled_deadline_ms: Option<i64>,
909    schedule_epoch: u64,
910    reconcile_running: bool,
911    wake_pending: bool,
912}
913
914#[cfg(target_arch = "wasm32")]
915thread_local! {
916    static BROWSER_AUTO_CONNECT_ACTORS: std::cell::RefCell<
917        std::collections::HashMap<usize, std::rc::Rc<std::cell::RefCell<BrowserAutoConnectActorState>>>
918    > = std::cell::RefCell::new(std::collections::HashMap::new());
919}
920
921fn browser_auto_connect_failure_backoff_ms(failure_count: u8) -> i64 {
922    500_i64
923        .saturating_mul(1_i64 << (failure_count as u32).min(4))
924        .min(8_000)
925}
926
927fn browser_peer_evidence(peer: &BrowserDesiredPeer) -> String {
928    format!(
929        "{}|{}|{}",
930        peer.node_id.as_deref().unwrap_or_default(),
931        peer.ticket
932            .as_deref()
933            .map(crate::session_token::token_fingerprint)
934            .unwrap_or_default(),
935        peer.session_id.as_deref().unwrap_or_default(),
936    )
937}
938
939fn desired_peer_evidence_is_current(
940    mailbox: &BrowserDesiredPeerMailbox,
941    peer: &BrowserDesiredPeer,
942) -> bool {
943    // Mailbox revisions describe the complete desired-peer set. An unrelated
944    // peer update must not cancel an in-flight connection for this peer after
945    // it has already established transport or admission. The peer's own
946    // evidence is the typed cancellation input.
947    mailbox
948        .peers
949        .get(&peer.device_id)
950        .is_some_and(|current| browser_peer_evidence(current) == browser_peer_evidence(peer))
951}
952
953#[cfg(target_arch = "wasm32")]
954fn browser_actor_key(client: &Arc<Client>) -> usize {
955    Arc::as_ptr(client) as usize
956}
957
958impl Client {
959    #[cfg(not(target_arch = "wasm32"))]
960    pub(crate) async fn native_external_auto_connect_owner_is_current(
961        &self,
962        user_id: &str,
963        local_device_id: &str,
964    ) -> bool {
965        let expected = (
966            user_id.trim().to_string(),
967            local_device_id.trim().to_string(),
968        );
969        let state = self.external_desired_peer_actor.lock().await;
970        let key = self
971            .auto_connect_loop_key
972            .lock()
973            .unwrap_or_else(|poisoned| poisoned.into_inner());
974        key.as_ref() == Some(&expected)
975            && state.active
976            && state.local_device_id == expected.1
977            && self
978                .auto_connect_generation
979                .load(std::sync::atomic::Ordering::SeqCst)
980                == state.generation
981    }
982
983    /// Returns the explicitly withdrawn device bound to `remote_node_id`, if
984    /// the active managed-session roster has retired that route.
985    ///
986    /// This is a read of the existing desired-peer owner, not a second
987    /// admission registry.  Accepted transport handlers use it before
988    /// projecting a replacement so cached peer tickets cannot resurrect a
989    /// device after gateway revocation.
990    pub(crate) async fn withdrawn_desired_device_for_node(
991        &self,
992        remote_node_id: &str,
993    ) -> Option<String> {
994        #[cfg(not(target_arch = "wasm32"))]
995        {
996            let state = self.external_desired_peer_actor.lock().await;
997            if !state.active
998                || self
999                    .auto_connect_generation
1000                    .load(std::sync::atomic::Ordering::SeqCst)
1001                    != state.generation
1002            {
1003                return None;
1004            }
1005            state.mailbox.withdrawn_device_for_node(remote_node_id)
1006        }
1007        #[cfg(target_arch = "wasm32")]
1008        {
1009            let actor_key = self as *const Client as usize;
1010            BROWSER_AUTO_CONNECT_ACTORS.with(|actors| {
1011                let actor = actors.borrow().get(&actor_key).cloned()?;
1012                let state = actor.borrow();
1013                browser_actor_is_current(&state)
1014                    .then(|| state.mailbox.withdrawn_device_for_node(remote_node_id))
1015                    .flatten()
1016            })
1017        }
1018    }
1019
1020    /// Fences a claimed same-user device identity after its authoritative
1021    /// roster entry has been withdrawn.  Room/ticket peers are unaffected
1022    /// unless that exact device previously belonged to this managed device
1023    /// roster and was explicitly retired.
1024    pub(crate) async fn desired_device_is_withdrawn(&self, device_id: &str) -> bool {
1025        #[cfg(not(target_arch = "wasm32"))]
1026        {
1027            let state = self.external_desired_peer_actor.lock().await;
1028            state.active
1029                && self
1030                    .auto_connect_generation
1031                    .load(std::sync::atomic::Ordering::SeqCst)
1032                    == state.generation
1033                && state.mailbox.device_is_withdrawn(device_id)
1034        }
1035        #[cfg(target_arch = "wasm32")]
1036        {
1037            let actor_key = self as *const Client as usize;
1038            BROWSER_AUTO_CONNECT_ACTORS.with(|actors| {
1039                actors
1040                    .borrow()
1041                    .get(&actor_key)
1042                    .cloned()
1043                    .is_some_and(|actor| {
1044                        let state = actor.borrow();
1045                        browser_actor_is_current(&state)
1046                            && state.mailbox.device_is_withdrawn(device_id)
1047                    })
1048            })
1049        }
1050    }
1051
1052    #[cfg(target_arch = "wasm32")]
1053    pub(crate) fn current_desired_user_device_for_node_now(
1054        &self,
1055        remote_node_id: &str,
1056    ) -> Option<String> {
1057        let actor_key = self as *const Client as usize;
1058        BROWSER_AUTO_CONNECT_ACTORS.with(|actors| {
1059            let actor = actors.borrow().get(&actor_key).cloned()?;
1060            let state = actor.borrow();
1061            browser_actor_is_current(&state)
1062                .then(|| state.mailbox.current_online_device_for_node(remote_node_id))
1063                .flatten()
1064        })
1065    }
1066
1067    #[cfg(target_arch = "wasm32")]
1068    pub(crate) async fn current_desired_user_device_for_node(
1069        &self,
1070        remote_node_id: &str,
1071    ) -> Option<String> {
1072        self.current_desired_user_device_for_node_now(remote_node_id)
1073    }
1074
1075    pub(crate) async fn current_desired_user_device_assignment_matches(
1076        &self,
1077        device_id: &str,
1078        remote_node_id: &str,
1079        ticket: &str,
1080        evidence: &str,
1081    ) -> bool {
1082        #[cfg(not(target_arch = "wasm32"))]
1083        {
1084            let state = self.external_desired_peer_actor.lock().await;
1085            state.active
1086                && self
1087                    .auto_connect_generation
1088                    .load(std::sync::atomic::Ordering::SeqCst)
1089                    == state.generation
1090                && state.mailbox.current_online_assignment_matches(
1091                    device_id,
1092                    remote_node_id,
1093                    ticket,
1094                    evidence,
1095                )
1096        }
1097        #[cfg(target_arch = "wasm32")]
1098        {
1099            let actor_key = self as *const Client as usize;
1100            BROWSER_AUTO_CONNECT_ACTORS.with(|actors| {
1101                actors
1102                    .borrow()
1103                    .get(&actor_key)
1104                    .cloned()
1105                    .is_some_and(|actor| {
1106                        let state = actor.borrow();
1107                        browser_actor_is_current(&state)
1108                            && state.mailbox.current_online_assignment_matches(
1109                                device_id,
1110                                remote_node_id,
1111                                ticket,
1112                                evidence,
1113                            )
1114                    })
1115            })
1116        }
1117    }
1118}
1119
1120#[derive(Debug, Clone, Copy, PartialEq, Eq)]
1121enum BrowserDesiredPeerDecision {
1122    ObserveExisting,
1123    WaitUntil(i64),
1124    Dial,
1125}
1126
1127fn browser_desired_peer_decision(
1128    local_node_id: &str,
1129    remote_node_id: &str,
1130    transport_alive: bool,
1131    now_ms: i64,
1132    non_initiator_not_before_ms: Option<i64>,
1133) -> BrowserDesiredPeerDecision {
1134    if transport_alive {
1135        return BrowserDesiredPeerDecision::ObserveExisting;
1136    }
1137    if local_node_id > remote_node_id {
1138        return BrowserDesiredPeerDecision::Dial;
1139    }
1140
1141    let not_before = non_initiator_not_before_ms.unwrap_or_else(|| now_ms.saturating_add(3_000));
1142    if now_ms >= not_before {
1143        BrowserDesiredPeerDecision::Dial
1144    } else {
1145        BrowserDesiredPeerDecision::WaitUntil(not_before)
1146    }
1147}
1148
1149fn should_replace_browser_retry_deadline(current: Option<i64>, next: Option<i64>) -> bool {
1150    match (current, next) {
1151        (None, None) => false,
1152        (Some(_), None) | (None, Some(_)) => true,
1153        (Some(current), Some(next)) => next < current,
1154    }
1155}
1156
1157fn browser_reconcile_deadlines(
1158    outcome: BrowserPeerReconcileOutcome,
1159    failure_retry_deadline: Option<i64>,
1160) -> (Option<i64>, Option<i64>) {
1161    match outcome {
1162        BrowserPeerReconcileOutcome::Retry => (failure_retry_deadline, None),
1163        BrowserPeerReconcileOutcome::WakeAt(deadline) => (Some(deadline), None),
1164        BrowserPeerReconcileOutcome::WaitUntil(deadline) => (None, Some(deadline)),
1165        BrowserPeerReconcileOutcome::Stable | BrowserPeerReconcileOutcome::Ineligible => {
1166            (None, None)
1167        }
1168    }
1169}
1170
1171#[derive(Debug, Clone, Copy, PartialEq, Eq)]
1172enum BrowserPeerReconcileOutcome {
1173    Stable,
1174    Ineligible,
1175    Retry,
1176    WakeAt(i64),
1177    WaitUntil(i64),
1178}
1179
1180#[cfg(target_arch = "wasm32")]
1181fn browser_actor_is_current(actor: &BrowserAutoConnectActorState) -> bool {
1182    actor.active
1183        && actor.client.upgrade().is_some_and(|client| {
1184            client
1185                .auto_connect_generation
1186                .load(std::sync::atomic::Ordering::SeqCst)
1187                == actor.generation
1188        })
1189}
1190
1191#[cfg(target_arch = "wasm32")]
1192fn browser_attempt_is_current(
1193    actor: &BrowserAutoConnectActorState,
1194    _revision: u64,
1195    peer: &BrowserDesiredPeer,
1196) -> bool {
1197    browser_actor_is_current(actor) && desired_peer_evidence_is_current(&actor.mailbox, peer)
1198}
1199
1200#[cfg(target_arch = "wasm32")]
1201fn deactivate_browser_actor(actor: &mut BrowserAutoConnectActorState) {
1202    actor.active = false;
1203    actor.mailbox.peers.clear();
1204    actor.peer_evidence.clear();
1205    actor.failure_count.clear();
1206    actor.retry_not_before_ms.clear();
1207    actor.non_initiator_not_before_ms.clear();
1208    actor.scheduled_deadline_ms = None;
1209    actor.schedule_epoch = actor.schedule_epoch.saturating_add(1);
1210    actor.wake_pending = false;
1211}
1212
1213#[cfg(target_arch = "wasm32")]
1214fn schedule_browser_auto_connect_retry(
1215    actor: &std::rc::Rc<std::cell::RefCell<BrowserAutoConnectActorState>>,
1216) {
1217    let (next_deadline, generation) = {
1218        let state = actor.borrow();
1219        if !browser_actor_is_current(&state) {
1220            return;
1221        }
1222        let next = state
1223            .retry_not_before_ms
1224            .values()
1225            .chain(state.non_initiator_not_before_ms.values())
1226            .copied()
1227            .min();
1228        (next, state.generation)
1229    };
1230
1231    let schedule = {
1232        let mut state = actor.borrow_mut();
1233        if !should_replace_browser_retry_deadline(state.scheduled_deadline_ms, next_deadline) {
1234            return;
1235        }
1236        state.schedule_epoch = state.schedule_epoch.saturating_add(1);
1237        state.scheduled_deadline_ms = next_deadline;
1238        next_deadline.map(|deadline| (deadline, state.schedule_epoch))
1239    };
1240
1241    let Some((deadline, epoch)) = schedule else {
1242        return;
1243    };
1244    let actor = actor.clone();
1245    wasm_bindgen_futures::spawn_local(async move {
1246        let delay_ms = deadline.saturating_sub(now_millis_i64()).max(0) as u64;
1247        gloo_timers::future::sleep(std::time::Duration::from_millis(delay_ms)).await;
1248        let should_wake = {
1249            let mut state = actor.borrow_mut();
1250            if !state.active
1251                || state.generation != generation
1252                || state.schedule_epoch != epoch
1253                || !browser_actor_is_current(&state)
1254            {
1255                false
1256            } else {
1257                state.scheduled_deadline_ms = None;
1258                true
1259            }
1260        };
1261        if should_wake {
1262            wake_browser_auto_connect_actor(actor);
1263        }
1264    });
1265}
1266
1267#[cfg(target_arch = "wasm32")]
1268fn wake_browser_auto_connect_actor(
1269    actor: std::rc::Rc<std::cell::RefCell<BrowserAutoConnectActorState>>,
1270) {
1271    let should_spawn = {
1272        let mut state = actor.borrow_mut();
1273        if !browser_actor_is_current(&state) {
1274            return;
1275        }
1276        if state.reconcile_running {
1277            state.wake_pending = true;
1278            false
1279        } else {
1280            state.reconcile_running = true;
1281            true
1282        }
1283    };
1284    if !should_spawn {
1285        return;
1286    }
1287
1288    wasm_bindgen_futures::spawn_local(async move {
1289        loop {
1290            run_browser_auto_connect_pass(actor.clone()).await;
1291            let run_again = {
1292                let mut state = actor.borrow_mut();
1293                if !browser_actor_is_current(&state) {
1294                    state.reconcile_running = false;
1295                    false
1296                } else if state.wake_pending {
1297                    state.wake_pending = false;
1298                    true
1299                } else {
1300                    state.reconcile_running = false;
1301                    false
1302                }
1303            };
1304            if !run_again {
1305                schedule_browser_auto_connect_retry(&actor);
1306                break;
1307            }
1308        }
1309    });
1310}
1311
1312#[cfg(target_arch = "wasm32")]
1313async fn reconcile_browser_desired_peer(
1314    actor: &std::rc::Rc<std::cell::RefCell<BrowserAutoConnectActorState>>,
1315    client: &Arc<Client>,
1316    generation: u64,
1317    revision: u64,
1318    local_device_id: &str,
1319    peer: &BrowserDesiredPeer,
1320    non_initiator_not_before_ms: Option<i64>,
1321) -> BrowserPeerReconcileOutcome {
1322    if client
1323        .auto_connect_generation
1324        .load(std::sync::atomic::Ordering::SeqCst)
1325        != generation
1326    {
1327        return BrowserPeerReconcileOutcome::Ineligible;
1328    }
1329    if client.is_app_execution_suspended() {
1330        return BrowserPeerReconcileOutcome::Ineligible;
1331    }
1332
1333    let parsed_ticket = peer.ticket.as_deref().and_then(|ticket| {
1334        let (iroh_ticket, _) = crate::session_token::split_ticket(ticket);
1335        parse_endpoint_ticket(iroh_ticket).ok()
1336    });
1337    let parsed_node_id = parsed_ticket.as_ref().map(|address| address.id.to_string());
1338    let binding_node_id = peer.node_id.as_deref().or(parsed_node_id.as_deref());
1339    if let Some(node_id) = binding_node_id {
1340        client.bind_node_device_id(node_id, &peer.device_id).await;
1341    }
1342
1343    if !actor
1344        .try_borrow()
1345        .is_ok_and(|state| browser_attempt_is_current(&state, revision, peer))
1346    {
1347        return BrowserPeerReconcileOutcome::Ineligible;
1348    }
1349    if !peer.online || peer.device_id == local_device_id {
1350        return BrowserPeerReconcileOutcome::Ineligible;
1351    }
1352    if remote_excludes_local_device(&peer.excluded_peers, local_device_id) {
1353        // `excludedPeers` is a durable auto-connect policy, not a transport
1354        // teardown command. A product may keep auto-connect disabled while the
1355        // user explicitly opens a session. Only the authenticated native
1356        // manual-disconnect notice (or an actual roster withdrawal) may retire
1357        // an already-admitted generation.
1358        return BrowserPeerReconcileOutcome::Ineligible;
1359    }
1360    if client.is_auto_connect_excluded(&peer.device_id) {
1361        return BrowserPeerReconcileOutcome::Ineligible;
1362    }
1363
1364    let Some(ticket) = peer.ticket.as_deref() else {
1365        return BrowserPeerReconcileOutcome::Ineligible;
1366    };
1367    let Some(endpoint_addr) = parsed_ticket else {
1368        return BrowserPeerReconcileOutcome::Ineligible;
1369    };
1370    let endpoint_id = endpoint_addr.id;
1371    let remote_node_id = endpoint_id.to_string();
1372    if peer
1373        .node_id
1374        .as_deref()
1375        .is_some_and(|node_id| node_id != remote_node_id)
1376    {
1377        return BrowserPeerReconcileOutcome::Ineligible;
1378    }
1379
1380    let Some(local_node_id) = client.current_node_id().await else {
1381        return BrowserPeerReconcileOutcome::WakeAt(now_millis_i64().saturating_add(250));
1382    };
1383    if local_node_id == remote_node_id {
1384        return BrowserPeerReconcileOutcome::Ineligible;
1385    }
1386
1387    client
1388        .reconcile_authoritative_device_node(&peer.device_id, &remote_node_id)
1389        .await;
1390    if !actor
1391        .try_borrow()
1392        .is_ok_and(|state| browser_attempt_is_current(&state, revision, peer))
1393    {
1394        return BrowserPeerReconcileOutcome::Ineligible;
1395    }
1396
1397    let transport_alive = client.is_connection_transport_alive(endpoint_id).await;
1398    match browser_desired_peer_decision(
1399        &local_node_id,
1400        &remote_node_id,
1401        transport_alive,
1402        now_millis_i64(),
1403        non_initiator_not_before_ms,
1404    ) {
1405        BrowserDesiredPeerDecision::WaitUntil(deadline) => {
1406            return BrowserPeerReconcileOutcome::WaitUntil(deadline);
1407        }
1408        BrowserDesiredPeerDecision::ObserveExisting | BrowserDesiredPeerDecision::Dial => {}
1409    }
1410
1411    let desired_peer_evidence = browser_peer_evidence(peer);
1412    let result = client
1413        .connect_desired_device(&peer.device_id, ticket, &desired_peer_evidence)
1414        .await;
1415    let (attempt_is_current, peer_was_withdrawn) = actor
1416        .try_borrow()
1417        .map(|state| {
1418            let actor_is_current = browser_actor_is_current(&state);
1419            (
1420                actor_is_current && browser_attempt_is_current(&state, revision, peer),
1421                superseded_desired_peer_was_withdrawn(
1422                    actor_is_current,
1423                    state.mailbox.revision,
1424                    revision,
1425                    state.mailbox.peers.contains_key(&peer.device_id),
1426                ),
1427            )
1428        })
1429        .unwrap_or((false, false));
1430    if !attempt_is_current {
1431        if peer_was_withdrawn {
1432            client
1433                .retire_withdrawn_desired_peer_connections(&peer.device_id, peer.node_id.as_deref())
1434                .await;
1435        }
1436        return BrowserPeerReconcileOutcome::Ineligible;
1437    }
1438    if client.is_auto_connect_excluded(&peer.device_id) {
1439        // This is a dialing preference, not a manual-disconnect command.
1440        // A concurrent explicit connect may share the completed transport.
1441        // Stop this automatic attempt without retiring that shared generation.
1442        return BrowserPeerReconcileOutcome::Ineligible;
1443    }
1444
1445    let Ok(result) = result else {
1446        return BrowserPeerReconcileOutcome::Retry;
1447    };
1448    if matches!(result.state.as_str(), "failed" | "closed")
1449        || !client.is_connection_transport_alive(endpoint_id).await
1450    {
1451        return BrowserPeerReconcileOutcome::Retry;
1452    }
1453
1454    let (iroh_ticket, token_suffix) = crate::session_token::split_ticket(ticket);
1455    if let Some(token) = token_suffix.and_then(|suffix| {
1456        crate::session_token::decode_payload(iroh_ticket, suffix).map(|payload| payload.token)
1457    }) {
1458        let has_proof = client
1459            .has_current_remote_session_admission_proof(&result.connection_id, endpoint_id, &token)
1460            .await;
1461        // An explicit product scope supersedes the provider user-device token
1462        // on this shared physical route. `connect_desired_device` substitutes
1463        // Rust's cached scoped intent and reports the host-approved scope.
1464        if !has_proof && result.approved_scope.as_deref().is_none_or(str::is_empty) {
1465            return BrowserPeerReconcileOutcome::Retry;
1466        }
1467    }
1468
1469    BrowserPeerReconcileOutcome::Stable
1470}
1471
1472#[cfg(target_arch = "wasm32")]
1473async fn run_browser_auto_connect_pass(
1474    actor: std::rc::Rc<std::cell::RefCell<BrowserAutoConnectActorState>>,
1475) {
1476    let (client, generation, revision, local_device_id, peers) = {
1477        let state = actor.borrow();
1478        if !browser_actor_is_current(&state) {
1479            return;
1480        }
1481        let Some(client) = state.client.upgrade() else {
1482            return;
1483        };
1484        (
1485            client,
1486            state.generation,
1487            state.mailbox.revision,
1488            state.local_device_id.clone(),
1489            state.mailbox.peers.values().cloned().collect::<Vec<_>>(),
1490        )
1491    };
1492
1493    for peer in peers {
1494        let (retry_not_before, non_initiator_not_before) = {
1495            let state = actor.borrow();
1496            if !browser_attempt_is_current(&state, revision, &peer) {
1497                continue;
1498            }
1499            (
1500                state.retry_not_before_ms.get(&peer.device_id).copied(),
1501                state
1502                    .non_initiator_not_before_ms
1503                    .get(&peer.device_id)
1504                    .copied(),
1505            )
1506        };
1507        let now = now_millis_i64();
1508        if retry_not_before.is_some_and(|deadline| now < deadline) {
1509            continue;
1510        }
1511
1512        let outcome = reconcile_browser_desired_peer(
1513            &actor,
1514            &client,
1515            generation,
1516            revision,
1517            &local_device_id,
1518            &peer,
1519            non_initiator_not_before,
1520        )
1521        .await;
1522        let backups_to_retire = {
1523            let mut state = actor.borrow_mut();
1524            if !browser_attempt_is_current(&state, revision, &peer) {
1525                continue;
1526            }
1527            let backups = if outcome == BrowserPeerReconcileOutcome::Stable {
1528                state
1529                    .mailbox
1530                    .acknowledge_topology_peer_ready(&peer.device_id, peer.topology_revision)
1531            } else {
1532                state
1533                    .mailbox
1534                    .observe_topology_peer_not_ready(&peer.device_id, peer.topology_revision);
1535                Vec::new()
1536            };
1537            match outcome {
1538                BrowserPeerReconcileOutcome::Stable | BrowserPeerReconcileOutcome::Ineligible => {
1539                    state.failure_count.remove(&peer.device_id);
1540                    state.retry_not_before_ms.remove(&peer.device_id);
1541                    state.non_initiator_not_before_ms.remove(&peer.device_id);
1542                }
1543                BrowserPeerReconcileOutcome::Retry => {
1544                    let count = state
1545                        .failure_count
1546                        .entry(peer.device_id.clone())
1547                        .or_insert(0);
1548                    *count = count.saturating_add(1).min(10);
1549                    let deadline = now_millis_i64()
1550                        .saturating_add(browser_auto_connect_failure_backoff_ms(*count));
1551                    let (retry_deadline, non_initiator_deadline) =
1552                        browser_reconcile_deadlines(outcome, Some(deadline));
1553                    match retry_deadline {
1554                        Some(deadline) => {
1555                            state
1556                                .retry_not_before_ms
1557                                .insert(peer.device_id.clone(), deadline);
1558                        }
1559                        None => {
1560                            state.retry_not_before_ms.remove(&peer.device_id);
1561                        }
1562                    }
1563                    match non_initiator_deadline {
1564                        Some(deadline) => {
1565                            state
1566                                .non_initiator_not_before_ms
1567                                .insert(peer.device_id.clone(), deadline);
1568                        }
1569                        None => {
1570                            state.non_initiator_not_before_ms.remove(&peer.device_id);
1571                        }
1572                    }
1573                }
1574                BrowserPeerReconcileOutcome::WakeAt(deadline) => {
1575                    state
1576                        .retry_not_before_ms
1577                        .insert(peer.device_id.clone(), deadline);
1578                    state.non_initiator_not_before_ms.remove(&peer.device_id);
1579                }
1580                BrowserPeerReconcileOutcome::WaitUntil(deadline) => {
1581                    state.retry_not_before_ms.remove(&peer.device_id);
1582                    state
1583                        .non_initiator_not_before_ms
1584                        .insert(peer.device_id.clone(), deadline);
1585                }
1586            }
1587            backups
1588        };
1589        for backup in backups_to_retire {
1590            client
1591                .retire_withdrawn_desired_peer_connections(
1592                    &backup.device_id,
1593                    backup.node_id.as_deref(),
1594                )
1595                .await;
1596        }
1597    }
1598}
1599
1600#[cfg(not(target_arch = "wasm32"))]
1601async fn native_external_attempt_is_current(
1602    client: &Arc<Client>,
1603    generation: u64,
1604    _revision: u64,
1605    peer: &BrowserDesiredPeer,
1606) -> bool {
1607    if client
1608        .auto_connect_generation
1609        .load(std::sync::atomic::Ordering::SeqCst)
1610        != generation
1611    {
1612        return false;
1613    }
1614    let state = client.external_desired_peer_actor.lock().await;
1615    state.active
1616        && state.generation == generation
1617        && desired_peer_evidence_is_current(&state.mailbox, peer)
1618}
1619
1620#[cfg(not(target_arch = "wasm32"))]
1621async fn reconcile_native_external_desired_peer(
1622    client: &Arc<Client>,
1623    generation: u64,
1624    revision: u64,
1625    local_device_id: &str,
1626    peer: &BrowserDesiredPeer,
1627    non_initiator_not_before_ms: Option<i64>,
1628) -> BrowserPeerReconcileOutcome {
1629    if !native_external_attempt_is_current(client, generation, revision, peer).await
1630        || client.is_app_execution_suspended()
1631    {
1632        return BrowserPeerReconcileOutcome::Ineligible;
1633    }
1634
1635    let parsed_ticket = peer.ticket.as_deref().and_then(|ticket| {
1636        let (iroh_ticket, _) = crate::session_token::split_ticket(ticket);
1637        parse_endpoint_ticket(iroh_ticket).ok()
1638    });
1639    let parsed_node_id = parsed_ticket.as_ref().map(|address| address.id.to_string());
1640    let binding_node_id = peer.node_id.as_deref().or(parsed_node_id.as_deref());
1641    if let Some(node_id) = binding_node_id {
1642        client.bind_node_device_id(node_id, &peer.device_id).await;
1643    }
1644
1645    if !peer.online || peer.device_id == local_device_id {
1646        return BrowserPeerReconcileOutcome::Ineligible;
1647    }
1648    if remote_excludes_local_device(&peer.excluded_peers, local_device_id) {
1649        // Roster exclusion suppresses automatic dialing only. It cannot
1650        // invalidate an explicit, authenticated session that shares the same
1651        // physical generation.
1652        return BrowserPeerReconcileOutcome::Ineligible;
1653    }
1654    if client.is_auto_connect_excluded(&peer.device_id) {
1655        return BrowserPeerReconcileOutcome::Ineligible;
1656    }
1657
1658    let Some(ticket) = peer.ticket.as_deref() else {
1659        return BrowserPeerReconcileOutcome::Ineligible;
1660    };
1661    let Some(endpoint_addr) = parsed_ticket else {
1662        return BrowserPeerReconcileOutcome::Ineligible;
1663    };
1664    let endpoint_id = endpoint_addr.id;
1665    let remote_node_id = endpoint_id.to_string();
1666    if peer
1667        .node_id
1668        .as_deref()
1669        .is_some_and(|node_id| node_id != remote_node_id)
1670    {
1671        return BrowserPeerReconcileOutcome::Ineligible;
1672    }
1673
1674    let Some(local_node_id) = client.current_node_id().await else {
1675        return BrowserPeerReconcileOutcome::WakeAt(now_millis_i64().saturating_add(250));
1676    };
1677    if local_node_id == remote_node_id {
1678        return BrowserPeerReconcileOutcome::Ineligible;
1679    }
1680
1681    client
1682        .reconcile_authoritative_device_node(&peer.device_id, &remote_node_id)
1683        .await;
1684    if !native_external_attempt_is_current(client, generation, revision, peer).await {
1685        return BrowserPeerReconcileOutcome::Ineligible;
1686    }
1687
1688    let transport_alive = client.is_connection_transport_alive(endpoint_id).await;
1689    match browser_desired_peer_decision(
1690        &local_node_id,
1691        &remote_node_id,
1692        transport_alive,
1693        now_millis_i64(),
1694        non_initiator_not_before_ms,
1695    ) {
1696        BrowserDesiredPeerDecision::WaitUntil(deadline) => {
1697            return BrowserPeerReconcileOutcome::WaitUntil(deadline);
1698        }
1699        BrowserDesiredPeerDecision::ObserveExisting | BrowserDesiredPeerDecision::Dial => {}
1700    }
1701
1702    let desired_peer_evidence = browser_peer_evidence(peer);
1703    let result = client
1704        .connect_desired_device(&peer.device_id, ticket, &desired_peer_evidence)
1705        .await;
1706    let transport_stable_id = client
1707        .get_connection(endpoint_id)
1708        .await
1709        .map(|connection| crate::transport_generation::for_connection(&connection));
1710    if let Some(transport_stable_id) = transport_stable_id {
1711        let connection_id = Client::deterministic_connection_id(&local_node_id, &remote_node_id);
1712        let _ = client
1713            .maybe_start_offline_proof_for_transport(
1714                &connection_id,
1715                endpoint_id,
1716                transport_stable_id,
1717            )
1718            .await;
1719    }
1720    if !native_external_attempt_is_current(client, generation, revision, peer).await {
1721        let peer_was_withdrawn = {
1722            let state = client.external_desired_peer_actor.lock().await;
1723            let actor_is_current = state.active && state.generation == generation;
1724            superseded_desired_peer_was_withdrawn(
1725                actor_is_current,
1726                state.mailbox.revision,
1727                revision,
1728                state.mailbox.peers.contains_key(&peer.device_id),
1729            )
1730        };
1731        if peer_was_withdrawn {
1732            client
1733                .retire_withdrawn_desired_peer_connections(&peer.device_id, peer.node_id.as_deref())
1734                .await;
1735        }
1736        return BrowserPeerReconcileOutcome::Ineligible;
1737    }
1738    if client.is_auto_connect_excluded(&peer.device_id) {
1739        // This is a dialing preference, not a manual-disconnect command.
1740        // A concurrent explicit connect may share the completed transport.
1741        // Stop this automatic attempt without retiring that shared generation.
1742        return BrowserPeerReconcileOutcome::Ineligible;
1743    }
1744
1745    let Ok(result) = result else {
1746        return BrowserPeerReconcileOutcome::Retry;
1747    };
1748    if matches!(result.state.as_str(), "failed" | "closed")
1749        || !client.is_connection_transport_alive(endpoint_id).await
1750    {
1751        return BrowserPeerReconcileOutcome::Retry;
1752    }
1753
1754    let (iroh_ticket, token_suffix) = crate::session_token::split_ticket(ticket);
1755    if let Some(token) = token_suffix.and_then(|suffix| {
1756        crate::session_token::decode_payload(iroh_ticket, suffix).map(|payload| payload.token)
1757    }) {
1758        let has_proof = client
1759            .has_current_remote_session_admission_proof(&result.connection_id, endpoint_id, &token)
1760            .await;
1761        if !has_proof && result.approved_scope.as_deref().is_none_or(str::is_empty) {
1762            return BrowserPeerReconcileOutcome::Retry;
1763        }
1764    }
1765
1766    BrowserPeerReconcileOutcome::Stable
1767}
1768
1769#[cfg(not(target_arch = "wasm32"))]
1770async fn run_native_external_auto_connect_pass(client: Arc<Client>) {
1771    let (generation, revision, local_device_id, peers, recovery_scoped) = {
1772        let mut state = client.external_desired_peer_actor.lock().await;
1773        if !state.active
1774            || client
1775                .auto_connect_generation
1776                .load(std::sync::atomic::Ordering::SeqCst)
1777                != state.generation
1778        {
1779            return;
1780        }
1781        let recovery_device_ids = std::mem::take(&mut state.recovery_device_ids);
1782        let recovery_scoped = !recovery_device_ids.is_empty();
1783        let peers = state
1784            .mailbox
1785            .peers
1786            .values()
1787            .filter(|peer| {
1788                recovery_device_ids.is_empty() || recovery_device_ids.contains(&peer.device_id)
1789            })
1790            .cloned()
1791            .collect::<Vec<_>>();
1792        (
1793            state.generation,
1794            state.mailbox.revision,
1795            state.local_device_id.clone(),
1796            peers,
1797            recovery_scoped,
1798        )
1799    };
1800
1801    for peer in peers {
1802        let (retry_not_before, non_initiator_not_before) = {
1803            let state = client.external_desired_peer_actor.lock().await;
1804            if !state.active
1805                || state.generation != generation
1806                || state.mailbox.revision != revision
1807                || !state
1808                    .mailbox
1809                    .peers
1810                    .get(&peer.device_id)
1811                    .is_some_and(|current| {
1812                        browser_peer_evidence(current) == browser_peer_evidence(&peer)
1813                    })
1814            {
1815                continue;
1816            }
1817            (
1818                state.retry_not_before_ms.get(&peer.device_id).copied(),
1819                state
1820                    .non_initiator_not_before_ms
1821                    .get(&peer.device_id)
1822                    .copied(),
1823            )
1824        };
1825        let now = now_millis_i64();
1826        if retry_not_before.is_some_and(|deadline| now < deadline) {
1827            continue;
1828        }
1829
1830        let outcome = reconcile_native_external_desired_peer(
1831            &client,
1832            generation,
1833            revision,
1834            &local_device_id,
1835            &peer,
1836            non_initiator_not_before,
1837        )
1838        .await;
1839        if recovery_scoped {
1840            eprintln!(
1841                "[pluto-rtc][auto-connect][recovery-pass] device_id={} outcome={:?}",
1842                peer.device_id, outcome,
1843            );
1844        }
1845        let backups_to_retire = {
1846            let mut state = client.external_desired_peer_actor.lock().await;
1847            if !state.active
1848                || state.generation != generation
1849                || state.mailbox.revision != revision
1850                || !state
1851                    .mailbox
1852                    .peers
1853                    .get(&peer.device_id)
1854                    .is_some_and(|current| {
1855                        browser_peer_evidence(current) == browser_peer_evidence(&peer)
1856                    })
1857            {
1858                continue;
1859            }
1860            let backups = if outcome == BrowserPeerReconcileOutcome::Stable {
1861                state
1862                    .mailbox
1863                    .acknowledge_topology_peer_ready(&peer.device_id, peer.topology_revision)
1864            } else {
1865                state
1866                    .mailbox
1867                    .observe_topology_peer_not_ready(&peer.device_id, peer.topology_revision);
1868                Vec::new()
1869            };
1870            match outcome {
1871                BrowserPeerReconcileOutcome::Stable | BrowserPeerReconcileOutcome::Ineligible => {
1872                    state.failure_count.remove(&peer.device_id);
1873                    state.retry_not_before_ms.remove(&peer.device_id);
1874                    state.non_initiator_not_before_ms.remove(&peer.device_id);
1875                }
1876                BrowserPeerReconcileOutcome::Retry => {
1877                    let count = state
1878                        .failure_count
1879                        .entry(peer.device_id.clone())
1880                        .or_insert(0);
1881                    *count = count.saturating_add(1).min(10);
1882                    let deadline = now_millis_i64()
1883                        .saturating_add(browser_auto_connect_failure_backoff_ms(*count));
1884                    let (retry_deadline, non_initiator_deadline) =
1885                        browser_reconcile_deadlines(outcome, Some(deadline));
1886                    match retry_deadline {
1887                        Some(deadline) => {
1888                            state
1889                                .retry_not_before_ms
1890                                .insert(peer.device_id.clone(), deadline);
1891                        }
1892                        None => {
1893                            state.retry_not_before_ms.remove(&peer.device_id);
1894                        }
1895                    }
1896                    match non_initiator_deadline {
1897                        Some(deadline) => {
1898                            state
1899                                .non_initiator_not_before_ms
1900                                .insert(peer.device_id.clone(), deadline);
1901                        }
1902                        None => {
1903                            state.non_initiator_not_before_ms.remove(&peer.device_id);
1904                        }
1905                    }
1906                    // A close-triggered pass is deliberately scoped to the
1907                    // affected durable device. Preserve that scope across its
1908                    // bounded retry: otherwise the timer below wakes an empty
1909                    // recovery set and re-runs every unrelated desired peer,
1910                    // turning one physical loss into mesh-wide transport churn.
1911                    if recovery_scoped {
1912                        state.recovery_device_ids.insert(peer.device_id.clone());
1913                    }
1914                }
1915                BrowserPeerReconcileOutcome::WakeAt(deadline) => {
1916                    state
1917                        .retry_not_before_ms
1918                        .insert(peer.device_id.clone(), deadline);
1919                    state.non_initiator_not_before_ms.remove(&peer.device_id);
1920                    if recovery_scoped {
1921                        state.recovery_device_ids.insert(peer.device_id.clone());
1922                    }
1923                }
1924                BrowserPeerReconcileOutcome::WaitUntil(deadline) => {
1925                    state.retry_not_before_ms.remove(&peer.device_id);
1926                    state
1927                        .non_initiator_not_before_ms
1928                        .insert(peer.device_id.clone(), deadline);
1929                    if recovery_scoped {
1930                        state.recovery_device_ids.insert(peer.device_id.clone());
1931                    }
1932                }
1933            }
1934            backups
1935        };
1936        for backup in backups_to_retire {
1937            client
1938                .retire_withdrawn_desired_peer_connections(
1939                    &backup.device_id,
1940                    backup.node_id.as_deref(),
1941                )
1942                .await;
1943        }
1944    }
1945}
1946
1947#[cfg(not(target_arch = "wasm32"))]
1948async fn schedule_native_external_auto_connect_retry(client: Arc<Client>) {
1949    let schedule = {
1950        let mut state = client.external_desired_peer_actor.lock().await;
1951        if !state.active
1952            || client
1953                .auto_connect_generation
1954                .load(std::sync::atomic::Ordering::SeqCst)
1955                != state.generation
1956        {
1957            return;
1958        }
1959        let next = state
1960            .retry_not_before_ms
1961            .values()
1962            .chain(state.non_initiator_not_before_ms.values())
1963            .copied()
1964            .min();
1965        if !should_replace_browser_retry_deadline(state.scheduled_deadline_ms, next) {
1966            return;
1967        }
1968        state.schedule_epoch = state.schedule_epoch.saturating_add(1);
1969        state.scheduled_deadline_ms = next;
1970        next.map(|deadline| (deadline, state.schedule_epoch, state.generation))
1971    };
1972    let Some((deadline, epoch, generation)) = schedule else {
1973        return;
1974    };
1975    tokio::spawn(async move {
1976        let delay_ms = deadline.saturating_sub(now_millis_i64()).max(0) as u64;
1977        tokio::time::sleep(std::time::Duration::from_millis(delay_ms)).await;
1978        let should_wake = {
1979            let mut state = client.external_desired_peer_actor.lock().await;
1980            if !state.active
1981                || state.generation != generation
1982                || state.schedule_epoch != epoch
1983                || client
1984                    .auto_connect_generation
1985                    .load(std::sync::atomic::Ordering::SeqCst)
1986                    != generation
1987            {
1988                false
1989            } else {
1990                state.scheduled_deadline_ms = None;
1991                true
1992            }
1993        };
1994        if should_wake {
1995            let _ = wake_native_external_auto_connect_boxed(client).await;
1996        }
1997    });
1998}
1999
2000#[cfg(not(target_arch = "wasm32"))]
2001fn wake_native_external_auto_connect_boxed(
2002    client: Arc<Client>,
2003) -> std::pin::Pin<Box<dyn std::future::Future<Output = bool> + Send>> {
2004    Box::pin(async move {
2005        let should_spawn = {
2006            let mut state = client.external_desired_peer_actor.lock().await;
2007            if !state.active
2008                || client
2009                    .auto_connect_generation
2010                    .load(std::sync::atomic::Ordering::SeqCst)
2011                    != state.generation
2012            {
2013                return false;
2014            }
2015            if state.reconcile_running {
2016                state.wake_pending = true;
2017                false
2018            } else {
2019                state.reconcile_running = true;
2020                true
2021            }
2022        };
2023        if !should_spawn {
2024            return true;
2025        }
2026        tokio::spawn(async move {
2027            loop {
2028                run_native_external_auto_connect_pass(client.clone()).await;
2029                let run_again = {
2030                    let mut state = client.external_desired_peer_actor.lock().await;
2031                    if !state.active
2032                        || client
2033                            .auto_connect_generation
2034                            .load(std::sync::atomic::Ordering::SeqCst)
2035                            != state.generation
2036                    {
2037                        state.reconcile_running = false;
2038                        false
2039                    } else if state.wake_pending {
2040                        state.wake_pending = false;
2041                        true
2042                    } else {
2043                        state.reconcile_running = false;
2044                        false
2045                    }
2046                };
2047                if !run_again {
2048                    schedule_native_external_auto_connect_retry(client.clone()).await;
2049                    break;
2050                }
2051            }
2052        });
2053        true
2054    })
2055}
2056
2057#[cfg(not(target_arch = "wasm32"))]
2058impl Client {
2059    /// Test-harness diagnostic for the provider-neutral desired-peer mailbox.
2060    ///
2061    /// Tickets are never returned. Only their endpoint identity and a one-way
2062    /// fingerprint are exposed so a production-shaped local test can prove
2063    /// that a gateway replacement revision crossed the host boundary.
2064    #[doc(hidden)]
2065    pub async fn external_desired_peer_debug_snapshot(&self) -> serde_json::Value {
2066        let state = self.external_desired_peer_actor.lock().await;
2067        let mut ready_active_peers = state
2068            .mailbox
2069            .pending_ready_active_peers
2070            .iter()
2071            .cloned()
2072            .collect::<Vec<_>>();
2073        ready_active_peers.sort();
2074        let peers = state
2075            .mailbox
2076            .peers
2077            .values()
2078            .map(|peer| {
2079                let ticket_node_id = peer.ticket.as_deref().and_then(|ticket| {
2080                    let (iroh_ticket, _) = crate::session_token::split_ticket(ticket);
2081                    parse_endpoint_ticket(iroh_ticket)
2082                        .ok()
2083                        .map(|address| address.id.to_string())
2084                });
2085                serde_json::json!({
2086                    "deviceId": peer.device_id,
2087                    "nodeId": peer.node_id,
2088                    "ticketNodeId": ticket_node_id,
2089                    "ticketFingerprint": peer.ticket.as_deref()
2090                        .map(crate::session_token::token_fingerprint),
2091                    "online": peer.online,
2092                    "sessionId": peer.session_id,
2093                })
2094            })
2095            .collect::<Vec<_>>();
2096        serde_json::json!({
2097            "active": state.active,
2098            "generation": state.generation,
2099            "revision": state.mailbox.revision,
2100            "reconcileRunning": state.reconcile_running,
2101            "wakePending": state.wake_pending,
2102            "scheduledDeadlineMs": state.scheduled_deadline_ms,
2103            "lastFailureCounts": state.failure_count,
2104            "retryNotBeforeMs": state.retry_not_before_ms,
2105            "topologyRevision": state.mailbox.pending_topology_revision,
2106            "readyActivePeers": ready_active_peers,
2107            "backupPeerCount": state.mailbox.peers.values()
2108                .filter(|peer| peer.topology_role == Some(SparseTopologyRole::Backup))
2109                .count(),
2110            "peers": peers,
2111        })
2112    }
2113
2114    pub async fn start_external_auto_connect(
2115        self: &Arc<Self>,
2116        user_id: String,
2117        local_device_id: String,
2118    ) -> anyhow::Result<()> {
2119        let user_id = user_id.trim().to_string();
2120        let local_device_id = local_device_id.trim().to_string();
2121        if user_id.is_empty() || local_device_id.is_empty() {
2122            return Err(anyhow::anyhow!(
2123                "external auto-connect requires non-empty user and local device ids"
2124            ));
2125        }
2126        let key = (user_id.clone(), local_device_id.clone());
2127        let duplicate_start = {
2128            let state = self.external_desired_peer_actor.lock().await;
2129            let guard = self
2130                .auto_connect_loop_key
2131                .lock()
2132                .unwrap_or_else(|poisoned| poisoned.into_inner());
2133            guard.as_ref() == Some(&key)
2134                && state.active
2135                && state.local_device_id == local_device_id
2136                && self
2137                    .auto_connect_generation
2138                    .load(std::sync::atomic::Ordering::SeqCst)
2139                    == state.generation
2140        };
2141        if duplicate_start {
2142            self.wake_native_external_auto_connect().await;
2143            return Ok(());
2144        }
2145        let generation = {
2146            let mut guard = self
2147                .auto_connect_loop_key
2148                .lock()
2149                .unwrap_or_else(|poisoned| poisoned.into_inner());
2150            self.known_device_ids_by_node
2151                .write()
2152                .unwrap_or_else(|poisoned| poisoned.into_inner())
2153                .clear();
2154            self.native_route_repair_credentials
2155                .write()
2156                .unwrap_or_else(|poisoned| poisoned.into_inner())
2157                .clear();
2158            self.scoped_route_repair_credentials
2159                .write()
2160                .unwrap_or_else(|poisoned| poisoned.into_inner())
2161                .clear();
2162            *guard = Some(key);
2163            self.auto_connect_generation
2164                .fetch_add(1, std::sync::atomic::Ordering::SeqCst)
2165                + 1
2166        };
2167        let mut state = self.external_desired_peer_actor.lock().await;
2168        *state = NativeExternalAutoConnectActorState {
2169            active: true,
2170            generation,
2171            local_device_id,
2172            ..NativeExternalAutoConnectActorState::default()
2173        };
2174        Ok(())
2175    }
2176
2177    pub async fn submit_external_desired_peers(
2178        self: &Arc<Self>,
2179        revision: u64,
2180        peers_json: &str,
2181    ) -> anyhow::Result<bool> {
2182        self.submit_external_desired_peer_update(revision, peers_json, None, false)
2183            .await
2184    }
2185
2186    /// Publish a capability withdrawal only if the root actor is still running.
2187    /// A stopped actor has already discarded its desired peers; retiring an old
2188    /// identity must not restart it or fail a subsequent identity startup.
2189    /// Ordinary admission/startup submissions remain strict.
2190    pub async fn withdraw_external_desired_peers(
2191        self: &Arc<Self>,
2192        revision: u64,
2193        peers_json: &str,
2194    ) -> anyhow::Result<bool> {
2195        self.submit_external_desired_peer_update(revision, peers_json, None, true)
2196            .await
2197    }
2198
2199    #[cfg(feature = "transport-lan")]
2200    pub(crate) async fn submit_external_desired_peer_observation_expired(
2201        self: &Arc<Self>,
2202        revision: u64,
2203        peers_json: &str,
2204        expired_device_id: &str,
2205    ) -> anyhow::Result<bool> {
2206        let expired_device_id = expired_device_id.trim();
2207        if expired_device_id.is_empty() {
2208            return Err(anyhow::anyhow!("expired offline device id is required"));
2209        }
2210        self.submit_external_desired_peer_update(revision, peers_json, Some(expired_device_id), false)
2211            .await
2212    }
2213
2214    async fn submit_external_desired_peer_update(
2215        self: &Arc<Self>,
2216        revision: u64,
2217        peers_json: &str,
2218        observation_expired_device_id: Option<&str>,
2219        withdrawing: bool,
2220    ) -> anyhow::Result<bool> {
2221        if peers_json.len() > 256 * 1024 {
2222            return Err(anyhow::anyhow!(
2223                "external desired-peer payload exceeds 256 KiB"
2224            ));
2225        }
2226        let peers: Vec<BrowserDesiredPeer> = serde_json::from_str(peers_json)
2227            .map_err(|error| anyhow::anyhow!("invalid external desired-peer payload: {error}"))?;
2228        if peers.len() > 100 {
2229            return Err(anyhow::anyhow!(
2230                "external desired-peer payload exceeds 100 peers"
2231            ));
2232        }
2233        if observation_expired_device_id
2234            .is_some_and(|expired| peers.iter().any(|peer| peer.device_id.trim() == expired))
2235        {
2236            return Err(anyhow::anyhow!(
2237                "expired offline device remains in desired-peer input"
2238            ));
2239        }
2240
2241        let (bindings, withdrawn) = {
2242            let mut state = self.external_desired_peer_actor.lock().await;
2243            if !state.active {
2244                if withdrawing {
2245                    return Ok(false);
2246                }
2247                return Err(anyhow::anyhow!("external auto-connect is not started"));
2248            }
2249            let accepted = match observation_expired_device_id {
2250                Some(expired) => state
2251                    .mailbox
2252                    .accept_with_observation_expired(revision, peers, expired),
2253                None => state.mailbox.accept_with_withdrawn(revision, peers),
2254            };
2255            let Some(withdrawn) = accepted else {
2256                return Ok(false);
2257            };
2258            let desired_ids = state
2259                .mailbox
2260                .peers
2261                .keys()
2262                .cloned()
2263                .collect::<std::collections::HashSet<_>>();
2264            state.peer_evidence.retain(|id, _| desired_ids.contains(id));
2265            state.failure_count.retain(|id, _| desired_ids.contains(id));
2266            state
2267                .retry_not_before_ms
2268                .retain(|id, _| desired_ids.contains(id));
2269            state
2270                .non_initiator_not_before_ms
2271                .retain(|id, _| desired_ids.contains(id));
2272
2273            let evidence = state
2274                .mailbox
2275                .peers
2276                .iter()
2277                .map(|(id, peer)| (id.clone(), browser_peer_evidence(peer)))
2278                .collect::<Vec<_>>();
2279            for (id, evidence) in evidence {
2280                let changed = state
2281                    .peer_evidence
2282                    .insert(id.clone(), evidence.clone())
2283                    .is_some_and(|previous| previous != evidence);
2284                if changed {
2285                    state.failure_count.remove(&id);
2286                    state.retry_not_before_ms.remove(&id);
2287                    state.non_initiator_not_before_ms.remove(&id);
2288                }
2289            }
2290            state.schedule_epoch = state.schedule_epoch.saturating_add(1);
2291            state.scheduled_deadline_ms = None;
2292            let bindings = state
2293                .mailbox
2294                .peers
2295                .values()
2296                .filter(|peer| peer.online)
2297                .filter_map(|peer| {
2298                    let ticket_node_id = peer.ticket.as_deref().and_then(|ticket| {
2299                        let (iroh_ticket, _) = crate::session_token::split_ticket(ticket);
2300                        parse_endpoint_ticket(iroh_ticket)
2301                            .ok()
2302                            .map(|address| address.id.to_string())
2303                    });
2304                    let node_id = match (peer.node_id.as_deref(), ticket_node_id.as_deref()) {
2305                        (Some(directory), Some(ticket)) if directory != ticket => return None,
2306                        (Some(directory), _) => directory.to_string(),
2307                        (None, Some(ticket)) => ticket.to_string(),
2308                        (None, None) => return None,
2309                    };
2310                    Some((peer.device_id.clone(), node_id))
2311                })
2312                .collect::<Vec<_>>();
2313            (bindings, withdrawn)
2314        };
2315        for peer in withdrawn {
2316            let still_withdrawn = {
2317                let state = self.external_desired_peer_actor.lock().await;
2318                state.active
2319                    && state.mailbox.revision == revision
2320                    && state
2321                        .mailbox
2322                        .peers
2323                        .get(&peer.device_id)
2324                        .is_none_or(|current| !current.online)
2325            };
2326            if still_withdrawn {
2327                self.retire_withdrawn_desired_peer_connections(
2328                    &peer.device_id,
2329                    peer.node_id.as_deref(),
2330                )
2331                .await;
2332            }
2333        }
2334        for (device_id, node_id) in bindings {
2335            self.observe_authoritative_device_node(&device_id, &node_id);
2336        }
2337        self.wake_native_external_auto_connect().await;
2338        Ok(true)
2339    }
2340
2341    pub async fn wake_native_external_auto_connect(&self) -> bool {
2342        // Explicit desired-peer input is authoritative and may affect the full
2343        // roster. It supersedes any narrow close-recovery timer/pass.
2344        {
2345            let mut state = self.external_desired_peer_actor.lock().await;
2346            state.recovery_device_ids.clear();
2347            state.recovery_suppressed_until_ms.clear();
2348        }
2349        wake_native_external_auto_connect_boxed(Arc::new(self.clone())).await
2350    }
2351
2352    /// Reconcile one already-installed replacement generation immediately.
2353    ///
2354    /// A close observer can schedule (and temporarily suppress duplicate)
2355    /// recovery before Iroh finishes selecting the successor connection. Once
2356    /// the incoming owner has atomically installed that successor and retired
2357    /// the preceding generation's native-main proofs, the old suppression no
2358    /// longer describes duplicate work: this is new typed lifecycle evidence
2359    /// that the current route itself needs admission repair. Feed that evidence
2360    /// to the existing desired-peer actor instead of waiting for another
2361    /// provider revision or starting a second retry owner.
2362    pub(crate) async fn wake_native_external_auto_connect_for_replaced_route(
2363        &self,
2364        remote_node_id: &str,
2365    ) -> bool {
2366        let remote_node_id = remote_node_id.trim();
2367        let Some(remote_device_id) = self
2368            .known_device_ids_by_node
2369            .read()
2370            .unwrap_or_else(|poisoned| poisoned.into_inner())
2371            .get(remote_node_id)
2372            .cloned()
2373        else {
2374            eprintln!(
2375                "[pluto-rtc][auto-connect][replacement-route] remote_node_id={} action=skip reason=device-binding-missing",
2376                remote_node_id,
2377            );
2378            return false;
2379        };
2380
2381        {
2382            let mut state = self.external_desired_peer_actor.lock().await;
2383            if !state.active
2384                || self
2385                    .auto_connect_generation
2386                    .load(std::sync::atomic::Ordering::SeqCst)
2387                    != state.generation
2388                || !state.mailbox.peers.contains_key(&remote_device_id)
2389            {
2390                eprintln!(
2391                    "[pluto-rtc][auto-connect][replacement-route] remote_node_id={} device_id={} action=skip reason=actor-or-peer-inactive",
2392                    remote_node_id, remote_device_id,
2393                );
2394                return false;
2395            }
2396
2397            state.recovery_device_ids.insert(remote_device_id.clone());
2398            state.recovery_suppressed_until_ms.remove(&remote_device_id);
2399            state.retry_not_before_ms.remove(&remote_device_id);
2400            state.non_initiator_not_before_ms.remove(&remote_device_id);
2401            state.schedule_epoch = state.schedule_epoch.saturating_add(1);
2402            state.scheduled_deadline_ms = None;
2403        }
2404
2405        eprintln!(
2406            "[pluto-rtc][auto-connect][replacement-route] remote_node_id={} device_id={} action=wake",
2407            remote_node_id, remote_device_id,
2408        );
2409        wake_native_external_auto_connect_boxed(Arc::new(self.clone())).await
2410    }
2411
2412    /// Coalesce a burst of recoverable native transport-loss notices before
2413    /// waking the one external desired-peer actor. A local interface can close
2414    /// several routes in one network transition; eagerly reconciling after
2415    /// each close would redial routes that the same transition is still
2416    /// retiring. A newer provider revision still wakes immediately and fences
2417    /// this scheduled notice through `schedule_epoch`.
2418    pub(crate) async fn schedule_native_external_auto_connect_recovery(
2419        &self,
2420        remote_node_id: &str,
2421    ) -> bool {
2422        // One physical QUIC loss produces close, accept, path-watcher, and
2423        // admission-stream notifications on independent tasks. Give the
2424        // successor transport time to win before the sole desired-peer actor
2425        // re-examines it; 75ms allowed those observers to redial against an
2426        // already-arriving replacement and created generation churn.
2427        const RECOVERY_COALESCE_MS: i64 = 1_000;
2428        // A single close is reported by the Iroh connection, the accept loop,
2429        // and sometimes the path watcher. Once the actor has been asked to
2430        // repair that loss, suppress duplicate reports long enough for the
2431        // tie-break winner and session admission to settle. This is a
2432        // per-device fence, not a second retry owner.
2433        const RECOVERY_DUPLICATE_SUPPRESSION_MS: i64 = MANAGED_SETTLE_DEADLINE_MS;
2434        let remote_node_id = remote_node_id.trim();
2435        let Some(remote_device_id) = self
2436            .known_device_ids_by_node
2437            .read()
2438            .unwrap_or_else(|poisoned| poisoned.into_inner())
2439            .get(remote_node_id)
2440            .cloned()
2441        else {
2442            // Do not fan a route-level loss into a full desired-peer scan when
2443            // the route has not been bound to a durable device yet. The next
2444            // provider revision remains the authority for that startup race.
2445            return false;
2446        };
2447        let schedule = {
2448            let mut state = self.external_desired_peer_actor.lock().await;
2449            if !state.active
2450                || self
2451                    .auto_connect_generation
2452                    .load(std::sync::atomic::Ordering::SeqCst)
2453                    != state.generation
2454            {
2455                return false;
2456            }
2457
2458            let now = now_millis_i64();
2459            if state
2460                .recovery_suppressed_until_ms
2461                .get(&remote_device_id)
2462                .is_some_and(|deadline| *deadline > now)
2463            {
2464                return true;
2465            }
2466            state.recovery_device_ids.insert(remote_device_id.clone());
2467            state.recovery_suppressed_until_ms.insert(
2468                remote_device_id,
2469                now.saturating_add(RECOVERY_DUPLICATE_SUPPRESSION_MS),
2470            );
2471
2472            let deadline = now.saturating_add(RECOVERY_COALESCE_MS);
2473            if state
2474                .scheduled_deadline_ms
2475                .is_some_and(|scheduled| scheduled <= deadline)
2476            {
2477                return true;
2478            }
2479            state.schedule_epoch = state.schedule_epoch.saturating_add(1);
2480            state.scheduled_deadline_ms = Some(deadline);
2481            (deadline, state.schedule_epoch, state.generation)
2482        };
2483
2484        let client = Arc::new(self.clone());
2485        tokio::spawn(async move {
2486            let (deadline, epoch, generation) = schedule;
2487            let delay_ms = deadline.saturating_sub(now_millis_i64()).max(0) as u64;
2488            tokio::time::sleep(std::time::Duration::from_millis(delay_ms)).await;
2489            let should_wake = {
2490                let mut state = client.external_desired_peer_actor.lock().await;
2491                if !state.active
2492                    || state.generation != generation
2493                    || state.schedule_epoch != epoch
2494                    || client
2495                        .auto_connect_generation
2496                        .load(std::sync::atomic::Ordering::SeqCst)
2497                        != generation
2498                {
2499                    false
2500                } else {
2501                    state.scheduled_deadline_ms = None;
2502                    true
2503                }
2504            };
2505            if should_wake {
2506                let _ = wake_native_external_auto_connect_boxed(client).await;
2507            }
2508        });
2509        true
2510    }
2511
2512    pub(crate) async fn external_auto_connect_is_active(&self) -> bool {
2513        self.external_desired_peer_actor.lock().await.active
2514    }
2515
2516    pub async fn stop_external_auto_connect(&self) {
2517        let mut state = self.external_desired_peer_actor.lock().await;
2518        state.active = false;
2519        state.mailbox.peers.clear();
2520        state.peer_evidence.clear();
2521        state.failure_count.clear();
2522        state.retry_not_before_ms.clear();
2523        state.non_initiator_not_before_ms.clear();
2524        state.scheduled_deadline_ms = None;
2525        state.schedule_epoch = state.schedule_epoch.saturating_add(1);
2526        state.wake_pending = false;
2527        state.recovery_device_ids.clear();
2528        state.recovery_suppressed_until_ms.clear();
2529    }
2530}
2531
2532#[cfg(target_arch = "wasm32")]
2533impl Client {
2534    pub(crate) fn start_browser_auto_connect(
2535        self: &Arc<Self>,
2536        user_id: String,
2537        local_device_id: String,
2538    ) -> anyhow::Result<()> {
2539        let user_id = user_id.trim().to_string();
2540        let local_device_id = local_device_id.trim().to_string();
2541        if user_id.is_empty() || local_device_id.is_empty() {
2542            return Err(anyhow::anyhow!(
2543                "browser auto-connect requires non-empty user and local device ids"
2544            ));
2545        }
2546
2547        let key = (user_id, local_device_id.clone());
2548        let actor_key = browser_actor_key(self);
2549        let duplicate_actor = BROWSER_AUTO_CONNECT_ACTORS.with(|actors| {
2550            actors.borrow().get(&actor_key).cloned().filter(|actor| {
2551                let state = actor.borrow();
2552                browser_actor_is_current(&state) && state.local_device_id == local_device_id
2553            })
2554        });
2555        {
2556            let guard = self
2557                .auto_connect_loop_key
2558                .lock()
2559                .unwrap_or_else(|poisoned| poisoned.into_inner());
2560            if guard.as_ref() == Some(&key) {
2561                if let Some(actor) = duplicate_actor {
2562                    wake_browser_auto_connect_actor(actor);
2563                    return Ok(());
2564                }
2565            }
2566        }
2567
2568        let generation = {
2569            let mut guard = self
2570                .auto_connect_loop_key
2571                .lock()
2572                .unwrap_or_else(|poisoned| poisoned.into_inner());
2573            self.known_device_ids_by_node
2574                .write()
2575                .unwrap_or_else(|poisoned| poisoned.into_inner())
2576                .clear();
2577            *guard = Some(key);
2578            self.auto_connect_generation
2579                .fetch_add(1, std::sync::atomic::Ordering::SeqCst)
2580                + 1
2581        };
2582
2583        let actor = std::rc::Rc::new(std::cell::RefCell::new(BrowserAutoConnectActorState {
2584            active: true,
2585            client: Arc::downgrade(self),
2586            generation,
2587            local_device_id,
2588            mailbox: BrowserDesiredPeerMailbox::default(),
2589            peer_evidence: std::collections::HashMap::new(),
2590            failure_count: std::collections::HashMap::new(),
2591            retry_not_before_ms: std::collections::HashMap::new(),
2592            non_initiator_not_before_ms: std::collections::HashMap::new(),
2593            scheduled_deadline_ms: None,
2594            schedule_epoch: 0,
2595            reconcile_running: false,
2596            wake_pending: false,
2597        }));
2598        BROWSER_AUTO_CONNECT_ACTORS.with(|actors| {
2599            if let Some(previous) = actors.borrow_mut().insert(actor_key, actor) {
2600                deactivate_browser_actor(&mut previous.borrow_mut());
2601            }
2602        });
2603        Ok(())
2604    }
2605
2606    pub(crate) fn submit_browser_desired_peers(
2607        self: &Arc<Self>,
2608        revision: u64,
2609        peers_json: &str,
2610    ) -> anyhow::Result<bool> {
2611        let peers: Vec<BrowserDesiredPeer> = serde_json::from_str(peers_json)
2612            .map_err(|error| anyhow::anyhow!("invalid browser desired-peer payload: {error}"))?;
2613        let actor = BROWSER_AUTO_CONNECT_ACTORS
2614            .with(|actors| actors.borrow().get(&browser_actor_key(self)).cloned());
2615        let Some(actor) = actor else {
2616            return Err(anyhow::anyhow!("browser auto-connect is not started"));
2617        };
2618
2619        let (accepted, current_bindings, withdrawn) = {
2620            let mut state = actor.borrow_mut();
2621            if !browser_actor_is_current(&state) {
2622                (false, Vec::new(), Vec::new())
2623            } else if let Some(withdrawn) = state.mailbox.accept_with_withdrawn(revision, peers) {
2624                let desired_ids = state
2625                    .mailbox
2626                    .peers
2627                    .keys()
2628                    .cloned()
2629                    .collect::<std::collections::HashSet<_>>();
2630                state
2631                    .peer_evidence
2632                    .retain(|device_id, _| desired_ids.contains(device_id));
2633                state
2634                    .failure_count
2635                    .retain(|device_id, _| desired_ids.contains(device_id));
2636                state
2637                    .retry_not_before_ms
2638                    .retain(|device_id, _| desired_ids.contains(device_id));
2639                state
2640                    .non_initiator_not_before_ms
2641                    .retain(|device_id, _| desired_ids.contains(device_id));
2642
2643                let evidence = state
2644                    .mailbox
2645                    .peers
2646                    .iter()
2647                    .map(|(device_id, peer)| (device_id.clone(), browser_peer_evidence(peer)))
2648                    .collect::<Vec<_>>();
2649                for (device_id, evidence) in evidence {
2650                    let changed = state
2651                        .peer_evidence
2652                        .insert(device_id.clone(), evidence.clone())
2653                        .is_some_and(|previous| previous != evidence);
2654                    if changed {
2655                        state.failure_count.remove(&device_id);
2656                        state.retry_not_before_ms.remove(&device_id);
2657                        state.non_initiator_not_before_ms.remove(&device_id);
2658                    }
2659                }
2660                state.schedule_epoch = state.schedule_epoch.saturating_add(1);
2661                state.scheduled_deadline_ms = None;
2662                let bindings = state
2663                    .mailbox
2664                    .peers
2665                    .values()
2666                    .filter(|peer| peer.online)
2667                    .filter_map(|peer| {
2668                        let ticket_node_id = peer.ticket.as_deref().and_then(|ticket| {
2669                            let (iroh_ticket, _) = crate::session_token::split_ticket(ticket);
2670                            parse_endpoint_ticket(iroh_ticket)
2671                                .ok()
2672                                .map(|address| address.id.to_string())
2673                        });
2674                        let node_id = match (peer.node_id.as_deref(), ticket_node_id.as_deref()) {
2675                            (Some(directory_node_id), Some(ticket_node_id))
2676                                if directory_node_id != ticket_node_id =>
2677                            {
2678                                return None;
2679                            }
2680                            (Some(directory_node_id), _) => directory_node_id.to_string(),
2681                            (None, Some(ticket_node_id)) => ticket_node_id.to_string(),
2682                            (None, None) => return None,
2683                        };
2684                        Some((peer.device_id.clone(), node_id))
2685                    })
2686                    .collect::<Vec<_>>();
2687                (true, bindings, withdrawn)
2688            } else {
2689                (false, Vec::new(), Vec::new())
2690            }
2691        };
2692        if accepted {
2693            // This is the cancellation token for any older desired-peer pass.
2694            // It must be visible before waking the serial actor, because an old
2695            // connect may currently be suspended inside Iroh or admission.
2696            for (device_id, node_id) in current_bindings {
2697                self.observe_authoritative_device_node(&device_id, &node_id);
2698            }
2699            if !withdrawn.is_empty() {
2700                let client = Arc::clone(self);
2701                let actor_for_retirement = actor.clone();
2702                wasm_bindgen_futures::spawn_local(async move {
2703                    for peer in withdrawn {
2704                        let still_withdrawn = {
2705                            let state = actor_for_retirement.borrow();
2706                            browser_actor_is_current(&state)
2707                                && state.mailbox.revision == revision
2708                                && state
2709                                    .mailbox
2710                                    .peers
2711                                    .get(&peer.device_id)
2712                                    .is_none_or(|current| !current.online)
2713                        };
2714                        if still_withdrawn {
2715                            client
2716                                .retire_withdrawn_desired_peer_connections(
2717                                    &peer.device_id,
2718                                    peer.node_id.as_deref(),
2719                                )
2720                                .await;
2721                        }
2722                    }
2723                });
2724            }
2725            wake_browser_auto_connect_actor(actor);
2726        }
2727        Ok(accepted)
2728    }
2729
2730    pub(crate) fn wake_browser_auto_connect(self: &Arc<Self>) -> bool {
2731        let actor = BROWSER_AUTO_CONNECT_ACTORS
2732            .with(|actors| actors.borrow().get(&browser_actor_key(self)).cloned());
2733        let Some(actor) = actor else {
2734            return false;
2735        };
2736        if !browser_actor_is_current(&actor.borrow()) {
2737            return false;
2738        }
2739        wake_browser_auto_connect_actor(actor);
2740        true
2741    }
2742
2743    pub(crate) fn stop_browser_auto_connect(self: &Arc<Self>) {
2744        let actor = BROWSER_AUTO_CONNECT_ACTORS
2745            .with(|actors| actors.borrow_mut().remove(&browser_actor_key(self)));
2746        if let Some(actor) = actor {
2747            deactivate_browser_actor(&mut actor.borrow_mut());
2748        }
2749    }
2750}
2751
2752#[cfg(test)]
2753mod tests {
2754    use super::*;
2755
2756    #[cfg(not(target_arch = "wasm32"))]
2757    #[tokio::test]
2758    async fn auto_connect_disabled_during_reconcile_preserves_manual_transport() {
2759        let client = Arc::new(Client::new_with_app_tag(
2760            crate::test_constants::TEST_PROJECT_ID.to_string(),
2761            "exclusion-race".to_string(),
2762            Box::new(|| None),
2763        ));
2764        let remote = Client::new_with_app_tag(
2765            crate::test_constants::TEST_PROJECT_ID.to_string(),
2766            "exclusion-race".to_string(),
2767            Box::new(|| None),
2768        );
2769        for owner in [&*client, &remote] {
2770            let endpoint = iroh::Endpoint::builder(iroh::endpoint::presets::N0)
2771                .alpns(vec![crate::native_node::PlutoniumProtocol::ALPN.to_vec()])
2772                .relay_mode(iroh::RelayMode::Disabled)
2773                .bind_addr("127.0.0.1:0".parse::<std::net::SocketAddr>().unwrap())
2774                .unwrap()
2775                .bind()
2776                .await
2777                .unwrap();
2778            owner.adopt_endpoint(endpoint).await;
2779        }
2780        let ticket = remote.endpoint_ticket().await.unwrap();
2781        let remote_node_id = remote.current_node_id().await.unwrap();
2782        let endpoint_id = remote_node_id.parse().unwrap();
2783        let connected = client
2784            .connect_device(Some("remote"), &ticket)
2785            .await
2786            .unwrap();
2787        let original = client.get_connection(endpoint_id).await.unwrap();
2788        let original_stable_id = crate::transport_generation::for_connection(&original);
2789        let peer = BrowserDesiredPeer {
2790            device_id: "remote".to_string(),
2791            node_id: Some(remote_node_id),
2792            ticket: Some(ticket),
2793            online: true,
2794            session_id: Some("runtime-1".to_string()),
2795            excluded_peers: Vec::new(),
2796            topology_role: None,
2797            topology_revision: None,
2798        };
2799        let generation = client
2800            .auto_connect_generation
2801            .load(std::sync::atomic::Ordering::SeqCst);
2802        {
2803            let mut actor = client.external_desired_peer_actor.lock().await;
2804            actor.active = true;
2805            actor.generation = generation;
2806            actor.local_device_id = "local".to_string();
2807            actor
2808                .mailbox
2809                .accept_with_withdrawn(1, vec![peer.clone()])
2810                .unwrap();
2811        }
2812        // Pause the production actor at the same connect gate shared by a
2813        // manual request. Restore the durable preference while it is awaiting
2814        // that gate, as a relaunched WebView can do during admission.
2815        let gate = client.managed_connect_gate(&connected.connection_id).await;
2816        let guard = gate.lock().await;
2817        let reconcile =
2818            reconcile_native_external_desired_peer(&client, generation, 1, "local", &peer, None);
2819        tokio::pin!(reconcile);
2820        assert!(
2821            tokio::time::timeout(std::time::Duration::from_millis(100), &mut reconcile)
2822                .await
2823                .is_err()
2824        );
2825        client.set_auto_connect_excluded("remote", true);
2826        drop(guard);
2827        assert_eq!(
2828            tokio::time::timeout(std::time::Duration::from_secs(5), reconcile)
2829                .await
2830                .unwrap(),
2831            BrowserPeerReconcileOutcome::Ineligible,
2832        );
2833        assert!(client.is_auto_connect_excluded("remote"));
2834        assert!(
2835            original.close_reason().is_none(),
2836            "auto-connect policy must not close a manual transport"
2837        );
2838        let current = client
2839            .get_connection(endpoint_id)
2840            .await
2841            .expect("manual connection survives");
2842        assert_eq!(
2843            crate::transport_generation::for_connection(&current),
2844            original_stable_id
2845        );
2846        assert!(client.is_connection_transport_alive(endpoint_id).await);
2847    }
2848
2849    fn replacement_snapshot(last_reconnect_reason: Option<&str>) -> ConnectionHealthView {
2850        ConnectionHealthView {
2851            connection_id: "conn-1".to_string(),
2852            device_id: Some("device-1".to_string()),
2853            device_id_hint: Some("device-1".to_string()),
2854            node_id: Some("node-1".to_string()),
2855            active_transport_stable_id: None,
2856            transport_generation: 2,
2857            route_generation: 0,
2858            status: ManagedHealth::AwaitingReplacement,
2859            settled_ready: false,
2860            readiness_state: ReadinessState::AwaitingReplacement,
2861            replacement_pending: true,
2862            last_lifecycle_transition_at_ms: 1_000,
2863            readiness_reason: "missing-active-transport".to_string(),
2864            transition_count: 1,
2865            connecting_transition_count: 0,
2866            replacement_count: 1,
2867            retire_count: 0,
2868            last_disconnect_reason: None,
2869            last_reconnect_reason: last_reconnect_reason.map(str::to_string),
2870        }
2871    }
2872
2873    #[test]
2874    fn runtime_instance_change_is_observed_only_after_the_initial_generation() {
2875        let mut last_known = std::collections::HashMap::new();
2876
2877        assert!(!observe_remote_runtime_instance(
2878            &mut last_known,
2879            "device-1",
2880            Some("runtime-1"),
2881        ));
2882        assert!(!observe_remote_runtime_instance(
2883            &mut last_known,
2884            "device-1",
2885            Some("runtime-1"),
2886        ));
2887        assert!(observe_remote_runtime_instance(
2888            &mut last_known,
2889            "device-1",
2890            Some("runtime-2"),
2891        ));
2892    }
2893
2894    #[cfg(not(target_arch = "wasm32"))]
2895    #[test]
2896    fn delayed_runtime_instance_update_preserves_a_responsive_replacement() {
2897        assert!(!runtime_instance_change_requires_replacement(true, true));
2898    }
2899
2900    #[cfg(not(target_arch = "wasm32"))]
2901    #[test]
2902    fn runtime_instance_update_replaces_an_unresponsive_transport() {
2903        assert!(runtime_instance_change_requires_replacement(true, false));
2904        assert!(!runtime_instance_change_requires_replacement(false, false));
2905    }
2906
2907    #[test]
2908    fn ticket_change_clears_transient_retry_without_retaining_ticket_material() {
2909        let mut last_known = std::collections::HashMap::new();
2910        let mut last_attempt_at = std::collections::HashMap::new();
2911        let mut failure_count = std::collections::HashMap::new();
2912        let mut failure_backoff_until = std::collections::HashMap::new();
2913
2914        assert!(!reconcile_remote_ticket_retry_state(
2915            &mut last_known,
2916            "device-1",
2917            "ticket-1.token-1",
2918            &mut last_attempt_at,
2919            &mut failure_count,
2920            &mut failure_backoff_until,
2921        ));
2922        last_attempt_at.insert("device-1".to_string(), 100);
2923        failure_count.insert("device-1".to_string(), 4);
2924        failure_backoff_until.insert("device-1".to_string(), 8_100);
2925
2926        assert!(!reconcile_remote_ticket_retry_state(
2927            &mut last_known,
2928            "device-1",
2929            "ticket-1.token-1",
2930            &mut last_attempt_at,
2931            &mut failure_count,
2932            &mut failure_backoff_until,
2933        ));
2934        assert_eq!(last_attempt_at.get("device-1"), Some(&100));
2935        assert_eq!(failure_count.get("device-1"), Some(&4));
2936        assert_eq!(failure_backoff_until.get("device-1"), Some(&8_100));
2937
2938        assert!(reconcile_remote_ticket_retry_state(
2939            &mut last_known,
2940            "device-1",
2941            "ticket-1.token-2",
2942            &mut last_attempt_at,
2943            &mut failure_count,
2944            &mut failure_backoff_until,
2945        ));
2946        assert!(!last_attempt_at.contains_key("device-1"));
2947        assert!(!failure_count.contains_key("device-1"));
2948        assert!(!failure_backoff_until.contains_key("device-1"));
2949
2950        let stored = last_known
2951            .get("device-1::ticket")
2952            .expect("ticket fingerprint");
2953        assert_eq!(stored.len(), 16);
2954        assert!(!stored.contains("token-2"));
2955    }
2956
2957    #[test]
2958    fn replacement_window_suppresses_auto_connect_for_active_replacement_churn() {
2959        let snapshot = replacement_snapshot(Some("incoming-replacement-churn"));
2960
2961        assert!(should_suppress_auto_connect_for_replacement_window(
2962            &snapshot,
2963            snapshot.last_lifecycle_transition_at_ms + 1_000,
2964            false,
2965        ));
2966        assert!(
2967            !should_suppress_auto_connect_for_replacement_window(
2968                &snapshot,
2969                snapshot.last_lifecycle_transition_at_ms + 1_000,
2970                true,
2971            ),
2972            "replacement suppression must not bypass positive liveness checks for an owned transport"
2973        );
2974    }
2975
2976    #[test]
2977    fn transient_disconnect_replacement_window_allows_auto_connect_redial() {
2978        let snapshot = replacement_snapshot(Some("transient-disconnect"));
2979
2980        assert!(!should_suppress_auto_connect_for_replacement_window(
2981            &snapshot,
2982            snapshot.last_lifecycle_transition_at_ms + 1_000,
2983            false,
2984        ));
2985    }
2986
2987    #[test]
2988    fn expired_replacement_window_allows_auto_connect_redial() {
2989        let snapshot = replacement_snapshot(Some("incoming-replacement-churn"));
2990
2991        assert!(!should_suppress_auto_connect_for_replacement_window(
2992            &snapshot,
2993            snapshot.last_lifecycle_transition_at_ms + MANAGED_SETTLE_DEADLINE_MS,
2994            false,
2995        ));
2996    }
2997
2998    #[test]
2999    fn remote_exclusion_matches_canonical_device_id_case_insensitively() {
3000        let excluded_peers = vec!["  DEVICE-LOCAL  ".to_string()];
3001
3002        assert!(remote_excludes_local_device(
3003            &excluded_peers,
3004            "device-local"
3005        ));
3006        assert!(!remote_excludes_local_device(
3007            &excluded_peers,
3008            "device-other"
3009        ));
3010        assert!(!remote_excludes_local_device(&excluded_peers, ""));
3011    }
3012
3013    #[test]
3014    fn browser_desired_peer_mailbox_fences_stale_revisions_and_replaces_state() {
3015        let mut mailbox = BrowserDesiredPeerMailbox::default();
3016        let peer = BrowserDesiredPeer {
3017            device_id: " device-1 ".to_string(),
3018            node_id: Some(" node-1 ".to_string()),
3019            ticket: Some(" ticket-1 ".to_string()),
3020            online: true,
3021            session_id: Some(" runtime-1 ".to_string()),
3022            excluded_peers: vec![" DEVICE-LOCAL ".to_string()],
3023            topology_role: None,
3024            topology_revision: None,
3025        };
3026
3027        assert!(mailbox.accept_with_withdrawn(2, vec![peer]).is_some());
3028        assert!(mailbox.accept_with_withdrawn(1, Vec::new()).is_none());
3029        assert!(mailbox.accept_with_withdrawn(2, Vec::new()).is_none());
3030        assert_eq!(mailbox.revision, 2);
3031        let stored = mailbox.peers.get("device-1").expect("desired peer");
3032        assert_eq!(stored.node_id.as_deref(), Some("node-1"));
3033        assert_eq!(stored.excluded_peers, vec!["device-local"]);
3034
3035        let withdrawn = mailbox
3036            .accept_with_withdrawn(3, Vec::new())
3037            .expect("newer revision is accepted");
3038        assert_eq!(withdrawn.len(), 1);
3039        assert_eq!(withdrawn[0].device_id, "device-1");
3040        assert!(mailbox.peers.is_empty());
3041        assert!(mailbox.device_is_withdrawn("device-1"));
3042        assert_eq!(
3043            mailbox.withdrawn_device_for_node("node-1").as_deref(),
3044            Some("device-1")
3045        );
3046
3047        let mut restored = withdrawn[0].clone();
3048        restored.online = true;
3049        assert!(mailbox
3050            .accept_with_withdrawn(4, vec![restored])
3051            .expect("restored revision is accepted")
3052            .is_empty());
3053        assert!(!mailbox.device_is_withdrawn("device-1"));
3054        assert_eq!(mailbox.withdrawn_device_for_node("node-1"), None);
3055    }
3056
3057    #[test]
3058    fn current_user_device_assignment_requires_exact_online_runtime_evidence() {
3059        let endpoint_id = iroh::SecretKey::generate().public();
3060        let node_id = endpoint_id.to_string();
3061        let iroh_ticket =
3062            iroh_tickets::endpoint::EndpointTicket::new(iroh::EndpointAddr::new(endpoint_id))
3063                .to_string();
3064        let ticket =
3065            crate::session_token::build_ticket(&iroh_ticket, "user-device-token", "user-device", 0);
3066        let peer = BrowserDesiredPeer {
3067            device_id: "device-1".to_string(),
3068            node_id: Some(node_id.clone()),
3069            ticket: Some(ticket.clone()),
3070            online: true,
3071            session_id: Some("runtime-1".to_string()),
3072            excluded_peers: Vec::new(),
3073            topology_role: None,
3074            topology_revision: None,
3075        };
3076        let evidence = browser_peer_evidence(&peer);
3077        let mut mailbox = BrowserDesiredPeerMailbox::default();
3078        assert!(mailbox.accept(1, vec![peer.clone()]));
3079        assert_eq!(
3080            mailbox.current_online_device_for_node(&node_id).as_deref(),
3081            Some("device-1")
3082        );
3083        assert!(
3084            mailbox.current_online_assignment_matches("device-1", &node_id, &ticket, &evidence,)
3085        );
3086        assert!(!mailbox.current_online_assignment_matches(
3087            "device-1",
3088            &node_id,
3089            "stale-ticket",
3090            &evidence,
3091        ));
3092
3093        let mut replacement = peer.clone();
3094        replacement.session_id = Some("runtime-2".to_string());
3095        assert!(mailbox.accept(2, vec![replacement.clone()]));
3096        assert!(
3097            !mailbox.current_online_assignment_matches("device-1", &node_id, &ticket, &evidence,)
3098        );
3099        let replacement_evidence = browser_peer_evidence(&replacement);
3100        assert!(mailbox.current_online_assignment_matches(
3101            "device-1",
3102            &node_id,
3103            &ticket,
3104            &replacement_evidence,
3105        ));
3106
3107        replacement.online = false;
3108        assert!(mailbox.accept(3, vec![replacement]));
3109        assert_eq!(mailbox.current_online_device_for_node(&node_id), None);
3110        assert!(!mailbox.current_online_assignment_matches(
3111            "device-1",
3112            &node_id,
3113            &ticket,
3114            &replacement_evidence,
3115        ));
3116    }
3117
3118    #[test]
3119    fn browser_desired_peer_mailbox_withdraws_live_route_when_device_goes_offline() {
3120        let mut mailbox = BrowserDesiredPeerMailbox::default();
3121        let online = BrowserDesiredPeer {
3122            device_id: "device-1".to_string(),
3123            node_id: Some("node-1".to_string()),
3124            ticket: Some("ticket-1".to_string()),
3125            online: true,
3126            session_id: Some("runtime-1".to_string()),
3127            excluded_peers: Vec::new(),
3128            topology_role: None,
3129            topology_revision: None,
3130        };
3131        assert!(mailbox
3132            .accept_with_withdrawn(1, vec![online.clone()])
3133            .expect("initial revision is accepted")
3134            .is_empty());
3135
3136        let mut offline = online.clone();
3137        offline.online = false;
3138        let withdrawn = mailbox
3139            .accept_with_withdrawn(2, vec![offline])
3140            .expect("offline revision is accepted");
3141        assert_eq!(withdrawn, vec![online]);
3142        assert!(mailbox.peers.contains_key("device-1"));
3143        assert!(!mailbox.peers["device-1"].online);
3144        assert!(mailbox.device_is_withdrawn("device-1"));
3145        assert_eq!(
3146            mailbox.withdrawn_device_for_node("node-1").as_deref(),
3147            Some("device-1")
3148        );
3149
3150        assert!(mailbox
3151            .accept_with_withdrawn(3, vec![mailbox.peers["device-1"].clone()])
3152            .expect("repeated offline revision is accepted")
3153            .is_empty());
3154    }
3155
3156    #[test]
3157    fn offline_boot_is_open() {
3158        let mut mailbox = BrowserDesiredPeerMailbox::default();
3159        let offline = BrowserDesiredPeer {
3160            device_id: "device-1".to_string(),
3161            node_id: Some("node-1".to_string()),
3162            ticket: Some("ticket-1".to_string()),
3163            online: false,
3164            session_id: None,
3165            excluded_peers: Vec::new(),
3166            topology_role: None,
3167            topology_revision: None,
3168        };
3169
3170        assert!(mailbox
3171            .accept_with_withdrawn(1, vec![offline])
3172            .expect("initial offline roster is accepted")
3173            .is_empty());
3174        assert!(!mailbox.device_is_withdrawn("device-1"));
3175        assert_eq!(mailbox.withdrawn_device_for_node("node-1"), None);
3176    }
3177
3178    #[test]
3179    fn candidate_observation_expiry_removes_dial_intent_without_withdrawal() {
3180        let mut mailbox = BrowserDesiredPeerMailbox::default();
3181        let peer = BrowserDesiredPeer {
3182            device_id: "device-1".to_string(),
3183            node_id: Some("node-1".to_string()),
3184            ticket: Some("ticket-1".to_string()),
3185            online: true,
3186            session_id: None,
3187            excluded_peers: Vec::new(),
3188            topology_role: None,
3189            topology_revision: None,
3190        };
3191        assert!(mailbox
3192            .accept_with_withdrawn(1, vec![peer.clone()])
3193            .unwrap()
3194            .is_empty());
3195        assert!(mailbox
3196            .accept_with_observation_expired(2, Vec::new(), "device-1")
3197            .expect("newer observation revision")
3198            .is_empty());
3199        assert!(mailbox.peers.is_empty());
3200        assert!(!mailbox.device_is_withdrawn("device-1"));
3201        assert_eq!(mailbox.withdrawn_device_for_node("node-1"), None);
3202        assert!(mailbox
3203            .accept_with_observation_expired(2, Vec::new(), "device-1")
3204            .is_none());
3205
3206        assert!(mailbox
3207            .accept_with_withdrawn(3, vec![peer])
3208            .expect("rediscovery revision")
3209            .is_empty());
3210        assert!(mailbox.peers.contains_key("device-1"));
3211        assert!(!mailbox.device_is_withdrawn("device-1"));
3212    }
3213
3214    #[test]
3215    fn topology_backups_retire_only_after_current_active_edges_are_ready() {
3216        let peer = |device_id: &str, role: SparseTopologyRole, topology_revision: u64| {
3217            BrowserDesiredPeer {
3218                device_id: device_id.to_string(),
3219                node_id: Some(format!("node-{device_id}")),
3220                ticket: Some(format!("ticket-{device_id}")),
3221                online: true,
3222                session_id: Some(format!("runtime-{device_id}")),
3223                excluded_peers: Vec::new(),
3224                topology_role: Some(role),
3225                topology_revision: Some(topology_revision),
3226            }
3227        };
3228        let active_a = peer("active-a", SparseTopologyRole::Active, 7);
3229        let active_b = peer("active-b", SparseTopologyRole::Active, 7);
3230        let backup = peer("backup", SparseTopologyRole::Backup, 7);
3231        let membership_only = |mut peer: BrowserDesiredPeer| {
3232            peer.node_id = None;
3233            peer.ticket = None;
3234            peer.topology_role = None;
3235            peer.topology_revision = None;
3236            peer
3237        };
3238        let mut mailbox = BrowserDesiredPeerMailbox::default();
3239        assert!(mailbox
3240            .accept_with_withdrawn(10, vec![active_a.clone(), active_b.clone(), backup.clone()],)
3241            .expect("initial topology")
3242            .is_empty());
3243
3244        assert!(mailbox
3245            .acknowledge_topology_peer_ready("active-a", Some(7))
3246            .is_empty());
3247        assert!(mailbox.peers.contains_key("backup"));
3248        mailbox.observe_topology_peer_not_ready("active-a", Some(7));
3249        assert!(mailbox
3250            .acknowledge_topology_peer_ready("active-b", Some(7))
3251            .is_empty());
3252        assert!(mailbox.peers.contains_key("backup"));
3253        assert!(mailbox
3254            .acknowledge_topology_peer_ready("active-b", Some(6))
3255            .is_empty());
3256        assert!(mailbox.peers.contains_key("backup"));
3257
3258        assert_eq!(
3259            mailbox.acknowledge_topology_peer_ready("active-a", Some(7)),
3260            vec![backup.clone()],
3261        );
3262        assert!(!mailbox.peers.contains_key("backup"));
3263        assert!(mailbox.device_is_withdrawn("backup"));
3264
3265        // An unrelated public-membership revision cannot revive a retired
3266        // backup while the gateway topology generation is unchanged.
3267        assert!(mailbox
3268            .accept_with_withdrawn(11, vec![active_a.clone(), active_b.clone(), backup.clone()],)
3269            .expect("same topology revision")
3270            .is_empty());
3271        assert!(!mailbox.peers.contains_key("backup"));
3272
3273        // A new topology clears readiness. A late acknowledgement from the
3274        // prior generation cannot retire its replacement overlap.
3275        let next_active = peer("active-next", SparseTopologyRole::Active, 8);
3276        let next_backup = peer("backup-next", SparseTopologyRole::Backup, 8);
3277        mailbox
3278            .accept_with_withdrawn(
3279                12,
3280                vec![
3281                    next_active.clone(),
3282                    next_backup.clone(),
3283                    membership_only(active_a.clone()),
3284                    membership_only(active_b.clone()),
3285                ],
3286            )
3287            .expect("next topology revision");
3288        assert!(mailbox
3289            .acknowledge_topology_peer_ready("active-a", Some(7))
3290            .is_empty());
3291        assert!(!mailbox.peers.contains_key("backup-next"));
3292        assert_eq!(
3293            mailbox.peers["active-a"].topology_role,
3294            Some(SparseTopologyRole::Backup),
3295        );
3296
3297        // A third lease arriving before revision 8 settles retains only the
3298        // last Rust-proven revision 7 active set. It does not accumulate the
3299        // speculative revision 8 active or backup set.
3300        let latest_active = peer("active-latest", SparseTopologyRole::Active, 9);
3301        let speculative_overlap = peer("active-next", SparseTopologyRole::Backup, 9);
3302        mailbox
3303            .accept_with_withdrawn(
3304                13,
3305                vec![
3306                    latest_active.clone(),
3307                    speculative_overlap,
3308                    membership_only(active_a.clone()),
3309                    membership_only(active_b.clone()),
3310                    membership_only(next_active.clone()),
3311                ],
3312            )
3313            .expect("rapid topology revision");
3314        assert_eq!(
3315            mailbox
3316                .peers
3317                .values()
3318                .filter(|peer| peer.topology_role == Some(SparseTopologyRole::Backup))
3319                .map(|peer| peer.device_id.as_str())
3320                .collect::<std::collections::BTreeSet<_>>(),
3321            std::collections::BTreeSet::from(["active-a", "active-b"]),
3322        );
3323        assert_eq!(
3324            mailbox.acknowledge_topology_peer_ready("active-latest", Some(9)),
3325            vec![active_a.clone(), active_b.clone()]
3326                .into_iter()
3327                .map(|mut peer| {
3328                    peer.topology_role = Some(SparseTopologyRole::Backup);
3329                    peer.topology_revision = Some(9);
3330                    peer
3331                })
3332                .collect::<Vec<_>>(),
3333        );
3334
3335        // Restarting the actor starts with no inherited readiness proof.
3336        let mut restarted = BrowserDesiredPeerMailbox::default();
3337        restarted
3338            .accept_with_withdrawn(1, vec![next_active, next_backup])
3339            .expect("restarted topology");
3340        assert!(restarted.peers.contains_key("backup-next"));
3341    }
3342
3343    #[test]
3344    fn browser_peer_evidence_fingerprints_ticket_instead_of_retaining_it() {
3345        let peer = BrowserDesiredPeer {
3346            device_id: "device-1".to_string(),
3347            node_id: Some("node-1".to_string()),
3348            ticket: Some("iroh-ticket.sensitive-token-material".to_string()),
3349            online: true,
3350            session_id: Some("runtime-1".to_string()),
3351            excluded_peers: Vec::new(),
3352            topology_role: None,
3353            topology_revision: None,
3354        };
3355
3356        let evidence = browser_peer_evidence(&peer);
3357        assert!(evidence.starts_with("node-1|"));
3358        assert!(!evidence.contains("sensitive-token-material"));
3359    }
3360
3361    #[test]
3362    fn desired_peer_attempt_survives_unrelated_mailbox_revision() {
3363        let mut mailbox = BrowserDesiredPeerMailbox::default();
3364        let tracked = BrowserDesiredPeer {
3365            device_id: "device-tracked".to_string(),
3366            node_id: Some("node-tracked".to_string()),
3367            ticket: Some("ticket-tracked".to_string()),
3368            online: true,
3369            session_id: Some("runtime-tracked".to_string()),
3370            excluded_peers: Vec::new(),
3371            topology_role: None,
3372            topology_revision: None,
3373        };
3374        let unrelated = BrowserDesiredPeer {
3375            device_id: "device-unrelated".to_string(),
3376            node_id: Some("node-unrelated".to_string()),
3377            ticket: Some("ticket-unrelated".to_string()),
3378            online: true,
3379            session_id: Some("runtime-unrelated".to_string()),
3380            excluded_peers: Vec::new(),
3381            topology_role: None,
3382            topology_revision: None,
3383        };
3384
3385        assert!(mailbox.accept(1, vec![tracked.clone()]));
3386        assert!(desired_peer_evidence_is_current(&mailbox, &tracked));
3387
3388        assert!(mailbox.accept(2, vec![tracked.clone(), unrelated.clone()]));
3389        assert!(desired_peer_evidence_is_current(&mailbox, &tracked));
3390
3391        let mut changed = tracked.clone();
3392        changed.session_id = Some("runtime-replacement".to_string());
3393        assert!(mailbox.accept(3, vec![changed, unrelated.clone()]));
3394        assert!(!desired_peer_evidence_is_current(&mailbox, &tracked));
3395
3396        assert!(mailbox.accept(4, vec![unrelated]));
3397        assert!(!desired_peer_evidence_is_current(&mailbox, &tracked));
3398    }
3399
3400    #[test]
3401    fn browser_desired_peer_tie_break_has_one_initial_dialer_and_bounded_takeover() {
3402        assert_eq!(
3403            browser_desired_peer_decision("node-z", "node-a", false, 1_000, None),
3404            BrowserDesiredPeerDecision::Dial
3405        );
3406        assert_eq!(
3407            browser_desired_peer_decision("node-a", "node-z", false, 1_000, None),
3408            BrowserDesiredPeerDecision::WaitUntil(4_000)
3409        );
3410        assert_eq!(
3411            browser_desired_peer_decision("node-a", "node-z", false, 4_000, Some(4_000)),
3412            BrowserDesiredPeerDecision::Dial
3413        );
3414        assert_eq!(
3415            browser_desired_peer_decision("node-a", "node-z", true, 1_000, None),
3416            BrowserDesiredPeerDecision::ObserveExisting
3417        );
3418    }
3419
3420    #[test]
3421    fn browser_retry_deadline_only_replaces_with_earlier_work_or_cancellation() {
3422        assert!(!should_replace_browser_retry_deadline(None, None));
3423        assert!(should_replace_browser_retry_deadline(None, Some(5_000)));
3424        assert!(!should_replace_browser_retry_deadline(
3425            Some(5_000),
3426            Some(6_000)
3427        ));
3428        assert!(should_replace_browser_retry_deadline(
3429            Some(5_000),
3430            Some(4_000)
3431        ));
3432        assert!(should_replace_browser_retry_deadline(Some(5_000), None));
3433    }
3434
3435    #[test]
3436    fn browser_failure_backoff_is_bounded_for_recovery() {
3437        assert_eq!(browser_auto_connect_failure_backoff_ms(1), 1_000);
3438        assert_eq!(browser_auto_connect_failure_backoff_ms(2), 2_000);
3439        assert_eq!(browser_auto_connect_failure_backoff_ms(10), 8_000);
3440    }
3441
3442    #[test]
3443    fn reconcile_deadlines_replace_expired_non_initiator_wait_with_failure_backoff() {
3444        let (retry_deadline, non_initiator_deadline) =
3445            browser_reconcile_deadlines(BrowserPeerReconcileOutcome::Retry, Some(9_000));
3446        assert_eq!(retry_deadline, Some(9_000));
3447        assert_eq!(non_initiator_deadline, None);
3448
3449        let (retry_deadline, non_initiator_deadline) =
3450            browser_reconcile_deadlines(BrowserPeerReconcileOutcome::WaitUntil(4_000), None);
3451        assert_eq!(retry_deadline, None);
3452        assert_eq!(non_initiator_deadline, Some(4_000));
3453    }
3454
3455    #[cfg(not(target_arch = "wasm32"))]
3456    #[tokio::test]
3457    async fn duplicate_external_auto_connect_preserves_gateway_mailbox_generation() {
3458        let client = std::sync::Arc::new(Client::new_with_app_tag(
3459            crate::test_constants::TEST_PROJECT_ID.to_string(),
3460            "test-app".to_string(),
3461            Box::new(|| None),
3462        ));
3463        client
3464            .start_external_auto_connect("test-user".to_string(), "local-device".to_string())
3465            .await
3466            .expect("external actor starts");
3467        assert!(client
3468            .submit_external_desired_peers(7, "[]")
3469            .await
3470            .expect("gateway revision is accepted"));
3471
3472        let before = client.external_desired_peer_debug_snapshot().await;
3473        client
3474            .start_external_auto_connect("test-user".to_string(), "local-device".to_string())
3475            .await
3476            .expect("duplicate start is idempotent");
3477        let after = client.external_desired_peer_debug_snapshot().await;
3478
3479        assert_eq!(after["generation"], before["generation"]);
3480        assert_eq!(after["revision"], serde_json::json!(7));
3481        assert_eq!(after["active"], serde_json::json!(true));
3482    }
3483}
3484
3485#[cfg(not(target_arch = "wasm32"))]
3486fn should_suppress_auto_connect_for_replacement_window(
3487    snapshot: &ConnectionHealthView,
3488    now_ms: i64,
3489    transport_connected: bool,
3490) -> bool {
3491    if !snapshot.replacement_pending || transport_connected {
3492        return false;
3493    }
3494
3495    if now_ms.saturating_sub(snapshot.last_lifecycle_transition_at_ms) >= MANAGED_SETTLE_DEADLINE_MS
3496    {
3497        return false;
3498    }
3499
3500    // A transient network drop clears the active transport but does not itself
3501    // create a successor transport. Auto-connect must redial immediately;
3502    // otherwise native->browser edges remain stuck in transport-only readiness.
3503    !matches!(
3504        snapshot.last_reconnect_reason.as_deref(),
3505        Some("transient-disconnect")
3506    )
3507}
3508
3509impl Client {
3510    async fn retire_desired_peer_connections(
3511        &self,
3512        remote_device_id: &str,
3513        remote_node_id: Option<&str>,
3514        reason: &str,
3515    ) {
3516        let mut records = self.resolve_peer_connection_records(remote_device_id).await;
3517        if records.is_empty() {
3518            if let Some(node_id) = remote_node_id
3519                .map(str::trim)
3520                .filter(|value| !value.is_empty())
3521            {
3522                records = self.resolve_peer_connection_records(node_id).await;
3523            }
3524        }
3525
3526        for record in records {
3527            let endpoint_id = record
3528                .endpoint_id
3529                .as_deref()
3530                .or(record.node_id.as_deref())
3531                .and_then(|value| value.trim().parse::<iroh::EndpointId>().ok());
3532            let disconnected = if let Some(endpoint_id) = endpoint_id {
3533                self.disconnect_with_reason(endpoint_id, reason)
3534                    .await
3535                    .is_ok()
3536            } else {
3537                false
3538            };
3539            if !disconnected {
3540                self.retire_managed_connection(&record.connection_id, Some(reason.to_string()))
3541                    .await;
3542            }
3543            eprintln!(
3544                "[pluto-rtc][auto-connect] retired undesired connection remote_device_id={} connection_id={} reason={} transport_closed={}",
3545                remote_device_id, record.connection_id, reason, disconnected
3546            );
3547        }
3548    }
3549
3550    async fn retire_withdrawn_desired_peer_connections(
3551        &self,
3552        remote_device_id: &str,
3553        remote_node_id: Option<&str>,
3554    ) {
3555        self.retire_desired_peer_connections(
3556            remote_device_id,
3557            remote_node_id,
3558            crate::lifecycle_reason::REASON_DESIRED_PEER_WITHDRAWN,
3559        )
3560        .await;
3561    }
3562
3563    #[cfg(not(target_arch = "wasm32"))]
3564    pub(crate) async fn retire_offline_desired_route(
3565        &self,
3566        remote_device_id: &str,
3567        remote_node_id: Option<&str>,
3568    ) {
3569        self.retire_withdrawn_desired_peer_connections(remote_device_id, remote_node_id)
3570            .await;
3571    }
3572
3573    pub async fn is_connection_transport_alive(&self, endpoint_id: iroh::EndpointId) -> bool {
3574        if !self.is_connected(endpoint_id).await {
3575            return false;
3576        }
3577
3578        #[cfg(not(target_arch = "wasm32"))]
3579        {
3580            let Some(connection) = self.get_connection(endpoint_id).await else {
3581                return false;
3582            };
3583
3584            return connection.close_reason().is_none();
3585        }
3586
3587        #[cfg(target_arch = "wasm32")]
3588        {
3589            let Some(connection) = self.get_connection(endpoint_id).await else {
3590                return false;
3591            };
3592
3593            connection.close_reason().is_none()
3594        }
3595    }
3596
3597    #[cfg(not(target_arch = "wasm32"))]
3598    pub(crate) async fn validate_active_connection(
3599        &self,
3600        endpoint_id: iroh::EndpointId,
3601    ) -> Option<crate::heartbeat::IrohConnectionProbe> {
3602        if !self.is_connection_transport_alive(endpoint_id).await {
3603            return None;
3604        }
3605        let node = self.iroh_node.read().await.as_ref().cloned()?;
3606        node.probe_connection(endpoint_id, std::time::Duration::from_secs(4))
3607            .await
3608    }
3609
3610    #[cfg(not(target_arch = "wasm32"))]
3611    pub(crate) async fn should_force_network_change_after_connect_failures(
3612        &self,
3613        remote_device_id: &str,
3614    ) -> bool {
3615        for record in self.connection_manager.list_active().await {
3616            if !matches!(
3617                record.state,
3618                crate::connection_manager::ConnectionState::Connected
3619            ) {
3620                continue;
3621            }
3622
3623            let Some(record_device_id) = record
3624                .device_id
3625                .as_deref()
3626                .or(record.device_id_hint.as_deref())
3627                .map(str::trim)
3628                .filter(|value| !value.is_empty())
3629            else {
3630                continue;
3631            };
3632
3633            if record_device_id == remote_device_id {
3634                continue;
3635            }
3636
3637            eprintln!(
3638                "[pluto-rtc][auto-connect] suppress network-change recovery remote_device_id={} because healthy peer remains connected peer_device_id={} connection_id={}",
3639                remote_device_id,
3640                record_device_id,
3641                record.connection_id
3642            );
3643            return false;
3644        }
3645
3646        true
3647    }
3648
3649    #[cfg(not(target_arch = "wasm32"))]
3650    pub(crate) async fn try_auto_connect_device(
3651        self: &Arc<Self>,
3652        user_id: &str,
3653        device: crate::signaling::Device,
3654        local_device_id: &str,
3655        last_attempt_at: &mut std::collections::HashMap<String, i64>,
3656        failure_count: &mut std::collections::HashMap<String, u8>,
3657        failure_backoff_until: &mut std::collections::HashMap<String, i64>,
3658        retry_throttle_ms: i64,
3659        non_initiator_wait_started_at: &mut std::collections::HashMap<String, i64>,
3660        non_initiator_escalation_count: &mut std::collections::HashMap<String, u8>,
3661        initiator_grace_ms: i64,
3662        skip_log_at: &mut std::collections::HashMap<String, i64>,
3663        connected_health_checked_at: &mut std::collections::HashMap<String, i64>,
3664        connected_health_state: &mut std::collections::HashMap<String, bool>,
3665        connected_health_failures: &mut std::collections::HashMap<String, u8>,
3666        connected_health_probe_ms: i64,
3667        connected_health_failure_threshold: u8,
3668        last_presence_republish_at_ms: &mut i64,
3669        presence_republish_interval_ms: i64,
3670        last_network_change_at_ms: &mut i64,
3671        network_change_recovery_interval_ms: i64,
3672        network_change_failure_threshold: u8,
3673        last_known_node_id: &mut std::collections::HashMap<String, String>,
3674    ) {
3675        if self.is_app_execution_suspended() {
3676            return;
3677        }
3678
3679        #[cfg(target_arch = "wasm32")]
3680        let _ = (
3681            user_id,
3682            last_presence_republish_at_ms,
3683            presence_republish_interval_ms,
3684            last_network_change_at_ms,
3685            network_change_recovery_interval_ms,
3686            network_change_failure_threshold,
3687        );
3688
3689        let remote_device_id = device.device_id.clone();
3690        let now = now_millis_i64();
3691        let skip_log_interval_ms = 2_000;
3692
3693        let mut log_skip = |reason: &str| {
3694            let is_noise_reason = matches!(
3695                reason,
3696                "already-connected"
3697                    | "tie-break-not-initiator-waiting"
3698                    | "retry-throttle"
3699                    | "self-ticket"
3700                    | "auto-connect-excluded"
3701                    | "remote-excluded-local-device"
3702            );
3703            if is_noise_reason && !auto_connect_verbose() {
3704                return;
3705            }
3706
3707            let skip_key = format!("{}::{}", remote_device_id, reason);
3708            let should_log = match skip_log_at.get(&skip_key) {
3709                Some(last_at) => now.saturating_sub(*last_at) >= skip_log_interval_ms,
3710                None => true,
3711            };
3712            if !should_log {
3713                return;
3714            }
3715            skip_log_at.insert(skip_key, now);
3716
3717            #[cfg(not(target_arch = "wasm32"))]
3718            eprintln!(
3719                "[pluto-rtc][auto-connect] skip remote_device_id={} local_device_id={} reason={}",
3720                remote_device_id, local_device_id, reason
3721            );
3722            #[cfg(target_arch = "wasm32")]
3723            web_sys::console::log_1(&wasm_bindgen::JsValue::from_str(&format!(
3724                "[pluto-rtc][auto-connect] skip remote_device_id={} local_device_id={} reason={}",
3725                remote_device_id, local_device_id, reason
3726            )));
3727        };
3728
3729        if !device.online {
3730            last_attempt_at.remove(&remote_device_id);
3731            non_initiator_wait_started_at.remove(&remote_device_id);
3732            non_initiator_escalation_count.remove(&remote_device_id);
3733            connected_health_checked_at.remove(&remote_device_id);
3734            connected_health_state.remove(&remote_device_id);
3735            connected_health_failures.remove(&remote_device_id);
3736            clear_auto_connect_failure_state(
3737                &remote_device_id,
3738                failure_count,
3739                failure_backoff_until,
3740            );
3741            return;
3742        }
3743
3744        if remote_device_id == local_device_id {
3745            last_attempt_at.remove(&remote_device_id);
3746            non_initiator_wait_started_at.remove(&remote_device_id);
3747            non_initiator_escalation_count.remove(&remote_device_id);
3748            connected_health_checked_at.remove(&remote_device_id);
3749            connected_health_state.remove(&remote_device_id);
3750            connected_health_failures.remove(&remote_device_id);
3751            clear_auto_connect_failure_state(
3752                &remote_device_id,
3753                failure_count,
3754                failure_backoff_until,
3755            );
3756            return;
3757        }
3758
3759        if remote_excludes_local_device(&device.excluded_peers, local_device_id) {
3760            log_skip("remote-excluded-local-device");
3761            return;
3762        }
3763
3764        if self.is_auto_connect_peer_excluded(&remote_device_id, device.node_id.as_deref()) {
3765            log_skip("auto-connect-excluded");
3766            return;
3767        }
3768
3769        let provider_ticket = match device.ticket.as_deref() {
3770            Some(ticket) if !ticket.trim().is_empty() => ticket,
3771            _ => {
3772                log_skip("missing-ticket");
3773                return;
3774            }
3775        };
3776
3777        let (provider_iroh_ticket, _) = crate::session_token::split_ticket(provider_ticket.trim());
3778        let provider_endpoint_addr = match parse_endpoint_ticket(provider_iroh_ticket) {
3779            Ok(addr) => addr,
3780            Err(error) => {
3781                log_skip("invalid-ticket");
3782                eprintln!(
3783                    "[pluto-rtc][auto-connect] invalid ticket remote_device_id={} error={}",
3784                    remote_device_id, error
3785                );
3786                return;
3787            }
3788        };
3789        let preferred_ticket = self.preferred_scoped_route_ticket(
3790            provider_endpoint_addr.id.to_string().as_str(),
3791            provider_iroh_ticket,
3792        );
3793        let ticket_str = preferred_ticket.as_deref().unwrap_or(provider_ticket);
3794
3795        // Strip compound ticket suffix for iroh parsing.  The token is
3796        // extracted and will be auto-presented to the host after connection.
3797        let (iroh_ticket, token_suffix) = crate::session_token::split_ticket(ticket_str.trim());
3798        let extracted_token_payload = token_suffix
3799            .and_then(|suffix| crate::session_token::decode_payload(iroh_ticket, suffix));
3800        let extracted_token = extracted_token_payload
3801            .as_ref()
3802            .map(|payload| payload.token.clone());
3803        let extracted_token_suffix = token_suffix
3804            .filter(|_| extracted_token_payload.is_some())
3805            .map(ToOwned::to_owned);
3806
3807        let endpoint_addr = provider_endpoint_addr;
3808        #[cfg(feature = "transport-lan")]
3809        if self.transport_config.read().await.network_policy
3810            == crate::client::NetworkPolicy::LocalOnly
3811            && !crate::local_discovery::endpoint_addr_is_local_only(&endpoint_addr, &[])
3812        {
3813            log_skip("local-only-address-rejected");
3814            return;
3815        }
3816        let endpoint_id = endpoint_addr.id;
3817        let node_id_str = endpoint_id.to_string();
3818        #[cfg(not(target_arch = "wasm32"))]
3819        if let Some(payload) = extracted_token_payload
3820            .as_ref()
3821            .filter(|payload| payload.scope.as_str() == "user-device")
3822        {
3823            self.update_native_route_repair_credential(
3824                &node_id_str,
3825                Some(&remote_device_id),
3826                extracted_token.as_deref(),
3827                extracted_token_suffix.as_deref(),
3828                &payload.scope,
3829            );
3830        }
3831        let runtime_instance_changed = observe_remote_runtime_instance(
3832            last_known_node_id,
3833            &remote_device_id,
3834            device.session_id.as_deref(),
3835        );
3836        let ticket_changed = reconcile_remote_ticket_retry_state(
3837            last_known_node_id,
3838            &remote_device_id,
3839            provider_ticket,
3840            last_attempt_at,
3841            failure_count,
3842            failure_backoff_until,
3843        );
3844
3845        if ticket_changed {
3846            eprintln!(
3847                "[pluto-rtc][auto-connect] peer ticket changed remote_device_id={} node_id={} - clearing stale retry deadline",
3848                remote_device_id, node_id_str
3849            );
3850        }
3851
3852        // Detect peer identity change (e.g. web refresh → new node_id).
3853        // When detected, clear all retry/health/backoff state so reconnection
3854        // proceeds immediately without being blocked by stale throttles.
3855        {
3856            let identity_changed = last_known_node_id
3857                .get(&remote_device_id)
3858                .map(|prev| prev.as_str() != node_id_str)
3859                .unwrap_or(false);
3860            last_known_node_id.insert(remote_device_id.clone(), node_id_str.clone());
3861            if identity_changed {
3862                #[cfg(not(target_arch = "wasm32"))]
3863                eprintln!(
3864                    "[pluto-rtc][auto-connect] peer identity changed remote_device_id={} new_node_id={} — resetting state for fast reconnect",
3865                    remote_device_id, node_id_str
3866                );
3867                #[cfg(target_arch = "wasm32")]
3868                web_sys::console::log_1(&wasm_bindgen::JsValue::from_str(&format!(
3869                    "[pluto-rtc][auto-connect] peer identity changed remote_device_id={} new_node_id={} — resetting state for fast reconnect",
3870                    remote_device_id, node_id_str
3871                )));
3872                last_attempt_at.remove(&remote_device_id);
3873                non_initiator_wait_started_at.remove(&remote_device_id);
3874                non_initiator_escalation_count.remove(&remote_device_id);
3875                connected_health_checked_at.remove(&remote_device_id);
3876                connected_health_state.remove(&remote_device_id);
3877                connected_health_failures.remove(&remote_device_id);
3878                clear_auto_connect_failure_state(
3879                    &remote_device_id,
3880                    failure_count,
3881                    failure_backoff_until,
3882                );
3883            }
3884        }
3885
3886        let local_node_id = match self.current_node_id().await {
3887            Some(id) => id,
3888            None => {
3889                log_skip("local-node-not-ready");
3890                return;
3891            }
3892        };
3893
3894        if local_node_id == node_id_str {
3895            log_skip("self-ticket");
3896            return;
3897        }
3898
3899        if runtime_instance_changed {
3900            // Presence is coordination evidence, not physical-lifecycle
3901            // authority. A newly launched peer can accept a replacement route
3902            // before its new gateway lease wins aggregation. Retiring that
3903            // already-responsive route when the delayed instance arrives
3904            // creates needless generation churn. Probe the exact current
3905            // generation and retire it only when the peer no longer responds.
3906            let current_transport = self.get_connection(endpoint_id).await;
3907            let current_transport_stable_id = current_transport
3908                .as_ref()
3909                .map(crate::transport_generation::for_connection);
3910            #[cfg(openrtc_iroh_retire_remote_connections_api)]
3911            if let Some(current_transport) = current_transport.as_ref() {
3912                let transport_stable_id =
3913                    crate::transport_generation::for_connection(current_transport);
3914                // One Iroh endpoint can carry OpenRTC control traffic and
3915                // independent application protocols. A durable node identity
3916                // survives process restart, so those other protocol
3917                // connections can otherwise remain attached to the retired
3918                // process generation even after the new OpenRTC connection is
3919                // healthy. Runtime-instance replacement is the typed lifecycle
3920                // evidence: fence every other connection for this peer while
3921                // preserving the exact admitted OpenRTC generation.
3922                let endpoint = { self.iroh_endpoint.read().await.clone() };
3923                if let Some(endpoint) = endpoint {
3924                    let retired = retire_stale_endpoint_connections(
3925                        &endpoint,
3926                        endpoint_id,
3927                        current_transport,
3928                        crate::lifecycle_reason::REASON_STALE_ACTIVE_CONNECTION_RECONNECT
3929                            .as_bytes(),
3930                    )
3931                    .await;
3932                    if retired > 0 {
3933                        eprintln!(
3934                            "[pluto-rtc][auto-connect] peer runtime instance changed remote_device_id={} node_id={} - retired {} stale shared-endpoint protocol connection(s), preserved transport_stable_id={}",
3935                            remote_device_id, node_id_str, retired, transport_stable_id
3936                        );
3937                    }
3938                }
3939            }
3940            let active_transport_responsive = if current_transport_stable_id.is_some() {
3941                self.validate_active_connection(endpoint_id)
3942                    .await
3943                    .is_some_and(|probe| probe.responsive)
3944            } else {
3945                false
3946            };
3947            if runtime_instance_change_requires_replacement(
3948                runtime_instance_changed,
3949                active_transport_responsive,
3950            ) {
3951                if let Some(transport_stable_id) = current_transport_stable_id {
3952                    match self
3953                        .retire_transport_generation_after_failed_probe(
3954                            &Self::deterministic_connection_id(&local_node_id, &node_id_str),
3955                            endpoint_id,
3956                            transport_stable_id,
3957                            connected_health_probe_ms,
3958                            crate::lifecycle_reason::REASON_STALE_ACTIVE_CONNECTION_RECONNECT,
3959                        )
3960                        .await
3961                    {
3962                        FailedProbeRetirementOutcome::Retired => {
3963                            eprintln!(
3964                                "[pluto-rtc][auto-connect] peer runtime instance changed remote_device_id={} node_id={} - replaced unresponsive physical transport",
3965                                remote_device_id, node_id_str
3966                            );
3967                        }
3968                        FailedProbeRetirementOutcome::DeferredAdmissionInFlight => {
3969                            eprintln!(
3970                                "[pluto-rtc][auto-connect] peer runtime instance changed remote_device_id={} node_id={} - deferred replacement while admission is in flight",
3971                                remote_device_id, node_id_str
3972                            );
3973                        }
3974                        FailedProbeRetirementOutcome::CancelledCurrentGenerationActivity => {
3975                            eprintln!(
3976                                "[pluto-rtc][auto-connect] peer runtime instance changed remote_device_id={} node_id={} - preserved generation with recent protocol activity",
3977                                remote_device_id, node_id_str
3978                            );
3979                        }
3980                        FailedProbeRetirementOutcome::GenerationChanged => {
3981                            eprintln!(
3982                                "[pluto-rtc][auto-connect] peer runtime instance changed remote_device_id={} node_id={} - ignored stale generation",
3983                                remote_device_id, node_id_str
3984                            );
3985                        }
3986                    }
3987                }
3988            } else {
3989                eprintln!(
3990                    "[pluto-rtc][auto-connect] peer runtime instance changed remote_device_id={} node_id={} - preserving responsive physical transport",
3991                    remote_device_id, node_id_str
3992                );
3993            }
3994            last_attempt_at.remove(&remote_device_id);
3995            non_initiator_wait_started_at.remove(&remote_device_id);
3996            non_initiator_escalation_count.remove(&remote_device_id);
3997            connected_health_checked_at.remove(&remote_device_id);
3998            connected_health_state.remove(&remote_device_id);
3999            connected_health_failures.remove(&remote_device_id);
4000            clear_auto_connect_failure_state(
4001                &remote_device_id,
4002                failure_count,
4003                failure_backoff_until,
4004            );
4005        }
4006
4007        if let Some(discovered_node_id) = device.node_id.as_deref() {
4008            if discovered_node_id != node_id_str {
4009                log_skip("ticket-node-mismatch-discovery");
4010                eprintln!(
4011                    "[pluto-rtc][auto-connect] rejected inconsistent discovery identity remote_device_id={} discovered_node_id={} ticket_node_id={}",
4012                    remote_device_id, discovered_node_id, node_id_str
4013                );
4014                return;
4015            }
4016        }
4017
4018        // The discovery ticket/node pair is the current coordination authority
4019        // for this durable device. A transport to an older node can remain open
4020        // until QUIC idle timeout after a browser/app restart; it must not keep
4021        // owning the logical device session or receive application payloads.
4022        // Run this even when the loop did not observe the previous node itself
4023        // (for example, subscription replacement between snapshots).
4024        self.reconcile_authoritative_device_node(&remote_device_id, &node_id_str)
4025            .await;
4026
4027        #[cfg(feature = "transport-lan")]
4028        if self.is_peer_locally_reachable(&node_id_str).await {
4029            eprintln!(
4030                "[pluto-rtc][auto-connect] LAN peer reachable remote_device_id={} node_id={}",
4031                remote_device_id, node_id_str
4032            );
4033        }
4034
4035        let connection_id = Self::deterministic_connection_id(&local_node_id, &node_id_str);
4036
4037        let managed_health_snapshot = self.managed_connection_health(&connection_id).await;
4038
4039        #[cfg(not(target_arch = "wasm32"))]
4040        {
4041            if let Some(existing) = self
4042                .connection_manager
4043                .get_by_connection_id(&connection_id)
4044                .await
4045            {
4046                let existing_device_id = existing.device_id.as_deref().map(str::trim);
4047                if existing_device_id != Some(remote_device_id.as_str()) {
4048                    self.connection_manager
4049                        .set_device_id(&connection_id, remote_device_id.clone())
4050                        .await;
4051                    eprintln!(
4052                        "[pluto-rtc][auto-connect] backfilled device_id hint connection_id={} node_id={} device_id={}",
4053                        connection_id, node_id_str, remote_device_id
4054                    );
4055                }
4056            }
4057        }
4058
4059        // The native managed actor owns the durable logical Iroh connection.
4060        // A carrier mechanism cannot repair a missing committed connection on
4061        // its own, so only generation-current Iroh liveness may suppress the
4062        // bounded redial after a network drop.
4063        #[cfg(not(target_arch = "wasm32"))]
4064        let transport_connected = self.is_connection_transport_alive(endpoint_id).await;
4065        #[cfg(target_arch = "wasm32")]
4066        let transport_connected = self.is_connected(endpoint_id).await;
4067        if let Some(snapshot) = managed_health_snapshot.as_ref() {
4068            // The Rust core owns replacement recovery. Suppress a competing
4069            // dial while no transport owns the replacement window, but never
4070            // let that suppression bypass positive liveness probes for an
4071            // already-owned generation.
4072            if should_suppress_auto_connect_for_replacement_window(
4073                &snapshot,
4074                now,
4075                transport_connected,
4076            ) {
4077                log_skip("replacement-pending");
4078                return;
4079            }
4080        }
4081        #[cfg(target_arch = "wasm32")]
4082        if !transport_connected {
4083            if let Some(existing) = self
4084                .connection_manager
4085                .get_by_connection_id(&connection_id)
4086                .await
4087            {
4088                if matches!(
4089                    existing.state,
4090                    crate::connection_manager::ConnectionState::Connected
4091                ) {
4092                    web_sys::console::warn_1(&wasm_bindgen::JsValue::from_str(&format!(
4093                        "[pluto-rtc][auto-connect][wasm] transport mismatch remote_device_id={} remote_node_id={} connection_id={} record_stable_id={:?} record_generation={} state={:?} status_reason={:?}",
4094                        remote_device_id,
4095                        node_id_str,
4096                        connection_id,
4097                        existing.transport_stable_id,
4098                        existing.transport_generation,
4099                        existing.state,
4100                        existing.status_reason
4101                    )));
4102                }
4103            }
4104        }
4105
4106        // Tie breaker only controls who should dial.
4107        // If an incoming transport is already connected, the non-initiator must
4108        // observe and promote that transport instead of waiting forever for the
4109        // initiator path to "win".
4110        //
4111        // Use exponential backoff with jitter so repeated escalations don't
4112        // produce duplicate-storm collisions (2s→4s→8s→16s cap, ±20% jitter).
4113        let esc_count = non_initiator_escalation_count
4114            .get(&remote_device_id)
4115            .copied()
4116            .unwrap_or(0);
4117        let effective_grace_ms = if local_node_id <= node_id_str {
4118            non_initiator_escalation_grace_ms(esc_count)
4119        } else {
4120            initiator_grace_ms
4121        };
4122        let waited_ms = if local_node_id <= node_id_str {
4123            let wait_started_at = non_initiator_wait_started_at
4124                .entry(remote_device_id.clone())
4125                .or_insert(now);
4126            now.saturating_sub(*wait_started_at)
4127        } else {
4128            0
4129        };
4130        match auto_connect_tie_break_decision(
4131            &local_node_id,
4132            &node_id_str,
4133            transport_connected,
4134            waited_ms,
4135            effective_grace_ms,
4136        ) {
4137            AutoConnectTieBreakDecision::ObserveConnectedTransport => {
4138                non_initiator_wait_started_at.remove(&remote_device_id);
4139                non_initiator_escalation_count.remove(&remote_device_id);
4140            }
4141            AutoConnectTieBreakDecision::WaitForInitiator => {
4142                if waited_ms < 1_000 {
4143                    log_skip("tie-break-not-initiator-waiting");
4144                }
4145                return;
4146            }
4147            AutoConnectTieBreakDecision::ActAsInitiator => {
4148                if local_node_id <= node_id_str {
4149                    // Non-initiator escalated after grace period expired.
4150                    let new_esc_count = esc_count.saturating_add(1);
4151                    non_initiator_escalation_count.insert(remote_device_id.clone(), new_esc_count);
4152                    #[cfg(not(target_arch = "wasm32"))]
4153                    eprintln!(
4154                        "[pluto-rtc][auto-connect] non-initiator escalated after {}ms grace remote_device_id={} local_device_id={} escalation_count={}",
4155                        waited_ms, remote_device_id, local_device_id, new_esc_count
4156                    );
4157                    #[cfg(target_arch = "wasm32")]
4158                    web_sys::console::log_1(&wasm_bindgen::JsValue::from_str(&format!(
4159                        "[pluto-rtc][auto-connect] non-initiator escalated after {}ms grace remote_device_id={} local_device_id={} escalation_count={}",
4160                        waited_ms, remote_device_id, local_device_id, new_esc_count
4161                    )));
4162                }
4163                non_initiator_wait_started_at.remove(&remote_device_id);
4164            }
4165        }
4166
4167        self.reconcile_authoritative_device_node(&remote_device_id, &node_id_str)
4168            .await;
4169
4170        if transport_connected {
4171            let current_scopes = self.connection_manager.get_scopes(&connection_id).await;
4172            let admitted_scope = match self.session_admission(&connection_id) {
4173                crate::session_token::SessionAdmission::Accepted {
4174                    scope: Some(scope), ..
4175                } => Some(scope.into_inner()),
4176                _ => None,
4177            };
4178            let presentation_in_flight = self
4179                .session_token_registry
4180                .is_session_token_presentation_in_flight(&connection_id);
4181            if !persistent_managed_actor_may_own_transport(
4182                &current_scopes,
4183                admitted_scope.as_deref(),
4184                presentation_in_flight,
4185            ) {
4186                log_skip(if presentation_in_flight {
4187                    "explicit-admission-presentation-owns-current-transport"
4188                } else {
4189                    "non-managed-scope-owns-current-transport"
4190                });
4191                return;
4192            }
4193
4194            connected_health_checked_at
4195                .entry(remote_device_id.clone())
4196                .or_insert(now);
4197            connected_health_state
4198                .entry(remote_device_id.clone())
4199                .or_insert(true);
4200
4201            let should_probe = match connected_health_checked_at.get(&remote_device_id) {
4202                Some(last_probe_at) => {
4203                    now.saturating_sub(*last_probe_at) >= connected_health_probe_ms
4204                }
4205                None => false,
4206            };
4207            let mut healthy = connected_health_state
4208                .get(&remote_device_id)
4209                .copied()
4210                .unwrap_or(true);
4211            let mut probed_transport_stable_id = None;
4212
4213            if should_probe {
4214                let active_probe = self.validate_active_connection(endpoint_id).await;
4215                probed_transport_stable_id =
4216                    active_probe.as_ref().map(|probe| probe.transport_stable_id);
4217                let active_probe_healthy =
4218                    active_probe.as_ref().is_some_and(|probe| probe.responsive);
4219                let passive_transport_alive = self.is_connection_transport_alive(endpoint_id).await;
4220                let recent_generation_activity = active_probe.as_ref().is_some_and(|probe| {
4221                    generation_activity_is_recent(
4222                        probe.last_inbound_activity_age,
4223                        connected_health_probe_ms,
4224                    ) || generation_activity_is_recent(
4225                        self.native_transport_protocol_activity_age(
4226                            &connection_id,
4227                            probe.transport_stable_id,
4228                        ),
4229                        connected_health_probe_ms,
4230                    )
4231                });
4232                let probe_decision = active_connection_probe_decision_with_generation_activity(
4233                    active_probe_healthy,
4234                    passive_transport_alive,
4235                    recent_generation_activity,
4236                );
4237                let current_failures = connected_health_failures
4238                    .get(&remote_device_id)
4239                    .copied()
4240                    .unwrap_or(0);
4241                let failures =
4242                    next_active_connection_probe_failure_count(probe_decision, current_failures);
4243                healthy = active_connection_probe_is_healthy(
4244                    probe_decision,
4245                    failures,
4246                    connected_health_failure_threshold,
4247                );
4248                connected_health_checked_at.insert(remote_device_id.clone(), now);
4249                connected_health_state.insert(remote_device_id.clone(), healthy);
4250                if matches!(probe_decision, ActiveConnectionProbeDecision::Healthy) {
4251                    connected_health_failures.remove(&remote_device_id);
4252                    // Keep the canonical peer snapshot aligned with the transport
4253                    // liveness decision. This is especially important after an
4254                    // in-place app relaunch: the old generation may have marked the
4255                    // peer stale while the replacement transport is already live.
4256                    // A matching pong is current-generation readiness evidence.
4257                    let _ = self
4258                        .confirm_managed_connection_readiness(&connection_id)
4259                        .await;
4260                } else {
4261                    connected_health_failures.insert(remote_device_id.clone(), failures);
4262                    if matches!(
4263                        probe_decision,
4264                        ActiveConnectionProbeDecision::PreserveLiveTransport
4265                    ) && healthy
4266                    {
4267                        log_skip("active-probe-missed-preserving-live-transport");
4268                    }
4269                }
4270            }
4271
4272            if !healthy {
4273                if !should_probe {
4274                    log_skip("awaiting-current-generation-health-probe");
4275                    return;
4276                }
4277                if self.is_app_execution_suspended() {
4278                    log_skip("backgrounded-after-health-probe");
4279                    return;
4280                }
4281                #[cfg(not(target_arch = "wasm32"))]
4282                {
4283                    let failures = connected_health_failures
4284                        .get(&remote_device_id)
4285                        .copied()
4286                        .unwrap_or(1);
4287                    if failures < connected_health_failure_threshold {
4288                        log_skip("active-connection-health-probe-failed");
4289                        return;
4290                    }
4291                    let Some(transport_stable_id) = probed_transport_stable_id else {
4292                        log_skip("active-probe-generation-disappeared");
4293                        connected_health_checked_at.remove(&remote_device_id);
4294                        connected_health_state.remove(&remote_device_id);
4295                        connected_health_failures.remove(&remote_device_id);
4296                        return;
4297                    };
4298                    match self
4299                        .retire_transport_generation_after_failed_probe(
4300                            &connection_id,
4301                            endpoint_id,
4302                            transport_stable_id,
4303                            connected_health_probe_ms,
4304                            crate::lifecycle_reason::REASON_STALE_ACTIVE_CONNECTION_RECONNECT,
4305                        )
4306                        .await
4307                    {
4308                        FailedProbeRetirementOutcome::Retired => {
4309                            log_skip("stale-active-connection-reconnect");
4310                        }
4311                        FailedProbeRetirementOutcome::DeferredAdmissionInFlight => {
4312                            log_skip("active-probe-retirement-deferred-admission-in-flight");
4313                            connected_health_checked_at.insert(remote_device_id.clone(), now);
4314                            connected_health_state.insert(remote_device_id.clone(), true);
4315                            connected_health_failures.remove(&remote_device_id);
4316                            return;
4317                        }
4318                        FailedProbeRetirementOutcome::CancelledCurrentGenerationActivity => {
4319                            log_skip(
4320                                "active-probe-retirement-cancelled-by-current-generation-activity",
4321                            );
4322                            connected_health_checked_at.insert(remote_device_id.clone(), now);
4323                            connected_health_state.insert(remote_device_id.clone(), true);
4324                            connected_health_failures.remove(&remote_device_id);
4325                            let _ = self
4326                                .confirm_managed_connection_readiness(&connection_id)
4327                                .await;
4328                            return;
4329                        }
4330                        FailedProbeRetirementOutcome::GenerationChanged => {
4331                            log_skip("active-probe-generation-disappeared");
4332                            connected_health_checked_at.remove(&remote_device_id);
4333                            connected_health_state.remove(&remote_device_id);
4334                            connected_health_failures.remove(&remote_device_id);
4335                            return;
4336                        }
4337                    }
4338                    self.maybe_refresh_gateway_presence_for_auto_connect(
4339                        user_id,
4340                        local_device_id,
4341                        "stale-active-connection-reconnect",
4342                        last_presence_republish_at_ms,
4343                        presence_republish_interval_ms,
4344                    )
4345                    .await;
4346                    self.reconcile_authoritative_device_node(&remote_device_id, &node_id_str)
4347                        .await;
4348                    connected_health_checked_at.remove(&remote_device_id);
4349                    connected_health_state.remove(&remote_device_id);
4350                    connected_health_failures.remove(&remote_device_id);
4351                    if failures >= network_change_failure_threshold {
4352                        self.maybe_force_network_change_for_auto_connect(
4353                            user_id,
4354                            local_device_id,
4355                            "stale-active-connection",
4356                            last_network_change_at_ms,
4357                            network_change_recovery_interval_ms,
4358                            last_presence_republish_at_ms,
4359                            presence_republish_interval_ms,
4360                        )
4361                        .await;
4362                    }
4363                }
4364                #[cfg(target_arch = "wasm32")]
4365                {
4366                    let failures = connected_health_failures
4367                        .get(&remote_device_id)
4368                        .copied()
4369                        .unwrap_or(1);
4370                    if failures < connected_health_failure_threshold {
4371                        eprintln!(
4372                            "[pluto-rtc][auto-connect][health] remote_device_id={} local_device_id={} remote_node_id={} healthy=false failures={} threshold={} action=wait",
4373                            remote_device_id,
4374                            local_device_id,
4375                            node_id_str,
4376                            failures,
4377                            connected_health_failure_threshold
4378                        );
4379                        log_skip("active-connection-health-probe-failed");
4380                        return;
4381                    }
4382                    eprintln!(
4383                        "[pluto-rtc][auto-connect][health] remote_device_id={} local_device_id={} remote_node_id={} healthy=false failures={} threshold={} action=disconnect-and-reconnect",
4384                        remote_device_id,
4385                        local_device_id,
4386                        node_id_str,
4387                        failures,
4388                        connected_health_failure_threshold
4389                    );
4390                    log_skip("stale-active-connection-reconnect");
4391                    let _ = self
4392                        .disconnect_with_reason(
4393                            endpoint_id,
4394                            crate::lifecycle_reason::REASON_STALE_ACTIVE_CONNECTION_RECONNECT,
4395                        )
4396                        .await;
4397                    connected_health_checked_at.remove(&remote_device_id);
4398                    connected_health_state.remove(&remote_device_id);
4399                    connected_health_failures.remove(&remote_device_id);
4400                }
4401            } else {
4402                // Do not clear a bounded passive-open miss here. A transport
4403                // handle can remain locally open after the remote process has
4404                // disappeared, so only the matching positive probe above may
4405                // reset `connected_health_failures`. Clearing on the
4406                // temporarily-healthy PreserveLiveTransport decision would
4407                // keep every pass at failure 1 and make the retirement
4408                // threshold unreachable.
4409                if self.is_auto_connect_peer_excluded(&remote_device_id, Some(&node_id_str)) {
4410                    log_skip("auto-connect-excluded-after-health-probe");
4411                    let _ = self
4412                        .disconnect_with_reason(
4413                            endpoint_id,
4414                            crate::lifecycle_reason::REASON_MANUAL_DISCONNECT,
4415                        )
4416                        .await;
4417                    return;
4418                }
4419                self.connection_manager
4420                    .upsert_pending(
4421                        connection_id.clone(),
4422                        Some(node_id_str.clone()),
4423                        Some(remote_device_id.clone()),
4424                        Some(node_id_str.clone()),
4425                    )
4426                    .await;
4427                let current_transport_stable_id = self
4428                    .get_connection(endpoint_id)
4429                    .await
4430                    .map(|connection| crate::transport_generation::for_connection(&connection));
4431                self.commit_current_native_transport_record(
4432                    &connection_id,
4433                    &node_id_str,
4434                    Some(remote_device_id.clone()),
4435                    current_transport_stable_id,
4436                    Some("health-check".to_string()),
4437                )
4438                .await;
4439                #[cfg(target_arch = "wasm32")]
4440                {
4441                    let active_transport =
4442                        self.iroh_path_kind(&node_id_str).await.transport_label();
4443                    let _ = self
4444                        .report_transport_status_for_current_generation(
4445                            &connection_id,
4446                            active_transport,
4447                            None,
4448                        )
4449                        .await;
4450                }
4451
4452                // Admission is bound to the physical transport generation, not
4453                // to whichever fresh ticket happens to be in the latest presence
4454                // row. Native peers routinely rotate advertised ticket tokens
4455                // while an already admitted connection remains live. Re-presenting
4456                // the replacement token creates a needless control stream and can
4457                // disturb an otherwise settled carrier generation.
4458                #[cfg(not(target_arch = "wasm32"))]
4459                let has_remote_admission_proof =
4460                    current_transport_stable_id.is_some_and(|transport_stable_id| {
4461                        self.remote_session_token_admitted_for_transport(
4462                            &connection_id,
4463                            transport_stable_id,
4464                        )
4465                    });
4466                #[cfg(target_arch = "wasm32")]
4467                let has_remote_admission_proof = if let Some(token) = extracted_token.as_deref() {
4468                    self.has_current_remote_session_admission_proof(
4469                        &connection_id,
4470                        endpoint_id,
4471                        token,
4472                    )
4473                    .await
4474                } else {
4475                    false
4476                };
4477                #[cfg(not(target_arch = "wasm32"))]
4478                let has_current_inbound_admission_proof =
4479                    current_transport_stable_id.is_some_and(|transport_stable_id| {
4480                        self.inbound_session_token_admitted_for_transport(
4481                            &connection_id,
4482                            transport_stable_id,
4483                        )
4484                    });
4485                #[cfg(not(target_arch = "wasm32"))]
4486                let peer_requested_reciprocal =
4487                    self.has_pending_reciprocal_session_admission_request(&connection_id);
4488                #[cfg(not(target_arch = "wasm32"))]
4489                if has_remote_admission_proof
4490                    && has_current_inbound_admission_proof
4491                    && peer_requested_reciprocal
4492                {
4493                    self.clear_reciprocal_session_admission_request(&connection_id);
4494                }
4495                #[cfg(not(target_arch = "wasm32"))]
4496                let presentation_decision = session_admission_presentation_decision(
4497                    extracted_token.is_some(),
4498                    has_remote_admission_proof,
4499                    has_current_inbound_admission_proof,
4500                    peer_requested_reciprocal,
4501                    local_node_id > node_id_str,
4502                );
4503                #[cfg(target_arch = "wasm32")]
4504                let presentation_decision = session_admission_presentation_decision(
4505                    extracted_token.is_some(),
4506                    has_remote_admission_proof,
4507                    true,
4508                    false,
4509                    true,
4510                );
4511                if presentation_decision.should_present {
4512                    if let Some(backoff_until) =
4513                        failure_backoff_until.get(&remote_device_id).copied()
4514                    {
4515                        if now < backoff_until {
4516                            log_skip("session-admission-backoff");
4517                            return;
4518                        }
4519                        failure_backoff_until.remove(&remote_device_id);
4520                    }
4521
4522                    let token = extracted_token
4523                        .as_ref()
4524                        .expect("representation requires an extracted token");
4525                    eprintln!(
4526                        "[pluto-rtc][auto-connect][session-admission] re-presenting token on live transport connection_id={} remote_device_id={} remote_node_id={} token_fp={}",
4527                        connection_id,
4528                        remote_device_id,
4529                        node_id_str,
4530                        super::core_impl::log_fingerprint(token.as_str())
4531                    );
4532                    #[cfg(not(target_arch = "wasm32"))]
4533                    let presentation_result = self
4534                        .present_and_accept_session_token_for_route_repair(
4535                            endpoint_id,
4536                            &connection_id,
4537                            token,
4538                            extracted_token_suffix.as_deref(),
4539                            Some(remote_device_id.clone()),
4540                            Some(local_device_id.to_string()),
4541                            has_remote_admission_proof,
4542                            presentation_decision.request_reciprocal,
4543                        )
4544                        .await;
4545                    #[cfg(target_arch = "wasm32")]
4546                    let presentation_result = self
4547                        .present_and_accept_session_token_with_local_claim(
4548                            endpoint_id,
4549                            &connection_id,
4550                            token,
4551                            extracted_token_suffix.as_deref(),
4552                            Some(remote_device_id.clone()),
4553                            Some(local_device_id.to_string()),
4554                        )
4555                        .await;
4556                    match presentation_result {
4557                        Ok(approval_scope) => {
4558                            clear_auto_connect_failure_state(
4559                                &remote_device_id,
4560                                failure_count,
4561                                failure_backoff_until,
4562                            );
4563                            eprintln!(
4564                                "[pluto-rtc][auto-connect][session-admission] live transport admitted connection_id={} remote_device_id={} remote_node_id={} scope={} token_fp={}",
4565                                connection_id,
4566                                remote_device_id,
4567                                node_id_str,
4568                                approval_scope,
4569                                super::core_impl::log_fingerprint(token.as_str())
4570                            );
4571                        }
4572                        Err(error) => {
4573                            let active_transport_stable_id =
4574                                self.get_connection(endpoint_id).await.map(|connection| {
4575                                    crate::transport_generation::for_connection(&connection)
4576                                });
4577                            if current_transport_stable_id != active_transport_stable_id {
4578                                eprintln!(
4579                                    "[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={:?}",
4580                                    connection_id,
4581                                    remote_device_id,
4582                                    current_transport_stable_id,
4583                                    active_transport_stable_id
4584                                );
4585                                return;
4586                            }
4587
4588                            #[cfg(not(target_arch = "wasm32"))]
4589                            let current_generation_is_routable = active_transport_stable_id
4590                                .is_some_and(|transport_stable_id| {
4591                                    self.native_application_stream_admitted_for_transport(
4592                                        &connection_id,
4593                                        transport_stable_id,
4594                                    )
4595                                });
4596                            #[cfg(target_arch = "wasm32")]
4597                            let current_generation_is_routable = false;
4598
4599                            if current_generation_is_routable {
4600                                clear_auto_connect_failure_state(
4601                                    &remote_device_id,
4602                                    failure_count,
4603                                    failure_backoff_until,
4604                                );
4605                                connected_health_failures.remove(&remote_device_id);
4606                                let _ = self
4607                                    .confirm_managed_connection_readiness(&connection_id)
4608                                    .await;
4609                                eprintln!(
4610                                    "[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={}",
4611                                    connection_id,
4612                                    remote_device_id,
4613                                    active_transport_stable_id,
4614                                    error,
4615                                );
4616                                return;
4617                            }
4618
4619                            // A duplicate never owns admission outcome. Ignore
4620                            // it even if the canonical exchange completed
4621                            // between returning this tagged error and handling
4622                            // it here; only the canonical presenter may count
4623                            // or retire its physical generation.
4624                            if session_token_presentation_was_duplicate(&error) {
4625                                eprintln!(
4626                                    "[pluto-rtc][auto-connect][session-admission] ignored duplicate presentation owned by canonical exchange connection_id={} remote_device_id={} transport_stable_id={:?}",
4627                                    connection_id,
4628                                    remote_device_id,
4629                                    active_transport_stable_id,
4630                                );
4631                                return;
4632                            }
4633
4634                            if self
4635                                .session_token_registry
4636                                .is_session_token_presentation_in_flight(&connection_id)
4637                            {
4638                                eprintln!(
4639                                    "[pluto-rtc][auto-connect][session-admission] ignored duplicate presentation while canonical exchange remains in flight connection_id={} remote_device_id={} transport_stable_id={:?}",
4640                                    connection_id,
4641                                    remote_device_id,
4642                                    active_transport_stable_id,
4643                                );
4644                                return;
4645                            }
4646
4647                            if session_token_presentation_failure_action(&error)
4648                                == SessionTokenPresentationFailureAction::RejectAndDisconnect
4649                            {
4650                                let rejection_reason =
4651                                    format!("session-token-presentation-failed: {}", error);
4652                                self.reject_session_connection(
4653                                    &connection_id,
4654                                    rejection_reason.as_str(),
4655                                );
4656                                self.connection_manager
4657                                    .set_failed(&connection_id, Some(rejection_reason.clone()))
4658                                    .await;
4659                                let _ = self
4660                                    .disconnect_with_reason(
4661                                        endpoint_id,
4662                                        crate::lifecycle_reason::REASON_SESSION_ADMISSION_REJECTED,
4663                                    )
4664                                    .await;
4665                                self.forget_session_connection(&connection_id);
4666                                register_auto_connect_admission_rejection(
4667                                    &remote_device_id,
4668                                    now_millis_i64(),
4669                                    failure_count,
4670                                    failure_backoff_until,
4671                                );
4672                                eprintln!(
4673                                    "[pluto-rtc][auto-connect][session-admission] live transport admission rejected connection_id={} remote_device_id={} remote_node_id={} token_fp={} error={}",
4674                                    connection_id,
4675                                    remote_device_id,
4676                                    node_id_str,
4677                                    super::core_impl::log_fingerprint(token.as_str()),
4678                                    error
4679                                );
4680                                return;
4681                            }
4682
4683                            register_auto_connect_failure(
4684                                &remote_device_id,
4685                                now_millis_i64(),
4686                                failure_count,
4687                                failure_backoff_until,
4688                            );
4689                            let failures =
4690                                failure_count.get(&remote_device_id).copied().unwrap_or(1);
4691                            match retryable_session_admission_failure_decision(
4692                                failures,
4693                                connected_health_failure_threshold,
4694                                current_transport_stable_id,
4695                                active_transport_stable_id,
4696                                runtime_instance_changed,
4697                            ) {
4698                                RetryableSessionAdmissionFailureDecision::PreserveCurrentTransport => {
4699                                    eprintln!(
4700                                        "[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={}",
4701                                        connection_id,
4702                                        remote_device_id,
4703                                        node_id_str,
4704                                        current_transport_stable_id,
4705                                        failures,
4706                                        connected_health_failure_threshold,
4707                                        super::core_impl::log_fingerprint(token.as_str()),
4708                                        error
4709                                    );
4710                                }
4711                                RetryableSessionAdmissionFailureDecision::RetireCurrentTransport => {
4712                                    eprintln!(
4713                                        "[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={}",
4714                                        connection_id,
4715                                        remote_device_id,
4716                                        node_id_str,
4717                                        current_transport_stable_id,
4718                                        failures,
4719                                        connected_health_failure_threshold,
4720                                        super::core_impl::log_fingerprint(token.as_str()),
4721                                        error
4722                                    );
4723                                    let _ = self
4724                                        .disconnect_with_reason(
4725                                            endpoint_id,
4726                                            crate::lifecycle_reason::REASON_STALE_ACTIVE_CONNECTION_RECONNECT,
4727                                        )
4728                                        .await;
4729                                    connected_health_checked_at.remove(&remote_device_id);
4730                                    connected_health_state.remove(&remote_device_id);
4731                                    connected_health_failures.remove(&remote_device_id);
4732                                }
4733                                RetryableSessionAdmissionFailureDecision::ReplacementAlreadyWon => {
4734                                    eprintln!(
4735                                        "[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={:?}",
4736                                        connection_id,
4737                                        remote_device_id,
4738                                        current_transport_stable_id,
4739                                        active_transport_stable_id
4740                                    );
4741                                }
4742                            }
4743                            return;
4744                        }
4745                    }
4746                }
4747
4748                #[cfg(not(target_arch = "wasm32"))]
4749                {
4750                    if self
4751                        .promote_known_native_user_device_connection(&connection_id, &node_id_str)
4752                        .await
4753                        .is_none()
4754                    {
4755                        let _ = self
4756                            .confirm_managed_connection_readiness(&connection_id)
4757                            .await;
4758                    }
4759                    // Health polling observes the current committed connection;
4760                    // it does not own carrier recovery. Initial connect, path
4761                    // changes, foreground resume, inbound signaling, and the
4762                    // Rust retry scheduler already feed the connection owner.
4763                    // Starting WebRTC here can transiently clear settlement on
4764                    // an otherwise healthy direct Iroh route.
4765                }
4766                #[cfg(target_arch = "wasm32")]
4767                {
4768                    let _ = self
4769                        .confirm_managed_connection_readiness(&connection_id)
4770                        .await;
4771                    self.emit_current_wasm_connection_state(&connection_id)
4772                        .await;
4773                }
4774                clear_auto_connect_failure_state(
4775                    &remote_device_id,
4776                    failure_count,
4777                    failure_backoff_until,
4778                );
4779                non_initiator_escalation_count.remove(&remote_device_id);
4780                log_skip("already-connected");
4781                return;
4782            }
4783        } else {
4784            connected_health_checked_at.remove(&remote_device_id);
4785            connected_health_state.remove(&remote_device_id);
4786            connected_health_failures.remove(&remote_device_id);
4787        }
4788
4789        if let Some(backoff_until) = failure_backoff_until.get(&remote_device_id).copied() {
4790            if now < backoff_until {
4791                log_skip("failure-backoff");
4792                return;
4793            }
4794            failure_backoff_until.remove(&remote_device_id);
4795        }
4796
4797        let last = last_attempt_at.get(&remote_device_id).cloned().unwrap_or(0);
4798        if now - last < retry_throttle_ms {
4799            log_skip("retry-throttle");
4800            return;
4801        }
4802        last_attempt_at.insert(remote_device_id.clone(), now);
4803        let connect_mode = "ticket-endpoint-addr";
4804
4805        #[cfg(not(target_arch = "wasm32"))]
4806        {
4807            self.connection_manager
4808                .upsert_pending(
4809                    connection_id.clone(),
4810                    Some(node_id_str.clone()),
4811                    Some(remote_device_id.clone()),
4812                    Some(node_id_str.clone()),
4813                )
4814                .await;
4815            self.connection_manager.set_connecting(&connection_id).await;
4816            let dial_attempt_fence = self
4817                .connection_manager
4818                .get_by_connection_id(&connection_id)
4819                .await
4820                .map(|record| (record.transport_generation, record.transport_stable_id));
4821
4822            eprintln!(
4823                "[pluto-rtc][auto-connect] attempt remote_device_id={} local_device_id={} local_node_id={} remote_node_id={} mode={}",
4824                remote_device_id,
4825                local_device_id,
4826                local_node_id,
4827                node_id_str,
4828                connect_mode
4829            );
4830
4831            eprintln!("[pluto-rtc] auto-connecting to {}", remote_device_id);
4832            let result = self.ensure_connected_addr(endpoint_id, endpoint_addr).await;
4833
4834            match result {
4835                Ok(()) => {
4836                    if self.is_auto_connect_peer_excluded(&remote_device_id, Some(&node_id_str)) {
4837                        log_skip("auto-connect-excluded-after-dial");
4838                        let _ = self
4839                            .disconnect_with_reason(
4840                                endpoint_id,
4841                                crate::lifecycle_reason::REASON_MANUAL_DISCONNECT,
4842                            )
4843                            .await;
4844                        self.connection_manager
4845                            .set_closed(
4846                                &connection_id,
4847                                Some(crate::lifecycle_reason::REASON_MANUAL_DISCONNECT.to_string()),
4848                            )
4849                            .await;
4850                        return;
4851                    }
4852                    let active_probe = self.validate_active_connection(endpoint_id).await;
4853                    let active_probe_stable =
4854                        active_probe.as_ref().is_some_and(|probe| probe.responsive);
4855                    let transport_alive = self.is_connection_transport_alive(endpoint_id).await;
4856                    let probe_decision =
4857                        active_connection_probe_decision(active_probe_stable, transport_alive);
4858                    let stable = matches!(
4859                        probe_decision,
4860                        ActiveConnectionProbeDecision::Healthy
4861                            | ActiveConnectionProbeDecision::PreserveLiveTransport
4862                    );
4863                    if !stable {
4864                        if self.is_app_execution_suspended() {
4865                            log_skip("backgrounded-after-stability-probe");
4866                            return;
4867                        }
4868                        self.connection_manager
4869                            .set_failed(
4870                                &connection_id,
4871                                Some("connection failed stability verification".to_string()),
4872                            )
4873                            .await;
4874                        if let Some(transport_stable_id) =
4875                            active_probe.as_ref().map(|probe| probe.transport_stable_id)
4876                        {
4877                            let _ = self
4878                                .disconnect_transport_generation_with_reason(
4879                                    endpoint_id,
4880                                    transport_stable_id,
4881                                    crate::lifecycle_reason::REASON_CONNECTION_FAILED_STABILITY,
4882                                )
4883                                .await;
4884                        }
4885                        self.reconcile_authoritative_device_node(&remote_device_id, &node_id_str)
4886                            .await;
4887                        register_auto_connect_failure(
4888                            &remote_device_id,
4889                            now_millis_i64(),
4890                            failure_count,
4891                            failure_backoff_until,
4892                        );
4893                        let current_failures =
4894                            failure_count.get(&remote_device_id).copied().unwrap_or(0);
4895                        if current_failures >= network_change_failure_threshold
4896                            && self
4897                                .should_force_network_change_after_connect_failures(
4898                                    &remote_device_id,
4899                                )
4900                                .await
4901                        {
4902                            self.maybe_force_network_change_for_auto_connect(
4903                                user_id,
4904                                local_device_id,
4905                                "unstable-connection",
4906                                last_network_change_at_ms,
4907                                network_change_recovery_interval_ms,
4908                                last_presence_republish_at_ms,
4909                                presence_republish_interval_ms,
4910                            )
4911                            .await;
4912                        }
4913                        eprintln!(
4914                            "[pluto-rtc][auto-connect] unstable connection suppressed remote_device_id={} local_device_id={} local_node_id={} remote_node_id={} connection_id={}",
4915                            remote_device_id,
4916                            local_device_id,
4917                            local_node_id,
4918                            node_id_str,
4919                            connection_id
4920                        );
4921                        return;
4922                    }
4923                    if matches!(
4924                        probe_decision,
4925                        ActiveConnectionProbeDecision::PreserveLiveTransport
4926                    ) {
4927                        log_skip("stability-probe-missed-preserving-live-transport");
4928                    }
4929                    let current_transport_stable_id = self
4930                        .get_connection(endpoint_id)
4931                        .await
4932                        .map(|connection| crate::transport_generation::for_connection(&connection));
4933                    self.commit_current_native_transport_record(
4934                        &connection_id,
4935                        &node_id_str,
4936                        Some(remote_device_id.clone()),
4937                        current_transport_stable_id,
4938                        Some(connect_mode.to_string()),
4939                    )
4940                    .await;
4941
4942                    if let Some(ref token) = extracted_token {
4943                        let bilateral_user_device = extracted_token_payload
4944                            .as_ref()
4945                            .is_some_and(|payload| payload.scope.as_str() == "user-device");
4946                        let has_remote_admission_proof =
4947                            current_transport_stable_id.is_some_and(|transport_stable_id| {
4948                                self.remote_session_token_admitted_for_transport(
4949                                    &connection_id,
4950                                    transport_stable_id,
4951                                )
4952                            });
4953                        let has_current_inbound_admission_proof = current_transport_stable_id
4954                            .is_some_and(|transport_stable_id| {
4955                                self.inbound_session_token_admitted_for_transport(
4956                                    &connection_id,
4957                                    transport_stable_id,
4958                                )
4959                            });
4960                        let peer_requested_reciprocal =
4961                            self.has_pending_reciprocal_session_admission_request(&connection_id);
4962                        let presentation_decision = if bilateral_user_device {
4963                            session_admission_presentation_decision(
4964                                true,
4965                                has_remote_admission_proof,
4966                                has_current_inbound_admission_proof,
4967                                peer_requested_reciprocal,
4968                                local_node_id > node_id_str,
4969                            )
4970                        } else {
4971                            SessionAdmissionPresentationDecision {
4972                                should_present: !has_remote_admission_proof,
4973                                request_reciprocal: false,
4974                            }
4975                        };
4976                        if !presentation_decision.should_present {
4977                            eprintln!(
4978                                "[pluto-rtc][auto-connect][session-admission] awaiting elected presenter on fresh transport connection_id={} remote_device_id={} remote_node_id={} local_node_id={}",
4979                                connection_id,
4980                                remote_device_id,
4981                                node_id_str,
4982                                local_node_id,
4983                            );
4984                            return;
4985                        }
4986                        match self
4987                            .present_and_accept_session_token_for_route_repair(
4988                                endpoint_id,
4989                                &connection_id,
4990                                token,
4991                                extracted_token_suffix.as_deref(),
4992                                Some(remote_device_id.clone()),
4993                                Some(local_device_id.to_string()),
4994                                has_remote_admission_proof,
4995                                presentation_decision.request_reciprocal,
4996                            )
4997                            .await
4998                        {
4999                            Ok(approval_scope) => {
5000                                eprintln!(
5001                                    "[pluto-rtc][auto-connect] session-token presented and approved connection_id={} scope={} device_id={} token_fp={} local_admission=accepted",
5002                                    connection_id,
5003                                    approval_scope,
5004                                    remote_device_id,
5005                                    super::core_impl::log_fingerprint(token.as_str())
5006                                );
5007                            }
5008                            Err(error) => {
5009                                if session_token_presentation_failure_action(&error)
5010                                    == SessionTokenPresentationFailureAction::RetryOnNextTransport
5011                                {
5012                                    register_auto_connect_failure(
5013                                        &remote_device_id,
5014                                        now_millis_i64(),
5015                                        failure_count,
5016                                        failure_backoff_until,
5017                                    );
5018                                    eprintln!(
5019                                        "[pluto-rtc][auto-connect] transient session-token presentation failure; preserving peer session for canonical transport retry connection_id={} device_id={} token_fp={} error={}",
5020                                        connection_id,
5021                                        remote_device_id,
5022                                        super::core_impl::log_fingerprint(token.as_str()),
5023                                        error
5024                                    );
5025                                    return;
5026                                }
5027                                let rejection_reason =
5028                                    format!("session-token-presentation-failed: {}", error);
5029                                self.reject_session_connection(
5030                                    &connection_id,
5031                                    rejection_reason.as_str(),
5032                                );
5033                                self.connection_manager
5034                                    .set_failed(&connection_id, Some(rejection_reason.clone()))
5035                                    .await;
5036                                let _ = self
5037                                    .disconnect_with_reason(
5038                                        endpoint_id,
5039                                        crate::lifecycle_reason::REASON_SESSION_ADMISSION_REJECTED,
5040                                    )
5041                                    .await;
5042                                self.reconcile_authoritative_device_node(
5043                                    &remote_device_id,
5044                                    &node_id_str,
5045                                )
5046                                .await;
5047                                // Clear the rejection state so the next auto-connect
5048                                // attempt can re-present a (potentially refreshed) token.
5049                                // Without this the session stays in Rejected and no
5050                                // further token presentation occurs for this peer.
5051                                self.forget_session_connection(&connection_id);
5052                                register_auto_connect_admission_rejection(
5053                                    &remote_device_id,
5054                                    now_millis_i64(),
5055                                    failure_count,
5056                                    failure_backoff_until,
5057                                );
5058                                eprintln!(
5059                                    "[pluto-rtc][auto-connect] session-token presentation failed connection_id={} device_id={} token_fp={} error={}",
5060                                    connection_id,
5061                                    remote_device_id,
5062                                    super::core_impl::log_fingerprint(token.as_str()),
5063                                    error
5064                                );
5065                                return;
5066                            }
5067                        }
5068                    }
5069
5070                    let _ = self
5071                        .confirm_managed_connection_readiness(&connection_id)
5072                        .await;
5073                    // Spawn a path-watcher for this outgoing connection so that
5074                    // relay → direct-QUIC (and back) transitions retrigger or
5075                    // suspend transport upgrades reactively — matching the
5076                    // behaviour we already have for incoming connections.
5077                    if let Some(outgoing_conn) = self.get_connection(endpoint_id).await {
5078                        // force_restart=false: the path-watcher fires on relay↔direct
5079                        // transitions after connection. A Connecting session started just
5080                        // above should not be replaced by path-change retriggers.
5081                        self.spawn_iroh_path_watcher(
5082                            &connection_id,
5083                            &node_id_str,
5084                            &outgoing_conn,
5085                            false,
5086                        );
5087                    }
5088                    clear_auto_connect_failure_state(
5089                        &remote_device_id,
5090                        failure_count,
5091                        failure_backoff_until,
5092                    );
5093                    non_initiator_escalation_count.remove(&remote_device_id);
5094                    eprintln!(
5095                        "[pluto-rtc][auto-connect] connected remote_device_id={} local_device_id={} local_node_id={} remote_node_id={} connection_id={}",
5096                        remote_device_id,
5097                        local_device_id,
5098                        local_node_id,
5099                        node_id_str,
5100                        connection_id
5101                    );
5102                }
5103                Err(e) => {
5104                    if let Some((transport_generation, transport_stable_id)) = dial_attempt_fence {
5105                        let failed = self
5106                            .connection_manager
5107                            .set_failed_if_dial_attempt_current(
5108                                &connection_id,
5109                                transport_generation,
5110                                transport_stable_id,
5111                                Some(
5112                                    crate::lifecycle_reason::REASON_AUTO_CONNECT_FAILURE
5113                                        .to_string(),
5114                                ),
5115                            )
5116                            .await;
5117                        if failed.is_none() {
5118                            eprintln!(
5119                                "[pluto-rtc][auto-connect] ignored stale dial failure connection_id={} remote_device_id={} remote_node_id={} expected_transport_generation={} expected_transport_stable_id={:?}",
5120                                connection_id,
5121                                remote_device_id,
5122                                node_id_str,
5123                                transport_generation,
5124                                transport_stable_id,
5125                            );
5126                            // The successor's accept/admission path owns all
5127                            // remaining settlement. Do not charge this stale
5128                            // attempt to backoff or let its failure threshold
5129                            // disconnect the current physical generation.
5130                            return;
5131                        }
5132                    }
5133                    register_auto_connect_failure(
5134                        &remote_device_id,
5135                        now_millis_i64(),
5136                        failure_count,
5137                        failure_backoff_until,
5138                    );
5139                    let current_failures =
5140                        failure_count.get(&remote_device_id).copied().unwrap_or(0);
5141                    if current_failures >= network_change_failure_threshold
5142                        && self
5143                            .should_force_network_change_after_connect_failures(&remote_device_id)
5144                            .await
5145                    {
5146                        let _ = self
5147                            .disconnect_with_reason(
5148                                endpoint_id,
5149                                crate::lifecycle_reason::REASON_NETWORK_CHANGE_RECONNECT,
5150                            )
5151                            .await;
5152                        self.reconcile_authoritative_device_node(&remote_device_id, &node_id_str)
5153                            .await;
5154                        self.maybe_force_network_change_for_auto_connect(
5155                            user_id,
5156                            local_device_id,
5157                            "repeated-connect-failures",
5158                            last_network_change_at_ms,
5159                            network_change_recovery_interval_ms,
5160                            last_presence_republish_at_ms,
5161                            presence_republish_interval_ms,
5162                        )
5163                        .await;
5164                    }
5165                    self.maybe_refresh_gateway_presence_for_auto_connect(
5166                        user_id,
5167                        local_device_id,
5168                        "connect-failed",
5169                        last_presence_republish_at_ms,
5170                        presence_republish_interval_ms,
5171                    )
5172                    .await;
5173                    eprintln!(
5174                        "[pluto-rtc] auto-connect failed to {}: {}",
5175                        remote_device_id, e
5176                    );
5177                }
5178            }
5179        }
5180
5181        #[cfg(target_arch = "wasm32")]
5182        {
5183            self.connection_manager
5184                .upsert_pending(
5185                    connection_id.clone(),
5186                    Some(node_id_str.clone()),
5187                    Some(remote_device_id.clone()),
5188                    Some(node_id_str.clone()),
5189                )
5190                .await;
5191            // Pre-populate device_id from the signaling record so that peer
5192            // lookups by device_id succeed immediately instead of waiting
5193            // for the accept probe to backfill it (~18 s in worst case).
5194            self.connection_manager
5195                .set_device_id(&connection_id, remote_device_id.clone())
5196                .await;
5197            self.connection_manager.set_connecting(&connection_id).await;
5198            self.emit_current_wasm_connection_state(&connection_id)
5199                .await;
5200            web_sys::console::log_1(&wasm_bindgen::JsValue::from_str(&format!(
5201                "[pluto-rtc][auto-connect] attempt remote_device_id={} local_device_id={} local_node_id={} remote_node_id={} mode={}",
5202                remote_device_id,
5203                local_device_id,
5204                local_node_id,
5205                node_id_str,
5206                connect_mode
5207            )));
5208            web_sys::console::log_1(&wasm_bindgen::JsValue::from_str(&format!(
5209                "[pluto-rtc] auto-connecting to {}",
5210                remote_device_id
5211            )));
5212            let result = self.ensure_connected_addr(endpoint_id, endpoint_addr).await;
5213
5214            match result {
5215                Ok(()) => {
5216                    let active_probe_stable = self.validate_active_connection(endpoint_id).await;
5217                    let transport_alive = self.is_connection_transport_alive(endpoint_id).await;
5218                    let probe_decision =
5219                        active_connection_probe_decision(active_probe_stable, transport_alive);
5220                    let stable = matches!(
5221                        probe_decision,
5222                        ActiveConnectionProbeDecision::Healthy
5223                            | ActiveConnectionProbeDecision::PreserveLiveTransport
5224                    );
5225                    if !stable {
5226                        if self.is_app_execution_suspended() {
5227                            log_skip("backgrounded-after-stability-probe");
5228                            return;
5229                        }
5230                        self.connection_manager
5231                            .upsert_pending(
5232                                connection_id.clone(),
5233                                Some(node_id_str.clone()),
5234                                Some(remote_device_id.clone()),
5235                                Some(node_id_str.clone()),
5236                            )
5237                            .await;
5238                        self.connection_manager
5239                            .set_failed(
5240                                &connection_id,
5241                                Some("connection failed stability verification".to_string()),
5242                            )
5243                            .await;
5244                        self.emit_current_wasm_connection_state(&connection_id)
5245                            .await;
5246                        register_auto_connect_failure(
5247                            &remote_device_id,
5248                            now_millis_i64(),
5249                            failure_count,
5250                            failure_backoff_until,
5251                        );
5252                        let _ = self
5253                            .disconnect_with_reason(
5254                                endpoint_id,
5255                                crate::lifecycle_reason::REASON_AUTO_CONNECT_FAILURE,
5256                            )
5257                            .await;
5258                        web_sys::console::warn_1(&wasm_bindgen::JsValue::from_str(&format!(
5259                            "[pluto-rtc][auto-connect] unstable connection suppressed remote_device_id={} local_device_id={} local_node_id={} remote_node_id={}",
5260                            remote_device_id,
5261                            local_device_id,
5262                            local_node_id,
5263                            node_id_str
5264                        )));
5265                        return;
5266                    }
5267                    if matches!(
5268                        probe_decision,
5269                        ActiveConnectionProbeDecision::PreserveLiveTransport
5270                    ) {
5271                        web_sys::console::log_1(&wasm_bindgen::JsValue::from_str(&format!(
5272                            "[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={}",
5273                            remote_device_id,
5274                            local_device_id,
5275                            local_node_id,
5276                            node_id_str,
5277                            connection_id
5278                        )));
5279                    }
5280
5281                    self.connection_manager
5282                        .upsert_pending(
5283                            connection_id.clone(),
5284                            Some(node_id_str.clone()),
5285                            Some(remote_device_id.clone()),
5286                            Some(node_id_str.clone()),
5287                        )
5288                        .await;
5289                    let pending_record = self
5290                        .connection_manager
5291                        .get_by_connection_id(&connection_id)
5292                        .await;
5293                    web_sys::console::log_1(&wasm_bindgen::JsValue::from_str(&format!(
5294                        "[pluto-rtc][auto-connect][wasm] upsert_pending remote_device_id={} remote_node_id={} connection_id={} pending_record_exists={} record={:?}",
5295                        remote_device_id,
5296                        node_id_str,
5297                        connection_id,
5298                        pending_record.is_some(),
5299                        pending_record,
5300                    )));
5301                    let transport_stable_id = self
5302                        .get_connection(endpoint_id)
5303                        .await
5304                        .map(|connection| crate::transport_generation::for_connection(&connection));
5305                    self.commit_current_native_transport_record(
5306                        &connection_id,
5307                        &node_id_str,
5308                        Some(remote_device_id.clone()),
5309                        transport_stable_id,
5310                        Some(connect_mode.to_string()),
5311                    )
5312                    .await;
5313                    let active_transport =
5314                        self.iroh_path_kind(&node_id_str).await.transport_label();
5315                    let _ = self
5316                        .report_transport_status_for_current_generation(
5317                            &connection_id,
5318                            active_transport,
5319                            None,
5320                        )
5321                        .await;
5322
5323                    if let Some(ref token) = extracted_token {
5324                        match self
5325                            .present_and_accept_session_token(
5326                                endpoint_id,
5327                                &connection_id,
5328                                token,
5329                                extracted_token_suffix.as_deref(),
5330                                Some(remote_device_id.clone()),
5331                            )
5332                            .await
5333                        {
5334                            Ok(approval_scope) => {
5335                                web_sys::console::log_1(&wasm_bindgen::JsValue::from_str(&format!(
5336                                    "[pluto-rtc][auto-connect][wasm] session-token presented and approved connection_id={} scope={} device_id={} token_fp={} local_admission=accepted",
5337                                    connection_id,
5338                                    approval_scope,
5339                                    remote_device_id,
5340                                    super::core_impl::log_fingerprint(token.as_str())
5341                                )));
5342                            }
5343                            Err(error) => {
5344                                if session_token_presentation_failure_action(&error)
5345                                    == SessionTokenPresentationFailureAction::RetryOnNextTransport
5346                                {
5347                                    register_auto_connect_failure(
5348                                        &remote_device_id,
5349                                        now_millis_i64(),
5350                                        failure_count,
5351                                        failure_backoff_until,
5352                                    );
5353                                    web_sys::console::warn_1(&wasm_bindgen::JsValue::from_str(&format!(
5354                                        "[pluto-rtc][auto-connect][wasm] transient session-token presentation failure; preserving peer session for canonical transport retry connection_id={} device_id={} token_fp={} error={}",
5355                                        connection_id,
5356                                        remote_device_id,
5357                                        super::core_impl::log_fingerprint(token.as_str()),
5358                                        error
5359                                    )));
5360                                    return;
5361                                }
5362                                let rejection_reason =
5363                                    format!("session-token-presentation-failed: {}", error);
5364                                self.reject_session_connection(
5365                                    &connection_id,
5366                                    rejection_reason.as_str(),
5367                                );
5368                                self.connection_manager
5369                                    .set_failed(&connection_id, Some(rejection_reason))
5370                                    .await;
5371                                self.emit_current_wasm_connection_state(&connection_id)
5372                                    .await;
5373                                let _ = self
5374                                    .disconnect_with_reason(
5375                                        endpoint_id,
5376                                        crate::lifecycle_reason::REASON_SESSION_ADMISSION_REJECTED,
5377                                    )
5378                                    .await;
5379                                // Token presentation failure belongs to this
5380                                // transport attempt. Let a later deterministic
5381                                // reconnect authenticate from a clean window.
5382                                self.forget_session_connection(&connection_id);
5383                                register_auto_connect_admission_rejection(
5384                                    &remote_device_id,
5385                                    now_millis_i64(),
5386                                    failure_count,
5387                                    failure_backoff_until,
5388                                );
5389                                web_sys::console::warn_1(&wasm_bindgen::JsValue::from_str(&format!(
5390                                    "[pluto-rtc][auto-connect][wasm] session-token presentation failed connection_id={} device_id={} token_fp={} error={}",
5391                                    connection_id,
5392                                    remote_device_id,
5393                                    super::core_impl::log_fingerprint(token.as_str()),
5394                                    error
5395                                )));
5396                                return;
5397                            }
5398                        }
5399                    }
5400
5401                    let _ = self
5402                        .confirm_managed_connection_readiness(&connection_id)
5403                        .await;
5404                    let connected_record = self
5405                        .connection_manager
5406                        .get_by_connection_id(&connection_id)
5407                        .await;
5408                    let record_count = self.connection_manager.list_all().await.len();
5409                    web_sys::console::log_1(&wasm_bindgen::JsValue::from_str(&format!(
5410                        "[pluto-rtc][auto-connect][wasm] set_connected remote_device_id={} remote_node_id={} connection_id={} connected_record_exists={} total_records={} record={:?}",
5411                        remote_device_id,
5412                        node_id_str,
5413                        connection_id,
5414                        connected_record.is_some(),
5415                        record_count,
5416                        connected_record,
5417                    )));
5418                    self.emit_current_wasm_connection_state(&connection_id)
5419                        .await;
5420                    clear_auto_connect_failure_state(
5421                        &remote_device_id,
5422                        failure_count,
5423                        failure_backoff_until,
5424                    );
5425                    non_initiator_escalation_count.remove(&remote_device_id);
5426                    web_sys::console::log_1(&wasm_bindgen::JsValue::from_str(&format!(
5427                        "[pluto-rtc][auto-connect] connected remote_device_id={} local_device_id={} local_node_id={} remote_node_id={}",
5428                        remote_device_id,
5429                        local_device_id,
5430                        local_node_id,
5431                        node_id_str
5432                    )));
5433                }
5434                Err(e) => {
5435                    let err = e.to_string();
5436                    self.connection_manager
5437                        .upsert_pending(
5438                            connection_id.clone(),
5439                            Some(node_id_str.clone()),
5440                            Some(remote_device_id.clone()),
5441                            Some(node_id_str.clone()),
5442                        )
5443                        .await;
5444                    self.connection_manager
5445                        .set_failed(&connection_id, Some(err.clone()))
5446                        .await;
5447                    self.emit_current_wasm_connection_state(&connection_id)
5448                        .await;
5449                    register_auto_connect_failure(
5450                        &remote_device_id,
5451                        now_millis_i64(),
5452                        failure_count,
5453                        failure_backoff_until,
5454                    );
5455                    web_sys::console::error_1(&wasm_bindgen::JsValue::from_str(&format!(
5456                        "[pluto-rtc] auto-connect failed to {}: {}",
5457                        remote_device_id, err
5458                    )));
5459                }
5460            }
5461        }
5462    }
5463
5464    #[cfg(not(target_arch = "wasm32"))]
5465    async fn run_auto_connect_snapshot(
5466        self: &Arc<Self>,
5467        user_id: &str,
5468        local_device_id: &str,
5469        last_attempt_at: &mut std::collections::HashMap<String, i64>,
5470        failure_count: &mut std::collections::HashMap<String, u8>,
5471        failure_backoff_until: &mut std::collections::HashMap<String, i64>,
5472        retry_throttle_ms: i64,
5473        non_initiator_wait_started_at: &mut std::collections::HashMap<String, i64>,
5474        non_initiator_escalation_count: &mut std::collections::HashMap<String, u8>,
5475        initiator_grace_ms: i64,
5476        skip_log_at: &mut std::collections::HashMap<String, i64>,
5477        connected_health_checked_at: &mut std::collections::HashMap<String, i64>,
5478        connected_health_state: &mut std::collections::HashMap<String, bool>,
5479        connected_health_failures: &mut std::collections::HashMap<String, u8>,
5480        connected_health_probe_ms: i64,
5481        connected_health_failure_threshold: u8,
5482        last_presence_republish_at_ms: &mut i64,
5483        presence_republish_interval_ms: i64,
5484        last_network_change_at_ms: &mut i64,
5485        network_change_recovery_interval_ms: i64,
5486        network_change_failure_threshold: u8,
5487        last_snapshot_signature: &mut Option<(usize, usize, usize)>,
5488        last_snapshot_log_at: &mut i64,
5489        last_known_node_id: &mut std::collections::HashMap<String, String>,
5490        known_devices: &mut std::collections::HashMap<String, crate::signaling::Device>,
5491    ) {
5492        if self.is_app_execution_suspended() {
5493            return;
5494        }
5495
5496        let now = now_millis_i64();
5497        match self.search_devices(user_id).await {
5498            Ok(devices) => {
5499                let total = devices.len();
5500                let online = devices.iter().filter(|d| d.online).count();
5501                let remote_online = devices
5502                    .iter()
5503                    .filter(|d| d.online && d.device_id != local_device_id)
5504                    .count();
5505                let signature = (total, online, remote_online);
5506                let changed = last_snapshot_signature
5507                    .map(|prev| prev != signature)
5508                    .unwrap_or(true);
5509                let should_log = if auto_connect_verbose() {
5510                    now.saturating_sub(*last_snapshot_log_at) >= 2_000
5511                } else {
5512                    changed || now.saturating_sub(*last_snapshot_log_at) >= 30_000
5513                };
5514                if should_log {
5515                    *last_snapshot_log_at = now;
5516                    *last_snapshot_signature = Some(signature);
5517                    #[cfg(not(target_arch = "wasm32"))]
5518                    eprintln!(
5519                        "[pluto-rtc][auto-connect][snapshot] user_id={} local_device_id={} total_devices={} online_devices={} remote_online_devices={}",
5520                        user_id, local_device_id, total, online, remote_online
5521                    );
5522                    #[cfg(target_arch = "wasm32")]
5523                    web_sys::console::log_1(&wasm_bindgen::JsValue::from_str(&format!(
5524                        "[pluto-rtc][auto-connect][snapshot] user_id={} local_device_id={} total_devices={} online_devices={} remote_online_devices={}",
5525                        user_id, local_device_id, total, online, remote_online
5526                    )));
5527                }
5528
5529                let known_device_ids: std::collections::HashSet<String> =
5530                    devices.iter().map(|d| d.device_id.clone()).collect();
5531
5532                for device in devices {
5533                    known_devices.insert(device.device_id.clone(), device.clone());
5534                    self.try_auto_connect_device(
5535                        user_id,
5536                        device,
5537                        local_device_id,
5538                        last_attempt_at,
5539                        failure_count,
5540                        failure_backoff_until,
5541                        retry_throttle_ms,
5542                        non_initiator_wait_started_at,
5543                        non_initiator_escalation_count,
5544                        initiator_grace_ms,
5545                        skip_log_at,
5546                        connected_health_checked_at,
5547                        connected_health_state,
5548                        connected_health_failures,
5549                        connected_health_probe_ms,
5550                        connected_health_failure_threshold,
5551                        last_presence_republish_at_ms,
5552                        presence_republish_interval_ms,
5553                        last_network_change_at_ms,
5554                        network_change_recovery_interval_ms,
5555                        network_change_failure_threshold,
5556                        last_known_node_id,
5557                    )
5558                    .await;
5559                }
5560
5561                // Prune tracking state for device IDs no longer returned by
5562                // the snapshot. Prevents unbounded growth as peers come and go
5563                // without always firing a DeviceEvent::Removed.
5564                last_attempt_at.retain(|k, _| known_device_ids.contains(k));
5565                failure_count.retain(|k, _| known_device_ids.contains(k));
5566                failure_backoff_until.retain(|k, _| known_device_ids.contains(k));
5567                non_initiator_wait_started_at.retain(|k, _| known_device_ids.contains(k));
5568                non_initiator_escalation_count.retain(|k, _| known_device_ids.contains(k));
5569                connected_health_checked_at.retain(|k, _| known_device_ids.contains(k));
5570                connected_health_state.retain(|k, _| known_device_ids.contains(k));
5571                connected_health_failures.retain(|k, _| known_device_ids.contains(k));
5572                last_known_node_id.retain(|key, _| {
5573                    let device_id = key
5574                        .strip_suffix("::runtime-instance")
5575                        .or_else(|| key.strip_suffix("::ticket"))
5576                        .unwrap_or(key.as_str());
5577                    known_device_ids.contains(device_id)
5578                });
5579                // skip_log_at keys are "device_id::reason" — prune by prefix
5580                skip_log_at.retain(|k, _| {
5581                    k.split_once("::")
5582                        .map_or(false, |(id, _)| known_device_ids.contains(id))
5583                });
5584            }
5585            Err(error) => {
5586                let min_interval = if auto_connect_verbose() {
5587                    2_000
5588                } else {
5589                    30_000
5590                };
5591                if now.saturating_sub(*last_snapshot_log_at) >= min_interval {
5592                    *last_snapshot_log_at = now;
5593                    #[cfg(not(target_arch = "wasm32"))]
5594                    eprintln!(
5595                        "[pluto-rtc][auto-connect][snapshot] user_id={} local_device_id={} search_devices_error={}",
5596                        user_id, local_device_id, error
5597                    );
5598                    #[cfg(target_arch = "wasm32")]
5599                    web_sys::console::error_1(&wasm_bindgen::JsValue::from_str(&format!(
5600                        "[pluto-rtc][auto-connect][snapshot] user_id={} local_device_id={} search_devices_error={}",
5601                        user_id, local_device_id, error
5602                    )));
5603                }
5604            }
5605        }
5606    }
5607
5608    #[cfg(not(target_arch = "wasm32"))]
5609    pub(crate) async fn auto_connect_loop(
5610        self: Arc<Self>,
5611        user_id: String,
5612        local_device_id: String,
5613        generation: u64,
5614    ) {
5615        use futures::StreamExt;
5616        let mut last_attempt_at: std::collections::HashMap<String, i64> =
5617            std::collections::HashMap::new();
5618        let mut non_initiator_wait_started_at: std::collections::HashMap<String, i64> =
5619            std::collections::HashMap::new();
5620        let mut non_initiator_escalation_count: std::collections::HashMap<String, u8> =
5621            std::collections::HashMap::new();
5622        let mut skip_log_at: std::collections::HashMap<String, i64> =
5623            std::collections::HashMap::new();
5624        let mut connected_health_checked_at: std::collections::HashMap<String, i64> =
5625            std::collections::HashMap::new();
5626        let mut connected_health_state: std::collections::HashMap<String, bool> =
5627            std::collections::HashMap::new();
5628        let mut connected_health_failures: std::collections::HashMap<String, u8> =
5629            std::collections::HashMap::new();
5630        let mut failure_count: std::collections::HashMap<String, u8> =
5631            std::collections::HashMap::new();
5632        let mut failure_backoff_until: std::collections::HashMap<String, i64> =
5633            std::collections::HashMap::new();
5634        let mut last_known_node_id: std::collections::HashMap<String, String> =
5635            std::collections::HashMap::new();
5636        // The provider subscription owns discovery updates. Keep its latest
5637        // device values in memory so transport health can run on its own
5638        // cadence without turning every ping into a Firestore, RTDB, gateway,
5639        // or future-provider roster read.
5640        let mut known_devices: std::collections::HashMap<String, crate::signaling::Device> =
5641            std::collections::HashMap::new();
5642        let mut last_snapshot_signature: Option<(usize, usize, usize)> = None;
5643        let mut last_snapshot_log_at = 0_i64;
5644        let mut last_presence_republish_at_ms = 0_i64;
5645        let retry_throttle_ms = 1_000;
5646        // Gateway events deliver roster and live-presence changes.
5647        // This is only a low-frequency safety reconciliation; it must never be
5648        // used as a five-second full-scope polling loop.
5649        let rescan_interval_ms_fg: u64 = 300_000;
5650        let rescan_interval_ms_bg: u64 = 300_000;
5651        let initiator_grace_ms = 3_000;
5652        // Foreground: probe every 5s, fail after 2 misses (~10s total).
5653        // Background: skip health probing entirely (i64::MAX disables the probe).
5654        let connected_health_probe_ms_fg = 5_000_i64;
5655        let connected_health_probe_ms_bg = i64::MAX;
5656        let connected_health_failure_threshold = 2;
5657        // Failure-triggered presence repair is throttled to once per 10s.
5658        // The native presence actor owns periodic gateway lease refreshes.
5659        let presence_republish_interval_ms_fg = 10_000_i64;
5660        let presence_republish_interval_ms_bg = i64::MAX;
5661        let mut last_network_change_at_ms = 0_i64;
5662        // Foreground: network change recovery every 15s. Background: skip.
5663        let network_change_recovery_interval_ms_fg = 15_000_i64;
5664        let network_change_recovery_interval_ms_bg = i64::MAX;
5665        let network_change_failure_threshold = auto_connect_network_change_failure_threshold();
5666
5667        loop {
5668            if !self.is_auto_connect_generation_current(generation) {
5669                #[cfg(not(target_arch = "wasm32"))]
5670                eprintln!(
5671                    "[pluto-rtc][auto-connect] stopping stale loop user_id={} local_device_id={} generation={}",
5672                    user_id, local_device_id, generation
5673                );
5674                #[cfg(target_arch = "wasm32")]
5675                web_sys::console::log_1(&wasm_bindgen::JsValue::from_str(&format!(
5676                    "[pluto-rtc][auto-connect] stopping stale loop user_id={} local_device_id={} generation={}",
5677                    user_id, local_device_id, generation
5678                )));
5679                break;
5680            }
5681
5682            // Snapshot pass so auto-connect does not depend on receiving a specific
5683            // event type from the subscription backend.
5684            if !self.is_app_execution_suspended() {
5685                self.run_auto_connect_snapshot(
5686                    &user_id,
5687                    &local_device_id,
5688                    &mut last_attempt_at,
5689                    &mut failure_count,
5690                    &mut failure_backoff_until,
5691                    retry_throttle_ms,
5692                    &mut non_initiator_wait_started_at,
5693                    &mut non_initiator_escalation_count,
5694                    initiator_grace_ms,
5695                    &mut skip_log_at,
5696                    &mut connected_health_checked_at,
5697                    &mut connected_health_state,
5698                    &mut connected_health_failures,
5699                    connected_health_probe_ms_fg,
5700                    connected_health_failure_threshold,
5701                    &mut last_presence_republish_at_ms,
5702                    presence_republish_interval_ms_fg,
5703                    &mut last_network_change_at_ms,
5704                    network_change_recovery_interval_ms_fg,
5705                    network_change_failure_threshold,
5706                    &mut last_snapshot_signature,
5707                    &mut last_snapshot_log_at,
5708                    &mut last_known_node_id,
5709                    &mut known_devices,
5710                )
5711                .await;
5712            }
5713
5714            #[allow(unused_mut)]
5715            let mut stream = match self.subscribe_devices(&user_id).await {
5716                Ok(s) => {
5717                    #[cfg(not(target_arch = "wasm32"))]
5718                    eprintln!(
5719                        "[pluto-rtc][auto-connect] subscribed user_id={} local_device_id={}",
5720                        user_id, local_device_id
5721                    );
5722                    #[cfg(target_arch = "wasm32")]
5723                    web_sys::console::log_1(&wasm_bindgen::JsValue::from_str(&format!(
5724                        "[pluto-rtc][auto-connect] subscribed user_id={} local_device_id={}",
5725                        user_id, local_device_id
5726                    )));
5727                    s
5728                }
5729                Err(e) => {
5730                    #[cfg(not(target_arch = "wasm32"))]
5731                    eprintln!(
5732                        "[pluto-rtc][auto-connect] subscribe failed user_id={} local_device_id={} error={}",
5733                        user_id, local_device_id, e
5734                    );
5735                    #[cfg(target_arch = "wasm32")]
5736                    web_sys::console::error_1(&wasm_bindgen::JsValue::from_str(&format!(
5737                        "[pluto-rtc][auto-connect] subscribe failed user_id={} local_device_id={} error={}",
5738                        user_id, local_device_id, e
5739                    )));
5740
5741                    #[cfg(not(target_arch = "wasm32"))]
5742                    tokio::time::sleep(std::time::Duration::from_millis(1500)).await;
5743                    #[cfg(target_arch = "wasm32")]
5744                    gloo_timers::future::sleep(std::time::Duration::from_millis(1500)).await;
5745                    continue;
5746                }
5747            };
5748
5749            #[cfg(not(target_arch = "wasm32"))]
5750            {
5751                let mut last_bg_state = self.is_app_execution_suspended();
5752                let mut rescan_interval =
5753                    tokio::time::interval(std::time::Duration::from_millis(if last_bg_state {
5754                        rescan_interval_ms_bg
5755                    } else {
5756                        rescan_interval_ms_fg
5757                    }));
5758                let mut health_interval = tokio::time::interval(std::time::Duration::from_millis(
5759                    connected_health_probe_ms_fg as u64,
5760                ));
5761                health_interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
5762                loop {
5763                    if !self.is_auto_connect_generation_current(generation) {
5764                        break;
5765                    }
5766                    // Recreate the interval when background state changes so the
5767                    // new period takes effect immediately rather than after the
5768                    // current (possibly 30s) tick expires.
5769                    let bg = self.is_app_execution_suspended();
5770                    if bg != last_bg_state {
5771                        last_bg_state = bg;
5772                        rescan_interval =
5773                            tokio::time::interval(std::time::Duration::from_millis(if bg {
5774                                rescan_interval_ms_bg
5775                            } else {
5776                                rescan_interval_ms_fg
5777                            }));
5778                    }
5779                    let connected_health_probe_ms = if bg {
5780                        connected_health_probe_ms_bg
5781                    } else {
5782                        connected_health_probe_ms_fg
5783                    };
5784                    let presence_republish_interval_ms = if bg {
5785                        presence_republish_interval_ms_bg
5786                    } else {
5787                        presence_republish_interval_ms_fg
5788                    };
5789                    let network_change_recovery_interval_ms = if bg {
5790                        network_change_recovery_interval_ms_bg
5791                    } else {
5792                        network_change_recovery_interval_ms_fg
5793                    };
5794                    // A discovery event can put this peer into the deterministic
5795                    // non-initiator grace window before the incoming transport
5796                    // arrives. Wake at that exact policy deadline so reciprocal
5797                    // admission is not accidentally coupled to a presence lease
5798                    // refresh or the low-frequency safety rescan.
5799                    let grace_wakeup_delay_ms = next_non_initiator_wakeup_delay_ms(
5800                        &non_initiator_wait_started_at,
5801                        &non_initiator_escalation_count,
5802                        now_millis_i64(),
5803                    );
5804                    let grace_wakeup = tokio::time::sleep(std::time::Duration::from_millis(
5805                        grace_wakeup_delay_ms.unwrap_or(rescan_interval_ms_fg),
5806                    ));
5807                    tokio::pin!(grace_wakeup);
5808                    tokio::select! {
5809                        _ = self.native_auto_connect_wake.notified() => {
5810                            if self.is_app_execution_suspended() {
5811                                continue;
5812                            }
5813                            // Admission proof changes are typed lifecycle
5814                            // input even when the provider roster and physical
5815                            // transport are unchanged. Reconcile through this
5816                            // actor's normal admission/tie-break path now.
5817                            self.run_auto_connect_snapshot(
5818                                &user_id,
5819                                &local_device_id,
5820                                &mut last_attempt_at,
5821                                &mut failure_count,
5822                                &mut failure_backoff_until,
5823                                retry_throttle_ms,
5824                                &mut non_initiator_wait_started_at,
5825                                &mut non_initiator_escalation_count,
5826                                initiator_grace_ms,
5827                                &mut skip_log_at,
5828                                &mut connected_health_checked_at,
5829                                &mut connected_health_state,
5830                                &mut connected_health_failures,
5831                                connected_health_probe_ms,
5832                                connected_health_failure_threshold,
5833                                &mut last_presence_republish_at_ms,
5834                                presence_republish_interval_ms,
5835                                &mut last_network_change_at_ms,
5836                                network_change_recovery_interval_ms,
5837                                network_change_failure_threshold,
5838                                &mut last_snapshot_signature,
5839                                &mut last_snapshot_log_at,
5840                                &mut last_known_node_id,
5841                                &mut known_devices,
5842                            ).await;
5843                        }
5844                        _ = rescan_interval.tick() => {
5845                            if self.is_app_execution_suspended() {
5846                                continue;
5847                            }
5848                            self.run_auto_connect_snapshot(
5849                                &user_id,
5850                                &local_device_id,
5851                                &mut last_attempt_at,
5852                                &mut failure_count,
5853                                &mut failure_backoff_until,
5854                                retry_throttle_ms,
5855                                &mut non_initiator_wait_started_at,
5856                                &mut non_initiator_escalation_count,
5857                                initiator_grace_ms,
5858                                &mut skip_log_at,
5859                                &mut connected_health_checked_at,
5860                                &mut connected_health_state,
5861                                &mut connected_health_failures,
5862                                connected_health_probe_ms,
5863                                connected_health_failure_threshold,
5864                                &mut last_presence_republish_at_ms,
5865                                presence_republish_interval_ms,
5866                                &mut last_network_change_at_ms,
5867                                network_change_recovery_interval_ms,
5868                                network_change_failure_threshold,
5869                                &mut last_snapshot_signature,
5870                                &mut last_snapshot_log_at,
5871                                &mut last_known_node_id,
5872                                &mut known_devices,
5873                            ).await;
5874                        }
5875                        _ = health_interval.tick() => {
5876                            if self.is_app_execution_suspended() {
5877                                continue;
5878                            }
5879
5880                            // Probe cached provider records, not the provider
5881                            // itself. Presence is coordination evidence; when a
5882                            // cached record has gone offline, an already-owned
5883                            // physical transport still receives positive
5884                            // liveness probes until that exact generation is
5885                            // retired or proves responsive.
5886                            let cached_devices: Vec<crate::signaling::Device> =
5887                                known_devices.values().cloned().collect();
5888                            for mut device in cached_devices {
5889                                if !device.online {
5890                                    let Some(ticket) = device.ticket.as_deref() else {
5891                                        continue;
5892                                    };
5893                                    let (iroh_ticket, _) =
5894                                        crate::session_token::split_ticket(ticket.trim());
5895                                    let Ok(endpoint_addr) = parse_endpoint_ticket(iroh_ticket) else {
5896                                        continue;
5897                                    };
5898                                    if !self.is_connected(endpoint_addr.id).await {
5899                                        continue;
5900                                    }
5901                                    device.online = true;
5902                                }
5903                                self.try_auto_connect_device(
5904                                    &user_id,
5905                                    device,
5906                                    &local_device_id,
5907                                    &mut last_attempt_at,
5908                                    &mut failure_count,
5909                                    &mut failure_backoff_until,
5910                                    retry_throttle_ms,
5911                                    &mut non_initiator_wait_started_at,
5912                                    &mut non_initiator_escalation_count,
5913                                    initiator_grace_ms,
5914                                    &mut skip_log_at,
5915                                    &mut connected_health_checked_at,
5916                                    &mut connected_health_state,
5917                                    &mut connected_health_failures,
5918                                    connected_health_probe_ms_fg,
5919                                    connected_health_failure_threshold,
5920                                    &mut last_presence_republish_at_ms,
5921                                    presence_republish_interval_ms_fg,
5922                                    &mut last_network_change_at_ms,
5923                                    network_change_recovery_interval_ms_fg,
5924                                    network_change_failure_threshold,
5925                                    &mut last_known_node_id,
5926                                )
5927                                .await;
5928                            }
5929                        }
5930                        _ = &mut grace_wakeup, if grace_wakeup_delay_ms.is_some() => {
5931                            if self.is_app_execution_suspended() {
5932                                continue;
5933                            }
5934                            self.run_auto_connect_snapshot(
5935                                &user_id,
5936                                &local_device_id,
5937                                &mut last_attempt_at,
5938                                &mut failure_count,
5939                                &mut failure_backoff_until,
5940                                retry_throttle_ms,
5941                                &mut non_initiator_wait_started_at,
5942                                &mut non_initiator_escalation_count,
5943                                initiator_grace_ms,
5944                                &mut skip_log_at,
5945                                &mut connected_health_checked_at,
5946                                &mut connected_health_state,
5947                                &mut connected_health_failures,
5948                                connected_health_probe_ms,
5949                                connected_health_failure_threshold,
5950                                &mut last_presence_republish_at_ms,
5951                                presence_republish_interval_ms,
5952                                &mut last_network_change_at_ms,
5953                                network_change_recovery_interval_ms,
5954                                network_change_failure_threshold,
5955                                &mut last_snapshot_signature,
5956                                &mut last_snapshot_log_at,
5957                                &mut last_known_node_id,
5958                                &mut known_devices,
5959                            ).await;
5960                        }
5961                        result = stream.next() => {
5962                            let Some(result) = result else { break; };
5963                            let events = match result {
5964                                Ok(e) => e,
5965                                Err(error) => {
5966                                    eprintln!(
5967                                        "[pluto-rtc][auto-connect] discovery subscription ended; recreating owner-managed stream: {}",
5968                                        error
5969                                    );
5970                                    break;
5971                                }
5972                            };
5973
5974                            for event in events {
5975                                match event {
5976                                    crate::signaling::DeviceEvent::Added { device }
5977                                    | crate::signaling::DeviceEvent::Modified { device } => {
5978                                        known_devices
5979                                            .insert(device.device_id.clone(), device.clone());
5980                                        self.try_auto_connect_device(
5981                                            &user_id,
5982                                            device,
5983                                            &local_device_id,
5984                                            &mut last_attempt_at,
5985                                            &mut failure_count,
5986                                            &mut failure_backoff_until,
5987                                            retry_throttle_ms,
5988                                            &mut non_initiator_wait_started_at,
5989                                            &mut non_initiator_escalation_count,
5990                                            initiator_grace_ms,
5991                                            &mut skip_log_at,
5992                                            &mut connected_health_checked_at,
5993                                            &mut connected_health_state,
5994                                            &mut connected_health_failures,
5995                                            connected_health_probe_ms_fg,
5996                                            connected_health_failure_threshold,
5997                                            &mut last_presence_republish_at_ms,
5998                                            presence_republish_interval_ms_fg,
5999                                            &mut last_network_change_at_ms,
6000                                            network_change_recovery_interval_ms_fg,
6001                                            network_change_failure_threshold,
6002                                            &mut last_known_node_id,
6003                                        )
6004                                        .await;
6005                                    }
6006                                    crate::signaling::DeviceEvent::Removed { device_id } => {
6007                                        known_devices.remove(&device_id);
6008                                        last_attempt_at.remove(&device_id);
6009                                        non_initiator_wait_started_at.remove(&device_id);
6010                                        non_initiator_escalation_count.remove(&device_id);
6011                                        skip_log_at.retain(|key, _| !key.starts_with(&format!("{}::", device_id)));
6012                                        connected_health_checked_at.remove(&device_id);
6013                                        connected_health_state.remove(&device_id);
6014                                        connected_health_failures.remove(&device_id);
6015                                        last_known_node_id
6016                                            .remove(&format!("{device_id}::runtime-instance"));
6017                                        last_known_node_id
6018                                            .remove(&format!("{device_id}::ticket"));
6019                                        clear_auto_connect_failure_state(
6020                                            &device_id,
6021                                            &mut failure_count,
6022                                            &mut failure_backoff_until,
6023                                        );
6024
6025                                        for record in self.connection_manager.get_by_device_id(&device_id).await {
6026                                            if matches!(
6027                                                record.state,
6028                                                crate::connection_manager::ConnectionState::Pending
6029                                                    | crate::connection_manager::ConnectionState::Connecting
6030                                            ) {
6031                                                self.connection_manager
6032                                                    .set_failed(
6033                                                        &record.connection_id,
6034                                                        Some("device removed from signaling snapshot".to_string()),
6035                                                    )
6036                                                    .await;
6037                                            }
6038                                        }
6039                                    }
6040                                }
6041                            }
6042                        }
6043                    }
6044                }
6045            }
6046
6047            #[cfg(target_arch = "wasm32")]
6048            {
6049                let mut stream = stream.fuse();
6050
6051                loop {
6052                    if !self.is_auto_connect_generation_current(generation) {
6053                        break;
6054                    }
6055                    let bg = self.is_app_execution_suspended();
6056                    let connected_health_probe_ms = if bg {
6057                        connected_health_probe_ms_bg
6058                    } else {
6059                        connected_health_probe_ms_fg
6060                    };
6061                    let presence_republish_interval_ms = if bg {
6062                        presence_republish_interval_ms_bg
6063                    } else {
6064                        presence_republish_interval_ms_fg
6065                    };
6066                    let network_change_recovery_interval_ms = if bg {
6067                        network_change_recovery_interval_ms_bg
6068                    } else {
6069                        network_change_recovery_interval_ms_fg
6070                    };
6071                    let rescan_ms = if bg {
6072                        rescan_interval_ms_bg
6073                    } else {
6074                        rescan_interval_ms_fg
6075                    };
6076                    let sleep =
6077                        gloo_timers::future::sleep(std::time::Duration::from_millis(rescan_ms))
6078                            .fuse();
6079                    futures::pin_mut!(sleep);
6080                    futures::select! {
6081                        _ = sleep => {
6082                            if self.is_app_execution_suspended() {
6083                                continue;
6084                            }
6085                            self.run_auto_connect_snapshot(
6086                                &user_id,
6087                                &local_device_id,
6088                                &mut last_attempt_at,
6089                                &mut failure_count,
6090                                &mut failure_backoff_until,
6091                                retry_throttle_ms,
6092                                &mut non_initiator_wait_started_at,
6093                                &mut non_initiator_escalation_count,
6094                                initiator_grace_ms,
6095                                &mut skip_log_at,
6096                                &mut connected_health_checked_at,
6097                                &mut connected_health_state,
6098                                &mut connected_health_failures,
6099                                connected_health_probe_ms,
6100                                connected_health_failure_threshold,
6101                                &mut last_presence_republish_at_ms,
6102                                presence_republish_interval_ms,
6103                                &mut last_network_change_at_ms,
6104                                network_change_recovery_interval_ms,
6105                                network_change_failure_threshold,
6106                                &mut last_snapshot_signature,
6107                                &mut last_snapshot_log_at,
6108                                &mut last_known_node_id,
6109                                &mut known_devices,
6110                            ).await;
6111                        }
6112                        result = stream.next() => {
6113                            let Some(result) = result else { break; };
6114                            let events = match result {
6115                                Ok(e) => e,
6116                                Err(error) => {
6117                                    web_sys::console::error_1(&wasm_bindgen::JsValue::from_str(
6118                                        &format!(
6119                                            "[pluto-rtc][auto-connect] discovery subscription ended; recreating owner-managed stream: {}",
6120                                            error
6121                                        ),
6122                                    ));
6123                                    break;
6124                                }
6125                            };
6126
6127                            for event in events {
6128                                match event {
6129                                    crate::signaling::DeviceEvent::Added { device }
6130                                    | crate::signaling::DeviceEvent::Modified { device } => {
6131                                        // Event-driven connects always use foreground parameters.
6132                                        self.try_auto_connect_device(
6133                                            &user_id,
6134                                            device,
6135                                            &local_device_id,
6136                                            &mut last_attempt_at,
6137                                            &mut failure_count,
6138                                            &mut failure_backoff_until,
6139                                            retry_throttle_ms,
6140                                            &mut non_initiator_wait_started_at,
6141                                            &mut non_initiator_escalation_count,
6142                                            initiator_grace_ms,
6143                                            &mut skip_log_at,
6144                                            &mut connected_health_checked_at,
6145                                            &mut connected_health_state,
6146                                            &mut connected_health_failures,
6147                                            connected_health_probe_ms_fg,
6148                                            connected_health_failure_threshold,
6149                                            &mut last_presence_republish_at_ms,
6150                                            presence_republish_interval_ms_fg,
6151                                            &mut last_network_change_at_ms,
6152                                            network_change_recovery_interval_ms_fg,
6153                                            network_change_failure_threshold,
6154                                            &mut last_known_node_id,
6155                                        )
6156                                        .await;
6157                                    }
6158                                    crate::signaling::DeviceEvent::Removed { device_id } => {
6159                                        last_attempt_at.remove(&device_id);
6160                                        non_initiator_wait_started_at.remove(&device_id);
6161                                        non_initiator_escalation_count.remove(&device_id);
6162                                        skip_log_at.retain(|key, _| !key.starts_with(&format!("{}::", device_id)));
6163                                        connected_health_checked_at.remove(&device_id);
6164                                        connected_health_state.remove(&device_id);
6165                                        connected_health_failures.remove(&device_id);
6166                                        last_known_node_id
6167                                            .remove(&format!("{device_id}::runtime-instance"));
6168                                        last_known_node_id
6169                                            .remove(&format!("{device_id}::ticket"));
6170                                        clear_auto_connect_failure_state(
6171                                            &device_id,
6172                                            &mut failure_count,
6173                                            &mut failure_backoff_until,
6174                                        );
6175
6176                                        for record in self.connection_manager.get_by_device_id(&device_id).await {
6177                                            if matches!(
6178                                                record.state,
6179                                                crate::connection_manager::ConnectionState::Pending
6180                                                    | crate::connection_manager::ConnectionState::Connecting
6181                                            ) {
6182                                                self.connection_manager
6183                                                    .set_failed(
6184                                                        &record.connection_id,
6185                                                        Some("device removed from signaling snapshot".to_string()),
6186                                                    )
6187                                                    .await;
6188                                            }
6189                                        }
6190                                    }
6191                                }
6192                            }
6193                        }
6194                    }
6195                }
6196            }
6197
6198            #[cfg(not(target_arch = "wasm32"))]
6199            eprintln!(
6200                "[pluto-rtc][auto-connect] device stream ended user_id={} local_device_id={}, resubscribing",
6201                user_id, local_device_id
6202            );
6203            #[cfg(target_arch = "wasm32")]
6204            web_sys::console::log_1(&wasm_bindgen::JsValue::from_str(&format!(
6205                "[pluto-rtc][auto-connect] device stream ended user_id={} local_device_id={}, resubscribing",
6206                user_id, local_device_id
6207            )));
6208
6209            #[cfg(not(target_arch = "wasm32"))]
6210            tokio::time::sleep(std::time::Duration::from_millis(400)).await;
6211            #[cfg(target_arch = "wasm32")]
6212            gloo_timers::future::sleep(std::time::Duration::from_millis(400)).await;
6213        }
6214    }
6215}