openrtc 2.0.0-rc.7

//! Provider-neutral OpenRTC 2.0 control-plane client for pure native hosts.
//!
//! Consumer applications own authentication, attestation registration, and
//! secure private-key storage. OpenRTC owns assertion exchange, device
//! enrollment, avenue-scoped gateway grants, renewal, and replay-safe device
//! proofs. No Firebase type or project identifier is exposed by this module.

use crate::native_coordination_gateway::{
    NativeCapabilities, NativeCapabilityHandle, NativeGatewayGrant, NativeGatewayGrantProvider,
    NativeGatewayGrantRequest,
};
use crate::signaling::{
    Device, DeviceCapabilities, DeviceEvent, SessionEvent, SignalingBackend, SignalingSession,
};
use anyhow::{anyhow, bail, Context, Result};
use async_trait::async_trait;
use base64::Engine as _;
use ed25519_dalek::{Signature, Verifier, VerifyingKey};
use futures::stream::BoxStream;
use serde::{Deserialize, Serialize};
use serde_json::{json, Value};
use std::fmt;
use std::sync::{Arc, Mutex as StdMutex};
use std::time::{SystemTime, UNIX_EPOCH};
use tokio::sync::{watch, Mutex};
use uuid::Uuid;

pub const OPENRTC_PRODUCTION_CONTROL_PLANE: &str = "https://api.openrtc.app";
const DEVICE_CERTIFICATE_RENEW_SKEW_MS: u64 = 24 * 60 * 60_000;
const SOURCE_RENEW_SKEW_MS: u64 = 5 * 60_000;
const REQUEST_TIMEOUT_SECONDS: u64 = 15;

#[derive(Debug)]
struct NativeControlPlaneHttpError {
    status: u16,
    message: String,
}

impl fmt::Display for NativeControlPlaneHttpError {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        write!(
            formatter,
            "OpenRTC control-plane request failed ({}): {}",
            self.status, self.message
        )
    }
}

impl std::error::Error for NativeControlPlaneHttpError {}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct NativeIdentityAssertion {
    pub token: String,
    pub provider_id: Option<String>,
}

/// Application-owned source of short-lived OIDC/JWKS assertions.
///
/// Implementations may use Firebase Auth, Auth0, Clerk, a custom backend, or
/// another registered issuer. The OpenRTC crate never initializes that IdP.
#[async_trait]
pub trait NativeIdentityAssertionProvider: Send + Sync {
    async fn assertion(&self, force_refresh: bool) -> Result<NativeIdentityAssertion>;

    /// Stable, non-secret local identifier for one signed-in principal. It
    /// changes on login/logout/account switch, never on access-token refresh.
    /// OpenRTC hashes it with the app and install-key thumbprint before asking
    /// the certificate store for a cold-start lookup. `None` preserves the RC
    /// assertion-first compatibility behavior.
    fn session_key(&self) -> Result<Option<String>> {
        Ok(None)
    }

    /// Monotonic application-owned login epoch. Increment it only for login,
    /// logout, account switch, or explicit revocation--never for an ordinary
    /// OAuth access-token refresh.
    fn identity_epoch(&self) -> u64 {
        0
    }

    /// Optional immediate epoch observer. OpenRTC closes every capability
    /// created under the old epoch when this value changes. Hosts without an
    /// observer must explicitly close their handles during logout/account
    /// switch; grant renewal still fails closed if `identity_epoch()` changed.
    fn subscribe_identity_epoch(&self) -> Option<watch::Receiver<u64>> {
        None
    }
}

/// Host secure-storage bridge for the per-install Ed25519 device key.
/// Private key bytes never enter the OpenRTC runtime.
pub trait NativeV2DeviceSigner: Send + Sync {
    fn public_jwk(&self, app_tag: &str) -> Result<Value>;
    fn sign(&self, app_tag: &str, challenge: &[u8]) -> Result<Vec<u8>>;
}

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct StoredNativeDeviceCertificate {
    pub app_tag: String,
    pub principal_id: String,
    pub device_id: String,
    pub token: String,
    pub expires_at_ms: u64,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub session_key_hash: Option<String>,
    /// Public verification material returned alongside the certificate. Old
    /// cache entries intentionally fail closed and re-enroll once.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub signing_public_jwk: Option<Value>,
}

/// Application-owned durable certificate storage. Desktop applications should
/// protect this value with owner-only file permissions or the OS credential
/// store. A certificate is public-key bound, but remains bearer authorization
/// and therefore must not be placed in logs or world-readable preferences.
#[async_trait]
pub trait NativeDeviceCertificateStore: Send + Sync {
    async fn load(
        &self,
        app_tag: &str,
        principal_id: &str,
        device_id: &str,
    ) -> Result<Option<StoredNativeDeviceCertificate>>;

    async fn save(&self, certificate: &StoredNativeDeviceCertificate) -> Result<()>;

    /// Optional cold-start lookup that avoids fetching an IdP assertion merely
    /// to rediscover the stable OpenRTC principal. Production stores should
    /// index the already-hashed value; raw consumer session identifiers must
    /// never be persisted by OpenRTC.
    async fn load_for_session(
        &self,
        _app_tag: &str,
        _session_key_hash: &str,
        _device_id: &str,
    ) -> Result<Option<StoredNativeDeviceCertificate>> {
        Ok(None)
    }

    async fn remove(&self, app_tag: &str, principal_id: &str, device_id: &str) -> Result<()>;
}

/// Network-idle, process-local certificate cache for prototypes and tests.
/// Production native applications should supply durable protected storage.
#[derive(Default)]
pub struct MemoryNativeDeviceCertificateStore {
    certificate: StdMutex<Option<StoredNativeDeviceCertificate>>,
}

#[async_trait]
impl NativeDeviceCertificateStore for MemoryNativeDeviceCertificateStore {
    async fn load(
        &self,
        app_tag: &str,
        principal_id: &str,
        device_id: &str,
    ) -> Result<Option<StoredNativeDeviceCertificate>> {
        Ok(self
            .certificate
            .lock()
            .map_err(|_| anyhow!("native device certificate cache is poisoned"))?
            .clone()
            .filter(|value| {
                value.app_tag == app_tag
                    && value.principal_id == principal_id
                    && value.device_id == device_id
            }))
    }

    async fn save(&self, certificate: &StoredNativeDeviceCertificate) -> Result<()> {
        *self
            .certificate
            .lock()
            .map_err(|_| anyhow!("native device certificate cache is poisoned"))? =
            Some(certificate.clone());
        Ok(())
    }

    async fn load_for_session(
        &self,
        app_tag: &str,
        session_key_hash: &str,
        device_id: &str,
    ) -> Result<Option<StoredNativeDeviceCertificate>> {
        Ok(self
            .certificate
            .lock()
            .map_err(|_| anyhow!("native device certificate cache is poisoned"))?
            .clone()
            .filter(|value| {
                value.app_tag == app_tag
                    && value.device_id == device_id
                    && value.session_key_hash.as_deref() == Some(session_key_hash)
            }))
    }

    async fn remove(&self, app_tag: &str, principal_id: &str, device_id: &str) -> Result<()> {
        let mut guard = self
            .certificate
            .lock()
            .map_err(|_| anyhow!("native device certificate cache is poisoned"))?;
        if guard.as_ref().is_some_and(|value| {
            value.app_tag == app_tag
                && value.principal_id == principal_id
                && value.device_id == device_id
        }) {
            *guard = None;
        }
        Ok(())
    }
}

#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct NativeAttestationEvidence {
    pub kind: String,
    pub token: String,
}

/// Optional application-owned managed-attestation bridge. It is invoked only
/// during enrollment or certificate renewal, never as a socket heartbeat.
#[async_trait]
pub trait NativeAttestationProvider: Send + Sync {
    async fn evidence(&self, challenge: &str, api_key: &str) -> Result<NativeAttestationEvidence>;
}

#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct NativeV2Features {
    pub relay: bool,
    pub moq: bool,
    pub ble: bool,
    pub advanced_fanout: bool,
    pub durable_membership: bool,
}

#[derive(Clone, Default)]
pub struct NativeV2DevicesOptions {
    pub max_peers: Option<u32>,
    pub features: NativeV2Features,
    pub attestation: Option<Arc<dyn NativeAttestationProvider>>,
    pub identity_relay: Option<NativeV2IdentityCredentialRelay>,
}

#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct NativeV2CapabilityOptions {
    pub max_peers: Option<u32>,
    pub features: NativeV2Features,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum NativeV2AnonymousKind {
    Space,
    Room,
    Ticket,
}

