openrtc 1.0.4

//! 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::*;

#[derive(Debug, Clone, PartialEq, Eq)]
enum IncomingApplicationAdmissionDecision {
    Forward,
    ClassifyAdmission,
    Reject(String),
}

fn incoming_application_admission_decision(
    registry_active: bool,
    admission: &crate::session_token::SessionAdmission,
    current_transport_is_admitted: bool,
) -> IncomingApplicationAdmissionDecision {
    use crate::session_token::SessionAdmission;

    if !registry_active {
        return IncomingApplicationAdmissionDecision::Forward;
    }
    if let SessionAdmission::Rejected { reason } = admission {
        return IncomingApplicationAdmissionDecision::Reject(reason.clone());
    }
    if current_transport_is_admitted {
        return IncomingApplicationAdmissionDecision::Forward;
    }
    IncomingApplicationAdmissionDecision::ClassifyAdmission
}

pub(super) fn session_scope_allows_device_binding(scope: &str) -> bool {
    let normalized = scope.trim();
    normalized == "user-device"
        || normalized == "persistent"
        || normalized == "trusted-device"
        || normalized.starts_with("trusted-device:")
}

pub(super) fn session_scope_uses_transient_peer_identity(scope: &str) -> bool {
    let normalized = scope.trim();
    !normalized.is_empty() && !session_scope_allows_device_binding(normalized)
}
#[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>,
    reciprocal_requested: bool,
}

fn outgoing_session_token_stream_contract(
    is_wasm: bool,
    application_crypto_active: bool,
    persistent_control_required: bool,
) -> crate::native_protocol::SessionTokenStreamContract {
    if is_wasm || (application_crypto_active && !persistent_control_required) {
        crate::native_protocol::SessionTokenStreamContract::OneShotAdmission
    } else {
        crate::native_protocol::SessionTokenStreamContract::PersistentControl
    }
}

