openrtc 0.2.0

//! Session-token admission and managed-scope grant persistence methods for `Client`.
//!
//! Extracted from `core_impl.rs` to keep that file focused on endpoint, transport,
//! and connection lifecycle. All items here are `impl Client` blocks that depend
//! on `super::*` (the same `use super::*;` pattern used by the other `_impl` files).

#[cfg(not(target_arch = "wasm32"))]
use super::core_impl::log_fingerprint;
#[cfg(native)]
use super::core_impl::{
    PERSISTENT_MANAGED_ADMISSION_SCOPE, PERSISTENT_MANAGED_SCOPE_TICKET_FILE_PREFIX,
};
use super::*;
#[cfg(native)]
use anyhow::Context;

/// Upper bound on how long a managed dial waits for a host's session-token
/// approval response before treating the presentation as failed. A real
/// granting host replies in well under a second; bounding the read keeps the
/// dial from wedging when the peer does not run the host token responder.
const SESSION_TOKEN_HOST_RESPONSE_TIMEOUT_MS: u64 = 5_000;

#[derive(Debug, Clone, Copy, Default)]
struct SessionTokenValidationOptions<'a> {
    payload_suffix: Option<&'a str>,
    run_side_effects: bool,
}

#[derive(Debug, Clone, Copy, Default)]
struct SessionTokenPresentationOptions<'a> {
    token_payload: Option<&'a str>,
    device_id: Option<&'a str>,
}

#[derive(Debug, Clone, Copy)]
enum SessionTokenPresentationTarget<'a> {
    Host {
        endpoint_id: iroh::EndpointId,
        connection_id: &'a str,
    },
    Endpoint {
        endpoint_id: iroh::EndpointId,
    },
}

#[cfg(native)]
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
#[serde(rename_all = "camelCase")]
pub(crate) struct PersistedManagedScopeGrantRecord {
    pub(crate) scope: String,
    pub(crate) token: String,
    pub(crate) max_connections: u32,
}

impl Client {
    /// Validate an incoming session token and consume one use.
    /// Returns Ok(scope) on success, Err(reason) on failure.
    /// Empty registry = backward-compat gate (all pass).
    pub fn validate_session_token(&self, token: &str) -> Result<String, String> {
        self.session_token_registry
            .validate_and_consume(token)
            .map(|grant_scope| grant_scope.into_inner())
    }

    /// Phase 1: verdict-only admission. Returns the deterministic scope
    /// associated with `(token, connection_id)` *without* running any
    /// post-admission side effects (replacement peer accept, native WebRTC
    /// recovery, etc.). Callers that own the response writer must:
    ///
    ///   1. Call this to compute the verdict.
    ///   2. Write + flush the approval/rejection response to the wire.
    ///   3. Then invoke
    ///      [`Self::run_post_session_token_admission_side_effects`] to fire
    ///      the lifecycle hooks (replacement, WebRTC restart, ...).
    ///
    /// This separation eliminates the
    /// `[session-token-response:protocol-byte] 0 bytes read` race where a
    /// concurrent `accept_replacement_peer` could retire the very transport
    /// the response writer was about to flush onto.
    ///
    /// Existing callers that do not own a wire-level response writer (e.g.
    /// the inline TS/handshake paths in
    /// `inspect_incoming_native_main_frame`) still get the legacy "validate
    /// + side effects" semantics via
    /// [`Self::validate_session_token_for_connection_with_side_effects`].
    pub async fn validate_session_token_for_connection(
        &self,
        token: &str,
        connection_id: &str,
    ) -> Result<String, String> {
        self.validate_session_token_for_connection_with_options(
            token,
            connection_id,
            SessionTokenValidationOptions::default(),
        )
        .await
    }

    pub async fn validate_session_token_for_connection_with_payload(
        &self,
        token: &str,
        connection_id: &str,
        payload_suffix: Option<&str>,
    ) -> Result<String, String> {
        self.validate_session_token_for_connection_with_options(
            token,
            connection_id,
            SessionTokenValidationOptions {
                payload_suffix,
                run_side_effects: false,
            },
        )
        .await
    }

    /// Pre-Phase-1 helper: validates and immediately runs post-admission
    /// side effects in the same call. Used by callers that do not own a
    /// dedicated response writer (i.e. paths where there is no opportunity
    /// to interleave a wire flush between the verdict and the side
    /// effects). Equivalent to the pre-Phase-1 behaviour of
    /// `validate_session_token_for_connection`.
    pub async fn validate_session_token_for_connection_with_side_effects(
        &self,
        token: &str,
        connection_id: &str,
    ) -> Result<String, String> {
        self.validate_session_token_for_connection_with_options(
            token,
            connection_id,
            SessionTokenValidationOptions {
                payload_suffix: None,
                run_side_effects: true,
            },
        )
        .await
    }

    pub async fn validate_session_token_for_connection_with_payload_and_side_effects(
        &self,
        token: &str,
        connection_id: &str,
        payload_suffix: Option<&str>,
    ) -> Result<String, String> {
        self.validate_session_token_for_connection_with_options(
            token,
            connection_id,
            SessionTokenValidationOptions {
                payload_suffix,
                run_side_effects: true,
            },
        )
        .await
    }

    async fn validate_session_token_for_connection_with_options(
        &self,
        token: &str,
        connection_id: &str,
        options: SessionTokenValidationOptions<'_>,
    ) -> Result<String, String> {
        let was_already_admitted = self
            .session_token_registry
            .is_session_token_admitted_for_connection(connection_id);
        let grant_scope = self
            .session_token_registry
            .validate_and_consume_for_connection_with_payload(
                token,
                Some(connection_id),
                options.payload_suffix,
            )?;
        if !grant_scope.as_str().trim().is_empty() {
            self.connection_manager
                .add_scope(connection_id, grant_scope.as_str())
                .await;
        }
        let scope = grant_scope.into_inner();
        if options.run_side_effects {
            self.run_post_session_token_admission_side_effects(
                connection_id,
                !was_already_admitted,
            )
            .await;
        }
        Ok(scope)
    }