impl NativeV2AnonymousKind {
    fn source_kind(self) -> &'static str {
        match self {
            Self::Space => "space",
            Self::Room => "room",
            Self::Ticket => "ticket",
        }
    }
}

/// Mutable opaque-credential relay for native runtimes constructed before the
/// application finishes login. It carries only OpenRTC-issued device
/// certificates and follows certificate renewal performed by the grant owner.
#[derive(Clone, Default)]
pub struct NativeV2IdentityCredentialRelay {
    value: Arc<StdMutex<Option<String>>>,
}

impl NativeV2IdentityCredentialRelay {
    pub fn provider(&self) -> Box<dyn Fn() -> Option<String> + Send + Sync> {
        let state = self.value.clone();
        Box::new(move || state.lock().ok().and_then(|value| value.clone()))
    }

    fn set(&self, value: Option<String>) -> Result<()> {
        *self
            .value
            .lock()
            .map_err(|_| anyhow!("native identity credential state is poisoned"))? = value;
        Ok(())
    }
}

/// One-time, fail-closed signaling installation point for native hosts whose
/// process is created before user authentication completes. Installation may
/// happen exactly once; replacement requires closing and rebuilding the
/// capability/runtime so this cannot become a shadow lifecycle owner.
#[derive(Default)]
pub struct NativeV2SignalingSlot {
    backend: tokio::sync::RwLock<Option<Arc<dyn SignalingBackend>>>,
}

impl NativeV2SignalingSlot {
    pub async fn install(&self, backend: Arc<dyn SignalingBackend>) -> Result<()> {
        let mut current = self.backend.write().await;
        if current.is_some() {
            bail!("native OpenRTC capability signaling is already installed");
        }
        *current = Some(backend);
        Ok(())
    }

    async fn current(&self) -> Result<Arc<dyn SignalingBackend>> {
        self.backend
            .read()
            .await
            .clone()
            .ok_or_else(|| anyhow!("native OpenRTC capability is not active"))
    }
}

#[async_trait]
impl SignalingBackend for NativeV2SignalingSlot {
    async fn update_presence(
        &self,
        user_id: &str,
        local_node_id: &str,
        ticket_str: &str,
        is_online: bool,
        name: &str,
        ttl_ms: u64,
        metadata: Option<&str>,
    ) -> Result<()> {
        self.current()
            .await?
            .update_presence(
                user_id,
                local_node_id,
                ticket_str,
                is_online,
                name,
                ttl_ms,
                metadata,
            )
            .await
    }

    async fn set_offline(&self, user_id: &str, local_node_id: &str) -> Result<()> {
        self.current()
            .await?
            .set_offline(user_id, local_node_id)
            .await
    }

    async fn update_live_presence(
        &self,
        user_id: &str,
        local_node_id: &str,
        ticket_str: &str,
        name: &str,
        metadata: Option<&str>,
    ) -> Result<()> {
        self.current()
            .await?
            .update_live_presence(user_id, local_node_id, ticket_str, name, metadata)
            .await
    }

    async fn set_live_presence_offline(&self, user_id: &str, local_node_id: &str) -> Result<()> {
        self.current()
            .await?
            .set_live_presence_offline(user_id, local_node_id)
            .await
    }

    async fn update_device(
        &self,
        user_id: &str,
        device_id: &str,
        device_name: Option<&str>,
        capabilities: Option<DeviceCapabilities>,
        metadata: Option<&str>,
    ) -> Result<()> {
        self.current()
            .await?
            .update_device(user_id, device_id, device_name, capabilities, metadata)
            .await
    }

    async fn delete_device(&self, user_id: &str, device_id: &str) -> Result<()> {
        self.current()
            .await?
            .delete_device(user_id, device_id)
            .await
    }

    async fn set_excluded_peers(
        &self,
        user_id: &str,
        local_node_id: &str,
        excluded_peers: &[String],
    ) -> Result<()> {
        self.current()
            .await?
            .set_excluded_peers(user_id, local_node_id, excluded_peers)
            .await
    }

    async fn search_devices(
        &self,
        user_id: &str,
        exclude_node_id: Option<&str>,
    ) -> Result<Vec<Device>> {
        self.current()
            .await?
            .search_devices(user_id, exclude_node_id)
            .await
    }

    async fn list_devices(
        &self,
        user_id: &str,
        exclude_node_id: Option<&str>,
    ) -> Result<Vec<Device>> {
        self.current()
            .await?
            .list_devices(user_id, exclude_node_id)
            .await
    }

    async fn send_message(
        &self,
        sender_id: &str,
        target_id: &str,
        payload: &str,
        state: Option<&str>,
        reply_payload: Option<&str>,
    ) -> Result<String> {
        self.current()
            .await?
            .send_message(sender_id, target_id, payload, state, reply_payload)
            .await
    }

    async fn subscribe_devices(
        &self,
        user_id: &str,
    ) -> Result<BoxStream<'static, Result<Vec<DeviceEvent>>>> {
        self.current().await?.subscribe_devices(user_id).await
    }

    async fn create_session(&self, session: SignalingSession) -> Result<()> {
        self.current().await?.create_session(session).await
    }

    async fn update_session(&self, session_id: &str, update_data: Value) -> Result<()> {
        self.current()
            .await?
            .update_session(session_id, update_data)
            .await
    }

    async fn subscribe_sessions(
        &self,
        local_device_id: &str,
    ) -> Result<BoxStream<'static, Result<Vec<SessionEvent>>>> {
        self.current()
            .await?
            .subscribe_sessions(local_device_id)
            .await
    }
}

#[derive(Debug, Clone)]
struct CredentialSource {
    token: String,
    expires_at_ms: u64,
    principal_id: String,
}

#[derive(Debug, Clone)]
struct AnonymousCredentialSource {
    token: String,
    expires_at_ms: u64,
    principal_id: String,
    device_id: String,
    requested_id: String,
    avenue_id: String,
}

#[derive(Clone)]
pub struct NativeV2ControlPlane {
    inner: Arc<NativeV2ControlPlaneInner>,
}

struct NativeV2ControlPlaneInner {
    api_key: String,
    app_tag: String,
    endpoint: String,
    gateway_endpoint: String,
    http: reqwest::Client,
    assertion_provider: Option<Arc<dyn NativeIdentityAssertionProvider>>,
    signer: Arc<dyn NativeV2DeviceSigner>,
    certificate_store: Arc<dyn NativeDeviceCertificateStore>,
    auth_epoch: Mutex<NativeAuthEpochState>,
}

#[derive(Debug, Clone)]
struct CachedNativeIdentity {
    response: AssertionExchangeResponse,
    expires_at_ms: u64,
}

#[derive(Debug, Default)]
struct NativeAuthEpochState {
    epoch: Option<u64>,
    identity: Option<CachedNativeIdentity>,
}

/// An authenticated device-mesh capability plus its stable app-scoped
/// principal. The handle remains network-idle until the runtime publishes its
/// initial presence, and `close()` stops its socket and grant-refresh owner.
pub struct NativeAuthenticatedDevices {
    pub principal_id: String,
    pub handle: NativeCapabilityHandle,
    identity_relay: NativeV2IdentityCredentialRelay,
    identity_epoch_monitor: Option<tokio::task::JoinHandle<()>>,
}

/// A backend-free native capability. Its install-bound principal and device
/// identity come from the host-secure Ed25519 key, and the handle owns exactly
/// one live avenue.
pub struct NativeAnonymousCapability {
    pub principal_id: String,
    pub handle: NativeCapabilityHandle,
}

impl NativeAuthenticatedDevices {
    /// Opaque device certificate used by Rust peer-session admission. The
    /// provider observes certificate renewal without exposing any private key.
    pub fn identity_credential_provider(&self) -> Box<dyn Fn() -> Option<String> + Send + Sync> {
        self.identity_relay.provider()
    }

    /// Install this one avenue in the shared Rust runtime. The returned
    /// backend remains owned by this disposable capability handle.
    pub fn signaling(&self) -> Arc<dyn SignalingBackend> {
        self.handle.signaling()
    }

    pub fn is_closed(&self) -> bool {
        self.handle.is_closed()
    }

    pub async fn close(&self) {
        if let Some(monitor) = &self.identity_epoch_monitor {
            monitor.abort();
        }
        self.handle.close().await;
    }
}

impl Drop for NativeAuthenticatedDevices {
    fn drop(&mut self) {
        if let Some(monitor) = &self.identity_epoch_monitor {
            monitor.abort();
        }
    }
}

impl NativeAnonymousCapability {
    /// Install this one avenue in the shared Rust runtime. Capability/grant
    /// renewal remains internal and does not replace a healthy socket.
    pub fn signaling(&self) -> Arc<dyn SignalingBackend> {
        self.handle.signaling()
    }

