openrtc 1.0.2

/// Session token registry for gating iroh connections during short-lived sessions
/// (e.g. share page, magic-link).
///
/// # Compound ticket format
/// `<iroh_endpoint_ticket>.<base64url_token_payload>`
///
/// The token payload suffix is appended after the raw iroh ticket using `.`.
/// We must split on the LAST `.` because real endpoint tickets can contain dots
/// (for example through embedded relay URL data).
///
/// Token payload is compact JSON:
/// `{"t":"<token>","s":"<scope>","m":<maxConnections>,"h":"<endpointTicketHash>","e":<expiresAtMs>,"a":"openrtc:endpoint-ticket:v1","n":"<nonce>"}`
///
/// # Backward compatibility
/// If the registry is empty, a tokenless low-level/manual connection may
/// proceed without a grant. Any presented bearer string still fails closed.
use std::collections::HashMap;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, RwLock};

use base64::{engine::general_purpose::URL_SAFE_NO_PAD, Engine as _};

/// Default TTL for newly minted restricted endpoint-ticket grants.
pub const DEFAULT_RESTRICTED_SESSION_TOKEN_TTL_MS: u64 = 15 * 60 * 1000;
pub const TOKEN_PAYLOAD_ENDPOINT_TICKET_AUDIENCE: &str = "openrtc:endpoint-ticket:v1";
const TOKEN_PAYLOAD_NONCE_BYTES: usize = 16;

pub fn now_unix_ms() -> u64 {
    #[cfg(target_arch = "wasm32")]
    {
        js_sys::Date::now().max(0.0) as u64
    }
    #[cfg(not(target_arch = "wasm32"))]
    {
        use std::time::{SystemTime, UNIX_EPOCH};
        SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .map(|duration| duration.as_millis().min(u128::from(u64::MAX)) as u64)
            .unwrap_or(0)
    }
}

#[derive(Debug, Clone, PartialEq, Eq, Hash, serde::Serialize, serde::Deserialize)]
#[serde(transparent)]
/// Admission-grant label carried by restricted session tokens and compound tickets.
///
/// This is distinct from the TypeScript `PeerScope` lifecycle label: `GrantScope`
/// governs authorization, validation, and revoke semantics in the Rust admission layer.
pub struct GrantScope(pub String);

impl GrantScope {
    pub fn new(value: impl Into<String>) -> Self {
        Self(value.into())
    }

    pub fn as_str(&self) -> &str {
        &self.0
    }

    pub fn into_inner(self) -> String {
        self.0
    }
}

impl From<&str> for GrantScope {
    fn from(value: &str) -> Self {
        Self::new(value)
    }
}

impl From<String> for GrantScope {
    fn from(value: String) -> Self {
        Self::new(value)
    }
}

impl From<&GrantScope> for GrantScope {
    fn from(value: &GrantScope) -> Self {
        value.clone()
    }
}

impl AsRef<str> for GrantScope {
    fn as_ref(&self) -> &str {
        self.as_str()
    }
}

impl PartialEq<&str> for GrantScope {
    fn eq(&self, other: &&str) -> bool {
        self.as_str() == *other
    }
}

impl PartialEq<String> for GrantScope {
    fn eq(&self, other: &String) -> bool {
        self.as_str() == other
    }
}

impl std::fmt::Display for GrantScope {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.write_str(&self.0)
    }
}

