host_sso/

manager.rs

//! `SsoManager` — coordinator for the SSO QR-pairing lifecycle.
//!
//! All I/O is delegated to injected trait adapters (`SsoTransport`,
//! `SsoSigner`, `SsoSessionStore`, `SsoEventSink`, `SsoProductKeyStore`) so
//! the crate has no direct I/O dependencies and is straightforward to unit-test.

use std::sync::atomic::{AtomicBool, AtomicI64, Ordering};
use std::sync::{Arc, Mutex};

use crate::{
    error::SsoError,
    session::PairingResult,
    state::SsoState,
    traits::{
        NoopProductKeyStore, PersistedSessionMeta, SsoEventSink, SsoProductKeyStore,
        SsoSessionStore, SsoSigner, SsoTransport,
    },
};

#[cfg(not(target_arch = "wasm32"))]
use crate::presence;
#[cfg(not(target_arch = "wasm32"))]
use crate::product_key_cache::ProductKeyCache;

// ---------------------------------------------------------------------------
// Timeout constant
// ---------------------------------------------------------------------------

/// Timeout for a product key request. A const so tests can read it.
#[cfg(not(target_arch = "wasm32"))]
pub const PRODUCT_KEY_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(30);

// ---------------------------------------------------------------------------
// Sign material
// ---------------------------------------------------------------------------

/// Session key material required to forward sign/product-key requests to the
/// paired phone.
///
/// Stored in `SsoManager` after a successful pairing and zeroized on unpair.
#[cfg(not(target_arch = "wasm32"))]
struct SignMaterial {
    /// AES-128 session key negotiated during P-256 ECDH pairing.
    session_key: zeroize::Zeroizing<[u8; 16]>,
    /// 32-byte sr25519 account ID (`//wallet//sso`) used as HMAC key.
    identity_pubkey: [u8; 32],
    /// 65-byte uncompressed P-256 public key from the pairing handshake.
    p256_pubkey: [u8; 65],
}

// ---------------------------------------------------------------------------
// SsoManager
// ---------------------------------------------------------------------------

/// Central coordinator for the SSO pairing lifecycle.
///
/// Generic over five injected adapters so hosts provide platform-specific
/// implementations without coupling this crate to any I/O runtime.
///
/// The fifth generic `Pk` defaults to [`NoopProductKeyStore`], so existing
/// callers constructed via [`SsoManager::new`] do not need to change.
///
/// All shared fields are wrapped in `Arc` so the background pairing thread
/// can update state and notify the sink without requiring `Arc<SsoManager>`.
pub struct SsoManager<T, Si, St, E, Pk = NoopProductKeyStore>
where
    T: SsoTransport + 'static,
    Si: SsoSigner,
    St: SsoSessionStore + 'static,
    E: SsoEventSink + 'static,
    Pk: SsoProductKeyStore + 'static,
{
    /// Current observable state; shared with the background pairing thread.
    state: Arc<Mutex<SsoState>>,
    /// Transport adapter shared with the pairing closures (FnOnce captures Arc).
    transport: Arc<T>,
    signer: Si,
    /// Session store shared with the background pairing thread.
    store: Arc<St>,
    /// Event sink shared with the background pairing thread.
    sink: Arc<E>,
    /// Product key persistence adapter.
    product_key_store: Arc<Pk>,
    /// Metadata URL embedded in the QR handshake (may be empty).
    metadata_url: String,
    /// Guards against launching two concurrent pairing attempts.
    pairing_in_progress: Arc<AtomicBool>,
    /// Monotonic counter incremented on each pair() call. The polling thread
    /// checks its generation matches the current one before mutating state,
    /// preventing a stale thread from corrupting a newer pairing session.
    #[cfg(not(target_arch = "wasm32"))]
    pairing_generation: Arc<std::sync::atomic::AtomicU64>,
    /// Cancel flag forwarded to the active `PairingSession`; `unpair()` sets it.
    #[cfg(not(target_arch = "wasm32"))]
    pairing_cancel: Mutex<Option<Arc<AtomicBool>>>,
    /// Session key material for forwarding sign requests to the paired phone.
    /// Set by `handle_pairing_result`, cleared by `unpair`.
    #[cfg(not(target_arch = "wasm32"))]
    sign_material: Mutex<Option<SignMaterial>>,
    /// In-memory product key cache, bound to the current phone identity.
    /// `None` until a successful pairing establishes a phone identity.
    #[cfg(not(target_arch = "wasm32"))]
    product_key_cache: Mutex<Option<ProductKeyCache>>,
    /// Cancel flag for the background presence monitor thread.
    /// Set to `true` by `unpair()` to stop the monitor.
    #[cfg(not(target_arch = "wasm32"))]
    presence_cancel: Mutex<Option<Arc<AtomicBool>>>,
    /// Unix timestamp (seconds) of the last heartbeat received from the phone.
    /// `0` means no heartbeat has been received; reads need no mutex.
    #[cfg(not(target_arch = "wasm32"))]
    last_heartbeat_ts: Arc<AtomicI64>,
}

// ---------------------------------------------------------------------------
// Constructor: new() — only available when Pk = NoopProductKeyStore
// ---------------------------------------------------------------------------

impl<T, Si, St, E> SsoManager<T, Si, St, E, NoopProductKeyStore>
where
    T: SsoTransport + 'static,
    Si: SsoSigner,
    St: SsoSessionStore + 'static,
    E: SsoEventSink + 'static,
{
    /// Construct a new manager starting in the `Idle` state with the default
    /// no-op product key store.
    ///
    /// If you need product key caching with persistence, use
    /// [`SsoManager::with_product_key_store`] instead.
    pub fn new(transport: T, signer: Si, store: St, sink: E, metadata_url: String) -> Self {
        Self::with_product_key_store(
            transport,
            signer,
            store,
            sink,
            NoopProductKeyStore,
            metadata_url,
        )
    }
}

// ---------------------------------------------------------------------------
// Constructor: with_product_key_store() — available for all Pk
// ---------------------------------------------------------------------------

impl<T, Si, St, E, Pk> SsoManager<T, Si, St, E, Pk>
where
    T: SsoTransport + 'static,
    Si: SsoSigner,
    St: SsoSessionStore + 'static,
    E: SsoEventSink + 'static,
    Pk: SsoProductKeyStore + 'static,
{
    /// Construct a new manager with a custom product key store.
    ///
    /// The manager starts in the `Idle` state.
    pub fn with_product_key_store(
        transport: T,
        signer: Si,
        store: St,
        sink: E,
        product_key_store: Pk,
        metadata_url: String,
    ) -> Self {
        Self {
            state: Arc::new(Mutex::new(SsoState::Idle)),
            transport: Arc::new(transport),
            signer,
            store: Arc::new(store),
            sink: Arc::new(sink),
            product_key_store: Arc::new(product_key_store),
            metadata_url,
            pairing_in_progress: Arc::new(AtomicBool::new(false)),
            #[cfg(not(target_arch = "wasm32"))]
            pairing_generation: Arc::new(std::sync::atomic::AtomicU64::new(0)),
            #[cfg(not(target_arch = "wasm32"))]
            pairing_cancel: Mutex::new(None),
            #[cfg(not(target_arch = "wasm32"))]
            sign_material: Mutex::new(None),
            #[cfg(not(target_arch = "wasm32"))]
            product_key_cache: Mutex::new(None),
            #[cfg(not(target_arch = "wasm32"))]
            presence_cancel: Mutex::new(None),
            #[cfg(not(target_arch = "wasm32"))]
            last_heartbeat_ts: Arc::new(AtomicI64::new(0)),
        }
    }
}