    pub fn is_closed(&self) -> bool {
        self.handle.is_closed()
    }

    pub async fn close(&self) {
        self.handle.close().await;
    }
}

impl NativeV2ControlPlane {
    pub fn new(
        api_key: &str,
        assertion_provider: Arc<dyn NativeIdentityAssertionProvider>,
        signer: Arc<dyn NativeV2DeviceSigner>,
        certificate_store: Arc<dyn NativeDeviceCertificateStore>,
    ) -> Result<Self> {
        let api_key = crate::validate_v2_public_api_key(api_key)?.to_string();
        let app_tag = crate::app_tag_from_api_key(&api_key);
        Ok(Self {
            inner: Arc::new(NativeV2ControlPlaneInner {
                api_key,
                app_tag,
                endpoint: OPENRTC_PRODUCTION_CONTROL_PLANE.to_string(),
                gateway_endpoint:
                    crate::native_coordination_gateway::OPENRTC_PRODUCTION_COORDINATION_GATEWAY
                        .to_string(),
                http: reqwest::Client::builder()
                    .timeout(std::time::Duration::from_secs(REQUEST_TIMEOUT_SECONDS))
                    .build()?,
                assertion_provider: Some(assertion_provider),
                signer,
                certificate_store,
                auth_epoch: Mutex::new(NativeAuthEpochState::default()),
            }),
        })
    }

    /// Construct the API-key-only native prototype surface. Construction is
    /// network-idle. A capability request occurs only when `join_space`,
    /// `join_room`, or `issue_ticket` is awaited.
    pub fn anonymous(api_key: &str, signer: Arc<dyn NativeV2DeviceSigner>) -> Result<Self> {
        let api_key = crate::validate_v2_public_api_key(api_key)?.to_string();
        let app_tag = crate::app_tag_from_api_key(&api_key);
        Ok(Self {
            inner: Arc::new(NativeV2ControlPlaneInner {
                api_key,
                app_tag,
                endpoint: OPENRTC_PRODUCTION_CONTROL_PLANE.to_string(),
                gateway_endpoint:
                    crate::native_coordination_gateway::OPENRTC_PRODUCTION_COORDINATION_GATEWAY
                        .to_string(),
                http: reqwest::Client::builder()
                    .timeout(std::time::Duration::from_secs(REQUEST_TIMEOUT_SECONDS))
                    .build()?,
                assertion_provider: None,
                signer,
                certificate_store: Arc::new(MemoryNativeDeviceCertificateStore::default()),
                auth_epoch: Mutex::new(NativeAuthEpochState::default()),
            }),
        })
    }

    pub fn app_tag(&self) -> &str {
        &self.inner.app_tag
    }

    #[cfg(any(test, feature = "testing-endpoints"))]
    pub fn with_testing_endpoints(
        mut self,
        control_plane: impl Into<String>,
        gateway: impl Into<String>,
    ) -> Result<Self> {
        let control_plane = validate_endpoint(control_plane.into())?;
        let gateway = validate_endpoint(gateway.into())?;
        Arc::get_mut(&mut self.inner)
            .ok_or_else(|| anyhow!("testing endpoints must be set before cloning the client"))?
            .endpoint = control_plane;
        Arc::get_mut(&mut self.inner)
            .ok_or_else(|| anyhow!("testing endpoints must be set before cloning the client"))?
            .gateway_endpoint = gateway;
        Ok(self)
    }

    pub async fn devices(
        &self,
        device_id: impl Into<String>,
        platform_type: impl Into<String>,
        options: NativeV2DevicesOptions,
    ) -> Result<NativeAuthenticatedDevices> {
        let device_id = bounded_id("device_id", device_id.into())?;
        let platform_type = bounded_id("platform_type", platform_type.into())?;
        let assertion_provider = self.inner.assertion_provider.as_ref().ok_or_else(|| {
            anyhow!("authenticated devices require a native identity assertion provider")
        })?;
        let mut epoch_receiver = assertion_provider.subscribe_identity_epoch();
        let identity_epoch = epoch_receiver
            .as_ref()
            .map(|receiver| *receiver.borrow())
            .unwrap_or_else(|| assertion_provider.identity_epoch());
        let source = self
            .device_certificate(&device_id, options.attestation.as_deref(), identity_epoch)
            .await?;
        let principal_id = source.principal_id.clone();
        let identity_relay = options.identity_relay.clone().unwrap_or_default();
        identity_relay.set(Some(source.token.clone()))?;
        let grant_provider: Arc<dyn NativeGatewayGrantProvider> =
            Arc::new(ControlPlaneGrantProvider {
                control_plane: self.clone(),
                device_id: device_id.clone(),
                source: Mutex::new(source),
                max_peers: options.max_peers,
                features: options.features,
                attestation: options.attestation,
                identity_relay: identity_relay.clone(),
                identity_epoch,
            });
        let capabilities = NativeCapabilities::new(
            &self.inner.api_key,
            device_id,
            platform_type,
            grant_provider,
        )?;
        #[cfg(any(test, feature = "testing-endpoints"))]
        let capabilities = capabilities.with_testing_endpoint(&self.inner.gateway_endpoint)?;
        let handle = capabilities.devices(principal_id.clone())?;
        let identity_epoch_monitor = epoch_receiver.take().map(|receiver| {
            spawn_identity_epoch_monitor(
                receiver,
                identity_epoch,
                handle.closer(),
                identity_relay.clone(),
            )
        });
        Ok(NativeAuthenticatedDevices {
            handle,
            principal_id,
            identity_relay,
            identity_epoch_monitor,
        })
    }

    pub async fn join_space(
        &self,
        id: impl Into<String>,
        platform_type: impl Into<String>,
        options: NativeV2CapabilityOptions,
    ) -> Result<NativeAnonymousCapability> {
        self.anonymous_capability(
            NativeV2AnonymousKind::Space,
            id.into(),
            platform_type.into(),
            options,
        )
        .await
    }

    pub async fn join_room(
        &self,
        id: impl Into<String>,
        platform_type: impl Into<String>,
        options: NativeV2CapabilityOptions,
    ) -> Result<NativeAnonymousCapability> {
        self.anonymous_capability(
            NativeV2AnonymousKind::Room,
            id.into(),
            platform_type.into(),
            options,
        )
        .await
    }

    pub async fn issue_ticket(
        &self,
        id: impl Into<String>,
        platform_type: impl Into<String>,
        options: NativeV2CapabilityOptions,
    ) -> Result<NativeAnonymousCapability> {
        self.anonymous_capability(
            NativeV2AnonymousKind::Ticket,
            id.into(),
            platform_type.into(),
            options,
        )
        .await
    }

    async fn anonymous_capability(
        &self,
        kind: NativeV2AnonymousKind,
        id: String,
        platform_type: String,
        options: NativeV2CapabilityOptions,
    ) -> Result<NativeAnonymousCapability> {
        let platform_type = bounded_id("platform_type", platform_type)?;
        let requested_id = bounded_id("capability_id", id)?;
        let source = self
            .anonymous_source(kind, requested_id, options.max_peers)
            .await?;
        let principal_id = source.principal_id.clone();
        let device_id = source.device_id.clone();
        let avenue_id = source.avenue_id.clone();
        let grant_provider: Arc<dyn NativeGatewayGrantProvider> =
            Arc::new(AnonymousControlPlaneGrantProvider {
                control_plane: self.clone(),
                source: Mutex::new(source),
                kind,
                max_peers: options.max_peers,
                features: options.features,
            });
        let capabilities = NativeCapabilities::new(
            &self.inner.api_key,
            device_id,
            platform_type,
            grant_provider,
        )?;
        #[cfg(any(test, feature = "testing-endpoints"))]
        let capabilities = capabilities.with_testing_endpoint(&self.inner.gateway_endpoint)?;
        let handle = match kind {
            NativeV2AnonymousKind::Space => capabilities.join_space(avenue_id)?,
            NativeV2AnonymousKind::Room => capabilities.join_room(avenue_id)?,
            NativeV2AnonymousKind::Ticket => capabilities.issue_ticket(avenue_id)?,
        };
        Ok(NativeAnonymousCapability {
            principal_id,
            handle,
        })
    }