    /// Run the lifecycle side effects that previously lived inline in
    /// `validate_session_token_for_connection`. Idempotent: if
    /// `is_first_presentation` is `false`, this is a no-op (matching the
    /// pre-Phase-1 behaviour where duplicate token frames were skipped).
    ///
    /// Only callers that own a session-token response writer should invoke
    /// this directly — and only *after* the response has been written and
    /// the wire flushed.
    pub async fn run_post_session_token_admission_side_effects(
        &self,
        connection_id: &str,
        is_first_presentation: bool,
    ) {
        if !is_first_presentation {
            return;
        }

        #[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
        {
            let current_device_identity = self
                .connection_manager
                .get_by_connection_id(connection_id)
                .await
                .and_then(|record| record.device_id.or(record.device_id_hint));
            self.accept_replacement_peer(
                connection_id,
                current_device_identity.as_deref(),
                "replacement-peer-admitted",
            )
            .await;

            // First session-token presentation for this connection_id: the
            // remote may have a fresh RTCPeerConnection (e.g. browser
            // refresh) — force-restart the WebRTC upgrade so stale ICE/SCTP
            // state is torn down. Security:
            // maybe_start_native_webrtc_upgrade re-checks session_admission —
            // this path runs after mark_accepted in
            // validate_and_consume_for_connection, so Rejected remotes
            // never reach here.
            let current_webrtc_state = self.native_webrtc_state_for_peer(connection_id).await;
            let force_restart = matches!(
                current_webrtc_state,
                Some((_, crate::transport::NativeWebRTCState::Connecting))
                    | Some((_, crate::transport::NativeWebRTCState::Connected))
            );
            if current_webrtc_state.is_some() {
                self.reset_native_webrtc_attempt_budget(connection_id, "fresh-token-presentation")
                    .await;
            }
            if let Err(error) = self
                .request_native_webrtc_recovery(
                    connection_id,
                    None,
                    crate::native_webrtc_policy::NativeWebRTCRecoveryTrigger::Native(
                        crate::native_webrtc_policy::NativeWebRTCNativeTrigger::AdmissionAccepted,
                    ),
                    crate::native_webrtc_policy::NativeWebRTCRecoveryOptions {
                        force_restart,
                        preferred_negotiation_id: None,
                        role_override: None,
                    },
                )
                .await
            {
                let ctx = self.correlation_for_connection(connection_id).await;
                crate::clog!(
                    "[NativeWebRTC]",
                    &ctx,
                    "post_admission_upgrade_trigger_failed state={:?} force_restart={} error={}",
                    current_webrtc_state,
                    force_restart,
                    error
                );
            } else if force_restart {
                let ctx = self.correlation_for_connection(connection_id).await;
                crate::clog!(
                    "[NativeWebRTC]",
                    &ctx,
                    "post_admission_upgrade_restart_requested prior_state={:?}",
                    current_webrtc_state
                );
            }
        }
        #[cfg(not(all(not(target_arch = "wasm32"), feature = "transport-webrtc")))]
        {
            let _ = connection_id;
        }
    }

