openrtc 0.2.1

//! Client module tests split out from `client.rs` for readability.

#[cfg(test)]
mod tests {
    use super::super::auto_connect_impl::{
        active_connection_probe_decision, ActiveConnectionProbeDecision,
    };
    use super::super::core_impl::{
        is_duplicate_kept_existing_close_reason, is_graceful_disconnect_close_reason,
        is_manual_disconnect_close_reason, is_replacement_churn_close_reason,
    };
    #[cfg(target_arch = "wasm32")]
    use super::super::state_signaling_impl::should_delay_wasm_stale_settled_snapshot_retirement;
    use super::super::state_signaling_impl::should_represent_session_token_on_connected_transport;
    #[cfg(all(not(target_arch = "wasm32"), feature = "transport-moq"))]
    use super::super::MoQConfig;
    use super::super::{
        auto_connect_failure_backoff_ms, auto_connect_tie_break_decision,
        clear_auto_connect_failure_state, duplicate_closed_handling,
        incoming_local_close_should_wait_for_replacement, merge_device_status_snapshots,
        register_auto_connect_failure, AutoConnectTieBreakDecision, Client, DeviceConnectionStatus,
        DevicePresenceStatus, DuplicateClosedHandling, IncomingTransportCloseResolution,
        ManagedConnectionBridgeAction, ManagedConnectionHealthStatus, ReadinessState,
        TransportConfig, MANAGED_SETTLE_DEADLINE_MS,
    };
    use crate::connection_manager::{ConnectionHealth, ConnectionState, PeerSnapshot};
    use crate::native_protocol::{InspectedMainFrame, NativeMainMessage, NativeMessageAction};
    #[cfg(all(feature = "transport-webrtc", not(target_arch = "wasm32")))]
    use crate::native_webrtc_policy::NativeWebRTCAttemptReservation;
    use crate::signaling::Device;
    use iroh::{Endpoint, RelayMode};
    use std::collections::HashMap;
    use tokio::time::{sleep, Duration};

    #[test]
    fn relay_only_endpoint_addr_strips_ip_addresses() {
        let secret = iroh::SecretKey::generate();
        let relay_url: iroh::RelayUrl = "https://relay.example.com".parse().unwrap();
        let ip_addr = "192.168.1.20:4433".parse().unwrap();
        let addr = iroh::EndpointAddr::new(secret.public())
            .with_relay_url(relay_url.clone())
            .with_ip_addr(ip_addr);

        let scrubbed = Client::relay_only_endpoint_addr(addr).expect("relay address");

        assert_eq!(
            scrubbed.relay_urls().cloned().collect::<Vec<_>>(),
            vec![relay_url]
        );
        assert_eq!(scrubbed.ip_addrs().count(), 0);
        assert!(scrubbed.addrs.iter().all(|addr| addr.is_relay()));
    }

    #[test]
    fn relay_only_endpoint_addr_fails_closed_without_relay() {
        let secret = iroh::SecretKey::generate();
        let addr = iroh::EndpointAddr::new(secret.public())
            .with_ip_addr("10.0.0.12:4433".parse().unwrap());

        let error = Client::relay_only_endpoint_addr(addr).unwrap_err();

        assert!(error.to_string().contains("no relay address"));
    }

    #[test]
    fn moq_empty_config_uses_default_relay_url() {
        let config: TransportConfig =
            serde_json::from_value(serde_json::json!({ "moq": {} })).unwrap();

        assert_eq!(
            config.moq.unwrap().relay_url,
            "https://relay.cloudflare.mediaoverquic.com:443/moq"
        );
    }

    async fn wait_for_endpoint_addr(
        endpoint: &Endpoint,
        timeout_ms: u64,
    ) -> anyhow::Result<iroh::EndpointAddr> {
        let started = std::time::Instant::now();
        loop {
            let addr = endpoint.addr();
            if !addr.is_empty() {
                return Ok(addr);
            }

            if started.elapsed() > Duration::from_millis(timeout_ms) {
                return Err(anyhow::anyhow!(
                    "endpoint {} did not publish a dialable address within {}ms",
                    endpoint.id(),
                    timeout_ms
                ));
            }

            sleep(Duration::from_millis(50)).await;
        }
    }

    #[cfg(feature = "transport-lan")]
    #[tokio::test]
    async fn native_iroh_loopback_connect_establishes_direct_lan_path() {
        use crate::native_node::IncomingStreamType;

        let client_a = Client::new_with_app_tag(
            crate::test_constants::TEST_PROJECT_ID.to_string(),
            "lan-e2e-a".to_string(),
            Box::new(|| None),
        );
        let client_b = Client::new_with_app_tag(
            crate::test_constants::TEST_PROJECT_ID.to_string(),
            "lan-e2e-b".to_string(),
            Box::new(|| None),
        );

        let endpoint_a = Endpoint::builder(iroh::endpoint::presets::N0)
            .alpns(vec![crate::native_node::PlutoniumProtocol::ALPN.to_vec()])
            .relay_mode(RelayMode::Disabled)
            .bind_addr(
                "127.0.0.1:0"
                    .parse::<std::net::SocketAddr>()
                    .expect("socket addr a"),
            )
            .expect("bind addr a")
            .bind()
            .await
            .expect("bind endpoint a");
        let endpoint_b = Endpoint::builder(iroh::endpoint::presets::N0)
            .alpns(vec![crate::native_node::PlutoniumProtocol::ALPN.to_vec()])
            .relay_mode(RelayMode::Disabled)
            .bind_addr(
                "127.0.0.1:0"
                    .parse::<std::net::SocketAddr>()
                    .expect("socket addr b"),
            )
            .expect("bind addr b")
            .bind()
            .await
            .expect("bind endpoint b");

        client_a.adopt_endpoint(endpoint_a).await;
        client_b.adopt_endpoint(endpoint_b).await;

        let endpoint_b = client_b.get_endpoint().await.expect("endpoint b");
        let endpoint_b_id = endpoint_b.id();
        let endpoint_b_addr = wait_for_endpoint_addr(&endpoint_b, 5_000)
            .await
            .expect("endpoint b addr");
        let incoming = client_b.incoming_streams().await.expect("incoming streams");

        client_a
            .ensure_connected_addr(endpoint_b_id, endpoint_b_addr)
            .await
            .expect("connect over loopback LAN");

        tokio::time::timeout(Duration::from_secs(5), async {
            loop {
                let kind = client_a.iroh_path_kind(&endpoint_b_id.to_string()).await;
                if kind == crate::client::IrohPathKind::DirectLan {
                    break;
                }
                sleep(Duration::from_millis(50)).await;
            }
        })
        .await
        .expect("iroh selected path should become direct LAN");

        let transport_rtt_ms = client_a
            .iroh_transport_rtt_ms(&endpoint_b_id.to_string())
            .await
            .expect("direct LAN path should expose iroh transport RTT");
        assert!(
            (1..1_000).contains(&transport_rtt_ms),
            "unexpected direct LAN transport RTT: {transport_rtt_ms}ms"
        );

        let connection = client_a
            .get_connection(endpoint_b_id)
            .await
            .expect("active connection");
        let (mut send, _recv) = connection.open_bi().await.expect("open bi stream");
        send.write_all(b"lan-e2e").await.expect("write payload");
        send.finish().expect("finish stream");

        let incoming_stream = tokio::time::timeout(Duration::from_secs(5), incoming.recv())
            .await
            .expect("incoming stream timeout")
            .expect("incoming stream");
        assert!(matches!(
            incoming_stream.stream,
            IncomingStreamType::Bi(_, _)
        ));
    }

    #[test]
    fn auto_connect_failure_backoff_schedule_is_bounded() {
        assert_eq!(auto_connect_failure_backoff_ms(0), 0);
        assert_eq!(auto_connect_failure_backoff_ms(1), 2_000);
        assert_eq!(auto_connect_failure_backoff_ms(2), 4_000);
        assert_eq!(auto_connect_failure_backoff_ms(3), 8_000);
        assert_eq!(auto_connect_failure_backoff_ms(4), 16_000);
        assert_eq!(auto_connect_failure_backoff_ms(5), 30_000);
        assert_eq!(auto_connect_failure_backoff_ms(20), 30_000);
    }

    #[test]
    fn auto_connect_failure_state_registers_and_clears() {
        let mut failure_count: HashMap<String, u8> = HashMap::new();
        let mut backoff_until: HashMap<String, i64> = HashMap::new();
        let remote = "remote-device-1";
        let now = 1_000_i64;

        register_auto_connect_failure(remote, now, &mut failure_count, &mut backoff_until);
        assert_eq!(failure_count.get(remote).copied(), Some(1));
        assert_eq!(backoff_until.get(remote).copied(), Some(now + 2_000));

        register_auto_connect_failure(remote, now + 10, &mut failure_count, &mut backoff_until);
        assert_eq!(failure_count.get(remote).copied(), Some(2));
        assert_eq!(backoff_until.get(remote).copied(), Some(now + 10 + 4_000));

        clear_auto_connect_failure_state(remote, &mut failure_count, &mut backoff_until);
        assert!(!failure_count.contains_key(remote));
        assert!(!backoff_until.contains_key(remote));
    }

    #[test]
    fn active_uni_probe_miss_preserves_live_transport() {
        assert_eq!(
            active_connection_probe_decision(true, true),
            ActiveConnectionProbeDecision::Healthy
        );
        assert_eq!(
            active_connection_probe_decision(true, false),
            ActiveConnectionProbeDecision::Healthy,
            "a successful active probe is authoritative positive liveness evidence"
        );
        assert_eq!(
            active_connection_probe_decision(false, true),
            ActiveConnectionProbeDecision::PreserveLiveTransport,
            "active uni-stream probes are diagnostic; a slow probe must not replace a live transport"
        );
        assert_eq!(
            active_connection_probe_decision(false, false),
            ActiveConnectionProbeDecision::Failed
        );
    }

    #[test]
    fn connected_transport_only_short_circuits_when_session_admission_is_not_blocked() {
        assert!(
            !should_represent_session_token_on_connected_transport(false, false),
            "bare tickets should still short-circuit on a live transport"
        );
        assert!(
            !should_represent_session_token_on_connected_transport(true, false),
            "admitted tokenized transports should not re-present the token"
        );
        assert!(
            should_represent_session_token_on_connected_transport(true, true),
            "pending/rejected tokenized transports must re-present the token"
        );
    }

    #[tokio::test]
    async fn network_change_recovery_is_suppressed_when_another_peer_is_connected() {
        let client = Client::new_with_app_tag(
            crate::test_constants::TEST_PROJECT_ID.to_string(),
            "test-app".to_string(),
            Box::new(|| None),
        );

        client
            .connection_manager
            .upsert_pending(
                "conn-healthy".to_string(),
                Some("node-healthy".to_string()),
                Some("device-healthy".to_string()),
                Some("node-healthy".to_string()),
            )
            .await;
        client
            .connection_manager
            .set_connected_with_transport(
                "conn-healthy",
                Some("node-healthy".to_string()),
                None,
                Some("health-check".to_string()),
            )
            .await
            .expect("healthy record");
        client
            .connection_manager
            .set_device_id("conn-healthy", "device-healthy".to_string())
            .await
            .expect("healthy device binding");

        client
            .connection_manager
            .upsert_pending(
                "conn-failing".to_string(),
                Some("node-failing".to_string()),
                Some("device-failing".to_string()),
                Some("node-failing".to_string()),
            )
            .await;
        client
            .connection_manager
            .set_failed("conn-failing", Some("timed out connecting".to_string()))
            .await;

        assert!(
            !client
                .should_force_network_change_after_connect_failures("device-failing")
                .await
        );
    }

    #[tokio::test]
    async fn network_change_recovery_remains_enabled_without_other_connected_peers() {
        let client = Client::new_with_app_tag(
            crate::test_constants::TEST_PROJECT_ID.to_string(),
            "test-app".to_string(),
            Box::new(|| None),
        );

        client
            .connection_manager
            .upsert_pending(
                "conn-failing".to_string(),
                Some("node-failing".to_string()),
                Some("device-failing".to_string()),
                Some("node-failing".to_string()),
            )
            .await;
        client
            .connection_manager
            .set_failed("conn-failing", Some("timed out connecting".to_string()))
            .await;

        assert!(
            client
                .should_force_network_change_after_connect_failures("device-failing")
                .await
        );
    }

    #[cfg(all(feature = "transport-webrtc", not(target_arch = "wasm32")))]
    #[test]
    fn native_webrtc_attempt_budget_latches_after_exhaustion() {
        assert_eq!(
            Client::reserve_native_webrtc_attempt_slot(0),
            NativeWebRTCAttemptReservation::Start(1)
        );
        assert_eq!(
            Client::reserve_native_webrtc_attempt_slot(Client::NATIVE_WEBRTC_MAX_UPGRADE_ATTEMPTS),
            NativeWebRTCAttemptReservation::JustExhausted(
                Client::NATIVE_WEBRTC_MAX_UPGRADE_ATTEMPTS + 1
            )
        );
        assert_eq!(
            Client::reserve_native_webrtc_attempt_slot(
                Client::NATIVE_WEBRTC_MAX_UPGRADE_ATTEMPTS + 1
            ),
            NativeWebRTCAttemptReservation::AlreadyExhausted
        );
    }

    #[cfg(target_arch = "wasm32")]
    #[test]
    fn wasm_stale_snapshot_retirement_uses_short_replacement_grace() {
        assert!(should_delay_wasm_stale_settled_snapshot_retirement(
            9_000, 13_000
        ));
        assert!(!should_delay_wasm_stale_settled_snapshot_retirement(
            9_000, 13_500
        ));
    }

    #[cfg(all(feature = "transport-webrtc", not(target_arch = "wasm32")))]
    #[tokio::test]
    async fn native_webrtc_start_gate_serializes_concurrent_callers() {
        let client = Client::new_with_app_tag(
            crate::test_constants::TEST_PROJECT_ID.to_string(),
            "test-app".to_string(),
            Box::new(|| None),
        );

        let first_gate = client
            .acquire_native_webrtc_start_gate("conn-1", "node-1", false, None)
            .await;

        let waiter_client = client.clone();
        let waiter = tokio::spawn(async move {
            let gate = waiter_client
                .acquire_native_webrtc_start_gate("conn-1", "node-1", false, None)
                .await;
            waiter_client
                .release_native_webrtc_start_gate("conn-1", gate)
                .await;
        });

        sleep(Duration::from_millis(100)).await;
        assert!(
            !waiter.is_finished(),
            "second caller should wait while the first start gate is held"
        );

        client
            .release_native_webrtc_start_gate("conn-1", first_gate)
            .await;

        tokio::time::timeout(Duration::from_secs(1), waiter)
            .await
            .expect("waiting caller should acquire the gate after release")
            .expect("waiting task should finish cleanly");
    }

    #[test]
    fn duplicate_kept_existing_close_reason_detection_is_specific() {
        // Legacy format (no stable_id suffix)
        assert!(is_duplicate_kept_existing_close_reason(Some(
            "ApplicationClosed(0, b\"duplicate-kept-existing\")"
        )));
        // New format with stable_id suffix
        assert!(is_duplicate_kept_existing_close_reason(Some(
            "ApplicationClosed(0, b\"duplicate-kept-existing:12345\")"
        )));
        // Raw new format (before iroh wraps it)
        assert!(is_duplicate_kept_existing_close_reason(Some(
            "duplicate-kept-existing:99"
        )));
        assert!(!is_duplicate_kept_existing_close_reason(Some(
            "ApplicationClosed(0, b\"disconnected by user\")"
        )));
        assert!(!is_duplicate_kept_existing_close_reason(None));
    }

    #[test]
    fn replacement_churn_close_reason_detection_is_specific() {
        assert!(is_replacement_churn_close_reason(Some(
            "ApplicationClosed(0, b\"duplicate-kept-existing\")"
        )));
        assert!(is_replacement_churn_close_reason(Some(
            "ApplicationClosed(ApplicationClose { error_code: 0, reason: b\"duplicate-accept-superseded\" })"
        )));
        assert!(is_replacement_churn_close_reason(Some(
            "ApplicationClosed(ApplicationClose { error_code: 0, reason: b\"duplicate-dial-superseded\" })"
        )));
        assert!(is_replacement_churn_close_reason(Some(
            "ApplicationClosed(ApplicationClose { error_code: 0, reason: b\"replaced-by-new-inbound\" })"
        )));
        // Mirror case: native_node emits "replaced-by-new-outbound" when a fresh dial
        // supersedes an earlier connection. Must be classified as churn so the close
        // handler doesn't retire the kept transport (regression: without this, glare
        // resolution leaves the peer without a live transport even when adoption
        // succeeded).
        assert!(is_replacement_churn_close_reason(Some(
            "ApplicationClosed(ApplicationClose { error_code: 0, reason: b\"replaced-by-new-outbound\" })"
        )));
        assert!(is_replacement_churn_close_reason(Some(
            "ApplicationClosed(0, b\"zombie-replaced-by-fresh-dial\")"
        )));
        assert!(!is_replacement_churn_close_reason(Some(
            "ApplicationClosed(0, b\"manual disconnect\")"
        )));
        assert!(!is_replacement_churn_close_reason(Some("LocallyClosed")));
        assert!(!is_replacement_churn_close_reason(None));
    }

    #[test]
    fn graceful_disconnect_close_reason_detection_is_specific() {
        assert!(is_graceful_disconnect_close_reason(Some(
            "ApplicationClosed(0, b\"closed\")"
        )));
        assert!(is_graceful_disconnect_close_reason(Some(
            "ApplicationClosed(1, b\"disconnected by user\")"
        )));
        assert!(is_graceful_disconnect_close_reason(Some(
            "manual disconnect"
        )));
        assert!(!is_graceful_disconnect_close_reason(Some(
            "ApplicationClosed(0, b\"duplicate-kept-existing\")"
        )));
        assert!(!is_graceful_disconnect_close_reason(Some(
            "ApplicationClosed(1, b\"session-token-revoked\")"
        )));
        assert!(!is_graceful_disconnect_close_reason(Some(
            "transport reset by peer"
        )));
        assert!(!is_graceful_disconnect_close_reason(None));
    }

    #[test]
    fn manual_disconnect_close_reason_detection_is_session_scoped() {
        assert!(is_manual_disconnect_close_reason(Some(
            "ApplicationClosed(1, b\"disconnected by user\")"
        )));
        assert!(is_manual_disconnect_close_reason(Some(
            "ApplicationClosed(0, b\"manual disconnect\")"
        )));
        assert!(!is_manual_disconnect_close_reason(Some(
            "ApplicationClosed(0, b\"closed\")"
        )));
        assert!(!is_manual_disconnect_close_reason(Some(
            "ApplicationClosed(1, b\"session-token-revoked\")"
        )));
        assert!(!is_manual_disconnect_close_reason(None));
    }

    #[test]
    fn device_status_snapshots_prefer_connected_peer_by_device_id() {
        let devices = vec![Device {
            app_tag: Some("app".to_string()),
            device_id: "device-1".to_string(),
            user_id: Some("user-1".to_string()),
            device_name: "Device 1".to_string(),
            platform_type: Some("desktop".to_string()),
            capabilities: None,
            session_id: None,
            node_id: Some("node-1".to_string()),
            tag: Some("app".to_string()),
            kind: Some("device".to_string()),
            metadata: None,
            online: true,
            ticket: Some("ticket".to_string()),
            last_seen_at: None,
            expires_at: None,
            created_at: None,
            updated_at: None,
            excluded_peers: vec![],
        }];

        let peers = vec![
            PeerSnapshot {
                peer_id: "device-1".to_string(),
                device_id: Some("device-1".to_string()),
                device_id_hint: None,
                node_id: Some("node-1".to_string()),
                connection_ids: vec!["conn-stale".to_string()],
                active_transport_stable_id: Some(1),
                active_transport_generation: 1,
                active_transport: "iroh".to_string(),
                parallel_transport: None,
                last_lifecycle_transition_at_ms: 1,
                status: ConnectionState::Connecting,
                health: ConnectionHealth::Suspect,
                transition_count: 1,
                connecting_transition_count: 1,
                replacement_count: 0,
                retire_count: 0,
                last_disconnect_reason: None,
                last_reconnect_reason: Some("initial".to_string()),
                scopes: vec!["session".to_string()],
                last_seen_at_ms: 1,
                error: None,
            },
            PeerSnapshot {
                peer_id: "device-1".to_string(),
                device_id: Some("device-1".to_string()),
                device_id_hint: Some("old-device-1".to_string()),
                node_id: Some("node-1".to_string()),
                connection_ids: vec!["conn-live".to_string()],
                active_transport_stable_id: Some(2),
                active_transport_generation: 2,
                active_transport: "webrtc".to_string(),
                parallel_transport: Some("iroh".to_string()),
                last_lifecycle_transition_at_ms: 2,
                status: ConnectionState::Connected,
                health: ConnectionHealth::Healthy,
                transition_count: 2,
                connecting_transition_count: 1,
                replacement_count: 1,
                retire_count: 0,
                last_disconnect_reason: None,
                last_reconnect_reason: Some("replacement".to_string()),
                scopes: vec!["persistent".to_string()],
                last_seen_at_ms: 2,
                error: None,
            },
        ];

        let snapshots = merge_device_status_snapshots(devices, peers);
        let snapshot = snapshots.first().expect("snapshot");
        assert_eq!(
            snapshot.connection_status,
            DeviceConnectionStatus::Connected
        );
        assert_eq!(snapshot.peer_health, ConnectionHealth::Healthy);
        assert_eq!(snapshot.connection_id.as_deref(), Some("conn-live"));
        assert_eq!(snapshot.device_id_hint.as_deref(), Some("old-device-1"));
    }

    #[test]
    fn device_status_snapshots_prefer_routable_peer_when_aliases_overlap() {
        let devices = vec![Device {
            app_tag: Some("app".to_string()),
            device_id: "device-routable".to_string(),
            user_id: Some("user-1".to_string()),
            device_name: "Device Routable".to_string(),
            platform_type: Some("web".to_string()),
            capabilities: None,
            session_id: None,
            node_id: Some("node-routable".to_string()),
            tag: Some("app".to_string()),
            kind: Some("device".to_string()),
            metadata: None,
            online: true,
            ticket: Some("ticket".to_string()),
            last_seen_at: None,
            expires_at: None,
            created_at: None,
            updated_at: None,
            excluded_peers: vec![],
        }];

        let peers = vec![
            PeerSnapshot {
                peer_id: "device-routable".to_string(),
                device_id: Some("device-routable".to_string()),
                device_id_hint: None,
                node_id: Some("node-routable".to_string()),
                connection_ids: vec!["conn-stale-connected".to_string()],
                active_transport_stable_id: Some(7),
                active_transport_generation: 1,
                active_transport: "iroh".to_string(),
                parallel_transport: None,
                last_lifecycle_transition_at_ms: 10,
                status: ConnectionState::Connected,
                health: ConnectionHealth::Stale,
                transition_count: 1,
                connecting_transition_count: 1,
                replacement_count: 0,
                retire_count: 0,
                last_disconnect_reason: None,
                last_reconnect_reason: Some("health-check".to_string()),
                scopes: vec![],
                last_seen_at_ms: 10,
                error: None,
            },
            PeerSnapshot {
                peer_id: "device-routable".to_string(),
                device_id: Some("device-routable".to_string()),
                device_id_hint: None,
                node_id: Some("node-routable".to_string()),
                connection_ids: vec!["conn-routable".to_string()],
                active_transport_stable_id: Some(8),
                active_transport_generation: 2,
                active_transport: "webrtc".to_string(),
                parallel_transport: Some("iroh".to_string()),
                last_lifecycle_transition_at_ms: 20,
                status: ConnectionState::Connected,
                health: ConnectionHealth::Healthy,
                transition_count: 2,
                connecting_transition_count: 1,
                replacement_count: 1,
                retire_count: 0,
                last_disconnect_reason: None,
                last_reconnect_reason: Some("replacement".to_string()),
                scopes: vec!["persistent".to_string()],
                last_seen_at_ms: 20,
                error: None,
            },
        ];

        let snapshots = merge_device_status_snapshots(devices, peers);
        let snapshot = snapshots.first().expect("snapshot");
        assert_eq!(
            snapshot.connection_status,
            DeviceConnectionStatus::Connected
        );
        assert_eq!(snapshot.settled_ready, true);
        assert_eq!(snapshot.peer_health, ConnectionHealth::Healthy);
        assert_eq!(snapshot.connection_id.as_deref(), Some("conn-routable"));
        assert_eq!(snapshot.active_transport_stable_id, Some(8));
        assert_eq!(snapshot.active_transport, "webrtc");
        assert_eq!(snapshot.parallel_transport.as_deref(), Some("iroh"));
    }