    async fn anonymous_source(
        &self,
        kind: NativeV2AnonymousKind,
        requested_id: String,
        max_peers: Option<u32>,
    ) -> Result<AnonymousCredentialSource> {
        let original_requested_id = requested_id.clone();
        let avenue_id = native_capability_avenue_id(&self.inner.api_key, kind, &requested_id);
        let proof = self.device_proof(|nonce, issued_at| {
            format!(
                "openrtc:v2:capability:{}:{}:{}:{}:{}",
                self.inner.api_key,
                kind.source_kind(),
                avenue_id,
                nonce,
                issued_at
            )
        })?;
        let mut body = json!({
            "avenue": { "kind": kind.source_kind(), "id": avenue_id },
            "deviceProof": proof,
        });
        if let Some(max_peers) = max_peers {
            body["maxPeers"] = json!(max_peers);
        }
        let response: CapabilityResponse = self.post("/v2/capabilities", body).await?;
        let token = required("capability", response.capability)?;
        let principal_id = required_claim_string(&token, "principalId")?;
        let thumbprint = required_claim_string(&token, "deviceKeyThumbprint")?;
        let claim_exp = decode_claims(&token)?
            .get("exp")
            .and_then(Value::as_u64)
            .ok_or_else(|| anyhow!("OpenRTC capability is missing exp"))?;
        if claim_exp != response.expires_at
            || response.avenue.kind != kind.source_kind()
            || response.avenue.id != avenue_id
        {
            bail!("OpenRTC capability response scope is inconsistent");
        }
        Ok(AnonymousCredentialSource {
            token,
            expires_at_ms: response.expires_at.saturating_mul(1_000),
            principal_id,
            device_id: format!("anon_{}", thumbprint.chars().take(32).collect::<String>()),
            requested_id: original_requested_id,
            avenue_id,
        })
    }

    async fn device_certificate(
        &self,
        device_id: &str,
        attestation: Option<&dyn NativeAttestationProvider>,
        identity_epoch: u64,
    ) -> Result<CredentialSource> {
        let device_key_thumbprint =
            native_device_key_thumbprint(&self.inner.signer.public_jwk(&self.inner.app_tag)?)?;
        let assertion_provider = self.inner.assertion_provider.as_ref().ok_or_else(|| {
            anyhow!("authenticated devices require a native identity assertion provider")
        })?;
        let session_key_hash = assertion_provider
            .session_key()?
            .filter(|value| !value.trim().is_empty())
            .map(|value| {
                native_session_key_hash(&self.inner.app_tag, &device_key_thumbprint, value.trim())
            });
        if let Some(session_key_hash) = session_key_hash.as_deref() {
            if let Some(cached) = self
                .inner
                .certificate_store
                .load_for_session(&self.inner.app_tag, session_key_hash, device_id)
                .await?
                .filter(|value| certificate_is_reusable(value, &device_key_thumbprint, now_ms()))
            {
                return Ok(CredentialSource {
                    token: cached.token,
                    expires_at_ms: cached.expires_at_ms,
                    principal_id: cached.principal_id,
                });
            }
        }

        let identity = self.identity_for_epoch(identity_epoch, false).await?;
        let principal_id = bounded_id("principal_id", identity.response.principal_id.clone())?;

        if let Some(cached) = self
            .inner
            .certificate_store
            .load(&self.inner.app_tag, &principal_id, device_id)
            .await?
            .filter(|value| certificate_is_reusable(value, &device_key_thumbprint, now_ms()))
        {
            return Ok(CredentialSource {
                token: cached.token,
                expires_at_ms: cached.expires_at_ms,
                principal_id,
            });
        }

        // Identity sessions are intentionally short-lived. Reuse one only
        // inside the same host-declared login epoch, and exchange again only
        // when enrollment actually needs a fresh session. The provider's own
        // cached assertion is tried first; force refresh remains 401-driven.
        let identity = if identity.expires_at_ms > now_ms().saturating_add(30_000) {
            identity
        } else {
            self.identity_for_epoch(identity_epoch, true).await?
        };
        if identity.response.principal_id != principal_id {
            bail!("native identity principal changed without advancing its login epoch");
        }

        let proof = self.device_proof(|nonce, issued_at| {
            format!(
                "openrtc:v2:device-enroll:{}:{}:{}:{}:{}",
                self.inner.app_tag, principal_id, device_id, nonce, issued_at
            )
        })?;
        let attestation_value = if let Some(provider) = attestation {
            let challenge = format!(
                "openrtc:v2:attestation:{}:{}:{}:{}",
                self.inner.app_tag, principal_id, device_id, proof.nonce
            );
            Some(provider.evidence(&challenge, &self.inner.api_key).await?)
        } else {
            None
        };
        let mut enrollment_body = json!({
            "identitySession": identity.response.identity_session,
            "deviceId": device_id,
            "deviceProof": proof,
        });
        if let Some(attestation) = attestation_value {
            enrollment_body["attestation"] = json!(attestation);
        }
        let enrollment: DeviceEnrollmentResponse = self
            .post("/v2/devices/enroll", enrollment_body)
            .await
            .context("enroll native OpenRTC device")?;
        let expires_at_ms = enrollment
            .expires_at
            .checked_mul(1_000)
            .ok_or_else(|| anyhow!("device certificate expiry overflow"))?;
        let stored = StoredNativeDeviceCertificate {
            app_tag: self.inner.app_tag.clone(),
            principal_id: principal_id.clone(),
            device_id: device_id.to_string(),
            token: required("device certificate", enrollment.device_certificate)?,
            expires_at_ms,
            session_key_hash,
            signing_public_jwk: Some(enrollment.signing_public_jwk),
        };
        validate_stored_certificate(&stored, &device_key_thumbprint)?;
        self.inner.certificate_store.save(&stored).await?;
        Ok(CredentialSource {
            token: stored.token,
            expires_at_ms,
            principal_id,
        })
    }

    async fn exchange_identity_after_rejection(
        &self,
        force_assertion_refresh: bool,
    ) -> Result<AssertionExchangeResponse> {
        Ok(
            match self.exchange_identity(force_assertion_refresh).await {
                Ok(identity) => identity,
                Err(error)
                    if !force_assertion_refresh
                        && error
                            .downcast_ref::<NativeControlPlaneHttpError>()
                            .is_some_and(|failure| failure.status == 401) =>
                {
                    // Match the browser/WASM contract: use the consumer's cached
                    // identity first and request one fresh assertion only after an
                    // explicit rejection. No timer or unbounded validation loop is
                    // introduced.
                    self.exchange_identity(true).await?
                }
                Err(error) => return Err(error),
            },
        )
    }

    async fn identity_for_epoch(
        &self,
        identity_epoch: u64,
        require_unexpired_session: bool,
    ) -> Result<CachedNativeIdentity> {
        let assertion_provider = self.inner.assertion_provider.as_ref().ok_or_else(|| {
            anyhow!("authenticated devices require a native identity assertion provider")
        })?;
        if assertion_provider.identity_epoch() != identity_epoch {
            bail!("native identity changed before OpenRTC capability activation");
        }
        let mut state = self.inner.auth_epoch.lock().await;
        if state.epoch != Some(identity_epoch) {
            state.epoch = Some(identity_epoch);
            state.identity = None;
        }
        if let Some(identity) = state.identity.clone() {
            if !require_unexpired_session
                || identity.expires_at_ms > now_ms().saturating_add(30_000)
            {
                if assertion_provider.identity_epoch() != identity_epoch {
                    state.identity = None;
                    bail!("native identity changed during capability activation");
                }
                return Ok(identity);
            }
        }
        let response = self.exchange_identity_after_rejection(false).await?;
        if assertion_provider.identity_epoch() != identity_epoch {
            state.identity = None;
            bail!("native identity changed during assertion exchange");
        }
        let expires_at_ms = response
            .expires_at
            .checked_mul(1_000)
            .ok_or_else(|| anyhow!("identity session expiry overflow"))?;
        let identity = CachedNativeIdentity {
            response,
            expires_at_ms,
        };
        state.identity = Some(identity.clone());
        Ok(identity)
    }

    async fn exchange_identity(&self, force_refresh: bool) -> Result<AssertionExchangeResponse> {
        let assertion_provider = self.inner.assertion_provider.as_ref().ok_or_else(|| {
            anyhow!("authenticated devices require a native identity assertion provider")
        })?;
        let assertion = assertion_provider
            .assertion(force_refresh)
            .await
            .context("obtain consumer identity assertion")?;
        let assertion_token = required("identity assertion", assertion.token)?;
        let mut assertion_body = json!({ "assertion": assertion_token });
        if let Some(provider_id) = assertion.provider_id {
            assertion_body["providerId"] = json!(provider_id);
        }
        self.post("/v2/assertions/exchange", assertion_body)
            .await
            .context("exchange consumer identity assertion")
    }