    #[cfg(native)]
    pub async fn inspect_incoming_native_main_frame(
        &self,
        connection_id: &str,
        remote_node_id: Option<&str>,
        known_device_id: Option<&str>,
        frame: &[u8],
    ) -> Result<crate::native_protocol::InspectedMainFrame, String> {
        use crate::native_protocol::{InspectedMainFrame, ParsedMainFrame};

        match crate::native_protocol::parse_main_frame(frame) {
            ParsedMainFrame::NativeMessage(message) => {
                let is_handshake = message.is_handshake();
                let is_session_token = message.is_session_token_presentation();

                // Session-token presentation: the connecting client presents a
                // token extracted from a compound ticket.  Validate it and admit
                // the connection before any handshake or data flows.
                if is_session_token && self.session_registry_active() {
                    let token = message
                        .presented_session_token()
                        .ok_or_else(|| "session-token-missing-in-presentation".to_string())?;
                    let token_payload = message.presented_session_token_payload();
                    let presented_device_id = message.claimed_device_id();
                    let token_fp = log_fingerprint(token.as_str());
                    let presentation_ctx = crate::client::correlation::CorrelationContext::new()
                        .connection_id(connection_id)
                        .token_fp(&token_fp);
                    crate::clog!(
                        "[PlutoRTC][session-token]",
                        &presentation_ctx,
                        "received_presentation token_len={}",
                        token.len()
                    );
                    // Inline path (no dedicated response-writer): keep
                    // pre-Phase-1 semantics so post-admission lifecycle
                    // hooks still fire for callers that aren't using the
                    // verdict-only `validate_session_token_for_connection`
                    // + explicit-flush ordering. The desktop consumer in
                    // `try_consume_pending_sdk_token_stream` *does* own a
                    // writer and explicitly bypasses this path.
                    self.validate_session_token_for_connection_with_payload_and_side_effects(
                        &token,
                        connection_id,
                        token_payload.as_deref(),
                    )
                    .await?;
                    if let Some(device_id) = presented_device_id.as_deref() {
                        let _ = self
                            .bind_session_admission_authoritative_device_id(
                                connection_id,
                                device_id,
                            )
                            .await;
                    }
                    let admitted_ctx = self
                        .correlation_for_connection(connection_id)
                        .await
                        .token_fp(&token_fp);
                    crate::clog!(
                        "[PlutoRTC][session-token]",
                        &admitted_ctx,
                        "validated_admitted"
                    );
                    // Token validated — return as ForwardOpaque so the app
                    // doesn't try to route this SDK-internal message.
                    return Ok(InspectedMainFrame::ForwardOpaque);
                }

                let claimed_device_id = message.claimed_device_id();
                let known_device_id_owned = known_device_id.map(ToOwned::to_owned);
                let admitted_device_id = if self.session_registry_active() {
                    if is_handshake {
                        // Handshake: full admission check, persists rejection on failure.
                        self.ensure_native_session_admitted(
                            connection_id,
                            remote_node_id,
                            known_device_id,
                            claimed_device_id.as_deref(),
                        )?
                    } else {
                        // Non-handshake: require existing admission without persisting
                        // rejection so a later handshake can still succeed.
                        self.require_existing_admission(
                            connection_id,
                            remote_node_id,
                            known_device_id,
                        )?
                    }
                } else {
                    None
                };

                // Only attach a handshake binding for actual handshake messages.
                let handshake = if is_handshake
                    && (claimed_device_id.is_some()
                        || known_device_id_owned.is_some()
                        || admitted_device_id.is_some())
                {
                    Some(crate::native_protocol::NativeHandshakeBinding {
                        known_device_id: known_device_id_owned,
                        claimed_device_id,
                        authoritative_device_id_hint: admitted_device_id
                            .clone()
                            .or_else(|| known_device_id.map(ToOwned::to_owned)),
                        admitted_device_id,
                    })
                } else {
                    None
                };

                Ok(InspectedMainFrame::NativeMessage { message, handshake })
            }
            ParsedMainFrame::TypeScriptHandshake(handshake) => {
                println!(
                    "[PlutoRTC] Received TS handshake on connection_id={} has_token={} registry_active={}",
                    connection_id,
                    handshake.session_token.is_some(),
                    self.session_registry_active()
                );
                if self.session_registry_active() {
                    let action = handshake.action.as_deref().unwrap_or("hello");
                    if action == "hello" {
                        match handshake.session_token.as_deref() {
                            Some(token) => {
                                // TS handshake path has no dedicated response writer
                                // we can flush before side effects, so use the
                                // pre-Phase-1 inline-side-effects helper.
                                self.validate_session_token_for_connection_with_payload_and_side_effects(
                                    token,
                                    connection_id,
                                    handshake.session_token_payload.as_deref(),
                                )
                                .await?;
                            }
                            None => {
                                // No token on a hello — this happens after a browser page
                                // refresh where the outbound connectPeer() path never ran
                                // (the desktop accepted the inbound transport and sent the
                                // hello via the native-signal-routing path, which has no
                                // pending token).  If the connection is already admitted
                                // from the previous session, reuse that admission so the
                                // capability-update handshake can proceed and trigger WebRTC
                                // upgrade.  If it is not admitted, require a fresh token.
                                self.require_existing_admission(
                                    connection_id,
                                    remote_node_id,
                                    known_device_id,
                                )?;
                            }
                        }
                    } else {
                        self.require_existing_admission(
                            connection_id,
                            remote_node_id,
                            known_device_id,
                        )?;
                    }
                }
                if let Some(device_id) = handshake.claimed_device_id.as_deref() {
                    let _ = self
                        .bind_session_admission_authoritative_device_id(connection_id, device_id)
                        .await;
                }
                self.maybe_handle_typescript_handshake_capabilities(
                    connection_id,
                    remote_node_id,
                    &handshake,
                )
                .await;
                Ok(InspectedMainFrame::ForwardOpaque)
            }
            ParsedMainFrame::TypeScriptJson(json) => {
                if self.session_registry_active() {
                    // Require existing admission without persisting rejection.
                    self.require_existing_admission(
                        connection_id,
                        remote_node_id,
                        known_device_id,
                    )?;
                }
                self.maybe_handle_typescript_json_frame(connection_id, remote_node_id, &json)
                    .await;
                Ok(InspectedMainFrame::ForwardOpaque)
            }
            ParsedMainFrame::Opaque => {
                if self.session_registry_active() {
                    // Require existing admission without persisting rejection.
                    self.require_existing_admission(
                        connection_id,
                        remote_node_id,
                        known_device_id,
                    )?;
                }
                Ok(InspectedMainFrame::ForwardOpaque)
            }
        }
    }

    #[cfg(native)]
    pub fn extract_native_handshake_device_id(&self, frame: &[u8]) -> Option<String> {
        match crate::native_protocol::parse_main_frame(frame) {
            crate::native_protocol::ParsedMainFrame::NativeMessage(message) => {
                message.claimed_device_id()
            }
            _ => None,
        }
    }

    pub fn session_registry_active(&self) -> bool {
        !self.session_token_registry.is_empty()
    }

    #[cfg(native)]
    pub fn ensure_native_stream_admitted(
        &self,
        connection_id: &str,
        remote_node_id: Option<&str>,
        known_device_id: Option<&str>,
    ) -> Result<Option<String>, String> {
        // File transfer streams should not permanently reject — only deny
        // if the connection hasn't been admitted yet.
        self.require_existing_admission(connection_id, remote_node_id, known_device_id)
    }

    pub fn session_admission(&self, connection_id: &str) -> crate::session_token::SessionAdmission {
        self.session_token_registry.admission(connection_id)
    }