    /// Data-plane truth: a Connected snapshot with a live transport (stable transport id +
    /// tracked connection ids) beats a Failed snapshot for the same device regardless of
    /// recency. This is the "winner-takes-all" outcome of glare resolution — when one
    /// direction's dial fails but the other direction has an admitted transport, the UI
    /// should reflect the working path, not the older failed attempt.
    ///
    /// The complementary ghost-protection case (Connected without a live transport vs a
    /// newer Failed dial) is covered by [`ghost_connected_without_transport_yields_to_newer_failed`].
    #[test]
    fn device_status_merge_prefers_connected_with_live_transport_over_newer_failed() {
        let devices = vec![Device {
            app_tag: Some("app".to_string()),
            device_id: "device-dup-fail".to_string(),
            user_id: Some("user-1".to_string()),
            device_name: "Dup fail".to_string(),
            platform_type: Some("web".to_string()),
            capabilities: None,
            session_id: None,
            node_id: Some("node-dup-fail".to_string()),
            tag: Some("app".to_string()),
            kind: Some("device".to_string()),
            metadata: None,
            online: true,
            ticket: Some("ticket".to_string()),
            last_seen_at: None,
            expires_at: None,
            created_at: None,
            updated_at: None,
            excluded_peers: vec![],
        }];

        let connected_with_transport = PeerSnapshot {
            peer_id: "peer-by-connection-id-live".to_string(),
            device_id: Some("device-dup-fail".to_string()),
            device_id_hint: None,
            node_id: Some("node-dup-fail".to_string()),
            connection_ids: vec!["live-conn".to_string()],
            active_transport_stable_id: Some(1),
            active_transport_generation: 1,
            active_transport: "iroh".to_string(),
            parallel_transport: None,
            last_lifecycle_transition_at_ms: 1000,
            status: ConnectionState::Connected,
            health: ConnectionHealth::Healthy,
            transition_count: 1,
            connecting_transition_count: 0,
            replacement_count: 0,
            retire_count: 0,
            last_disconnect_reason: None,
            last_reconnect_reason: None,
            scopes: vec![],
            last_seen_at_ms: 1000,
            error: None,
        };

        let failed_newer = PeerSnapshot {
            peer_id: "device-dup-fail".to_string(),
            device_id: Some("device-dup-fail".to_string()),
            device_id_hint: None,
            node_id: Some("node-dup-fail".to_string()),
            connection_ids: vec!["dial-conn-failed".to_string()],
            active_transport_stable_id: None,
            active_transport_generation: 0,
            active_transport: "iroh".to_string(),
            parallel_transport: None,
            last_lifecycle_transition_at_ms: 2000,
            status: ConnectionState::Failed,
            health: ConnectionHealth::Stale,
            transition_count: 1,
            connecting_transition_count: 1,
            replacement_count: 0,
            retire_count: 0,
            last_disconnect_reason: None,
            last_reconnect_reason: None,
            scopes: vec![],
            last_seen_at_ms: 2000,
            error: Some("timeout".to_string()),
        };

        for peers in [
            vec![connected_with_transport.clone(), failed_newer.clone()],
            vec![failed_newer.clone(), connected_with_transport.clone()],
        ] {
            let snapshots = merge_device_status_snapshots(devices.clone(), peers);
            let snapshot = snapshots.first().expect("snapshot");
            assert_eq!(
                snapshot.connection_status,
                DeviceConnectionStatus::Connected,
                "Connected snapshot with live transport must beat newer Failed (merge order invariance)",
            );
            assert!(
                snapshot.settled_ready,
                "Live-transport Connected snapshot must report settled_ready",
            );
        }
    }

    /// Ghost-protection: a Connected snapshot without a live transport (e.g. a backfill
    /// row whose transport never materialised) must yield to a newer Failed dial for the
    /// same device. This is the case the old `merge_prefers_newer_failed_over_older_routable_connected`
    /// test was guarding; with data-plane truth we narrow it to the "no live transport"
    /// branch only.
    #[test]
    fn ghost_connected_without_transport_yields_to_newer_failed() {
        let devices = vec![Device {
            app_tag: Some("app".to_string()),
            device_id: "device-ghost".to_string(),
            user_id: Some("user-1".to_string()),
            device_name: "Ghost".to_string(),
            platform_type: Some("web".to_string()),
            capabilities: None,
            session_id: None,
            node_id: Some("node-ghost".to_string()),
            tag: Some("app".to_string()),
            kind: Some("device".to_string()),
            metadata: None,
            online: true,
            ticket: Some("ticket".to_string()),
            last_seen_at: None,
            expires_at: None,
            created_at: None,
            updated_at: None,
            excluded_peers: vec![],
        }];

        let connected_ghost = PeerSnapshot {
            peer_id: "peer-by-connection-id-ghost".to_string(),
            device_id: Some("device-ghost".to_string()),
            device_id_hint: None,
            node_id: Some("node-ghost".to_string()),
            connection_ids: vec!["ghost-conn".to_string()],
            // No active transport — ghost row.
            active_transport_stable_id: None,
            active_transport_generation: 0,
            active_transport: "iroh".to_string(),
            parallel_transport: None,
            last_lifecycle_transition_at_ms: 1000,
            status: ConnectionState::Connected,
            health: ConnectionHealth::Healthy,
            transition_count: 1,
            connecting_transition_count: 0,
            replacement_count: 0,
            retire_count: 0,
            last_disconnect_reason: None,
            last_reconnect_reason: None,
            scopes: vec![],
            last_seen_at_ms: 1000,
            error: None,
        };

        let failed_newer = PeerSnapshot {
            peer_id: "device-ghost".to_string(),
            device_id: Some("device-ghost".to_string()),
            device_id_hint: None,
            node_id: Some("node-ghost".to_string()),
            connection_ids: vec!["dial-conn-failed".to_string()],
            active_transport_stable_id: None,
            active_transport_generation: 0,
            active_transport: "iroh".to_string(),
            parallel_transport: None,
            last_lifecycle_transition_at_ms: 2000,
            status: ConnectionState::Failed,
            health: ConnectionHealth::Stale,
            transition_count: 1,
            connecting_transition_count: 1,
            replacement_count: 0,
            retire_count: 0,
            last_disconnect_reason: None,
            last_reconnect_reason: None,
            scopes: vec![],
            last_seen_at_ms: 2000,
            error: Some("timeout".to_string()),
        };

        for peers in [
            vec![connected_ghost.clone(), failed_newer.clone()],
            vec![failed_newer.clone(), connected_ghost.clone()],
        ] {
            let snapshots = merge_device_status_snapshots(devices.clone(), peers);
            let snapshot = snapshots.first().expect("snapshot");
            assert_eq!(
                snapshot.connection_status,
                DeviceConnectionStatus::Failed,
                "Connected snapshot without live transport must yield to newer Failed",
            );
        }
    }

    #[test]
    fn device_status_snapshots_report_online_without_peer() {
        let devices = vec![Device {
            app_tag: Some("app".to_string()),
            device_id: "device-2".to_string(),
            user_id: Some("user-1".to_string()),
            device_name: "Device 2".to_string(),
            platform_type: Some("web".to_string()),
            capabilities: None,
            session_id: None,
            node_id: Some("node-2".to_string()),
            tag: Some("app".to_string()),
            kind: Some("device".to_string()),
            metadata: None,
            online: true,
            ticket: Some("ticket".to_string()),
            last_seen_at: None,
            expires_at: None,
            created_at: None,
            updated_at: None,
            excluded_peers: vec![],
        }];

        let snapshots = merge_device_status_snapshots(devices, Vec::new());
        let snapshot = snapshots.first().expect("snapshot");
        assert_eq!(snapshot.connection_status, DeviceConnectionStatus::Online);
        assert_eq!(snapshot.presence_status, DevicePresenceStatus::Online);
        assert!(snapshot.connectable);
        assert_eq!(snapshot.peer_health, ConnectionHealth::Healthy);
        assert!(snapshot.connection_id.is_none());
    }

    #[test]
    fn device_status_snapshots_report_stale_online_roster_entries_as_idle() {
        let now = crate::firebase::now_millis_u64().min(i64::MAX as u64) as i64;
        let devices = vec![Device {
            app_tag: Some("app".to_string()),
            device_id: "device-idle".to_string(),
            user_id: Some("user-1".to_string()),
            device_name: "Idle Device".to_string(),
            platform_type: Some("desktop".to_string()),
            capabilities: None,
            session_id: None,
            node_id: Some("node-idle".to_string()),
            tag: Some("app".to_string()),
            kind: Some("device".to_string()),
            metadata: None,
            online: true,
            ticket: Some("ticket".to_string()),
            last_seen_at: None,
            expires_at: None,
            created_at: None,
            updated_at: Some(serde_json::Value::from(now.saturating_sub(
                crate::firebase::device_registry::DEVICE_STALE_HEARTBEAT_MS + 1,
            ))),
            excluded_peers: vec![],
        }];

        let snapshots = merge_device_status_snapshots(devices, Vec::new());
        let snapshot = snapshots.first().expect("snapshot");
        assert_eq!(snapshot.connection_status, DeviceConnectionStatus::Online);
        assert_eq!(snapshot.presence_status, DevicePresenceStatus::Idle);
        assert!(!snapshot.connectable);
        assert!(snapshot.presence_updated_at.is_some());
    }

    #[test]
    fn device_status_snapshots_preserve_offline_device_without_peer() {
        let devices = vec![Device {
            app_tag: Some("app".to_string()),
            device_id: "device-offline".to_string(),
            user_id: Some("user-1".to_string()),
            device_name: "Offline Device".to_string(),
            platform_type: Some("desktop".to_string()),
            capabilities: None,
            session_id: None,
            node_id: Some("node-offline".to_string()),
            tag: Some("app".to_string()),
            kind: Some("device".to_string()),
            metadata: None,
            online: false,
            ticket: None,
            last_seen_at: None,
            expires_at: None,
            created_at: None,
            updated_at: None,
            excluded_peers: vec![],
        }];

        let snapshots = merge_device_status_snapshots(devices, Vec::new());
        let snapshot = snapshots.first().expect("snapshot");
        assert_eq!(
            snapshot.connection_status,
            DeviceConnectionStatus::Disconnected
        );
        assert_eq!(snapshot.presence_status, DevicePresenceStatus::Offline);
        assert!(!snapshot.connectable);
        assert_eq!(snapshot.readiness_reason, "device-offline");
        assert!(snapshot.connection_id.is_none());
    }

    #[tokio::test]
    async fn device_status_snapshots_report_connected_peer_with_hint_and_scope() {
        let client = Client::new_with_app_tag(
            crate::test_constants::TEST_PROJECT_ID.to_string(),
            "test-app".to_string(),
            Box::new(|| None),
        );

        client
            .connection_manager
            .upsert_pending(
                "conn-1".to_string(),
                Some("node-1".to_string()),
                Some("device-old-1".to_string()),
                Some("node-1".to_string()),
            )
            .await;
        client.connection_manager.set_connecting("conn-1").await;
        client
            .connection_manager
            .set_device_id("conn-1", "device-1".to_string())
            .await;
        client
            .connection_manager
            .set_connected("conn-1", Some("node-1".to_string()))
            .await;
        client
            .connection_manager
            .set_health("conn-1", ConnectionHealth::Healthy)
            .await;
        client
            .connection_manager
            .add_scope("conn-1", "persistent")
            .await;

        let devices = vec![Device {
            app_tag: Some("test-app".to_string()),
            device_id: "device-1".to_string(),
            user_id: Some("user-1".to_string()),
            device_name: "Device 1".to_string(),
            platform_type: Some("desktop".to_string()),
            capabilities: None,
            session_id: None,
            node_id: Some("node-1".to_string()),
            tag: Some("test-app".to_string()),
            kind: Some("device".to_string()),
            metadata: None,
            online: true,
            ticket: Some("ticket".to_string()),
            last_seen_at: None,
            expires_at: None,
            created_at: None,
            updated_at: None,
            excluded_peers: vec![],
        }];

        let snapshots = merge_device_status_snapshots(
            devices,
            client.connection_manager.list_peer_snapshots().await,
        );
        let snapshot = snapshots.first().expect("snapshot");
        assert_eq!(
            snapshot.connection_status,
            DeviceConnectionStatus::Connected
        );
        assert_eq!(snapshot.peer_health, ConnectionHealth::Healthy);
        assert_eq!(snapshot.connection_id.as_deref(), Some("conn-1"));
        assert_eq!(snapshot.device_id_hint.as_deref(), Some("device-old-1"));
        assert_eq!(snapshot.scopes, vec!["persistent".to_string()]);
    }

    // ──────────────────────────────────────────────────────────────────────────
    // Auto-connect exclusion set tests
    // ──────────────────────────────────────────────────────────────────────────

    #[test]
    fn auto_connect_excluded_set_and_cleared() {
        let client = Client::new_with_app_tag(
            crate::test_constants::TEST_PROJECT_ID.to_string(),
            "test-app".to_string(),
            Box::new(|| None),
        );

        assert!(!client.is_auto_connect_excluded("device-1"));

        client.set_auto_connect_excluded("device-1", true);
        assert!(client.is_auto_connect_excluded("device-1"));

        client.set_auto_connect_excluded("device-1", false);
        assert!(!client.is_auto_connect_excluded("device-1"));
    }

    #[test]
    fn auto_connect_excluded_is_case_insensitive() {
        let client = Client::new_with_app_tag(
            crate::test_constants::TEST_PROJECT_ID.to_string(),
            "test-app".to_string(),
            Box::new(|| None),
        );

        client.set_auto_connect_excluded("Device-ABC", true);
        assert!(client.is_auto_connect_excluded("device-abc"));
        assert!(client.is_auto_connect_excluded("DEVICE-ABC"));
        assert!(client.is_auto_connect_excluded("Device-ABC"));
    }

    #[test]
    fn auto_connect_excluded_does_not_affect_other_devices() {
        let client = Client::new_with_app_tag(
            crate::test_constants::TEST_PROJECT_ID.to_string(),
            "test-app".to_string(),
            Box::new(|| None),
        );

        client.set_auto_connect_excluded("device-1", true);
        assert!(client.is_auto_connect_excluded("device-1"));
        assert!(!client.is_auto_connect_excluded("device-2"));
        assert!(!client.is_auto_connect_excluded("device-3"));
    }

    #[test]
    fn auto_connect_excluded_empty_device_id_is_ignored() {
        let client = Client::new_with_app_tag(
            crate::test_constants::TEST_PROJECT_ID.to_string(),
            "test-app".to_string(),
            Box::new(|| None),
        );

        client.set_auto_connect_excluded("", true);
        client.set_auto_connect_excluded("   ", true);
        // Neither should have been stored; the empty check should not crash.
        assert!(!client.is_auto_connect_excluded(""));
        assert!(!client.is_auto_connect_excluded("   "));
    }

    #[test]
    fn auto_connect_excluded_multiple_devices_independent() {
        let client = Client::new_with_app_tag(
            crate::test_constants::TEST_PROJECT_ID.to_string(),
            "test-app".to_string(),
            Box::new(|| None),
        );

        client.set_auto_connect_excluded("device-A", true);
        client.set_auto_connect_excluded("device-B", true);
        assert!(client.is_auto_connect_excluded("device-a")); // normalised
        assert!(client.is_auto_connect_excluded("device-b"));

        client.set_auto_connect_excluded("device-A", false);
        assert!(!client.is_auto_connect_excluded("device-a"));
        assert!(client.is_auto_connect_excluded("device-b")); // B still excluded
    }

    #[test]
    fn auto_connect_excluded_peer_tracks_node_alias_without_publishing_alias() {
        let client = Client::new_with_app_tag(
            crate::test_constants::TEST_PROJECT_ID.to_string(),
            "test-app".to_string(),
            Box::new(|| None),
        );

        client.set_auto_connect_excluded_peer("Device-A", Some("Node-XYZ"), true);

        assert!(client.is_auto_connect_excluded("device-a"));
        assert!(client.is_auto_connect_peer_excluded("device-a", Some("node-xyz")));
        assert!(client.is_auto_connect_peer_excluded("other-device", Some("node-xyz")));
        assert!(!client.is_auto_connect_excluded("node-xyz"));
        assert_eq!(client.current_excluded_peers_snapshot(), vec!["device-a"]);

        client.set_auto_connect_excluded("Device-A", false);

        assert!(!client.is_auto_connect_excluded("device-a"));
        assert!(!client.is_auto_connect_peer_excluded("device-a", Some("node-xyz")));
        assert!(!client.is_auto_connect_peer_excluded("other-device", Some("node-xyz")));
        assert!(client.current_excluded_peers_snapshot().is_empty());
    }

    /// Helper: build a minimal Device record with a valid ticket from a live endpoint addr.
    fn make_test_device(
        device_id: &str,
        node_id: &str,
        ticket_str: &str,
    ) -> crate::signaling::Device {
        make_test_device_with_excluded(device_id, node_id, ticket_str, vec![])
    }

    /// Like `make_test_device` but also sets the `excluded_peers` list.
    fn make_test_device_with_excluded(
        device_id: &str,
        node_id: &str,
        ticket_str: &str,
        excluded_peers: Vec<String>,
    ) -> crate::signaling::Device {
        crate::signaling::Device {
            app_tag: None,
            device_id: device_id.to_string(),
            user_id: None,
            device_name: "Test Device".to_string(),
            platform_type: None,
            capabilities: None,
            session_id: None,
            node_id: Some(node_id.to_string()),
            tag: None,
            kind: None,
            metadata: None,
            online: true,
            ticket: Some(ticket_str.to_string()),
            last_seen_at: None,
            expires_at: None,
            created_at: None,
            updated_at: None,
            excluded_peers,
        }
    }

    fn make_loop_state() -> (
        std::collections::HashMap<String, i64>,
        std::collections::HashMap<String, u8>,
        std::collections::HashMap<String, i64>,
        std::collections::HashMap<String, i64>,
        std::collections::HashMap<String, u8>,
        std::collections::HashMap<String, i64>,
        std::collections::HashMap<String, i64>,
        std::collections::HashMap<String, bool>,
        std::collections::HashMap<String, u8>,
        i64,
        i64,
        std::collections::HashMap<String, String>,
    ) {
        (
            std::collections::HashMap::new(), // last_attempt_at
            std::collections::HashMap::new(), // failure_count
            std::collections::HashMap::new(), // failure_backoff_until
            std::collections::HashMap::new(), // non_initiator_wait_started_at
            std::collections::HashMap::new(), // non_initiator_escalation_count
            std::collections::HashMap::new(), // skip_log_at
            std::collections::HashMap::new(), // connected_health_checked_at
            std::collections::HashMap::new(), // connected_health_state
            std::collections::HashMap::new(), // connected_health_failures
            0i64,                             // last_presence_republish_at_ms
            0i64,                             // last_network_change_at_ms
            std::collections::HashMap::new(), // last_known_node_id
        )
    }

    /// Excluded device is skipped by try_auto_connect_device before any attempt
    /// is made, regardless of whether the local node is the "initiator"
    /// (local_node_id > remote_node_id) or not.
    ///
    /// The exclusion check fires before the ticket is parsed, so no iroh node
    /// needs to be running for this test.
    #[tokio::test]
    async fn auto_connect_loop_skips_excluded_device_as_initiator() {
        // Build two real iroh endpoints to get valid node IDs and a ticket.
        let ep_local = Endpoint::builder(iroh::endpoint::presets::N0)
            .relay_mode(RelayMode::Disabled)
            .bind()
            .await
            .expect("local endpoint");
        let ep_remote = Endpoint::builder(iroh::endpoint::presets::N0)
            .relay_mode(RelayMode::Disabled)
            .bind()
            .await
            .expect("remote endpoint");

        let remote_addr = wait_for_endpoint_addr(&ep_remote, 3_000)
            .await
            .expect("remote addr");

        // Ensure local is the "initiator" (greater node_id lexicographically).
        let id_a = ep_local.id().to_string();
        let id_b = ep_remote.id().to_string();
        let (local_node_id_str, remote_node_id_str) = if id_a > id_b {
            (id_a, id_b)
        } else {
            (id_b, id_a)
        };

        let ticket = iroh_tickets::endpoint::EndpointTicket::new(remote_addr);
        let ticket_str = ticket.to_string();
        let remote_device_id = "53874465-3225-407b-8012-21292c3835b8";

        let client = std::sync::Arc::new(Client::new_with_app_tag(
            crate::test_constants::TEST_PROJECT_ID.to_string(),
            "test-app".to_string(),
            Box::new(|| None),
        ));
        // Inject node_id so the loop can proceed past the "local-node-not-ready" guard.
        *client.node_id.write().await = Some(local_node_id_str.clone());

        let device = make_test_device(remote_device_id, &remote_node_id_str, &ticket_str);

        // ── Excluded: no attempt recorded ──
        client.set_auto_connect_excluded(remote_device_id, true);

        let (
            mut last_attempt_at,
            mut failure_count,
            mut failure_backoff_until,
            mut non_initiator_wait_started_at,
            mut non_initiator_escalation_count,
            mut skip_log_at,
            mut connected_health_checked_at,
            mut connected_health_state,
            mut connected_health_failures,
            mut last_presence_republish_at_ms,
            mut last_network_change_at_ms,
            mut last_known_node_id,
        ) = make_loop_state();

        client
            .try_auto_connect_device(
                "user-1",
                device.clone(),
                "local-device",
                &mut last_attempt_at,
                &mut failure_count,
                &mut failure_backoff_until,
                500,
                &mut non_initiator_wait_started_at,
                &mut non_initiator_escalation_count,
                30_000,
                &mut skip_log_at,
                &mut connected_health_checked_at,
                &mut connected_health_state,
                &mut connected_health_failures,
                5_000,
                3,
                &mut last_presence_republish_at_ms,
                60_000,
                &mut last_network_change_at_ms,
                30_000,
                2,
                &mut last_known_node_id,
            )
            .await;

        assert!(
            !last_attempt_at.contains_key(remote_device_id),
            "initiator-role excluded device must not appear in last_attempt_at"
        );

        // ── Cleared: exclusion removed ──
        client.set_auto_connect_excluded(remote_device_id, false);
        assert!(!client.is_auto_connect_excluded(remote_device_id));

        ep_local.close().await;
        ep_remote.close().await;
    }