// ---------------------------------------------------------------------------
// Public API — general impl block for all Pk
// ---------------------------------------------------------------------------

impl<T, Si, St, E, Pk> SsoManager<T, Si, St, E, Pk>
where
    T: SsoTransport + 'static,
    Si: SsoSigner,
    St: SsoSessionStore + 'static,
    E: SsoEventSink + 'static,
    Pk: SsoProductKeyStore + 'static,
{
    /// Return a snapshot of the current state.
    pub fn state(&self) -> SsoState {
        self.state.lock().unwrap_or_else(|e| e.into_inner()).clone()
    }

    /// Initiate a QR-pairing session in a background thread.
    ///
    /// Transitions: `Idle` / `Failed` → `AwaitingScan` (when QR is ready) →
    /// `Paired` on success, or `Failed` on error or cancellation.
    ///
    /// Returns `Err(PairingAlreadyInProgress)` if a pairing is already running.
    #[cfg(not(target_arch = "wasm32"))]
    pub fn pair(&self) -> Result<(), SsoError> {
        // Reject duplicate pairing attempts atomically.
        if self
            .pairing_in_progress
            .compare_exchange(false, true, Ordering::AcqRel, Ordering::Acquire)
            .is_err()
        {
            return Err(SsoError::PairingAlreadyInProgress);
        }

        let identity_pubkey = self
            .signer
            .public_key()
            .map_err(|e| SsoError::SignerError(e.to_string()))?;
        let metadata_url = self.metadata_url.clone();

        // Each PairingConfig closure captures an Arc clone of the transport so
        // the FnOnce can call &self methods after the manager is potentially dropped.
        let transport_sub = Arc::clone(&self.transport);
        let transport_unsub = Arc::clone(&self.transport);

        let config = host_wallet::pairing::PairingConfig {
            subscribe: Box::new(move |topic| transport_sub.subscribe(topic)),
            unsubscribe: Box::new(move |id| transport_unsub.unsubscribe(id)),
            identity_pubkey,
            metadata_url,
        };

        // start_pairing spawns its own handshake thread internally and returns
        // immediately with a channel receiver for state updates.
        let session = host_wallet::pairing::start_pairing(config);

        // Store the cancel flag so unpair() can abort in-flight pairing.
        *self
            .pairing_cancel
            .lock()
            .unwrap_or_else(|e| e.into_inner()) = Some(Arc::clone(&session.cancel));

        // Increment generation so stale polling threads discard their events.
        let generation = self.pairing_generation.fetch_add(1, Ordering::AcqRel) + 1;

        // Capture shared state for the polling thread.
        let state_arc = Arc::clone(&self.state);
        let sink_arc = Arc::clone(&self.sink);
        let in_progress = Arc::clone(&self.pairing_in_progress);
        let gen_arc = Arc::clone(&self.pairing_generation);
        let state_rx = session.state_rx;

        log::debug!("host-sso: pair() called — pairing thread started (gen={generation})");

        // Spawn a lightweight thread that drains state_rx and applies transitions.
        std::thread::spawn(move || {
            poll_pairing_states(
                state_rx,
                state_arc,
                sink_arc,
                in_progress,
                gen_arc,
                generation,
            );
        });

        Ok(())
    }

    /// Clear the current session and return to the `Idle` state.
    ///
    /// If a pairing is in progress it is cancelled before clearing.
    /// Safe to call from any state, including `Idle`.
    pub fn unpair(&self) -> Result<(), SsoError> {
        // Cancel any in-flight pairing handshake.
        #[cfg(not(target_arch = "wasm32"))]
        if let Some(cancel) = self
            .pairing_cancel
            .lock()
            .unwrap_or_else(|e| e.into_inner())
            .take()
        {
            cancel.store(true, Ordering::Release);
        }

        // Zeroize sign material so the AES session key is overwritten in memory.
        #[cfg(not(target_arch = "wasm32"))]
        {
            *self.sign_material.lock().unwrap_or_else(|e| e.into_inner()) = None;
        }

        // Clear the in-memory product key cache and remove persisted cache.
        #[cfg(not(target_arch = "wasm32"))]
        {
            *self
                .product_key_cache
                .lock()
                .unwrap_or_else(|e| e.into_inner()) = None;
        }
        self.product_key_store.delete().map_err(SsoError::Store)?;

        // Stop the presence monitor and reset the heartbeat timestamp.
        #[cfg(not(target_arch = "wasm32"))]
        {
            if let Some(cancel) = self
                .presence_cancel
                .lock()
                .unwrap_or_else(|e| e.into_inner())
                .take()
            {
                cancel.store(true, Ordering::Release);
            }
            self.last_heartbeat_ts.store(0, Ordering::Release);
        }

        self.store.clear().map_err(SsoError::Store)?;
        self.pairing_in_progress.store(false, Ordering::Release);
        self.transition(&self.state, &self.sink, SsoState::Idle);
        log::debug!("host-sso: unpaired — session cleared");
        Ok(())
    }

    /// Load a previously persisted session from the store.
    ///
    /// On success, transitions to `Paired`. If no session is stored, stays
    /// `Idle`. Returns `Err(Store(_))` if the store call itself fails.
    pub fn restore_session(&self) -> Result<(), SsoError> {
        match self.store.load().map_err(SsoError::Store)? {
            Some(meta) => {
                log::debug!("host-sso: restored session for address={}", meta.address);
                self.transition(
                    &self.state,
                    &self.sink,
                    SsoState::Paired {
                        address: meta.address,
                        display_name: meta.display_name,
                        phone_online: false,
                    },
                );
                Ok(())
            }
            None => {
                log::debug!("host-sso: no persisted session found");
                Ok(())
            }
        }
    }

    /// Finalise a completed pairing handshake.
    ///
    /// Persists the session metadata, stores sign material for future
    /// `request_sign` calls, initialises the product key cache with the phone
    /// identity (loading any persisted entries), clears the in-progress guard,
    /// and transitions to `Paired`. The `PairingResult` is consumed and its
    /// `session_key` is zeroized on drop — callers must not use it after this call.
    pub fn handle_pairing_result(&self, result: PairingResult) -> Result<(), SsoError> {
        // Cache sign material before the result fields are moved into the ctx.
        #[cfg(not(target_arch = "wasm32"))]
        {
            let material = SignMaterial {
                session_key: zeroize::Zeroizing::new(*result.session_key),
                identity_pubkey: result.identity_pubkey,
                p256_pubkey: result.p256_pubkey,
            };
            *self.sign_material.lock().unwrap_or_else(|e| e.into_inner()) = Some(material);
        }

        // Initialise the product key cache with the phone wallet identity.
        #[cfg(not(target_arch = "wasm32"))]
        if let Some(phone_pubkey) = result.phone_wallet_pubkey {
            self.init_product_key_cache(phone_pubkey);
        }

        let phone_wallet_pubkey_hex = result.phone_wallet_pubkey.map(|pk| {
            pk.iter().fold(String::with_capacity(64), |mut s, b| {
                use std::fmt::Write as _;
                let _ = write!(s, "{b:02x}");
                s
            })
        });

        handle_established(EstablishedCtx {
            address: result.address,
            display_name: result.display_name,
            identity_pubkey: result.identity_pubkey,
            p256_pubkey: result.p256_pubkey,
            phone_wallet_pubkey_hex,
            store: &self.store,
            in_progress: &self.pairing_in_progress,
            state: &self.state,
            sink: &self.sink,
        })
    }

    /// Send a sign request to the paired phone and block until response or timeout.
    ///
    /// Must be called from a background thread (blocking I/O).
    /// Returns the signature bytes on success.
    ///
    /// Returns `Err(NotPaired)` if no paired session exists, or `Err(SignFailed)`
    /// if the transport or phone returns an error.
    #[cfg(not(target_arch = "wasm32"))]
    pub fn request_sign(&self, payload: &[u8]) -> Result<Vec<u8>, SsoError> {
        // Guard: must be in Paired state.
        match self.state() {
            SsoState::Paired { .. } => {}
            _ => return Err(SsoError::NotPaired),
        }

        // Extract session material under the lock, then release immediately.
        // The state check and material extraction are separate critical sections;
        // unpair() could clear sign_material between them, which is handled by
        // the None arm returning NotPaired.
        let (session_key, identity_pubkey, p256_pubkey) = {
            let guard = self.sign_material.lock().unwrap_or_else(|e| e.into_inner());
            match guard.as_ref() {
                Some(m) => (
                    zeroize::Zeroizing::new(*m.session_key),
                    m.identity_pubkey,
                    m.p256_pubkey,
                ),
                None => return Err(SsoError::NotPaired),
            }
        };

        let transport_write = Arc::clone(&self.transport);
        let transport_sub = Arc::clone(&self.transport);
        let transport_unsub = Arc::clone(&self.transport);

        let config = host_wallet::pairing::PairedSignConfig {
            write: Box::new(move |topic, value| transport_write.write(topic, value)),
            subscribe: Box::new(move |topic| transport_sub.subscribe(topic)),
            unsubscribe: Box::new(move |id| transport_unsub.unsubscribe(id)),
        };

        let (result_tx, result_rx) = std::sync::mpsc::sync_channel(1);

        host_wallet::pairing::sign_via_paired(
            payload.to_vec(),
            *session_key,
            identity_pubkey,
            p256_pubkey,
            config,
            result_tx,
        );

        result_rx
            .recv()
            .map_err(|_| SsoError::SignFailed("result channel closed".to_string()))?
            .map_err(SsoError::SignFailed)
    }

    /// Request a product public key from the paired phone.
    ///
    /// Checks the in-memory cache first (no network round-trip on hit). On a
    /// cache miss, verifies the phone is online before sending a
    /// `ProductKeyRequest` over the encrypted channel and waiting up to 30
    /// seconds for the phone's response.
    ///
    /// Returns `Err(NotPaired)` if no paired session exists.
    /// Returns `Err(PhoneOffline)` if the phone has not sent a heartbeat
    ///   within [`presence::HEARTBEAT_TIMEOUT_SECS`] (cache misses only).
    /// Returns `Err(ProductKeyCapabilityAbsent)` if the phone does not advertise
    ///   the `"product_key"` capability.
    /// Returns `Err(ProductKeyRejected)` if the phone denies the request.
    /// Returns `Err(ProductKeyTimeout)` if no response arrives within 30 s.
    #[cfg(not(target_arch = "wasm32"))]
    pub fn request_product_key(&self, product_id: &str, index: u32) -> Result<[u8; 32], SsoError> {
        // Guard: must be in Paired state.
        match self.state() {
            SsoState::Paired { .. } => {}
            _ => return Err(SsoError::NotPaired),
        }

        // Single store load: check capability and extract phone pubkey together.
        let phone_pubkey = self.load_product_key_prereqs()?;
        if let Some(cached) = self.cache_get(phone_pubkey, product_id, index) {
            return Ok(cached);
        }

        // Best-effort guard: the phone could go offline between here and the
        // network round-trip. If it does, the request times out after PRODUCT_KEY_TIMEOUT.
        if !self.is_phone_online() {
            return Err(SsoError::PhoneOffline);
        }

        // Send request to the phone.
        let pubkey = self.send_product_key_request(phone_pubkey, product_id, index)?;

        // Insert into cache and persist.
        self.cache_insert_and_persist(phone_pubkey, product_id, index, pubkey);

        Ok(pubkey)
    }

    // -----------------------------------------------------------------------
    // Private helpers
    // -----------------------------------------------------------------------

    /// Update the internal state and notify the event sink.
    fn transition(&self, state: &Arc<Mutex<SsoState>>, sink: &Arc<E>, new_state: SsoState) {
        transition_state(state, sink, new_state);
    }

    /// Load session meta once and extract both capability check and phone pubkey.
    /// Avoids double store load (review finding C-1).
    #[cfg(not(target_arch = "wasm32"))]
    fn load_product_key_prereqs(&self) -> Result<[u8; 32], SsoError> {
        let meta = self
            .store
            .load()
            .map_err(SsoError::Store)?
            .ok_or(SsoError::NotPaired)?;

        if !meta.capabilities.iter().any(|c| c == "product_key") {
            return Err(SsoError::ProductKeyCapabilityAbsent);
        }

        let hex = meta.phone_wallet_pubkey_hex.ok_or(SsoError::NotPaired)?;
        let bytes =
            hex::decode(&hex).map_err(|e| SsoError::Store(format!("bad pubkey hex: {e}")))?;
        bytes
            .try_into()
            .map_err(|_| SsoError::Store("phone_wallet_pubkey_hex is not 32 bytes".to_string()))
    }

    /// Try to get a product key from the in-memory cache.
    #[cfg(not(target_arch = "wasm32"))]
    fn cache_get(&self, phone_pubkey: [u8; 32], product_id: &str, index: u32) -> Option<[u8; 32]> {
        let mut guard = self
            .product_key_cache
            .lock()
            .unwrap_or_else(|e| e.into_inner());
        guard
            .as_mut()
            .and_then(|cache| cache.get(&phone_pubkey, product_id, index))
    }

    /// Insert a product key into the cache and persist the snapshot.
    #[cfg(not(target_arch = "wasm32"))]
    fn cache_insert_and_persist(
        &self,
        phone_pubkey: [u8; 32],
        product_id: &str,
        index: u32,
        pubkey: [u8; 32],
    ) {
        let mut guard = self
            .product_key_cache
            .lock()
            .unwrap_or_else(|e| e.into_inner());
        if let Some(cache) = guard.as_mut() {
            if let Err(e) = cache.insert(&phone_pubkey, product_id, index, pubkey) {
                // Identity mismatch after a concurrent re-pair — log and skip persist.
                log::warn!("host-sso: product key cache insert skipped: {e}");
                return;
            }
            let entries = cache.to_entries();
            drop(guard); // release lock before I/O
            if let Err(e) = self.product_key_store.save(&phone_pubkey, &entries) {
                log::warn!("host-sso: product key cache persist failed: {e}");
            }
        }
    }

    /// Initialise (or reinitialise) the product key cache for `phone_identity`.
    ///
    /// Loads any matching persisted entries from the store.
    #[cfg(not(target_arch = "wasm32"))]
    fn init_product_key_cache(&self, phone_identity: [u8; 32]) {
        let mut cache = ProductKeyCache::new(phone_identity);

        // Load persisted entries for this identity, if any.
        match self.product_key_store.load() {
            Ok(Some((stored_identity, entries))) => {
                cache.load_from(&stored_identity, &entries);
            }
            Ok(None) => {}
            Err(e) => {
                log::warn!("host-sso: failed to load persisted product key cache: {e}");
            }
        }

        *self
            .product_key_cache
            .lock()
            .unwrap_or_else(|e| e.into_inner()) = Some(cache);
    }

    /// Send a `ProductKeyRequest` over the encrypted channel and await the response.
    #[cfg(not(target_arch = "wasm32"))]
    fn send_product_key_request(
        &self,
        phone_pubkey: [u8; 32],
        product_id: &str,
        index: u32,
    ) -> Result<[u8; 32], SsoError> {
        let (session_key, identity_pubkey, p256_pubkey) = self.extract_sign_material()?;

        let message_id = derive_message_id(&phone_pubkey, product_id, index);
        let req = crate::product_key::ProductKeyRequest {
            message_id,
            product_id: product_id.to_string(),
            index,
        };
        let req_bytes = req.encode().map_err(SsoError::SignFailed)?;

        let transport_write = Arc::clone(&self.transport);
        let transport_sub = Arc::clone(&self.transport);
        let transport_unsub = Arc::clone(&self.transport);

        let config = host_wallet::pairing::PairedSignConfig {
            write: Box::new(move |topic, value| transport_write.write(topic, value)),
            subscribe: Box::new(move |topic| transport_sub.subscribe(topic)),
            unsubscribe: Box::new(move |id| transport_unsub.unsubscribe(id)),
        };

        let (result_tx, result_rx) = std::sync::mpsc::sync_channel::<Result<Vec<u8>, String>>(1);

        // Re-use sign_via_paired: the encrypted response bytes are the SCALE-encoded
        // ProductKeyResponse. The phone must decrypt our request and encrypt its response
        // using the same session key and channel derivation.
        host_wallet::pairing::sign_via_paired(
            req_bytes,
            *session_key,
            identity_pubkey,
            p256_pubkey,
            config,
            result_tx,
        );

        let raw = result_rx
            .recv_timeout(PRODUCT_KEY_TIMEOUT)
            .map_err(|_| SsoError::ProductKeyTimeout)?
            .map_err(|e| {
                if e.contains("timed out") {
                    SsoError::ProductKeyTimeout
                } else {
                    SsoError::SignFailed(e)
                }
            })?;

        let response = crate::product_key::ProductKeyResponse::decode(&raw)
            .map_err(|e| SsoError::SignFailed(format!("ProductKeyResponse decode failed: {e}")))?;

        match response.payload {
            crate::product_key::ProductKeyPayload::Ok(pubkey) => Ok(pubkey),
            crate::product_key::ProductKeyPayload::Unauthorized(reason) => {
                Err(SsoError::ProductKeyRejected(reason))
            }
            crate::product_key::ProductKeyPayload::DerivationError(reason) => {
                Err(SsoError::ProductKeyRejected(reason))
            }
        }
    }

    /// Extract sign material from the mutex. Returns `Err(NotPaired)` if absent.
    #[cfg(not(target_arch = "wasm32"))]
    #[allow(clippy::type_complexity)]
    fn extract_sign_material(
        &self,
    ) -> Result<(zeroize::Zeroizing<[u8; 16]>, [u8; 32], [u8; 65]), SsoError> {
        let guard = self.sign_material.lock().unwrap_or_else(|e| e.into_inner());
        match guard.as_ref() {
            Some(m) => Ok((
                zeroize::Zeroizing::new(*m.session_key),
                m.identity_pubkey,
                m.p256_pubkey,
            )),
            None => Err(SsoError::NotPaired),
        }
    }

    /// Return `true` if the phone sent a heartbeat within the presence timeout.
    ///
    /// Delegates to [`presence::is_phone_online`] using the atomic timestamp so
    /// no mutex is required.
    #[cfg(not(target_arch = "wasm32"))]
    pub fn is_phone_online(&self) -> bool {
        presence::is_phone_online(&self.last_heartbeat_ts)
    }

    /// Remove a single entry from the product key cache and persist the snapshot.
    ///
    /// No-op if the cache is uninitialised or the entry does not exist.
    #[cfg(not(target_arch = "wasm32"))]
    pub fn cache_remove_and_persist(&self, product_id: &str, index: u32) {
        let (identity, entries) = {
            let mut guard = self
                .product_key_cache
                .lock()
                .unwrap_or_else(|e| e.into_inner());
            let cache = match guard.as_mut() {
                Some(c) => c,
                None => return,
            };
            cache.remove(product_id, index);
            (*cache.identity(), cache.to_entries())
        };
        if let Err(e) = self.product_key_store.save(&identity, &entries) {
            log::warn!("host-sso: product key cache persist failed after revocation: {e}");
        }
    }
}