    /// Complete an already-accepted session admission with the peer's
    /// authoritative device id once a later handshake/native binding proves it.
    ///
    /// Some reconnect paths validate the session token before the managed
    /// connection record has been tagged with the browser/desktop device id.
    /// The admission is valid, but trusted-device features that require an
    /// authoritative device id (for example drive-view bucket discovery) must
    /// not remain stuck in that incomplete state.
    pub async fn bind_session_admission_authoritative_device_id(
        &self,
        connection_id: &str,
        device_id: &str,
    ) -> bool {
        let device_id = device_id.trim();
        if device_id.is_empty() {
            return false;
        }

        use crate::session_token::SessionAdmission;

        match self.session_admission(connection_id) {
            SessionAdmission::Accepted {
                authoritative_device_id: Some(existing),
                ..
            } => existing == device_id,
            SessionAdmission::Accepted {
                scope: Some(scope),
                authoritative_device_id: None,
                ..
            } => {
                self.connection_manager
                    .set_device_id(connection_id, device_id.to_string())
                    .await;
                self.session_token_registry.bind_connection_scope(
                    connection_id,
                    scope.clone(),
                    Some(device_id.to_string()),
                );
                println!(
                    "[PlutoRTC][session-admission][late-authoritative-device] connection_id={} scope={} authoritative_device_id={}",
                    connection_id,
                    scope.as_str(),
                    device_id
                );
                true
            }
            SessionAdmission::Accepted {
                mechanism,
                scope: None,
                authoritative_device_id: None,
            } => {
                self.connection_manager
                    .set_device_id(connection_id, device_id.to_string())
                    .await;
                self.session_token_registry.mark_accepted(
                    connection_id,
                    mechanism,
                    None,
                    Some(device_id.to_string()),
                );
                println!(
                    "[PlutoRTC][session-admission][late-authoritative-device] connection_id={} scope= authoritative_device_id={}",
                    connection_id,
                    device_id
                );
                true
            }
            _ => false,
        }
    }

    pub fn reject_session_connection(&self, connection_id: &str, reason: &str) {
        self.session_token_registry
            .mark_rejected(connection_id, reason);
    }

    pub fn forget_session_connection(&self, connection_id: &str) {
        self.session_token_registry.forget_connection(connection_id);
    }

    pub fn ensure_native_session_admitted(
        &self,
        connection_id: &str,
        remote_node_id: Option<&str>,
        known_device_id: Option<&str>,
        claimed_device_id: Option<&str>,
    ) -> Result<Option<String>, String> {
        self.ensure_native_session_admitted_inner(
            connection_id,
            remote_node_id,
            known_device_id,
            claimed_device_id,
            true,
        )
    }

    /// Like `ensure_native_session_admitted` but does NOT permanently mark
    /// the connection as Rejected when admission fails.  Used for
    /// non-handshake messages so that a legitimate handshake arriving later
    /// can still succeed.
    #[cfg(native)]
    pub(crate) fn require_existing_admission(
        &self,
        connection_id: &str,
        remote_node_id: Option<&str>,
        known_device_id: Option<&str>,
    ) -> Result<Option<String>, String> {
        self.ensure_native_session_admitted_inner(
            connection_id,
            remote_node_id,
            known_device_id,
            None,
            false,
        )
    }

    pub(crate) fn ensure_native_session_admitted_inner(
        &self,
        connection_id: &str,
        remote_node_id: Option<&str>,
        known_device_id: Option<&str>,
        claimed_device_id: Option<&str>,
        persist_rejection: bool,
    ) -> Result<Option<String>, String> {
        use crate::session_token::{
            NativeTrustedConnectionContext, SessionAdmission, SessionAdmissionMechanism,
        };

        if !self.session_registry_active() {
            return Ok(None);
        }

        match self.session_token_registry.admission(connection_id) {
            SessionAdmission::Accepted {
                authoritative_device_id,
                ..
            } => {
                return Ok(authoritative_device_id);
            }
            SessionAdmission::Rejected { reason } => {
                return Err(reason);
            }
            SessionAdmission::Pending => {}
        }

        let context = NativeTrustedConnectionContext {
            connection_id: connection_id.to_string(),
            remote_node_id: remote_node_id.map(ToOwned::to_owned),
            known_device_id: known_device_id
                .map(str::trim)
                .filter(|value| !value.is_empty())
                .map(ToOwned::to_owned),
            claimed_device_id: claimed_device_id
                .map(str::trim)
                .filter(|value| !value.is_empty())
                .map(ToOwned::to_owned),
        };

        if let Some(authoritative_device_id) = self
            .session_token_registry
            .evaluate_trusted_native_connection(&context)
        {
            self.session_token_registry.mark_accepted(
                connection_id,
                SessionAdmissionMechanism::TrustedNativeBinding,
                None,
                Some(authoritative_device_id.clone()),
            );
            return Ok(Some(authoritative_device_id));
        }

        let reason = "session-token-required".to_string();
        if persist_rejection {
            self.session_token_registry
                .mark_rejected(connection_id, reason.clone());
        }
        Err(reason)
    }

    /// Register a session token in the registry.
    pub fn register_session_token(&self, token: String, scope: String, max_connections: u32) {
        self.session_token_registry.register(
            token,
            crate::session_token::GrantScope::from(scope),
            max_connections,
        );
    }

    /// Register a session token with an absolute Unix-millisecond expiry.
    pub fn register_session_token_with_expiry_ms(
        &self,
        token: String,
        scope: String,
        max_connections: u32,
        expires_at_ms: u64,
    ) {
        self.session_token_registry.register_with_expiry_ms(
            token,
            crate::session_token::GrantScope::from(scope),
            max_connections,
            Some(expires_at_ms),
        );
    }

    /// Clear all registered short-lived session tokens and admission state.
    pub fn clear_session_tokens(&self) {
        self.session_token_registry.clear();
        #[cfg(native)]
        if let Ok(mut cache) = self.managed_scope_tickets.write() {
            cache.clear();
        }
    }

    /// Keep the admission gate active even before any explicit share-style
    /// tokens are issued. This lets the app enforce "all native connections
    /// must be admitted" from startup onward while still allowing trusted
    /// native bindings to pass through the verifier path.
    pub fn ensure_default_admission_gate(&self, scope: &str) {
        if self.session_registry_active() {
            return;
        }

        let token = crate::session_token::generate_token();
        self.session_token_registry.register(
            token,
            crate::session_token::GrantScope::from(scope),
            0,
        );
    }