#[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 {
    #[cfg(native)]
    async fn write_private_managed_scope_grant(
        path: &std::path::Path,
        payload: &[u8],
    ) -> anyhow::Result<()> {
        use tokio::io::AsyncWriteExt;

        let temp_path = path.with_extension(format!(
            "json.tmp-{}",
            crate::session_token::generate_payload_nonce()
        ));
        let mut options = tokio::fs::OpenOptions::new();
        options.write(true).create_new(true);
        #[cfg(unix)]
        {
            options.mode(0o600);
        }

        let write_result = async {
            let mut file = options.open(&temp_path).await.with_context(|| {
                format!(
                    "failed creating private managed scope grant: {}",
                    temp_path.display()
                )
            })?;
            file.write_all(payload).await.with_context(|| {
                format!(
                    "failed writing private managed scope grant: {}",
                    temp_path.display()
                )
            })?;
            file.flush().await.with_context(|| {
                format!(
                    "failed flushing private managed scope grant: {}",
                    temp_path.display()
                )
            })?;
            file.sync_all().await.with_context(|| {
                format!(
                    "failed syncing private managed scope grant: {}",
                    temp_path.display()
                )
            })?;
            drop(file);

            #[cfg(windows)]
            match tokio::fs::remove_file(path).await {
                Ok(()) => {}
                Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
                Err(error) => {
                    return Err(anyhow::Error::new(error).context(format!(
                        "failed replacing managed scope grant: {}",
                        path.display()
                    )));
                }
            }

            tokio::fs::rename(&temp_path, path).await.with_context(|| {
                format!(
                    "failed installing private managed scope grant: {}",
                    path.display()
                )
            })?;

            #[cfg(unix)]
            {
                use std::os::unix::fs::PermissionsExt;
                tokio::fs::set_permissions(path, std::fs::Permissions::from_mode(0o600))
                    .await
                    .with_context(|| {
                        format!(
                            "failed securing managed scope grant permissions: {}",
                            path.display()
                        )
                    })?;
            }

            Ok::<(), anyhow::Error>(())
        }
        .await;

        if write_result.is_err() {
            let _ = tokio::fs::remove_file(&temp_path).await;
        }
        write_result
    }

    pub(crate) fn admission_trace_enabled() -> bool {
        #[cfg(not(target_arch = "wasm32"))]
        {
            return std::env::var("OPENRTC_ADMISSION_VERBOSE")
                .ok()
                .map(|value| {
                    matches!(
                        value.trim().to_ascii_lowercase().as_str(),
                        "1" | "true" | "yes" | "on"
                    )
                })
                .unwrap_or(false);
        }

        #[cfg(target_arch = "wasm32")]
        {
            false
        }
    }

    /// Consume the SDK-owned session-token stream while this physical peer is
    /// pending admission or provisionally trusted through a durable native
    /// device binding. A trusted binding is local, directional evidence; the
    /// peer can still owe this side its session-token presentation. Once a
    /// session-token admission is recorded, streams pass through untouched.
    ///
    /// This method is the native single-consumer boundary: Tauri and other host
    /// bridges must call it before forwarding an accepted Iroh stream to an
    /// application runtime. Keeping the responder in Rust prevents webview
    /// startup, remount, or IPC subscription timing from stranding admission.
    #[cfg(not(target_arch = "wasm32"))]
    pub async fn route_incoming_bi_stream_for_admission(
        &self,
        remote_endpoint_id: iroh::EndpointId,
        transport_stable_id: u64,
        mut send: iroh::endpoint::SendStream,
        recv: iroh::endpoint::RecvStream,
    ) -> Result<IncomingBiStreamDisposition, String> {
        use crate::native_protocol::NativeMainMessage;
        use crate::session_token::SessionAdmission;

        let remote_node_id = remote_endpoint_id.to_string();
        let local_node_id = self
            .current_node_id()
            .await
            .ok_or_else(|| "missing local node id while routing incoming stream".to_string())?;
        let connection_id = Self::deterministic_connection_id(&local_node_id, &remote_node_id);

        let current_transport_stable_id = self
            .get_connection(remote_endpoint_id)
            .await
            .map(|connection| connection.stable_id() as u64);
        if current_transport_stable_id != Some(transport_stable_id) {
            return Err(format!(
                "stale incoming Iroh stream generation connection_id={} incoming_transport_stable_id={} current_transport_stable_id={:?}",
                connection_id, transport_stable_id, current_transport_stable_id
            ));
        }

        let registry_active = self.session_registry_active();
        let admission = self.session_admission(&connection_id);
        let session_token_admitted_logically =
            matches!(&admission, SessionAdmission::Accepted { .. });
        let application_stream_admitted = self
            .native_application_stream_admitted_for_transport(&connection_id, transport_stable_id);
        let admission_decision = incoming_application_admission_decision(
            registry_active,
            &admission,
            application_stream_admitted,
        );
        if let IncomingApplicationAdmissionDecision::Reject(reason) = &admission_decision {
            return Err(format!(
                "rejected connection cannot open application streams: {reason}"
            ));
        }
        let trusted_user_device_crypto_policy =
            self.trusted_user_device_application_crypto_is_required();
        if trusted_user_device_crypto_policy {
            self.set_connection_application_crypto_required(&connection_id);
        }
        let application_crypto_key =
            self.application_crypto_key_for_connection(Some(&connection_id));
        let application_crypto_ready = !self.connection_requires_application_crypto(&connection_id)
            || (application_crypto_key.is_some()
                && (!self.connection_requires_application_crypto_confirmation(&connection_id)
                    || self.connection_application_crypto_is_confirmed(&connection_id)));
        if admission_decision == IncomingApplicationAdmissionDecision::Forward
            && application_crypto_ready
        {
            if Self::admission_trace_enabled() {
                eprintln!(
                    "[OpenRTC][native-stream-router] classified application bi-stream connection_id={} remote_node_id={} transport_stable_id={} registry_active={} session_token_admitted_logically={} application_stream_admitted={} admission_projection={:?} inspected_bytes=0 disposition=forward",
                    connection_id,
                    remote_node_id,
                    transport_stable_id,
                    registry_active,
                    session_token_admitted_logically,
                    application_stream_admitted,
                    admission
                );
            }
            return Ok(IncomingBiStreamDisposition::Forward { send, recv });
        }

        if Self::admission_trace_enabled() {
            eprintln!(
                "[OpenRTC][native-stream-router] classifying SDK admission bi-stream connection_id={} remote_node_id={} admission_projection={:?} session_token_admitted_logically={} application_stream_admitted={}",
                connection_id,
                remote_node_id,
                admission,
                session_token_admitted_logically,
                application_stream_admitted
            );
        }

        // SDK admission control is plaintext inside Iroh's authenticated QUIC
        // stream. Product application crypto can be installed on only one side
        // when a path drops, so reauthorization must not depend on that key.
        // Retain encrypted-frame parsing for compatibility with older senders.
        let mut recv = recv;
        let mut wire_prefix = [0u8; 4];
        recv.read_exact(&mut wire_prefix)
            .await
            .map_err(|error| format!("pending admission wire prefix: {error}"))?;
        let wire_frame_len = u32::from_be_bytes(wire_prefix) as usize;

        // A token-approved route is still SDK-control-only until reciprocal
        // application key agreement is confirmed. Browser peers send their
        // key-agreement response and transport signaling as one-shot
        // native-main frames (`u32 length || 0x00 || JSON`). Consume those in
        // Rust instead of forwarding them to product code as encrypted data.
        // Observation does not advance lifecycle: only the existing handshake
        // handler may install/confirm the key and make the route eligible for
        // the Forward branch above.
        if (session_token_admitted_logically || trusted_user_device_crypto_policy)
            && !application_crypto_ready
            && wire_frame_len != 0
        {
            const MAX_PENDING_CONTROL_FRAME_BYTES: usize = 1024 * 1024;
            if wire_frame_len > MAX_PENDING_CONTROL_FRAME_BYTES {
                return Err(format!(
                    "pending application-crypto control frame too large: {wire_frame_len}"
                ));
            }
            let mut framed = vec![0u8; wire_frame_len];
            recv.read_exact(&mut framed)
                .await
                .map_err(|error| format!("pending application-crypto control frame: {error}"))?;
            let Some((&protocol, frame)) = framed.split_first() else {
                return Err("pending application-crypto control frame was empty".to_string());
            };
            if protocol != 0x00 {
                return Err(format!(
                    "product stream arrived before reciprocal application crypto was ready: protocol={protocol}"
                ));
            }
            match crate::native_protocol::parse_main_frame(frame) {
                crate::native_protocol::ParsedMainFrame::TypeScriptHandshake(handshake)
                    if trusted_user_device_crypto_policy
                        && handshake.application_key_agreement_public_key.is_some() =>
                {
                    self.maybe_handle_typescript_handshake_capabilities(
                        &connection_id,
                        Some(&remote_node_id),
                        &handshake,
                    )
                    .await;
                }
                crate::native_protocol::ParsedMainFrame::TypeScriptHandshake(_)
                | crate::native_protocol::ParsedMainFrame::TypeScriptJson(_)
                    if session_token_admitted_logically =>
                {
                    self.inspect_incoming_native_main_frame_for_transport(
                        &connection_id,
                        Some(&remote_node_id),
                        None,
                        Some(transport_stable_id),
                        frame,
                    )
                    .await?;
                }
                _ => {
                    return Err(
                        "non-control stream arrived before reciprocal application crypto was ready"
                            .to_string(),
                    );
                }
            }
            let _ = send.finish();
            if Self::admission_trace_enabled() {
                eprintln!(
                    "[OpenRTC][native-stream-router] consumed pre-confirmation SDK control frame connection_id={} remote_node_id={} transport_stable_id={} inspected_bytes={} disposition=consume",
                    connection_id,
                    remote_node_id,
                    transport_stable_id,
                    4 + wire_frame_len,
                );
            }
            return Ok(IncomingBiStreamDisposition::Consumed);
        }
        // Plain SDK framing starts `0x00 || u32(label_len)`, so its first four
        // bytes decode to zero. Older encrypted admission framing starts with a
        // non-zero ciphertext length; sniff it only for wire compatibility.
        let admission_stream_encrypted = application_crypto_key.is_some() && wire_frame_len != 0;
        let (mut send, mut recv) = match (application_crypto_key, admission_stream_encrypted) {
            (Some(key), true) => (
                crate::application_crypto_streams::PeerSendStream::encrypted(send, key),
                crate::application_crypto_streams::PeerRecvStream::encrypted_with_prefix(
                    recv,
                    key,
                    &wire_prefix,
                )
                .map_err(|error| format!("wrap encrypted pending admission stream: {error}"))?,
            ),
            _ => (
                crate::application_crypto_streams::PeerSendStream::plain(send),
                crate::application_crypto_streams::PeerRecvStream::plain(recv),
            ),
        };

        async fn read_admission_exact(
            recv: &mut crate::application_crypto_streams::PeerRecvStream,
            buffer: &mut [u8],
            context: &str,
        ) -> Result<(), String> {
            let mut offset = 0;
            while offset < buffer.len() {
                let read = recv
                    .read(&mut buffer[offset..])
                    .await
                    .map_err(|error| format!("{context}: {error}"))?;
                if read == 0 {
                    return Err(format!("{context}: stream finished early"));
                }
                offset += read;
            }
            Ok(())
        }

        const MAX_LABEL_BYTES: usize = 64;
        const MAX_NATIVE_MAIN_FRAME_BYTES: usize = 1024 * 1024;

        let mut protocol = [0u8; 1];
        if admission_stream_encrypted {
            read_admission_exact(&mut recv, &mut protocol, "pending admission protocol byte")
                .await?;
        } else {
            protocol[0] = wire_prefix[0];
        }
        if protocol[0] != 0x00 {
            return Err(format!(
                "pending admission expected native protocol byte 0, received {}",
                protocol[0]
            ));
        }

        let mut label_len = [0u8; 4];
        if admission_stream_encrypted {
            read_admission_exact(&mut recv, &mut label_len, "pending admission label length")
                .await?;
        } else {
            label_len[..3].copy_from_slice(&wire_prefix[1..]);
            read_admission_exact(
                &mut recv,
                &mut label_len[3..],
                "pending admission label length",
            )
            .await?;
        }
        let label_len = u32::from_be_bytes(label_len) as usize;
        if label_len > MAX_LABEL_BYTES {
            return Err(format!("pending admission label too large: {label_len}"));
        }
        let mut label = vec![0u8; label_len];
        read_admission_exact(&mut recv, &mut label, "pending admission label").await?;
        if label != b"main" {
            return Err(format!(
                "pending admission expected main label, received {}",
                String::from_utf8_lossy(&label)
            ));
        }

        let mut frame_len = [0u8; 4];
        read_admission_exact(&mut recv, &mut frame_len, "pending admission frame length").await?;
        let frame_len = u32::from_be_bytes(frame_len) as usize;
        if frame_len > MAX_NATIVE_MAIN_FRAME_BYTES {
            return Err(format!("pending admission frame too large: {frame_len}"));
        }
        let mut frame = vec![0u8; frame_len];
        read_admission_exact(&mut recv, &mut frame, "pending admission frame").await?;

        let parsed_frame = crate::native_protocol::parse_main_frame(&frame);
        if (session_token_admitted_logically || trusted_user_device_crypto_policy)
            && !application_crypto_ready
            && matches!(
                &parsed_frame,
                crate::native_protocol::ParsedMainFrame::TypeScriptHandshake(_)
                    | crate::native_protocol::ParsedMainFrame::TypeScriptJson(_)
            )
        {
            match &parsed_frame {
                crate::native_protocol::ParsedMainFrame::TypeScriptHandshake(handshake)
                    if trusted_user_device_crypto_policy
                        && handshake.application_key_agreement_public_key.is_some() =>
                {
                    self.install_native_control_stream(
                        &connection_id,
                        remote_endpoint_id,
                        transport_stable_id,
                        send.into_plain(),
                        recv.into_plain(),
                        "application-crypto-negotiation",
                    )
                    .await;
                    self.maybe_handle_typescript_handshake_capabilities(
                        &connection_id,
                        Some(&remote_node_id),
                        handshake,
                    )
                    .await;
                    if Self::admission_trace_enabled() {
                        eprintln!(
                            "[OpenRTC][native-stream-router] installed labeled application-crypto control route connection_id={} remote_node_id={} transport_stable_id={} inspected_bytes={} disposition=consume",
                            connection_id,
                            remote_node_id,
                            transport_stable_id,
                            1 + 4 + label_len + 4 + frame_len,
                        );
                    }
                    return Ok(IncomingBiStreamDisposition::Consumed);
                }
                _ if session_token_admitted_logically => {
                    self.inspect_incoming_native_main_frame_for_transport(
                        &connection_id,
                        Some(&remote_node_id),
                        None,
                        Some(transport_stable_id),
                        &frame,
                    )
                    .await?;
                }
                _ => {
                    return Err(
                        "non-key-agreement stream arrived before role admission".to_string()
                    );
                }
            }
            let _ = send.finish();
            if Self::admission_trace_enabled() {
                eprintln!(
                    "[OpenRTC][native-stream-router] consumed labeled pre-confirmation SDK control frame connection_id={} remote_node_id={} transport_stable_id={} inspected_bytes={} disposition=consume",
                    connection_id,
                    remote_node_id,
                    transport_stable_id,
                    1 + 4 + label_len + 4 + frame_len,
                );
            }
            return Ok(IncomingBiStreamDisposition::Consumed);
        }

        let message = match parsed_frame {
            crate::native_protocol::ParsedMainFrame::NativeMessage(message)
                if message.is_session_token_presentation() =>
            {
                message
            }
            _ => {
                if Self::admission_trace_enabled() {
                    eprintln!(
                        "[OpenRTC][native-stream-router] rejected non-token stream before role admission connection_id={} remote_node_id={} admission={:?} inspected_bytes={} disposition=deny",
                        connection_id,
                        remote_node_id,
                        admission,
                        1 + 4 + label_len + 4 + frame_len
                    );
                }
                return Err(
                    "current transport requires an SDK session-token presentation".to_string(),
                );
            }
        };
        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 claimed_device_id = message.claimed_device_id();
        let stream_contract = message.session_token_stream_contract();
        let reciprocal_requested = message.session_token_reciprocal_requested();
        let response_ack_requested = message.session_token_response_ack_requested();
        let was_already_admitted = self
            .session_token_registry
            .is_session_token_admitted_for_connection(&connection_id);
        let response_guard = self
            .session_token_registry
            .try_begin_admission_response(&connection_id)
            .ok_or_else(|| "session-token-retirement-in-progress".to_string())?;

        let mut verdict = self
            .validate_session_token_for_connection_with_payload(
                &token,
                &connection_id,
                token_payload.as_deref(),
            )
            .await;
        if matches!(verdict.as_deref(), Ok("user-device")) {
            let expected_device_id = self
                .known_remote_device_id_for_incoming_transport(&connection_id, &remote_node_id)
                .await;
            if claimed_device_id.as_deref() != expected_device_id.as_deref() {
                let reason = format!(
                    "authoritative-device-mismatch: expected={:?} claimed={:?}",
                    expected_device_id, claimed_device_id,
                );
                self.reject_session_connection(&connection_id, reason.as_str());
                verdict = Err(reason);
            } else if let Some(device_id) = claimed_device_id.as_deref() {
                if !self
                    .bind_session_admission_authoritative_device_id(&connection_id, device_id)
                    .await
                {
                    let reason = "authoritative-device-bind-refused".to_string();
                    self.reject_session_connection(&connection_id, reason.as_str());
                    verdict = Err(reason);
                }
            }
        }
        let mut response = match &verdict {
            Ok(scope) => NativeMainMessage::session_token_approval(
                (!scope.is_empty()).then_some(scope.as_str()),
                &connection_id,
            ),
            Err(reason) => NativeMainMessage::session_token_rejection(reason, &connection_id),
        };
        let inline_reciprocal_stream_instance_id = format!(
            "native-admission:{}:{}",
            transport_stable_id,
            crate::session_token::generate_token(),
        );
        if verdict.is_ok()
            && reciprocal_requested
            && message.reciprocal_session_token_mode() == Some("inline-v1")
        {
            let credential = self
                .native_route_repair_credentials
                .read()
                .ok()
                .and_then(|credentials| credentials.get(&remote_node_id).cloned());
            let known_remote_device_id = self
                .known_remote_device_id_for_incoming_transport(&connection_id, &remote_node_id)
                .await;
            let local_device_id = self
                .active_session_identity()
                .map(|(_, device_id)| device_id)
                .filter(|device_id| !device_id.trim().is_empty());
            if let (Some(credential), Some(local_device_id)) = (credential, local_device_id) {
                if known_remote_device_id.as_deref()
                    == Some(credential.authoritative_device_id.as_str())
                {
                    let presentation_id = crate::session_token::generate_token();
                    match self
                        .prepare_inline_reciprocal_session_admission(
                            remote_endpoint_id,
                            transport_stable_id,
                            inline_reciprocal_stream_instance_id.as_str(),
                            presentation_id.as_str(),
                            credential.token.as_str(),
                            credential.token_payload.as_str(),
                            local_device_id.as_str(),
                            stream_contract,
                        )
                        .await
                    {
                        Ok(presentation) => {
                            response = response
                                .with_reciprocal_session_token_presentation(presentation);
                        }
                        Err(error) => eprintln!(
                            "[PlutoRTC][session-admission][inline-reciprocal-prepare] connection_id={} remote_node_id={} result=deferred error={}",
                            connection_id, remote_node_id, error,
                        ),
                    }
                } else {
                    eprintln!(
                        "[PlutoRTC][session-admission][inline-reciprocal-prepare] connection_id={} remote_node_id={} result=deferred reason=authoritative-device-mismatch",
                        connection_id, remote_node_id,
                    );
                }
            }
        }
        let serialized = serde_json::to_vec(&response)
            .map_err(|error| format!("serialize session-token response: {error}"))?;
        let mut response_frame = Vec::with_capacity(1 + 4 + 4 + 4 + serialized.len());
        response_frame.push(0x00);
        response_frame.extend_from_slice(&(4u32).to_be_bytes());
        response_frame.extend_from_slice(b"main");
        response_frame.extend_from_slice(&(serialized.len() as u32).to_be_bytes());
        response_frame.extend_from_slice(&serialized);
        send.write_all(&response_frame)
            .await
            .map_err(|error| format!("write session-token response: {error}"))?;
        send.flush()
            .await
            .map_err(|error| format!("flush session-token response: {error}"))?;
        let response_ack_message = if response_ack_requested
            && !admission_stream_encrypted
            && verdict.is_ok()
        {
            let ack_result = tokio::time::timeout(std::time::Duration::from_secs(2), async {
                let mut protocol = [0u8; 1];
                read_admission_exact(
                    &mut recv,
                    &mut protocol,
                    "session-token response ack protocol byte",
                )
                .await?;
                if protocol[0] != 0x00 {
                    return Err(format!(
                        "session-token response ack expected native protocol byte 0, received {}",
                        protocol[0]
                    ));
                }
                let mut label_len = [0u8; 4];
                read_admission_exact(
                    &mut recv,
                    &mut label_len,
                    "session-token response ack label length",
                )
                .await?;
                let label_len = u32::from_be_bytes(label_len) as usize;
                if label_len > MAX_LABEL_BYTES {
                    return Err(format!(
                        "session-token response ack label too large: {label_len}"
                    ));
                }
                let mut label = vec![0u8; label_len];
                read_admission_exact(&mut recv, &mut label, "session-token response ack label")
                    .await?;
                if label != b"main" {
                    return Err("session-token response ack expected main label".to_string());
                }
                let mut frame_len = [0u8; 4];
                read_admission_exact(
                    &mut recv,
                    &mut frame_len,
                    "session-token response ack frame length",
                )
                .await?;
                let frame_len = u32::from_be_bytes(frame_len) as usize;
                if frame_len > MAX_NATIVE_MAIN_FRAME_BYTES {
                    return Err(format!(
                        "session-token response ack frame too large: {frame_len}"
                    ));
                }
                let mut frame = vec![0u8; frame_len];
                read_admission_exact(&mut recv, &mut frame, "session-token response ack frame")
                    .await?;
                match crate::native_protocol::parse_main_frame(&frame) {
                    crate::native_protocol::ParsedMainFrame::NativeMessage(message)
                        if message.is_session_token_response_ack(&connection_id) =>
                    {
                        Ok(message)
                    }
                    _ => Err("session-token response ack frame was not a matching ACK".to_string()),
                }
            })
            .await;
            match ack_result {
                Ok(Ok(message)) => Some(message),
                Ok(Err(error)) => {
                    eprintln!(
                        "[PlutoRTC][session-token] response ACK rejected connection_id={} remote_node_id={} error={}",
                        connection_id, remote_node_id, error,
                    );
                    None
                }
                Err(_) => {
                    eprintln!(
                        "[PlutoRTC][session-token] response ACK timed out connection_id={} remote_node_id={}",
                        connection_id, remote_node_id,
                    );
                    None
                }
            }
        } else {
            None
        };
        let response_acknowledged = !response_ack_requested || response_ack_message.is_some();
        if Self::admission_trace_enabled() {
            eprintln!(
                "[OpenRTC][native-stream-router] classified SDK session-token presentation connection_id={} remote_node_id={} prior_admission={:?} inspected_bytes={} accepted={} disposition=consume",
                connection_id,
                remote_node_id,
                admission,
                1 + 4 + label_len + 4 + frame_len,
                verdict.is_ok()
            );
        }
        let scope = match verdict {
            Ok(scope) => scope,
            Err(reason) => {
                eprintln!(
                    "[PlutoRTC][session-token] native host rejected connection_id={} remote_node_id={} reason={}",
                    connection_id, remote_node_id, reason
                );
                return Ok(IncomingBiStreamDisposition::Consumed);
            }
        };

        // A written verdict is not yet a generation-bound security proof. If
        // its requested ACK never arrives, the presenter cannot know it was
        // approved and must be allowed to retry on this same physical route.
        // Leaving the inbound proof unset keeps the Rust admission router in
        // ClassifyAdmission instead of misrouting that retry as encrypted
        // product data.
        if !response_acknowledged {
            drop(response_guard);
            eprintln!(
                "[PlutoRTC][session-token] approval delivery unacknowledged connection_id={} remote_node_id={} action=leave-route-pending",
                connection_id, remote_node_id,
            );
            return Ok(IncomingBiStreamDisposition::Consumed);
        }

        let effective_stream_contract = if response_acknowledged {
            stream_contract
        } else {
            crate::native_protocol::SessionTokenStreamContract::OneShotAdmission
        };
        self.bind_inbound_native_admission_stream_contract(
            &connection_id,
            effective_stream_contract,
        );
        self.bind_inbound_session_token_transport(&connection_id, transport_stable_id);
        if reciprocal_requested {
            self.request_reciprocal_session_admission(&connection_id);
        }

        self.connection_manager
            .upsert_pending(
                connection_id.clone(),
                Some(remote_node_id.clone()),
                None,
                Some(remote_node_id.clone()),
            )
            .await;
        if self
            .get_connection(remote_endpoint_id)
            .await
            .is_some_and(|connection| connection.stable_id() as u64 == transport_stable_id)
        {
            self.connection_manager
                .set_connected_with_transport(
                    &connection_id,
                    Some(remote_node_id.clone()),
                    Some(transport_stable_id),
                    Some("incoming".to_string()),
                )
                .await;
        } else {
            self.connection_manager
                .set_connected(&connection_id, Some(remote_node_id.clone()))
                .await;
        }
        // Token validation can complete before the incoming connection has
        // been indexed. Re-apply the accepted scope after materializing the
        // record so the Rust-owned peer-session projection cannot lose it.
        if !scope.trim().is_empty() {
            self.connection_manager
                .add_scope(&connection_id, scope.as_str())
                .await;
        }
        let authoritative_device_bound = if let Some(device_id) = claimed_device_id.as_deref() {
            let bound = self
                .bind_session_admission_authoritative_device_id(&connection_id, device_id)
                .await;
            if scope == "user-device"
                && self.resume_peer_requested_auto_connect_if_authenticated(device_id)
            {
                println!(
                    "[PlutoRTC][session-admission][peer-reconnect] cleared peer-requested auto-connect suppression connection_id={} device_id={}",
                    connection_id, device_id,
                );
            }
            bound
        } else {
            false
        };
        let reciprocal_committed = if let Some(ack) = response_ack_message.as_ref() {
            if let (Some(presentation_id), Some(accepted)) = (
                ack.reciprocal_session_token_ack_presentation_id(),
                ack.reciprocal_session_token_ack_accepted(),
            ) {
                match self
                    .confirm_inline_reciprocal_session_admission(
                        remote_endpoint_id,
                        transport_stable_id,
                        inline_reciprocal_stream_instance_id.as_str(),
                        presentation_id.as_str(),
                        accepted,
                        ack.reciprocal_session_token_ack_scope().as_deref(),
                    )
                    .await
                {
                    Ok(()) => true,
                    Err(error) => {
                        eprintln!(
                            "[PlutoRTC][session-admission][inline-reciprocal-commit] connection_id={} remote_node_id={} result=pending error={}",
                            connection_id, remote_node_id, error,
                        );
                        false
                    }
                }
            } else {
                false
            }
        } else {
            false
        };
        if reciprocal_committed {
            self.clear_reciprocal_session_admission_request(&connection_id);
        }
        match effective_stream_contract {
            crate::native_protocol::SessionTokenStreamContract::PersistentControl => {
                if admission_stream_encrypted {
                    // Keyed reauthorization is always one-shot. Treat an older
                    // peer's persistent marker as one-shot too rather than
                    // installing a raw control reader behind the crypto frame.
                    let _ = send.finish();
                    drop(recv);
                } else {
                    self.install_native_control_stream(
                        &connection_id,
                        remote_endpoint_id,
                        transport_stable_id,
                        send.into_plain(),
                        recv.into_plain(),
                        "admission-host",
                    )
                    .await;
                }
            }
            crate::native_protocol::SessionTokenStreamContract::OneShotAdmission => {
                if let Err(error) = send
                    .finish_and_wait_for_peer(std::time::Duration::from_secs(2))
                    .await
                {
                    eprintln!(
                        "[PlutoRTC][session-token] host response delivery wait ended connection_id={} remote_node_id={} error={}",
                        connection_id, remote_node_id, error,
                    );
                }
                drop(recv);
            }
        }
        // Keep terminal retirement fenced until a one-shot verdict has either
        // reached the peer or exhausted its bounded delivery wait.
        drop(response_guard);
        // The initial incoming-transport readiness probe runs before
        // `accept_external_connection` has installed this leg in the native
        // node, so it may conservatively record Stale. Admission is the first
        // point where both the current physical generation and its Rust-owned
        // application route are proven. Re-confirm here, fenced to that exact
        // generation, so the accepting host projects the same routable truth as
        // the dialer without introducing another lifecycle owner.
        if self
            .connection_manager
            .current_transport_matches(&connection_id, Some(transport_stable_id))
            .await
        {
            let _ = self
                .confirm_managed_connection_readiness(&connection_id)
                .await;
        }
        if !admission_stream_encrypted {
            if let Err(error) = self
                .send_typescript_capability_update(&connection_id, "native-admission-host", None)
                .await
            {
                eprintln!(
                    "[OpenRTC][capability] native admission host advertisement failed connection_id={} error={}",
                    connection_id, error,
                );
            }
        }
        self.run_post_session_token_admission_side_effects(&connection_id, !was_already_admitted)
            .await;
        eprintln!(
            "[PlutoRTC][session-token] native host approved connection_id={} remote_node_id={} scope={} claimed_device_id={} authoritative_device_bound={}",
            connection_id,
            remote_node_id,
            scope,
            claimed_device_id.as_deref().unwrap_or("<none>"),
            authoritative_device_bound,
        );
        Ok(IncomingBiStreamDisposition::Consumed)
    }

    #[cfg(not(target_arch = "wasm32"))]
    pub(crate) async fn native_peer_is_pending_admission(
        &self,
        remote_endpoint_id: iroh::EndpointId,
    ) -> bool {
        if !self.session_registry_active() {
            return false;
        }
        let Some(local_node_id) = self.current_node_id().await else {
            return true;
        };
        let connection_id =
            Self::deterministic_connection_id(&local_node_id, &remote_endpoint_id.to_string());
        let Some(transport_stable_id) = self
            .get_connection(remote_endpoint_id)
            .await
            .map(|connection| connection.stable_id() as u64)
        else {
            return true;
        };
        !self.native_application_stream_admitted_for_transport(&connection_id, transport_stable_id)
    }

    #[cfg(not(target_arch = "wasm32"))]
    pub(crate) fn inbound_session_token_admitted_for_transport(
        &self,
        connection_id: &str,
        transport_stable_id: u64,
    ) -> bool {
        self.inbound_session_admission_transport_ids
            .read()
            .ok()
            .and_then(|proofs| proofs.get(connection_id).copied())
            == Some(transport_stable_id)
    }

    #[cfg(not(target_arch = "wasm32"))]
    pub(crate) fn remote_session_token_admitted_for_transport(
        &self,
        connection_id: &str,
        transport_stable_id: u64,
    ) -> bool {
        self.remote_session_admission_proofs
            .read()
            .ok()
            .and_then(|proofs| {
                proofs
                    .get(connection_id)
                    .map(|proof| proof.transport_stable_id)
            })
            == Some(transport_stable_id)
    }

    /// Product streams are allowed once this runtime has accepted the logical
    /// session and holds every directional proof declared by the wire contract
    /// on the current physical generation. A one-shot presentation proves only
    /// the receiver's inbound direction. Persistent native control declares the
    /// directions that must remain generation-bound for managed sessions.
    #[cfg(not(target_arch = "wasm32"))]
    pub(crate) fn native_application_stream_admitted_for_transport(
        &self,
        connection_id: &str,
        transport_stable_id: u64,
    ) -> bool {
        let admission = self.session_admission(connection_id);
        if !matches!(
            admission,
            crate::session_token::SessionAdmission::Accepted { .. }
        ) {
            return false;
        }

        let contracts = self.native_admission_stream_contracts(connection_id);
        // A durable user-device relationship is bilateral. After this side's
        // outbound presentation succeeds, the peer can still owe its reverse
        // presentation on the same physical Iroh generation. Treating the
        // single outbound proof as application-ready forwards that reverse
        // token frame into product code and strands the peer asymmetrically.
        let persistent_control_declared = matches!(
            contracts.inbound,
            Some(crate::native_protocol::SessionTokenStreamContract::PersistentControl)
        ) || matches!(
            contracts.outbound,
            Some(crate::native_protocol::SessionTokenStreamContract::PersistentControl)
        );
        let bilateral_user_device = persistent_control_declared
            && matches!(
                &admission,
                crate::session_token::SessionAdmission::Accepted {
                    scope: Some(scope),
                    ..
                } if scope.as_str() == "user-device"
            );
        let inbound_required = bilateral_user_device || contracts.inbound.is_some();
        let outbound_required = bilateral_user_device || contracts.outbound.is_some();
        if !inbound_required && !outbound_required {
            return false;
        }

        let inbound_proven =
            self.inbound_session_token_admitted_for_transport(connection_id, transport_stable_id);
        let outbound_proven =
            self.remote_session_token_admitted_for_transport(connection_id, transport_stable_id);
        let admission_proven =
            (!inbound_required || inbound_proven) && (!outbound_required || outbound_proven);
        if !admission_proven {
            return false;
        }

        if self.connection_requires_application_crypto(connection_id) {
            if self
                .application_crypto_key_for_connection(Some(connection_id))
                .is_none()
            {
                return false;
            }
            if self.connection_requires_application_crypto_confirmation(connection_id)
                && !self.connection_application_crypto_is_confirmed(connection_id)
            {
                return false;
            }
        }
        true
    }

    #[cfg(not(target_arch = "wasm32"))]
    pub(crate) fn native_application_stream_pending_diagnostic(
        &self,
        connection_id: &str,
        transport_stable_id: u64,
    ) -> String {
        let contracts = self.native_admission_stream_contracts(connection_id);
        let admission = self.session_admission(connection_id);
        let bilateral_user_device = matches!(
            &admission,
            crate::session_token::SessionAdmission::Accepted {
                scope: Some(scope),
                ..
            } if scope.as_str() == "user-device"
        );
        let inbound_required = bilateral_user_device || contracts.inbound.is_some();
        let outbound_required = bilateral_user_device || contracts.outbound.is_some();
        let inbound_proven =
            self.inbound_session_token_admitted_for_transport(connection_id, transport_stable_id);
        let outbound_proven =
            self.remote_session_token_admitted_for_transport(connection_id, transport_stable_id);
        let crypto_required = self.connection_requires_application_crypto(connection_id);
        let crypto_key_installed = self
            .application_crypto_key_for_connection(Some(connection_id))
            .is_some();
        let crypto_confirmation_required =
            self.connection_requires_application_crypto_confirmation(connection_id);
        let crypto_confirmed = self.connection_application_crypto_is_confirmed(connection_id);

        format!(
            "admission={admission:?} contracts={contracts:?} inbound_required={inbound_required} inbound_proven={inbound_proven} outbound_required={outbound_required} outbound_proven={outbound_proven} crypto_required={crypto_required} crypto_key_installed={crypto_key_installed} crypto_confirmation_required={crypto_confirmation_required} crypto_confirmed={crypto_confirmed}",
        )
    }

    #[cfg(not(target_arch = "wasm32"))]
    pub(crate) fn bind_inbound_session_token_transport(
        &self,
        connection_id: &str,
        transport_stable_id: u64,
    ) {
        if let Ok(mut proofs) = self.inbound_session_admission_transport_ids.write() {
            proofs.insert(connection_id.to_string(), transport_stable_id);
        }
        println!(
            "[PlutoRTC][session-admission][inbound-proof] connection_id={} transport_stable_id={}",
            connection_id, transport_stable_id,
        );
    }

    #[cfg(not(target_arch = "wasm32"))]
    pub(crate) fn request_reciprocal_session_admission(&self, connection_id: &str) {
        if let Ok(mut requests) = self.pending_reciprocal_session_admission_requests.write() {
            requests.insert(connection_id.to_string());
        }
        println!(
            "[PlutoRTC][session-admission][reciprocal-request] connection_id={} state=pending",
            connection_id,
        );
    }

    #[cfg(not(target_arch = "wasm32"))]
    pub(crate) fn has_pending_reciprocal_session_admission_request(
        &self,
        connection_id: &str,
    ) -> bool {
        self.pending_reciprocal_session_admission_requests
            .read()
            .ok()
            .is_some_and(|requests| requests.contains(connection_id))
    }

    #[cfg(not(target_arch = "wasm32"))]
    pub(crate) fn clear_reciprocal_session_admission_request(&self, connection_id: &str) {
        let removed = self
            .pending_reciprocal_session_admission_requests
            .write()
            .ok()
            .is_some_and(|mut requests| requests.remove(connection_id));
        if removed {
            println!(
                "[PlutoRTC][session-admission][reciprocal-request] connection_id={} state=satisfied",
                connection_id,
            );
        }
    }

    #[cfg(not(target_arch = "wasm32"))]
    pub(crate) fn bind_inbound_native_admission_stream_contract(
        &self,
        connection_id: &str,
        contract: crate::native_protocol::SessionTokenStreamContract,
    ) {
        if let Ok(mut contracts) = self.native_admission_stream_contracts.write() {
            contracts
                .entry(connection_id.to_string())
                .or_default()
                .inbound = Some(contract);
        }
        println!(
            "[PlutoRTC][session-admission][stream-contract] connection_id={} direction=inbound contract={:?}",
            connection_id, contract,
        );
    }

    #[cfg(not(target_arch = "wasm32"))]
    pub(crate) fn bind_outbound_native_admission_stream_contract(
        &self,
        connection_id: &str,
        contract: crate::native_protocol::SessionTokenStreamContract,
    ) {
        if let Ok(mut contracts) = self.native_admission_stream_contracts.write() {
            contracts
                .entry(connection_id.to_string())
                .or_default()
                .outbound = Some(contract);
        }
        println!(
            "[PlutoRTC][session-admission][stream-contract] connection_id={} direction=outbound contract={:?}",
            connection_id, contract,
        );
    }

    #[cfg(not(target_arch = "wasm32"))]
    pub(crate) fn native_admission_stream_contracts(
        &self,
        connection_id: &str,
    ) -> crate::client::NativeAdmissionStreamContracts {
        self.native_admission_stream_contracts
            .read()
            .ok()
            .and_then(|contracts| contracts.get(connection_id).copied())
            .unwrap_or_default()
    }

    /// Return whether every required direction has a current-generation proof.
    /// Persistent native-main streams are bidirectional data routes, but the
    /// admission proofs remain directional security evidence. Stream ranking
    /// may retain one route only after both user-device presentations complete.
    #[cfg(not(target_arch = "wasm32"))]
    pub(crate) fn native_admission_route_is_ready_for_transport(
        &self,
        connection_id: &str,
        transport_stable_id: Option<u64>,
    ) -> bool {
        if let Some(transport_stable_id) = transport_stable_id {
            return self.native_application_stream_admitted_for_transport(
                connection_id,
                transport_stable_id,
            );
        }

        let contracts = self.native_admission_stream_contracts(connection_id);
        let requires_persistent_proof = matches!(
            contracts.inbound,
            Some(crate::native_protocol::SessionTokenStreamContract::PersistentControl)
        ) || matches!(
            contracts.outbound,
            Some(crate::native_protocol::SessionTokenStreamContract::PersistentControl)
        );
        if requires_persistent_proof {
            return self.native_declared_persistent_routes_are_proven(connection_id);
        }
        let inbound_ready = contracts.inbound.is_none()
            || self
                .inbound_session_admission_transport_ids
                .read()
                .ok()
                .is_some_and(|proofs| proofs.contains_key(connection_id));
        let outbound_ready = contracts.outbound.is_none()
            || self
                .remote_session_admission_proofs
                .read()
                .ok()
                .is_some_and(|proofs| proofs.contains_key(connection_id));
        inbound_ready && outbound_ready
    }

    #[cfg(not(target_arch = "wasm32"))]
    pub(crate) fn native_admission_stream_is_persistent(&self, connection_id: &str) -> bool {
        let contracts = self.native_admission_stream_contracts(connection_id);
        contracts.inbound
            == Some(crate::native_protocol::SessionTokenStreamContract::PersistentControl)
            || contracts.outbound
                == Some(crate::native_protocol::SessionTokenStreamContract::PersistentControl)
    }

    /// Invalidate the generation-bound native-main route proofs owned by one
    /// physical Iroh leg without forgetting the accepted logical session.
    ///
    /// Re-presenting the same token for the same deterministic connection is
    /// idempotent in `SessionTokenRegistry`. Keeping these proofs stale would
    /// let application streams race ahead of the peer's replacement control
    /// stream; clearing them forces the existing admission flow to establish a
    /// declared Rust-owned route before the session becomes routable again.
    #[cfg(not(target_arch = "wasm32"))]
    pub(crate) fn invalidate_native_main_route_proofs_for_transport(
        &self,
        connection_id: &str,
        transport_stable_id: u64,
    ) -> bool {
        let inbound_removed = self
            .inbound_session_admission_transport_ids
            .write()
            .ok()
            .is_some_and(|mut proofs| {
                if proofs.get(connection_id).copied() == Some(transport_stable_id) {
                    proofs.remove(connection_id);
                    true
                } else {
                    false
                }
            });
        let remote_removed = self
            .remote_session_admission_proofs
            .write()
            .ok()
            .is_some_and(|mut proofs| {
                if proofs
                    .get(connection_id)
                    .is_some_and(|proof| proof.transport_stable_id == transport_stable_id)
                {
                    proofs.remove(connection_id);
                    true
                } else {
                    false
                }
            });

        let removed = inbound_removed || remote_removed;
        if removed {
            println!(
                "[PlutoRTC][session-admission][route-invalidated] connection_id={} transport_stable_id={} inbound_removed={} remote_removed={}",
                connection_id, transport_stable_id, inbound_removed, remote_removed,
            );
        }
        removed
    }

    /// Invalidate only proofs whose wire contract requires the cached
    /// persistent native-main stream to remain alive. A one-shot admission is
    /// transport-generation evidence; it survives control-stream recreation
    /// on that same live Iroh leg and is retired only with the physical leg.
    #[cfg(not(target_arch = "wasm32"))]
    pub(crate) fn invalidate_native_persistent_route_proofs_for_transport(
        &self,
        connection_id: &str,
        transport_stable_id: u64,
    ) -> bool {
        let contracts = self.native_admission_stream_contracts(connection_id);
        let inbound_removed = if contracts.inbound
            == Some(crate::native_protocol::SessionTokenStreamContract::PersistentControl)
        {
            self.inbound_session_admission_transport_ids
                .write()
                .ok()
                .is_some_and(|mut proofs| {
                    if proofs.get(connection_id).copied() == Some(transport_stable_id) {
                        proofs.remove(connection_id);
                        true
                    } else {
                        false
                    }
                })
        } else {
            false
        };
        let outbound_removed = if contracts.outbound
            == Some(crate::native_protocol::SessionTokenStreamContract::PersistentControl)
        {
            self.remote_session_admission_proofs
                .write()
                .ok()
                .is_some_and(|mut proofs| {
                    if proofs
                        .get(connection_id)
                        .is_some_and(|proof| proof.transport_stable_id == transport_stable_id)
                    {
                        proofs.remove(connection_id);
                        true
                    } else {
                        false
                    }
                })
        } else {
            false
        };

        let removed = inbound_removed || outbound_removed;
        if removed {
            println!(
                "[PlutoRTC][session-admission][persistent-route-invalidated] connection_id={} transport_stable_id={} inbound_removed={} outbound_removed={}",
                connection_id, transport_stable_id, inbound_removed, outbound_removed,
            );
        }
        removed
    }

    #[cfg(not(target_arch = "wasm32"))]
    pub(crate) fn invalidate_native_main_route_proofs_except_transport(
        &self,
        connection_id: &str,
        current_transport_stable_id: u64,
    ) -> bool {
        let inbound_stale = self
            .inbound_session_admission_transport_ids
            .read()
            .ok()
            .and_then(|proofs| proofs.get(connection_id).copied())
            .filter(|stable_id| *stable_id != current_transport_stable_id);
        let remote_stale = self
            .remote_session_admission_proofs
            .read()
            .ok()
            .and_then(|proofs| {
                proofs
                    .get(connection_id)
                    .map(|proof| proof.transport_stable_id)
            })
            .filter(|stable_id| *stable_id != current_transport_stable_id);

        let mut removed = false;
        if let Some(stable_id) = inbound_stale {
            removed |=
                self.invalidate_native_main_route_proofs_for_transport(connection_id, stable_id);
        }
        if let Some(stable_id) = remote_stale {
            removed |=
                self.invalidate_native_main_route_proofs_for_transport(connection_id, stable_id);
        }
        removed
    }

    #[cfg(not(target_arch = "wasm32"))]
    pub(crate) fn native_declared_persistent_routes_are_proven(&self, connection_id: &str) -> bool {
        let contracts = self.native_admission_stream_contracts(connection_id);
        let bilateral_user_device = matches!(
            self.session_admission(connection_id),
            crate::session_token::SessionAdmission::Accepted {
                scope: Some(scope),
                ..
            } if scope.as_str() == "user-device"
        );
        let inbound_required = bilateral_user_device
            || contracts.inbound
                == Some(crate::native_protocol::SessionTokenStreamContract::PersistentControl);
        let outbound_required = bilateral_user_device
            || contracts.outbound
                == Some(crate::native_protocol::SessionTokenStreamContract::PersistentControl);
        if !inbound_required && !outbound_required {
            return false;
        }

        let inbound_transport_stable_id = self
            .inbound_session_admission_transport_ids
            .read()
            .ok()
            .and_then(|proofs| proofs.get(connection_id).copied());
        let outbound_transport_stable_id = self
            .remote_session_admission_proofs
            .read()
            .ok()
            .and_then(|proofs| {
                proofs
                    .get(connection_id)
                    .map(|proof| proof.transport_stable_id)
            });

        if inbound_required && inbound_transport_stable_id.is_none() {
            return false;
        }
        if outbound_required && outbound_transport_stable_id.is_none() {
            return false;
        }

        if inbound_required && outbound_required {
            return inbound_transport_stable_id == outbound_transport_stable_id;
        }

        true
    }

    /// 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
    }

    /// Reserve the connection while a host adapter writes an SDK-owned
    /// session-token verdict. The guard must remain alive until the response
    /// has been flushed, then be dropped before post-admission lifecycle work.
    pub fn try_begin_session_token_response(
        &self,
        connection_id: &str,
    ) -> Option<crate::session_token::AdmissionResponseGuard> {
        self.session_token_registry
            .try_begin_admission_response(connection_id)
    }

    /// 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,
    ) {
        self.commit_session_token_application_security_epoch(connection_id)
            .await;
        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;
        }
    }

    async fn normalize_session_token_scope_trust_class(
        &self,
        connection_id: &str,
        scope: Option<&str>,
    ) -> bool {
        let Some(scope) = scope.map(str::trim).filter(|scope| !scope.is_empty()) else {
            return false;
        };
        let next_is_transient = session_scope_uses_transient_peer_identity(scope);
        let replace_mixed_trust_scopes = self
            .connection_manager
            .get_scopes(connection_id)
            .await
            .iter()
            .any(|existing| {
                session_scope_uses_transient_peer_identity(existing) != next_is_transient
            });
        if replace_mixed_trust_scopes {
            self.connection_manager
                .release_scope(connection_id, None)
                .await;
            self.connection_manager
                .add_scope(connection_id, scope)
                .await;
        }
        replace_mixed_trust_scopes
    }

    /// Start an outbound capability epoch before presenting the token. Keeping
    /// crypto required closes the interval where an old settled projection
    /// could otherwise admit product traffic while replacement key agreement
    /// is still in flight.
    pub(crate) fn begin_outbound_session_token_application_security_epoch(
        &self,
        connection_id: &str,
        token: &str,
    ) -> bool {
        let current_fingerprint = crate::session_token::token_fingerprint(token);
        let committed_fingerprint = self
            .outbound_application_security_epoch_fingerprints
            .read()
            .ok()
            .and_then(|epochs| epochs.get(connection_id).cloned());
        let security_epoch_changed = match committed_fingerprint.as_deref() {
            Some(committed) => committed != current_fingerprint,
            // A key agreement may settle just before the first token verdict.
            // With no previously committed capability there is no older epoch
            // to retire; logout/revoke boundaries explicitly clear both maps.
            None => false,
        };
        if security_epoch_changed {
            self.clear_connection_application_crypto_key(connection_id);
            eprintln!(
                "[OpenRTC][admission-security] began outbound security epoch connection_id={} previous_token_fp={} current_token_fp={}",
                connection_id,
                committed_fingerprint.as_deref().unwrap_or("<uncommitted>"),
                current_fingerprint,
            );
        }
        self.set_connection_application_crypto_required(connection_id);
        security_epoch_changed
    }

    pub(crate) fn commit_outbound_session_token_application_security_epoch(
        &self,
        connection_id: &str,
        token: &str,
    ) {
        if let Ok(mut epochs) = self
            .outbound_application_security_epoch_fingerprints
            .write()
        {
            epochs.insert(
                connection_id.to_string(),
                crate::session_token::token_fingerprint(token),
            );
        }
    }

    /// Commit the capability token as an application-security epoch after its
    /// approval has reached the presenter. A different token on the same
    /// deterministic node-pair connection must not inherit the previous
    /// application key or a scope from the opposite trust class. Re-presenting
    /// the same token is intentionally idempotent so transport replacement does
    /// not reset crypto sequence state.
    async fn commit_session_token_application_security_epoch(&self, connection_id: &str) {
        let Some(current_fingerprint) = self
            .session_token_registry
            .admission_fingerprint(connection_id)
        else {
            return;
        };
        let committed_fingerprint = self
            .session_token_registry
            .application_security_epoch_fingerprint(connection_id);
        let security_epoch_changed = match committed_fingerprint.as_deref() {
            Some(committed) => committed != current_fingerprint,
            // The first approved capability adopts any key established during
            // its transport handshake. Only a committed older capability can
            // make the current key stale.
            None => false,
        };

        let admitted_scope = match self.session_admission(connection_id) {
            crate::session_token::SessionAdmission::Accepted {
                scope: Some(scope), ..
            } => Some(scope.into_inner()),
            _ => None,
        };
        if security_epoch_changed {
            self.clear_connection_application_crypto_key(connection_id);
        }
        self.set_connection_application_crypto_required(connection_id);
        let replace_mixed_trust_scopes = self
            .normalize_session_token_scope_trust_class(connection_id, admitted_scope.as_deref())
            .await;
        if security_epoch_changed {
            eprintln!(
                "[OpenRTC][admission-security] committed new security epoch connection_id={} previous_token_fp={} current_token_fp={} scopes_replaced={}",
                connection_id,
                committed_fingerprint.as_deref().unwrap_or("<uncommitted>"),
                current_fingerprint,
                replace_mixed_trust_scopes,
            );
        }
        self.session_token_registry
            .commit_application_security_epoch(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> {
        self.inspect_incoming_native_main_frame_for_transport(
            connection_id,
            remote_node_id,
            known_device_id,
            None,
            frame,
        )
        .await
    }

    #[cfg(native)]
    pub(crate) async fn inspect_incoming_native_main_frame_for_transport(
        &self,
        connection_id: &str,
        remote_node_id: Option<&str>,
        known_device_id: Option<&str>,
        transport_stable_id: Option<u64>,
        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()
                    );
                    let was_already_admitted = self
                        .session_token_registry
                        .is_session_token_admitted_for_connection(connection_id);
                    let verdict = self
                        .validate_session_token_for_connection_with_payload(
                            &token,
                            connection_id,
                            token_payload.as_deref(),
                        )
                        .await;
                    let (response, accepted) = match verdict {
                        Ok(scope) => {
                            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"
                            );
                            (
                                crate::native_protocol::NativeMainMessage::session_token_approval(
                                    (!scope.is_empty()).then_some(scope.as_str()),
                                    connection_id,
                                ),
                                true,
                            )
                        }
                        Err(reason) => {
                            crate::clog!(
                                "[PlutoRTC][session-token]",
                                &presentation_ctx,
                                "validated_rejected reason={}",
                                reason
                            );
                            (
                                crate::native_protocol::NativeMainMessage::session_token_rejection(
                                    &reason,
                                    connection_id,
                                ),
                                false,
                            )
                        }
                    };
                    return Ok(InspectedMainFrame::SessionTokenResponse {
                        response,
                        accepted,
                        is_first_presentation: accepted && !was_already_admitted,
                    });
                }

                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
                };

                if message.is_peer_data() {
                    let generation = self
                        .current_native_peer_data_generation(connection_id, transport_stable_id)
                        .await
                        .ok_or_else(|| "native-peer-data-stale-generation".to_string())?;
                    let protected_payload = message
                        .peer_data_payload()
                        .ok_or_else(|| "native-peer-data-payload-missing".to_string())?;
                    let payload = self
                        .open_inbound_application_payload(connection_id, &protected_payload)
                        .map_err(|error| error.to_string())?;
                    self.emit_native_peer_data(crate::client::NativePeerDataEvent {
                        connection_id: connection_id.to_string(),
                        remote_node_id: remote_node_id
                            .map(str::trim)
                            .filter(|value| !value.is_empty())
                            .map(ToOwned::to_owned),
                        transport: crate::transport_label::IROH.to_string(),
                        transport_stable_id: generation.transport_stable_id,
                        transport_generation: generation.transport_generation,
                        route_generation: generation.route_generation,
                        payload,
                    });
                    return Ok(InspectedMainFrame::ForwardOpaque);
                }

                // 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);
        // A deterministic connection id can be reused after logout, explicit
        // disconnect, revocation, or a fresh admission window. Its prior
        // application key and ephemeral agreement belong to the retired
        // logical session, not to the node pair forever. Keeping either lets a
        // replacement scope skip key agreement and encrypt with stale material.
        // Transient transport replacement does not call this boundary and
        // therefore keeps its established application session intact.
        self.clear_connection_application_crypto_key(connection_id);
        #[cfg(not(target_arch = "wasm32"))]
        if let Ok(mut proofs) = self.inbound_session_admission_transport_ids.write() {
            proofs.remove(connection_id);
        }
        if let Ok(mut proofs) = self.remote_session_admission_proofs.write() {
            proofs.remove(connection_id);
        }
        if let Ok(mut pending) = self.pending_inline_reciprocal_admissions.write() {
            pending.retain(|_, transcript| transcript.connection_id != connection_id);
        }
        if let Ok(mut epochs) = self
            .outbound_application_security_epoch_fingerprints
            .write()
        {
            epochs.remove(connection_id);
        }
        #[cfg(not(target_arch = "wasm32"))]
        if let Ok(mut contracts) = self.native_admission_stream_contracts.write() {
            contracts.remove(connection_id);
        }
        #[cfg(not(target_arch = "wasm32"))]
        if let Ok(mut requests) = self.pending_reciprocal_session_admission_requests.write() {
            requests.remove(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(not(target_arch = "wasm32"))]
        if let Ok(mut proofs) = self.inbound_session_admission_transport_ids.write() {
            proofs.clear();
        }
        if let Ok(mut proofs) = self.remote_session_admission_proofs.write() {
            proofs.clear();
        }
        if let Ok(mut pending) = self.pending_inline_reciprocal_admissions.write() {
            pending.clear();
        }
        if let Ok(mut epochs) = self
            .outbound_application_security_epoch_fingerprints
            .write()
        {
            epochs.clear();
        }
        #[cfg(not(target_arch = "wasm32"))]
        if let Ok(mut contracts) = self.native_admission_stream_contracts.write() {
            contracts.clear();
        }
        #[cfg(not(target_arch = "wasm32"))]
        if let Ok(mut requests) = self.pending_reciprocal_session_admission_requests.write() {
            requests.clear();
        }
        #[cfg(not(target_arch = "wasm32"))]
        if let Ok(mut credentials) = self.native_route_repair_credentials.write() {
            credentials.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) -> Vec<String> {
        let token_fp = crate::session_token::token_fingerprint(token);
        eprintln!("[PlutoRTC][teardown-trace] revoke_session_token token_fp={token_fp}");
        let affected_connections = self.session_token_registry.revoke(token);
        #[cfg(not(target_arch = "wasm32"))]
        if let Ok(mut proofs) = self.inbound_session_admission_transport_ids.write() {
            for connection_id in &affected_connections {
                proofs.remove(connection_id);
            }
        }
        if let Ok(mut proofs) = self.remote_session_admission_proofs.write() {
            for connection_id in &affected_connections {
                proofs.remove(connection_id);
            }
        }
        if let Ok(mut pending) = self.pending_inline_reciprocal_admissions.write() {
            pending
                .retain(|_, transcript| !affected_connections.contains(&transcript.connection_id));
        }
        if let Ok(mut epochs) = self
            .outbound_application_security_epoch_fingerprints
            .write()
        {
            for connection_id in &affected_connections {
                epochs.remove(connection_id);
            }
        }
        #[cfg(not(target_arch = "wasm32"))]
        if let Ok(mut contracts) = self.native_admission_stream_contracts.write() {
            for connection_id in &affected_connections {
                contracts.remove(connection_id);
            }
        }
        #[cfg(not(target_arch = "wasm32"))]
        if let Ok(mut requests) = self.pending_reciprocal_session_admission_requests.write() {
            for connection_id in &affected_connections {
                requests.remove(connection_id);
            }
        }
        #[cfg(not(target_arch = "wasm32"))]
        if let Ok(mut credentials) = self.native_route_repair_credentials.write() {
            credentials.retain(|_, credential| credential.token != token);
        }
        if !affected_connections.is_empty() {
            let client = self.clone();
            let retirement_connections = affected_connections.clone();
            n0_future::task::spawn(async move {
                client
                    .retire_revoked_session_connections(&retirement_connections)
                    .await;
            });
        }
        #[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
                            ),
                        }
                    }
                }
            }
        }
        affected_connections
    }

    /// 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));
        if let Ok(mut pending) = self.pending_inline_reciprocal_admissions.write() {
            pending
                .retain(|_, transcript| !affected_connections.contains(&transcript.connection_id));
        }
        if let Ok(mut epochs) = self
            .outbound_application_security_epoch_fingerprints
            .write()
        {
            for connection_id in &affected_connections {
                epochs.remove(connection_id);
            }
        }
        #[cfg(not(target_arch = "wasm32"))]
        if let Ok(mut credentials) = self.native_route_repair_credentials.write() {
            // Repair credentials are derived from remote compound tickets. A
            // scope revocation cannot identify their scope without trusting the
            // cached payload again, so discard all and repopulate from fresh
            // signaling observations.
            credentials.clear();
        }
        #[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 endpoint_disconnects = self
            .retire_revoked_session_connections(&affected_connections)
            .await;

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

        affected_connections
    }

    async fn retire_revoked_session_connections(&self, connection_ids: &[String]) -> usize {
        let mut endpoint_ids = HashSet::new();
        for connection_id in connection_ids {
            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;
                    }
                }
            }
        }
        endpoint_disconnects
    }

    #[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")?;
        Self::write_private_managed_scope_grant(&path, &serialized).await?;
        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)
        };
        if let Some(device_id) = authoritative_device_id.as_deref() {
            self.connection_manager
                .set_device_id(connection_id, device_id.to_string())
                .await;
        }

        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;
        }

        let admission = self.session_admission(connection_id);
        if self.session_registry_active() {
            match &admission {
                crate::session_token::SessionAdmission::Rejected { .. } => return false,
                crate::session_token::SessionAdmission::Accepted {
                    scope: Some(scope), ..
                } if !session_scope_allows_device_binding(scope.as_str()) => {
                    println!(
                        "[PlutoRTC][session-admission][device-bind-refused] connection_id={} scope={}",
                        connection_id,
                        scope.as_str(),
                    );
                    return false;
                }
                crate::session_token::SessionAdmission::Accepted { scope: None, .. }
                | crate::session_token::SessionAdmission::Pending => return false,
                crate::session_token::SessionAdmission::Accepted { scope: Some(_), .. } => {}
            }

            if !self
                .bind_session_admission_authoritative_device_id(
                    connection_id,
                    authoritative_device_id,
                )
                .await
            {
                println!(
                    "[PlutoRTC][session-admission][device-bind-refused] connection_id={} reason=authoritative-device-mismatch",
                    connection_id,
                );
                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 admission {
                crate::session_token::SessionAdmission::Accepted { .. } => {}
                crate::session_token::SessionAdmission::Rejected { .. }
                | crate::session_token::SessionAdmission::Pending => {
                    unreachable!("non-accepted admission returned before trusted-device mutation")
                }
            }
        }

        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,
                reciprocal_requested: false,
            },
        )
        .await
    }

    pub(crate) async fn present_and_accept_session_token(
        &self,
        endpoint_id: iroh::EndpointId,
        connection_id: &str,
        token: &str,
        token_payload: Option<&str>,
        remote_authoritative_device_id: Option<String>,
    ) -> Result<String, String> {
        self.present_and_accept_session_token_with_options(
            endpoint_id,
            connection_id,
            token,
            token_payload,
            remote_authoritative_device_id,
            None,
            false,
            false,
        )
        .await
    }

    #[cfg(not(target_arch = "wasm32"))]
    pub(crate) async fn present_and_accept_session_token_for_route_repair(
        &self,
        endpoint_id: iroh::EndpointId,
        connection_id: &str,
        token: &str,
        token_payload: Option<&str>,
        remote_authoritative_device_id: Option<String>,
        claimed_local_device_id: Option<String>,
        force_presentation: bool,
        request_reciprocal: bool,
    ) -> Result<String, String> {
        self.present_and_accept_session_token_with_options(
            endpoint_id,
            connection_id,
            token,
            token_payload,
            remote_authoritative_device_id,
            claimed_local_device_id,
            force_presentation,
            request_reciprocal,
        )
        .await
    }

    pub(crate) async fn present_and_accept_session_token_with_local_claim(
        &self,
        endpoint_id: iroh::EndpointId,
        connection_id: &str,
        token: &str,
        token_payload: Option<&str>,
        remote_authoritative_device_id: Option<String>,
        claimed_local_device_id: Option<String>,
    ) -> Result<String, String> {
        self.present_and_accept_session_token_with_options(
            endpoint_id,
            connection_id,
            token,
            token_payload,
            remote_authoritative_device_id,
            claimed_local_device_id,
            false,
            false,
        )
        .await
    }

    async fn present_and_accept_session_token_with_options(
        &self,
        endpoint_id: iroh::EndpointId,
        connection_id: &str,
        token: &str,
        token_payload: Option<&str>,
        remote_authoritative_device_id: Option<String>,
        claimed_local_device_id: Option<String>,
        force_presentation: bool,
        request_reciprocal: bool,
    ) -> Result<String, String> {
        // Admission presentation is a per-physical-generation single-flight.
        // Explicit connect and the managed-session actor can legitimately meet
        // here with the same token. Join that canonical exchange instead of
        // surfacing an internal ownership race to the caller. If the owner
        // exits without producing a matching proof, acquire the guard and make
        // this caller the replacement presenter.
        #[cfg(target_arch = "wasm32")]
        let presentation_join_deadline_ms =
            js_sys::Date::now() + SESSION_TOKEN_HOST_RESPONSE_TIMEOUT_MS as f64 + 1_000.0;
        #[cfg(not(target_arch = "wasm32"))]
        let presentation_join_deadline = std::time::Instant::now()
            + std::time::Duration::from_millis(SESSION_TOKEN_HOST_RESPONSE_TIMEOUT_MS + 1_000);
        let _presentation_guard = loop {
            if let Some(guard) = self
                .session_token_registry
                .try_begin_admission_presentation(connection_id)
            {
                break guard;
            }
            if let Some(approval_scope) = self
                .current_remote_session_admission_scope(connection_id, endpoint_id, token)
                .await
            {
                println!(
                    "[PlutoRTC][session-admission][single-flight] joined canonical presentation connection_id={} endpoint_id={} token_fp={} scope={}",
                    connection_id,
                    endpoint_id,
                    super::core_impl::log_fingerprint(token),
                    approval_scope,
                );
                return Ok(approval_scope);
            }
            #[cfg(target_arch = "wasm32")]
            let expired = js_sys::Date::now() >= presentation_join_deadline_ms;
            #[cfg(not(target_arch = "wasm32"))]
            let expired = std::time::Instant::now() >= presentation_join_deadline;
            if expired {
                return Err(format!(
                    "[session-token-presentation:join-timeout] canonical presenter did not finish for {}",
                    connection_id
                ));
            }
            #[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;
        };
        // Close the check-then-present race with negotiated transport handoffs
        // and duplicate app-level connect requests. The proof is fenced to the
        // exact token and physical transport generation, so replaying the same
        // request cannot add authority. The host's original device binding also
        // remains authoritative: skipping a duplicate never rebinds it to a new
        // local claim. Route repair can still force a fresh presentation after
        // it has invalidated the generation-bound proof.
        if !force_presentation {
            if let Some(approval_scope) = self
                .current_remote_session_admission_scope(connection_id, endpoint_id, token)
                .await
            {
                println!(
                    "[PlutoRTC][session-admission][remote-proof] presentation skipped connection_id={} endpoint_id={} token_fp={} scope={} reason=current-transport-already-approved",
                    connection_id,
                    endpoint_id,
                    super::core_impl::log_fingerprint(token),
                    approval_scope,
                );
                return Ok(approval_scope);
            }
        }
        let local_device_id = claimed_local_device_id
            .filter(|device_id| !device_id.trim().is_empty())
            .or_else(|| {
                self.active_session_identity()
                    .map(|(_, device_id)| device_id)
            })
            .filter(|device_id| !device_id.trim().is_empty());
        let approval_scope = self
            .present_session_token(
                SessionTokenPresentationTarget::Host {
                    endpoint_id,
                    connection_id,
                },
                token,
                SessionTokenPresentationOptions {
                    token_payload,
                    device_id: local_device_id.as_deref(),
                    reciprocal_requested: request_reciprocal,
                },
            )
            .await?;
        self.mark_connection_admitted_by_host(
            connection_id,
            if approval_scope.trim().is_empty() {
                None
            } else {
                Some(approval_scope.as_str())
            },
            remote_authoritative_device_id,
        )
        .await;
        self.normalize_session_token_scope_trust_class(
            connection_id,
            if approval_scope.trim().is_empty() {
                None
            } else {
                Some(approval_scope.as_str())
            },
        )
        .await;
        #[cfg(target_arch = "wasm32")]
        if let Some(transport_stable_id) = self
            .get_connection(endpoint_id)
            .await
            .map(|connection| connection.stable_id() as u64)
        {
            if let Ok(mut proofs) = self.remote_session_admission_proofs.write() {
                proofs.insert(
                    connection_id.to_string(),
                    RemoteSessionAdmissionProof {
                        transport_stable_id,
                        token_fingerprint: crate::session_token::token_fingerprint(token),
                        approval_scope: approval_scope.clone(),
                    },
                );
            }
            println!(
                "[PlutoRTC][session-admission][remote-proof] connection_id={} endpoint_id={} transport_stable_id={} token_fp={}",
                connection_id,
                endpoint_id,
                transport_stable_id,
                super::core_impl::log_fingerprint(token),
            );
        }
        Ok(approval_scope)
    }

    pub(crate) async fn has_current_remote_session_admission_proof(
        &self,
        connection_id: &str,
        endpoint_id: iroh::EndpointId,
        token: &str,
    ) -> bool {
        self.current_remote_session_admission_scope(connection_id, endpoint_id, token)
            .await
            .is_some()
    }

    /// Return whether this side has completed the outbound admission required
    /// by a compound endpoint ticket. Plain endpoint tickets require no
    /// session-token proof and are therefore already ready.
    pub async fn remote_session_admission_ready_for_ticket(
        &self,
        endpoint_ticket: &str,
    ) -> anyhow::Result<bool> {
        let (iroh_ticket, token_suffix) =
            crate::session_token::split_compound_ticket(endpoint_ticket.trim());
        let Some(token_suffix) = token_suffix else {
            return Ok(true);
        };
        let payload =
            crate::session_token::decode_token_payload_for_ticket(iroh_ticket, token_suffix)
                .ok_or_else(|| {
                    anyhow::anyhow!("invalid compound ticket payload for endpoint ticket")
                })?;
        let endpoint_addr = parse_endpoint_ticket(iroh_ticket)?;
        let Some(local_node_id) = self.current_node_id().await else {
            return Ok(false);
        };
        let connection_id =
            Self::deterministic_connection_id(&local_node_id, &endpoint_addr.id.to_string());
        Ok(self
            .has_current_remote_session_admission_proof(
                &connection_id,
                endpoint_addr.id,
                payload.token.as_str(),
            )
            .await)
    }

    async fn current_remote_session_admission_scope(
        &self,
        connection_id: &str,
        endpoint_id: iroh::EndpointId,
        token: &str,
    ) -> Option<String> {
        let Some(transport_stable_id) = self
            .get_connection(endpoint_id)
            .await
            .map(|connection| connection.stable_id() as u64)
        else {
            return None;
        };
        let token_fingerprint = crate::session_token::token_fingerprint(token);
        let proof = self
            .remote_session_admission_proofs
            .read()
            .ok()
            .and_then(|proofs| proofs.get(connection_id).cloned());
        let matches = proof.as_ref().is_some_and(|proof| {
            proof.transport_stable_id == transport_stable_id
                && proof.token_fingerprint == token_fingerprint
        });
        if Self::admission_trace_enabled() {
            match proof.as_ref() {
                Some(proof) => println!(
                    "[PlutoRTC][session-admission][remote-proof-check] connection_id={} endpoint_id={} current_stable_id={} proof_stable_id={} stable_id_matches={} token_matches={} result={}",
                    connection_id,
                    endpoint_id,
                    transport_stable_id,
                    proof.transport_stable_id,
                    proof.transport_stable_id == transport_stable_id,
                    proof.token_fingerprint == token_fingerprint,
                    matches,
                ),
                None => println!(
                    "[PlutoRTC][session-admission][remote-proof-check] connection_id={} endpoint_id={} current_stable_id={} result=false reason=missing-proof",
                    connection_id, endpoint_id, transport_stable_id,
                ),
            }
        }
        matches.then(|| proof.expect("matched proof exists").approval_scope)
    }

    /// Prepare one reciprocal presentation in the Rust admission owner.
    /// Browser adapters may provide the credential they learned from the
    /// authoritative directory, but the resulting transcript is fenced to the
    /// current inbound admission epoch, physical generation, and exact stream
    /// instance before any bytes are sent.
    pub async fn prepare_inline_reciprocal_session_admission(
        &self,
        endpoint_id: iroh::EndpointId,
        expected_transport_stable_id: u64,
        stream_instance_id: &str,
        presentation_id: &str,
        token: &str,
        token_payload: &str,
        device_id: &str,
        stream_contract: crate::native_protocol::SessionTokenStreamContract,
    ) -> Result<crate::native_protocol::ReciprocalSessionTokenPresentation, String> {
        let stream_instance_id = stream_instance_id.trim();
        let presentation_id = presentation_id.trim();
        let token = token.trim();
        let token_payload = token_payload.trim();
        let device_id = device_id.trim();
        if stream_instance_id.is_empty()
            || presentation_id.is_empty()
            || token.is_empty()
            || token_payload.is_empty()
            || device_id.is_empty()
        {
            return Err("reciprocal admission transcript contains an empty field".to_string());
        }
        let current_transport_stable_id = self
            .get_connection(endpoint_id)
            .await
            .map(|connection| connection.stable_id() as u64);
        if current_transport_stable_id != Some(expected_transport_stable_id) {
            return Err(format!(
                "reciprocal admission preparation belongs to a retired transport generation: expected={} current={:?}",
                expected_transport_stable_id, current_transport_stable_id,
            ));
        }
        let local_node_id = self
            .current_node_id()
            .await
            .ok_or_else(|| "missing local node id for reciprocal admission".to_string())?;
        let remote_node_id = endpoint_id.to_string();
        let connection_id = Self::deterministic_connection_id(&local_node_id, &remote_node_id);
        let inbound_admission_fingerprint = self
            .session_token_registry
            .admission_fingerprint(&connection_id)
            .ok_or_else(|| {
                "reciprocal admission preparation requires an accepted inbound token epoch"
                    .to_string()
            })?;
        let inbound_admission_epoch = self
            .session_token_registry
            .admission_epoch(&connection_id)
            .ok_or_else(|| {
                "reciprocal admission preparation requires a current admission epoch".to_string()
            })?;
        let remote_device_id = self
            .known_remote_device_id_for_incoming_transport(&connection_id, &remote_node_id)
            .await
            .filter(|remote_device_id| !remote_device_id.trim().is_empty())
            .ok_or_else(|| {
                "reciprocal admission preparation requires an authoritative remote device"
                    .to_string()
            })?;
        match self.session_admission(&connection_id) {
            crate::session_token::SessionAdmission::Accepted {
                authoritative_device_id: Some(admitted_device_id),
                ..
            } if admitted_device_id == remote_device_id => {}
            crate::session_token::SessionAdmission::Accepted { .. } => {
                return Err(
                    "reciprocal admission preparation has no matching admitted remote identity"
                        .to_string(),
                );
            }
            _ => {
                return Err(
                    "reciprocal admission preparation requires an accepted inbound admission"
                        .to_string(),
                );
            }
        }
        let active_local_device_id = self
            .active_session_identity()
            .map(|(_, local_device_id)| local_device_id)
            .filter(|local_device_id| !local_device_id.trim().is_empty())
            .ok_or_else(|| "local managed-session identity is unavailable".to_string())?;
        if active_local_device_id != device_id {
            return Err(format!(
                "reciprocal admission local device mismatch: active={} claimed={}",
                active_local_device_id, device_id,
            ));
        }
        let payload = crate::session_token::decode_token_payload(token_payload)
            .filter(crate::session_token::token_payload_metadata_is_valid)
            .ok_or_else(|| {
                "reciprocal admission token payload is invalid or expired".to_string()
            })?;
        if payload.token != token {
            return Err("reciprocal admission token payload does not match the token".to_string());
        }
        let expected_scope = payload.scope.as_str().trim().to_string();
        if expected_scope.is_empty() {
            return Err("reciprocal admission token scope is empty".to_string());
        }

        let transcript = crate::client::PendingInlineReciprocalAdmission {
            connection_id: connection_id.clone(),
            remote_node_id: remote_node_id.clone(),
            local_device_id: active_local_device_id,
            remote_device_id,
            transport_stable_id: expected_transport_stable_id,
            stream_instance_id: stream_instance_id.to_string(),
            presentation_id: presentation_id.to_string(),
            token: token.to_string(),
            token_fingerprint: crate::session_token::token_fingerprint(token),
            expected_scope,
            inbound_admission_fingerprint,
            inbound_admission_epoch,
        };
        let mut pending = self
            .pending_inline_reciprocal_admissions
            .write()
            .map_err(|_| "reciprocal admission transcript registry is unavailable".to_string())?;
        pending.retain(|_, existing| {
            existing.remote_node_id != remote_node_id
                || existing.transport_stable_id != expected_transport_stable_id
                || existing.stream_instance_id != stream_instance_id
        });
        pending.insert(presentation_id.to_string(), transcript);

        Ok(crate::native_protocol::ReciprocalSessionTokenPresentation {
            presentation_id: presentation_id.to_string(),
            token: token.to_string(),
            token_payload: token_payload.to_string(),
            device_id: device_id.to_string(),
            stream_contract,
        })
    }

    /// Commit the peer's ACK for an inline reciprocal presentation to the
    /// Rust-owned admission state. The adapter reports only the correlated ACK;
    /// token, scope expectation, identity epoch, and route generation all come
    /// from the Rust-owned preparation transcript.
    pub async fn confirm_inline_reciprocal_session_admission(
        &self,
        endpoint_id: iroh::EndpointId,
        expected_transport_stable_id: u64,
        stream_instance_id: &str,
        presentation_id: &str,
        accepted: bool,
        approval_scope: Option<&str>,
    ) -> Result<(), String> {
        let stream_instance_id = stream_instance_id.trim();
        let presentation_id = presentation_id.trim();
        let transcript = self
            .pending_inline_reciprocal_admissions
            .read()
            .map_err(|_| "reciprocal admission transcript registry is unavailable".to_string())?
            .get(presentation_id)
            .cloned()
            .ok_or_else(|| "reciprocal admission ACK has no pending transcript".to_string())?;
        if transcript.presentation_id != presentation_id
            || transcript.remote_node_id != endpoint_id.to_string()
            || transcript.transport_stable_id != expected_transport_stable_id
            || transcript.stream_instance_id != stream_instance_id
        {
            return Err(
                "reciprocal admission ACK does not match its prepared transcript".to_string(),
            );
        }
        if !accepted {
            if let Ok(mut pending) = self.pending_inline_reciprocal_admissions.write() {
                pending.remove(presentation_id);
            }
            return Ok(());
        }
        let current_transport_stable_id = self
            .get_connection(endpoint_id)
            .await
            .map(|connection| connection.stable_id() as u64);
        if current_transport_stable_id != Some(expected_transport_stable_id) {
            return Err(format!(
                "reciprocal admission ACK belongs to a retired transport generation: expected={} current={:?}",
                expected_transport_stable_id, current_transport_stable_id,
            ));
        }
        let connection_id = transcript.connection_id.clone();
        let active_local_device_id = self
            .active_session_identity()
            .map(|(_, local_device_id)| local_device_id)
            .filter(|local_device_id| !local_device_id.trim().is_empty());
        if active_local_device_id.as_deref() != Some(transcript.local_device_id.as_str()) {
            return Err("reciprocal admission ACK belongs to a retired local identity".to_string());
        }
        let current_remote_device_id = self
            .known_remote_device_id_for_incoming_transport(
                &connection_id,
                &transcript.remote_node_id,
            )
            .await;
        if current_remote_device_id.as_deref() != Some(transcript.remote_device_id.as_str()) {
            return Err(
                "reciprocal admission ACK belongs to a retired remote identity".to_string(),
            );
        }
        match self.session_admission(&connection_id) {
            crate::session_token::SessionAdmission::Accepted {
                authoritative_device_id: Some(admitted_device_id),
                ..
            } if admitted_device_id == transcript.remote_device_id => {}
            _ => {
                return Err(
                    "reciprocal admission ACK arrived outside its admitted remote identity"
                        .to_string(),
                );
            }
        }
        if !self
            .connection_manager
            .current_transport_matches(&connection_id, Some(expected_transport_stable_id))
            .await
        {
            return Err(
                "reciprocal admission ACK does not match the managed transport generation"
                    .to_string(),
            );
        }
        if self
            .session_token_registry
            .admission_fingerprint(&connection_id)
            != Some(transcript.inbound_admission_fingerprint.clone())
        {
            return Err(
                "reciprocal admission ACK belongs to a retired admission epoch".to_string(),
            );
        }
        if self.session_token_registry.admission_epoch(&connection_id)
            != Some(transcript.inbound_admission_epoch)
        {
            return Err(
                "reciprocal admission ACK belongs to a retired admission generation".to_string(),
            );
        }

        let approval_scope = approval_scope.unwrap_or_default().trim().to_string();
        if approval_scope != transcript.expected_scope {
            return Err(format!(
                "reciprocal admission ACK scope mismatch: expected={} received={}",
                transcript.expected_scope, approval_scope,
            ));
        }
        if let Ok(mut proofs) = self.remote_session_admission_proofs.write() {
            proofs.insert(
                connection_id.clone(),
                RemoteSessionAdmissionProof {
                    transport_stable_id: expected_transport_stable_id,
                    token_fingerprint: transcript.token_fingerprint.clone(),
                    approval_scope: approval_scope.clone(),
                },
            );
        } else {
            return Err("remote admission proof registry is unavailable".to_string());
        }
        self.commit_outbound_session_token_application_security_epoch(
            &connection_id,
            transcript.token.as_str(),
        );
        if let Ok(mut pending) = self.pending_inline_reciprocal_admissions.write() {
            pending.remove(presentation_id);
        }
        let _ = self
            .confirm_managed_connection_readiness_from_transport_proof(
                &connection_id,
                expected_transport_stable_id,
            )
            .await;
        println!(
            "[PlutoRTC][session-admission][inline-reciprocal-commit] connection_id={} endpoint_id={} transport_stable_id={} token_fp={} scope={}",
            connection_id,
            endpoint_id,
            expected_transport_stable_id,
            super::core_impl::log_fingerprint(transcript.token.as_str()),
            approval_scope,
        );
        Ok(())
    }

    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,
                reciprocal_requested: false,
            },
        )
        .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,
                )
            }
        };
        self.begin_outbound_session_token_application_security_epoch(connection_id.as_ref(), token);

        async fn read_peer_exact(
            recv: &mut crate::application_crypto_streams::PeerRecvStream,
            buffer: &mut [u8],
            context: &str,
        ) -> Result<(), String> {
            let mut offset = 0;
            while offset < buffer.len() {
                let read = recv
                    .read(&mut buffer[offset..])
                    .await
                    .map_err(|error| format!("[{context}] {error}"))?;
                if read == 0 {
                    return Err(format!("[{context}] stream finished early (0 bytes read)"));
                }
                offset += read;
            }
            Ok(())
        }

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

            let mut label_len_buf = [0u8; 4];
            read_peer_exact(recv, &mut label_len_buf, "session-token-response:label-len").await?;
            let label_len = u32::from_be_bytes(label_len_buf) as usize;
            let mut label_buf = vec![0u8; label_len];
            read_peer_exact(recv, &mut label_buf, "session-token-response:label").await?;
            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];
            read_peer_exact(recv, &mut frame_len_buf, "session-token-response:frame-len").await?;
            let frame_len = u32::from_be_bytes(frame_len_buf) as usize;
            let mut frame = vec![0u8; frame_len];
            read_peer_exact(recv, &mut frame, "session-token-response:frame-body").await?;

            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
                )),
            }
        }

        // 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 application_crypto_active = self
            .application_crypto_key_for_connection(Some(connection_id.as_ref()))
            .is_some();
        #[cfg(not(target_arch = "wasm32"))]
        let persistent_control_required =
            self.native_admission_stream_is_persistent(connection_id.as_ref());
        #[cfg(target_arch = "wasm32")]
        let persistent_control_required = false;
        // Once a native managed session declares a persistent control route,
        // route repair must replace that route with the same contract. Falling
        // back to a one-shot keyed presentation leaves the QUIC leg alive but
        // removes its generation-bound application-route proof when the short
        // stream closes.
        let stream_contract = outgoing_session_token_stream_contract(
            cfg!(target_arch = "wasm32"),
            application_crypto_active,
            persistent_control_required,
        );
        let token_msg = crate::native_protocol::NativeMainMessage::session_token_presentation_with_payload_and_device_id(
            token,
            options.token_payload,
            options.device_id,
        )
        .with_session_token_stream_contract(stream_contract)
        .with_session_token_reciprocal_requested(options.reciprocal_requested);
        let response_ack_requested = token_msg.session_token_response_ack_requested();
        let serialized = serde_json::to_vec(&token_msg).map_err(|error| {
            format!("failed to serialize session-token presentation: {}", error)
        })?;

        #[cfg(not(target_arch = "wasm32"))]
        let transport_stable_id = connection.stable_id() as u64;
        let (send, recv) = connection.open_bi().await.map_err(|error| {
            format!(
                "failed to open bi-stream for session-token presentation: {}",
                error
            )
        })?;
        // Admission is SDK control, not product data. Keep it independent of
        // directional application-key readiness so a reconnect can reauthorize
        // after one-sided key agreement. Iroh already authenticates and
        // encrypts this QUIC stream end to end.
        let (mut send, mut recv) = (
            crate::application_crypto_streams::PeerSendStream::plain(send),
            crate::application_crypto_streams::PeerRecvStream::plain(recv),
        );

        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);

        send.write_all(&buf)
            .await
            .map_err(|error| format!("failed to write session-token presentation: {}", error))?;
        send.flush()
            .await
            .map_err(|error| format!("failed to flush 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
        })?;
        if !response.is_session_token_response() {
            let _ = send.finish();
            return Err(
                "host returned unexpected response to session-token presentation".to_string(),
            );
        }

        if response.session_token_approved() == Some(true) {
            #[cfg(not(target_arch = "wasm32"))]
            let (reciprocal_acceptance, reciprocal_ack) = if options.reciprocal_requested {
                let current_transport_stable_id = self
                    .get_connection(endpoint_id)
                    .await
                    .map(|connection| connection.stable_id() as u64);
                if current_transport_stable_id != Some(transport_stable_id) {
                    let _ = send.finish();
                    return Err(format!(
                        "physical transport changed before reciprocal session admission could be validated: response={} current={:?}",
                        transport_stable_id, current_transport_stable_id,
                    ));
                }
                if response.reciprocal_session_token_mode() != Some("inline-v1") {
                    (None, None)
                } else {
                    match response.reciprocal_session_token_presentation_state() {
                    crate::native_protocol::ReciprocalSessionTokenPresentationState::Present(
                        presentation,
                    ) => {
                        let presentation_id = presentation.presentation_id.clone();
                        let reciprocal_remote_node_id = endpoint_id.to_string();
                        let expected_remote_device_id = self
                            .known_remote_device_id_for_incoming_transport(
                                connection_id.as_ref(),
                                reciprocal_remote_node_id.as_str(),
                            )
                            .await;
                        if expected_remote_device_id.as_deref()
                            != Some(presentation.device_id.as_str())
                        {
                            (
                                None,
                                Some((
                                    presentation_id,
                                    false,
                                    None,
                                    Some("authoritative-device-mismatch".to_string()),
                                )),
                            )
                        } else if presentation.stream_contract != stream_contract {
                            (
                                None,
                                Some((
                                    presentation_id,
                                    false,
                                    None,
                                    Some("stream-contract-mismatch".to_string()),
                                )),
                            )
                        } else {
                            let was_already_admitted = self
                                .session_token_registry
                                .is_session_token_admitted_for_connection(connection_id.as_ref());
                            match self
                                .validate_session_token_for_connection_with_payload(
                                    presentation.token.as_str(),
                                    connection_id.as_ref(),
                                    Some(presentation.token_payload.as_str()),
                                )
                                .await
                            {
                                Ok(scope) => {
                                    let device_bound = self
                                        .bind_session_admission_authoritative_device_id(
                                            connection_id.as_ref(),
                                            presentation.device_id.as_str(),
                                        )
                                        .await;
                                    if !device_bound {
                                        self.reject_session_connection(
                                            connection_id.as_ref(),
                                            "reciprocal-authoritative-device-mismatch",
                                        );
                                        (
                                            None,
                                            Some((
                                                presentation_id,
                                                false,
                                                None,
                                                Some("authoritative-device-mismatch".to_string()),
                                            )),
                                        )
                                    } else {
                                        self.bind_inbound_native_admission_stream_contract(
                                            connection_id.as_ref(),
                                            stream_contract,
                                        );
                                        self.bind_inbound_session_token_transport(
                                            connection_id.as_ref(),
                                            transport_stable_id,
                                        );
                                        (
                                            Some((
                                                presentation,
                                                scope.clone(),
                                                was_already_admitted,
                                            )),
                                            Some((presentation_id, true, Some(scope), None)),
                                        )
                                    }
                                }
                                Err(error) => (
                                    None,
                                    Some((
                                        presentation_id,
                                        false,
                                        None,
                                        Some(format!("token-rejected:{error}")),
                                    )),
                                ),
                            }
                        }
                    }
                    crate::native_protocol::ReciprocalSessionTokenPresentationState::Malformed => (
                        None,
                        response.reciprocal_session_token_presentation_id().map(
                            |presentation_id| {
                                (
                                    presentation_id,
                                    false,
                                    None,
                                    Some("malformed-presentation".to_string()),
                                )
                            },
                        ),
                    ),
                    crate::native_protocol::ReciprocalSessionTokenPresentationState::Absent => {
                        (None, None)
                    }
                }
                }
            } else {
                (None, None)
            };
            if response_ack_requested {
                let ack = crate::native_protocol::NativeMainMessage::session_token_response_ack(
                    connection_id.as_ref(),
                );
                #[cfg(not(target_arch = "wasm32"))]
                let ack = match reciprocal_ack.as_ref() {
                    Some((presentation_id, accepted, scope, reason)) => ack
                        .with_reciprocal_session_token_ack(
                            presentation_id,
                            *accepted,
                            scope.as_deref(),
                            reason.as_deref(),
                        ),
                    None => ack,
                };
                let serialized_ack = serde_json::to_vec(&ack).map_err(|error| {
                    format!("failed to serialize session-token response ACK: {error}")
                })?;
                let mut ack_frame = Vec::with_capacity(1 + 4 + 4 + 4 + serialized_ack.len());
                ack_frame.push(0x00);
                ack_frame.extend_from_slice(&(4u32).to_be_bytes());
                ack_frame.extend_from_slice(b"main");
                ack_frame.extend_from_slice(&(serialized_ack.len() as u32).to_be_bytes());
                ack_frame.extend_from_slice(&serialized_ack);
                send.write_all(&ack_frame).await.map_err(|error| {
                    format!("failed to write session-token response ACK: {error}")
                })?;
                send.flush().await.map_err(|error| {
                    format!("failed to flush session-token response ACK: {error}")
                })?;
                println!(
                    "[PlutoRTC][session-token-presentation] dialer acknowledged host response connection_id={} endpoint_id={} stream_contract={:?}",
                    connection_id.as_ref(),
                    endpoint_id,
                    stream_contract,
                );
            }
            #[cfg(not(target_arch = "wasm32"))]
            let reciprocal_committed = reciprocal_acceptance.is_some();
            #[cfg(not(target_arch = "wasm32"))]
            if let Some((presentation, reciprocal_scope, was_already_admitted)) =
                reciprocal_acceptance
            {
                let current_transport_stable_id = self
                    .get_connection(endpoint_id)
                    .await
                    .map(|connection| connection.stable_id() as u64);
                if current_transport_stable_id != Some(transport_stable_id) {
                    let _ = send.finish();
                    return Err(format!(
                        "physical transport changed before reciprocal session admission could be bound: response={} current={:?}",
                        transport_stable_id, current_transport_stable_id,
                    ));
                }
                self.run_post_session_token_admission_side_effects(
                    connection_id.as_ref(),
                    !was_already_admitted,
                )
                .await;
                println!(
                    "[PlutoRTC][session-admission][reciprocal-response] connection_id={} endpoint_id={} transport_stable_id={} scope={} claimed_device_id={} result=bound",
                    connection_id.as_ref(),
                    endpoint_id,
                    transport_stable_id,
                    reciprocal_scope,
                    presentation.device_id,
                );
            }
            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
                );
            }
            self.commit_outbound_session_token_application_security_epoch(
                connection_id.as_ref(),
                token,
            );
            #[cfg(not(target_arch = "wasm32"))]
            {
                let current_transport_stable_id = self
                    .get_connection(endpoint_id)
                    .await
                    .map(|connection| connection.stable_id() as u64);
                if current_transport_stable_id != Some(transport_stable_id) {
                    let _ = send.finish();
                    return Err(format!(
                        "physical transport changed before session approval could be bound: approved={} current={:?}",
                        transport_stable_id, current_transport_stable_id,
                    ));
                }
                if let Ok(mut proofs) = self.remote_session_admission_proofs.write() {
                    proofs.insert(
                        connection_id.to_string(),
                        RemoteSessionAdmissionProof {
                            transport_stable_id,
                            token_fingerprint: crate::session_token::token_fingerprint(token),
                            approval_scope: scope.clone(),
                        },
                    );
                }
                if !options.reciprocal_requested || reciprocal_committed {
                    self.clear_reciprocal_session_admission_request(connection_id.as_ref());
                }
                println!(
                    "[PlutoRTC][session-admission][remote-proof] connection_id={} endpoint_id={} transport_stable_id={} token_fp={}",
                    connection_id.as_ref(),
                    endpoint_id,
                    transport_stable_id,
                    super::core_impl::log_fingerprint(token),
                );
                self.bind_outbound_native_admission_stream_contract(
                    connection_id.as_ref(),
                    stream_contract,
                );
                match (send, recv, stream_contract) {
                    (
                        crate::application_crypto_streams::PeerSendStream::Plain(send),
                        crate::application_crypto_streams::PeerRecvStream::Plain(recv),
                        crate::native_protocol::SessionTokenStreamContract::PersistentControl,
                    ) => {
                        self.install_native_control_stream(
                            connection_id.as_ref(),
                            endpoint_id,
                            transport_stable_id,
                            send,
                            recv,
                            "admission-dialer",
                        )
                        .await;
                    }
                    (send, _, _) => {
                        let _ = send.finish();
                    }
                }
                if !application_crypto_active {
                    if let Err(error) = self
                        .send_typescript_capability_update(
                            connection_id.as_ref(),
                            "native-admission-dialer",
                            None,
                        )
                        .await
                    {
                        eprintln!(
                            "[OpenRTC][capability] native admission dialer advertisement failed connection_id={} error={}",
                            connection_id.as_ref(), error,
                        );
                    }
                }
                // A successful response is already a round trip over this
                // exact physical Iroh generation. Consume that stronger proof
                // directly instead of launching a second ping that can race a
                // replacement. Application crypto remains an independent
                // product-routability gate.
                let _ = self
                    .confirm_managed_connection_readiness_from_transport_proof(
                        connection_id.as_ref(),
                        transport_stable_id,
                    )
                    .await;
            }
            #[cfg(target_arch = "wasm32")]
            let _ = send.finish();
            return Ok(scope);
        }

        let _ = send.finish();
        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,
                reciprocal_requested: false,
            },
        )
        .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,
                reciprocal_requested: false,
            },
        )
        .await
    }
}