    /// Manual-disconnect suppression is keyed by the stable device id for
    /// signaling, but native reconnect races can still surface through a node-id
    /// keyed record. The local alias must suppress those stale/different device
    /// ids without publishing the node id as an excluded peer.
    #[tokio::test]
    async fn auto_connect_loop_skips_node_alias_excluded_peer_with_different_device_id() {
        let ep_local = Endpoint::builder(iroh::endpoint::presets::N0)
            .relay_mode(RelayMode::Disabled)
            .bind()
            .await
            .expect("local endpoint");
        let ep_remote = Endpoint::builder(iroh::endpoint::presets::N0)
            .relay_mode(RelayMode::Disabled)
            .bind()
            .await
            .expect("remote endpoint");

        let remote_addr = wait_for_endpoint_addr(&ep_remote, 3_000)
            .await
            .expect("remote addr");

        let local_node_id_str = ep_local.id().to_string();
        let remote_node_id_str = ep_remote.id().to_string();
        let ticket = iroh_tickets::endpoint::EndpointTicket::new(remote_addr);
        let ticket_str = ticket.to_string();

        let old_remote_device_id = "manual-disconnected-device";
        let stale_remote_device_id = "stale-presence-device";

        let client = std::sync::Arc::new(Client::new_with_app_tag(
            crate::test_constants::TEST_PROJECT_ID.to_string(),
            "test-app".to_string(),
            Box::new(|| None),
        ));
        *client.node_id.write().await = Some(local_node_id_str.clone());

        client.set_auto_connect_excluded_peer(
            old_remote_device_id,
            Some(&remote_node_id_str),
            true,
        );

        let device = make_test_device(stale_remote_device_id, &remote_node_id_str, &ticket_str);
        let (
            mut last_attempt_at,
            mut failure_count,
            mut failure_backoff_until,
            mut non_initiator_wait_started_at,
            mut non_initiator_escalation_count,
            mut skip_log_at,
            mut connected_health_checked_at,
            mut connected_health_state,
            mut connected_health_failures,
            mut last_presence_republish_at_ms,
            mut last_network_change_at_ms,
            mut last_known_node_id,
        ) = make_loop_state();

        client
            .try_auto_connect_device(
                "user-1",
                device,
                "local-device",
                &mut last_attempt_at,
                &mut failure_count,
                &mut failure_backoff_until,
                500,
                &mut non_initiator_wait_started_at,
                &mut non_initiator_escalation_count,
                0,
                &mut skip_log_at,
                &mut connected_health_checked_at,
                &mut connected_health_state,
                &mut connected_health_failures,
                5_000,
                3,
                &mut last_presence_republish_at_ms,
                60_000,
                &mut last_network_change_at_ms,
                30_000,
                2,
                &mut last_known_node_id,
            )
            .await;

        assert!(
            !last_attempt_at.contains_key(stale_remote_device_id),
            "node-id alias should suppress stale/different device ids for the same peer"
        );
        assert_eq!(
            client.current_excluded_peers_snapshot(),
            vec![old_remote_device_id.to_string()],
            "published excludedPeers must remain device-id scoped and not include node aliases"
        );

        ep_local.close().await;
        ep_remote.close().await;
    }

    /// Non-initiator variant: when local_node_id < remote_node_id the local side
    /// would normally wait the grace period before escalating to dial.
    /// The exclusion must prevent any attempt even with zero grace (immediate escalation).
    #[tokio::test]
    async fn auto_connect_loop_skips_excluded_device_as_non_initiator() {
        let ep_local = Endpoint::builder(iroh::endpoint::presets::N0)
            .relay_mode(RelayMode::Disabled)
            .bind()
            .await
            .expect("local endpoint");
        let ep_remote = Endpoint::builder(iroh::endpoint::presets::N0)
            .relay_mode(RelayMode::Disabled)
            .bind()
            .await
            .expect("remote endpoint");

        let remote_addr = wait_for_endpoint_addr(&ep_remote, 3_000)
            .await
            .expect("remote addr");

        // Force non-initiator role: local_node_id < remote_node_id.
        let id_a = ep_local.id().to_string();
        let id_b = ep_remote.id().to_string();
        let (local_node_id_str, remote_node_id_str) = if id_a < id_b {
            (id_a, id_b)
        } else {
            (id_b, id_a)
        };

        let ticket = iroh_tickets::endpoint::EndpointTicket::new(remote_addr);
        let ticket_str = ticket.to_string();
        let remote_device_id = "non-initiator-device-test";

        let client = std::sync::Arc::new(Client::new_with_app_tag(
            crate::test_constants::TEST_PROJECT_ID.to_string(),
            "test-app".to_string(),
            Box::new(|| None),
        ));
        *client.node_id.write().await = Some(local_node_id_str.clone());

        let device = make_test_device(remote_device_id, &remote_node_id_str, &ticket_str);

        client.set_auto_connect_excluded(remote_device_id, true);

        let (
            mut last_attempt_at,
            mut failure_count,
            mut failure_backoff_until,
            mut non_initiator_wait_started_at,
            mut non_initiator_escalation_count,
            mut skip_log_at,
            mut connected_health_checked_at,
            mut connected_health_state,
            mut connected_health_failures,
            mut last_presence_republish_at_ms,
            mut last_network_change_at_ms,
            mut last_known_node_id,
        ) = make_loop_state();

        // Call twice with zero grace to simulate the escalation cycle a non-initiator
        // would normally go through — the exclusion must block both calls.
        for _ in 0..2 {
            client
                .try_auto_connect_device(
                    "user-1",
                    device.clone(),
                    "local-device",
                    &mut last_attempt_at,
                    &mut failure_count,
                    &mut failure_backoff_until,
                    500,
                    &mut non_initiator_wait_started_at,
                    &mut non_initiator_escalation_count,
                    0, // zero grace so non-initiator escalates immediately
                    &mut skip_log_at,
                    &mut connected_health_checked_at,
                    &mut connected_health_state,
                    &mut connected_health_failures,
                    5_000,
                    3,
                    &mut last_presence_republish_at_ms,
                    60_000,
                    &mut last_network_change_at_ms,
                    30_000,
                    2,
                    &mut last_known_node_id,
                )
                .await;
        }

        assert!(
            !last_attempt_at.contains_key(remote_device_id),
            "non-initiator excluded device must not appear in last_attempt_at"
        );

        client.set_auto_connect_excluded(remote_device_id, false);
        assert!(!client.is_auto_connect_excluded(remote_device_id));

        ep_local.close().await;
        ep_remote.close().await;
    }

    // ── excluded_peers (remote session-side auto-connect veto) ────────────────

    /// When a remote online device says it has manually disconnected from this
    /// local device, auto-connect must not immediately dial it back in.
    #[tokio::test]
    async fn auto_connect_skips_remote_excluded_local_device() {
        let ep_local = Endpoint::builder(iroh::endpoint::presets::N0)
            .relay_mode(RelayMode::Disabled)
            .bind()
            .await
            .expect("local endpoint");
        let ep_remote = Endpoint::builder(iroh::endpoint::presets::N0)
            .relay_mode(RelayMode::Disabled)
            .bind()
            .await
            .expect("remote endpoint");

        let addr_a = wait_for_endpoint_addr(&ep_local, 3_000)
            .await
            .expect("local addr");
        let addr_b = wait_for_endpoint_addr(&ep_remote, 3_000)
            .await
            .expect("remote addr");

        // Force initiator role so the non-initiator grace is not an obstacle.
        let id_a = ep_local.id().to_string();
        let id_b = ep_remote.id().to_string();
        let (local_node_id_str, remote_node_id_str, remote_addr) = if id_a > id_b {
            (id_a, id_b, addr_b)
        } else {
            (id_b, id_a, addr_a)
        };

        let ticket = iroh_tickets::endpoint::EndpointTicket::new(remote_addr);
        let ticket_str = ticket.to_string();

        let remote_device_id = "remote-excluder-device";
        let local_device_id = "my-local-device";

        let client = std::sync::Arc::new(Client::new_with_app_tag(
            crate::test_constants::TEST_PROJECT_ID.to_string(),
            "test-app".to_string(),
            Box::new(|| None),
        ));
        *client.node_id.write().await = Some(local_node_id_str.clone());

        let device = make_test_device_with_excluded(
            remote_device_id,
            &remote_node_id_str,
            &ticket_str,
            vec![local_device_id.to_string()],
        );

        assert!(!client.is_auto_connect_excluded(remote_device_id));

        let (
            mut last_attempt_at,
            mut failure_count,
            mut failure_backoff_until,
            mut non_initiator_wait_started_at,
            mut non_initiator_escalation_count,
            mut skip_log_at,
            mut connected_health_checked_at,
            mut connected_health_state,
            mut connected_health_failures,
            mut last_presence_republish_at_ms,
            mut last_network_change_at_ms,
            mut last_known_node_id,
        ) = make_loop_state();

        client
            .try_auto_connect_device(
                "user-1",
                device.clone(),
                local_device_id,
                &mut last_attempt_at,
                &mut failure_count,
                &mut failure_backoff_until,
                500,
                &mut non_initiator_wait_started_at,
                &mut non_initiator_escalation_count,
                0,
                &mut skip_log_at,
                &mut connected_health_checked_at,
                &mut connected_health_state,
                &mut connected_health_failures,
                5_000,
                3,
                &mut last_presence_republish_at_ms,
                60_000,
                &mut last_network_change_at_ms,
                30_000,
                2,
                &mut last_known_node_id,
            )
            .await;

        assert!(
            !last_attempt_at.contains_key(remote_device_id),
            "remote excluded_peers containing local device_id must suppress auto-connect"
        );

        ep_local.close().await;
        ep_remote.close().await;
    }

    /// When the remote device's excluded_peers does NOT contain our local device_id,
    /// the dial is not blocked by the remote-side check (local exclusion still works
    /// independently but is not set here — to isolate the remote field).
    #[tokio::test]
    async fn auto_connect_not_skipped_when_remote_excludes_different_device() {
        let ep_local = Endpoint::builder(iroh::endpoint::presets::N0)
            .relay_mode(RelayMode::Disabled)
            .bind()
            .await
            .expect("local endpoint");
        let ep_remote = Endpoint::builder(iroh::endpoint::presets::N0)
            .relay_mode(RelayMode::Disabled)
            .bind()
            .await
            .expect("remote endpoint");

        let addr_a = wait_for_endpoint_addr(&ep_local, 3_000)
            .await
            .expect("local addr");
        let addr_b = wait_for_endpoint_addr(&ep_remote, 3_000)
            .await
            .expect("remote addr");

        // Force initiator role so the non-initiator grace is not an obstacle.
        // local_node_id_str must be the larger ID so our client is the initiator.
        let id_a = ep_local.id().to_string();
        let id_b = ep_remote.id().to_string();
        let (local_node_id_str, remote_node_id_str, remote_addr) = if id_a > id_b {
            (id_a, id_b, addr_b)
        } else {
            (id_b, id_a, addr_a)
        };

        let ticket = iroh_tickets::endpoint::EndpointTicket::new(remote_addr);
        let ticket_str = ticket.to_string();

        let remote_device_id = "not-excluding-device";
        let local_device_id = "my-local-device";

        let client = std::sync::Arc::new(Client::new_with_app_tag(
            crate::test_constants::TEST_PROJECT_ID.to_string(),
            "test-app".to_string(),
            Box::new(|| None),
        ));
        *client.node_id.write().await = Some(local_node_id_str.clone());

        // Remote device excludes a *different* device, not ours.
        let device = make_test_device_with_excluded(
            remote_device_id,
            &remote_node_id_str,
            &ticket_str,
            vec!["some-other-device".to_string()],
        );

        let (
            mut last_attempt_at,
            mut failure_count,
            mut failure_backoff_until,
            mut non_initiator_wait_started_at,
            mut non_initiator_escalation_count,
            mut skip_log_at,
            mut connected_health_checked_at,
            mut connected_health_state,
            mut connected_health_failures,
            mut last_presence_republish_at_ms,
            mut last_network_change_at_ms,
            mut last_known_node_id,
        ) = make_loop_state();

        client
            .try_auto_connect_device(
                "user-1",
                device.clone(),
                local_device_id,
                &mut last_attempt_at,
                &mut failure_count,
                &mut failure_backoff_until,
                500,
                &mut non_initiator_wait_started_at,
                &mut non_initiator_escalation_count,
                0,
                &mut skip_log_at,
                &mut connected_health_checked_at,
                &mut connected_health_state,
                &mut connected_health_failures,
                5_000,
                3,
                &mut last_presence_republish_at_ms,
                60_000,
                &mut last_network_change_at_ms,
                30_000,
                2,
                &mut last_known_node_id,
            )
            .await;

        // An attempt SHOULD be recorded (the throttle kicks in, but the key exists).
        assert!(
            last_attempt_at.contains_key(remote_device_id),
            "dial should proceed when remote excluded_peers does not contain local device_id"
        );

        ep_local.close().await;
        ep_remote.close().await;
    }

    /// Remote exclusion entries are matched case-insensitively so mixed-case
    /// device ids cannot reconnect-loop after a manual disconnect.
    #[tokio::test]
    async fn auto_connect_skips_remote_excluded_local_device_case_insensitive_match() {
        let ep_local = Endpoint::builder(iroh::endpoint::presets::N0)
            .relay_mode(RelayMode::Disabled)
            .bind()
            .await
            .expect("local endpoint");
        let ep_remote = Endpoint::builder(iroh::endpoint::presets::N0)
            .relay_mode(RelayMode::Disabled)
            .bind()
            .await
            .expect("remote endpoint");

        let addr_a = wait_for_endpoint_addr(&ep_local, 3_000)
            .await
            .expect("local addr");
        let addr_b = wait_for_endpoint_addr(&ep_remote, 3_000)
            .await
            .expect("remote addr");

        // Force initiator role.
        let id_a = ep_local.id().to_string();
        let id_b = ep_remote.id().to_string();
        let (local_node_id_str, remote_node_id_str, remote_addr) = if id_a > id_b {
            (id_a, id_b, addr_b)
        } else {
            (id_b, id_a, addr_a)
        };

        let ticket = iroh_tickets::endpoint::EndpointTicket::new(remote_addr);
        let ticket_str = ticket.to_string();

        let remote_device_id = "case-test-remote";
        let local_device_id = "MY-LOCAL-DEVICE";

        let client = std::sync::Arc::new(Client::new_with_app_tag(
            crate::test_constants::TEST_PROJECT_ID.to_string(),
            "test-app".to_string(),
            Box::new(|| None),
        ));
        *client.node_id.write().await = Some(local_node_id_str.clone());

        let device = make_test_device_with_excluded(
            remote_device_id,
            &remote_node_id_str,
            &ticket_str,
            vec!["my-local-device".to_string()],
        );

        let (
            mut last_attempt_at,
            mut failure_count,
            mut failure_backoff_until,
            mut non_initiator_wait_started_at,
            mut non_initiator_escalation_count,
            mut skip_log_at,
            mut connected_health_checked_at,
            mut connected_health_state,
            mut connected_health_failures,
            mut last_presence_republish_at_ms,
            mut last_network_change_at_ms,
            mut last_known_node_id,
        ) = make_loop_state();

        client
            .try_auto_connect_device(
                "user-1",
                device.clone(),
                local_device_id,
                &mut last_attempt_at,
                &mut failure_count,
                &mut failure_backoff_until,
                500,
                &mut non_initiator_wait_started_at,
                &mut non_initiator_escalation_count,
                0,
                &mut skip_log_at,
                &mut connected_health_checked_at,
                &mut connected_health_state,
                &mut connected_health_failures,
                5_000,
                3,
                &mut last_presence_republish_at_ms,
                60_000,
                &mut last_network_change_at_ms,
                30_000,
                2,
                &mut last_known_node_id,
            )
            .await;

        assert!(
            !last_attempt_at.contains_key(remote_device_id),
            "remote excluded_peers must block by case-insensitive match"
        );

        ep_local.close().await;
        ep_remote.close().await;
    }

    /// current_excluded_peers_snapshot returns the in-memory exclusion set as a sorted Vec.
    #[test]
    fn current_excluded_peers_snapshot_reflects_set() {
        let client = std::sync::Arc::new(Client::new_with_app_tag(
            crate::test_constants::TEST_PROJECT_ID.to_string(),
            "test-app".to_string(),
            Box::new(|| None),
        ));

        assert!(client.current_excluded_peers_snapshot().is_empty());

        client.set_auto_connect_excluded("device-b", true);
        client.set_auto_connect_excluded("device-a", true);

        let snapshot = client.current_excluded_peers_snapshot();
        assert_eq!(snapshot, vec!["device-a", "device-b"]); // sorted

        client.set_auto_connect_excluded("device-a", false);
        let snapshot2 = client.current_excluded_peers_snapshot();
        assert_eq!(snapshot2, vec!["device-b"]);
    }

    #[tokio::test]
    async fn connect_device_rejects_invalid_ticket_before_runtime_lookup() {
        let client = Client::new_with_app_tag(
            crate::test_constants::TEST_PROJECT_ID.to_string(),
            "test-app".to_string(),
            Box::new(|| None),
        );

        let error = client
            .connect_device(Some("device-1"), "not-a-valid-ticket")
            .await
            .expect_err("invalid endpoint ticket should fail");

        let message = error.to_string().to_ascii_lowercase();
        assert!(
            message.contains("endpoint") || message.contains("ticket"),
            "unexpected connect_device error: {}",
            error
        );
    }

    #[tokio::test]
    async fn connect_device_coalesces_concurrent_dials_for_same_endpoint() {
        let client_a = Client::new_with_app_tag(
            crate::test_constants::TEST_PROJECT_ID.to_string(),
            "connect-coalesce-a".to_string(),
            Box::new(|| None),
        );
        let client_b = Client::new_with_app_tag(
            crate::test_constants::TEST_PROJECT_ID.to_string(),
            "connect-coalesce-b".to_string(),
            Box::new(|| None),
        );

        let endpoint_a = Endpoint::builder(iroh::endpoint::presets::N0)
            .alpns(vec![crate::native_node::PlutoniumProtocol::ALPN.to_vec()])
            .relay_mode(RelayMode::Disabled)
            .bind_addr(
                "127.0.0.1:0"
                    .parse::<std::net::SocketAddr>()
                    .expect("socket addr"),
            )
            .expect("bind addr a")
            .bind()
            .await
            .expect("bind endpoint a");
        let endpoint_b = Endpoint::builder(iroh::endpoint::presets::N0)
            .alpns(vec![crate::native_node::PlutoniumProtocol::ALPN.to_vec()])
            .relay_mode(RelayMode::Disabled)
            .bind_addr(
                "127.0.0.1:0"
                    .parse::<std::net::SocketAddr>()
                    .expect("socket addr"),
            )
            .expect("bind addr b")
            .bind()
            .await
            .expect("bind endpoint b");

        client_a.adopt_endpoint(endpoint_a).await;
        client_b.adopt_endpoint(endpoint_b).await;

        let endpoint_b = client_b.get_endpoint().await.expect("endpoint b");
        let _ = wait_for_endpoint_addr(&endpoint_b, 5_000)
            .await
            .expect("endpoint b addr");
        let ticket_b = client_b.endpoint_ticket().await.expect("endpoint ticket");

        let (first, second) = tokio::join!(
            client_a.connect_device(Some("device-b"), ticket_b.as_str()),
            client_a.connect_device(Some("device-b"), ticket_b.as_str()),
        );

        let first = first.expect("first connect succeeds");
        let second = second.expect("second connect coalesces");
        assert_eq!(first.connection_id, second.connection_id);

        let record = client_a
            .connection_manager
            .get_by_connection_id(&first.connection_id)
            .await
            .expect("managed record");
        assert_eq!(record.state, ConnectionState::Connected);
        assert_eq!(
            record.transport_generation, 2,
            "coalesced connect should keep one native transport binding plus one direct-path status generation"
        );
        assert_eq!(
            record.replacement_count, 1,
            "coalesced connect should not create a replacement transport"
        );
        assert_eq!(
            client_a.connection_manager.list_all().await.len(),
            1,
            "duplicate connect callers should share the same managed record"
        );
    }

    #[tokio::test]
    async fn connect_device_reports_connected_connection_state_on_return() {
        let client_a = Client::new_with_app_tag(
            crate::test_constants::TEST_PROJECT_ID.to_string(),
            "connect-state-a".to_string(),
            Box::new(|| None),
        );
        let client_b = Client::new_with_app_tag(
            crate::test_constants::TEST_PROJECT_ID.to_string(),
            "connect-state-b".to_string(),
            Box::new(|| None),
        );

        let endpoint_a = Endpoint::builder(iroh::endpoint::presets::N0)
            .alpns(vec![crate::native_node::PlutoniumProtocol::ALPN.to_vec()])
            .relay_mode(RelayMode::Disabled)
            .bind_addr(
                "127.0.0.1:0"
                    .parse::<std::net::SocketAddr>()
                    .expect("socket addr"),
            )
            .expect("bind addr a")
            .bind()
            .await
            .expect("bind endpoint a");
        let endpoint_b = Endpoint::builder(iroh::endpoint::presets::N0)
            .alpns(vec![crate::native_node::PlutoniumProtocol::ALPN.to_vec()])
            .relay_mode(RelayMode::Disabled)
            .bind_addr(
                "127.0.0.1:0"
                    .parse::<std::net::SocketAddr>()
                    .expect("socket addr"),
            )
            .expect("bind addr b")
            .bind()
            .await
            .expect("bind endpoint b");

        client_a.adopt_endpoint(endpoint_a).await;
        client_b.adopt_endpoint(endpoint_b).await;

        let ticket_b = client_b.endpoint_ticket().await.expect("endpoint ticket");

        let result = client_a
            .connect_device(Some("device-b"), ticket_b.as_str())
            .await
            .expect("managed connect succeeds");

        let record = client_a
            .connection_manager
            .get_by_connection_id(&result.connection_id)
            .await
            .expect("managed record");
        assert_eq!(record.state, ConnectionState::Connected);

        let snapshot = client_a
            .connection_state(&result.connection_id)
            .await
            .expect("connection state snapshot");
        assert_eq!(snapshot.state, "connected");
        assert_eq!(snapshot.transport_state, "connected");
    }

    #[tokio::test]
    async fn ensure_connected_addr_commits_connection_manager_before_return() {
        let client_a = Client::new_with_app_tag(
            crate::test_constants::TEST_PROJECT_ID.to_string(),
            "ensure-connected-record-a".to_string(),
            Box::new(|| None),
        );
        let client_b = Client::new_with_app_tag(
            crate::test_constants::TEST_PROJECT_ID.to_string(),
            "ensure-connected-record-b".to_string(),
            Box::new(|| None),
        );

        let endpoint_a = Endpoint::builder(iroh::endpoint::presets::N0)
            .alpns(vec![crate::native_node::PlutoniumProtocol::ALPN.to_vec()])
            .relay_mode(RelayMode::Disabled)
            .bind_addr(
                "127.0.0.1:0"
                    .parse::<std::net::SocketAddr>()
                    .expect("socket addr"),
            )
            .expect("bind addr a")
            .bind()
            .await
            .expect("bind endpoint a");
        let endpoint_b = Endpoint::builder(iroh::endpoint::presets::N0)
            .alpns(vec![crate::native_node::PlutoniumProtocol::ALPN.to_vec()])
            .relay_mode(RelayMode::Disabled)
            .bind_addr(
                "127.0.0.1:0"
                    .parse::<std::net::SocketAddr>()
                    .expect("socket addr"),
            )
            .expect("bind addr b")
            .bind()
            .await
            .expect("bind endpoint b");

        client_a.adopt_endpoint(endpoint_a).await;
        client_b.adopt_endpoint(endpoint_b).await;

        let endpoint_b = client_b.get_endpoint().await.expect("endpoint b");
        let endpoint_b_id = endpoint_b.id();
        let endpoint_b_addr = wait_for_endpoint_addr(&endpoint_b, 5_000)
            .await
            .expect("endpoint b addr");
        let remote_node_id = endpoint_b_id.to_string();
        let local_node_id = client_a.current_node_id().await.expect("local node id");
        let connection_id = Client::deterministic_connection_id(&local_node_id, &remote_node_id);

        client_a
            .connection_manager
            .upsert_pending(
                connection_id.clone(),
                Some(remote_node_id.clone()),
                Some("device-b".to_string()),
                Some(remote_node_id.clone()),
            )
            .await;
        client_a
            .connection_manager
            .set_connecting(&connection_id)
            .await;

        client_a
            .ensure_connected_addr(endpoint_b_id, endpoint_b_addr)
            .await
            .expect("transport dial succeeds");

        let record = client_a
            .connection_manager
            .get_by_connection_id(&connection_id)
            .await
            .expect("managed record exists before bi-stream open");
        assert_eq!(record.state, ConnectionState::Connected);

        let snapshot = client_a
            .connection_state(&connection_id)
            .await
            .expect("connection state snapshot");
        assert_eq!(snapshot.state, "connected");
        assert_eq!(snapshot.transport_state, "connected");

        let (_send, _recv) = client_a
            .open_bi(endpoint_b_id)
            .await
            .expect("bi-stream open succeeds after managed record commit");
    }