    /// Revoke a specific session token.
    pub fn revoke_session_token(&self, token: &str) {
        let token_fp = if token.len() > 8 {
            format!("{}…{}", &token[..4], &token[token.len() - 4..])
        } else {
            "(short)".to_string()
        };
        eprintln!("[PlutoRTC][teardown-trace] revoke_session_token token_fp={token_fp}");
        self.session_token_registry.revoke(token);
        #[cfg(native)]
        if let Ok(mut cache) = self.managed_scope_tickets.write() {
            // We only need to know whether the persistent managed scope token
            // is being removed; avoid cloning all matching scope keys.
            let remove_persistent_scope = cache.iter().any(|(scope, entry)| {
                scope.as_str() == PERSISTENT_MANAGED_ADMISSION_SCOPE && entry.token == token
            });
            cache.retain(|_, entry| entry.token != token);

            if remove_persistent_scope {
                if let Ok(base_dir_guard) = self.native_device_base_dir.try_read() {
                    if let Some(base_dir) = base_dir_guard.clone() {
                        let path = base_dir.join(format!(
                            "{}_{}.json",
                            PERSISTENT_MANAGED_SCOPE_TICKET_FILE_PREFIX,
                            PERSISTENT_MANAGED_ADMISSION_SCOPE
                        ));
                        match std::fs::remove_file(&path) {
                            Ok(()) => println!(
                                "[PlutoRTC][ticket][managed-persist-remove] scope={} path={}",
                                PERSISTENT_MANAGED_ADMISSION_SCOPE,
                                path.display()
                            ),
                            Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
                            Err(error) => eprintln!(
                                "[PlutoRTC][ticket][managed-persist-remove] failed scope={} path={} error={}",
                                PERSISTENT_MANAGED_ADMISSION_SCOPE,
                                path.display(),
                                error
                            ),
                        }
                    }
                }
            }
        }
    }

    /// Revoke all tokens with the given scope.
    pub async fn revoke_tokens_by_scope(&self, scope: &str) -> Vec<String> {
        let affected_connections = self
            .session_token_registry
            .revoke_by_scope(&crate::session_token::GrantScope::from(scope));
        #[cfg(native)]
        if let Ok(mut cache) = self.managed_scope_tickets.write() {
            cache.retain(|entry_scope, _| entry_scope != scope);
        }
        #[cfg(native)]
        let _ = self.delete_persisted_managed_scope_grant(scope).await;
        let mut endpoint_ids = HashSet::new();

        for connection_id in &affected_connections {
            if let Some(record) = self
                .connection_manager
                .get_by_connection_id(connection_id)
                .await
            {
                let endpoint_candidate = record
                    .endpoint_id
                    .clone()
                    .or_else(|| record.node_id.clone());
                if let Some(endpoint_id) = endpoint_candidate {
                    endpoint_ids.insert(endpoint_id);
                } else {
                    self.connection_manager
                        .set_closed(
                            connection_id,
                            Some(crate::lifecycle_reason::REASON_SESSION_TOKEN_REVOKED.to_string()),
                        )
                        .await;
                }
            }
        }

        let endpoint_disconnects = endpoint_ids.len();
        for endpoint_id in endpoint_ids {
            match endpoint_id.parse::<iroh::EndpointId>() {
                Ok(parsed) => {
                    let _ = self
                        .disconnect_with_reason(
                            parsed,
                            crate::lifecycle_reason::REASON_SESSION_TOKEN_REVOKED,
                        )
                        .await;
                }
                Err(_) => {
                    let records = self.connection_manager.get_by_node_id(&endpoint_id).await;
                    for record in records {
                        self.connection_manager
                            .set_closed(
                                &record.connection_id,
                                Some(
                                    crate::lifecycle_reason::REASON_SESSION_TOKEN_REVOKED
                                        .to_string(),
                                ),
                            )
                            .await;
                    }
                }
            }
        }

        eprintln!(
            "[PlutoRTC][teardown-trace] revoke_tokens_by_scope scope={} affected_connections={} endpoint_disconnects={}",
            scope,
            affected_connections.len(),
            endpoint_disconnects
        );

        affected_connections
    }

    #[cfg(native)]
    async fn managed_scope_grant_path(&self, scope: &str) -> anyhow::Result<std::path::PathBuf> {
        let base_dir = self
            .native_device_base_dir
            .read()
            .await
            .clone()
            .ok_or_else(|| anyhow::anyhow!("native device identity not initialized"))?;
        let sanitized_scope = scope
            .trim()
            .chars()
            .map(|value| match value {
                'a'..='z' | 'A'..='Z' | '0'..='9' | '-' | '_' => value,
                _ => '-',
            })
            .collect::<String>();
        Ok(base_dir.join(format!(
            "{}_{}.json",
            PERSISTENT_MANAGED_SCOPE_TICKET_FILE_PREFIX, sanitized_scope
        )))
    }

    #[cfg(native)]
    pub(crate) async fn load_persisted_managed_scope_grant(
        &self,
        scope: &str,
    ) -> anyhow::Result<Option<PersistedManagedScopeGrantRecord>> {
        let path = match self.managed_scope_grant_path(scope).await {
            Ok(path) => path,
            Err(_) => return Ok(None),
        };

        let payload = match tokio::fs::read_to_string(&path).await {
            Ok(payload) => payload,
            Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(None),
            Err(error) => {
                return Err(anyhow::Error::new(error).context(format!(
                    "failed reading persisted managed scope grant: {}",
                    path.display()
                )));
            }
        };

        let persisted = serde_json::from_str::<PersistedManagedScopeGrantRecord>(&payload)
            .with_context(|| {
                format!(
                    "failed parsing persisted managed scope grant: {}",
                    path.display()
                )
            })?;
        Ok(Some(persisted))
    }