    fn device_proof(
        &self,
        challenge: impl FnOnce(&str, u64) -> String,
    ) -> Result<NativeDeviceProof> {
        let public_key_jwk = self.inner.signer.public_jwk(&self.inner.app_tag)?;
        validate_public_jwk(&public_key_jwk)?;
        let nonce = Uuid::new_v4().simple().to_string();
        let issued_at = now_seconds();
        let challenge = challenge(&nonce, issued_at);
        let signature = self
            .inner
            .signer
            .sign(&self.inner.app_tag, challenge.as_bytes())?;
        if signature.len() != 64 {
            bail!("native device signer returned an invalid Ed25519 signature");
        }
        Ok(NativeDeviceProof {
            public_key_jwk,
            signature: base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(signature),
            nonce,
            issued_at,
        })
    }

    async fn post<T: for<'de> Deserialize<'de>>(&self, path: &str, body: Value) -> Result<T> {
        let mut object = body
            .as_object()
            .cloned()
            .ok_or_else(|| anyhow!("native control-plane body must be an object"))?;
        object.insert("apiKey".to_string(), json!(self.inner.api_key));
        let url = format!("{}{}", self.inner.endpoint.trim_end_matches('/'), path);
        let request_id = format!("native_{}", Uuid::new_v4().simple());
        let serialized = serde_json::to_vec(&object).context("encode OpenRTC request")?;

        for attempt in 0..2 {
            let response = self
                .inner
                .http
                .post(&url)
                .header("X-OpenRTC-Idempotency-Key", &request_id)
                .header(reqwest::header::CONTENT_TYPE, "application/json")
                .body(serialized.clone())
                .send()
                .await;
            let response = match response {
                Ok(response) => response,
                Err(_error) if attempt == 0 => {
                    tokio::time::sleep(std::time::Duration::from_millis(100)).await;
                    continue;
                }
                Err(error) => return Err(error).context("send OpenRTC request"),
            };
            let status = response.status();
            if attempt == 0
                && matches!(
                    status,
                    reqwest::StatusCode::BAD_GATEWAY
                        | reqwest::StatusCode::SERVICE_UNAVAILABLE
                        | reqwest::StatusCode::GATEWAY_TIMEOUT
                )
            {
                tokio::time::sleep(std::time::Duration::from_millis(100)).await;
                continue;
            }
            if !status.is_success() {
                let payload = response.json::<Value>().await.unwrap_or(Value::Null);
                let message = payload
                    .get("error")
                    .and_then(Value::as_str)
                    .unwrap_or("OpenRTC control-plane request failed")
                    .chars()
                    .take(256)
                    .collect::<String>();
                return Err(anyhow::Error::new(NativeControlPlaneHttpError {
                    status: status.as_u16(),
                    message,
                }));
            }
            return response
                .json::<T>()
                .await
                .context("decode OpenRTC response");
        }
        unreachable!("native control-plane retry loop has two terminal attempts")
    }
}

fn spawn_identity_epoch_monitor(
    mut receiver: watch::Receiver<u64>,
    identity_epoch: u64,
    closer: crate::native_coordination_gateway::NativeCapabilityCloser,
    relay: NativeV2IdentityCredentialRelay,
) -> tokio::task::JoinHandle<()> {
    tokio::spawn(async move {
        loop {
            if receiver.changed().await.is_err() {
                return;
            }
            if *receiver.borrow_and_update() != identity_epoch {
                let _ = relay.set(None);
                closer.close().await;
                return;
            }
        }
    })
}

struct ControlPlaneGrantProvider {
    control_plane: NativeV2ControlPlane,
    device_id: String,
    source: Mutex<CredentialSource>,
    max_peers: Option<u32>,
    features: NativeV2Features,
    attestation: Option<Arc<dyn NativeAttestationProvider>>,
    identity_relay: NativeV2IdentityCredentialRelay,
    identity_epoch: u64,
}

struct AnonymousControlPlaneGrantProvider {
    control_plane: NativeV2ControlPlane,
    source: Mutex<AnonymousCredentialSource>,
    kind: NativeV2AnonymousKind,
    max_peers: Option<u32>,
    features: NativeV2Features,
}

#[async_trait]
impl NativeGatewayGrantProvider for ControlPlaneGrantProvider {
    async fn grant(&self, request: NativeGatewayGrantRequest) -> Result<NativeGatewayGrant> {
        if request.device_id != self.device_id {
            bail!("gateway grant request device does not match enrolled device");
        }
        let mut source = self.source.lock().await;
        if source.expires_at_ms <= now_ms().saturating_add(SOURCE_RENEW_SKEW_MS) {
            *source = self
                .control_plane
                .device_certificate(
                    &self.device_id,
                    self.attestation.as_deref(),
                    self.identity_epoch,
                )
                .await?;
            self.identity_relay.set(Some(source.token.clone()))?;
        }
        let source_jti = required_claim_string(&source.token, "jti")?;
        let proof = self.control_plane.device_proof(|nonce, issued_at| {
            format!(
                "openrtc:v2:gateway-grant:{}:{}:{}:{}:{}:{}",
                source_jti,
                request.avenue.kind,
                request.avenue.id,
                request.runtime_instance_id,
                nonce,
                issued_at
            )
        })?;
        let mut grant_body = json!({
            "credentialType": "device-certificate",
            "credential": source.token,
            "avenue": request.avenue,
            "runtimeInstanceId": request.runtime_instance_id,
            "ticketFingerprint": request.ticket_fingerprint,
            "deviceProof": proof,
            "features": self.features,
        });
        if let Some(refresh_grant) = request.refresh_grant {
            grant_body["refreshGrant"] = json!(refresh_grant);
        }
        if let Some(max_peers) = self.max_peers {
            grant_body["maxPeers"] = json!(max_peers);
        }
        let response: GatewayGrantResponse = self
            .control_plane
            .post("/v2/gateway/grants", grant_body)
            .await?;
        if response.gateway_url.trim_end_matches('/')
            != self
                .control_plane
                .inner
                .gateway_endpoint
                .trim_end_matches('/')
        {
            bail!("OpenRTC returned an unexpected native gateway origin");
        }
        Ok(NativeGatewayGrant {
            protocol_version: response.protocol_version,
            gateway_url: response.gateway_url,
            route_key: response.route_key,
            token: response.token,
            expires_at_ms: response.expires_at_ms,
        })
    }
}

#[async_trait]
impl NativeGatewayGrantProvider for AnonymousControlPlaneGrantProvider {
    async fn grant(&self, request: NativeGatewayGrantRequest) -> Result<NativeGatewayGrant> {
        let mut source = self.source.lock().await;
        if request.device_id != source.device_id {
            bail!("gateway grant request device does not match the capability install key");
        }
        if source.expires_at_ms <= now_ms().saturating_add(SOURCE_RENEW_SKEW_MS) {
            *source = self
                .control_plane
                .anonymous_source(self.kind, source.requested_id.clone(), self.max_peers)
                .await?;
        }
        let expected_kind = match self.kind {
            NativeV2AnonymousKind::Ticket => "session",
            _ => self.kind.source_kind(),
        };
        if request.avenue.kind != expected_kind || request.avenue.id != source.avenue_id {
            bail!("gateway grant request avenue does not match the capability");
        }
        let source_jti = required_claim_string(&source.token, "jti")?;
        let proof = self.control_plane.device_proof(|nonce, issued_at| {
            format!(
                "openrtc:v2:gateway-grant:{}:{}:{}:{}:{}:{}",
                source_jti,
                request.avenue.kind,
                request.avenue.id,
                request.runtime_instance_id,
                nonce,
                issued_at
            )
        })?;
        let mut grant_body = json!({
            "credentialType": "capability",
            "credential": source.token,
            "avenue": request.avenue,
            "runtimeInstanceId": request.runtime_instance_id,
            "ticketFingerprint": request.ticket_fingerprint,
            "deviceProof": proof,
            "features": self.features,
        });
        if let Some(refresh_grant) = request.refresh_grant {
            grant_body["refreshGrant"] = json!(refresh_grant);
        }
        if let Some(max_peers) = self.max_peers {
            grant_body["maxPeers"] = json!(max_peers);
        }
        let response: GatewayGrantResponse = self
            .control_plane
            .post("/v2/gateway/grants", grant_body)
            .await?;
        if response.gateway_url.trim_end_matches('/')
            != self
                .control_plane
                .inner
                .gateway_endpoint
                .trim_end_matches('/')
        {
            bail!("OpenRTC returned an unexpected native gateway origin");
        }
        Ok(NativeGatewayGrant {
            protocol_version: response.protocol_version,
            gateway_url: response.gateway_url,
            route_key: response.route_key,
            token: response.token,
            expires_at_ms: response.expires_at_ms,
        })
    }
}