/// A registered session token.
#[derive(Debug, Clone)]
pub struct SessionToken {
    /// The raw token string (random, high-entropy).
    pub token: String,
    /// Application-defined admission grant (e.g. "share", "magic-link", "friend").
    /// Distinct from the TypeScript `PeerScope` lifecycle label.
    pub scope: GrantScope,
    /// Max number of connections allowed. 0 = unlimited.
    pub max_connections: u32,
    /// Number of connections that have consumed this token so far.
    pub use_count: u32,
    /// Absolute Unix-millisecond expiry. None means legacy/unbounded.
    pub expires_at_ms: Option<u64>,
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub enum SessionAdmissionMechanism {
    SessionToken,
    TrustedNativeBinding,
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub enum SessionAdmission {
    Pending,
    Accepted {
        mechanism: SessionAdmissionMechanism,
        scope: Option<GrantScope>,
        authoritative_device_id: Option<String>,
    },
    Rejected {
        reason: String,
    },
}

/// Stable, low-cardinality fingerprint of a token value, used to correlate
/// duplicate presentations of the *same* token on the *same* connection
/// (the Phase 1 idempotency case) vs. presentations of a *different* token
/// on the same connection (e.g. a token rotation, which must re-validate).
///
/// The fingerprint is the first 16 hex chars of SHA-256 over the raw token
/// bytes — short enough to log, long enough to make collisions astronomically
/// unlikely in practice. It is intentionally not cryptographically reversible:
/// even logged in production, it does not leak the token.
pub fn token_fingerprint(token: &str) -> String {
    use sha2::{Digest, Sha256};
    let digest = Sha256::digest(token.as_bytes());
    hex::encode(&digest[..8])
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct NativeTrustedConnectionContext {
    pub connection_id: String,
    pub remote_node_id: Option<String>,
    pub known_device_id: Option<String>,
    pub claimed_device_id: Option<String>,
}

pub type NativeTrustedConnectionVerifier =
    Arc<dyn Fn(&NativeTrustedConnectionContext) -> Option<String> + Send + Sync>;

/// RAII handle returned by [`SessionTokenRegistry::begin_admission_response`].
/// While this guard exists, the registry reports
/// `is_session_token_admission_in_flight(connection_id) == true`. Drop it
/// only after the response has been fully written to the wire.
pub struct AdmissionResponseGuard {
    connection_id: String,
    activities: Arc<RwLock<HashMap<String, AdmissionActivity>>>,
}

/// Exclusive outbound session-token presenter for one logical connection.
/// Duplicate auto-connect passes may observe the same transport concurrently,
/// but only this guard's owner may open an admission request stream.
pub struct AdmissionPresentationGuard {
    connection_id: String,
    activities: Arc<RwLock<HashMap<String, AdmissionActivity>>>,
}

impl Drop for AdmissionPresentationGuard {
    fn drop(&mut self) {
        if let Ok(mut activities) = self.activities.write() {
            if let Some(activity) = activities.get_mut(&self.connection_id) {
                activity.presentation_active = false;
                if activity.response_writers == 0
                    && !activity.presentation_active
                    && !activity.retirement_active
                {
                    activities.remove(&self.connection_id);
                }
            }
        }
    }
}

impl AdmissionResponseGuard {
    pub fn connection_id(&self) -> &str {
        &self.connection_id
    }
}

impl Drop for AdmissionResponseGuard {
    fn drop(&mut self) {
        if let Ok(mut activities) = self.activities.write() {
            if let Some(activity) = activities.get_mut(&self.connection_id) {
                if activity.response_writers > 0 {
                    activity.response_writers -= 1;
                }
                if activity.response_writers == 0
                    && !activity.presentation_active
                    && !activity.retirement_active
                {
                    activities.remove(&self.connection_id);
                }
            }
        }
    }
}

#[derive(Debug, Default)]
struct AdmissionActivity {
    response_writers: u32,
    presentation_active: bool,
    retirement_active: bool,
}

/// Exclusive terminal-retirement ownership for one connection. Admission
/// response writers cannot begin until this guard is dropped.
pub struct AdmissionRetirementGuard {
    connection_id: String,
    activities: Arc<RwLock<HashMap<String, AdmissionActivity>>>,
}

impl Drop for AdmissionRetirementGuard {
    fn drop(&mut self) {
        if let Ok(mut activities) = self.activities.write() {
            if let Some(activity) = activities.get_mut(&self.connection_id) {
                activity.retirement_active = false;
                if activity.response_writers == 0 && !activity.presentation_active {
                    activities.remove(&self.connection_id);
                }
            }
        }
    }
}

/// Thread-safe registry of active session tokens.
#[derive(Default, Clone)]
pub struct SessionTokenRegistry {
    inner: Arc<RwLock<HashMap<String, SessionToken>>>,
    /// Maps connection_id → scope for connections validated with a token.
    /// Used by `revoke_by_scope` to return connection IDs that should be disconnected.
    validated_connections: Arc<RwLock<HashMap<String, GrantScope>>>,
    /// Tokens this runtime validated from the remote peer. Keep this separate
    /// from `admissions`, which is also updated when the remote host approves
    /// our outbound presentation (the opposite direction of trust).
    inbound_token_admissions: Arc<RwLock<HashMap<String, (GrantScope, String)>>>,
    admissions: Arc<RwLock<HashMap<String, SessionAdmission>>>,
    /// Per-connection record of the token fingerprint that produced the
    /// current admission verdict. Used by Phase 1 callers to distinguish
    /// "same token replayed" (idempotent) from "different token presented on
    /// the same connection" (must re-validate).
    admission_fingerprints: Arc<RwLock<HashMap<String, String>>>,
    /// Monotonic logical-admission generation for each connection. Unlike the
    /// token fingerprint, this advances when identity/scope admission state is
    /// replaced even if the same bearer token is reused.
    admission_epochs: Arc<RwLock<HashMap<String, u64>>>,
    next_admission_epoch: Arc<AtomicU64>,
    /// Token fingerprint whose application-security epoch has been committed
    /// after the approval response reached the peer. Validation is phase one;
    /// this map advances only at the existing post-response commit point so a
    /// failed response write cannot rotate crypto behind the presenter's back.
    application_security_epoch_fingerprints: Arc<RwLock<HashMap<String, String>>>,
    /// Per-connection counter of in-flight session-token response writers.
    /// While this is non-zero for a connection, the transport must not be
    /// retired/replaced (otherwise the response stream the dialer is reading
    /// closes mid-flight → `0 bytes read`).
    admission_activities: Arc<RwLock<HashMap<String, AdmissionActivity>>>,
    /// Maps `token_fingerprint:payload_nonce` → connection_id. Payload nonce
    /// replay on a different connection is rejected when modern peers present
    /// the compact token payload suffix alongside the raw token.
    seen_payload_nonces: Arc<RwLock<HashMap<String, String>>>,
    native_trusted_connection_verifier: Arc<RwLock<Option<NativeTrustedConnectionVerifier>>>,
}

impl SessionTokenRegistry {
    pub fn new() -> Self {
        Self::default()
    }

    /// Register a new token. Replaces any existing token with the same value.
    pub fn register(&self, token: String, scope: impl Into<GrantScope>, max_connections: u32) {
        self.register_with_expiry_ms(token, scope, max_connections, None);
    }

    /// Register a new token with an optional absolute Unix-millisecond expiry.
    pub fn register_with_expiry_ms(
        &self,
        token: String,
        scope: impl Into<GrantScope>,
        max_connections: u32,
        expires_at_ms: Option<u64>,
    ) {
        let scope = scope.into();
        if token.trim().is_empty() || scope.as_str().trim().is_empty() {
            return;
        }
        let entry = SessionToken {
            token: token.clone(),
            scope,
            max_connections,
            use_count: 0,
            expires_at_ms,
        };
        if let Ok(mut map) = self.inner.write() {
            map.insert(token, entry);
        }
    }

    /// Validate a token and increment its use count.
    ///
    /// Returns `Ok(scope)` on success, `Err(reason)` on failure.
    /// If the registry is empty, only a tokenless low-level/manual connection
    /// receives the compatibility verdict. A presented bearer token must
    /// always match a registered capability; otherwise arbitrary strings could
    /// be promoted into trust by higher layers.
    pub fn validate_and_consume(&self, token: &str) -> Result<GrantScope, String> {
        self.validate_and_consume_for_connection(token, None)
    }

    /// Validate a token, increment its use count, and optionally record the
    /// connection_id so `revoke_by_scope` can return affected connections.
    pub fn validate_and_consume_for_connection(
        &self,
        token: &str,
        connection_id: Option<&str>,
    ) -> Result<GrantScope, String> {
        self.validate_and_consume_for_connection_with_payload(token, connection_id, None)
    }

    /// Validate a raw session token with optional compact payload metadata.
    ///
    /// Legacy callers may omit `payload_suffix`; modern compound-ticket callers
    /// should provide it so the host can enforce the payload audience, expiry,
    /// token binding, and nonce replay checks instead of seeing only the raw
    /// bearer token.
    pub fn validate_and_consume_for_connection_with_payload(
        &self,
        token: &str,
        connection_id: Option<&str>,
        payload_suffix: Option<&str>,
    ) -> Result<GrantScope, String> {
        let presented_fp = if !token.is_empty() {
            Some(token_fingerprint(token))
        } else {
            None
        };

        let cached_inbound = if let Some(conn_id) = connection_id {
            if let SessionAdmission::Rejected { reason } = self.admission(conn_id) {
                return Err(reason);
            }
            self.inbound_token_admissions
                .read()
                .ok()
                .and_then(|admissions| admissions.get(conn_id).cloned())
        } else {
            None
        };

        // Validation and use-count consumption share one write lock. A token
        // with max_connections=1 therefore cannot be admitted twice by two
        // concurrent streams that both observed the old count.
        let mut map = self
            .inner
            .write()
            .map_err(|_| "registry lock poisoned".to_string())?;

        // Preserve tokenless low-level/manual operation, but never treat an
        // unvalidated presented bearer token as an accepted capability.
        if map.is_empty() {
            return if token.trim().is_empty() {
                Ok(GrantScope::new(String::new()))
            } else {
                Err("unknown session token".to_string())
            };
        }

        let entry = map
            .get_mut(token)
            .ok_or_else(|| "unknown session token".to_string())?;

        if let Some(suffix) = payload_suffix {
            self.validate_presented_payload_suffix(
                token,
                connection_id,
                suffix,
                &entry.scope,
                entry.max_connections,
            )?;
        }

        if entry
            .expires_at_ms
            .is_some_and(|expires_at_ms| expires_at_ms <= now_unix_ms())
        {
            map.remove(token);
            return Err("session token expired".to_string());
        }

        // Phase 1 idempotency is valid only while the registered capability
        // itself is still live. Expiry/revocation must not be bypassed by a
        // deterministic connection id re-presenting an old token.
        if let Some((cached_scope, cached_fp)) = cached_inbound {
            if presented_fp.as_deref() == Some(cached_fp.as_str()) {
                if cached_scope == entry.scope {
                    return Ok(cached_scope);
                }
                return Err("session token scope changed after admission".to_string());
            }
        }

        if entry.max_connections > 0 && entry.use_count >= entry.max_connections {
            return Err(format!(
                "session token exhausted (max={} used={})",
                entry.max_connections, entry.use_count
            ));
        }

        let scope = entry.scope.clone();
        entry.use_count = entry.use_count.saturating_add(1);
        drop(map);

        // Record the connection → scope binding for revocation, the
        // admission verdict, and the token fingerprint that produced it.
        if let Some(conn_id) = connection_id {
            if let Ok(mut conns) = self.validated_connections.write() {
                conns.insert(conn_id.to_string(), scope.clone());
            }
            if let Some(fp) = presented_fp.as_ref() {
                if let Ok(mut inbound) = self.inbound_token_admissions.write() {
                    inbound.insert(conn_id.to_string(), (scope.clone(), fp.clone()));
                }
            }
            self.mark_accepted(
                conn_id,
                SessionAdmissionMechanism::SessionToken,
                Some(scope.clone()),
                None,
            );
            if let Some(fp) = presented_fp.as_ref() {
                if let Ok(mut fps) = self.admission_fingerprints.write() {
                    fps.insert(conn_id.to_string(), fp.clone());
                }
            }
        }

        Ok(scope)
    }

    /// Phase 1 verdict-only API. Returns the deterministic admission verdict
    /// for `(token, connection_id)` *without* running any post-admission side
    /// effects (no `accept_replacement_peer`, no WebRTC recovery, no
    /// transport replacement). Callers must:
    ///   1. Call this to compute the verdict.
    ///   2. Write + flush the response to the wire.
    ///   3. Then run their post-admission side effects (e.g. via
    ///      `Client::run_post_session_token_admission_side_effects`).
    ///
    /// This separation is what eliminates the `0 bytes read` race where the
    /// host's lifecycle hooks would retire the very transport the response
    /// writer was using before it could flush.
    pub fn validate_with_cached_response(
        &self,
        token: &str,
        connection_id: &str,
    ) -> Result<GrantScope, String> {
        self.validate_with_cached_response_and_payload(token, connection_id, None)
    }

    pub fn validate_with_cached_response_and_payload(
        &self,
        token: &str,
        connection_id: &str,
        payload_suffix: Option<&str>,
    ) -> Result<GrantScope, String> {
        self.validate_and_consume_for_connection_with_payload(
            token,
            Some(connection_id),
            payload_suffix,
        )
    }

    /// Mark a session-token response writer as in-flight for `connection_id`.
    /// Returns a guard whose `Drop` decrements the counter. While any guard
    /// is alive for a given connection, the transport must not be
    /// closed/retired by lifecycle code (see `is_session_token_admission_in_flight`).
    pub fn begin_admission_response(&self, connection_id: &str) -> AdmissionResponseGuard {
        self.try_begin_admission_response(connection_id)
            .expect("admission retirement is not active")
    }

    /// Atomically begin a response writer unless terminal retirement already
    /// owns this connection.
    pub fn try_begin_admission_response(
        &self,
        connection_id: &str,
    ) -> Option<AdmissionResponseGuard> {
        let mut activities = self.admission_activities.write().ok()?;
        let activity = activities.entry(connection_id.to_string()).or_default();
        if activity.retirement_active {
            return None;
        }
        activity.response_writers = activity.response_writers.saturating_add(1);
        Some(AdmissionResponseGuard {
            connection_id: connection_id.to_string(),
            activities: self.admission_activities.clone(),
        })
    }

    /// Atomically acquire the sole outbound presenter for one connection.
    /// Host response writers may coexist because they operate in the opposite
    /// direction, while duplicate outbound attempts and retirement are fenced.
    pub fn try_begin_admission_presentation(
        &self,
        connection_id: &str,
    ) -> Option<AdmissionPresentationGuard> {
        let mut activities = self.admission_activities.write().ok()?;
        let activity = activities.entry(connection_id.to_string()).or_default();
        if activity.retirement_active || activity.presentation_active {
            return None;
        }
        activity.presentation_active = true;
        Some(AdmissionPresentationGuard {
            connection_id: connection_id.to_string(),
            activities: self.admission_activities.clone(),
        })
    }

    /// Atomically acquire terminal retirement unless a response writer already
    /// owns the connection. This closes the check-then-retire race at the
    /// admission deadline.
    pub fn try_begin_admission_retirement(
        &self,
        connection_id: &str,
    ) -> Option<AdmissionRetirementGuard> {
        let mut activities = self.admission_activities.write().ok()?;
        let activity = activities.entry(connection_id.to_string()).or_default();
        if activity.retirement_active
            || activity.response_writers > 0
            || activity.presentation_active
        {
            return None;
        }
        activity.retirement_active = true;
        Some(AdmissionRetirementGuard {
            connection_id: connection_id.to_string(),
            activities: self.admission_activities.clone(),
        })
    }

    /// True while a session-token response writer or the sole outbound
    /// presenter is mid-flight for `connection_id`. Lifecycle code must defer
    /// transport teardown while this returns true.
    pub fn is_session_token_admission_in_flight(&self, connection_id: &str) -> bool {
        self.admission_activities
            .read()
            .ok()
            .and_then(|activities| {
                activities
                    .get(connection_id)
                    .map(|activity| activity.response_writers > 0 || activity.presentation_active)
            })
            .unwrap_or(false)
    }

    pub fn is_session_token_presentation_in_flight(&self, connection_id: &str) -> bool {
        self.admission_activities
            .read()
            .ok()
            .and_then(|activities| {
                activities
                    .get(connection_id)
                    .map(|activity| activity.presentation_active)
            })
            .unwrap_or(false)
    }

    pub fn admission_fingerprint(&self, connection_id: &str) -> Option<String> {
        self.admission_fingerprints
            .read()
            .ok()
            .and_then(|fps| fps.get(connection_id).cloned())
    }

    pub fn admission_epoch(&self, connection_id: &str) -> Option<u64> {
        self.admission_epochs
            .read()
            .ok()
            .and_then(|epochs| epochs.get(connection_id).copied())
    }

    fn advance_admission_epoch(&self, connection_id: &str) {
        let epoch = self.next_admission_epoch.fetch_add(1, Ordering::Relaxed) + 1;
        if let Ok(mut epochs) = self.admission_epochs.write() {
            epochs.insert(connection_id.to_string(), epoch);
        }
    }

    pub(crate) fn application_security_epoch_fingerprint(
        &self,
        connection_id: &str,
    ) -> Option<String> {
        self.application_security_epoch_fingerprints
            .read()
            .ok()
            .and_then(|fingerprints| fingerprints.get(connection_id).cloned())
    }

    pub(crate) fn commit_application_security_epoch(&self, connection_id: &str) {
        let Some(fingerprint) = self.admission_fingerprint(connection_id) else {
            return;
        };
        if let Ok(mut committed) = self.application_security_epoch_fingerprints.write() {
            committed.insert(connection_id.to_string(), fingerprint);
        }
    }

    /// Revoke a specific token.
    pub fn revoke(&self, token: &str) -> Vec<String> {
        if let Ok(mut map) = self.inner.write() {
            map.remove(token);
        }
        let fingerprint = token_fingerprint(token);
        let nonce_key_prefix = format!("{}:", fingerprint);
        if let Ok(mut seen) = self.seen_payload_nonces.write() {
            seen.retain(|key, _| !key.starts_with(&nonce_key_prefix));
        }
        let mut affected = Vec::new();
        if let Ok(mut inbound) = self.inbound_token_admissions.write() {
            inbound.retain(|connection_id, (_, admitted_fingerprint)| {
                if admitted_fingerprint == &fingerprint {
                    affected.push(connection_id.clone());
                    false
                } else {
                    true
                }
            });
        }
        if let Ok(mut validated) = self.validated_connections.write() {
            for connection_id in &affected {
                validated.remove(connection_id);
            }
        }
        if let Ok(mut admissions) = self.admissions.write() {
            for connection_id in &affected {
                admissions.insert(
                    connection_id.clone(),
                    SessionAdmission::Rejected {
                        reason: "session token revoked".to_string(),
                    },
                );
            }
        }
        for connection_id in &affected {
            self.advance_admission_epoch(connection_id);
        }
        if let Ok(mut fingerprints) = self.admission_fingerprints.write() {
            for connection_id in &affected {
                fingerprints.remove(connection_id);
            }
        }
        if let Ok(mut fingerprints) = self.application_security_epoch_fingerprints.write() {
            for connection_id in &affected {
                fingerprints.remove(connection_id);
            }
        }
        affected.sort();
        affected.dedup();
        affected
    }

    /// Revoke all tokens and logical admissions with the given scope.
    ///
    /// `validated_connections` records tokens this runtime validated from a
    /// peer. `admissions` also records the opposite direction after a remote
    /// host approves this runtime's token. Scope revocation is a logical-session
    /// boundary, so both directions must be returned to the lifecycle owner.
    pub fn revoke_by_scope(&self, scope: impl AsRef<str>) -> Vec<String> {
        let scope = scope.as_ref();
        if let Ok(mut map) = self.inner.write() {
            map.retain(|_, v| v.scope.as_str() != scope);
        }

        let mut affected = std::collections::HashSet::new();
        if let Ok(mut conns) = self.validated_connections.write() {
            conns.retain(|conn_id, conn_scope| {
                if conn_scope.as_str() == scope {
                    affected.insert(conn_id.clone());
                    false
                } else {
                    true
                }
            });
        }
        if let Ok(mut admissions) = self.admissions.write() {
            admissions.retain(|connection_id, admission| {
                let matches_scope = matches!(
                    admission,
                    SessionAdmission::Accepted {
                        scope: Some(admission_scope),
                        ..
                    } if admission_scope.as_str() == scope
                );
                if matches_scope {
                    affected.insert(connection_id.clone());
                }
                !matches_scope
            });
        }
        if let Ok(mut inbound) = self.inbound_token_admissions.write() {
            for connection_id in &affected {
                inbound.remove(connection_id);
            }
        }
        if let Ok(mut fps) = self.admission_fingerprints.write() {
            for connection_id in &affected {
                fps.remove(connection_id);
            }
        }
        if let Ok(mut fps) = self.application_security_epoch_fingerprints.write() {
            for connection_id in &affected {
                fps.remove(connection_id);
            }
        }
        if let Ok(mut epochs) = self.admission_epochs.write() {
            for connection_id in &affected {
                epochs.remove(connection_id);
            }
        }
        let mut affected = affected.into_iter().collect::<Vec<_>>();
        affected.sort();
        affected
    }

    /// Remove all tokens and validated connection bindings.
    pub fn clear(&self) {
        if let Ok(mut map) = self.inner.write() {
            map.clear();
        }
        if let Ok(mut conns) = self.validated_connections.write() {
            conns.clear();
        }
        if let Ok(mut admissions) = self.admissions.write() {
            admissions.clear();
        }
        if let Ok(mut inbound) = self.inbound_token_admissions.write() {
            inbound.clear();
        }
        if let Ok(mut fps) = self.admission_fingerprints.write() {
            fps.clear();
        }
        if let Ok(mut fps) = self.application_security_epoch_fingerprints.write() {
            fps.clear();
        }
        if let Ok(mut epochs) = self.admission_epochs.write() {
            epochs.clear();
        }
        if let Ok(mut activities) = self.admission_activities.write() {
            activities.clear();
        }
        if let Ok(mut seen) = self.seen_payload_nonces.write() {
            seen.clear();
        }
    }

    /// True if no restricted session tokens are registered.
    pub fn is_empty(&self) -> bool {
        self.inner.read().map(|m| m.is_empty()).unwrap_or(true)
    }

    pub fn admission(&self, connection_id: &str) -> SessionAdmission {
        self.admissions
            .read()
            .ok()
            .and_then(|admissions| admissions.get(connection_id).cloned())
            .unwrap_or(SessionAdmission::Pending)
    }

    /// True when this runtime validated a token presented by the remote peer.
    ///
    /// Used to detect duplicate token frames on the same logical connection (e.g. new iroh
    /// substreams re-sending the compound ticket) — those must not re-trigger native WebRTC
    /// upgrade recovery, which is only appropriate for the **first** token presentation.
    pub fn is_session_token_admitted_for_connection(&self, connection_id: &str) -> bool {
        self.inbound_token_admissions
            .read()
            .map(|admissions| admissions.contains_key(connection_id))
            .unwrap_or(false)
    }

    pub fn mark_rejected(&self, connection_id: &str, reason: impl Into<String>) {
        if let Ok(mut admissions) = self.admissions.write() {
            admissions.insert(
                connection_id.to_string(),
                SessionAdmission::Rejected {
                    reason: reason.into(),
                },
            );
        }
        self.advance_admission_epoch(connection_id);
    }

    pub fn mark_accepted(
        &self,
        connection_id: &str,
        mechanism: SessionAdmissionMechanism,
        scope: Option<GrantScope>,
        authoritative_device_id: Option<String>,
    ) {
        if let Ok(mut admissions) = self.admissions.write() {
            admissions.insert(
                connection_id.to_string(),
                SessionAdmission::Accepted {
                    mechanism,
                    scope,
                    authoritative_device_id,
                },
            );
        }
        self.advance_admission_epoch(connection_id);
    }

    pub fn bind_connection_scope(
        &self,
        connection_id: &str,
        scope: impl Into<GrantScope>,
        authoritative_device_id: Option<String>,
    ) {
        let scope = scope.into();
        if let Ok(mut conns) = self.validated_connections.write() {
            conns.insert(connection_id.to_string(), scope.clone());
        }
        self.mark_accepted(
            connection_id,
            SessionAdmissionMechanism::SessionToken,
            Some(scope),
            authoritative_device_id,
        );
    }

    pub fn forget_connection(&self, connection_id: &str) {
        if let Ok(mut conns) = self.validated_connections.write() {
            conns.remove(connection_id);
        }
        if let Ok(mut admissions) = self.admissions.write() {
            admissions.remove(connection_id);
        }
        if let Ok(mut inbound) = self.inbound_token_admissions.write() {
            inbound.remove(connection_id);
        }
        if let Ok(mut fps) = self.admission_fingerprints.write() {
            fps.remove(connection_id);
        }
        if let Ok(mut fps) = self.application_security_epoch_fingerprints.write() {
            fps.remove(connection_id);
        }
        if let Ok(mut epochs) = self.admission_epochs.write() {
            epochs.remove(connection_id);
        }
        if let Ok(mut activities) = self.admission_activities.write() {
            activities.remove(connection_id);
        }
    }

    pub fn set_native_trusted_connection_verifier(
        &self,
        verifier: Option<NativeTrustedConnectionVerifier>,
    ) {
        if let Ok(mut slot) = self.native_trusted_connection_verifier.write() {
            *slot = verifier;
        }
    }

    pub fn evaluate_trusted_native_connection(
        &self,
        context: &NativeTrustedConnectionContext,
    ) -> Option<String> {
        self.native_trusted_connection_verifier
            .read()
            .ok()
            .and_then(|slot| slot.as_ref().and_then(|verifier| verifier(context)))
    }

    fn validate_presented_payload_suffix(
        &self,
        token: &str,
        connection_id: Option<&str>,
        payload_suffix: &str,
        expected_scope: &GrantScope,
        max_connections: u32,
    ) -> Result<(), String> {
        let payload = decode_token_payload(payload_suffix)
            .ok_or_else(|| "invalid session token payload".to_string())?;
        if payload.token != token {
            return Err("session token payload mismatch".to_string());
        }
        if !token_payload_metadata_is_valid(&payload) {
            return Err("invalid session token payload metadata".to_string());
        }
        if &payload.scope != expected_scope || payload.max_connections != max_connections {
            return Err("session token payload grant mismatch".to_string());
        }

        // The per-connection nonce binding is single-use anti-replay protection
        // for out-of-band share tickets (`max_connections == 1`). A multi-use or
        // unlimited presence ticket (e.g. `user-device`, `max_connections != 1`)
        // is advertised once and legitimately presented by EVERY dialing peer:
        // each dialer forwards the same signed payload nonce verbatim because it
        // cannot re-mint the ticket-bound payload (see
        // `auto_connect_impl::extracted_token_suffix`). Binding that shared nonce
        // to the first connection would falsely reject every other device as a
        // "nonce replayed" attacker and churn the mesh. Multi-use bounds are
        // enforced independently by `use_count >= max_connections`, so only
        // genuine single-use tokens need the per-connection nonce binding.
        if max_connections != 1 {
            return Ok(());
        }

        let Some(nonce) = payload.nonce.as_deref() else {
            return Ok(());
        };
        let Some(connection_id) = connection_id else {
            return Ok(());
        };

        let replay_key = format!("{}:{}", token_fingerprint(token), nonce);
        let mut seen = self
            .seen_payload_nonces
            .write()
            .map_err(|_| "registry lock poisoned".to_string())?;
        match seen.get(&replay_key) {
            Some(existing_connection_id) if existing_connection_id != connection_id => {
                Err("session token payload nonce replayed".to_string())
            }
            Some(_) => Ok(()),
            None => {
                seen.insert(replay_key, connection_id.to_string());
                Ok(())
            }
        }
    }
}

// --------------------------------------------------------------------------
// Compact token payload for embedding in ticket strings
// --------------------------------------------------------------------------

/// Decoded token payload embedded in a compound ticket.
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub struct TokenPayload {
    /// Raw token value.
    #[serde(rename = "t")]
    pub token: String,
    /// Scope string.
    #[serde(rename = "s")]
    pub scope: GrantScope,
    /// Max connections (0 = unlimited).
    #[serde(rename = "m")]
    pub max_connections: u32,
    /// SHA-256 hash of the endpoint ticket this payload was minted for.
    #[serde(rename = "h", default, skip_serializing_if = "Option::is_none")]
    pub ticket_hash: Option<String>,
    /// Absolute Unix-millisecond expiry.
    #[serde(rename = "e", default, skip_serializing_if = "Option::is_none")]
    pub expires_at_ms: Option<u64>,
    /// Explicit intended consumer for this compact payload.
    #[serde(rename = "a", default, skip_serializing_if = "Option::is_none")]
    pub audience: Option<String>,
    /// Per-payload nonce used to distinguish otherwise equivalent grants.
    #[serde(rename = "n", default, skip_serializing_if = "Option::is_none")]
    pub nonce: Option<String>,
}

/// Encode a token payload as a base64url (no padding) string.
pub fn encode_token_payload(
    token: &str,
    scope: impl Into<GrantScope>,
    max_connections: u32,
) -> String {
    let scope = scope.into();
    let payload = TokenPayload {
        token: token.to_string(),
        scope,
        max_connections,
        ticket_hash: None,
        expires_at_ms: None,
        audience: None,
        nonce: None,
    };
    let json = serde_json::to_string(&payload).unwrap_or_default();
    URL_SAFE_NO_PAD.encode(json.as_bytes())
}

/// Stable hash that binds a compact token payload to the endpoint ticket it
/// grants access to. The full SHA-256 digest is kept because this value is
/// carried inside bearer material rather than used only for logs.
pub fn endpoint_ticket_hash(iroh_ticket: &str) -> String {
    use sha2::{Digest, Sha256};
    let digest = Sha256::digest(iroh_ticket.as_bytes());
    hex::encode(digest)
}

/// Encode a token payload and bind it to a specific endpoint ticket.
pub fn encode_token_payload_for_ticket(
    iroh_ticket: &str,
    token: &str,
    scope: impl Into<GrantScope>,
    max_connections: u32,
) -> String {
    encode_token_payload_for_ticket_with_expiry(iroh_ticket, token, scope, max_connections, None)
}

/// Encode a token payload, bind it to a specific endpoint ticket, and optionally
/// include the absolute expiry enforced by the host registry.
pub fn encode_token_payload_for_ticket_with_expiry(
    iroh_ticket: &str,
    token: &str,
    scope: impl Into<GrantScope>,
    max_connections: u32,
    expires_at_ms: Option<u64>,
) -> String {
    let scope = scope.into();
    let payload = TokenPayload {
        token: token.to_string(),
        scope,
        max_connections,
        ticket_hash: Some(endpoint_ticket_hash(iroh_ticket)),
        expires_at_ms,
        audience: Some(TOKEN_PAYLOAD_ENDPOINT_TICKET_AUDIENCE.to_string()),
        nonce: Some(generate_payload_nonce()),
    };
    let json = serde_json::to_string(&payload).unwrap_or_default();
    URL_SAFE_NO_PAD.encode(json.as_bytes())
}

/// Decode a base64url token payload string.
pub fn decode_token_payload(encoded: &str) -> Option<TokenPayload> {
    let bytes = URL_SAFE_NO_PAD.decode(encoded).ok()?;
    serde_json::from_slice(&bytes).ok()
}

/// Check whether a decoded payload is bound to the provided endpoint ticket.
///
/// Legacy payloads without `h` are accepted for backward compatibility.
pub fn token_payload_matches_ticket(iroh_ticket: &str, payload: &TokenPayload) -> bool {
    let ticket_matches = payload
        .ticket_hash
        .as_deref()
        .map_or(true, |hash| hash == endpoint_ticket_hash(iroh_ticket));
    let not_expired = payload
        .expires_at_ms
        .map_or(true, |expires_at_ms| expires_at_ms > now_unix_ms());
    ticket_matches && not_expired && token_payload_metadata_is_valid(payload)
}

pub fn token_payload_metadata_is_valid(payload: &TokenPayload) -> bool {
    let audience_matches = payload.audience.as_deref().map_or(true, |audience| {
        audience == TOKEN_PAYLOAD_ENDPOINT_TICKET_AUDIENCE
    });
    let nonce_is_valid = payload
        .nonce
        .as_deref()
        .map_or(true, is_valid_payload_nonce);
    let not_expired = payload
        .expires_at_ms
        .map_or(true, |expires_at_ms| expires_at_ms > now_unix_ms());
    audience_matches && nonce_is_valid && not_expired
}

/// Decode a token payload and reject it if its endpoint-ticket binding does
/// not match the accompanying iroh ticket.
pub fn decode_token_payload_for_ticket(iroh_ticket: &str, encoded: &str) -> Option<TokenPayload> {
    let payload = decode_token_payload(encoded)?;
    token_payload_matches_ticket(iroh_ticket, &payload).then_some(payload)
}

/// Split a (possibly compound) ticket string into `(iroh_ticket, Option<token_payload>)`.
///
/// If the string contains a `.`, everything before the last `.` is the iroh ticket
/// and everything after is the base64url-encoded token payload.
pub fn split_compound_ticket(compound: &str) -> (&str, Option<&str>) {
    match compound.rfind('.') {
        Some(pos) => (&compound[..pos], Some(&compound[pos + 1..])),
        None => (compound, None),
    }
}

/// Build a compound ticket by appending a token suffix bound to the endpoint ticket.
pub fn build_compound_ticket(
    iroh_ticket: &str,
    token: &str,
    scope: impl Into<GrantScope>,
    max_connections: u32,
) -> String {
    build_compound_ticket_with_expiry(iroh_ticket, token, scope, max_connections, None)
}

/// Build a compound ticket with an optional host-enforced absolute expiry.
pub fn build_compound_ticket_with_expiry(
    iroh_ticket: &str,
    token: &str,
    scope: impl Into<GrantScope>,
    max_connections: u32,
    expires_at_ms: Option<u64>,
) -> String {
    let scope = scope.into();
    let suffix = encode_token_payload_for_ticket_with_expiry(
        iroh_ticket,
        token,
        scope,
        max_connections,
        expires_at_ms,
    );
    format!("{}.{}", iroh_ticket, suffix)
}

// --------------------------------------------------------------------------
// Token generation
// --------------------------------------------------------------------------

/// Generate a cryptographically random 32-character alphanumeric token.
/// Uses the platform's secure RNG (getrandom on WASM, OsRng on native).
pub fn generate_token() -> String {
    let mut bytes = [0u8; 24]; // 24 bytes → 32 base64url chars (no padding)
    getrandom::getrandom(&mut bytes).expect("getrandom failed");
    URL_SAFE_NO_PAD.encode(bytes)
}

pub fn generate_payload_nonce() -> String {
    let mut bytes = [0u8; TOKEN_PAYLOAD_NONCE_BYTES];
    getrandom::getrandom(&mut bytes).expect("getrandom failed");
    URL_SAFE_NO_PAD.encode(bytes)
}

fn is_valid_payload_nonce(nonce: &str) -> bool {
    URL_SAFE_NO_PAD
        .decode(nonce)
        .map(|bytes| bytes.len() == TOKEN_PAYLOAD_NONCE_BYTES)
        .unwrap_or(false)
}

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

    #[test]
    fn test_split_no_dot() {
        let ticket = "endpointabcdef123456";
        let (iroh, token) = split_compound_ticket(ticket);
        assert_eq!(iroh, "endpointabcdef123456");
        assert!(token.is_none());
    }

    #[test]
    fn test_split_with_dot() {
        let encoded = encode_token_payload("my-token", "share", 5);
        let compound = format!("endpointabcdef123456.{}", encoded);
        let (iroh, token_str) = split_compound_ticket(&compound);
        assert_eq!(iroh, "endpointabcdef123456");
        let payload = decode_token_payload(token_str.unwrap()).unwrap();
        assert_eq!(payload.token, "my-token");
        assert_eq!(payload.scope, "share");
        assert_eq!(payload.max_connections, 5);
    }

    #[test]
    fn test_split_uses_last_dot_for_ticket_with_embedded_dots() {
        let encoded = encode_token_payload("my-token", "trusted", 0);
        let iroh_ticket = "ticket.with.embedded.dots";
        let compound = format!("{}.{}", iroh_ticket, encoded);
        let (recovered_ticket, token_str) = split_compound_ticket(&compound);

        assert_eq!(recovered_ticket, iroh_ticket);
        let payload = decode_token_payload(token_str.expect("token suffix")).expect("payload");
        assert_eq!(payload.token, "my-token");
        assert_eq!(payload.scope, "trusted");
    }

    #[test]
    fn test_registry_empty_gate() {
        let registry = SessionTokenRegistry::new();
        assert!(registry.validate_and_consume("").is_ok());
        assert_eq!(
            registry.validate_and_consume("anything").unwrap_err(),
            "unknown session token",
        );
    }

    #[test]
    fn test_registry_validate_consume() {
        let registry = SessionTokenRegistry::new();
        registry.register("tok1".to_string(), "share".to_string(), 2);

        assert_eq!(registry.validate_and_consume("tok1").unwrap(), "share");
        assert_eq!(registry.validate_and_consume("tok1").unwrap(), "share");
        // Third attempt should fail (max=2).
        assert!(registry.validate_and_consume("tok1").is_err());
    }

    #[test]
    fn test_registry_unlimited() {
        let registry = SessionTokenRegistry::new();
        registry.register("tok2".to_string(), "magic-link".to_string(), 0);

        for _ in 0..100 {
            assert!(registry.validate_and_consume("tok2").is_ok());
        }
    }

    #[test]
    fn test_registry_rejects_expired_token() {
        let registry = SessionTokenRegistry::new();
        registry.register_with_expiry_ms(
            "expired-token".to_string(),
            "share".to_string(),
            0,
            Some(now_unix_ms().saturating_sub(1)),
        );
        registry.register("sentinel".to_string(), "sentinel".to_string(), 0);

        assert_eq!(
            registry.validate_and_consume("expired-token").unwrap_err(),
            "session token expired"
        );
        assert_eq!(
            registry.validate_and_consume("expired-token").unwrap_err(),
            "unknown session token"
        );
    }

    #[test]
    fn test_revoke_by_scope() {
        let registry = SessionTokenRegistry::new();
        registry.register("tok1".to_string(), "share".to_string(), 0);
        registry.register("tok2".to_string(), "share".to_string(), 0);
        registry.register("tok3".to_string(), "other".to_string(), 0);

        registry.revoke_by_scope("share");

        // Share tokens are gone → registry still has "other" so empty-gate doesn't apply.
        assert!(registry.validate_and_consume("tok1").is_err());
        assert!(registry.validate_and_consume("tok2").is_err());
        assert!(registry.validate_and_consume("tok3").is_ok());
    }

    #[test]
    fn test_generate_token_length() {
        let t = generate_token();
        // 24 bytes → 32 base64url chars (no padding).
        assert_eq!(t.len(), 32);
    }

    #[test]
    fn test_build_and_split_compound_ticket_roundtrip() {
        let iroh_ticket = "abcdef1234567890abcdef";
        let token = "my-secret-token";
        let scope = "share";
        let max_conn = 3;

        let compound = build_compound_ticket(iroh_ticket, token, scope, max_conn);

        // Compound ticket contains exactly one dot
        assert_eq!(compound.matches('.').count(), 1);

        // Split recovers the iroh ticket and decodes the token payload
        let (recovered_ticket, suffix) = split_compound_ticket(&compound);
        assert_eq!(recovered_ticket, iroh_ticket);

        let payload = decode_token_payload(suffix.unwrap()).unwrap();
        assert_eq!(payload.token, token);
        assert_eq!(payload.scope, scope);
        assert_eq!(payload.max_connections, max_conn);
        assert_eq!(
            payload.ticket_hash.as_deref(),
            Some(endpoint_ticket_hash(iroh_ticket).as_str())
        );
        assert_eq!(
            payload.audience.as_deref(),
            Some(TOKEN_PAYLOAD_ENDPOINT_TICKET_AUDIENCE)
        );
        assert!(payload.nonce.as_deref().is_some_and(is_valid_payload_nonce));
        assert!(token_payload_matches_ticket(iroh_ticket, &payload));
        assert!(
            decode_token_payload_for_ticket("different-endpoint-ticket", suffix.unwrap()).is_none()
        );
    }

    #[test]
    fn test_bound_compound_ticket_rejects_wrong_audience_and_bad_nonce() {
        let iroh_ticket = "endpoint-ticket";
        let encoded = encode_token_payload_for_ticket(iroh_ticket, "token", "share", 1);
        let mut payload = decode_token_payload(&encoded).unwrap();

        payload.audience = Some("openrtc:other-audience:v1".to_string());
        assert!(!token_payload_matches_ticket(iroh_ticket, &payload));

        payload.audience = Some(TOKEN_PAYLOAD_ENDPOINT_TICKET_AUDIENCE.to_string());
        payload.nonce = Some("not-valid-base64url-nonce".to_string());
        assert!(!token_payload_matches_ticket(iroh_ticket, &payload));
    }

    #[test]
    fn test_bound_compound_ticket_nonces_are_unique() {
        let first = build_compound_ticket("endpoint-ticket", "token-one", "share", 1);
        let second = build_compound_ticket("endpoint-ticket", "token-two", "share", 1);
        let (_, first_suffix) = split_compound_ticket(&first);
        let (_, second_suffix) = split_compound_ticket(&second);
        let first_payload = decode_token_payload(first_suffix.unwrap()).unwrap();
        let second_payload = decode_token_payload(second_suffix.unwrap()).unwrap();

        assert_ne!(first_payload.nonce, second_payload.nonce);
        assert!(first_payload
            .nonce
            .as_deref()
            .is_some_and(is_valid_payload_nonce));
        assert!(second_payload
            .nonce
            .as_deref()
            .is_some_and(is_valid_payload_nonce));
    }

    #[test]
    fn test_single_use_payload_nonce_replay_rejected_across_connections() {
        // A single-use share ticket (max_connections == 1) binds its nonce to the
        // first admitting connection; a different connection presenting the same
        // out-of-band nonce is a replay and must be rejected.
        let registry = SessionTokenRegistry::new();
        registry.register("tok".to_string(), "share".to_string(), 1);
        let suffix = encode_token_payload_for_ticket("endpoint-ticket", "tok", "share", 1);

        assert_eq!(
            registry
                .validate_and_consume_for_connection_with_payload(
                    "tok",
                    Some("conn-A"),
                    Some(&suffix),
                )
                .unwrap(),
            "share"
        );
        // Idempotent re-presentation on the SAME connection still succeeds.
        assert_eq!(
            registry
                .validate_and_consume_for_connection_with_payload(
                    "tok",
                    Some("conn-A"),
                    Some(&suffix),
                )
                .unwrap(),
            "share"
        );
        assert_eq!(
            registry
                .validate_and_consume_for_connection_with_payload(
                    "tok",
                    Some("conn-B"),
                    Some(&suffix),
                )
                .unwrap_err(),
            "session token payload nonce replayed"
        );
    }

    #[test]
    fn test_multi_connection_payload_nonce_allowed_across_connections() {
        // A multi-use / unlimited presence ticket (max_connections == 0, e.g.
        // `user-device`) is advertised once and presented by every dialing peer,
        // each forwarding the SAME signed nonce. This must NOT be treated as a
        // replay — otherwise the second device in a same-user mesh is falsely
        // rejected and the connection churns. Regression test for the
        // browser↔native mesh stranding at session-admission-pending.
        let registry = SessionTokenRegistry::new();
        registry.register("tok".to_string(), "user-device".to_string(), 0);
        let suffix = encode_token_payload_for_ticket("endpoint-ticket", "tok", "user-device", 0);

        assert_eq!(
            registry
                .validate_and_consume_for_connection_with_payload(
                    "tok",
                    Some("conn-A"),
                    Some(&suffix),
                )
                .unwrap(),
            "user-device"
        );
        // A different connection presenting the same advertised nonce is a
        // legitimate second mesh peer, not a replay attacker.
        assert_eq!(
            registry
                .validate_and_consume_for_connection_with_payload(
                    "tok",
                    Some("conn-B"),
                    Some(&suffix),
                )
                .unwrap(),
            "user-device"
        );
        assert_eq!(
            registry
                .validate_and_consume_for_connection_with_payload(
                    "tok",
                    Some("conn-C"),
                    Some(&suffix),
                )
                .unwrap(),
            "user-device"
        );
    }

    #[test]
    fn test_payload_metadata_must_match_presented_token() {
        let registry = SessionTokenRegistry::new();
        registry.register("real-token".to_string(), "share".to_string(), 0);
        let suffix = encode_token_payload_for_ticket("endpoint-ticket", "other-token", "share", 0);

        assert_eq!(
            registry
                .validate_and_consume_for_connection_with_payload(
                    "real-token",
                    Some("conn-A"),
                    Some(&suffix),
                )
                .unwrap_err(),
            "session token payload mismatch"
        );
    }

    #[test]
    fn test_payload_grant_must_match_registered_capability() {
        let registry = SessionTokenRegistry::new();
        registry.register("real-token".to_string(), "share:read".to_string(), 1);
        let suffix =
            encode_token_payload_for_ticket("endpoint-ticket", "real-token", "share:write", 2);

        assert_eq!(
            registry
                .validate_and_consume_for_connection_with_payload(
                    "real-token",
                    Some("conn-A"),
                    Some(&suffix),
                )
                .unwrap_err(),
            "session token payload grant mismatch"
        );
    }

    #[test]
    fn test_bound_compound_ticket_rejects_suffix_graft() {
        let first_ticket = "endpoint-ticket-one";
        let second_ticket = "endpoint-ticket-two";
        let first_compound = build_compound_ticket(first_ticket, "token-one", "share", 1);
        let second_compound = build_compound_ticket(second_ticket, "token-two", "share", 1);

        let (_, first_suffix) = split_compound_ticket(&first_compound);
        let grafted = format!("{}.{}", second_ticket, first_suffix.unwrap());
        let (grafted_ticket, grafted_suffix) = split_compound_ticket(&grafted);

        assert_eq!(grafted_ticket, second_ticket);
        assert!(
            decode_token_payload_for_ticket(grafted_ticket, grafted_suffix.unwrap()).is_none(),
            "a suffix minted for one endpoint ticket must not validate on another endpoint ticket"
        );

        let (_, second_suffix) = split_compound_ticket(&second_compound);
        let payload =
            decode_token_payload_for_ticket(second_ticket, second_suffix.unwrap()).unwrap();
        assert_eq!(payload.token, "token-two");
    }

    #[test]
    fn test_compound_ticket_payload_carries_expiry() {
        let expires_at_ms = now_unix_ms().saturating_add(DEFAULT_RESTRICTED_SESSION_TOKEN_TTL_MS);
        let compound = build_compound_ticket_with_expiry(
            "endpoint-ticket",
            "token-with-expiry",
            "share",
            1,
            Some(expires_at_ms),
        );
        let (iroh_ticket, suffix) = split_compound_ticket(&compound);
        let payload = decode_token_payload_for_ticket(iroh_ticket, suffix.unwrap()).unwrap();

        assert_eq!(payload.token, "token-with-expiry");
        assert_eq!(payload.expires_at_ms, Some(expires_at_ms));
        assert_eq!(
            payload.audience.as_deref(),
            Some(TOKEN_PAYLOAD_ENDPOINT_TICKET_AUDIENCE)
        );
        assert!(payload.nonce.as_deref().is_some_and(is_valid_payload_nonce));
    }

    #[test]
    fn test_build_compound_ticket_then_registry_validate() {
        // Simulates: host generates compound ticket → guest extracts token → host validates
        let registry = SessionTokenRegistry::new();
        let token = generate_token();
        let scope = "share";
        registry.register(token.clone(), scope.to_string(), 1);

        let compound = build_compound_ticket("irohticketbase32", &token, scope, 1);
        let (_iroh, suffix) = split_compound_ticket(&compound);
        let payload = decode_token_payload(suffix.unwrap()).unwrap();
        assert!(token_payload_matches_ticket("irohticketbase32", &payload));

        // Validate the extracted token against the registry
        let result = registry.validate_and_consume(&payload.token);
        assert!(result.is_ok());
        assert_eq!(result.unwrap(), "share");

        // Second use should fail (max_connections = 1)
        let result2 = registry.validate_and_consume(&payload.token);
        assert!(result2.is_err());
    }

    #[test]
    fn test_revoke_single_token() {
        let registry = SessionTokenRegistry::new();
        registry.register("tok-a".to_string(), "share".to_string(), 0);
        registry.register("tok-b".to_string(), "share".to_string(), 0);

        registry
            .validate_and_consume_for_connection("tok-a", Some("conn-a"))
            .expect("initial admission");
        assert_eq!(registry.revoke("tok-a"), vec!["conn-a"]);

        // tok-a gone, tok-b still works
        assert!(registry.validate_and_consume("tok-a").is_err());
        assert!(registry.validate_and_consume("tok-b").is_ok());
        assert!(matches!(
            registry.admission("conn-a"),
            SessionAdmission::Rejected { ref reason } if reason == "session token revoked"
        ));
    }

    #[test]
    fn test_registry_clear() {
        let registry = SessionTokenRegistry::new();
        registry.register("tok-x".to_string(), "share".to_string(), 0);
        assert!(!registry.is_empty());

        registry.clear();
        assert!(registry.is_empty());

        // Clearing capabilities must not make a previously protected path
        // accept arbitrary bearer strings.
        assert!(registry.validate_and_consume("").is_ok());
        assert!(registry.validate_and_consume("anything").is_err());
    }

    #[test]
    fn test_ts_encoded_payload_is_decodable() {
        // This payload was generated by the TS test helpers — ensures cross-language compat
        // TS: JSON.stringify({t:"test-token",s:"share",m:0}) → base64url
        let json = r#"{"t":"test-token","s":"share","m":0}"#;
        let encoded = URL_SAFE_NO_PAD.encode(json.as_bytes());
        let payload = decode_token_payload(&encoded).unwrap();
        assert_eq!(payload.token, "test-token");
        assert_eq!(payload.scope, "share");
        assert_eq!(payload.max_connections, 0);
    }

    #[test]
    fn test_generated_token_uniqueness() {
        let tokens: Vec<String> = (0..100).map(|_| generate_token()).collect();
        let unique: std::collections::HashSet<&String> = tokens.iter().collect();
        assert_eq!(unique.len(), 100, "generated tokens must be unique");
    }

    #[test]
    fn test_revoke_by_scope_returns_affected_connections() {
        let registry = SessionTokenRegistry::new();
        registry.register("tok-share-1".to_string(), "share".to_string(), 0);
        registry.register("tok-share-2".to_string(), "share".to_string(), 0);
        registry.register("tok-other".to_string(), "other".to_string(), 0);

        // Validate with connection IDs
        registry
            .validate_and_consume_for_connection("tok-share-1", Some("conn-A"))
            .unwrap();
        registry
            .validate_and_consume_for_connection("tok-share-2", Some("conn-B"))
            .unwrap();
        registry
            .validate_and_consume_for_connection("tok-other", Some("conn-C"))
            .unwrap();

        // Revoke "share" scope — should return conn-A and conn-B
        let mut affected = registry.revoke_by_scope("share");
        affected.sort();
        assert_eq!(affected, vec!["conn-A", "conn-B"]);

        // conn-C should still be tracked
        let affected2 = registry.revoke_by_scope("other");
        assert_eq!(affected2, vec!["conn-C"]);
    }

    #[test]
    fn test_revoke_scoped_guest_scope_isolates_grants() {
        let registry = SessionTokenRegistry::new();
        registry.register("tok-a".to_string(), "scoped:grant-a".to_string(), 0);
        registry.register("tok-b".to_string(), "scoped:grant-b".to_string(), 0);
        registry
            .validate_and_consume_for_connection("tok-a", Some("conn-a"))
            .unwrap();
        registry
            .validate_and_consume_for_connection("tok-b", Some("conn-b"))
            .unwrap();
        let affected = registry.revoke_by_scope("scoped:grant-a");
        assert_eq!(affected, vec!["conn-a"]);
        assert!(registry.validate_and_consume("tok-b").is_ok());
        assert!(registry.validate_and_consume("tok-a").is_err());
    }

    #[test]
    fn test_validate_for_connection_rejects_without_token() {
        let registry = SessionTokenRegistry::new();
        registry.register("real-tok".to_string(), "share".to_string(), 0);

        // No token → rejected
        let result = registry.validate_and_consume_for_connection("", Some("conn-X"));
        assert!(result.is_err());

        // Wrong token → rejected
        let result = registry.validate_and_consume_for_connection("wrong-tok", Some("conn-Y"));
        assert!(result.is_err());

        // Correct token → accepted
        let result = registry.validate_and_consume_for_connection("real-tok", Some("conn-Z"));
        assert!(result.is_ok());
        assert_eq!(result.unwrap(), "share");
    }

    #[test]
    fn test_connection_admission_tracks_token_acceptance_and_revoke() {
        let registry = SessionTokenRegistry::new();
        registry.register("real-tok".to_string(), "share".to_string(), 0);

        assert_eq!(registry.admission("conn-1"), SessionAdmission::Pending);

        registry
            .validate_and_consume_for_connection("real-tok", Some("conn-1"))
            .unwrap();

        assert_eq!(
            registry.admission("conn-1"),
            SessionAdmission::Accepted {
                mechanism: SessionAdmissionMechanism::SessionToken,
                scope: Some(GrantScope::from("share")),
                authoritative_device_id: None,
            }
        );

        assert_eq!(registry.revoke_by_scope("share"), vec!["conn-1"]);
        assert_eq!(registry.admission("conn-1"), SessionAdmission::Pending);
    }

    #[test]
    fn is_session_token_admitted_for_connection_tracks_first_vs_duplicate_presentation() {
        let registry = SessionTokenRegistry::new();
        registry.register("tok".to_string(), "share".to_string(), 0);

        assert!(
            !registry.is_session_token_admitted_for_connection("conn-dup"),
            "before first consume, not yet session-token admitted"
        );

        registry
            .validate_and_consume_for_connection("tok", Some("conn-dup"))
            .unwrap();

        assert!(
            registry.is_session_token_admitted_for_connection("conn-dup"),
            "after first consume, duplicate token frames must be detectable for WebRTC gating"
        );

        registry.forget_connection("conn-dup");
        assert!(
            !registry.is_session_token_admitted_for_connection("conn-dup"),
            "forget clears admission so a reconnect can run first-presentation recovery again"
        );
    }

    #[test]
    fn outbound_host_approval_does_not_impersonate_inbound_token_admission() {
        let registry = SessionTokenRegistry::new();
        registry.register("tok".to_string(), "user-device".to_string(), 0);

        // `bind_connection_scope` is used after the remote host approves this
        // runtime's outbound token. It may update the compatibility projection,
        // but it must not claim that this runtime validated the peer's token.
        registry.bind_connection_scope("conn-bilateral", "user-device", None);

        assert!(matches!(
            registry.admission("conn-bilateral"),
            SessionAdmission::Accepted {
                mechanism: SessionAdmissionMechanism::SessionToken,
                ..
            }
        ));
        assert!(
            !registry.is_session_token_admitted_for_connection("conn-bilateral"),
            "outbound approval and inbound token validation are independent"
        );
        assert_eq!(
            registry.revoke_by_scope("user-device"),
            vec!["conn-bilateral".to_string()],
            "logical scope revocation still owns the remote-host-approved direction",
        );
        assert!(matches!(
            registry.admission("conn-bilateral"),
            SessionAdmission::Pending,
        ));
    }

    #[test]
    fn test_native_trusted_verifier_can_accept_authoritative_device() {
        let registry = SessionTokenRegistry::new();
        registry.register("sentinel".to_string(), "share".to_string(), 0);
        registry.set_native_trusted_connection_verifier(Some(Arc::new(|context| {
            if context.remote_node_id.as_deref() == Some("node-123")
                && context.claimed_device_id.as_deref() == Some("device-abc")
            {
                Some("device-abc".to_string())
            } else {
                None
            }
        })));

        let context = NativeTrustedConnectionContext {
            connection_id: "conn-native".to_string(),
            remote_node_id: Some("node-123".to_string()),
            known_device_id: None,
            claimed_device_id: Some("device-abc".to_string()),
        };

        let authoritative = registry.evaluate_trusted_native_connection(&context);
        assert_eq!(authoritative.as_deref(), Some("device-abc"));

        registry.mark_accepted(
            "conn-native",
            SessionAdmissionMechanism::TrustedNativeBinding,
            None,
            authoritative,
        );

        assert_eq!(
            registry.admission("conn-native"),
            SessionAdmission::Accepted {
                mechanism: SessionAdmissionMechanism::TrustedNativeBinding,
                scope: None,
                authoritative_device_id: Some("device-abc".to_string()),
            }
        );
    }

    // -----------------------------------------------------------------------
    // End-to-end scenario tests: native host ↔ web client token gating
    // -----------------------------------------------------------------------

    /// Simulates: native host generates compound ticket, web client extracts
    /// token, host validates token on incoming handshake. Verifies the full
    /// round-trip from ticket generation through validation.
    #[test]
    fn test_native_host_web_client_full_flow() {
        let registry = SessionTokenRegistry::new();

        // Host generates a compound ticket (native side)
        let host_token = generate_token();
        registry.register(host_token.clone(), "share".to_string(), 0);
        let compound = build_compound_ticket("nativeticketbase32", &host_token, "share", 0);

        // Client extracts the token from the compound ticket (web side)
        let (_iroh_ticket, suffix) = split_compound_ticket(&compound);
        let payload = decode_token_payload(suffix.unwrap()).unwrap();

        // Host validates the extracted token for a specific connection
        let result =
            registry.validate_and_consume_for_connection(&payload.token, Some("web-client-conn-1"));
        assert!(result.is_ok());
        assert_eq!(result.unwrap(), "share");
    }

    /// Client without a token is rejected when registry is non-empty.
    #[test]
    fn test_client_without_token_rejected() {
        let registry = SessionTokenRegistry::new();
        registry.register(generate_token(), "share".to_string(), 0);

        // Client arrives without a token
        let result = registry.validate_and_consume_for_connection("", Some("bad-conn"));
        assert!(result.is_err());
        assert!(result.unwrap_err().contains("unknown"));
    }

    /// Client with wrong token is rejected.
    #[test]
    fn test_client_with_wrong_token_rejected() {
        let registry = SessionTokenRegistry::new();
        let real_token = generate_token();
        registry.register(real_token, "share".to_string(), 0);

        let result = registry
            .validate_and_consume_for_connection("completely-wrong-token", Some("bad-conn"));
        assert!(result.is_err());
    }

    /// Revoking a scope disconnects all connections that used tokens of that
    /// scope and prevents new connections from using those tokens.
    #[test]
    fn test_revoke_scope_disconnects_and_blocks() {
        let registry = SessionTokenRegistry::new();
        let tok1 = generate_token();
        let tok2 = generate_token();
        registry.register(tok1.clone(), "share".to_string(), 0);
        registry.register(tok2.clone(), "share".to_string(), 0);
        // Keep a sentinel token to prove scope-selective revocation leaves
        // unrelated capabilities intact.
        registry.register("sentinel".to_string(), "sentinel".to_string(), 0);

        // Two clients connect with valid tokens
        assert!(registry
            .validate_and_consume_for_connection(&tok1, Some("conn-web-1"))
            .is_ok());
        assert!(registry
            .validate_and_consume_for_connection(&tok2, Some("conn-web-2"))
            .is_ok());

        // Host revokes "share" scope
        let mut affected = registry.revoke_by_scope("share");
        affected.sort();
        assert_eq!(affected, vec!["conn-web-1", "conn-web-2"]);

        // New connection attempt with tok1 fails (token revoked)
        let result = registry.validate_and_consume_for_connection(&tok1, Some("conn-web-3"));
        assert!(result.is_err());

        // New connection attempt with tok2 also fails
        let result = registry.validate_and_consume_for_connection(&tok2, Some("conn-web-4"));
        assert!(result.is_err());
    }

    /// Mixed scopes: revoking "share" does not affect "other" connections.
    #[test]
    fn test_revoke_scope_only_affects_matching_scope() {
        let registry = SessionTokenRegistry::new();
        let share_tok = generate_token();
        let other_tok = generate_token();
        registry.register(share_tok.clone(), "share".to_string(), 0);
        registry.register(other_tok.clone(), "other".to_string(), 0);

        assert!(registry
            .validate_and_consume_for_connection(&share_tok, Some("share-conn"))
            .is_ok());
        assert!(registry
            .validate_and_consume_for_connection(&other_tok, Some("other-conn"))
            .is_ok());

        let affected = registry.revoke_by_scope("share");
        assert_eq!(affected, vec!["share-conn"]);

        // "other" token still works for new connections
        let new_other_tok = generate_token();
        registry.register(new_other_tok.clone(), "other".to_string(), 0);
        assert!(registry
            .validate_and_consume_for_connection(&new_other_tok, Some("other-conn-2"))
            .is_ok());
    }

    /// maxConnections enforcement: token with max=1 rejects second use.
    #[test]
    fn test_max_connections_enforced_per_token() {
        let registry = SessionTokenRegistry::new();
        let token = generate_token();
        registry.register(token.clone(), "share".to_string(), 1);

        // First use succeeds
        assert!(registry
            .validate_and_consume_for_connection(&token, Some("conn-1"))
            .is_ok());

        // Second use is rejected (exhausted)
        let result = registry.validate_and_consume_for_connection(&token, Some("conn-2"));
        assert!(result.is_err());
        assert!(result.unwrap_err().contains("exhausted"));
    }

    #[test]
    fn max_connections_is_atomic_under_concurrent_presentations() {
        let registry = SessionTokenRegistry::new();
        let token = generate_token();
        registry.register(token.clone(), "share".to_string(), 1);
        let start = Arc::new(std::sync::Barrier::new(16));

        let handles = (0..16)
            .map(|index| {
                let registry = registry.clone();
                let token = token.clone();
                let start = start.clone();
                std::thread::spawn(move || {
                    start.wait();
                    registry
                        .validate_and_consume_for_connection(
                            &token,
                            Some(&format!("concurrent-{index}")),
                        )
                        .is_ok()
                })
            })
            .collect::<Vec<_>>();
        let accepted = handles
            .into_iter()
            .map(|handle| handle.join().expect("validation thread"))
            .filter(|accepted| *accepted)
            .count();

        assert_eq!(
            accepted, 1,
            "single-use capability must admit exactly one connection"
        );
    }

    #[test]
    fn empty_token_or_scope_never_activates_a_grant() {
        let registry = SessionTokenRegistry::new();
        registry.register(String::new(), "user-device".to_string(), 0);
        registry.register("token".to_string(), String::new(), 0);

        assert!(registry.is_empty());
        assert!(registry.validate_and_consume("").is_ok());
        assert_eq!(
            registry.validate_and_consume("token").unwrap_err(),
            "unknown session token"
        );
    }

    /// Backward-compat: empty registry accepts tokenless manual connections,
    /// but never claims that an arbitrary bearer token was validated.
    #[test]
    fn test_empty_registry_accepts_only_tokenless_manual_connections() {
        let registry = SessionTokenRegistry::new();

        assert!(registry
            .validate_and_consume_for_connection("any-token", Some("conn-1"))
            .is_err());
        assert!(registry
            .validate_and_consume_for_connection("", Some("conn-2"))
            .is_ok());
    }

    /// Connection-level caching: once a connection_id is accepted with a token,
    /// repeat validations for that connection should not re-consume the token.
    #[test]
    fn test_validate_for_connection_is_idempotent_after_accept() {
        let registry = SessionTokenRegistry::new();
        let token = generate_token();
        registry.register(token.clone(), "share".to_string(), 1);

        // First validation consumes the single allowed use.
        assert!(registry
            .validate_and_consume_for_connection(&token, Some("conn-1"))
            .is_ok());

        // Subsequent validation for the same connection should be accepted
        // from cached admission, not rejected as exhausted.
        assert!(registry
            .validate_and_consume_for_connection(&token, Some("conn-1"))
            .is_ok());

        // A second connection should still be rejected (max_connections=1).
        let result = registry.validate_and_consume_for_connection(&token, Some("conn-2"));
        assert!(result.is_err());
        assert!(result.unwrap_err().contains("exhausted"));
    }

    /// Web host ↔ native client: host creates token in WASM registry,
    /// native client extracts from compound ticket and presents for validation.
    #[test]
    fn test_web_host_native_client_flow() {
        let registry = SessionTokenRegistry::new();

        // Web host generates compound ticket
        let host_token = generate_token();
        registry.register(host_token.clone(), "share".to_string(), 0);
        let compound = build_compound_ticket("wasmticketbase32", &host_token, "share", 0);

        // Native client parses compound ticket
        let (iroh_ticket, suffix) = split_compound_ticket(&compound);
        assert_eq!(iroh_ticket, "wasmticketbase32");
        let payload = decode_token_payload(suffix.unwrap()).unwrap();
        assert_eq!(payload.scope, "share");

        // Native client presents token → host validates
        let result = registry
            .validate_and_consume_for_connection(&payload.token, Some("native-client-conn"));
        assert!(result.is_ok());

        // Keep a sentinel so registry stays non-empty after revocation
        registry.register("sentinel".to_string(), "sentinel".to_string(), 0);

        // After revoke, same token fails
        registry.revoke_by_scope("share");
        let new_attempt = registry
            .validate_and_consume_for_connection(&payload.token, Some("native-client-conn-2"));
        assert!(new_attempt.is_err());
    }

    // -----------------------------------------------------------------------
    // Edge case coverage
    // -----------------------------------------------------------------------

    #[test]
    fn test_grant_scope_serde_roundtrip() {
        let scope = GrantScope::new("share");
        let json = serde_json::to_string(&scope).unwrap();
        assert_eq!(json, r#""share""#); // transparent serialization
        let deserialized: GrantScope = serde_json::from_str(&json).unwrap();
        assert_eq!(deserialized, "share");
    }

    #[test]
    fn test_grant_scope_equality() {
        let scope = GrantScope::new("share");
        assert_eq!(scope, "share");
        assert_eq!(scope, "share".to_string());
        assert_eq!(scope, GrantScope::from("share"));
        assert_ne!(scope, "other");
    }

    #[test]
    fn test_forget_connection_cleans_up() {
        let registry = SessionTokenRegistry::new();
        registry.register("tok".to_string(), "share".to_string(), 0);

        registry
            .validate_and_consume_for_connection("tok", Some("conn-forget"))
            .unwrap();
        assert_eq!(
            registry.admission("conn-forget"),
            SessionAdmission::Accepted {
                mechanism: SessionAdmissionMechanism::SessionToken,
                scope: Some(GrantScope::from("share")),
                authoritative_device_id: None,
            }
        );

        registry.forget_connection("conn-forget");
        assert_eq!(registry.admission("conn-forget"), SessionAdmission::Pending);
    }

    #[test]
    fn test_forget_connection_does_not_clear_payload_nonce_replay_memory() {
        // Single-use ticket: the nonce binding must survive forget_connection so a
        // forgotten connection cannot be used to replay the one-shot token.
        let registry = SessionTokenRegistry::new();
        registry.register("tok".to_string(), "share".to_string(), 1);
        let suffix = encode_token_payload_for_ticket("endpoint-ticket", "tok", "share", 1);

        registry
            .validate_and_consume_for_connection_with_payload("tok", Some("conn-A"), Some(&suffix))
            .expect("first presentation");
        registry.forget_connection("conn-A");

        assert_eq!(
            registry
                .validate_and_consume_for_connection_with_payload(
                    "tok",
                    Some("conn-B"),
                    Some(&suffix),
                )
                .unwrap_err(),
            "session token payload nonce replayed"
        );
    }

    #[test]
    fn test_mark_rejected_and_readback() {
        let registry = SessionTokenRegistry::new();

        registry.mark_rejected("conn-bad", "invalid token");
        assert_eq!(
            registry.admission("conn-bad"),
            SessionAdmission::Rejected {
                reason: "invalid token".to_string(),
            }
        );
    }

    #[test]
    fn test_trusted_verifier_returns_none_does_not_admit() {
        let registry = SessionTokenRegistry::new();
        registry.register("sentinel".to_string(), "share".to_string(), 0);
        registry.set_native_trusted_connection_verifier(Some(Arc::new(|_context| {
            None // Always reject
        })));

        let context = NativeTrustedConnectionContext {
            connection_id: "conn-untrusted".to_string(),
            remote_node_id: Some("unknown-node".to_string()),
            known_device_id: None,
            claimed_device_id: Some("fake-device".to_string()),
        };

        let result = registry.evaluate_trusted_native_connection(&context);
        assert!(result.is_none());
        // Admission should still be Pending — verifier returning None doesn't mark rejected
        assert_eq!(
            registry.admission("conn-untrusted"),
            SessionAdmission::Pending
        );
    }

    #[test]
    fn test_validate_without_connection_id_does_not_track_admission() {
        let registry = SessionTokenRegistry::new();
        registry.register("tok".to_string(), "share".to_string(), 0);

        // Validate without connection_id (simple path)
        let result = registry.validate_and_consume("tok");
        assert!(result.is_ok());

        // No admission should be tracked since no connection_id was given
        // (there's nothing to look up)
    }

    #[test]
    fn test_revoke_by_scope_with_no_matching_connections() {
        let registry = SessionTokenRegistry::new();
        registry.register("tok-a".to_string(), "share".to_string(), 0);
        registry.register("tok-b".to_string(), "other".to_string(), 0);

        // Validate tok-b but revoke "share" — no connections should be affected
        registry
            .validate_and_consume_for_connection("tok-b", Some("conn-other"))
            .unwrap();

        let affected = registry.revoke_by_scope("share");
        assert!(affected.is_empty());

        // conn-other should still be admitted
        assert_eq!(
            registry.admission("conn-other"),
            SessionAdmission::Accepted {
                mechanism: SessionAdmissionMechanism::SessionToken,
                scope: Some(GrantScope::from("other")),
                authoritative_device_id: None,
            }
        );
    }

    #[test]
    fn test_clear_removes_all_admission_state() {
        let registry = SessionTokenRegistry::new();
        registry.register("tok".to_string(), "share".to_string(), 0);
        registry
            .validate_and_consume_for_connection("tok", Some("conn-1"))
            .unwrap();
        registry.mark_rejected("conn-2", "bad");

        assert!(matches!(
            registry.admission("conn-1"),
            SessionAdmission::Accepted { .. }
        ));
        assert!(matches!(
            registry.admission("conn-2"),
            SessionAdmission::Rejected { .. }
        ));

        registry.clear();

        assert_eq!(registry.admission("conn-1"), SessionAdmission::Pending);
        assert_eq!(registry.admission("conn-2"), SessionAdmission::Pending);
        assert!(registry.is_empty());
    }

    // -----------------------------------------------------------------------
    // Phase 1 (idempotent admission + flush-before-side-effects) coverage.
    // -----------------------------------------------------------------------

    /// Phase 1: presenting the *same* token on the *same* connection_id twice
    /// must be idempotent — second call returns the cached verdict, never
    /// re-consumes the token, and never errors with "exhausted".
    #[test]
    fn validate_with_cached_response_is_idempotent_for_same_token() {
        let registry = SessionTokenRegistry::new();
        registry.register("tok".to_string(), "scoped:abc".to_string(), 1);

        let first = registry
            .validate_with_cached_response("tok", "conn-A")
            .expect("first admission");
        assert_eq!(first, "scoped:abc");

        let second = registry
            .validate_with_cached_response("tok", "conn-A")
            .expect("replay must succeed deterministically");
        assert_eq!(second, "scoped:abc");
        assert_eq!(first, second);
    }

    /// Phase 1: same token presented on *different* connection_ids consumes
    /// the token's max_connections budget per connection (not idempotent
    /// across connections).
    #[test]
    fn validate_with_cached_response_is_not_idempotent_across_connection_ids() {
        let registry = SessionTokenRegistry::new();
        registry.register("tok".to_string(), "scoped:abc".to_string(), 1);

        registry
            .validate_with_cached_response("tok", "conn-A")
            .expect("first connection accepted");

        // max_connections=1 → second connection must be rejected.
        let err = registry
            .validate_with_cached_response("tok", "conn-B")
            .expect_err("second connection rejected");
        assert!(err.contains("exhausted"), "expected exhausted, got: {err}");
    }

    /// Phase 1: presenting a *different* token on a connection that is
    /// already admitted must NOT auto-accept from cache. A rotated/forged
    /// token must be re-validated.
    #[test]
    fn validate_with_cached_response_rejects_different_token_on_admitted_connection() {
        let registry = SessionTokenRegistry::new();
        registry.register("real-tok".to_string(), "scoped:abc".to_string(), 0);

        registry
            .validate_with_cached_response("real-tok", "conn-X")
            .expect("first admission");

        let err = registry
            .validate_with_cached_response("forged-tok", "conn-X")
            .expect_err("rotated/forged token must be re-validated");
        assert!(
            err.contains("unknown"),
            "expected 'unknown session token', got: {err}",
        );
    }

    /// Phase 1: token fingerprints are deterministic, hex-encoded, low
    /// cardinality, and *not* equal to the raw token.
    #[test]
    fn token_fingerprint_is_deterministic_and_short() {
        let fp1 = token_fingerprint("hello-world");
        let fp2 = token_fingerprint("hello-world");
        assert_eq!(fp1, fp2);
        assert_eq!(fp1.len(), 16);
        assert_ne!(fp1, "hello-world");
        assert!(fp1.chars().all(|c| c.is_ascii_hexdigit()));
        assert_ne!(fp1, token_fingerprint("hello-worldX"));
    }

    /// Phase 1: in-flight admission counter is correct under interleaved
    /// guard scopes (matches the shape used by the desktop response writer).
    #[test]
    fn in_flight_admission_guard_tracks_concurrent_writers() {
        let registry = SessionTokenRegistry::new();

        assert!(!registry.is_session_token_admission_in_flight("conn-1"));

        let g1 = registry.begin_admission_response("conn-1");
        assert!(registry.is_session_token_admission_in_flight("conn-1"));

        // Nested guard for the same connection (e.g. two concurrent
        // session-token streams on a single transport): both must be alive
        // before the connection is "quiet" again.
        let g2 = registry.begin_admission_response("conn-1");
        assert!(registry.is_session_token_admission_in_flight("conn-1"));

        drop(g1);
        assert!(
            registry.is_session_token_admission_in_flight("conn-1"),
            "still in-flight while second guard is alive",
        );
        drop(g2);
        assert!(!registry.is_session_token_admission_in_flight("conn-1"));
    }

    /// Phase 1: in-flight tracking is per-connection (a writer on conn-1 must
    /// not block lifecycle action on conn-2).
    #[test]
    fn in_flight_admission_guard_is_per_connection() {
        let registry = SessionTokenRegistry::new();

        let _g = registry.begin_admission_response("conn-A");
        assert!(registry.is_session_token_admission_in_flight("conn-A"));
        assert!(!registry.is_session_token_admission_in_flight("conn-B"));
    }

    #[test]
    fn admission_response_and_terminal_retirement_are_mutually_exclusive() {
        let registry = SessionTokenRegistry::new();

        let response = registry
            .try_begin_admission_response("conn-gated")
            .expect("response should acquire an idle connection");
        assert!(registry
            .try_begin_admission_retirement("conn-gated")
            .is_none());
        drop(response);

        let retirement = registry
            .try_begin_admission_retirement("conn-gated")
            .expect("retirement should acquire after response flush");
        assert!(registry
            .try_begin_admission_response("conn-gated")
            .is_none());
        drop(retirement);

        assert!(registry
            .try_begin_admission_response("conn-gated")
            .is_some());
    }

    #[test]
    fn outbound_admission_presentation_is_single_flight_and_blocks_retirement() {
        let registry = SessionTokenRegistry::new();

        let presentation = registry
            .try_begin_admission_presentation("conn-present")
            .expect("first presenter should acquire the connection");
        assert!(registry.is_session_token_presentation_in_flight("conn-present"));
        assert!(registry.is_session_token_admission_in_flight("conn-present"));
        assert!(registry
            .try_begin_admission_presentation("conn-present")
            .is_none());
        assert!(registry
            .try_begin_admission_retirement("conn-present")
            .is_none());

        drop(presentation);
        assert!(!registry.is_session_token_presentation_in_flight("conn-present"));
        assert!(registry
            .try_begin_admission_presentation("conn-present")
            .is_some());
    }

    #[test]
    fn admission_response_and_outbound_presentation_preserve_each_others_ownership() {
        let registry = SessionTokenRegistry::new();

        let response = registry
            .try_begin_admission_response("conn-bilateral")
            .expect("response writer should acquire an idle connection");
        let presentation = registry
            .try_begin_admission_presentation("conn-bilateral")
            .expect("opposite-direction presenter may coexist");

        drop(response);
        assert!(registry.is_session_token_presentation_in_flight("conn-bilateral"));
        assert!(registry.is_session_token_admission_in_flight("conn-bilateral"));
        assert!(registry
            .try_begin_admission_retirement("conn-bilateral")
            .is_none());

        let response = registry
            .try_begin_admission_response("conn-bilateral")
            .expect("response writer can reacquire while presentation remains active");
        drop(presentation);
        assert!(!registry.is_session_token_presentation_in_flight("conn-bilateral"));
        assert!(registry.is_session_token_admission_in_flight("conn-bilateral"));
        assert!(registry
            .try_begin_admission_retirement("conn-bilateral")
            .is_none());

        drop(response);
        assert!(!registry.is_session_token_admission_in_flight("conn-bilateral"));
        assert!(registry
            .try_begin_admission_retirement("conn-bilateral")
            .is_some());
    }

    /// Phase 1: forget_connection clears all per-connection state including
    /// fingerprints + in-flight counters.
    #[test]
    fn forget_connection_clears_phase1_state() {
        let registry = SessionTokenRegistry::new();
        registry.register("tok".to_string(), "scoped:abc".to_string(), 0);

        registry
            .validate_with_cached_response("tok", "conn-A")
            .expect("admit");
        assert!(registry.admission_fingerprint("conn-A").is_some());

        let _g = registry.begin_admission_response("conn-A");
        assert!(registry.is_session_token_admission_in_flight("conn-A"));
        drop(_g);

        registry.forget_connection("conn-A");
        assert_eq!(registry.admission("conn-A"), SessionAdmission::Pending);
        assert!(registry.admission_fingerprint("conn-A").is_none());
        assert!(!registry.is_session_token_admission_in_flight("conn-A"));
    }

    #[test]
    fn test_token_payload_with_grant_scope_cross_language_compat() {
        // Verify that GrantScope serializes transparently in token payloads,
        // matching the flat string format that TypeScript expects.
        let json = r#"{"t":"test-token","s":"share","m":0}"#;
        let encoded = URL_SAFE_NO_PAD.encode(json.as_bytes());
        let payload = decode_token_payload(&encoded).unwrap();
        assert_eq!(payload.scope, GrantScope::from("share"));

        // Re-encode and verify roundtrip
        let re_encoded = encode_token_payload(
            &payload.token,
            payload.scope.clone(),
            payload.max_connections,
        );
        let re_decoded = decode_token_payload(&re_encoded).unwrap();
        assert_eq!(re_decoded.token, "test-token");
        assert_eq!(re_decoded.scope, "share");
    }
}