// ---------------------------------------------------------------------------
// Free helpers used by both the manager and the background polling thread
// ---------------------------------------------------------------------------

/// Bundled arguments for `handle_established` to stay under clippy's
/// `too_many_arguments` limit.
struct EstablishedCtx<'a, St, E>
where
    St: SsoSessionStore,
    E: SsoEventSink,
{
    address: String,
    display_name: String,
    identity_pubkey: [u8; 32],
    p256_pubkey: [u8; 65],
    phone_wallet_pubkey_hex: Option<String>,
    store: &'a Arc<St>,
    in_progress: &'a Arc<AtomicBool>,
    state: &'a Arc<Mutex<SsoState>>,
    sink: &'a Arc<E>,
}

/// Persist session metadata, clear the in-progress flag, and transition
/// to `Paired`. Extracted so both `handle_pairing_result` and the polling
/// thread can call it without an `&SsoManager` reference.
fn handle_established<St, E>(ctx: EstablishedCtx<'_, St, E>) -> Result<(), SsoError>
where
    St: SsoSessionStore,
    E: SsoEventSink,
{
    let p256_pubkey_hex = hex_encode_p256(&ctx.p256_pubkey);
    let session_id = derive_session_id(&ctx.identity_pubkey, &ctx.p256_pubkey);

    let meta = PersistedSessionMeta {
        session_id,
        address: ctx.address.clone(),
        display_name: ctx.display_name.clone(),
        p256_pubkey_hex,
        phone_wallet_pubkey_hex: ctx.phone_wallet_pubkey_hex,
        // TODO(backlog:sso-session-manager): Capabilities are always empty until
        // the phone firmware sends them in the pairing JSON response. Until then,
        // request_product_key() returns ProductKeyCapabilityAbsent. To test the
        // product key flow, manually inject capabilities into the persisted meta.
        capabilities: Vec::new(),
    };

    ctx.store.save(&meta).map_err(SsoError::Store)?;
    ctx.in_progress.store(false, Ordering::Release);

    log::debug!("host-sso: pairing complete for address={}", ctx.address);
    transition_state(
        ctx.state,
        ctx.sink,
        SsoState::Paired {
            address: ctx.address,
            display_name: ctx.display_name,
            phone_online: true,
        },
    );
    Ok(())
}

/// Set `state` to `new_state` and notify `sink`. Used by free-function helpers
/// that do not have a `&SsoManager` reference (e.g., the polling thread).
fn transition_state<E>(state: &Arc<Mutex<SsoState>>, sink: &Arc<E>, new_state: SsoState)
where
    E: SsoEventSink,
{
    let mut guard = state.lock().unwrap_or_else(|e| e.into_inner());
    *guard = new_state.clone();
    drop(guard);
    sink.on_state_changed(&new_state);
}

/// Drain `state_rx` from a `PairingSession` and apply state transitions.
///
/// Runs on a dedicated thread spawned by `pair()`. The thread exits when the
/// channel is closed (i.e., the pairing background thread finishes).
#[cfg(not(target_arch = "wasm32"))]
fn poll_pairing_states<E>(
    state_rx: std::sync::mpsc::Receiver<host_wallet::pairing::PairingState>,
    state: Arc<Mutex<SsoState>>,
    sink: Arc<E>,
    in_progress: Arc<AtomicBool>,
    generation_counter: Arc<std::sync::atomic::AtomicU64>,
    my_generation: u64,
) where
    E: SsoEventSink,
{
    // We receive a fixed sequence: AwaitingScan → Established|Failed.
    // Loop until the sender drops (channel closed).
    for pairing_state in state_rx {
        // If a newer pair() call has started, this thread is stale — discard events.
        if generation_counter.load(Ordering::Acquire) != my_generation {
            log::debug!("host-sso: stale polling thread (gen={my_generation}), exiting");
            return;
        }
        match pairing_state {
            host_wallet::pairing::PairingState::AwaitingScan { qr_uri } => {
                log::debug!("host-sso: QR ready — transitioning to AwaitingScan");
                transition_state(&state, &sink, SsoState::AwaitingScan { qr_uri });
            }
            host_wallet::pairing::PairingState::Established {
                address,
                display_name,
            } => {
                log::debug!("host-sso: pairing established for address={}", address);
                // TODO(backlog:sso-session-manager): PairingState::Established does
                // not carry identity_pubkey or p256_pubkey. Session persistence is
                // skipped here — the host must call handle_pairing_result() with the
                // full PairingResult (including real key material) to persist the session.
                // Phase 2 will extend PairingState to carry these fields.
                in_progress.store(false, Ordering::Release);
                transition_state(
                    &state,
                    &sink,
                    SsoState::Paired {
                        address,
                        display_name,
                        phone_online: true,
                    },
                );
                // Channel will close after this; loop exits naturally.
            }
            host_wallet::pairing::PairingState::Failed(reason) => {
                log::warn!("host-sso: pairing failed: {reason}");
                in_progress.store(false, Ordering::Release);
                transition_state(&state, &sink, SsoState::Failed { reason });
            }
        }
    }
}

/// Hex-encode a 65-byte P-256 public key.
fn hex_encode_p256(pubkey: &[u8; 65]) -> String {
    pubkey.iter().fold(String::with_capacity(130), |mut s, b| {
        use std::fmt::Write as _;
        let _ = write!(s, "{b:02x}");
        s
    })
}

/// Derive a stable session ID from the identity pubkey and P-256 pubkey.
///
/// TODO(backlog:sso-session-manager): Replace with HMAC-SHA256 domain-separated
/// hash. Current concatenation is not unique if the same identity key pairs with
/// a different P-256 key (re-pairing edge case).
fn derive_session_id(identity_pubkey: &[u8; 32], p256_pubkey: &[u8; 65]) -> String {
    let mut id = String::with_capacity(64 + 130);
    for b in identity_pubkey {
        use std::fmt::Write as _;
        let _ = write!(id, "{b:02x}");
    }
    for b in p256_pubkey {
        use std::fmt::Write as _;
        let _ = write!(id, "{b:02x}");
    }
    id
}

/// Derive a stable, deterministic message ID for a product key request.
///
/// Using a deterministic ID means the phone can deduplicate re-sent requests
/// for the same (product_id, index) tuple within the same session.
#[cfg(not(target_arch = "wasm32"))]
fn derive_message_id(phone_pubkey: &[u8; 32], product_id: &str, index: u32) -> String {
    // Simple concatenation — sufficient for uniqueness within a session.
    // The phone treats message_id as opaque; it only echoes it back.
    let mut id = String::new();
    for b in phone_pubkey.iter().take(8) {
        use std::fmt::Write as _;
        let _ = write!(id, "{b:02x}");
    }
    let _ = std::fmt::Write::write_fmt(&mut id, format_args!("-{product_id}-{index}"));
    id
}

// ---------------------------------------------------------------------------
// Tests
// ---------------------------------------------------------------------------

#[cfg(test)]
mod tests {
    use super::*;
    use crate::traits::{PersistedSessionMeta, SsoProductKeyStore};
    use std::sync::{Arc, Mutex};

    // --- Stub implementations ---

    struct NoopTransport;

    impl SsoTransport for NoopTransport {
        fn subscribe(
            &self,
            _topic_hex: &str,
        ) -> Result<(u64, std::sync::mpsc::Receiver<(String, String)>), String> {
            let (_tx, rx) = std::sync::mpsc::channel();
            Ok((0, rx))
        }

        fn unsubscribe(&self, _id: u64) {}

        fn write(&self, _topic_hex: &str, _value: &str) -> Result<(), String> {
            Ok(())
        }
    }

    struct StubSigner {
        pubkey: [u8; 32],
    }

    impl host_wallet::HostSigner for StubSigner {
        fn public_key(&self) -> Result<[u8; 32], host_wallet::SignerError> {
            Ok(self.pubkey)
        }

        fn sign(&self, _payload: &[u8]) -> Result<[u8; 64], host_wallet::SignerError> {
            Ok([0u8; 64])
        }
    }

    #[derive(Default)]
    struct MemoryStore {
        data: Mutex<Option<PersistedSessionMeta>>,
    }

    impl SsoSessionStore for MemoryStore {
        fn save(&self, session: &PersistedSessionMeta) -> Result<(), String> {
            *self.data.lock().unwrap_or_else(|e| e.into_inner()) = Some(session.clone());
            Ok(())
        }

        fn load(&self) -> Result<Option<PersistedSessionMeta>, String> {
            Ok(self.data.lock().unwrap_or_else(|e| e.into_inner()).clone())
        }

        fn clear(&self) -> Result<(), String> {
            *self.data.lock().unwrap_or_else(|e| e.into_inner()) = None;
            Ok(())
        }
    }

    #[derive(Default)]
    struct RecordingSink {
        states: Arc<Mutex<Vec<SsoState>>>,
    }

    impl SsoEventSink for RecordingSink {
        fn on_state_changed(&self, state: &SsoState) {
            self.states
                .lock()
                .unwrap_or_else(|e| e.into_inner())
                .push(state.clone());
        }
    }

    // A memory-backed product key store for tests.
    #[derive(Default)]
    struct MemoryProductKeyStore {
        data: Mutex<
            Option<(
                [u8; 32],
                Vec<crate::product_key_cache::ProductKeyCacheEntry>,
            )>,
        >,
    }

    impl SsoProductKeyStore for MemoryProductKeyStore {
        fn save(
            &self,
            identity: &[u8; 32],
            entries: &[crate::product_key_cache::ProductKeyCacheEntry],
        ) -> Result<(), String> {
            *self.data.lock().unwrap_or_else(|e| e.into_inner()) =
                Some((*identity, entries.to_vec()));
            Ok(())
        }

        fn load(&self) -> crate::traits::ProductKeyStoreLoadResult {
            Ok(self.data.lock().unwrap_or_else(|e| e.into_inner()).clone())
        }

        fn delete(&self) -> Result<(), String> {
            *self.data.lock().unwrap_or_else(|e| e.into_inner()) = None;
            Ok(())
        }
    }

    // Helper: build a manager with the default NoopProductKeyStore.
    fn make_manager() -> SsoManager<NoopTransport, StubSigner, MemoryStore, RecordingSink> {
        SsoManager::new(
            NoopTransport,
            StubSigner {
                pubkey: [0x01u8; 32],
            },
            MemoryStore::default(),
            RecordingSink::default(),
            "https://example.com/metadata".to_string(),
        )
    }

    // Helper: build a manager with the MemoryProductKeyStore.
    fn make_manager_with_pk_store(
    ) -> SsoManager<NoopTransport, StubSigner, MemoryStore, RecordingSink, MemoryProductKeyStore>
    {
        SsoManager::with_product_key_store(
            NoopTransport,
            StubSigner {
                pubkey: [0x01u8; 32],
            },
            MemoryStore::default(),
            RecordingSink::default(),
            MemoryProductKeyStore::default(),
            "https://example.com/metadata".to_string(),
        )
    }

    fn make_pairing_result() -> PairingResult {
        PairingResult {
            address: "5GrwvaEF5zXb26Fz9rcQpDWS57CtERHpNehXCPcNoHGKutQY".to_string(),
            display_name: "Alice's Phone".to_string(),
            session_key: zeroize::Zeroizing::new([0xABu8; 16]),
            identity_pubkey: [0x01u8; 32],
            p256_pubkey: [0x04u8; 65],
            phone_wallet_pubkey: None,
        }
    }

    fn make_pairing_result_with_phone_pubkey(phone_pubkey: [u8; 32]) -> PairingResult {
        PairingResult {
            phone_wallet_pubkey: Some(phone_pubkey),
            ..make_pairing_result()
        }
    }

    // -----------------------------------------------------------------------
    // Existing tests (preserved unchanged)
    // -----------------------------------------------------------------------

    #[test]
    fn test_new_starts_in_idle_state() {
        let manager = make_manager();
        assert_eq!(manager.state(), SsoState::Idle);
    }

    #[test]
    fn test_restore_session_transitions_to_paired() {
        let manager = make_manager();

        // Pre-populate the store.
        manager
            .store
            .save(&PersistedSessionMeta {
                session_id: "sid".to_string(),
                address: "5GrwvaEF5zXb26Fz9rcQpDWS57CtERHpNehXCPcNoHGKutQY".to_string(),
                display_name: "Alice's Phone".to_string(),
                p256_pubkey_hex: "04".repeat(65),
                phone_wallet_pubkey_hex: None,
                capabilities: Vec::new(),
            })
            .unwrap();

        manager.restore_session().unwrap();

        assert!(matches!(
            manager.state(),
            SsoState::Paired { address, .. } if address == "5GrwvaEF5zXb26Fz9rcQpDWS57CtERHpNehXCPcNoHGKutQY"
        ));

        // Verify the sink was notified of the state transition.
        let states = manager
            .sink
            .states
            .lock()
            .unwrap_or_else(|e| e.into_inner());
        assert!(!states.is_empty(), "sink must be notified on restore");
        assert!(matches!(states.last(), Some(SsoState::Paired { .. })));
    }

    #[test]
    fn test_restore_session_stays_idle_when_no_stored_session() {
        let manager = make_manager();
        manager.restore_session().unwrap();
        assert_eq!(manager.state(), SsoState::Idle);
    }

    #[test]
    fn test_unpair_clears_store_and_returns_to_idle() {
        let manager = make_manager();

        // Establish a paired state first.
        manager
            .handle_pairing_result(make_pairing_result())
            .unwrap();
        assert!(matches!(manager.state(), SsoState::Paired { .. }));

        manager.unpair().unwrap();
        assert_eq!(manager.state(), SsoState::Idle);

        // Store must be empty after unpair.
        assert!(manager.store.load().unwrap().is_none());
    }

    #[cfg(not(target_arch = "wasm32"))]
    #[test]
    fn test_pair_rejects_when_already_pairing() {
        let manager = make_manager();

        // First call should succeed.
        manager.pair().unwrap();

        // Second call must fail with PairingAlreadyInProgress.
        let err = manager.pair().unwrap_err();
        assert!(
            matches!(err, SsoError::PairingAlreadyInProgress),
            "expected PairingAlreadyInProgress, got {err:?}"
        );

        // Clean up so the flag is reset.
        manager.unpair().unwrap();
    }

    #[test]
    fn test_handle_pairing_result_persists_and_transitions() {
        let manager = make_manager();

        manager
            .handle_pairing_result(make_pairing_result())
            .unwrap();

        // State must be Paired.
        let state = manager.state();
        assert!(
            matches!(state, SsoState::Paired { ref address, .. } if address == "5GrwvaEF5zXb26Fz9rcQpDWS57CtERHpNehXCPcNoHGKutQY"),
            "unexpected state: {state:?}"
        );

        // Metadata must be persisted.
        let meta = manager.store.load().unwrap();
        assert!(meta.is_some(), "session should be persisted");
        let meta = meta.unwrap();
        assert_eq!(
            meta.address,
            "5GrwvaEF5zXb26Fz9rcQpDWS57CtERHpNehXCPcNoHGKutQY"
        );
        assert_eq!(meta.display_name, "Alice's Phone");

        // pairing_in_progress flag must be cleared.
        assert!(
            !manager.pairing_in_progress.load(Ordering::Acquire),
            "pairing_in_progress should be false after success"
        );
    }

    /// Verify that `pair()` transitions to `AwaitingScan` with a real QR URI.
    ///
    /// The `NoopTransport` subscribe closure returns a channel whose sender is
    /// dropped immediately, causing `start_pairing`'s handshake thread to time
    /// out. The polling thread will eventually emit `Failed`, but we only
    /// observe the intermediate `AwaitingScan` transition here.
    #[cfg(not(target_arch = "wasm32"))]
    #[test]
    fn test_pair_emits_awaiting_scan_with_real_qr_uri() {
        let manager = make_manager();

        manager.pair().unwrap();

        // The pairing thread emits AwaitingScan almost immediately after generating
        // the P-256 keypair. Poll for up to 2 seconds to accommodate scheduler variance.
        let deadline = std::time::Instant::now() + std::time::Duration::from_secs(2);
        loop {
            {
                let states = manager
                    .sink
                    .states
                    .lock()
                    .unwrap_or_else(|e| e.into_inner());
                if let Some(SsoState::AwaitingScan { qr_uri }) = states.first() {
                    assert!(
                        qr_uri.starts_with("polkadotapp://pair?handshake="),
                        "QR URI must use the polkadotapp scheme, got: {qr_uri}"
                    );
                    assert!(!qr_uri.is_empty(), "QR URI must be non-empty");
                    return;
                }
            }
            if std::time::Instant::now() >= deadline {
                let states = manager
                    .sink
                    .states
                    .lock()
                    .unwrap_or_else(|e| e.into_inner());
                panic!("AwaitingScan not received within 2s; states so far: {states:?}");
            }
            std::thread::sleep(std::time::Duration::from_millis(20));
        }
    }

    /// Verify that `pair()` eventually transitions to `Failed` when the
    /// transport returns a closed channel (no mobile response).
    #[cfg(not(target_arch = "wasm32"))]
    #[test]
    #[ignore = "takes 120s (pairing timeout); run manually with --include-ignored"]
    fn test_pair_transitions_to_failed_on_timeout() {
        let manager = make_manager();
        manager.pair().unwrap();

        // Block until we see a Failed state. Pairing times out after 120s.
        let deadline = std::time::Instant::now() + std::time::Duration::from_secs(125);
        loop {
            {
                let states = manager
                    .sink
                    .states
                    .lock()
                    .unwrap_or_else(|e| e.into_inner());
                if states.iter().any(|s| matches!(s, SsoState::Failed { .. })) {
                    return;
                }
            }
            if std::time::Instant::now() >= deadline {
                panic!("Failed state not received within expected timeout window");
            }
            std::thread::sleep(std::time::Duration::from_millis(500));
        }
    }

    #[cfg(not(target_arch = "wasm32"))]
    #[test]
    fn test_request_sign_returns_not_paired_when_idle() {
        let manager = make_manager();
        let err = manager.request_sign(b"payload").unwrap_err();
        assert!(
            matches!(err, SsoError::NotPaired),
            "expected NotPaired when idle, got {err:?}"
        );
    }

    #[cfg(not(target_arch = "wasm32"))]
    #[test]
    fn test_request_sign_stores_and_clears_sign_material() {
        let manager = make_manager();

        // handle_pairing_result populates sign_material.
        manager
            .handle_pairing_result(make_pairing_result())
            .unwrap();
        {
            let guard = manager
                .sign_material
                .lock()
                .unwrap_or_else(|e| e.into_inner());
            assert!(
                guard.is_some(),
                "sign_material must be populated after pairing"
            );
        }

        // unpair must zeroize sign_material.
        manager.unpair().unwrap();
        {
            let guard = manager
                .sign_material
                .lock()
                .unwrap_or_else(|e| e.into_inner());
            assert!(
                guard.is_none(),
                "sign_material must be cleared after unpair"
            );
        }
    }

    #[cfg(not(target_arch = "wasm32"))]
    #[test]
    fn test_request_sign_returns_not_paired_after_unpair() {
        let manager = make_manager();
        manager
            .handle_pairing_result(make_pairing_result())
            .unwrap();
        manager.unpair().unwrap();

        let err = manager.request_sign(b"payload").unwrap_err();
        assert!(
            matches!(err, SsoError::NotPaired),
            "expected NotPaired after unpair, got {err:?}"
        );
    }

    /// Verify that calling `unpair()` during pairing cancels the handshake and
    /// returns to `Idle`.
    #[cfg(not(target_arch = "wasm32"))]
    #[test]
    fn test_unpair_cancels_in_progress_pairing() {
        let manager = make_manager();

        manager.pair().unwrap();
        assert!(
            manager.pairing_in_progress.load(Ordering::Acquire),
            "pairing_in_progress should be set after pair()"
        );

        manager.unpair().unwrap();

        // After unpair the flag must be cleared and state must be Idle.
        assert_eq!(manager.state(), SsoState::Idle);
        assert!(
            !manager.pairing_in_progress.load(Ordering::Acquire),
            "pairing_in_progress should be cleared after unpair()"
        );
    }

    // -----------------------------------------------------------------------
    // New product key tests
    // -----------------------------------------------------------------------

    /// `request_product_key` must return `NotPaired` when the manager is Idle.
    #[cfg(not(target_arch = "wasm32"))]
    #[test]
    fn test_request_product_key_returns_not_paired_when_idle() {
        let manager = make_manager();
        let err = manager.request_product_key("acme.dot", 0).unwrap_err();
        assert!(
            matches!(err, SsoError::NotPaired),
            "expected NotPaired when idle, got {err:?}"
        );
    }

    /// `request_product_key` must return `ProductKeyCapabilityAbsent` when the
    /// persisted session does not list `"product_key"` in capabilities.
    #[cfg(not(target_arch = "wasm32"))]
    #[test]
    fn test_request_product_key_returns_capability_absent_when_unsupported() {
        let manager = make_manager();
        // Pair without advertising the product_key capability.
        manager
            .handle_pairing_result(make_pairing_result())
            .unwrap();
        // The capabilities list is empty by default after handle_pairing_result.

        let err = manager.request_product_key("acme.dot", 0).unwrap_err();
        assert!(
            matches!(err, SsoError::ProductKeyCapabilityAbsent),
            "expected ProductKeyCapabilityAbsent, got {err:?}"
        );
    }

    /// `request_product_key` must return a cached key without calling the
    /// transport when a matching entry is already in the cache.
    #[cfg(not(target_arch = "wasm32"))]
    #[test]
    fn test_request_product_key_returns_cached_on_hit() {
        const PHONE_PUBKEY: [u8; 32] = [0x02u8; 32];
        const CACHED_KEY: [u8; 32] = [0xAAu8; 32];

        let manager = make_manager_with_pk_store();

        // Pair with a phone that advertises the product_key capability.
        let mut result = make_pairing_result_with_phone_pubkey(PHONE_PUBKEY);
        result.phone_wallet_pubkey = Some(PHONE_PUBKEY);
        manager.handle_pairing_result(result).unwrap();

        // Manually advertise the capability in the persisted meta.
        {
            let mut meta = manager.store.load().unwrap().unwrap();
            meta.capabilities = vec!["product_key".to_string()];
            manager.store.save(&meta).unwrap();
        }

        // Manually seed the in-memory cache.
        {
            let mut guard = manager
                .product_key_cache
                .lock()
                .unwrap_or_else(|e| e.into_inner());
            let cache = guard.get_or_insert_with(|| ProductKeyCache::new(PHONE_PUBKEY));
            cache
                .insert(&PHONE_PUBKEY, "acme.dot", 0, CACHED_KEY)
                .unwrap();
        }

        // The key must be returned from the cache — NoopTransport would fail if
        // the transport were actually called (write() returns Ok but subscribe
        // returns a closed channel, causing a timeout).
        let key = manager.request_product_key("acme.dot", 0).unwrap();
        assert_eq!(key, CACHED_KEY, "must return the cached key");
    }

    /// After `unpair()`, the product key cache must be cleared and the
    /// persisted store entry must be deleted.
    #[cfg(not(target_arch = "wasm32"))]
    #[test]
    fn test_request_product_key_clears_cache_on_unpair() {
        const PHONE_PUBKEY: [u8; 32] = [0x03u8; 32];
        const CACHED_KEY: [u8; 32] = [0xBBu8; 32];

        let manager = make_manager_with_pk_store();

        // Pair and seed the cache.
        manager
            .handle_pairing_result(make_pairing_result_with_phone_pubkey(PHONE_PUBKEY))
            .unwrap();
        {
            let mut guard = manager
                .product_key_cache
                .lock()
                .unwrap_or_else(|e| e.into_inner());
            let cache = guard.get_or_insert_with(|| ProductKeyCache::new(PHONE_PUBKEY));
            cache
                .insert(&PHONE_PUBKEY, "acme.dot", 0, CACHED_KEY)
                .unwrap();
        }
        // Persist the cache manually so we can verify deletion.
        manager
            .product_key_store
            .save(
                &PHONE_PUBKEY,
                &[crate::product_key_cache::ProductKeyCacheEntry {
                    product_id: "acme.dot".to_string(),
                    index: 0,
                    pubkey: CACHED_KEY,
                }],
            )
            .unwrap();
        assert!(
            manager.product_key_store.load().unwrap().is_some(),
            "store must contain an entry before unpair"
        );

        manager.unpair().unwrap();

        // In-memory cache must be cleared.
        let guard = manager
            .product_key_cache
            .lock()
            .unwrap_or_else(|e| e.into_inner());
        assert!(
            guard.is_none(),
            "product_key_cache must be None after unpair"
        );
        drop(guard);

        // Persisted store entry must be deleted.
        assert!(
            manager.product_key_store.load().unwrap().is_none(),
            "product key store must be empty after unpair"
        );
    }

    // -----------------------------------------------------------------------
    // Phase 3: presence + revocation tests
    // -----------------------------------------------------------------------

    /// `is_phone_online()` must return `false` on a freshly constructed manager
    /// because `last_heartbeat_ts` is initialised to 0.
    #[cfg(not(target_arch = "wasm32"))]
    #[test]
    fn test_is_phone_online_returns_false_initially() {
        let manager = make_manager();
        assert!(
            !manager.is_phone_online(),
            "phone must be offline before any heartbeat"
        );
    }

    /// `request_product_key` must return `PhoneOffline` on a cache miss when
    /// no heartbeat has been received (default initial state).
    #[cfg(not(target_arch = "wasm32"))]
    #[test]
    fn test_request_product_key_short_circuits_on_phone_offline() {
        const PHONE_PUBKEY: [u8; 32] = [0x04u8; 32];

        let manager = make_manager_with_pk_store();

        // Pair and advertise the product_key capability.
        manager
            .handle_pairing_result(make_pairing_result_with_phone_pubkey(PHONE_PUBKEY))
            .unwrap();
        {
            let mut meta = manager.store.load().unwrap().unwrap();
            meta.capabilities = vec!["product_key".to_string()];
            manager.store.save(&meta).unwrap();
        }

        // last_heartbeat_ts is 0 (never received) — phone is offline.
        let err = manager.request_product_key("acme.dot", 0).unwrap_err();
        assert!(
            matches!(err, SsoError::PhoneOffline),
            "expected PhoneOffline on cache miss with no heartbeat, got {err:?}"
        );
    }

    /// `cache_remove_and_persist` must evict the specified entry and write the
    /// updated snapshot to the product key store.
    #[cfg(not(target_arch = "wasm32"))]
    #[test]
    fn test_cache_remove_and_persist_removes_entry() {
        const PHONE_PUBKEY: [u8; 32] = [0x05u8; 32];
        const PUBKEY_A: [u8; 32] = [0xAAu8; 32];
        const PUBKEY_B: [u8; 32] = [0xBBu8; 32];

        let manager = make_manager_with_pk_store();
        manager
            .handle_pairing_result(make_pairing_result_with_phone_pubkey(PHONE_PUBKEY))
            .unwrap();

        // Seed cache with two entries.
        {
            let mut guard = manager
                .product_key_cache
                .lock()
                .unwrap_or_else(|e| e.into_inner());
            let cache = guard.get_or_insert_with(|| ProductKeyCache::new(PHONE_PUBKEY));
            cache
                .insert(&PHONE_PUBKEY, "acme.dot", 0, PUBKEY_A)
                .unwrap();
            cache.insert(&PHONE_PUBKEY, "foo.dot", 1, PUBKEY_B).unwrap();
        }

        // Revoke acme.dot/0.
        manager.cache_remove_and_persist("acme.dot", 0);

        // In-memory cache must no longer contain acme.dot/0.
        {
            let mut guard = manager
                .product_key_cache
                .lock()
                .unwrap_or_else(|e| e.into_inner());
            let result = guard
                .as_mut()
                .and_then(|c| c.get(&PHONE_PUBKEY, "acme.dot", 0));
            assert!(result.is_none(), "revoked entry must be absent from cache");
        }

        // Persisted snapshot must still contain foo.dot/1 but not acme.dot/0.
        let stored = manager.product_key_store.load().unwrap();
        let (_, entries) = stored.expect("store must have an entry after persist");
        assert!(
            !entries
                .iter()
                .any(|e| e.product_id == "acme.dot" && e.index == 0),
            "revoked entry must be absent from persistent store"
        );
        assert!(
            entries
                .iter()
                .any(|e| e.product_id == "foo.dot" && e.index == 1),
            "non-revoked entry must remain in persistent store"
        );
    }

    /// `unpair()` must reset `last_heartbeat_ts` to 0 so a subsequent
    /// `is_phone_online()` call returns `false`.
    #[cfg(not(target_arch = "wasm32"))]
    #[test]
    fn test_unpair_resets_heartbeat_timestamp() {
        let manager = make_manager();

        // Simulate a received heartbeat by writing a recent timestamp.
        manager.last_heartbeat_ts.store(
            std::time::SystemTime::now()
                .duration_since(std::time::UNIX_EPOCH)
                .unwrap_or_default()
                .as_secs() as i64,
            Ordering::Release,
        );
        assert!(
            manager.is_phone_online(),
            "phone must appear online after heartbeat"
        );

        manager
            .handle_pairing_result(make_pairing_result())
            .unwrap();
        manager.unpair().unwrap();

        assert!(
            !manager.is_phone_online(),
            "phone must be offline after unpair resets heartbeat timestamp"
        );
    }

    // -----------------------------------------------------------------------
    // Issue #108 — SsoManager happy-path tests
    // -----------------------------------------------------------------------

    /// A freshly constructed manager must start in the `Idle` state.
    #[test]
    fn test_initial_state_is_idle() {
        let manager = make_manager();
        assert_eq!(manager.state(), SsoState::Idle);
    }

    /// When the store contains a persisted session, `restore_session` must
    /// transition to `Paired` with the stored address.
    #[test]
    fn test_restore_session_transitions_to_paired_issue108() {
        let manager = make_manager();

        manager
            .store
            .save(&PersistedSessionMeta {
                session_id: "sid-108".to_string(),
                address: "5FHneW46xGXgs5mUiveU4sbTyGBzmstUspZC92UhjJM694ty".to_string(),
                display_name: "Bob's Phone".to_string(),
                p256_pubkey_hex: "04".repeat(65),
                phone_wallet_pubkey_hex: None,
                capabilities: Vec::new(),
            })
            .unwrap();

        manager.restore_session().unwrap();

        assert!(
            matches!(
                manager.state(),
                SsoState::Paired { ref address, .. }
                    if address == "5FHneW46xGXgs5mUiveU4sbTyGBzmstUspZC92UhjJM694ty"
            ),
            "expected Paired state after restore, got {:?}",
            manager.state()
        );
    }

    /// When the store is empty, `restore_session` must leave the manager `Idle`.
    #[test]
    fn test_restore_session_stays_idle_when_no_session() {
        let manager = make_manager();
        manager.restore_session().unwrap();
        assert_eq!(manager.state(), SsoState::Idle);
    }

    /// After pairing, `unpair()` must return the manager to the `Idle` state.
    #[test]
    fn test_unpair_from_paired_returns_to_idle() {
        let manager = make_manager();

        manager
            .handle_pairing_result(make_pairing_result())
            .unwrap();
        assert!(
            matches!(manager.state(), SsoState::Paired { .. }),
            "must be Paired before calling unpair()"
        );

        manager.unpair().unwrap();
        assert_eq!(manager.state(), SsoState::Idle);
    }

    /// The event sink must receive a `Paired` notification when
    /// `restore_session` finds a persisted session.
    #[test]
    fn test_sink_receives_state_change_on_restore() {
        let manager = make_manager();

        manager
            .store
            .save(&PersistedSessionMeta {
                session_id: "sid-sink".to_string(),
                address: "5GNJqTPyNqANBkUVMN1LPPrxXnFouWXoe2wNSmmEoLctxiZY".to_string(),
                display_name: "Charlie's Phone".to_string(),
                p256_pubkey_hex: "04".repeat(65),
                phone_wallet_pubkey_hex: None,
                capabilities: Vec::new(),
            })
            .unwrap();

        manager.restore_session().unwrap();

        let states = manager
            .sink
            .states
            .lock()
            .unwrap_or_else(|e| e.into_inner());
        assert!(
            !states.is_empty(),
            "sink must receive at least one state notification on restore"
        );
        assert!(
            matches!(states.last(), Some(SsoState::Paired { .. })),
            "last notified state must be Paired, got {:?}",
            states.last()
        );
    }
}