#[cfg(test)]
mod admission_security_tests {
    use super::{
        incoming_application_admission_decision, outgoing_session_token_stream_contract,
        IncomingApplicationAdmissionDecision,
    };
    use crate::client::Client;
    use crate::native_protocol::SessionTokenStreamContract;
    use crate::session_token::{GrantScope, SessionAdmission, SessionAdmissionMechanism};

    #[test]
    fn rejected_connection_never_forwards_later_application_streams() {
        let decision = incoming_application_admission_decision(
            true,
            &SessionAdmission::Rejected {
                reason: "invalid token".to_string(),
            },
            false,
        );
        assert_eq!(
            decision,
            IncomingApplicationAdmissionDecision::Reject("invalid token".to_string())
        );
    }

    #[test]
    fn active_registry_requires_current_transport_proof() {
        let accepted = SessionAdmission::Accepted {
            mechanism: SessionAdmissionMechanism::SessionToken,
            scope: Some(GrantScope::from("user-device")),
            authoritative_device_id: None,
        };
        assert_eq!(
            incoming_application_admission_decision(true, &accepted, false),
            IncomingApplicationAdmissionDecision::ClassifyAdmission
        );
        assert_eq!(
            incoming_application_admission_decision(true, &accepted, true),
            IncomingApplicationAdmissionDecision::Forward
        );
        assert_eq!(
            incoming_application_admission_decision(false, &SessionAdmission::Pending, false),
            IncomingApplicationAdmissionDecision::Forward
        );
    }