    #[cfg(native)]
    pub(crate) async fn persist_managed_scope_grant(
        &self,
        scope: &str,
        token: &str,
        max_connections: u32,
    ) -> anyhow::Result<()> {
        if scope.trim() != PERSISTENT_MANAGED_ADMISSION_SCOPE {
            return Ok(());
        }

        let path = self.managed_scope_grant_path(scope).await?;
        if let Some(parent) = path.parent() {
            tokio::fs::create_dir_all(parent).await.with_context(|| {
                format!(
                    "failed creating managed scope persistence directory: {}",
                    parent.display()
                )
            })?;
        }

        let payload = PersistedManagedScopeGrantRecord {
            scope: scope.trim().to_string(),
            token: token.to_string(),
            max_connections,
        };
        let serialized = serde_json::to_vec_pretty(&payload)
            .context("failed serializing persisted managed scope grant")?;
        tokio::fs::write(&path, serialized)
            .await
            .with_context(|| format!("failed writing managed scope grant: {}", path.display()))?;
        println!(
            "[PlutoRTC][ticket][managed-persist-store] scope={} token_fp={} max_connections={} path={}",
            payload.scope,
            log_fingerprint(token),
            max_connections,
            path.display()
        );
        Ok(())
    }

    #[cfg(native)]
    pub(crate) async fn delete_persisted_managed_scope_grant(
        &self,
        scope: &str,
    ) -> anyhow::Result<()> {
        if scope.trim() != PERSISTENT_MANAGED_ADMISSION_SCOPE {
            return Ok(());
        }

        let path = match self.managed_scope_grant_path(scope).await {
            Ok(path) => path,
            Err(_) => return Ok(()),
        };
        match tokio::fs::remove_file(&path).await {
            Ok(()) => {
                println!(
                    "[PlutoRTC][ticket][managed-persist-remove] scope={} path={}",
                    scope.trim(),
                    path.display()
                );
                Ok(())
            }
            Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(()),
            Err(error) => Err(anyhow::Error::new(error).context(format!(
                "failed removing managed scope grant: {}",
                path.display()
            ))),
        }
    }

    #[cfg(native)]
    pub(crate) async fn rehydrate_persistent_managed_scope_ticket(
        &self,
        scope: &str,
    ) -> anyhow::Result<()> {
        if scope.trim() != PERSISTENT_MANAGED_ADMISSION_SCOPE {
            return Ok(());
        }

        let Some(persisted) = self.load_persisted_managed_scope_grant(scope).await? else {
            return Ok(());
        };

        self.session_token_registry.register(
            persisted.token.clone(),
            crate::session_token::GrantScope::from(persisted.scope.clone()),
            persisted.max_connections,
        );

        let mut cache = match self.managed_scope_tickets.write() {
            Ok(guard) => guard,
            Err(poisoned) => poisoned.into_inner(),
        };
        cache.insert(
            scope.trim().to_string(),
            CachedManagedScopeTicket {
                scope: crate::session_token::GrantScope::from(persisted.scope.clone()),
                token: persisted.token.clone(),
                max_connections: persisted.max_connections,
                compound_ticket: String::new(),
                iroh_ticket: String::new(),
            },
        );

        println!(
            "[PlutoRTC][ticket][managed-persist-rehydrate] scope={} token_fp={} max_connections={}",
            persisted.scope,
            log_fingerprint(persisted.token.as_str()),
            persisted.max_connections
        );
        Ok(())
    }

    pub async fn mark_connection_admitted_by_host(
        &self,
        connection_id: &str,
        scope: Option<&str>,
        authoritative_device_id: Option<String>,
    ) {
        // On some reconnect/refresh paths, the host approves the session token before
        // the admission layer has an authoritative device id handy. If we already
        // bound a device id to this connection (via native bind / handshake), use it
        // as the authoritative id so trusted-device flows (e.g. drive-view over
        // native WebRTC) don't get spuriously rejected.
        let authoritative_device_id = if authoritative_device_id.is_some() {
            authoritative_device_id
        } else {
            self.connection_manager
                .get_by_connection_id(connection_id)
                .await
                .and_then(|record| record.device_id)
        };

        let normalized_scope = scope
            .map(str::trim)
            .filter(|value| !value.is_empty())
            .map(ToOwned::to_owned);

        if let Some(scope_name) = normalized_scope.as_deref() {
            self.connection_manager
                .add_scope(connection_id, scope_name)
                .await;
            self.session_token_registry.bind_connection_scope(
                connection_id,
                crate::session_token::GrantScope::from(scope_name.to_string()),
                authoritative_device_id.clone(),
            );
        } else {
            self.session_token_registry.mark_accepted(
                connection_id,
                crate::session_token::SessionAdmissionMechanism::SessionToken,
                None,
                authoritative_device_id.clone(),
            );
        }

        #[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
        {
            self.accept_replacement_peer(
                connection_id,
                authoritative_device_id.as_deref(),
                "replacement-peer-admitted",
            )
            .await;
            let _ = self
                .clear_native_webrtc_suppression(connection_id, None)
                .await;
        }

        println!(
            "[PlutoRTC][session-admission][local-accept] connection_id={} scope={} authoritative_device_id={}",
            connection_id,
            normalized_scope.as_deref().unwrap_or(""),
            authoritative_device_id.as_deref().unwrap_or("pending")
        );
    }

    /// Admit a runtime-discovered managed user-device connection after the
    /// transport has been bound to an authoritative device id.
    ///
    /// Managed same-user device discovery is already an admission source for
    /// auto-connect; keep the session-token registry, connection scopes, and
    /// connection identity in sync so read-side admission guards do not strand
    /// a healthy connection at `session-admission-pending`.
    pub async fn mark_trusted_user_device_connection_admitted(
        &self,
        connection_id: &str,
        authoritative_device_id: &str,
    ) -> bool {
        let authoritative_device_id = authoritative_device_id.trim();
        if authoritative_device_id.is_empty() {
            return false;
        }

        self.connection_manager
            .set_device_id(connection_id, authoritative_device_id.to_string())
            .await;
        self.connection_manager
            .add_scope(connection_id, "user-device")
            .await;

        if self.session_registry_active() {
            match self.session_admission(connection_id) {
                crate::session_token::SessionAdmission::Accepted { .. } => {}
                crate::session_token::SessionAdmission::Rejected { .. } => return false,
                crate::session_token::SessionAdmission::Pending => {
                    self.session_token_registry.mark_accepted(
                        connection_id,
                        crate::session_token::SessionAdmissionMechanism::TrustedNativeBinding,
                        Some(crate::session_token::GrantScope::from("user-device")),
                        Some(authoritative_device_id.to_string()),
                    );
                }
            }
        }

        true
    }