#[derive(Debug, Serialize)]
#[serde(rename_all = "camelCase")]
struct NativeDeviceProof {
    public_key_jwk: Value,
    signature: String,
    nonce: String,
    issued_at: u64,
}

#[derive(Debug, Clone, Deserialize)]
#[serde(rename_all = "camelCase")]
struct AssertionExchangeResponse {
    identity_session: String,
    principal_id: String,
    expires_at: u64,
}

#[derive(Deserialize)]
#[serde(rename_all = "camelCase")]
struct DeviceEnrollmentResponse {
    device_certificate: String,
    expires_at: u64,
    signing_public_jwk: Value,
}

#[derive(Deserialize)]
#[serde(rename_all = "camelCase")]
struct CapabilityResponse {
    capability: String,
    expires_at: u64,
    avenue: CapabilityAvenueResponse,
}

#[derive(Deserialize)]
struct CapabilityAvenueResponse {
    kind: String,
    id: String,
}

#[derive(Deserialize)]
#[serde(rename_all = "camelCase")]
struct GatewayGrantResponse {
    protocol_version: u8,
    gateway_url: String,
    route_key: String,
    token: String,
    expires_at_ms: u64,
}

fn now_seconds() -> u64 {
    SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .unwrap_or_default()
        .as_secs()
}

fn now_ms() -> u64 {
    now_seconds().saturating_mul(1_000)
}

fn required(field: &str, value: String) -> Result<String> {
    let value = value.trim().to_string();
    if value.is_empty() {
        bail!("{field} is required");
    }
    Ok(value)
}

fn bounded_id(field: &str, value: String) -> Result<String> {
    let value = required(field, value)?;
    if value.len() > 160
        || !value.bytes().all(|byte| {
            byte.is_ascii_alphanumeric() || matches!(byte, b'_' | b'-' | b'.' | b':' | b'@')
        })
    {
        bail!("{field} is invalid");
    }
    Ok(value)
}

fn native_capability_avenue_id(
    api_key: &str,
    kind: NativeV2AnonymousKind,
    requested_id: &str,
) -> String {
    if kind != NativeV2AnonymousKind::Space {
        return requested_id.to_string();
    }
    let digest =
        <sha2::Sha256 as sha2::Digest>::digest(format!("{api_key}:{requested_id}").as_bytes());
    hex::encode(digest)
}

fn validate_endpoint(value: String) -> Result<String> {
    let parsed = reqwest::Url::parse(value.trim())?;
    if !matches!(parsed.scheme(), "https" | "http")
        || parsed.host_str().is_none()
        || !parsed.username().is_empty()
        || parsed.password().is_some()
        || parsed.query().is_some()
        || parsed.fragment().is_some()
    {
        bail!("OpenRTC testing endpoint is invalid");
    }
    Ok(parsed.as_str().trim_end_matches('/').to_string())
}

fn validate_public_jwk(value: &Value) -> Result<()> {
    let object = value
        .as_object()
        .ok_or_else(|| anyhow!("native device public key is invalid"))?;
    let x = object.get("x").and_then(Value::as_str).unwrap_or_default();
    if object.get("kty").and_then(Value::as_str) != Some("OKP")
        || object.get("crv").and_then(Value::as_str) != Some("Ed25519")
        || object.contains_key("d")
        || x.len() != 43
        || !x
            .bytes()
            .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'_' | b'-'))
    {
        bail!("native device public key must be a public Ed25519 JWK");
    }
    Ok(())
}

fn decode_claims(token: &str) -> Result<Value> {
    let payload = token
        .split('.')
        .nth(1)
        .ok_or_else(|| anyhow!("OpenRTC credential is malformed"))?;
    let bytes = base64::engine::general_purpose::URL_SAFE_NO_PAD
        .decode(payload)
        .context("decode OpenRTC credential claims")?;
    serde_json::from_slice(&bytes).context("parse OpenRTC credential claims")
}

fn required_claim_string(token: &str, name: &str) -> Result<String> {
    decode_claims(token)?
        .get(name)
        .and_then(Value::as_str)
        .map(str::to_string)
        .filter(|value| !value.is_empty())
        .ok_or_else(|| anyhow!("OpenRTC credential is missing {name}"))
}

fn native_device_key_thumbprint(public_key_jwk: &Value) -> Result<String> {
    validate_public_jwk(public_key_jwk)?;
    let canonical = json!({
        "crv": public_key_jwk.get("crv").and_then(Value::as_str),
        "kty": public_key_jwk.get("kty").and_then(Value::as_str),
        "x": public_key_jwk.get("x").and_then(Value::as_str),
    });
    let bytes = serde_json::to_vec(&canonical)?;
    let digest = <sha2::Sha256 as sha2::Digest>::digest(bytes);
    Ok(base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(digest))
}

fn native_session_key_hash(
    app_tag: &str,
    device_key_thumbprint: &str,
    session_key: &str,
) -> String {
    let digest = <sha2::Sha256 as sha2::Digest>::digest(
        format!(
            "openrtc:v2:session\0{}\0{}\0{}",
            app_tag, device_key_thumbprint, session_key,
        )
        .as_bytes(),
    );
    base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(digest)
}

fn validate_stored_certificate(
    value: &StoredNativeDeviceCertificate,
    device_key_thumbprint: &str,
) -> Result<()> {
    let signing_public_jwk = value
        .signing_public_jwk
        .as_ref()
        .ok_or_else(|| anyhow!("stored OpenRTC device certificate has no verification key"))?;
    verify_device_certificate_signature(&value.token, signing_public_jwk)?;
    let claims = decode_claims(&value.token)?;
    let issued_at = claims.get("iat").and_then(Value::as_u64);
    let expires_at = claims.get("exp").and_then(Value::as_u64);
    if claims.get("iss").and_then(Value::as_str) != Some("openrtc")
        || claims.get("aud").and_then(Value::as_str) != Some("openrtc:v2:gateway-grant")
        || claims.get("typ").and_then(Value::as_str) != Some("device-certificate")
        || claims.get("appTag").and_then(Value::as_str) != Some(value.app_tag.as_str())
        || claims.get("principalId").and_then(Value::as_str) != Some(value.principal_id.as_str())
        || claims.get("principalKey").and_then(Value::as_str).is_none()
        || claims.get("deviceId").and_then(Value::as_str) != Some(value.device_id.as_str())
        || claims.get("deviceKeyThumbprint").and_then(Value::as_str) != Some(device_key_thumbprint)
        || claims.get("jti").and_then(Value::as_str).is_none()
        || issued_at.is_none()
        || expires_at.and_then(|exp| exp.checked_mul(1_000)) != Some(value.expires_at_ms)
        || issued_at
            .zip(expires_at)
            .is_none_or(|(iat, exp)| iat >= exp)
    {
        bail!("stored OpenRTC device certificate scope is invalid");
    }
    Ok(())
}

fn verify_device_certificate_signature(token: &str, public_jwk: &Value) -> Result<()> {
    let mut parts = token.split('.');
    let header_segment = parts
        .next()
        .ok_or_else(|| anyhow!("invalid certificate JWT"))?;
    let claims_segment = parts
        .next()
        .ok_or_else(|| anyhow!("invalid certificate JWT"))?;
    let signature_segment = parts
        .next()
        .ok_or_else(|| anyhow!("invalid certificate JWT"))?;
    if parts.next().is_some() {
        bail!("invalid certificate JWT");
    }
    let header: Value = serde_json::from_slice(
        &base64::engine::general_purpose::URL_SAFE_NO_PAD
            .decode(header_segment)
            .context("decode certificate header")?,
    )?;
    if header.get("alg").and_then(Value::as_str) != Some("EdDSA")
        || public_jwk.get("kty").and_then(Value::as_str) != Some("OKP")
        || public_jwk.get("crv").and_then(Value::as_str) != Some("Ed25519")
        || header.get("kid").and_then(Value::as_str)
            != public_jwk.get("kid").and_then(Value::as_str)
    {
        bail!("unsupported OpenRTC certificate signing key");
    }
    let public_key = base64::engine::general_purpose::URL_SAFE_NO_PAD
        .decode(
            public_jwk
                .get("x")
                .and_then(Value::as_str)
                .ok_or_else(|| anyhow!("certificate signing key is missing x"))?,
        )
        .context("decode certificate signing key")?;
    let public_key: [u8; 32] = public_key
        .try_into()
        .map_err(|_| anyhow!("certificate signing key has invalid length"))?;
    let verifying_key =
        VerifyingKey::from_bytes(&public_key).context("parse certificate signing key")?;
    let signature = base64::engine::general_purpose::URL_SAFE_NO_PAD
        .decode(signature_segment)
        .context("decode certificate signature")?;
    let signature = Signature::from_slice(&signature).context("parse certificate signature")?;
    verifying_key
        .verify(
            format!("{header_segment}.{claims_segment}").as_bytes(),
            &signature,
        )
        .context("verify OpenRTC device certificate")
}