    #[test]
    fn native_route_repair_preserves_declared_persistent_control() {
        assert_eq!(
            outgoing_session_token_stream_contract(false, true, true),
            SessionTokenStreamContract::PersistentControl
        );
        assert_eq!(
            outgoing_session_token_stream_contract(false, true, false),
            SessionTokenStreamContract::OneShotAdmission
        );
        assert_eq!(
            outgoing_session_token_stream_contract(false, false, false),
            SessionTokenStreamContract::PersistentControl
        );
        assert_eq!(
            outgoing_session_token_stream_contract(true, false, true),
            SessionTokenStreamContract::OneShotAdmission
        );
    }

    #[cfg(all(native, unix))]
    #[tokio::test]
    async fn persisted_managed_scope_grant_is_atomically_replaced_and_owner_only() {
        use std::os::unix::fs::PermissionsExt;

        let directory = std::env::temp_dir().join(format!(
            "openrtc-private-grant-{}",
            crate::session_token::generate_payload_nonce()
        ));
        tokio::fs::create_dir_all(&directory)
            .await
            .expect("create test directory");
        let path = directory.join("openrtc_managed_scope_ticket_user-device.json");

        Client::write_private_managed_scope_grant(&path, br#"{"token":"first"}"#)
            .await
            .expect("write first grant");
        Client::write_private_managed_scope_grant(&path, br#"{"token":"second"}"#)
            .await
            .expect("replace grant");

        assert_eq!(
            tokio::fs::read(&path).await.expect("read grant"),
            br#"{"token":"second"}"#
        );
        let mode = tokio::fs::metadata(&path)
            .await
            .expect("stat grant")
            .permissions()
            .mode()
            & 0o777;
        assert_eq!(mode, 0o600);

        tokio::fs::remove_dir_all(directory)
            .await
            .expect("remove test directory");
    }
}