    #[cfg(native)]
    pub fn set_native_trusted_connection_verifier(
        &self,
        verifier: Option<crate::session_token::NativeTrustedConnectionVerifier>,
    ) {
        self.session_token_registry
            .set_native_trusted_connection_verifier(verifier);
    }

    pub async fn present_session_token_to_host(
        &self,
        endpoint_id: iroh::EndpointId,
        connection_id: &str,
        token: &str,
    ) -> Result<String, String> {
        self.present_session_token(
            SessionTokenPresentationTarget::Host {
                endpoint_id,
                connection_id,
            },
            token,
            SessionTokenPresentationOptions::default(),
        )
        .await
    }

    pub async fn present_session_token_to_host_with_payload(
        &self,
        endpoint_id: iroh::EndpointId,
        connection_id: &str,
        token: &str,
        token_payload: Option<&str>,
    ) -> Result<String, String> {
        self.present_session_token(
            SessionTokenPresentationTarget::Host {
                endpoint_id,
                connection_id,
            },
            token,
            SessionTokenPresentationOptions {
                token_payload,
                device_id: None,
            },
        )
        .await
    }

    pub async fn present_session_token_to_host_with_payload_and_device_id(
        &self,
        endpoint_id: iroh::EndpointId,
        connection_id: &str,
        token: &str,
        token_payload: Option<&str>,
        device_id: Option<&str>,
    ) -> Result<String, String> {
        self.present_session_token(
            SessionTokenPresentationTarget::Host {
                endpoint_id,
                connection_id,
            },
            token,
            SessionTokenPresentationOptions {
                token_payload,
                device_id,
            },
        )
        .await
    }