fn certificate_is_reusable(
    value: &StoredNativeDeviceCertificate,
    device_key_thumbprint: &str,
    at_ms: u64,
) -> bool {
    value.expires_at_ms > at_ms.saturating_add(DEVICE_CERTIFICATE_RENEW_SKEW_MS)
        && validate_stored_certificate(value, device_key_thumbprint).is_ok()
}

#[cfg(test)]
mod tests {
    use super::*;
    use ed25519_dalek::{Signer as _, SigningKey};
    use std::sync::atomic::{AtomicU64, Ordering};
    use tokio::io::{AsyncReadExt, AsyncWriteExt};
    use tokio::net::TcpListener;

    struct TestSigner;

    struct NeverGrantProvider;

    #[async_trait]
    impl NativeGatewayGrantProvider for NeverGrantProvider {
        async fn grant(&self, _request: NativeGatewayGrantRequest) -> Result<NativeGatewayGrant> {
            bail!("inactive epoch-retirement test must not request a gateway grant")
        }
    }

    #[derive(Default)]
    struct RecordingAssertionProvider {
        force_refreshes: StdMutex<Vec<bool>>,
        identity_epoch: AtomicU64,
    }

    #[async_trait]
    impl NativeIdentityAssertionProvider for RecordingAssertionProvider {
        async fn assertion(&self, force_refresh: bool) -> Result<NativeIdentityAssertion> {
            self.force_refreshes.lock().unwrap().push(force_refresh);
            Ok(NativeIdentityAssertion {
                token: if force_refresh { "fresh" } else { "stale" }.to_string(),
                provider_id: None,
            })
        }

        fn identity_epoch(&self) -> u64 {
            self.identity_epoch.load(Ordering::Acquire)
        }
    }

    impl NativeV2DeviceSigner for TestSigner {
        fn public_jwk(&self, _app_tag: &str) -> Result<Value> {
            Ok(json!({
                "kty": "OKP",
                "crv": "Ed25519",
                "x": "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA",
            }))
        }

        fn sign(&self, _app_tag: &str, _challenge: &[u8]) -> Result<Vec<u8>> {
            Ok(vec![0; 64])
        }
    }

    fn signed_certificate(
        app_tag: &str,
        principal_id: &str,
        device_id: &str,
        expires_at: u64,
    ) -> (String, Value) {
        let signing_key = SigningKey::from_bytes(&[7_u8; 32]);
        let kid = "test-signing-key";
        let header = base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(
            serde_json::to_vec(&json!({ "alg": "EdDSA", "typ": "JWT", "kid": kid })).unwrap(),
        );
        let claims = base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(
            serde_json::to_vec(&json!({
                "iss": "openrtc",
                "aud": "openrtc:v2:gateway-grant",
                "typ": "device-certificate",
                "appTag": app_tag,
                "principalId": principal_id,
                "principalKey": "principal-key",
                "deviceId": device_id,
                "deviceKeyThumbprint": "thumbprint",
                "iat": expires_at - 60,
                "exp": expires_at,
                "jti": "test-jti",
            }))
            .unwrap(),
        );
        let signing_input = format!("{header}.{claims}");
        let signature = signing_key.sign(signing_input.as_bytes());
        let token = format!(
            "{signing_input}.{}",
            base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(signature.to_bytes())
        );
        let public_jwk = json!({
            "kty": "OKP",
            "crv": "Ed25519",
            "alg": "EdDSA",
            "kid": kid,
            "x": base64::engine::general_purpose::URL_SAFE_NO_PAD
                .encode(signing_key.verifying_key().to_bytes()),
        });
        (token, public_jwk)
    }

    #[test]
    fn cached_certificate_is_scope_bound_and_renews_a_day_early() {
        let now = now_ms();
        let expires_at = now / 1_000 + 2 * 24 * 60 * 60;
        let (token, public_jwk) =
            signed_certificate("app_0000000000000000", "principal", "device", expires_at);
        let value = StoredNativeDeviceCertificate {
            app_tag: "app_0000000000000000".to_string(),
            principal_id: "principal".to_string(),
            device_id: "device".to_string(),
            token,
            expires_at_ms: expires_at * 1_000,
            session_key_hash: Some("session-hash".to_string()),
            signing_public_jwk: Some(public_jwk),
        };
        assert!(certificate_is_reusable(&value, "thumbprint", now));
        assert!(!certificate_is_reusable(
            &value,
            "thumbprint",
            value.expires_at_ms - DEVICE_CERTIFICATE_RENEW_SKEW_MS + 1,
        ));

        let mut wrong = value.clone();
        wrong.device_id = "other-device".to_string();
        assert!(!certificate_is_reusable(&wrong, "thumbprint", now));
        assert!(!certificate_is_reusable(&value, "rotated-thumbprint", now));

        let mut tampered = value.clone();
        tampered.principal_id = "attacker".to_string();
        assert!(!certificate_is_reusable(&tampered, "thumbprint", now));

        let mut missing_key = value.clone();
        missing_key.signing_public_jwk = None;
        assert!(!certificate_is_reusable(&missing_key, "thumbprint", now));
    }

    #[test]
    fn device_public_jwk_rejects_private_material() {
        assert!(validate_public_jwk(&json!({
            "kty": "OKP",
            "crv": "Ed25519",
            "x": "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA",
        }))
        .is_ok());
        assert!(validate_public_jwk(&json!({
            "kty": "OKP",
            "crv": "Ed25519",
            "x": "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA",
            "d": "secret",
        }))
        .is_err());
    }

    #[tokio::test]
    async fn anonymous_constructor_is_network_idle_and_rejects_authenticated_devices() {
        let control = NativeV2ControlPlane::anonymous(
            crate::test_constants::TEST_API_KEY,
            Arc::new(TestSigner),
        )
        .expect("anonymous v2 control plane");
        assert_eq!(
            control.app_tag(),
            crate::app_tag_from_api_key(crate::test_constants::TEST_API_KEY),
        );
        let error = control
            .device_certificate("device", None, 0)
            .await
            .expect_err("anonymous clients do not own consumer authentication");
        assert!(error.to_string().contains("identity assertion provider"));
    }