    #[tokio::test]
    async fn resolve_peer_connection_ids_uses_core_peer_identity_aliases() {
        let client = Client::new_with_app_tag(
            crate::test_constants::TEST_PROJECT_ID.to_string(),
            "test-app".to_string(),
            Box::new(|| None),
        );

        client
            .connection_manager
            .upsert_pending(
                "conn-1".to_string(),
                Some("node1".to_string()),
                Some("device-hint-1".to_string()),
                Some("node1".to_string()),
            )
            .await;
        client.connection_manager.set_connecting("conn-1").await;
        client
            .connection_manager
            .set_device_id("conn-1", "device-1".to_string())
            .await;

        assert_eq!(
            client.resolve_peer_connection_ids("device-1").await,
            vec!["conn-1".to_string()]
        );
        assert_eq!(
            client.resolve_peer_connection_ids("device-hint-1").await,
            vec!["conn-1".to_string()]
        );
        assert_eq!(
            client.resolve_peer_connection_ids("node1").await,
            vec!["conn-1".to_string()]
        );
        assert_eq!(
            client.resolve_peer_connection_ids("conn-1").await,
            vec!["conn-1".to_string()]
        );
        assert_eq!(
            client
                .resolve_peer_connection_ids("node1-transientroutenode")
                .await,
            vec!["conn-1".to_string()],
            "a transport-specific route id that shares the peer node should resolve through the core alias index",
        );

        client
            .connection_manager
            .upsert_pending("node1-transientroutenode".to_string(), None, None, None)
            .await;

        assert_eq!(
            client
                .resolve_peer_connection_ids("node1-transientroutenode")
                .await,
            vec!["node1-transientroutenode".to_string(), "conn-1".to_string()],
            "a pending transient route record must not hide the canonical peer aliases",
        );

        client
            .connection_manager
            .upsert_pending(
                "desktopnode-browsernode".to_string(),
                Some("desktopnode".to_string()),
                None,
                Some("desktopnode".to_string()),
            )
            .await;

        assert_eq!(
            client
                .resolve_peer_connection_ids("transientroute-browsernode")
                .await,
            vec!["desktopnode-browsernode".to_string()],
            "compound route ids should resolve canonical peers by shared deterministic id halves even when node indexes are incomplete",
        );
    }

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

        client
            .connection_manager
            .upsert_pending(
                "canonical-conn".to_string(),
                Some("remotepeer".to_string()),
                Some("device-hint-1".to_string()),
                Some("remotepeer".to_string()),
            )
            .await;
        client
            .connection_manager
            .set_connecting("canonical-conn")
            .await;