    async fn present_session_token(
        &self,
        target: SessionTokenPresentationTarget<'_>,
        token: &str,
        options: SessionTokenPresentationOptions<'_>,
    ) -> Result<String, String> {
        let (endpoint_id, connection_id, is_endpoint_target) = match target {
            SessionTokenPresentationTarget::Host {
                endpoint_id,
                connection_id,
            } => (
                endpoint_id,
                std::borrow::Cow::Borrowed(connection_id),
                false,
            ),
            SessionTokenPresentationTarget::Endpoint { endpoint_id } => {
                let remote_node_id = endpoint_id.to_string();
                let local_node_id = self.current_node_id().await.ok_or_else(|| {
                    "missing local node id for session-token presentation".to_string()
                })?;
                (
                    endpoint_id,
                    std::borrow::Cow::Owned(Self::deterministic_connection_id(
                        &local_node_id,
                        &remote_node_id,
                    )),
                    true,
                )
            }
        };

        async fn read_native_main_message(
            recv: &mut iroh::endpoint::RecvStream,
        ) -> Result<crate::native_protocol::NativeMainMessage, String> {
            let mut protocol_byte = [0u8; 1];
            recv.read_exact(&mut protocol_byte)
                .await
                .map_err(|error| format!("[session-token-response:protocol-byte] {}", error))?;
            if protocol_byte[0] != 0x00 {
                return Err(format!(
                    "unexpected session-token response protocol byte: {}",
                    protocol_byte[0]
                ));
            }

            let mut label_len_buf = [0u8; 4];
            recv.read_exact(&mut label_len_buf)
                .await
                .map_err(|error| format!("[session-token-response:label-len] {}", error))?;
            let label_len = u32::from_be_bytes(label_len_buf) as usize;
            let mut label_buf = vec![0u8; label_len];
            recv.read_exact(&mut label_buf)
                .await
                .map_err(|error| format!("[session-token-response:label] {}", error))?;
            if String::from_utf8_lossy(&label_buf) != "main" {
                return Err("unexpected label in session-token response".to_string());
            }

            let mut frame_len_buf = [0u8; 4];
            recv.read_exact(&mut frame_len_buf)
                .await
                .map_err(|error| format!("[session-token-response:frame-len] {}", error))?;
            let frame_len = u32::from_be_bytes(frame_len_buf) as usize;
            let mut frame = vec![0u8; frame_len];
            recv.read_exact(&mut frame)
                .await
                .map_err(|error| format!("[session-token-response:frame-body] {}", error))?;

            match crate::native_protocol::parse_main_frame(&frame) {
                crate::native_protocol::ParsedMainFrame::NativeMessage(message) => Ok(message),
                other => Err(format!(
                    "unexpected session-token response frame: {:?}",
                    other
                )),
            }
        }

        let token_msg = crate::native_protocol::NativeMainMessage::session_token_presentation_with_payload_and_device_id(
            token,
            options.token_payload,
            options.device_id,
        );
        let serialized = serde_json::to_vec(&token_msg).map_err(|error| {
            format!("failed to serialize session-token presentation: {}", error)
        })?;

        // The dial path can briefly race connection registration (especially in wasm
        // during replacement churn). Give the endpoint a short window to appear
        // before failing the entire admission flow.
        let connection = {
            #[cfg(target_arch = "wasm32")]
            let deadline_ms = js_sys::Date::now() + 2_000.0;
            #[cfg(not(target_arch = "wasm32"))]
            let deadline = std::time::Instant::now() + std::time::Duration::from_millis(2_000);
            loop {
                if let Some(conn) = self.get_connection(endpoint_id).await {
                    break conn;
                }
                #[cfg(target_arch = "wasm32")]
                let expired = js_sys::Date::now() >= deadline_ms;
                #[cfg(not(target_arch = "wasm32"))]
                let expired = std::time::Instant::now() >= deadline;
                if expired {
                    return Err(format!(
                        "missing connection for session-token presentation: {}",
                        connection_id.as_ref()
                    ));
                }
                #[cfg(target_arch = "wasm32")]
                gloo_timers::future::sleep(std::time::Duration::from_millis(25)).await;
                #[cfg(not(target_arch = "wasm32"))]
                tokio::time::sleep(std::time::Duration::from_millis(25)).await;
            }
        };

        let (mut send, mut recv) = connection.open_bi().await.map_err(|error| {
            format!(
                "failed to open bi-stream for session-token presentation: {}",
                error
            )
        })?;

        println!(
            "[PlutoRTC][session-token-presentation] dialer sending token connection_id={} endpoint_id={} token_fp={} payload_len={}",
            connection_id.as_ref(),
            endpoint_id,
            super::core_impl::log_fingerprint(token),
            serialized.len(),
        );

        let label = b"main";
        let label_len = (label.len() as u32).to_be_bytes();
        let frame_len = (serialized.len() as u32).to_be_bytes();

        let mut buf = Vec::with_capacity(1 + 4 + label.len() + 4 + serialized.len());
        buf.push(0x00);
        buf.extend_from_slice(&label_len);
        buf.extend_from_slice(label);
        buf.extend_from_slice(&frame_len);
        buf.extend_from_slice(&serialized);

        tokio::io::AsyncWriteExt::write_all(&mut send, &buf)
            .await
            .map_err(|error| format!("failed to write session-token presentation: {}", error))?;
        // Intentionally do NOT call `send.finish()` here. On mobile (carrier
        // NAT / iOS Network.framework), finishing the send half before the
        // response arrives lets the inbound UDP pinhole close: there are no
        // more outbound stream frames to keep the path warm, and the host's
        // response packet gets dropped at the NAT, surfacing as "connection
        // lost" on the dialer's recv. The host parser at
        // `consume_pending_sdk_token_stream` reads length-prefixed bytes —
        // it never depends on seeing the FIN bit to know the payload boundary.
        // We finish the send half AFTER the response byte is read.
        println!(
            "[PlutoRTC][session-token-presentation] dialer write done connection_id={} endpoint_id={} awaiting host response",
            connection_id.as_ref(),
            endpoint_id,
        );

        // Bound the host-response read. A genuine granting host replies promptly,
        // but a peer that does not run the host token responder (e.g. a browser
        // dialing another user-owned browser in a mesh) would otherwise leave this
        // `read_exact` awaiting forever — wedging the entire managed dial so it
        // never returns, and starving the higher-level Client handshake (hello +
        // application key agreement) that the dialer must still perform. The
        // caller treats a presentation error as non-fatal, so a timeout simply
        // lets the connection fall through to the trust-based application route.
        let response = {
            use futures::FutureExt;
            let read_fut = read_native_main_message(&mut recv).fuse();
            futures::pin_mut!(read_fut);
            #[cfg(target_arch = "wasm32")]
            let timeout = gloo_timers::future::sleep(std::time::Duration::from_millis(
                SESSION_TOKEN_HOST_RESPONSE_TIMEOUT_MS,
            ))
            .fuse();
            #[cfg(not(target_arch = "wasm32"))]
            let timeout = tokio::time::sleep(std::time::Duration::from_millis(
                SESSION_TOKEN_HOST_RESPONSE_TIMEOUT_MS,
            ))
            .fuse();
            futures::pin_mut!(timeout);
            futures::select! {
                result = read_fut => result,
                _ = timeout => Err(format!(
                    "[session-token-response:timeout] no host response within {}ms",
                    SESSION_TOKEN_HOST_RESPONSE_TIMEOUT_MS
                )),
            }
        }
        .map_err(|error| {
            eprintln!(
                "[PlutoRTC][session-token-presentation] dialer recv FAILED connection_id={} endpoint_id={} token_fp={} error={}",
                connection_id.as_ref(),
                endpoint_id,
                super::core_impl::log_fingerprint(token),
                error
            );
            error
        })?;
        let _ = send.finish();
        if !response.is_session_token_response() {
            return Err(
                "host returned unexpected response to session-token presentation".to_string(),
            );
        }

        if response.session_token_approved() == Some(true) {
            let scope = response.approved_session_scope().unwrap_or_default();
            println!(
                "[PlutoRTC] Host approved session-token presentation connection_id={} endpoint_id={} scope={}",
                connection_id.as_ref(),
                endpoint_id,
                scope
            );
            if is_endpoint_target && !scope.is_empty() {
                println!(
                    "[PlutoRTC] Session-token presentation approved for connection_id={} scope={}",
                    connection_id.as_ref(),
                    scope
                );
            }
            return Ok(scope);
        }

        let reason = response
            .session_token_error()
            .unwrap_or_else(|| "session-token-rejected".to_string());
        Err(reason)
    }

    pub async fn present_session_token_to_endpoint(
        &self,
        endpoint_id: iroh::EndpointId,
        token: &str,
    ) -> Result<String, String> {
        self.present_session_token(
            SessionTokenPresentationTarget::Endpoint { endpoint_id },
            token,
            SessionTokenPresentationOptions::default(),
        )
        .await
    }

    pub async fn present_session_token_to_endpoint_with_payload(
        &self,
        endpoint_id: iroh::EndpointId,
        token: &str,
        token_payload: Option<&str>,
    ) -> Result<String, String> {
        self.present_session_token(
            SessionTokenPresentationTarget::Endpoint { endpoint_id },
            token,
            SessionTokenPresentationOptions {
                token_payload,
                device_id: None,
            },
        )
        .await
    }

    pub async fn present_session_token_to_endpoint_with_payload_and_device_id(
        &self,
        endpoint_id: iroh::EndpointId,
        token: &str,
        token_payload: Option<&str>,
        device_id: Option<&str>,
    ) -> Result<String, String> {
        self.present_session_token(
            SessionTokenPresentationTarget::Endpoint { endpoint_id },
            token,
            SessionTokenPresentationOptions {
                token_payload,
                device_id,
            },
        )
        .await
    }
}