    #[tokio::test]
    async fn control_plane_retry_reuses_the_exact_body_and_idempotency_key() {
        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
        let address = listener.local_addr().unwrap();
        let observed = Arc::new(Mutex::new(Vec::<(String, Vec<u8>)>::new()));
        let server_observed = observed.clone();
        let server = tokio::spawn(async move {
            for attempt in 0..2 {
                let (mut stream, _) = listener.accept().await.unwrap();
                let mut bytes = Vec::new();
                let mut buffer = [0_u8; 4096];
                let header_end = loop {
                    let read = stream.read(&mut buffer).await.unwrap();
                    assert!(read > 0);
                    bytes.extend_from_slice(&buffer[..read]);
                    if let Some(index) = bytes.windows(4).position(|part| part == b"\r\n\r\n") {
                        break index + 4;
                    }
                };
                let headers = String::from_utf8_lossy(&bytes[..header_end]).to_string();
                let content_length = headers
                    .lines()
                    .find_map(|line| {
                        let (name, value) = line.split_once(':')?;
                        name.eq_ignore_ascii_case("content-length")
                            .then(|| value.trim().parse::<usize>().unwrap())
                    })
                    .unwrap_or(0);
                while bytes.len() < header_end + content_length {
                    let read = stream.read(&mut buffer).await.unwrap();
                    assert!(read > 0);
                    bytes.extend_from_slice(&buffer[..read]);
                }
                let request_id = headers
                    .lines()
                    .find_map(|line| {
                        let (name, value) = line.split_once(':')?;
                        name.eq_ignore_ascii_case("x-openrtc-idempotency-key")
                            .then(|| value.trim().to_string())
                    })
                    .expect("idempotency header");
                server_observed.lock().await.push((
                    request_id,
                    bytes[header_end..header_end + content_length].to_vec(),
                ));

                let (status, body) = if attempt == 0 {
                    ("503 Service Unavailable", r#"{"error":"retry"}"#)
                } else {
                    ("200 OK", r#"{"ok":true}"#)
                };
                stream
                    .write_all(
                        format!(
                            "HTTP/1.1 {status}\r\ncontent-type: application/json\r\ncontent-length: {}\r\nconnection: close\r\n\r\n{body}",
                            body.len(),
                        )
                        .as_bytes(),
                    )
                    .await
                    .unwrap();
            }
        });

        let control = NativeV2ControlPlane::anonymous(
            crate::test_constants::TEST_API_KEY,
            Arc::new(TestSigner),
        )
        .unwrap()
        .with_testing_endpoints(format!("http://{address}"), "http://127.0.0.1:1")
        .unwrap();
        let response: Value = control
            .post("/retry", json!({ "intent": "same" }))
            .await
            .unwrap();
        assert_eq!(response, json!({ "ok": true }));
        server.await.unwrap();

        let observed = observed.lock().await;
        assert_eq!(observed.len(), 2);
        assert_eq!(observed[0], observed[1]);
    }

    #[tokio::test]
    async fn native_identity_refreshes_once_only_after_unauthorized() {
        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
        let address = listener.local_addr().unwrap();
        let server = tokio::spawn(async move {
            for attempt in 0..2 {
                let (mut stream, _) = listener.accept().await.unwrap();
                let mut bytes = Vec::new();
                let mut buffer = [0_u8; 4096];
                let header_end = loop {
                    let read = stream.read(&mut buffer).await.unwrap();
                    assert!(read > 0);
                    bytes.extend_from_slice(&buffer[..read]);
                    if let Some(index) = bytes.windows(4).position(|part| part == b"\r\n\r\n") {
                        break index + 4;
                    }
                };
                let headers = String::from_utf8_lossy(&bytes[..header_end]);
                let content_length = headers
                    .lines()
                    .find_map(|line| {
                        let (name, value) = line.split_once(':')?;
                        name.eq_ignore_ascii_case("content-length")
                            .then(|| value.trim().parse::<usize>().unwrap())
                    })
                    .unwrap_or(0);
                while bytes.len() < header_end + content_length {
                    let read = stream.read(&mut buffer).await.unwrap();
                    assert!(read > 0);
                    bytes.extend_from_slice(&buffer[..read]);
                }
                let request_body: Value =
                    serde_json::from_slice(&bytes[header_end..header_end + content_length])
                        .unwrap();
                let expected_assertion = if attempt == 0 { "stale" } else { "fresh" };
                assert_eq!(
                    request_body.get("assertion").and_then(Value::as_str),
                    Some(expected_assertion),
                );
                let (status, body) = if attempt == 0 {
                    ("401 Unauthorized", r#"{"error":"assertion rejected"}"#)
                } else {
                    (
                        "200 OK",
                        r#"{"identitySession":"identity","principalId":"principal","expiresAt":4102444800}"#,
                    )
                };
                stream
                    .write_all(
                        format!(
                            "HTTP/1.1 {status}\r\ncontent-type: application/json\r\ncontent-length: {}\r\nconnection: close\r\n\r\n{body}",
                            body.len(),
                        )
                        .as_bytes(),
                    )
                    .await
                    .unwrap();
            }
        });

        let provider = Arc::new(RecordingAssertionProvider::default());
        let control = NativeV2ControlPlane::new(
            crate::test_constants::TEST_API_KEY,
            provider.clone(),
            Arc::new(TestSigner),
            Arc::new(MemoryNativeDeviceCertificateStore::default()),
        )
        .unwrap()
        .with_testing_endpoints(format!("http://{address}"), "http://127.0.0.1:1")
        .unwrap();

        let identity = control
            .exchange_identity_after_rejection(false)
            .await
            .unwrap();
        assert_eq!(identity.principal_id, "principal");
        assert_eq!(*provider.force_refreshes.lock().unwrap(), vec![false, true]);
        server.await.unwrap();
    }

    #[tokio::test]
    async fn native_identity_exchange_is_coalesced_per_host_login_epoch() {
        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
        let address = listener.local_addr().unwrap();
        let server = tokio::spawn(async move {
            for attempt in 0..2 {
                let (mut stream, _) = listener.accept().await.unwrap();
                let mut bytes = Vec::new();
                let mut buffer = [0_u8; 4096];
                let header_end = loop {
                    let read = stream.read(&mut buffer).await.unwrap();
                    assert!(read > 0);
                    bytes.extend_from_slice(&buffer[..read]);
                    if let Some(index) = bytes.windows(4).position(|part| part == b"\r\n\r\n") {
                        break index + 4;
                    }
                };
                let headers = String::from_utf8_lossy(&bytes[..header_end]);
                let content_length = headers
                    .lines()
                    .find_map(|line| {
                        let (name, value) = line.split_once(':')?;
                        name.eq_ignore_ascii_case("content-length")
                            .then(|| value.trim().parse::<usize>().unwrap())
                    })
                    .unwrap_or(0);
                while bytes.len() < header_end + content_length {
                    let read = stream.read(&mut buffer).await.unwrap();
                    assert!(read > 0);
                    bytes.extend_from_slice(&buffer[..read]);
                }
                let body = format!(
                    r#"{{"identitySession":"identity-{attempt}","principalId":"principal-{attempt}","expiresAt":4102444800}}"#,
                );
                stream
                    .write_all(
                        format!(
                            "HTTP/1.1 200 OK\r\ncontent-type: application/json\r\ncontent-length: {}\r\nconnection: close\r\n\r\n{body}",
                            body.len(),
                        )
                        .as_bytes(),
                    )
                    .await
                    .unwrap();
            }
        });

        let provider = Arc::new(RecordingAssertionProvider::default());
        let control = NativeV2ControlPlane::new(
            crate::test_constants::TEST_API_KEY,
            provider.clone(),
            Arc::new(TestSigner),
            Arc::new(MemoryNativeDeviceCertificateStore::default()),
        )
        .unwrap()
        .with_testing_endpoints(format!("http://{address}"), "http://127.0.0.1:1")
        .unwrap();

        let first = control.identity_for_epoch(0, false).await.unwrap();
        let reused = control.identity_for_epoch(0, false).await.unwrap();
        assert_eq!(first.response.principal_id, "principal-0");
        assert_eq!(reused.response.principal_id, "principal-0");
        assert_eq!(*provider.force_refreshes.lock().unwrap(), vec![false]);

        provider.identity_epoch.store(1, Ordering::Release);
        let next = control.identity_for_epoch(1, false).await.unwrap();
        assert_eq!(next.response.principal_id, "principal-1");
        assert_eq!(
            *provider.force_refreshes.lock().unwrap(),
            vec![false, false]
        );
        server.await.unwrap();
    }

    #[tokio::test]
    async fn native_login_epoch_change_retires_the_old_capability_without_network_work() {
        let capabilities = NativeCapabilities::new(
            crate::test_constants::TEST_API_KEY,
            "device",
            "native",
            Arc::new(NeverGrantProvider),
        )
        .unwrap();
        let handle = capabilities.devices("principal").unwrap();
        let relay = NativeV2IdentityCredentialRelay::default();
        relay.set(Some("device-certificate".to_string())).unwrap();
        let credential = relay.provider();
        let (epoch_sender, epoch_receiver) = watch::channel(7_u64);
        let monitor = spawn_identity_epoch_monitor(epoch_receiver, 7, handle.closer(), relay);

        epoch_sender.send(8).unwrap();
        tokio::time::timeout(std::time::Duration::from_secs(1), monitor)
            .await
            .expect("epoch retirement must be prompt")
            .unwrap();
        assert!(handle.is_closed());
        assert_eq!(credential(), None);
    }

    #[test]
    fn native_space_namespace_matches_browser_v2_derivation() {
        let api_key = crate::test_constants::TEST_API_KEY;
        let requested = "portfolio-cursors";
        let expected = hex::encode(<sha2::Sha256 as sha2::Digest>::digest(
            format!("{api_key}:{requested}").as_bytes(),
        ));
        assert_eq!(
            native_capability_avenue_id(api_key, NativeV2AnonymousKind::Space, requested),
            expected,
        );
        assert_eq!(
            native_capability_avenue_id(api_key, NativeV2AnonymousKind::Room, "match-123"),
            "match-123",
        );
    }

    #[tokio::test]
    async fn signaling_slot_installs_once_and_fails_closed_before_activation() {
        let slot = NativeV2SignalingSlot::default();
        assert!(slot.search_devices("principal", None).await.is_err());
        slot.install(Arc::new(crate::signaling::GatewayRequiredSignalingBackend))
            .await
            .expect("first capability installs");
        assert!(slot
            .install(Arc::new(crate::signaling::GatewayRequiredSignalingBackend,))
            .await
            .is_err());
    }
}