        assert_eq!(
            client
                .resolve_transport_connection_ids("transientroute-remotepeer")
                .await,
            vec![
                "transientroute-remotepeer".to_string(),
                "canonical-conn".to_string()
            ],
            "transport lookup must check the exact session key before canonical peer aliases"
        );
    }

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

        client
            .connection_manager
            .upsert_pending(
                "localnode-remotenode".to_string(),
                Some("remotenode".to_string()),
                Some("remote-device".to_string()),
                Some("remotenode".to_string()),
            )
            .await;
        client
            .connection_manager
            .set_connected("localnode-remotenode", Some("remotenode".to_string()))
            .await;

        assert_eq!(
            client.resolve_transport_connection_ids("remotenode").await,
            vec!["remotenode".to_string(), "localnode-remotenode".to_string()],
            "transport lookup must not stop at a node-id candidate before finding the managed connection record"
        );
    }

    #[tokio::test]
    async fn resolve_peer_connection_records_returns_sorted_records() {
        let client = Client::new_with_app_tag(
            crate::test_constants::TEST_PROJECT_ID.to_string(),
            "test-app".to_string(),
            Box::new(|| None),
        );

        client
            .connection_manager
            .upsert_pending(
                "conn-1".to_string(),
                Some("node-1".to_string()),
                Some("device-1".to_string()),
                Some("node-1".to_string()),
            )
            .await;
        client.connection_manager.set_connecting("conn-1").await;

        tokio::time::sleep(std::time::Duration::from_millis(2)).await;

        client
            .connection_manager
            .upsert_pending(
                "conn-2".to_string(),
                Some("node-2".to_string()),
                Some("device-1".to_string()),
                Some("node-2".to_string()),
            )
            .await;
        client
            .connection_manager
            .set_connected("conn-2", Some("node-2".to_string()))
            .await;

        let records = client.resolve_peer_connection_records("device-1").await;
        assert_eq!(records.len(), 2);
        assert_eq!(records[0].connection_id, "conn-2");
        assert_eq!(records[1].connection_id, "conn-1");
    }

    #[tokio::test]
    async fn managed_connection_device_hint_prefers_authoritative_identity() {
        let client = Client::new_with_app_tag(
            crate::test_constants::TEST_PROJECT_ID.to_string(),
            "test-app".to_string(),
            Box::new(|| None),
        );

        client
            .connection_manager
            .upsert_pending(
                "conn-1".to_string(),
                Some("node-1".to_string()),
                Some("device-hint-1".to_string()),
                Some("node-1".to_string()),
            )
            .await;
        client
            .connection_manager
            .set_device_id("conn-1", "device-1".to_string())
            .await;

        assert_eq!(
            client
                .managed_connection_device_hint("conn-1")
                .await
                .as_deref(),
            Some("device-1")
        );
        assert_eq!(
            client
                .managed_connection_device_hint("device-hint-1")
                .await
                .as_deref(),
            Some("device-1")
        );
    }

    #[tokio::test]
    async fn managed_connection_bridge_action_requests_rebind_for_stale_transport() {
        let client = Client::new_with_app_tag(
            crate::test_constants::TEST_PROJECT_ID.to_string(),
            "test-app".to_string(),
            Box::new(|| None),
        );

        let node_id =
            "5d388fcaaa61efa536d78051e060928f6c1fff0983a32f6b532176b12328c3ac".to_string();

        client
            .connection_manager
            .upsert_pending(
                "conn-1".to_string(),
                Some(node_id.clone()),
                Some("device-1".to_string()),
                Some(node_id.clone()),
            )
            .await;
        client
            .connection_manager
            .set_connected("conn-1", Some(node_id.clone()))
            .await;

        assert_eq!(
            client.managed_connection_bridge_action("conn-1").await,
            Some(ManagedConnectionBridgeAction::Rebind {
                remote_node_id: Some(node_id),
                transport_generation: 0,
            })
        );
    }

    #[tokio::test]
    async fn transport_connected_peer_session_stays_connecting_until_settled() {
        let client = Client::new_with_app_tag(
            crate::test_constants::TEST_PROJECT_ID.to_string(),
            "test-app".to_string(),
            Box::new(|| None),
        );

        client
            .connection_manager
            .upsert_pending(
                "conn-transport-only-1".to_string(),
                Some("node-transport-only-1".to_string()),
                Some("device-transport-only-1".to_string()),
                Some("node-transport-only-1".to_string()),
            )
            .await;
        client
            .connection_manager
            .set_connected(
                "conn-transport-only-1",
                Some("node-transport-only-1".to_string()),
            )
            .await;

        let session = client
            .peer_session("device-transport-only-1")
            .await
            .expect("peer session snapshot");
        assert_eq!(session.peer_id, "device-transport-only-1");
        assert_eq!(session.status, ConnectionState::Connecting);
        assert_eq!(session.health, ConnectionHealth::Unknown);
        assert!(!session.settled_ready);
        assert_eq!(
            session.active_connection_id.as_deref(),
            Some("conn-transport-only-1")
        );

        let connection_state = client
            .connection_state("conn-transport-only-1")
            .await
            .expect("connection state snapshot");
        assert_eq!(connection_state.state, "connecting");
        assert_eq!(connection_state.transport_state, "connected");
        assert_eq!(connection_state.protocol_state, "transport-only");
        assert!(!connection_state.routable);

        let remote_device = Device {
            app_tag: Some("test-app".to_string()),
            device_id: "device-transport-only-1".to_string(),
            user_id: Some("user-1".to_string()),
            device_name: "Remote Device".to_string(),
            platform_type: Some("web".to_string()),
            capabilities: None,
            session_id: None,
            node_id: Some("node-transport-only-1".to_string()),
            online: true,
            ticket: Some("ticket-transport-only-1".to_string()),
            tag: Some("test-app".to_string()),
            kind: Some("device".to_string()),
            metadata: None,
            last_seen_at: None,
            expires_at: None,
            created_at: None,
            updated_at: None,
            excluded_peers: vec![],
        };
        let snapshots = merge_device_status_snapshots(
            vec![remote_device],
            client.connection_manager.list_peer_snapshots().await,
        );
        let snapshot = snapshots.first().expect("device status snapshot");
        assert_eq!(
            snapshot.connection_status,
            DeviceConnectionStatus::Connecting
        );
        assert!(!snapshot.settled_ready);
    }

    #[tokio::test]
    async fn managed_connection_bridge_action_retires_after_settle_timeout() {
        let client = Client::new_with_app_tag(
            crate::test_constants::TEST_PROJECT_ID.to_string(),
            "test-app".to_string(),
            Box::new(|| None),
        );

        let node_id =
            "5d388fcaaa61efa536d78051e060928f6c1fff0983a32f6b532176b12328c3ac".to_string();

        client
            .connection_manager
            .upsert_pending(
                "conn-timeout-1".to_string(),
                Some(node_id.clone()),
                Some("device-timeout-1".to_string()),
                Some(node_id.clone()),
            )
            .await;
        client
            .connection_manager
            .set_connected("conn-timeout-1", Some(node_id))
            .await;
        client
            .connection_manager
            .set_last_transport_change_at_ms_for_test(
                "conn-timeout-1",
                crate::firebase::now_millis_u64()
                    .min(i64::MAX as u64)
                    .saturating_sub((MANAGED_SETTLE_DEADLINE_MS + 1) as u64) as i64,
            )
            .await
            .expect("managed record");

        let snapshot = client
            .managed_connection_health("conn-timeout-1")
            .await
            .expect("managed connection health");
        assert_eq!(snapshot.status, ManagedConnectionHealthStatus::Dead);
        assert!(!snapshot.settled_ready);

        assert_eq!(
            client
                .managed_connection_bridge_action("conn-timeout-1")
                .await,
            Some(ManagedConnectionBridgeAction::Retire {
                reason: "settle-timeout".to_string(),
            })
        );
    }

    #[tokio::test]
    async fn managed_connection_bridge_action_keeps_recovered_transport_after_settle_deadline() {
        let client_a = Client::new_with_app_tag(
            crate::test_constants::TEST_PROJECT_ID.to_string(),
            "deadline-a".to_string(),
            Box::new(|| None),
        );
        let client_b = Client::new_with_app_tag(
            crate::test_constants::TEST_PROJECT_ID.to_string(),
            "deadline-b".to_string(),
            Box::new(|| None),
        );

        let endpoint_a = Endpoint::builder(iroh::endpoint::presets::N0)
            .alpns(vec![crate::native_node::PlutoniumProtocol::ALPN.to_vec()])
            .relay_mode(RelayMode::Disabled)
            .bind_addr(
                "127.0.0.1:0"
                    .parse::<std::net::SocketAddr>()
                    .expect("socket addr"),
            )
            .expect("bind addr a")
            .bind()
            .await
            .expect("bind endpoint a");
        let endpoint_b = Endpoint::builder(iroh::endpoint::presets::N0)
            .alpns(vec![crate::native_node::PlutoniumProtocol::ALPN.to_vec()])
            .relay_mode(RelayMode::Disabled)
            .bind_addr(
                "127.0.0.1:0"
                    .parse::<std::net::SocketAddr>()
                    .expect("socket addr"),
            )
            .expect("bind addr b")
            .bind()
            .await
            .expect("bind endpoint b");

        client_a.adopt_endpoint(endpoint_a).await;
        client_b.adopt_endpoint(endpoint_b).await;

        let endpoint_a = client_a.get_endpoint().await.expect("endpoint a");
        let endpoint_b = client_b.get_endpoint().await.expect("endpoint b");
        let endpoint_a_id = endpoint_a.id();
        let endpoint_b_id = endpoint_b.id();
        let endpoint_b_addr = wait_for_endpoint_addr(&endpoint_b, 5_000)
            .await
            .expect("endpoint b addr");

        client_a
            .ensure_connected_addr(endpoint_b_id, endpoint_b_addr)
            .await
            .expect("initial connection");

        let kept_connection = tokio::time::timeout(Duration::from_secs(5), async {
            loop {
                if let Some(connection) = client_b.get_connection(endpoint_a_id).await {
                    break connection;
                }
                sleep(Duration::from_millis(50)).await;
            }
        })
        .await
        .expect("timed out waiting for kept connection");
        let kept_stable_id = kept_connection.stable_id() as u64;

        let remote_node_id = endpoint_a_id.to_string();
        let local_node_id = endpoint_b_id.to_string();
        let connection_id = Client::deterministic_connection_id(&local_node_id, &remote_node_id);
        let synthetic_closed_stable_id = kept_stable_id.saturating_add(100);

        client_b
            .connection_manager
            .upsert_pending(
                connection_id.clone(),
                Some(remote_node_id.clone()),
                Some("device-a".to_string()),
                Some(remote_node_id.clone()),
            )
            .await;
        client_b
            .connection_manager
            .set_connected_with_transport(
                &connection_id,
                Some(remote_node_id.clone()),
                Some(synthetic_closed_stable_id),
                Some("incoming".to_string()),
            )
            .await
            .expect("connected record");
        client_b
            .connection_manager
            .set_device_id(&connection_id, "device-a".to_string())
            .await
            .expect("device binding");

        let resolution = client_b
            .reconcile_incoming_transport_after_local_close(
                &connection_id,
                endpoint_a_id,
                &remote_node_id,
                synthetic_closed_stable_id,
                "LocallyClosed",
            )
            .await;

        assert_eq!(
            resolution,
            IncomingTransportCloseResolution::PreservedKeptTransport
        );

        let settled = client_b
            .wait_for_settled_peer("device-a", Some(2_000))
            .await
            .expect("settled peer snapshot");
        assert!(
            settled.settled_ready,
            "expected recovered peer to be settled"
        );
        assert_eq!(
            settled.active_connection_id.as_deref(),
            Some(connection_id.as_str())
        );

        client_b
            .connection_manager
            .set_last_transport_change_at_ms_for_test(
                &connection_id,
                crate::firebase::now_millis_u64()
                    .min(i64::MAX as u64)
                    .saturating_sub((MANAGED_SETTLE_DEADLINE_MS + 1) as u64) as i64,
            )
            .await
            .expect("managed record");

        let snapshot = client_b
            .managed_connection_health(&connection_id)
            .await
            .expect("managed connection health");
        assert_eq!(snapshot.status, ManagedConnectionHealthStatus::Healthy);
        assert!(snapshot.settled_ready);
        assert_eq!(snapshot.active_transport_stable_id, Some(kept_stable_id));

        assert_eq!(
            client_b
                .managed_connection_bridge_action(&connection_id)
                .await,
            Some(ManagedConnectionBridgeAction::Healthy)
        );
    }

    #[test]
    fn duplicate_closed_handling_preserves_any_healthy_replacement_transport() {
        assert_eq!(
            duplicate_closed_handling(
                "ApplicationClosed(0, b\"duplicate-kept-existing\")",
                true,
                true,
                2,
                1
            ),
            DuplicateClosedHandling::RebindKeptTransport
        );
        assert_eq!(
            duplicate_closed_handling("LocallyClosed", true, true, 2, 1),
            DuplicateClosedHandling::RebindKeptTransport
        );
        assert_eq!(
            duplicate_closed_handling(
                "ApplicationClosed(0, b\"replaced-by-fresh-incoming\")",
                true,
                true,
                2,
                1
            ),
            DuplicateClosedHandling::RebindKeptTransport
        );
        assert_eq!(
            duplicate_closed_handling(
                "ApplicationClosed(0, b\"duplicate-kept-existing\")",
                true,
                false,
                2,
                1
            ),
            DuplicateClosedHandling::RebindKeptTransport
        );
        assert_eq!(
            duplicate_closed_handling(
                "ApplicationClosed(0, b\"duplicate-kept-existing\")",
                true,
                true,
                1,
                1
            ),
            DuplicateClosedHandling::RetireClosedTransport
        );
    }

    #[test]
    fn incoming_local_timeout_close_waits_for_replacement_grace() {
        assert!(incoming_local_close_should_wait_for_replacement("TimedOut"));
        assert!(incoming_local_close_should_wait_for_replacement(
            "Some(TimedOut)"
        ));
        assert!(incoming_local_close_should_wait_for_replacement(
            "transport timed out while replacement was active"
        ));
        assert!(incoming_local_close_should_wait_for_replacement(
            "ApplicationClosed(0, b\"replaced-by-new-outbound\")"
        ));
        assert!(!incoming_local_close_should_wait_for_replacement(
            "ApplicationClosed(0, b\"manual disconnect\")"
        ));
        assert!(!incoming_local_close_should_wait_for_replacement("Closed"));
    }

    #[tokio::test]
    async fn incoming_local_timeout_close_preserves_manager_record_for_replacement_grace() {
        let client = Client::new_with_app_tag(
            crate::test_constants::TEST_PROJECT_ID.to_string(),
            "test-app".to_string(),
            Box::new(|| None),
        );
        client
            .connection_manager
            .upsert_pending(
                "conn-timeout-close".to_string(),
                Some("node-timeout-close".to_string()),
                Some("device-timeout-close".to_string()),
                Some("node-timeout-close".to_string()),
            )
            .await;
        client
            .connection_manager
            .set_connected_with_transport(
                "conn-timeout-close",
                Some("node-timeout-close".to_string()),
                Some(10),
                Some("incoming".to_string()),
            )
            .await
            .expect("connected record");

        let endpoint_id = "5d388fcaaa61efa536d78051e060928f6c1fff0983a32f6b532176b12328c3ac"
            .parse()
            .expect("endpoint id");
        let resolution = client
            .reconcile_incoming_transport_after_local_close(
                "conn-timeout-close",
                endpoint_id,
                "node-timeout-close",
                10,
                "Some(TimedOut)",
            )
            .await;

        assert_eq!(
            resolution,
            IncomingTransportCloseResolution::PreservedKeptTransport
        );
        let record = client
            .connection_manager
            .get_by_connection_id("conn-timeout-close")
            .await
            .expect("record");
        assert_eq!(
            record.status_reason.as_deref(),
            Some(crate::lifecycle_reason::REASON_REPLACEMENT_IN_PROGRESS)
        );
        assert!(matches!(
            record.state,
            ConnectionState::Connecting | ConnectionState::Pending | ConnectionState::Connected
        ));
    }

    #[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn retire_managed_connection_defers_while_native_webrtc_is_connecting() {
        let client = Client::new_with_app_tag(
            crate::test_constants::TEST_PROJECT_ID.to_string(),
            "test-app".to_string(),
            Box::new(|| None),
        );

        client
            .connection_manager
            .upsert_pending(
                "conn-webrtc".to_string(),
                Some("node-remote".to_string()),
                Some("device-remote".to_string()),
                Some("node-remote".to_string()),
            )
            .await;
        client
            .connection_manager
            .set_connected_with_transport(
                "conn-webrtc",
                Some("node-remote".to_string()),
                None,
                Some("incoming".to_string()),
            )
            .await
            .expect("connected record");

        let session = std::sync::Arc::new(
            crate::transport::NativeWebRTCDataChannel::new(
                "node-z",
                "node-a",
                &crate::client::WebRTCConfig {
                    ice_servers: vec![],
                    privacy_mode: false,
                    lan_mode: false,
                },
                std::sync::Arc::new(|_| Box::pin(async move {})),
                "negotiation-test",
            )
            .await
            .expect("native webrtc session"),
        );
        session.start().await.expect("session started");
        client
            .native_webrtc_sessions
            .write()
            .await
            .insert("conn-webrtc".to_string(), session.clone());

        // Not a terminal retire reason — `disconnected by user` skips deferral by design.
        client
            .retire_managed_connection("conn-webrtc", Some("transport-replaced".to_string()))
            .await;

        assert!(
            client
                .connection_manager
                .get_by_connection_id("conn-webrtc")
                .await
                .is_some(),
            "connection record should remain while webrtc is still connecting",
        );
        assert!(
            client
                .native_webrtc_sessions
                .read()
                .await
                .contains_key("conn-webrtc"),
            "webrtc session should remain registered while cleanup is deferred",
        );
        assert_eq!(
            client
                .deferred_managed_retirements
                .read()
                .await
                .get("conn-webrtc")
                .cloned(),
            Some(Some("transport-replaced".to_string())),
        );

        session.close();
        client
            .native_webrtc_sessions
            .write()
            .await
            .remove("conn-webrtc");
        assert!(
            client
                .finalize_deferred_managed_retirement_if_needed("conn-webrtc")
                .await,
            "deferred retirement should finalize after the session disappears",
        );
        assert!(
            client
                .connection_manager
                .get_by_connection_id("conn-webrtc")
                .await
                .is_none(),
            "finalized deferred retirement should remove the connection record",
        );
    }

    #[test]
    fn auto_connect_tie_break_observes_existing_transport_before_waiting() {
        assert_eq!(
            auto_connect_tie_break_decision("node-a", "node-z", true, 0, 5_000),
            AutoConnectTieBreakDecision::ObserveConnectedTransport
        );
        assert_eq!(
            auto_connect_tie_break_decision("node-a", "node-z", false, 500, 5_000),
            AutoConnectTieBreakDecision::WaitForInitiator
        );
        // Non-initiator escalates to ActAsInitiator after grace period expires
        assert_eq!(
            auto_connect_tie_break_decision("node-a", "node-z", false, 6_000, 5_000),
            AutoConnectTieBreakDecision::ActAsInitiator
        );
        assert_eq!(
            auto_connect_tie_break_decision("node-z", "node-a", false, 0, 5_000),
            AutoConnectTieBreakDecision::ActAsInitiator
        );
    }

    #[tokio::test]
    async fn managed_connection_health_reports_dead_for_closed_connection() {
        let client = Client::new_with_app_tag(
            crate::test_constants::TEST_PROJECT_ID.to_string(),
            "test-app".to_string(),
            Box::new(|| None),
        );

        client
            .connection_manager
            .upsert_pending(
                "conn-1".to_string(),
                Some("node-1".to_string()),
                Some("device-1".to_string()),
                Some("node-1".to_string()),
            )
            .await;
        client
            .connection_manager
            .set_closed("conn-1", Some("closed-for-test".to_string()))
            .await;

        let snapshot = client
            .managed_connection_health("conn-1")
            .await
            .expect("snapshot");
        assert_eq!(snapshot.status, ManagedConnectionHealthStatus::Dead);
        assert!(!snapshot.settled_ready);
        assert_eq!(snapshot.device_id_hint.as_deref(), Some("device-1"));
    }

    #[tokio::test]
    async fn managed_connection_health_reports_replacement_pending_without_successor_transport() {
        let client = Client::new_with_app_tag(
            crate::test_constants::TEST_PROJECT_ID.to_string(),
            "test-app".to_string(),
            Box::new(|| None),
        );

        client
            .connection_manager
            .upsert_pending(
                "conn-replacement-1".to_string(),
                Some("node-replacement-1".to_string()),
                Some("device-replacement-1".to_string()),
                Some("node-replacement-1".to_string()),
            )
            .await;
        client
            .connection_manager
            .set_connected_with_transport(
                "conn-replacement-1",
                Some("node-replacement-1".to_string()),
                Some(41),
                Some("outgoing".to_string()),
            )
            .await
            .expect("connected record");
        client
            .connection_manager
            .mark_transport_replaced(
                "conn-replacement-1",
                None,
                Some("outgoing-closed".to_string()),
                Some("replacement-in-progress".to_string()),
            )
            .await
            .expect("replacement record");

        let snapshot = client
            .managed_connection_health("conn-replacement-1")
            .await
            .expect("snapshot");
        assert_eq!(
            snapshot.status,
            ManagedConnectionHealthStatus::AwaitingReplacement
        );
        assert!(snapshot.replacement_pending);
        assert_eq!(snapshot.active_transport_stable_id, None);
        assert_eq!(
            snapshot.readiness_state,
            ReadinessState::AwaitingReplacement
        );
        assert_eq!(
            snapshot.readiness_reason,
            "missing-active-transport".to_string()
        );
    }

    #[tokio::test]
    async fn connection_state_snapshot_reports_connecting_during_in_place_reconnect_close() {
        // Regression: an in-place transport reconnect closes the base transport with a
        // replacement reason and re-binds on the SAME connection (relay -> direct ->
        // webrtc upgrade). The intermediate Closing/Closed frame must surface as the
        // reconnecting family ("connecting"), not "closed", so consumers do not flicker
        // connected -> disconnected -> connected on each transport generation bump.
        let client = Client::new_with_app_tag(
            crate::test_constants::TEST_PROJECT_ID.to_string(),
            "test-app".to_string(),
            Box::new(|| None),
        );

        client
            .connection_manager
            .upsert_pending(
                "conn-inplace-reconnect".to_string(),
                Some("node-inplace-reconnect".to_string()),
                Some("device-inplace-reconnect".to_string()),
                Some("node-inplace-reconnect".to_string()),
            )
            .await;
        client
            .connection_manager
            .set_connected_with_transport(
                "conn-inplace-reconnect",
                Some("node-inplace-reconnect".to_string()),
                Some(41),
                Some("outgoing".to_string()),
            )
            .await
            .expect("connected record");
        // Base transport drops mid-replacement (old transport closed, successor pending).
        client
            .connection_manager
            .set_closed(
                "conn-inplace-reconnect",
                Some(crate::lifecycle_reason::REASON_REPLACEMENT_IN_PROGRESS.to_string()),
            )
            .await
            .expect("closed record");

        let snapshot = client
            .connection_state("conn-inplace-reconnect")
            .await
            .expect("connection state snapshot");
        assert_eq!(
            snapshot.state, "connecting",
            "in-place reconnect close must surface as connecting, not closed/disconnected"
        );
        assert_eq!(snapshot.transport_state, "connecting");
        assert_eq!(snapshot.protocol_state, "connecting");
    }

    #[tokio::test]
    async fn connection_state_snapshot_reports_closed_for_genuine_close() {
        // Guard the negative: a genuine (non-reconnect) close must still surface as
        // closed, so the in-place-reconnect smoothing does not mask real disconnects.
        let client = Client::new_with_app_tag(
            crate::test_constants::TEST_PROJECT_ID.to_string(),
            "test-app".to_string(),
            Box::new(|| None),
        );

        client
            .connection_manager
            .upsert_pending(
                "conn-genuine-close".to_string(),
                Some("node-genuine-close".to_string()),
                Some("device-genuine-close".to_string()),
                Some("node-genuine-close".to_string()),
            )
            .await;
        client
            .connection_manager
            .set_connected_with_transport(
                "conn-genuine-close",
                Some("node-genuine-close".to_string()),
                Some(41),
                Some("outgoing".to_string()),
            )
            .await
            .expect("connected record");
        client
            .connection_manager
            .set_closed(
                "conn-genuine-close",
                Some(crate::lifecycle_reason::REASON_MANUAL_DISCONNECT.to_string()),
            )
            .await
            .expect("closed record");

        let snapshot = client
            .connection_state("conn-genuine-close")
            .await
            .expect("connection state snapshot");
        assert_eq!(snapshot.state, "closed");
        assert_eq!(snapshot.transport_state, "closed");
        assert_eq!(snapshot.protocol_state, "closed");
    }

    #[tokio::test]
    async fn managed_connection_bridge_action_awaits_replacement_without_successor_transport() {
        let client = Client::new_with_app_tag(
            crate::test_constants::TEST_PROJECT_ID.to_string(),
            "test-app".to_string(),
            Box::new(|| None),
        );

        client
            .connection_manager
            .upsert_pending(
                "conn-replacement-retire-repro".to_string(),
                Some("node-replacement-retire-repro".to_string()),
                Some("device-replacement-retire-repro".to_string()),
                Some("node-replacement-retire-repro".to_string()),
            )
            .await;
        client
            .connection_manager
            .set_connected_with_transport(
                "conn-replacement-retire-repro",
                Some("node-replacement-retire-repro".to_string()),
                Some(77),
                Some("outgoing".to_string()),
            )
            .await
            .expect("connected record");
        client
            .connection_manager
            .mark_transport_replaced(
                "conn-replacement-retire-repro",
                None,
                Some("outgoing-closed".to_string()),
                Some("replacement-in-progress".to_string()),
            )
            .await
            .expect("replacement record");

        let snapshot = client
            .managed_connection_health("conn-replacement-retire-repro")
            .await
            .expect("snapshot");
        assert_eq!(
            snapshot.status,
            ManagedConnectionHealthStatus::AwaitingReplacement
        );
        assert!(snapshot.replacement_pending);
        assert_eq!(snapshot.active_transport_stable_id, None);

        assert_eq!(
            client
                .managed_connection_bridge_action("conn-replacement-retire-repro")
                .await,
            Some(ManagedConnectionBridgeAction::AwaitReplacement {
                reason: "missing-active-transport".to_string(),
            })
        );
    }

    #[tokio::test]
    async fn reconcile_incoming_close_ignores_stale_closed_transport_after_generation_advance() {
        let client = Client::new_with_app_tag(
            crate::test_constants::TEST_PROJECT_ID.to_string(),
            "test-app".to_string(),
            Box::new(|| None),
        );

        client
            .connection_manager
            .upsert_pending(
                "conn-stale-close".to_string(),
                Some("node-stale-close".to_string()),
                Some("device-stale-close".to_string()),
                Some("node-stale-close".to_string()),
            )
            .await;
        client
            .connection_manager
            .set_connected_with_transport(
                "conn-stale-close",
                Some("node-stale-close".to_string()),
                Some(10),
                Some("incoming".to_string()),
            )
            .await
            .expect("connected record");
        client
            .connection_manager
            .mark_transport_replaced(
                "conn-stale-close",
                Some(11),
                Some("incoming-replacement".to_string()),
                Some("replacement-in-progress".to_string()),
            )
            .await
            .expect("replacement record");

        let endpoint_id = "5d388fcaaa61efa536d78051e060928f6c1fff0983a32f6b532176b12328c3ac"
            .parse()
            .expect("endpoint id");

        let resolution = client
            .reconcile_incoming_transport_after_local_close(
                "conn-stale-close",
                endpoint_id,
                "node-stale-close",
                10,
                "LocallyClosed",
            )
            .await;

        assert_eq!(
            resolution,
            IncomingTransportCloseResolution::PreservedKeptTransport
        );
        let record = client
            .connection_manager
            .get_by_connection_id("conn-stale-close")
            .await
            .expect("record");
        assert_eq!(record.state, ConnectionState::Connected);
        assert_eq!(record.transport_stable_id, Some(11));
        assert_eq!(record.transport_generation, 2);
        assert_eq!(
            record.status_reason.as_deref(),
            Some("replacement-in-progress")
        );
    }

    #[tokio::test]
    async fn reconcile_incoming_close_reconfirms_readiness_for_kept_transport() {
        let client_a = Client::new_with_app_tag(
            crate::test_constants::TEST_PROJECT_ID.to_string(),
            "test-a".to_string(),
            Box::new(|| None),
        );
        let client_b = Client::new_with_app_tag(
            crate::test_constants::TEST_PROJECT_ID.to_string(),
            "test-b".to_string(),
            Box::new(|| None),
        );

        let endpoint_a = Endpoint::builder(iroh::endpoint::presets::N0)
            .alpns(vec![crate::native_node::PlutoniumProtocol::ALPN.to_vec()])
            .relay_mode(RelayMode::Disabled)
            .bind_addr(
                "127.0.0.1:0"
                    .parse::<std::net::SocketAddr>()
                    .expect("socket addr"),
            )
            .expect("bind addr a")
            .bind()
            .await
            .expect("bind endpoint a");
        let endpoint_b = Endpoint::builder(iroh::endpoint::presets::N0)
            .alpns(vec![crate::native_node::PlutoniumProtocol::ALPN.to_vec()])
            .relay_mode(RelayMode::Disabled)
            .bind_addr(
                "127.0.0.1:0"
                    .parse::<std::net::SocketAddr>()
                    .expect("socket addr"),
            )
            .expect("bind addr b")
            .bind()
            .await
            .expect("bind endpoint b");

        client_a.adopt_endpoint(endpoint_a).await;
        client_b.adopt_endpoint(endpoint_b).await;

        let endpoint_a = client_a.get_endpoint().await.expect("endpoint a");
        let endpoint_b = client_b.get_endpoint().await.expect("endpoint b");
        let endpoint_a_id = endpoint_a.id();
        let endpoint_b_id = endpoint_b.id();
        let endpoint_b_addr = wait_for_endpoint_addr(&endpoint_b, 5_000)
            .await
            .expect("endpoint b addr");

        client_a
            .ensure_connected_addr(endpoint_b_id, endpoint_b_addr.clone())
            .await
            .expect("initial connection");

        let kept_connection = tokio::time::timeout(Duration::from_secs(5), async {
            loop {
                if let Some(connection) = client_b.get_connection(endpoint_a_id).await {
                    break connection;
                }
                sleep(Duration::from_millis(50)).await;
            }
        })
        .await
        .expect("timed out waiting for kept connection");
        let kept_stable_id = kept_connection.stable_id() as u64;

        let remote_node_id = endpoint_a_id.to_string();
        let local_node_id = endpoint_b_id.to_string();
        let connection_id = Client::deterministic_connection_id(&local_node_id, &remote_node_id);
        let synthetic_closed_stable_id = kept_stable_id.saturating_add(100);

        client_b
            .connection_manager
            .upsert_pending(
                connection_id.clone(),
                Some(remote_node_id.clone()),
                Some("device-a".to_string()),
                Some(remote_node_id.clone()),
            )
            .await;
        client_b
            .connection_manager
            .set_connected_with_transport(
                &connection_id,
                Some(remote_node_id.clone()),
                Some(synthetic_closed_stable_id),
                Some("incoming".to_string()),
            )
            .await
            .expect("connected record");
        client_b
            .connection_manager
            .set_device_id(&connection_id, "device-a".to_string())
            .await
            .expect("device binding");

        let resolution = client_b
            .reconcile_incoming_transport_after_local_close(
                &connection_id,
                endpoint_a_id,
                &remote_node_id,
                synthetic_closed_stable_id,
                "LocallyClosed",
            )
            .await;

        assert_eq!(
            resolution,
            IncomingTransportCloseResolution::PreservedKeptTransport
        );

        let settled = client_b
            .wait_for_settled_peer("device-a", Some(2_000))
            .await
            .expect("settled peer snapshot");
        assert!(
            settled.settled_ready,
            "expected recovered peer to be settled"
        );
        assert_eq!(
            settled.active_connection_id.as_deref(),
            Some(connection_id.as_str())
        );

        let record = client_b
            .connection_manager
            .get_by_connection_id(&connection_id)
            .await
            .expect("record");
        assert_eq!(record.transport_stable_id, Some(kept_stable_id));

        let managed_health = client_b
            .managed_connection_health(&connection_id)
            .await
            .expect("managed health snapshot");
        assert_eq!(
            managed_health.status,
            ManagedConnectionHealthStatus::Healthy
        );
        assert!(!managed_health.replacement_pending);
        assert_eq!(
            managed_health.active_transport_stable_id,
            Some(kept_stable_id)
        );
        assert_eq!(managed_health.readiness_state, ReadinessState::Routable);

        assert_eq!(
            client_b
                .managed_connection_bridge_action(&connection_id)
                .await,
            Some(ManagedConnectionBridgeAction::Healthy)
        );

        let health = client_b
            .connection_manager
            .peer_snapshot("device-a")
            .await
            .expect("peer snapshot");
        assert_eq!(health.health, ConnectionHealth::Healthy);
    }

    #[tokio::test]
    async fn identity_change_retires_conflicting_live_native_record() {
        let client_a = Client::new_with_app_tag(
            crate::test_constants::TEST_PROJECT_ID.to_string(),
            "identity-old".to_string(),
            Box::new(|| None),
        );
        let client_b = Client::new_with_app_tag(
            crate::test_constants::TEST_PROJECT_ID.to_string(),
            "identity-target".to_string(),
            Box::new(|| None),
        );

        let endpoint_a = Endpoint::builder(iroh::endpoint::presets::N0)
            .alpns(vec![crate::native_node::PlutoniumProtocol::ALPN.to_vec()])
            .relay_mode(RelayMode::Disabled)
            .bind_addr(
                "127.0.0.1:0"
                    .parse::<std::net::SocketAddr>()
                    .expect("socket addr"),
            )
            .expect("bind addr a")
            .bind()
            .await
            .expect("bind endpoint a");
        let endpoint_b = Endpoint::builder(iroh::endpoint::presets::N0)
            .alpns(vec![crate::native_node::PlutoniumProtocol::ALPN.to_vec()])
            .relay_mode(RelayMode::Disabled)
            .bind_addr(
                "127.0.0.1:0"
                    .parse::<std::net::SocketAddr>()
                    .expect("socket addr"),
            )
            .expect("bind addr b")
            .bind()
            .await
            .expect("bind endpoint b");

        client_a.adopt_endpoint(endpoint_a).await;
        client_b.adopt_endpoint(endpoint_b).await;

        let endpoint_a = client_a.get_endpoint().await.expect("endpoint a");
        let endpoint_b = client_b.get_endpoint().await.expect("endpoint b");
        let endpoint_a_id = endpoint_a.id();
        let endpoint_b_id = endpoint_b.id();
        let endpoint_b_addr = wait_for_endpoint_addr(&endpoint_b, 5_000)
            .await
            .expect("endpoint b addr");

        client_a
            .ensure_connected_addr(endpoint_b_id, endpoint_b_addr)
            .await
            .expect("initial connection");

        let kept_connection = tokio::time::timeout(Duration::from_secs(5), async {
            loop {
                if let Some(connection) = client_b.get_connection(endpoint_a_id).await {
                    break connection;
                }
                sleep(Duration::from_millis(50)).await;
            }
        })
        .await
        .expect("timed out waiting for target connection");

        let old_remote_node_id = endpoint_a_id.to_string();
        let local_node_id = endpoint_b_id.to_string();
        let connection_id =
            Client::deterministic_connection_id(&local_node_id, &old_remote_node_id);

        client_b
            .connection_manager
            .upsert_pending(
                connection_id.clone(),
                Some(old_remote_node_id.clone()),
                Some("device-a".to_string()),
                Some(old_remote_node_id.clone()),
            )
            .await;
        client_b
            .connection_manager
            .set_connected_with_transport(
                &connection_id,
                Some(old_remote_node_id.clone()),
                Some(kept_connection.stable_id() as u64),
                Some("incoming".to_string()),
            )
            .await
            .expect("connected record");
        client_b
            .connection_manager
            .set_device_id(&connection_id, "device-a".to_string())
            .await
            .expect("device binding");

        client_b
            .retire_conflicting_device_node_records("device-a", "node-new-identity")
            .await;

        tokio::time::timeout(Duration::from_secs(5), async {
            loop {
                if client_b
                    .connection_manager
                    .get_by_connection_id(&connection_id)
                    .await
                    .is_none()
                {
                    break;
                }
                sleep(Duration::from_millis(50)).await;
            }
        })
        .await
        .expect("timed out waiting for stale record retirement");

        tokio::time::timeout(Duration::from_secs(5), async {
            loop {
                if !client_b.is_connected(endpoint_a_id).await {
                    break;
                }
                sleep(Duration::from_millis(50)).await;
            }
        })
        .await
        .expect("timed out waiting for old live transport disconnect");
    }

    #[tokio::test]
    async fn device_status_snapshots_mark_connected_healthy_peer_as_settled_ready() {
        let client = Client::new_with_app_tag(
            crate::test_constants::TEST_PROJECT_ID.to_string(),
            "test-app".to_string(),
            Box::new(|| None),
        );

        let remote_device = crate::signaling::Device {
            app_tag: Some("test-app".to_string()),
            device_id: "device-remote-1".to_string(),
            user_id: Some("user-1".to_string()),
            device_name: "Remote Device".to_string(),
            platform_type: Some("web".to_string()),
            capabilities: None,
            session_id: None,
            node_id: Some("node-remote-1".to_string()),
            online: true,
            ticket: Some("ticket-1".to_string()),
            tag: Some("test-app".to_string()),
            kind: Some("device".to_string()),
            metadata: None,
            last_seen_at: None,
            expires_at: None,
            created_at: None,
            updated_at: None,
            excluded_peers: vec![],
        };
        client
            .connection_manager
            .upsert_pending(
                "conn-remote-1".to_string(),
                Some("node-remote-1".to_string()),
                Some(remote_device.device_id.clone()),
                Some("node-remote-1".to_string()),
            )
            .await;
        client
            .connection_manager
            .set_connected("conn-remote-1", Some("node-remote-1".to_string()))
            .await;
        client
            .connection_manager
            .set_health(
                "conn-remote-1",
                crate::connection_manager::ConnectionHealth::Healthy,
            )
            .await;

        let snapshots = merge_device_status_snapshots(
            vec![remote_device],
            client.connection_manager.list_peer_snapshots().await,
        );
        let snapshot = snapshots.first().expect("snapshot");
        assert_eq!(
            snapshot.connection_status,
            DeviceConnectionStatus::Connected
        );
        assert!(snapshot.settled_ready);
    }

    #[tokio::test]
    async fn device_status_snapshots_match_pending_native_peer_by_node_id_before_device_binding() {
        let client = Client::new_with_app_tag(
            crate::test_constants::TEST_PROJECT_ID.to_string(),
            "test-app".to_string(),
            Box::new(|| None),
        );

        let remote_device = crate::signaling::Device {
            app_tag: Some("test-app".to_string()),
            device_id: "device-remote-node-alias-1".to_string(),
            user_id: Some("user-1".to_string()),
            device_name: "Remote Device".to_string(),
            platform_type: Some("desktop".to_string()),
            capabilities: None,
            session_id: None,
            node_id: Some("node-remote-node-alias-1".to_string()),
            online: true,
            ticket: Some("ticket-node-alias-1".to_string()),
            tag: Some("test-app".to_string()),
            kind: Some("device".to_string()),
            metadata: None,
            last_seen_at: None,
            expires_at: None,
            created_at: None,
            updated_at: None,
            excluded_peers: vec![],
        };

        client
            .connection_manager
            .upsert_pending(
                "conn-node-alias-1".to_string(),
                Some("node-remote-node-alias-1".to_string()),
                None,
                Some("node-remote-node-alias-1".to_string()),
            )
            .await;
        client
            .connection_manager
            .set_connected(
                "conn-node-alias-1",
                Some("node-remote-node-alias-1".to_string()),
            )
            .await;
        client
            .connection_manager
            .set_health(
                "conn-node-alias-1",
                crate::connection_manager::ConnectionHealth::Healthy,
            )
            .await;

        let snapshots = merge_device_status_snapshots(
            vec![remote_device],
            client.connection_manager.list_peer_snapshots().await,
        );
        let snapshot = snapshots.first().expect("snapshot");
        assert_eq!(
            snapshot.connection_status,
            DeviceConnectionStatus::Connected
        );
        assert!(snapshot.settled_ready);
        assert_eq!(snapshot.connection_id.as_deref(), Some("conn-node-alias-1"));
    }

    #[tokio::test]
    async fn peer_session_snapshot_reports_active_and_candidate_connections() {
        let client = Client::new_with_app_tag(
            crate::test_constants::TEST_PROJECT_ID.to_string(),
            "test-app".to_string(),
            Box::new(|| None),
        );

        client
            .connection_manager
            .upsert_pending(
                "conn-1".to_string(),
                Some("node-1".to_string()),
                Some("device-1".to_string()),
                Some("node-1".to_string()),
            )
            .await;
        client
            .connection_manager
            .set_connected("conn-1", Some("node-1".to_string()))
            .await;
        client
            .connection_manager
            .set_health(
                "conn-1",
                crate::connection_manager::ConnectionHealth::Healthy,
            )
            .await;
        client
            .connection_manager
            .upsert_pending(
                "conn-2".to_string(),
                Some("node-2".to_string()),
                Some("device-1".to_string()),
                Some("node-2".to_string()),
            )
            .await;

        let snapshot = client
            .peer_session("device-1")
            .await
            .expect("peer session snapshot");
        assert_eq!(snapshot.peer_id, "device-1");
        assert_eq!(snapshot.active_connection_id.as_deref(), Some("conn-1"));
        assert_eq!(
            snapshot.candidate_connection_ids,
            vec!["conn-2".to_string()]
        );
        assert!(snapshot.settled_ready);
    }

    #[tokio::test]
    async fn wait_for_settled_peer_returns_settled_snapshot() {
        let client = std::sync::Arc::new(Client::new_with_app_tag(
            crate::test_constants::TEST_PROJECT_ID.to_string(),
            "test-app".to_string(),
            Box::new(|| None),
        ));

        let client_clone = client.clone();
        tokio::spawn(async move {
            tokio::time::sleep(std::time::Duration::from_millis(50)).await;
            client_clone
                .connection_manager
                .upsert_pending(
                    "conn-wait-1".to_string(),
                    Some("node-wait-1".to_string()),
                    Some("device-wait-1".to_string()),
                    Some("node-wait-1".to_string()),
                )
                .await;
            client_clone
                .connection_manager
                .set_connected("conn-wait-1", Some("node-wait-1".to_string()))
                .await;
            client_clone
                .connection_manager
                .set_health(
                    "conn-wait-1",
                    crate::connection_manager::ConnectionHealth::Healthy,
                )
                .await;
        });

        let snapshot = client
            .wait_for_settled_peer("device-wait-1", Some(1_000))
            .await
            .expect("settled snapshot");
        assert_eq!(snapshot.peer_id, "device-wait-1");
        assert!(snapshot.settled_ready);
    }

    #[tokio::test]
    async fn peer_session_adopts_live_transport_from_shared_endpoint_by_node_id() {
        let client_a = Client::new_with_app_tag(
            crate::test_constants::TEST_PROJECT_ID.to_string(),
            "test-app".to_string(),
            Box::new(|| None),
        );
        let client_b = Client::new_with_app_tag(
            crate::test_constants::TEST_PROJECT_ID.to_string(),
            "test-app".to_string(),
            Box::new(|| None),
        );

        let endpoint_a = Endpoint::builder(iroh::endpoint::presets::N0)
            .alpns(vec![crate::native_node::PlutoniumProtocol::ALPN.to_vec()])
            .relay_mode(RelayMode::Disabled)
            .bind_addr(
                "127.0.0.1:0"
                    .parse::<std::net::SocketAddr>()
                    .expect("socket addr"),
            )
            .expect("bind addr a")
            .bind()
            .await
            .expect("bind endpoint a");
        let endpoint_b = Endpoint::builder(iroh::endpoint::presets::N0)
            .alpns(vec![crate::native_node::PlutoniumProtocol::ALPN.to_vec()])
            .relay_mode(RelayMode::Disabled)
            .bind_addr(
                "127.0.0.1:0"
                    .parse::<std::net::SocketAddr>()
                    .expect("socket addr"),
            )
            .expect("bind addr b")
            .bind()
            .await
            .expect("bind endpoint b");

        client_a.adopt_endpoint(endpoint_a).await;
        client_b.adopt_endpoint(endpoint_b).await;

        let endpoint_a = client_a.get_endpoint().await.expect("endpoint a");
        let endpoint_b = client_b.get_endpoint().await.expect("endpoint b");
        let endpoint_a_id = endpoint_a.id();
        let endpoint_b_id = endpoint_b.id();
        let endpoint_b_addr = wait_for_endpoint_addr(&endpoint_b, 5_000)
            .await
            .expect("endpoint b addr");

        client_a
            .ensure_connected_addr(endpoint_b_id, endpoint_b_addr)
            .await
            .expect("initial connection");

        tokio::time::timeout(Duration::from_secs(5), async {
            loop {
                if client_b.get_connection(endpoint_a_id).await.is_some() {
                    break;
                }
                sleep(Duration::from_millis(50)).await;
            }
        })
        .await
        .expect("timed out waiting for live transport");

        assert!(
            client_b.connection_manager.list_all().await.is_empty(),
            "client should start with a live transport but no managed connection record"
        );

        let remote_device = Device {
            app_tag: Some("test-app".to_string()),
            device_id: "device-status-live-transport".to_string(),
            user_id: Some("user-1".to_string()),
            device_name: "Live Transport Device".to_string(),
            platform_type: Some("desktop".to_string()),
            capabilities: None,
            session_id: None,
            node_id: Some(endpoint_a_id.to_string()),
            online: true,
            ticket: Some("ticket-live-transport".to_string()),
            tag: Some("test-app".to_string()),
            kind: Some("device".to_string()),
            metadata: None,
            last_seen_at: None,
            expires_at: None,
            created_at: None,
            updated_at: None,
            excluded_peers: vec![],
        };
        client_b
            .refresh_device_status_peer_snapshots_from_devices(&[remote_device.clone()])
            .await;
        let peers = client_b.connection_manager.list_peer_snapshots().await;
        client_b
            .backfill_peer_device_ids_from_devices(&[remote_device.clone()], &peers)
            .await;
        let status_snapshots = merge_device_status_snapshots(
            vec![remote_device],
            client_b.connection_manager.list_peer_snapshots().await,
        );
        let status = status_snapshots
            .first()
            .expect("device status snapshot adopted live transport");
        assert_eq!(status.connection_status, DeviceConnectionStatus::Connected);
        assert!(
            status.settled_ready,
            "device status should expose the adopted live transport as settled"
        );
        assert_eq!(status.device.device_id, "device-status-live-transport");
        assert!(status.connection_id.is_some());

        let endpoint_lookup = client_b
            .peer_session(&endpoint_a_id.to_string())
            .await
            .expect("peer session adopted from live transport");
        let endpoint_a_id_string = endpoint_a_id.to_string();
        let connection_id =
            Client::deterministic_connection_id(&endpoint_b_id.to_string(), &endpoint_a_id_string);

        assert_eq!(
            endpoint_lookup.node_id.as_deref(),
            Some(endpoint_a_id_string.as_str())
        );
        assert_eq!(
            endpoint_lookup.active_connection_id.as_deref(),
            Some(connection_id.as_str())
        );
        assert_eq!(client_b.connection_manager.list_all().await.len(), 1);

        let settled = client_b
            .wait_for_settled_peer(&connection_id, Some(2_000))
            .await
            .expect("settled snapshot from adopted live transport");
        assert!(settled.settled_ready);
        assert_eq!(
            settled.active_connection_id.as_deref(),
            Some(connection_id.as_str())
        );

        client_b
            .connection_manager
            .set_connecting(&connection_id)
            .await;
        let stale = client_b
            .connection_manager
            .peer_snapshot(&connection_id)
            .await
            .expect("stale cached snapshot");
        assert_eq!(
            stale.status,
            crate::connection_manager::ConnectionState::Connecting
        );

        let rebound = client_b
            .peer_session(&connection_id)
            .await
            .expect("cached peer session rebound from live transport");
        assert!(rebound.settled_ready);
        assert!(rebound.active_transport_stable_id.is_some());
    }

    #[tokio::test]
    async fn open_peer_bi_rejects_unsettled_peer_before_transport_lookup() {
        let client = Client::new_with_app_tag(
            crate::test_constants::TEST_PROJECT_ID.to_string(),
            "test-app".to_string(),
            Box::new(|| None),
        );

        client
            .connection_manager
            .upsert_pending(
                "conn-open-1".to_string(),
                Some("node-open-1".to_string()),
                Some("device-open-1".to_string()),
                Some("node-open-1".to_string()),
            )
            .await;
        client
            .connection_manager
            .set_connected("conn-open-1", Some("node-open-1".to_string()))
            .await;

        let error = client
            .open_peer_bi("device-open-1", Some(10))
            .await
            .expect_err("unsettled peer should fail before transport open");

        assert!(
            error.to_string().contains("not settled"),
            "unexpected error: {}",
            error
        );
    }

    #[tokio::test]
    async fn open_peer_bi_transport_only_does_not_require_settled_readiness() {
        let client = Client::new_with_app_tag(
            crate::test_constants::TEST_PROJECT_ID.to_string(),
            "test-app".to_string(),
            Box::new(|| None),
        );

        client
            .connection_manager
            .upsert_pending(
                "conn-transport-only-1".to_string(),
                Some("node-transport-only-1".to_string()),
                Some("device-transport-only-1".to_string()),
                Some("node-transport-only-1".to_string()),
            )
            .await;
        client
            .connection_manager
            .set_connected(
                "conn-transport-only-1",
                Some("node-transport-only-1".to_string()),
            )
            .await;

        let error = client
            .open_peer_bi_transport_only("device-transport-only-1", Some(10))
            .await
            .expect_err("missing live transport should still fail");

        assert!(
            !error.to_string().contains("not settled"),
            "transport-only open must not reuse strict settled readiness errors: {}",
            error
        );
    }

    #[tokio::test]
    async fn open_peer_bi_rejects_stale_settled_peer_without_active_transport() {
        let client = Client::new_with_app_tag(
            crate::test_constants::TEST_PROJECT_ID.to_string(),
            "test-app".to_string(),
            Box::new(|| None),
        );

        let remote_node_id = "844465046e89fce03a8f96df9e289ab1a4336f05d1d4ef67f2d4598dd7ad0a2a";

        client
            .connection_manager
            .upsert_pending(
                "conn-open-2".to_string(),
                Some(remote_node_id.to_string()),
                Some("device-open-2".to_string()),
                Some(remote_node_id.to_string()),
            )
            .await;
        client
            .connection_manager
            .set_connected("conn-open-2", Some(remote_node_id.to_string()))
            .await;
        client
            .connection_manager
            .set_health(
                "conn-open-2",
                crate::connection_manager::ConnectionHealth::Healthy,
            )
            .await;

        let error = client
            .open_peer_bi("device-open-2", Some(10))
            .await
            .expect_err("stale settled peer should fail before raw transport open");

        assert!(
            error.to_string().contains("no active transport")
                || error.to_string().contains("not settled")
                || error.to_string().contains("not found"),
            "unexpected error: {}",
            error
        );
    }

    #[tokio::test]
    async fn refresh_cached_peer_snapshot_preserves_settled_webrtc_peer_when_iroh_is_gone() {
        let client = Client::new_with_app_tag(
            crate::test_constants::TEST_PROJECT_ID.to_string(),
            "test-app".to_string(),
            Box::new(|| None),
        );

        let remote_node_id = "844465046e89fce03a8f96df9e289ab1a4336f05d1d4ef67f2d4598dd7ad0a2a";

        client
            .connection_manager
            .upsert_pending(
                "conn-open-webrtc".to_string(),
                Some(remote_node_id.to_string()),
                Some("device-open-webrtc".to_string()),
                Some(remote_node_id.to_string()),
            )
            .await;
        client
            .connection_manager
            .set_connected_with_transport(
                "conn-open-webrtc",
                Some(remote_node_id.to_string()),
                Some(77),
                Some("incoming".to_string()),
            )
            .await
            .expect("connected record");
        client
            .connection_manager
            .set_health(
                "conn-open-webrtc",
                crate::connection_manager::ConnectionHealth::Healthy,
            )
            .await;
        client
            .connection_manager
            .report_transport_status(
                "conn-open-webrtc",
                "webrtc".to_string(),
                Some("iroh-relay".to_string()),
            )
            .await
            .expect("peer snapshot");

        let snapshot = client
            .peer_snapshot("conn-open-webrtc")
            .await
            .expect("snapshot should remain");

        assert_eq!(snapshot.active_transport, "webrtc");
        assert_eq!(snapshot.parallel_transport.as_deref(), Some("iroh-relay"));
        assert!(
            client
                .connection_manager
                .get_by_connection_id("conn-open-webrtc")
                .await
                .is_some(),
            "settled upgraded peer should not be retired just because iroh rotated",
        );
    }

    #[tokio::test]
    async fn inspect_incoming_native_main_frame_parses_native_handshake() {
        let client = Client::new_with_app_tag(
            crate::test_constants::TEST_PROJECT_ID.to_string(),
            "test-app".to_string(),
            Box::new(|| None),
        );

        let frame = serde_json::json!({
            "channel": "handshake",
            "action": "request",
            "payload": {
                "deviceId": "c7d6b730-ff20-4cc3-bfdf-c2606f2a70ef",
            },
            "timestamp": 123,
        })
        .to_string()
        .into_bytes();

        let inspected = client
            .inspect_incoming_native_main_frame("conn-native-1", Some("node-1"), None, &frame)
            .await
            .expect("native handshake should parse");

        match inspected {
            crate::native_protocol::InspectedMainFrame::NativeMessage { message, handshake } => {
                assert_eq!(message.channel, "handshake");
                assert_eq!(
                    handshake
                        .as_ref()
                        .and_then(|value| value.claimed_device_id.as_deref()),
                    Some("c7d6b730-ff20-4cc3-bfdf-c2606f2a70ef")
                );
                assert_eq!(handshake.and_then(|value| value.admitted_device_id), None);
            }
            other => panic!("expected native message, got {:?}", other),
        }
    }

    #[tokio::test]
    async fn inspect_incoming_native_main_frame_rejects_missing_ts_token_when_registry_active() {
        let client = Client::new_with_app_tag(
            crate::test_constants::TEST_PROJECT_ID.to_string(),
            "test-app".to_string(),
            Box::new(|| None),
        );

        client.register_session_token("token-1".into(), "share".into(), 0);

        let json = serde_json::json!({
            "type": "handshake",
            "action": "hello",
        });
        let mut frame = vec![0x00];
        frame.extend_from_slice(json.to_string().as_bytes());

        let error = client
            .inspect_incoming_native_main_frame("conn-ts-1", Some("node-1"), None, &frame)
            .await
            .expect_err("missing token should be rejected");
        assert_eq!(error, "session-token-required");
    }

    #[tokio::test]
    async fn managed_connection_adoption_returns_identity_aware_snapshot() {
        let client = Client::new_with_app_tag(
            crate::test_constants::TEST_PROJECT_ID.to_string(),
            "test-app".to_string(),
            Box::new(|| None),
        );

        client
            .connection_manager
            .upsert_pending(
                "conn-adopt-1".to_string(),
                Some("node-adopt-1".to_string()),
                Some("device-adopt-1".to_string()),
                Some("node-adopt-1".to_string()),
            )
            .await;
        client
            .connection_manager
            .set_connected_with_transport(
                "conn-adopt-1",
                Some("node-adopt-1".to_string()),
                Some(22),
                Some("managed".to_string()),
            )
            .await
            .expect("managed record");
        client
            .connection_manager
            .set_device_id("conn-adopt-1", "device-adopt-1".to_string())
            .await
            .expect("managed device binding");

        let record = client
            .connection_manager
            .get_by_connection_id("conn-adopt-1")
            .await
            .expect("record exists");
        let adoption = client
            .managed_connection_adoption(&record)
            .await
            .expect("adoption snapshot");

        assert_eq!(adoption.connection_id, "conn-adopt-1");
        assert_eq!(adoption.node_id, "node-adopt-1");
        assert_eq!(adoption.device_id.as_deref(), Some("device-adopt-1"));
        assert_eq!(adoption.transport_generation, 1);
        assert!(adoption.main_stream_ready);
    }

    #[tokio::test]
    async fn inspect_ts_handshake_with_valid_token_succeeds() {
        let client = Client::new_with_app_tag(
            crate::test_constants::TEST_PROJECT_ID.to_string(),
            "test-app".to_string(),
            Box::new(|| None),
        );

        client.register_session_token("token-valid".into(), "share".into(), 1);

        let json = serde_json::json!({
            "type": "handshake",
            "action": "hello",
            "sessionToken": "token-valid",
        });
        let mut frame = vec![0x00];
        frame.extend_from_slice(json.to_string().as_bytes());

        let result = client
            .inspect_incoming_native_main_frame("conn-ts-valid", Some("node-1"), None, &frame)
            .await
            .expect("valid token should be accepted");

        assert!(matches!(
            result,
            crate::native_protocol::InspectedMainFrame::ForwardOpaque
        ));
    }

    #[tokio::test]
    async fn inspect_ts_capability_update_after_admission_does_not_require_token() {
        let client = Client::new_with_app_tag(
            crate::test_constants::TEST_PROJECT_ID.to_string(),
            "test-app".to_string(),
            Box::new(|| None),
        );

        client.register_session_token("token-valid".into(), "share".into(), 1);

        let hello = serde_json::json!({
            "type": "handshake",
            "action": "hello",
            "sessionToken": "token-valid",
        });
        let mut hello_frame = vec![0x00];
        hello_frame.extend_from_slice(hello.to_string().as_bytes());
        client
            .inspect_incoming_native_main_frame(
                "conn-ts-update",
                Some("node-1"),
                None,
                &hello_frame,
            )
            .await
            .expect("hello should admit the connection");

        let update = serde_json::json!({
            "type": "handshake",
            "action": "capability-update",
            "capabilities": {
                "webrtc": true,
            },
        });
        let mut update_frame = vec![0x00];
        update_frame.extend_from_slice(update.to_string().as_bytes());

        let result = client
            .inspect_incoming_native_main_frame(
                "conn-ts-update",
                Some("node-1"),
                None,
                &update_frame,
            )
            .await
            .expect("capability-update should reuse existing admission");

        assert!(matches!(
            result,
            crate::native_protocol::InspectedMainFrame::ForwardOpaque
        ));
    }

    #[tokio::test]
    async fn inspect_opaque_frame_rejected_when_registry_active_and_not_admitted() {
        let client = Client::new_with_app_tag(
            crate::test_constants::TEST_PROJECT_ID.to_string(),
            "test-app".to_string(),
            Box::new(|| None),
        );

        client.register_session_token("token-1".into(), "share".into(), 0);

        let error = client
            .inspect_incoming_native_main_frame(
                "conn-opaque-1",
                Some("node-1"),
                None,
                b"\x01\x02\x03",
            )
            .await
            .expect_err("opaque frame should be rejected when not admitted");
        assert_eq!(error, "session-token-required");
    }

    #[tokio::test]
    async fn inspect_opaque_frame_allowed_when_registry_inactive() {
        let client = Client::new_with_app_tag(
            crate::test_constants::TEST_PROJECT_ID.to_string(),
            "test-app".to_string(),
            Box::new(|| None),
        );

        let result = client
            .inspect_incoming_native_main_frame(
                "conn-opaque-2",
                Some("node-1"),
                None,
                b"\x01\x02\x03",
            )
            .await
            .expect("opaque frame should pass when no registry");

        assert!(matches!(
            result,
            crate::native_protocol::InspectedMainFrame::ForwardOpaque
        ));
    }

    #[tokio::test]
    async fn non_handshake_before_handshake_does_not_permanently_reject() {
        let client = Client::new_with_app_tag(
            crate::test_constants::TEST_PROJECT_ID.to_string(),
            "test-app".to_string(),
            Box::new(|| None),
        );

        client.register_session_token("token-1".into(), "share".into(), 1);

        // Send a non-handshake native message first — should fail transiently
        let event_frame = serde_json::json!({
            "channel": "status",
            "action": "event",
            "payload": {},
            "timestamp": 100,
        })
        .to_string()
        .into_bytes();

        let err = client
            .inspect_incoming_native_main_frame("conn-race-1", Some("node-1"), None, &event_frame)
            .await
            .expect_err("non-handshake should be rejected when pending");
        assert_eq!(err, "session-token-required");

        // Now send a handshake with a valid token — should NOT be permanently rejected
        let _handshake_frame = serde_json::json!({
            "channel": "handshake",
            "action": "request",
            "payload": { "deviceId": "device-race-1" },
            "timestamp": 101,
        })
        .to_string()
        .into_bytes();

        // The handshake itself won't have a session token (native path uses
        // trusted verifier), so it will still be rejected without a verifier.
        // But the key assertion is that admission is still Pending, not
        // permanently Rejected from the earlier event message.
        let admission = client.session_admission("conn-race-1");
        assert!(
            matches!(admission, crate::session_token::SessionAdmission::Pending),
            "admission should still be Pending after transient rejection, got {:?}",
            admission
        );
    }

    #[tokio::test]
    async fn non_handshake_binding_not_attached_to_non_handshake_messages() {
        let client = Client::new_with_app_tag(
            crate::test_constants::TEST_PROJECT_ID.to_string(),
            "test-app".to_string(),
            Box::new(|| None),
        );

        // No registry active — just test that binding is only on handshakes
        let event_frame = serde_json::json!({
            "channel": "status",
            "action": "event",
            "payload": {},
            "timestamp": 100,
        })
        .to_string()
        .into_bytes();

        let result = client
            .inspect_incoming_native_main_frame(
                "conn-no-binding",
                Some("node-1"),
                Some("known-device-1"),
                &event_frame,
            )
            .await
            .expect("should parse");

        match result {
            crate::native_protocol::InspectedMainFrame::NativeMessage { handshake, .. } => {
                assert!(
                    handshake.is_none(),
                    "non-handshake messages should not carry a NativeHandshakeBinding"
                );
            }
            other => panic!("expected NativeMessage, got {:?}", other),
        }
    }

    #[tokio::test]
    async fn ensure_native_stream_admitted_does_not_persist_rejection() {
        let client = Client::new_with_app_tag(
            crate::test_constants::TEST_PROJECT_ID.to_string(),
            "test-app".to_string(),
            Box::new(|| None),
        );

        client.register_session_token("token-1".into(), "share".into(), 0);

        // Stream admission fails transiently
        let err = client
            .ensure_native_stream_admitted("conn-stream-1", Some("node-1"), None)
            .expect_err("should reject unadmitted stream");
        assert_eq!(err, "session-token-required");

        // Admission should still be Pending, not permanently Rejected
        let admission = client.session_admission("conn-stream-1");
        assert!(
            matches!(admission, crate::session_token::SessionAdmission::Pending),
            "stream rejection should not persist, got {:?}",
            admission
        );
    }

    #[tokio::test]
    async fn inspect_session_token_presentation_validates_and_admits() {
        let client = Client::new_with_app_tag(
            crate::test_constants::TEST_PROJECT_ID.to_string(),
            "test-app".to_string(),
            Box::new(|| None),
        );

        client.register_session_token("share-token-1".into(), "share".into(), 1);

        // Build a session-token presentation frame
        let msg =
            crate::native_protocol::NativeMainMessage::session_token_presentation("share-token-1");
        let frame = serde_json::to_vec(&msg).unwrap();

        let result = client
            .inspect_incoming_native_main_frame("conn-token-1", Some("node-1"), None, &frame)
            .await
            .expect("valid token presentation should succeed");

        // Should return ForwardOpaque (SDK-internal, not routed to app)
        assert!(
            matches!(
                result,
                crate::native_protocol::InspectedMainFrame::ForwardOpaque
            ),
            "session-token presentation should return ForwardOpaque"
        );

        // Connection should now be admitted
        let admission = client.session_admission("conn-token-1");
        assert!(
            matches!(
                admission,
                crate::session_token::SessionAdmission::Accepted { .. }
            ),
            "connection should be admitted after token presentation, got {:?}",
            admission
        );
    }

    #[tokio::test]
    async fn inspect_session_token_presentation_rejects_invalid_token() {
        let client = Client::new_with_app_tag(
            crate::test_constants::TEST_PROJECT_ID.to_string(),
            "test-app".to_string(),
            Box::new(|| None),
        );

        client.register_session_token("real-token".into(), "share".into(), 1);

        let msg =
            crate::native_protocol::NativeMainMessage::session_token_presentation("wrong-token");
        let frame = serde_json::to_vec(&msg).unwrap();

        let err = client
            .inspect_incoming_native_main_frame("conn-bad-token", Some("node-1"), None, &frame)
            .await
            .expect_err("invalid token should be rejected");
        assert!(
            err.contains("invalid") || err.contains("token"),
            "error should mention token: {}",
            err
        );
    }

    #[tokio::test]
    async fn inspect_session_token_presentation_ignored_when_registry_inactive() {
        let client = Client::new_with_app_tag(
            crate::test_constants::TEST_PROJECT_ID.to_string(),
            "test-app".to_string(),
            Box::new(|| None),
        );

        // No tokens registered — registry inactive
        let msg =
            crate::native_protocol::NativeMainMessage::session_token_presentation("some-token");
        let frame = serde_json::to_vec(&msg).unwrap();

        let result = client
            .inspect_incoming_native_main_frame("conn-no-reg", Some("node-1"), None, &frame)
            .await
            .expect("should pass when registry inactive");

        // When registry is inactive, session-token message is treated as a
        // regular native message (not intercepted), forwarded as NativeMessage.
        assert!(matches!(
            result,
            crate::native_protocol::InspectedMainFrame::NativeMessage { .. }
        ));
    }

    #[tokio::test]
    async fn native_user_device_compound_ticket_connection_allows_token_then_handshake() {
        let host = Client::new_with_app_tag(
            crate::test_constants::TEST_PROJECT_ID.to_string(),
            "test-app".to_string(),
            Box::new(|| None),
        );
        host.register_session_token("user-device-token".into(), "user-device".into(), 0);

        let connection_id = "conn-user-device-sequence";
        let peer_node_id = "peer-node-1";
        let token_message = crate::native_protocol::NativeMainMessage::session_token_presentation(
            "user-device-token",
        );
        let token_frame = serde_json::to_vec(&token_message).expect("serialize token message");

        let token_result = host
            .inspect_incoming_native_main_frame(
                connection_id,
                Some(peer_node_id),
                None,
                &token_frame,
            )
            .await
            .expect("token presentation should be accepted");
        assert!(matches!(token_result, InspectedMainFrame::ForwardOpaque));
        assert!(matches!(
            host.session_admission(connection_id),
            crate::session_token::SessionAdmission::Accepted { .. }
        ));

        let handshake = NativeMainMessage {
            channel: "handshake".to_string(),
            action: NativeMessageAction::Request,
            request_id: None,
            payload: serde_json::json!({ "deviceId": "device-native-token-test" }),
            timestamp: 123,
            from: None,
        };
        let handshake_frame = serde_json::to_vec(&handshake).expect("serialize handshake");

        let handshake_result = host
            .inspect_incoming_native_main_frame(
                connection_id,
                Some(peer_node_id),
                None,
                &handshake_frame,
            )
            .await
            .expect("handshake after token should be accepted");
        match handshake_result {
            InspectedMainFrame::NativeMessage { message, .. } => {
                assert!(message.is_handshake());
                assert_eq!(
                    message.claimed_device_id().as_deref(),
                    Some("device-native-token-test")
                );
            }
            other => panic!("expected native handshake message, got {:?}", other),
        }
    }

    #[tokio::test]
    async fn default_admission_gate_requires_auth_from_startup() {
        let client = Client::new_with_app_tag(
            crate::test_constants::TEST_PROJECT_ID.to_string(),
            "test-app".to_string(),
            Box::new(|| None),
        );

        client.ensure_default_admission_gate("__startup");
        assert!(client.session_registry_active());

        let handshake = serde_json::json!({
            "type": "handshake",
            "action": "hello"
        });
        let mut frame = vec![0x00];
        frame.extend_from_slice(handshake.to_string().as_bytes());

        let err = client
            .inspect_incoming_native_main_frame("conn-startup-gate", Some("node-1"), None, &frame)
            .await
            .expect_err("startup gate should reject tokenless handshake");
        assert_eq!(err, "session-token-required");
    }

    #[tokio::test]
    async fn managed_adoption_blocked_when_registry_active_and_not_admitted() {
        let client = Client::new_with_app_tag(
            crate::test_constants::TEST_PROJECT_ID.to_string(),
            "test-app".to_string(),
            Box::new(|| None),
        );

        // Register a token so the registry is active
        client.register_session_token("token-1".into(), "user-device".into(), 0);

        // Create a connected managed record
        client
            .connection_manager
            .upsert_pending(
                "conn-blocked-1".to_string(),
                Some("node-blocked-1".to_string()),
                None,
                Some("node-blocked-1".to_string()),
            )
            .await;
        client
            .connection_manager
            .set_connected_with_transport(
                "conn-blocked-1",
                Some("node-blocked-1".to_string()),
                Some(42),
                Some("incoming".to_string()),
            )
            .await
            .expect("set connected");

        let record = client
            .connection_manager
            .get_by_connection_id("conn-blocked-1")
            .await
            .expect("record exists");

        // Adoption should be blocked because connection is not admitted
        let adoption = client.managed_connection_adoption(&record).await;
        assert!(
            adoption.is_none(),
            "adoption should be blocked when registry active and connection not admitted"
        );
    }

    #[tokio::test]
    async fn managed_adoption_allowed_after_token_admission() {
        let client = Client::new_with_app_tag(
            crate::test_constants::TEST_PROJECT_ID.to_string(),
            "test-app".to_string(),
            Box::new(|| None),
        );

        client.register_session_token("token-adopt".into(), "user-device".into(), 0);

        // Create a connected managed record with device_id
        client
            .connection_manager
            .upsert_pending(
                "conn-admitted-1".to_string(),
                Some("node-admitted-1".to_string()),
                Some("device-admitted-1".to_string()),
                Some("node-admitted-1".to_string()),
            )
            .await;
        client
            .connection_manager
            .set_connected_with_transport(
                "conn-admitted-1",
                Some("node-admitted-1".to_string()),
                Some(43),
                Some("incoming".to_string()),
            )
            .await
            .expect("set connected");
        client
            .connection_manager
            .set_device_id("conn-admitted-1", "device-admitted-1".to_string())
            .await
            .expect("set device_id");

        // Validate a token to admit the connection
        client
            .validate_session_token_for_connection("token-adopt", "conn-admitted-1")
            .await
            .expect("token should validate");

        let record = client
            .connection_manager
            .get_by_connection_id("conn-admitted-1")
            .await
            .expect("record exists");

        // Adoption should now succeed
        let adoption = client
            .managed_connection_adoption(&record)
            .await
            .expect("adoption should succeed after token admission");

        assert_eq!(adoption.connection_id, "conn-admitted-1");
        assert_eq!(adoption.node_id, "node-admitted-1");
    }

    #[tokio::test]
    async fn outgoing_host_approval_marks_connection_locally_admitted_for_adoption() {
        let client = Client::new_with_app_tag(
            crate::test_constants::TEST_PROJECT_ID.to_string(),
            "test-app".to_string(),
            Box::new(|| None),
        );

        client.register_session_token("gate-token".into(), "user-device".into(), 0);

        let connection_id = "conn-host-approved-1";
        let node_id = "node-host-approved-1";
        let device_id = "device-host-approved-1";

        client
            .connection_manager
            .upsert_pending(
                connection_id.to_string(),
                Some(node_id.to_string()),
                Some(device_id.to_string()),
                Some(node_id.to_string()),
            )
            .await;
        client
            .connection_manager
            .set_connected_with_transport(
                connection_id,
                Some(node_id.to_string()),
                Some(44),
                Some("outgoing".to_string()),
            )
            .await
            .expect("set connected");
        client
            .connection_manager
            .set_device_id(connection_id, device_id.to_string())
            .await
            .expect("set device_id");

        client
            .mark_connection_admitted_by_host(
                connection_id,
                Some("user-device"),
                Some(device_id.to_string()),
            )
            .await;

        assert!(matches!(
            client.session_admission(connection_id),
            crate::session_token::SessionAdmission::Accepted { .. }
        ));

        let record = client
            .connection_manager
            .get_by_connection_id(connection_id)
            .await
            .expect("record exists");

        let adoption = client
            .managed_connection_adoption(&record)
            .await
            .expect("host-approved outgoing connection should be adoptable");

        assert_eq!(adoption.connection_id, connection_id);
        assert_eq!(adoption.node_id, node_id);
        assert_eq!(adoption.device_id.as_deref(), Some(device_id));
        assert!(adoption.main_stream_ready);
    }

    #[cfg(all(feature = "transport-webrtc", not(target_arch = "wasm32")))]
    #[tokio::test]
    async fn accepting_replacement_peer_retires_stale_connection_for_same_device() {
        let client = Client::new_with_app_tag(
            crate::test_constants::TEST_PROJECT_ID.to_string(),
            "test-app".to_string(),
            Box::new(|| None),
        );

        client
            .connection_manager
            .upsert_pending(
                "conn-old".to_string(),
                Some("node-old".to_string()),
                Some("device-a".to_string()),
                Some("node-old".to_string()),
            )
            .await;
        client
            .connection_manager
            .set_connected_with_transport(
                "conn-old",
                Some("node-old".to_string()),
                Some(41),
                Some("incoming".to_string()),
            )
            .await
            .expect("set old connected");
        client
            .connection_manager
            .set_device_id("conn-old", "device-a".to_string())
            .await
            .expect("set old device");

        client
            .connection_manager
            .upsert_pending(
                "conn-new".to_string(),
                Some("node-new".to_string()),
                Some("device-a".to_string()),
                Some("node-new".to_string()),
            )
            .await;
        client
            .connection_manager
            .set_connected_with_transport(
                "conn-new",
                Some("node-new".to_string()),
                Some(42),
                Some("incoming".to_string()),
            )
            .await
            .expect("set new connected");
        client
            .connection_manager
            .set_device_id("conn-new", "device-a".to_string())
            .await
            .expect("set new device");

        client
            .native_webrtc_attempt_counts
            .write()
            .await
            .insert("conn-old".to_string(), 5);
        client.native_webrtc_suppressions.write().await.insert(
            "conn-old".to_string(),
            crate::client::NativeWebRTCSuppression {
                until_ms: i64::MAX,
                reason: "upgrade-attempts-exhausted".to_string(),
            },
        );

        client
            .accept_replacement_peer("conn-new", Some("device-a"), "replacement-peer-admitted")
            .await;

        assert!(client
            .connection_manager
            .get_by_connection_id("conn-old")
            .await
            .is_none());
        assert!(client
            .connection_manager
            .get_by_connection_id("conn-new")
            .await
            .is_some());
        assert!(client
            .native_webrtc_attempt_counts
            .read()
            .await
            .get("conn-old")
            .is_none());
        assert!(client
            .native_webrtc_suppressions
            .read()
            .await
            .get("conn-old")
            .is_none());
    }

    #[tokio::test]
    async fn native_user_device_ticket_cache_reuses_backend_token_within_app_session() {
        let client = Client::new_with_app_tag(
            crate::test_constants::TEST_PROJECT_ID.to_string(),
            "test-app".to_string(),
            Box::new(|| None),
        );

        let base_dir = std::env::temp_dir().join(format!(
            "openrtc-user-device-cache-{}",
            std::time::SystemTime::now()
                .duration_since(std::time::UNIX_EPOCH)
                .expect("unix epoch")
                .as_nanos()
        ));

        client
            .init_native_device_identity(base_dir, Some("Managed Cache Device"))
            .await
            .expect("init native device identity");
        client.init_iroh(None, Vec::new()).await.expect("init iroh");

        let first = client
            .endpoint_ticket_with_token("user-device", 0)
            .await
            .expect("first managed ticket");
        let second = client
            .endpoint_ticket_with_token("user-device", 0)
            .await
            .expect("second managed ticket");

        let (_, first_suffix) = crate::session_token::split_compound_ticket(&first);
        let (_, second_suffix) = crate::session_token::split_compound_ticket(&second);
        let first_payload =
            crate::session_token::decode_token_payload(first_suffix.expect("first token suffix"))
                .expect("first payload");
        let second_payload =
            crate::session_token::decode_token_payload(second_suffix.expect("second token suffix"))
                .expect("second payload");

        assert_eq!(first_payload.scope, "user-device");
        assert_eq!(second_payload.scope, "user-device");
        assert_eq!(
            first_payload.token, second_payload.token,
            "native managed ticket cache should reuse the same backend-owned token within one app session"
        );
    }

    #[tokio::test]
    async fn native_user_device_ticket_persists_across_app_restarts() {
        let base_dir = std::env::temp_dir().join(format!(
            "openrtc-user-device-persist-{}",
            std::time::SystemTime::now()
                .duration_since(std::time::UNIX_EPOCH)
                .expect("unix epoch")
                .as_nanos()
        ));

        let client_a = Client::new_with_app_tag(
            crate::test_constants::TEST_PROJECT_ID.to_string(),
            "test-app".to_string(),
            Box::new(|| None),
        );
        client_a
            .init_native_device_identity(base_dir.clone(), None)
            .await
            .expect("init native device identity a");
        client_a
            .init_iroh(None, Vec::new())
            .await
            .expect("init iroh a");

        let first = client_a
            .endpoint_ticket_with_token("user-device", 0)
            .await
            .expect("first managed ticket");

        let client_b = Client::new_with_app_tag(
            crate::test_constants::TEST_PROJECT_ID.to_string(),
            "test-app".to_string(),
            Box::new(|| None),
        );
        client_b
            .init_native_device_identity(base_dir.clone(), None)
            .await
            .expect("init native device identity b");
        client_b
            .init_iroh(None, Vec::new())
            .await
            .expect("init iroh b");

        let second = client_b
            .endpoint_ticket_with_token("user-device", 0)
            .await
            .expect("second managed ticket");

        let (_, first_suffix) = crate::session_token::split_compound_ticket(&first);
        let (_, second_suffix) = crate::session_token::split_compound_ticket(&second);
        let first_payload =
            crate::session_token::decode_token_payload(first_suffix.expect("first token suffix"))
                .expect("first payload");
        let second_payload =
            crate::session_token::decode_token_payload(second_suffix.expect("second token suffix"))
                .expect("second payload");

        assert_eq!(first_payload.token, second_payload.token);

        let _ = tokio::fs::remove_dir_all(&base_dir).await;
    }

    #[tokio::test]
    async fn revoking_persistent_user_device_scope_rotates_next_native_ticket() {
        let base_dir = std::env::temp_dir().join(format!(
            "openrtc-user-device-rotate-{}",
            std::time::SystemTime::now()
                .duration_since(std::time::UNIX_EPOCH)
                .expect("unix epoch")
                .as_nanos()
        ));

        let client = Client::new_with_app_tag(
            crate::test_constants::TEST_PROJECT_ID.to_string(),
            "test-app".to_string(),
            Box::new(|| None),
        );
        client
            .init_native_device_identity(base_dir.clone(), None)
            .await
            .expect("init native device identity");
        client.init_iroh(None, Vec::new()).await.expect("init iroh");

        let first = client
            .endpoint_ticket_with_token("user-device", 0)
            .await
            .expect("first managed ticket");
        client.revoke_tokens_by_scope("user-device").await;
        let second = client
            .endpoint_ticket_with_token("user-device", 0)
            .await
            .expect("second managed ticket");

        let (_, first_suffix) = crate::session_token::split_compound_ticket(&first);
        let (_, second_suffix) = crate::session_token::split_compound_ticket(&second);
        let first_payload =
            crate::session_token::decode_token_payload(first_suffix.expect("first token suffix"))
                .expect("first payload");
        let second_payload =
            crate::session_token::decode_token_payload(second_suffix.expect("second token suffix"))
                .expect("second payload");

        assert_ne!(first_payload.token, second_payload.token);

        let _ = tokio::fs::remove_dir_all(&base_dir).await;
    }

    #[tokio::test]
    async fn user_device_tokens_allow_multiple_connection_admissions_when_unlimited() {
        let client = Client::new_with_app_tag(
            crate::test_constants::TEST_PROJECT_ID.to_string(),
            "test-app".to_string(),
            Box::new(|| None),
        );

        client.register_session_token("user-device-token".into(), "user-device".into(), 0);

        let first_scope = client
            .validate_session_token_for_connection("user-device-token", "conn-user-device-1")
            .await
            .expect("first connection should be admitted");
        let second_scope = client
            .validate_session_token_for_connection("user-device-token", "conn-user-device-2")
            .await
            .expect("second connection should be admitted");

        assert_eq!(first_scope, "user-device");
        assert_eq!(second_scope, "user-device");
        assert!(matches!(
            client.session_admission("conn-user-device-1"),
            crate::session_token::SessionAdmission::Accepted { .. }
        ));
        assert!(matches!(
            client.session_admission("conn-user-device-2"),
            crate::session_token::SessionAdmission::Accepted { .. }
        ));
    }

    // -----------------------------------------------------------------------
    // Auto-connect with user-device compound tickets and tie-breaking
    // -----------------------------------------------------------------------

    #[test]
    fn compound_ticket_splits_correctly_for_user_device_scope() {
        let iroh_ticket = "endpointABCDEF1234567890";
        let token = "test-user-device-token-abc";
        let compound =
            crate::session_token::build_compound_ticket(iroh_ticket, token, "user-device", 0);

        let (recovered_ticket, suffix) = crate::session_token::split_compound_ticket(&compound);
        assert_eq!(recovered_ticket, iroh_ticket);
        assert!(suffix.is_some());

        let payload = crate::session_token::decode_token_payload(suffix.unwrap())
            .expect("token payload should decode");
        assert_eq!(payload.token, token);
        assert_eq!(payload.scope, "user-device");
        assert_eq!(payload.max_connections, 0);
    }

    #[tokio::test]
    async fn deterministic_connection_id_is_symmetric() {
        let id_ab = Client::deterministic_connection_id("node-aaa", "node-zzz");
        let id_ba = Client::deterministic_connection_id("node-zzz", "node-aaa");
        assert_eq!(
            id_ab, id_ba,
            "connection_id must be identical regardless of dial direction"
        );
    }

    #[tokio::test]
    async fn rejection_reset_allows_fresh_admission_after_timeout() {
        let client = Client::new_with_app_tag(
            crate::test_constants::TEST_PROJECT_ID.to_string(),
            "test-app".to_string(),
            Box::new(|| None),
        );

        client.register_session_token("ud-token".into(), "user-device".into(), 0);

        let connection_id = "conn-rejection-reset";

        // Simulate a timeout rejection (what handle_incoming_connection does)
        client.reject_session_connection(connection_id, "session-admission-timeout");

        assert!(
            matches!(
                client.session_admission(connection_id),
                crate::session_token::SessionAdmission::Rejected { .. }
            ),
            "connection should be rejected after timeout"
        );

        // Reset rejection (Fix #2 — what handle_incoming_connection now does on fresh transport)
        client.forget_session_connection(connection_id);

        assert!(
            matches!(
                client.session_admission(connection_id),
                crate::session_token::SessionAdmission::Pending
            ),
            "connection should be Pending after forget, got {:?}",
            client.session_admission(connection_id)
        );

        // Now a fresh token admission should succeed
        let scope = client
            .validate_session_token_for_connection("ud-token", connection_id)
            .await
            .expect("token should validate after rejection reset");
        assert_eq!(scope, "user-device");

        assert!(matches!(
            client.session_admission(connection_id),
            crate::session_token::SessionAdmission::Accepted { .. }
        ));
    }

    #[cfg(not(target_arch = "wasm32"))]
    #[tokio::test]
    async fn native_peer_snapshot_read_does_not_retire_transiently_missing_transport() {
        let client = Client::new_with_app_tag(
            crate::test_constants::TEST_PROJECT_ID.to_string(),
            "test-app".to_string(),
            Box::new(|| None),
        );
        let endpoint = Endpoint::builder(iroh::endpoint::presets::N0)
            .relay_mode(RelayMode::Disabled)
            .bind_addr(
                "127.0.0.1:0"
                    .parse::<std::net::SocketAddr>()
                    .expect("socket addr"),
            )
            .expect("bind addr")
            .bind()
            .await
            .expect("bind endpoint");
        let remote_node_id = endpoint.id().to_string();
        let connection_id = "conn-transient-path-refresh";

        client
            .connection_manager
            .upsert_pending(
                connection_id.to_string(),
                Some(remote_node_id.clone()),
                Some("device-transient-path-refresh".to_string()),
                Some(remote_node_id),
            )
            .await;
        client
            .connection_manager
            .set_connected_with_transport(
                connection_id,
                None,
                Some(42),
                Some("native-quic".to_string()),
            )
            .await
            .expect("connected record");

        let snapshot = client
            .peer_snapshot(connection_id)
            .await
            .expect("read-side projection must preserve the peer");
        assert_eq!(snapshot.connection_ids, vec![connection_id.to_string()]);
        assert!(
            client
                .connection_manager
                .get_by_connection_id(connection_id)
                .await
                .is_some(),
            "native read path must leave retirement to thresholded health reconciliation"
        );

        endpoint.close().await;
    }

    #[tokio::test]
    async fn tie_break_both_peers_share_deterministic_connection_id() {
        // Simulates what happens when both peers dial each other simultaneously:
        // both compute the same deterministic connection_id, so admission state
        // is shared and must be handled carefully.
        let client_a = Client::new_with_app_tag(
            crate::test_constants::TEST_PROJECT_ID.to_string(),
            "test-app".to_string(),
            Box::new(|| None),
        );

        client_a.register_session_token("host-token".into(), "user-device".into(), 0);

        let node_a = "aaaa1111aaaa1111aaaa1111aaaa1111";
        let node_b = "zzzz9999zzzz9999zzzz9999zzzz9999";

        // Both sides compute the same connection_id
        let connection_id = Client::deterministic_connection_id(node_a, node_b);
        let connection_id_reverse = Client::deterministic_connection_id(node_b, node_a);
        assert_eq!(connection_id, connection_id_reverse);

        // Simulate: B's incoming connection gets timeout-rejected on A
        client_a.reject_session_connection(&connection_id, "session-admission-timeout");

        // B re-dials. Fix #2: A resets the rejection for the deterministic connection_id
        client_a.forget_session_connection(&connection_id);

        // B presents its token — should succeed on the fresh admission window
        let scope = client_a
            .validate_session_token_for_connection("host-token", &connection_id)
            .await
            .expect("token should be accepted after rejection reset");
        assert_eq!(scope, "user-device");
    }

    #[tokio::test]
    async fn managed_adoption_uses_node_id_fallback_for_share_scope_anonymous_peer() {
        // Anonymous share-style peers are allowed to fall back to node_id so
        // the managed bridge can surface a routable connection before any app-
        // level identity arrives.
        let client = Client::new_with_app_tag(
            crate::test_constants::TEST_PROJECT_ID.to_string(),
            "test-app".to_string(),
            Box::new(|| None),
        );

        client.register_session_token("anon-token".into(), "share".into(), 0);

        let connection_id = "conn-anon-peer";
        let node_id = "node-anon-peer-1234";

        // Admit the connection via token
        client
            .validate_session_token_for_connection("anon-token", connection_id)
            .await
            .expect("token should validate");

        // Create a connected record WITHOUT a device_id (anonymous peer)
        client
            .connection_manager
            .upsert_pending(
                connection_id.to_string(),
                Some(node_id.to_string()),
                None, // no device_id
                Some(node_id.to_string()),
            )
            .await;
        client
            .connection_manager
            .set_connected_with_transport(
                connection_id,
                Some(node_id.to_string()),
                Some(99),
                Some("managed".to_string()),
            )
            .await
            .expect("set connected");

        let record = client
            .connection_manager
            .get_by_connection_id(connection_id)
            .await
            .expect("record exists");

        let adoption = client
            .managed_connection_adoption(&record)
            .await
            .expect("adoption should succeed with node_id fallback");

        assert_eq!(adoption.connection_id, connection_id);
        assert_eq!(adoption.node_id, node_id);
        // Share scope keeps the anonymous node_id fallback.
        assert_eq!(
            adoption.device_id.as_deref(),
            Some(node_id),
            "device_id should fall back to node_id for anonymous share peers"
        );
        assert!(
            adoption.main_stream_ready,
            "main_stream_ready should be true when device_id fallback is used"
        );
    }

    #[tokio::test]
    async fn managed_adoption_requires_real_device_identity_for_user_device_scope() {
        let client = Client::new_with_app_tag(
            crate::test_constants::TEST_PROJECT_ID.to_string(),
            "test-app".to_string(),
            Box::new(|| None),
        );

        client.register_session_token("user-device-token".into(), "user-device".into(), 0);

        let connection_id = "conn-user-device-anon";
        let node_id = "node-user-device-anon";

        client
            .validate_session_token_for_connection("user-device-token", connection_id)
            .await
            .expect("token should validate");

        client
            .connection_manager
            .upsert_pending(
                connection_id.to_string(),
                Some(node_id.to_string()),
                None,
                Some(node_id.to_string()),
            )
            .await;
        client
            .connection_manager
            .set_connected_with_transport(
                connection_id,
                Some(node_id.to_string()),
                Some(109),
                Some("managed".to_string()),
            )
            .await
            .expect("set connected");

        let record = client
            .connection_manager
            .get_by_connection_id(connection_id)
            .await
            .expect("record exists");

        let adoption = client
            .managed_connection_adoption(&record)
            .await
            .expect("adoption snapshot should still be produced");

        assert_eq!(adoption.connection_id, connection_id);
        assert_eq!(adoption.node_id, node_id);
        assert_eq!(
            adoption.device_id, None,
            "user-device scope must not fall back to node_id as app identity"
        );
        assert!(
            !adoption.main_stream_ready,
            "user-device adoption should wait for an authoritative device identity"
        );
    }

    #[tokio::test]
    async fn managed_adoption_blocked_for_unadmitted_peer_even_with_device_id() {
        // When the session registry is active, a connected peer with a device_id
        // but no token admission should NOT be adoptable.
        let client = Client::new_with_app_tag(
            crate::test_constants::TEST_PROJECT_ID.to_string(),
            "test-app".to_string(),
            Box::new(|| None),
        );

        client.register_session_token("gate-token".into(), "user-device".into(), 0);

        let connection_id = "conn-unadmitted-with-device";
        let node_id = "node-unadmitted";

        client
            .connection_manager
            .upsert_pending(
                connection_id.to_string(),
                Some(node_id.to_string()),
                Some("real-device-id".to_string()),
                Some(node_id.to_string()),
            )
            .await;
        client
            .connection_manager
            .set_connected_with_transport(
                connection_id,
                Some(node_id.to_string()),
                Some(50),
                Some("incoming".to_string()),
            )
            .await
            .expect("set connected");
        client
            .connection_manager
            .set_device_id(connection_id, "real-device-id".to_string())
            .await
            .expect("set device_id");

        let record = client
            .connection_manager
            .get_by_connection_id(connection_id)
            .await
            .expect("record exists");

        let adoption = client.managed_connection_adoption(&record).await;
        assert!(
            adoption.is_none(),
            "unadmitted connection should not be adoptable even with device_id"
        );
    }

    #[tokio::test]
    async fn late_device_binding_completes_user_device_session_admission() {
        let client = Client::new_with_app_tag(
            crate::test_constants::TEST_PROJECT_ID.to_string(),
            "test-app".to_string(),
            Box::new(|| None),
        );

        let connection_id = "conn-late-authoritative-device";
        let device_id = "0807631e-97af-4fb1-b245-93a63ff4dcad";

        client
            .connection_manager
            .upsert_pending(
                connection_id.to_string(),
                Some("node-1".to_string()),
                None,
                None,
            )
            .await;
        client
            .mark_connection_admitted_by_host(connection_id, Some("user-device"), None)
            .await;

        assert!(matches!(
            client.session_admission(connection_id),
            crate::session_token::SessionAdmission::Accepted {
                scope: Some(_),
                authoritative_device_id: None,
                ..
            }
        ));

        assert!(
            client
                .bind_session_admission_authoritative_device_id(connection_id, device_id)
                .await,
            "late binding should fill the missing authoritative device id"
        );

        assert!(matches!(
            client.session_admission(connection_id),
            crate::session_token::SessionAdmission::Accepted {
                scope: Some(scope),
                authoritative_device_id: Some(bound),
                ..
            } if scope.as_str() == "user-device" && bound == device_id
        ));

        let record = client
            .connection_manager
            .get_by_connection_id(connection_id)
            .await
            .expect("managed connection record should remain indexed");
        assert_eq!(record.device_id.as_deref(), Some(device_id));
    }

    #[tokio::test]
    async fn bind_connection_device_id_admits_managed_user_device_session() {
        let client = Client::new_with_app_tag(
            crate::test_constants::TEST_PROJECT_ID.to_string(),
            "test-app".to_string(),
            Box::new(|| None),
        );

        client.register_session_token("active-gate-token".into(), "user-device".into(), 0);

        let connection_id = "conn-bound-managed-user-device";
        let device_id = "managed-device-bound-from-handshake";
        client
            .connection_manager
            .upsert_pending(
                connection_id.to_string(),
                Some("node-bound-managed-user-device".to_string()),
                None,
                Some("node-bound-managed-user-device".to_string()),
            )
            .await;
        client
            .connection_manager
            .set_connected_with_transport(
                connection_id,
                Some("node-bound-managed-user-device".to_string()),
                Some(123),
                Some("wasm-accept".to_string()),
            )
            .await
            .expect("set connected");

        let snapshot = client
            .bind_connection_device_id(connection_id, device_id)
            .await
            .expect("bound peer snapshot");

        assert_eq!(snapshot.peer_id, device_id);
        assert!(snapshot.scopes.iter().any(|scope| scope == "user-device"));
        assert!(matches!(
            client.session_admission(connection_id),
            crate::session_token::SessionAdmission::Accepted {
                mechanism: crate::session_token::SessionAdmissionMechanism::TrustedNativeBinding,
                scope: Some(scope),
                authoritative_device_id: Some(bound),
            } if scope.as_str() == "user-device" && bound == device_id
        ));

        let session = client
            .peer_session(device_id)
            .await
            .expect("peer session snapshot");
        assert_eq!(session.peer_id, device_id);
        assert!(session.settled_ready);
        assert_eq!(session.readiness_reason, "transport-healthy");
        assert!(session.scopes.iter().any(|scope| scope == "user-device"));
    }

    #[tokio::test]
    async fn user_device_token_survives_multiple_rejection_reset_cycles() {
        // Simulates repeated reconnection attempts: each time the old connection
        // times out, gets rejected, then a fresh connection resets and re-admits.
        let client = Client::new_with_app_tag(
            crate::test_constants::TEST_PROJECT_ID.to_string(),
            "test-app".to_string(),
            Box::new(|| None),
        );

        // user-device with maxConnections=0 (unlimited)
        client.register_session_token("reconnect-token".into(), "user-device".into(), 0);

        let connection_id = "conn-reconnect-cycle";

        for cycle in 0..5 {
            // Simulate timeout rejection
            client.reject_session_connection(connection_id, "session-admission-timeout");
            assert!(
                matches!(
                    client.session_admission(connection_id),
                    crate::session_token::SessionAdmission::Rejected { .. }
                ),
                "cycle {}: should be rejected",
                cycle
            );

            // Reset for fresh transport
            client.forget_session_connection(connection_id);
            assert!(
                matches!(
                    client.session_admission(connection_id),
                    crate::session_token::SessionAdmission::Pending
                ),
                "cycle {}: should be pending after reset",
                cycle
            );

            // Re-admit with the same unlimited token
            let scope = client
                .validate_session_token_for_connection("reconnect-token", connection_id)
                .await
                .expect(&format!("cycle {}: token should still validate", cycle));
            assert_eq!(scope, "user-device");
        }
    }

    #[tokio::test]
    async fn session_token_presentation_message_roundtrips_correctly() {
        // Verify the session-token presentation message format that
        // present_session_token_to_host sends over the wire.
        let msg = NativeMainMessage::session_token_presentation("my-secret-token");
        let serialized = serde_json::to_vec(&msg).expect("serialize");

        let deserialized: NativeMainMessage =
            serde_json::from_slice(&serialized).expect("deserialize");

        assert!(
            deserialized.is_session_token_presentation(),
            "deserialized message should be a session-token presentation"
        );

        let extracted_token = deserialized
            .payload
            .get("token")
            .and_then(|v| v.as_str())
            .expect("token field");
        assert_eq!(extracted_token, "my-secret-token");
    }

    #[tokio::test]
    async fn auto_connect_tie_break_initiator_role_determined_by_node_id_comparison() {
        // Verify the tie-break decision function works with realistic node IDs
        // and that the initiator role is determined by lexicographic comparison.
        let high_node = "zzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzz";
        let low_node = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa";

        // high > low, so high is the initiator
        let decision = auto_connect_tie_break_decision(
            high_node, low_node, false, // not yet connected
            0,     // no wait time
            3_000, // grace_ms
        );
        assert!(
            matches!(decision, AutoConnectTieBreakDecision::ActAsInitiator),
            "higher node_id should act as initiator"
        );

        // low < high, so low is the non-initiator and should wait
        let decision = auto_connect_tie_break_decision(low_node, high_node, false, 0, 3_000);
        assert!(
            matches!(decision, AutoConnectTieBreakDecision::WaitForInitiator),
            "lower node_id should wait for initiator"
        );

        // After grace period, non-initiator escalates
        let decision = auto_connect_tie_break_decision(
            low_node, high_node, false, 4_000, // exceeded 3000ms grace
            3_000,
        );
        assert!(
            matches!(decision, AutoConnectTieBreakDecision::ActAsInitiator),
            "non-initiator should escalate after grace period"
        );
    }

    #[tokio::test]
    async fn full_auto_connect_sequence_native_initiator_to_wasm_peer() {
        // End-to-end unit test of the auto-connect admission flow from the
        // host (desktop/native) perspective when a WASM peer connects in.
        //
        // Sequence:
        // 1. Host registers a user-device token and publishes compound ticket
        // 2. WASM peer extracts the token from the compound ticket
        // 3. WASM peer connects transport
        // 4. WASM peer presents token → host validates and admits
        // 5. managed_connection_adoption produces a routable snapshot
        let host = Client::new_with_app_tag(
            crate::test_constants::TEST_PROJECT_ID.to_string(),
            "test-app".to_string(),
            Box::new(|| None),
        );

        // Step 1: Host generates compound ticket
        let token = "host-user-device-token-xyz";
        host.register_session_token(token.into(), "user-device".into(), 0);
        let iroh_ticket = "endpointABCDEF1234567890abcdef";
        let compound =
            crate::session_token::build_compound_ticket(iroh_ticket, token, "user-device", 0);

        // Step 2: WASM peer splits the compound ticket (same as try_auto_connect_device)
        let (recovered_iroh, suffix) = crate::session_token::split_compound_ticket(&compound);
        assert_eq!(recovered_iroh, iroh_ticket);
        let extracted_token = suffix
            .and_then(|value| {
                crate::session_token::decode_token_payload_for_ticket(recovered_iroh, value)
            })
            .map(|p| p.token)
            .expect("token should be extractable");
        assert_eq!(extracted_token, token);

        let wasm_node_id = "wasm-peer-node-id-1234";
        let wasm_device_id = "wasm-peer-device-id-1234";
        let connection_id = Client::deterministic_connection_id("host-node-id", wasm_node_id);

        // Step 3: Transport established — peer creates a managed connection record
        host.connection_manager
            .upsert_pending(
                connection_id.clone(),
                Some(wasm_node_id.to_string()),
                Some(wasm_device_id.to_string()),
                Some(wasm_node_id.to_string()),
            )
            .await;
        host.connection_manager
            .set_connected_with_transport(
                &connection_id,
                Some(wasm_node_id.to_string()),
                Some(77),
                Some("managed".to_string()),
            )
            .await
            .expect("set connected");

        // Step 4: WASM peer presents its extracted token
        let scope = host
            .validate_session_token_for_connection(&extracted_token, &connection_id)
            .await
            .expect("token presentation should succeed");
        assert_eq!(scope, "user-device");

        // Step 5: managed_connection_adoption should produce a routable snapshot
        let record = host
            .connection_manager
            .get_by_connection_id(&connection_id)
            .await
            .expect("record exists");
        let adoption = host
            .managed_connection_adoption(&record)
            .await
            .expect("adoption should succeed for admitted connection");

        assert_eq!(adoption.connection_id, connection_id);
        assert_eq!(adoption.node_id, wasm_node_id);
        assert_eq!(adoption.device_id.as_deref(), Some(wasm_device_id));
        assert!(adoption.main_stream_ready);
    }

    #[tokio::test]
    async fn full_auto_connect_sequence_wasm_initiator_to_native_host() {
        // Simulates the reverse direction: WASM initiates to native host.
        // The host has a compound ticket published. WASM dials in, presents
        // token, and gets admitted. This covers the desktop-as-host scenario.
        let host = Client::new_with_app_tag(
            crate::test_constants::TEST_PROJECT_ID.to_string(),
            "test-app".to_string(),
            Box::new(|| None),
        );

        let token = "native-host-token-for-wasm";
        host.register_session_token(token.into(), "user-device".into(), 0);

        let native_node = "native-host-node-aaa";
        let wasm_node = "wasm-client-node-zzz";
        let connection_id = Client::deterministic_connection_id(native_node, wasm_node);

        // Simulate: host sees incoming connection from wasm
        host.connection_manager
            .upsert_pending(
                connection_id.clone(),
                Some(wasm_node.to_string()),
                Some("wasm-device-id-123".to_string()),
                Some(wasm_node.to_string()),
            )
            .await;
        host.connection_manager
            .set_connected_with_transport(
                &connection_id,
                Some(wasm_node.to_string()),
                Some(88),
                Some("incoming".to_string()),
            )
            .await
            .expect("set connected");
        host.connection_manager
            .set_device_id(&connection_id, "wasm-device-id-123".to_string())
            .await
            .expect("set device_id");

        // Before token admission, adoption should be blocked
        let record = host
            .connection_manager
            .get_by_connection_id(&connection_id)
            .await
            .expect("record exists");
        assert!(
            host.managed_connection_adoption(&record).await.is_none(),
            "adoption should be blocked before token admission"
        );

        // Token presentation
        let scope = host
            .validate_session_token_for_connection(token, &connection_id)
            .await
            .expect("token should validate");
        assert_eq!(scope, "user-device");

        // Now adoption should succeed with the real device_id
        let adoption = host
            .managed_connection_adoption(&record)
            .await
            .expect("adoption should succeed after token admission");

        assert_eq!(adoption.device_id.as_deref(), Some("wasm-device-id-123"));
        assert!(adoption.main_stream_ready);
    }

    #[tokio::test]
    async fn duplicate_resolution_with_token_admission_preserves_admitted_state() {
        // When duplicate resolution keeps one connection and closes the other,
        // the admitted state on the kept connection should be preserved.
        let host = Client::new_with_app_tag(
            crate::test_constants::TEST_PROJECT_ID.to_string(),
            "test-app".to_string(),
            Box::new(|| None),
        );

        host.register_session_token("dup-token".into(), "user-device".into(), 0);

        let connection_id = "conn-dup-test";

        // Admit the connection
        host.validate_session_token_for_connection("dup-token", connection_id)
            .await
            .expect("token validates");

        assert!(matches!(
            host.session_admission(connection_id),
            crate::session_token::SessionAdmission::Accepted { .. }
        ));

        // After duplicate resolution, the kept connection's admission should still be intact.
        // (The closed connection's admission state is irrelevant since it's discarded.)
        // The key invariant: we do NOT call forget_session_connection on the KEPT connection.
        assert!(matches!(
            host.session_admission(connection_id),
            crate::session_token::SessionAdmission::Accepted { .. }
        ));
    }

    #[tokio::test]
    async fn forget_on_retirement_enables_reconnect_for_same_peer() {
        // Tests Fix #3: when a connection is retired (removed from bridged_connections),
        // calling forget_session_connection ensures a future reconnect for the same
        // peer (same deterministic connection_id) gets a fresh admission window.
        let host = Client::new_with_app_tag(
            crate::test_constants::TEST_PROJECT_ID.to_string(),
            "test-app".to_string(),
            Box::new(|| None),
        );

        host.register_session_token("retire-token".into(), "user-device".into(), 0);

        let node_a = "node-local-host";
        let node_b = "node-remote-peer";
        let connection_id = Client::deterministic_connection_id(node_a, node_b);

        // First connection: admit, then simulate retirement
        host.validate_session_token_for_connection("retire-token", &connection_id)
            .await
            .expect("first admission");
        assert!(matches!(
            host.session_admission(&connection_id),
            crate::session_token::SessionAdmission::Accepted { .. }
        ));

        // Simulate retirement: forget the connection (Fix #3)
        host.forget_session_connection(&connection_id);
        assert!(matches!(
            host.session_admission(&connection_id),
            crate::session_token::SessionAdmission::Pending
        ));

        // Second connection from same peer (same deterministic connection_id)
        let scope = host
            .validate_session_token_for_connection("retire-token", &connection_id)
            .await
            .expect("second admission should succeed after retirement forget");
        assert_eq!(scope, "user-device");
    }

    #[test]
    fn ensure_default_admission_gate_activates_registry_before_any_connection() {
        // On app startup, ensure_default_admission_gate should activate the
        // session registry so that all connections are token-gated from the start.
        // This prevents a window where connections arrive before any token is registered.
        let client = Client::new_with_app_tag(
            crate::test_constants::TEST_PROJECT_ID.to_string(),
            "test-app".to_string(),
            Box::new(|| None),
        );

        assert!(
            !client.session_registry_active(),
            "registry should be inactive before startup"
        );

        client.ensure_default_admission_gate("__startup");

        assert!(
            client.session_registry_active(),
            "registry should be active after startup gate"
        );

        // Connections without tokens should now be rejected
        assert!(matches!(
            client.session_admission("any-connection"),
            crate::session_token::SessionAdmission::Pending
        ));
    }

    #[test]
    fn protect_outbound_application_payload_fails_closed_when_required_key_missing() {
        let client = Client::new_with_app_tag(
            crate::test_constants::TEST_PROJECT_ID.to_string(),
            "test-app".to_string(),
            Box::new(|| None),
        );
        client.set_connection_application_crypto_required("conn-1");
        let err = client
            .protect_outbound_application_payload("conn-1", b"hello")
            .expect_err("missing key should fail when crypto is required");
        assert!(err.to_string().contains("application crypto required"));
    }

    #[test]
    fn protect_outbound_application_payload_passthrough_when_not_required() {
        let client = Client::new_with_app_tag(
            crate::test_constants::TEST_PROJECT_ID.to_string(),
            "test-app".to_string(),
            Box::new(|| None),
        );
        let protected = client
            .protect_outbound_application_payload("conn-1", b"hello")
            .expect("plaintext passthrough allowed when crypto not required");
        assert_eq!(protected, b"hello");
    }

    #[cfg(all(
        not(target_arch = "wasm32"),
        any(feature = "transport-webrtc", feature = "transport-moq")
    ))]
    #[tokio::test]
    async fn native_transport_peer_data_event_opens_application_crypto_payload() {
        let client = Client::new_with_app_tag(
            crate::test_constants::TEST_PROJECT_ID.to_string(),
            "test-app".to_string(),
            Box::new(|| None),
        );
        let mut rx = client.subscribe_native_peer_data();
        let key = [7u8; crate::application_crypto::APPLICATION_KEY_BYTES];
        let nonce = [3u8; crate::application_crypto::APPLICATION_NONCE_BYTES];
        client.set_connection_application_crypto_key("conn-1", key);
        let protected = crate::application_crypto::protect_application_payload_with_nonce(
            &key,
            0,
            b"hello over moq",
            &nonce,
        )
        .expect("payload should encrypt");

        client
            .handle_native_transport_peer_data(
                "conn-1",
                Some("remote-node"),
                "moq",
                bytes::Bytes::from(protected),
            )
            .await
            .expect("transport payload should open and emit");

        let event = tokio::time::timeout(Duration::from_secs(1), rx.recv())
            .await
            .expect("native peer data event should arrive")
            .expect("native peer data channel should stay open");
        assert_eq!(event.connection_id, "conn-1");
        assert_eq!(event.remote_node_id.as_deref(), Some("remote-node"));
        assert_eq!(event.transport, "moq");
        assert_eq!(event.payload, b"hello over moq");
    }

    #[cfg(all(
        not(target_arch = "wasm32"),
        any(feature = "transport-webrtc", feature = "transport-moq")
    ))]
    #[tokio::test]
    async fn native_moq_peer_data_event_opens_raw_stream_application_crypto_payload() {
        let client = Client::new_with_app_tag(
            crate::test_constants::TEST_PROJECT_ID.to_string(),
            "test-app".to_string(),
            Box::new(|| None),
        );
        let mut rx = client.subscribe_native_peer_data();
        let key = [9u8; crate::application_crypto::APPLICATION_KEY_BYTES];
        let nonce = [4u8; crate::application_crypto::APPLICATION_NONCE_BYTES];
        client.set_connection_application_crypto_key("conn-1", key);
        let protected = crate::application_crypto::protect_application_payload_with_nonce(
            &key,
            crate::application_crypto::RAW_STREAM_TYPE_ID,
            b"hello over direct moq",
            &nonce,
        )
        .expect("payload should encrypt");

        client
            .handle_native_transport_peer_data(
                "conn-1",
                Some("remote-node"),
                "moq",
                bytes::Bytes::from(protected),
            )
            .await
            .expect("direct moq payload should open and emit");

        let event = tokio::time::timeout(Duration::from_secs(1), rx.recv())
            .await
            .expect("native peer data event should arrive")
            .expect("native peer data channel should stay open");
        assert_eq!(event.connection_id, "conn-1");
        assert_eq!(event.remote_node_id.as_deref(), Some("remote-node"));
        assert_eq!(event.transport, "moq");
        assert_eq!(event.payload, b"hello over direct moq");
    }

    #[cfg(all(not(target_arch = "wasm32"), feature = "transport-moq"))]
    #[tokio::test(start_paused = true)]
    async fn native_moq_explicit_send_repairs_route_before_reporting_not_ready() {
        let client = Client::new_with_app_tag(
            crate::test_constants::TEST_PROJECT_ID.to_string(),
            "test-app".to_string(),
            Box::new(|| None),
        );
        *client.node_id.write().await = Some("local-node".to_string());
        client
            .update_transport_config(TransportConfig {
                moq: Some(MoQConfig {
                    relay_url: "https://relay.invalid/moq".to_string(),
                }),
                ..TransportConfig::default()
            })
            .await
            .expect("transport config");
        client
            .connection_manager
            .upsert_pending(
                "conn-moq".to_string(),
                Some("remote-node".to_string()),
                Some("remote-device".to_string()),
                Some("remote-node".to_string()),
            )
            .await;
        client
            .connection_manager
            .set_connected_with_transport(
                "conn-moq",
                Some("remote-node".to_string()),
                None,
                Some("incoming".to_string()),
            )
            .await
            .expect("connected record");
        client.set_connection_application_crypto_key(
            "conn-moq",
            [11u8; crate::application_crypto::APPLICATION_KEY_BYTES],
        );
        assert_eq!(
            client.resolve_peer_connection_ids("remote-device").await,
            vec!["conn-moq".to_string()],
            "device id should resolve to the managed connection before MoQ repair",
        );
        assert_eq!(
            client
                .resolve_transport_connection_ids("remote-device")
                .await,
            vec!["remote-device".to_string(), "conn-moq".to_string()],
            "transport lookup checks the raw transport key first, then the managed connection",
        );

        let error = client
            .send_peer_over_moq("remote-device", b"hello over moq")
            .await
            .expect_err("invalid relay should still fail the final MoQ-only proof");
        assert!(
            error
                .to_string()
                .contains("MoQ transport is not data-ready"),
            "unexpected error: {error}",
        );
        assert!(
            error
                .to_string()
                .contains("connection_id=conn-moq state=Failed data_ready=false"),
            "error should include all native MoQ candidate readiness details: {error}",
        );
        let session_keys = client
            .native_moq_sessions
            .read()
            .await
            .keys()
            .cloned()
            .collect::<Vec<_>>();
        assert_eq!(
            session_keys,
            vec!["conn-moq".to_string()],
            "explicit send should request native MoQ route repair before reporting not ready",
        );
        let state = client
            .native_moq_state_for_peer("remote-device")
            .await
            .expect("explicit send should request native MoQ route repair first");
        assert_eq!(state.0, "conn-moq");
        assert_eq!(state.1, crate::transport::NativeMoQState::Failed);
    }

    #[cfg(all(not(target_arch = "wasm32"), feature = "transport-moq"))]
    #[tokio::test]
    async fn native_moq_repair_replaces_failed_session_before_restart() {
        let client = Client::new_with_app_tag(
            crate::test_constants::TEST_PROJECT_ID.to_string(),
            "test-app".to_string(),
            Box::new(|| None),
        );
        *client.node_id.write().await = Some("local-node".to_string());
        client
            .update_transport_config(TransportConfig {
                moq: Some(MoQConfig {
                    relay_url: "https://relay.invalid/moq".to_string(),
                }),
                ..TransportConfig::default()
            })
            .await
            .expect("transport config");
        client
            .connection_manager
            .upsert_pending(
                "conn-moq-repair".to_string(),
                Some("remote-node-repair".to_string()),
                Some("remote-device-repair".to_string()),
                Some("remote-node-repair".to_string()),
            )
            .await;
        client
            .connection_manager
            .set_connected_with_transport(
                "conn-moq-repair",
                Some("remote-node-repair".to_string()),
                None,
                Some("incoming".to_string()),
            )
            .await
            .expect("connected record");

        assert!(client
            .request_moq_upgrade(
                "remote-device-repair",
                Some("remote-node-repair"),
                Some("test-initial")
            )
            .await
            .expect("initial request"));
        let first = {
            let sessions = client.native_moq_sessions.read().await;
            sessions
                .get("conn-moq-repair")
                .cloned()
                .expect("first failed session")
        };
        assert_eq!(first.state(), crate::transport::NativeMoQState::Failed);

        assert!(client
            .request_moq_upgrade(
                "remote-device-repair",
                Some("remote-node-repair"),
                Some("test-repair")
            )
            .await
            .expect("repair request"));
        let second = {
            let sessions = client.native_moq_sessions.read().await;
            sessions
                .get("conn-moq-repair")
                .cloned()
                .expect("replacement failed session")
        };

        assert!(
            !std::sync::Arc::ptr_eq(&first, &second),
            "repair must replace failed native MoQ sessions instead of reusing stale relay state",
        );
        assert_eq!(first.state(), crate::transport::NativeMoQState::Closed);
        assert_eq!(second.state(), crate::transport::NativeMoQState::Failed);
    }

    #[cfg(all(not(target_arch = "wasm32"), feature = "transport-moq"))]
    #[tokio::test]
    async fn native_moq_repair_replaces_connected_session_without_peer_data_subscription() {
        let client = Client::new_with_app_tag(
            crate::test_constants::TEST_PROJECT_ID.to_string(),
            "test-app".to_string(),
            Box::new(|| None),
        );
        *client.node_id.write().await = Some("local-node".to_string());
        let moq_config = MoQConfig {
            relay_url: "https://relay.invalid/moq".to_string(),
        };
        client
            .update_transport_config(TransportConfig {
                moq: Some(moq_config.clone()),
                ..TransportConfig::default()
            })
            .await
            .expect("transport config");
        client
            .connection_manager
            .upsert_pending(
                "conn-moq-repair-not-ready".to_string(),
                Some("remote-node-repair-not-ready".to_string()),
                Some("remote-device-repair-not-ready".to_string()),
                Some("remote-node-repair-not-ready".to_string()),
            )
            .await;
        client
            .connection_manager
            .set_connected_with_transport(
                "conn-moq-repair-not-ready",
                Some("remote-node-repair-not-ready".to_string()),
                None,
                Some("incoming".to_string()),
            )
            .await
            .expect("connected record");

        let first = std::sync::Arc::new(
            crate::transport::NativeMoQSession::new(
                "local-node",
                "remote-node-repair-not-ready",
                &moq_config,
            )
            .await
            .expect("test session"),
        );
        first.force_state_for_test(crate::transport::NativeMoQState::Connected);
        client
            .native_moq_sessions
            .write()
            .await
            .insert("conn-moq-repair-not-ready".to_string(), first.clone());
        assert!(!first.is_peer_data_subscribed().await);
        let (connection_id, state, data_ready) = client
            .native_moq_state_detail_for_peer("remote-device-repair-not-ready")
            .await
            .expect("native MoQ state detail");
        assert_eq!(connection_id, "conn-moq-repair-not-ready");
        assert_eq!(state, crate::transport::NativeMoQState::Connected);
        assert!(
            !data_ready,
            "connected native MoQ sessions are not ready until peer data subscription is active"
        );

        assert!(client
            .request_moq_repair(
                "remote-device-repair-not-ready",
                Some("remote-node-repair-not-ready"),
                Some("test-repair-not-ready")
            )
            .await
            .expect("repair request"));
        let second = {
            let sessions = client.native_moq_sessions.read().await;
            sessions
                .get("conn-moq-repair-not-ready")
                .cloned()
                .expect("replacement failed session")
        };

        assert!(
            !std::sync::Arc::ptr_eq(&first, &second),
            "repair must replace connected-but-not-data-ready native MoQ sessions",
        );
        assert_eq!(first.state(), crate::transport::NativeMoQState::Closed);
        assert_eq!(second.state(), crate::transport::NativeMoQState::Failed);
    }
}