ping_core/

client.rs

//! `MessagingClient` — top-level handle. Owns the OpenMLS provider, identity, local device,
//! and the set of open conversations.
//!
//! All operations are `async`. The intent is that the FFI generators emit Swift `async`,
//! Kotlin `suspend`, and the WASM glue exposes Promises.

use openmls::framing::MlsMessageOut;
use openmls::prelude::{
    tls_codec::Serialize as TlsSerialize, BasicCredential, Ciphersuite, CredentialWithKey,
    KeyPackageBuilder,
};
use openmls_basic_credential::SignatureKeyPair;
use openmls_traits::OpenMlsProvider;
use parking_lot::RwLock;
use ping_mls_store::{PersistentMlsProvider, StorageBackend};
use std::collections::HashMap;
use std::sync::Arc;
use zeroize::Zeroizing;

use crate::{
    codec,
    conversation::{Conversation, ConversationId, ConversationMeta, MemberInfo},
    device::{
        CatchupAppEventEntry, CatchupConversationEntry, CatchupSnapshot, DeviceId, DeviceInfo,
        LinkingTicket, LocalDevice, CATCHUP_SNAPSHOT_SOFT_CAP, CATCHUP_SNAPSHOT_VERSION,
    },
    error::{Error, Result},
    identity::{Identity, UserId},
    message::{IncomingMessage, MessageEnvelope, MessageKind},
    storage::Storage,
    sync::SyncCursor,
    transport::Transport,
};

const DEFAULT_CIPHERSUITE: Ciphersuite = Ciphersuite::MLS_128_DHKEMX25519_AES128GCM_SHA256_Ed25519;

/// Whether a transport send failure is a DEFINITE server rejection (the server
/// returned an HTTP error response) rather than an ambiguous network failure
/// where the server may actually have applied the message.
///
/// This decides whether a staged Commit can be safely rolled back: only a
/// definite rejection guarantees the server did NOT apply it. The host transport
/// embeds the HTTP status in the error string (e.g. "network: http 409"); 4xx +
/// known server error codes are definite, while timeouts / "fetch failed" / 5xx
/// are ambiguous (could be a masked success). When unsure we treat it as
/// ambiguous (return false) — merging on a masked success is recoverable, but
/// rolling back a Commit the server DID apply strands us a step behind forever.
fn is_definite_rejection(err: &Error) -> bool {
    let Error::Transport(s) = err else {
        return false;
    };
    let s = s.to_ascii_lowercase();
    s.contains("http 4")
        || s.contains("epoch_advanced")
        || s.contains("invalid_request")
        || s.contains("not_found")
        || s.contains("conflict")
        || s.contains("forbidden")
        || s.contains("unauthorized")
}

/// Per-chat result reported by [`MessagingClient::admit_device_to_chats`].
#[derive(Debug, Clone)]
pub struct AdmitChatOutcome {
    pub conversation_id: ConversationId,
    pub status: AdmitChatStatus,
}

#[derive(Debug, Clone)]
pub enum AdmitChatStatus {
    /// The new device is now an MLS leaf in this chat. Both the Commit
    /// and the addressed Welcome have been sent.
    Admitted,
    /// We chose not to admit (e.g. the conversation is a DeviceGroup,
    /// which was already handled at linking-ticket build time).
    Skipped { reason: String },
    /// MLS or transport rejected the admission. `error` is the underlying
    /// message — typically a `transport error: ...` or an OpenMLS error.
    Failed { error: String },
}

#[derive(Debug)]
pub struct ClientConfig {
    pub identity: Identity,
    pub device_label: String,
    pub storage: Arc<dyn Storage>,
    pub transport: Arc<dyn Transport>,
    /// Wall clock in ms. Pulled from the host so we can use a synthetic clock in tests.
    pub now_ms: u64,
    /// [CR-4] OpenMLS-provider backend. Defaults to in-memory; iOS NSE and web SW
    /// cold-start paths MUST pass `StorageBackend::Sqlite { path, encryption_key }`
    /// (native) or `StorageBackend::IndexedDb { db_name }` (WASM, when that lands).
    /// See `docs/design/CR4_CR7_PERSISTENCE.md`.
    pub storage_backend: StorageBackend,
    /// Optional 32-byte Ed25519 secret key the SDK should use as the
    /// device signing key. When set AND no `LocalDevice` is yet
    /// persisted in `storage`, the SDK constructs its first
    /// `LocalDevice` from this key instead of generating a fresh
    /// random one — so `device_id = SHA-256(public_key_of(secret))`
    /// is fully determined by what the host provided.
    ///
    /// Use case: align the SDK's `device_id` (which it stamps into
    /// every envelope's `sender_device` field) with an externally-
    /// computed device id — typically `SHA-256(device_signing_pubkey)`
    /// in the host's auth layer, where the JWT carries that same
    /// value as its `device_id` claim. Without this alignment, a
    /// server that validates `envelope.sender_device ==
    /// jwt.device_id` would reject every send.
    ///
    /// Ignored on re-init (when storage already has a persisted
    /// `LocalDevice`) so the device identity remains stable across
    /// restarts.
    pub device_signing_secret_key: Option<[u8; 32]>,
}

impl ClientConfig {
    /// Construct a config with `StorageBackend::Memory` — convenient for tests and
    /// the existing v0.1 in-memory flow.
    pub fn new_in_memory(
        identity: Identity,
        device_label: String,
        storage: Arc<dyn Storage>,
        transport: Arc<dyn Transport>,
        now_ms: u64,
    ) -> Self {
        Self {
            identity,
            device_label,
            storage,
            transport,
            now_ms,
            storage_backend: StorageBackend::Memory,
            device_signing_secret_key: None,
        }
    }
}

pub struct MessagingClient {
    pub(crate) identity: Identity,
    pub(crate) local_device: LocalDevice,
    pub(crate) crypto: Arc<PersistentMlsProvider>,
    pub(crate) signing: Arc<SignatureKeyPair>,
    pub(crate) storage: Arc<dyn Storage>,
    pub(crate) transport: Arc<dyn Transport>,
    conversations: RwLock<HashMap<ConversationId, Conversation>>,
    /// Conversations detected as STRANDED during catch-up: a full page of
    /// events was fetched but nothing could be applied (every envelope failed
    /// to decrypt / was wrong-epoch), meaning a Commit was missed and the group
    /// can no longer advance from local state. The host polls
    /// [`MessagingClient::stranded_conversations`] after a sync and recovers
    /// each (re-Welcome / same-user state snapshot). Cleared automatically once
    /// the conversation makes progress again.
    stranded: RwLock<std::collections::HashSet<ConversationId>>,
}

impl std::fmt::Debug for MessagingClient {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("MessagingClient")
            .field("user_id", &self.identity.user_id().as_hex())
            .field("device_id", &self.local_device.device_id.as_hex())
            .field("conversation_count", &self.conversations.read().len())
            .finish()
    }
}

impl MessagingClient {
    /// Initialise. Creates a new local device if none is recorded in storage; otherwise rehydrates.
    pub async fn init(cfg: ClientConfig) -> Result<Arc<Self>> {
        // [CR-4] OpenMLS provider is now pluggable. For `StorageBackend::Memory` this
        // behaves like the old `OpenMlsRustCrypto::default()`. For `Sqlite`, the
        // working set is hydrated from the on-disk blob; subsequent `checkpoint` calls
        // flush it back. iOS NSE / web SW cold-start lives here.
        //
        // Use `open_async` so the WASM `StorageBackend::IndexedDb` variant can read
        // its snapshot blob through the host-supplied `AsyncBlobStore` before
        // returning — without this, the provider's `MemoryStorage` would be empty
        // and `MlsGroup::load` would silently return `None` for every group on
        // cold restart, breaking chat persistence across reloads. Native targets
        // (Memory + Sqlite) delegate to the sync path under the hood, so the
        // `.await` is free there.
        let crypto = PersistentMlsProvider::open_async(cfg.storage_backend.clone())
            .await
            .map_err(|e| Error::Storage(format!("provider open: {e}")))?;
        let local_device = match cfg.storage.get("device", "local").await? {
            Some(bytes) => decode_local_device(&bytes, cfg.identity.user_id().clone())?,
            None => {
                // First-init path. If the host supplied a signing secret
                // (typically to align the device_id with their auth
                // layer), use it; otherwise mint a fresh random key.
                // Either way, the constructed `LocalDevice` is
                // immediately persisted so future inits load from
                // storage without consulting the override again.
                let dev = match cfg.device_signing_secret_key.as_ref() {
                    Some(secret) => LocalDevice::from_signing_secret(
                        cfg.identity.user_id().clone(),
                        cfg.device_label,
                        cfg.now_ms,
                        secret,
                    ),
                    None => LocalDevice::generate(
                        cfg.identity.user_id().clone(),
                        cfg.device_label,
                        cfg.now_ms,
                    ),
                };
                let bytes = encode_local_device(&dev)?;
                cfg.storage.put("device", "local", bytes).await?;
                dev
            }
        };

        // [CR-4] MLS signing keypair MUST be stable across cold restarts — otherwise the
        // leaf-key stored on disk no longer matches the per-client key on re-init, and any
        // send-after-restart silently misroutes. We derive deterministically from the
        // already-persistent `LocalDevice::signing` (Ed25519, 32 raw bytes), and the
        // ciphersuite's signature scheme is Ed25519 too — so the device signing key and the
        // MLS leaf signing key are the same bytes. The MLS storage provider also receives
        // a copy via `store()` so OpenMLS-internal lookups (process_message, etc.) succeed.
        let signing = {
            let sk_bytes = local_device.signing.to_bytes().to_vec();
            let pk_bytes = local_device.signing.verifying_key().to_bytes().to_vec();
            let kp = SignatureKeyPair::from_raw(
                DEFAULT_CIPHERSUITE.signature_algorithm(),
                sk_bytes,
                pk_bytes,
            );
            kp.store(crypto.storage()).map_err(Error::mls)?;
            Arc::new(kp)
        };

        let client = Arc::new(Self {
            identity: cfg.identity,
            local_device,
            crypto,
            signing,
            storage: cfg.storage,
            transport: cfg.transport,
            conversations: RwLock::new(HashMap::new()),
            stranded: RwLock::new(std::collections::HashSet::new()),
        });

        client.rehydrate_conversations(cfg.now_ms).await?;

        // [CR-10] Ensure the DeviceGroup exists at init, not lazily inside
        // build_linking_ticket. Single-device users need somewhere to write
        // personal events (drafts, read pointers, notes, vault wrapper)
        // even before they pair a second device. Lazy creation in
        // build_linking_ticket left them with no DG → no place for
        // personal state to land.
        //
        // Idempotent — re-init after a cold restart finds the DG via
        // rehydrate_conversations and this becomes a no-op.
        client.ensure_device_group(cfg.now_ms).await?;

        Ok(client)
    }

    /// [CR-10] Idempotently ensures this user's DeviceGroup exists in
    /// `self.conversations`. Called from `init` (so single-device users
    /// have a DG immediately) and from `build_linking_ticket` (the legacy
    /// lazy path; still safe to call when the DG already exists, since
    /// rehydrate_conversations would have re-attached it before init
    /// returned).
    ///
    /// The DeviceGroup is a one-leaf MLS group at creation time —
    /// `add_members` (called by `build_linking_ticket` when a second
    /// device pairs in) is what grows it. We persist the snapshot so a
    /// cold restart picks it up before this function runs again.
    pub(crate) async fn ensure_device_group(self: &Arc<Self>, now_ms: u64) -> Result<()> {
        let dg_id = device_group_id_for(self.identity.user_id());
        if self.conversations.read().contains_key(&dg_id) {
            return Ok(());
        }
        let mut new_dg = Conversation::create(
            dg_id,
            Some("device-group".into()),
            self.local_device.device_id.clone(),
            self.identity.user_id(),
            self.crypto.clone(),
            self.signing.clone(),
            self.storage.clone(),
            now_ms,
        )?;
        new_dg.meta.is_device_group = true;
        new_dg.snapshot_to_storage().await?;
        self.conversations.write().insert(dg_id, new_dg);
        Ok(())
    }

    pub fn user_id(&self) -> UserId {
        self.identity.user_id().clone()
    }
    /// Export this client's account identity (the Ed25519 seed, CBOR-wrapped —
    /// same format `Identity::import` / `MessagingClient::init(identity_export)`
    /// accept). SECRET. Hosts use this to TRANSFER the account identity to a
    /// newly-linked device over the sealed linking channel, so every linked
    /// device shares ONE `user_id` and `IncomingMessage.sender_user_id` equals
    /// the local `user_id()` for any of the account's own devices — the basis
    /// for cross-device self-attribution. Never log or persist in cleartext.
    pub fn export_identity(&self) -> Zeroizing<Vec<u8>> {
        self.identity.export()
    }
    pub fn device_id(&self) -> DeviceId {
        self.local_device.device_id.clone()
    }
    pub fn device_info(&self, now_ms: u64) -> DeviceInfo {
        self.local_device.info(now_ms)
    }

    /// Generate a fresh KeyPackage to publish to the directory. Hosts call this when registering
    /// a device or topping up the directory.
    ///
    /// `build()` writes the private init + encryption keys into the storage
    /// provider's working set, but ON ITS OWN that write is NOT durable: on the
    /// WASM/AsyncBlob backend the working set only reaches IndexedDB at the next
    /// `checkpoint_async`, so a page reload before the next state-changing op
    /// loses the private keys while the PUBLIC KeyPackage has already been
    /// published. Any Welcome later bound to that KeyPackage then fails with
    /// "No matching key package was found in the key store" (breaking calls and
    /// every invite to this device). So we checkpoint HERE, before returning the
    /// bytes the host will publish — the published KeyPackage is durable the
    /// instant it leaves this function. Hence `async`.
    pub async fn fresh_key_package(&self) -> Result<Vec<u8>> {
        self.build_key_package(false).await
    }

    /// Generate a fresh LAST-RESORT KeyPackage.
    ///
    /// A normal KeyPackage is single-use: once a Welcome consumes it, the
    /// private init key is deleted and the directory entry is burned. When a
    /// device's published pool runs dry, every invite to that device hard-fails
    /// ("user unavailable") until it comes online and tops up — the classic
    /// "I added them but they never got it" complaint.
    ///
    /// A last-resort KeyPackage (RFC 9420 §10) carries the `LastResort`
    /// extension, signalling the server it may serve this KeyPackage MORE THAN
    /// ONCE when no single-use KeyPackages remain. The host publishes exactly
    /// one per device; the server keeps it as the always-available fallback so
    /// an invite never fails purely because the pool emptied. Forward secrecy
    /// for the joining epoch is slightly weaker (the init key is reused until
    /// replenishment), which is the accepted RFC trade-off for availability.
    pub async fn fresh_last_resort_key_package(&self) -> Result<Vec<u8>> {
        self.build_key_package(true).await
    }

    async fn build_key_package(&self, last_resort: bool) -> Result<Vec<u8>> {
        let credential_with_key = CredentialWithKey {
            credential: BasicCredential::new(self.identity.user_id().0.clone()).into(),
            signature_key: self.signing.public().to_vec().into(),
        };
        let mut builder = KeyPackageBuilder::new()
            // Advertise the group-name extension capability so a later
            // `set_name` (rename / avatar change via GroupContextExtensions)
            // passes openmls' per-member capability check on every group this
            // device joins. For a last-resort KeyPackage the leaf must also
            // advertise the LastResort extension it carries, or openmls
            // validation rejects the KeyPackage at add-member time. See
            // `conversation::ping_leaf_capabilities_for`.
            .leaf_node_capabilities(crate::conversation::ping_leaf_capabilities_for(last_resort));
        if last_resort {
            builder = builder.mark_as_last_resort();
        }
        let bundle = builder
            .build(
                DEFAULT_CIPHERSUITE,
                self.crypto.as_ref(),
                self.signing.as_ref(),
                credential_with_key,
            )
            .map_err(Error::mls)?;
        // Durably persist the freshly-generated private keys BEFORE the public
        // KeyPackage is handed to the host to publish (see doc comment).
        self.crypto
            .checkpoint_async()
            .await
            .map_err(|e| Error::Storage(format!("key package checkpoint: {e}")))?;
        // KeyPackages are serialized as MlsMessage(KeyPackage) per the MLS framing spec.
        let msg: MlsMessageOut = bundle.key_package().clone().into();
        msg.tls_serialize_detached().map_err(Error::mls)
    }

    /// Create a new conversation owned by this client (and seeded with a single member: this device).
    pub async fn create_conversation(
        self: &Arc<Self>,
        name: Option<String>,
        now_ms: u64,
    ) -> Result<ConversationId> {
        self.create_conversation_with_id(ConversationId::new(), name, now_ms)
            .await
    }

    /// Create an ephemeral per-call MLS group. Identical to
    /// [`create_conversation`] except the id carries the `0xFF 0xCC` call-group
    /// sentinel (see [`ConversationId::new_call_group`]), so every device — the
    /// callee joining via a name-stripped Welcome, a freshly-linked sibling —
    /// recognises it as a call group by its id alone and keeps it out of the chat
    /// list, with no dependence on the (creator-local) `call:` name or a
    /// per-device registry. Hosts MUST mint call groups via this method rather
    /// than `create_conversation(name: "call:…")` to get the structural guarantee.
    pub async fn create_call_conversation(
        self: &Arc<Self>,
        name: Option<String>,
        now_ms: u64,
    ) -> Result<ConversationId> {
        self.create_conversation_with_id(ConversationId::new_call_group(), name, now_ms)
            .await
    }

    async fn create_conversation_with_id(
        self: &Arc<Self>,
        id: ConversationId,
        name: Option<String>,
        now_ms: u64,
    ) -> Result<ConversationId> {
        let convo = Conversation::create(
            id,
            name,
            self.local_device.device_id.clone(),
            self.identity.user_id(),
            self.crypto.clone(),
            self.signing.clone(),
            self.storage.clone(),
            now_ms,
        )?;
        convo.snapshot_to_storage().await?;
        self.conversations.write().insert(id, convo);
        Ok(id)
    }

    /// Join via a Welcome bundled in a [`MessageEnvelope`] of kind `Welcome`.
    pub async fn join_conversation(
        self: &Arc<Self>,
        welcome_envelope: &MessageEnvelope,
        now_ms: u64,
    ) -> Result<ConversationId> {
        if welcome_envelope.kind != MessageKind::Welcome {
            return Err(Error::Invalid("expected Welcome envelope".into()));
        }
        let convo = Conversation::join(
            &welcome_envelope.payload,
            self.local_device.device_id.clone(),
            self.crypto.clone(),
            self.signing.clone(),
            self.storage.clone(),
            now_ms,
        )?;
        let id = convo.id();
        convo.snapshot_to_storage().await?;
        self.conversations.write().insert(id, convo);
        // Joining (re-Welcome) recovers a previously-stranded conversation.
        self.stranded.write().remove(&id);
        Ok(id)
    }

    /// Conversations detected as STRANDED during catch-up — a Commit was missed
    /// and the group can no longer advance from local state. The host should
    /// recover each (re-Welcome from a peer, or a same-user state-snapshot
    /// import) so messages start delivering again. The set self-clears as a
    /// conversation makes progress or is re-joined.
    pub fn stranded_conversations(&self) -> Vec<ConversationId> {
        let mut ids: Vec<ConversationId> = self.stranded.read().iter().copied().collect();
        ids.sort_by_key(|a| a.0);
        ids
    }

    pub fn list_conversations(&self) -> Vec<ConversationMeta> {
        self.conversations
            .read()
            .values()
            .map(|c| c.meta.clone())
            .collect()
    }

    /// Member roster for a conversation, recovered locally from the MLS
    /// group's leaf credentials. Empty if the conversation is unknown to
    /// this client. Lets any device (including one that just joined via a
    /// linking Welcome) resolve a 1:1 peer's `UserId` without the
    /// out-of-band `ping.profile` re-send.
    pub fn members(&self, conv_id: ConversationId) -> Vec<MemberInfo> {
        self.conversations
            .read()
            .get(&conv_id)
            .map(|c| c.members())
            .unwrap_or_default()
    }

    /// Send an application message. Returns once the envelope has been handed to the transport.
    pub async fn send(
        &self,
        conv_id: ConversationId,
        plaintext: Vec<u8>,
        now_ms: u64,
    ) -> Result<MessageEnvelope> {
        // FAST-FAIL on a stranded conversation. A missed Commit forked our epoch,
        // so anything we encrypt now is at a stale epoch: the server either 409s
        // it (epoch occupied) or accepts a frame the peers can never decrypt — a
        // durable-but-undelivered message. Surface `EpochStranded` immediately so
        // the host renders "needs repair" (and recovers via
        // `stranded_conversations()` → re-Welcome / snapshot import, which clears
        // the mark) instead of silently 409-looping every send. Without this the
        // SDK knew the conversation was forked but sent anyway.
        if self.stranded.read().contains(&conv_id) {
            return Err(Error::EpochStranded(conv_id.as_hex()));
        }
        let envelope = {
            let mut guard = self.conversations.write();
            let convo = guard
                .get_mut(&conv_id)
                .ok_or_else(|| Error::UnknownConversation(conv_id.as_hex()))?;
            convo.send_application(&plaintext, now_ms)?
        };
        self.transport.send(envelope.clone()).await?;
        // The OpenMLS sender ratchet advances on every Application message — `seq` + `hlc`
        // are bumped on the conversation, and the underlying group keystore stores new
        // generation keys. Without a checkpoint here, a reload rolls back to the pre-send
        // state and the next send re-uses an already-consumed generation that receivers
        // silently drop. Mirrors the snapshot calls after every Commit/Welcome op.
        //
        // Capture the snapshot inputs UNDER the read guard, then DROP the
        // guard (end of the `let` statement) before the async flush — never
        // hold a `parking_lot` guard across `.await` (see
        // `Conversation::snapshot_inputs`).
        let snap = self
            .conversations
            .read()
            .get(&conv_id)
            .map(|c| c.snapshot_inputs())
            .transpose()?;
        if let Some(snap) = snap {
            snap.flush().await?;
        }
        Ok(envelope)
    }

    /// Add members. The Commit goes on the wire; the Welcome should be delivered to the new
    /// devices' inboxes (the host transport implements that — typically as a separate addressed
    /// envelope).
    ///
    /// [CR-2] Each entry is `(DeviceId, KeyPackage_bytes)`. The host typically gets the
    /// device_id from the directory at the same time it gets the KeyPackage; we use it to
    /// record a per-conversation `device_id → leaf_index` map so [`Self::revoke_device`]
    /// can later locate the leaf without a fresh directory lookup. The SDK does not
    /// cryptographically verify the host's device-id claim — that's a directory policy
    /// concern.
    //
    // The `conversations` lock is taken only for the SYNCHRONOUS MLS work
    // (the add commit) and the synchronous snapshot capture, then dropped
    // BEFORE every `.await`. We must never hold a `parking_lot` guard
    // across an await — see `Conversation::snapshot_inputs` for why (the
    // single-threaded wasm worker would panic in `parking_lot`'s parker
    // stub). `parking_lot/send_guard` is still set so any guard that DOES
    // briefly cross a yield-free boundary stays `Send`.
    pub async fn add_members(
        &self,
        conv_id: ConversationId,
        entries: Vec<(DeviceId, Vec<u8>)>,
        now_ms: u64,
    ) -> Result<()> {
        // Phase 1 — stage the Commit WITHOUT merging (local epoch unchanged).
        let staged = {
            let mut guard = self.conversations.write();
            let convo = guard
                .get_mut(&conv_id)
                .ok_or_else(|| Error::UnknownConversation(conv_id.as_hex()))?;
            convo.stage_add_members(entries, now_ms)?
        };

        // Phase 2 — send the Commit FIRST, then merge only if the server accepts
        // it (send-then-merge). A Commit the server REJECTS is rolled back, so the
        // local epoch can never run ahead of the server — the desync that
        // permanently bricks a group (every later Commit 409s; peers can't decrypt
        // our epoch). A network failure with NO response is ambiguous (the server
        // may have applied it), so there we merge to match a possible masked
        // success rather than strand ourselves a step behind.
        if let Err(send_err) = self.transport.send(staged.commit.clone()).await {
            let merged = {
                let mut guard = self.conversations.write();
                match guard.get_mut(&conv_id) {
                    Some(convo) if is_definite_rejection(&send_err) => {
                        let _ = convo.abort_staged();
                        false
                    }
                    Some(convo) => {
                        convo.confirm_staged(&staged, now_ms)?;
                        true
                    }
                    None => false,
                }
            };
            if merged {
                self.flush_conversation(&conv_id).await?;
            }
            return Err(send_err);
        }

        // Phase 3 — Commit accepted: merge locally + persist (so the advanced
        // epoch survives a crash even if the Welcome below fails).
        {
            let mut guard = self.conversations.write();
            let convo = guard
                .get_mut(&conv_id)
                .ok_or_else(|| Error::UnknownConversation(conv_id.as_hex()))?;
            convo.confirm_staged(&staged, now_ms)?;
        }
        self.flush_conversation(&conv_id).await?;

        // Phase 4 — deliver the Welcome to the new members. Best-effort: they are
        // in the group server-side now; a failed Welcome is recoverable
        // (re-invite) and must NOT roll back the merged Commit.
        if let Some(welcome) = staged.welcome {
            self.transport.send(welcome).await?;
        }
        Ok(())
    }

    /// Change a conversation's `name` (carried in the GroupContext) and broadcast
    /// the change to every member as an MLS GroupContextExtensions Commit. Unlike
    /// a hydration broadcast, the new name rides MLS group STATE, so every member
    /// — and every future joiner via the GroupInfo — converges on it. Hosts use
    /// this to make a rename or an embedded avatar-media-id change bulletproof
    /// (the `name` carries the `ping:meta:v1:` blob).
    ///
    /// No Welcome (membership is unchanged). Uses the same send-then-merge
    /// rollback discipline as [`Self::add_members`] so a server-rejected Commit
    /// never desyncs the local epoch. All members must have re-linked since the
    /// group-name capability shipped (see `conversation::ping_leaf_capabilities`),
    /// else openmls rejects the Commit.
    pub async fn set_conversation_name(
        &self,
        conv_id: ConversationId,
        name: Option<String>,
        now_ms: u64,
    ) -> Result<()> {
        // Phase 1 — stage the Commit WITHOUT merging (local epoch unchanged).
        let staged = {
            let mut guard = self.conversations.write();
            let convo = guard
                .get_mut(&conv_id)
                .ok_or_else(|| Error::UnknownConversation(conv_id.as_hex()))?;
            convo.stage_set_name(name, now_ms)?
        };

        // Phase 2 — send-then-merge (see `add_members` for the rollback rationale).
        if let Err(send_err) = self.transport.send(staged.commit.clone()).await {
            let merged = {
                let mut guard = self.conversations.write();
                match guard.get_mut(&conv_id) {
                    Some(convo) if is_definite_rejection(&send_err) => {
                        let _ = convo.abort_staged();
                        false
                    }
                    Some(convo) => {
                        convo.confirm_staged(&staged, now_ms)?;
                        true
                    }
                    None => false,
                }
            };
            if merged {
                self.flush_conversation(&conv_id).await?;
            }
            return Err(send_err);
        }

        // Phase 3 — Commit accepted: merge locally + persist.
        {
            let mut guard = self.conversations.write();
            let convo = guard
                .get_mut(&conv_id)
                .ok_or_else(|| Error::UnknownConversation(conv_id.as_hex()))?;
            convo.confirm_staged(&staged, now_ms)?;
        }
        self.flush_conversation(&conv_id).await?;
        Ok(())
    }

    /// Snapshot + flush a conversation's persistable state. Captures the snapshot
    /// synchronously under the read guard, drops the guard, then awaits the flush
    /// (never hold a `parking_lot` guard across an await — wasm parker panics).
    async fn flush_conversation(&self, conv_id: &ConversationId) -> Result<()> {
        let snap = self
            .conversations
            .read()
            .get(conv_id)
            .map(|c| c.snapshot_inputs())
            .transpose()?;
        if let Some(snap) = snap {
            snap.flush().await?;
        }
        Ok(())
    }

    /// Admits `new_device_id` to every conversation in `kps_per_chat` via
    /// the standard MLS `add_members` flow — one Commit + one Welcome per
    /// chat. This is the SDK-side replacement for the host's previous
    /// per-chat reconciler loop after device linking; centralising it
    /// here means iOS/Android/web hosts all share the orchestration and
    /// the transport's Welcome-recipient priming is automatic.
    ///
    /// Inputs:
    /// - `new_device_id`: the device being admitted (matches the
    ///   `device_binding_sig` recipient in the linking ticket).
    /// - `kps_per_chat`: one freshly-claimed KeyPackage per chat. The
    ///   host claims these via the auth-layer's per-account KP pool
    ///   (`GET /v1/devices/{accountId}`) AFTER the new device's
    ///   bootstrap has uploaded its KP batch.
    /// - `now_ms`: wall-clock used to stamp HLCs on the emitted
    ///   envelopes.
    ///
    /// Per-chat failures (unknown conversation, MLS error, transport
    /// error, etc.) are CAPTURED in the returned vec rather than
    /// short-circuiting the whole call — losing one chat shouldn't
    /// strand the new device on every other chat. The caller decides
    /// whether to retry the failed entries (e.g. with a fresh KP).
    pub async fn admit_device_to_chats(
        &self,
        new_device_id: DeviceId,
        kps_per_chat: Vec<(ConversationId, Vec<u8>)>,
        now_ms: u64,
    ) -> Result<Vec<AdmitChatOutcome>> {
        let mut outcomes = Vec::with_capacity(kps_per_chat.len());
        for (conv_id, kp_bytes) in kps_per_chat {
            // Belt-and-braces: skip the DeviceGroup. The DG was already
            // welcomed via the linking ticket — re-adding the new
            // device there would produce a duplicate-add Commit that
            // BE de-dups, but the noise is avoidable.
            let is_dg = self
                .conversations
                .read()
                .get(&conv_id)
                .map(|c| c.meta().is_device_group)
                .unwrap_or(false);
            if is_dg {
                outcomes.push(AdmitChatOutcome {
                    conversation_id: conv_id,
                    status: AdmitChatStatus::Skipped {
                        reason: "device_group".to_string(),
                    },
                });
                continue;
            }

            // Prime the host transport with the welcome recipient BEFORE
            // we mutate MLS state. If priming fails (non-web hosts use
            // the default no-op), continue — the host's transport will
            // either route some other way or surface a 4xx on the
            // welcome send and we'll catch it below.
            let _ = self
                .transport
                .set_next_welcome_recipients(conv_id, vec![new_device_id.clone()])
                .await;

            let entry = (new_device_id.clone(), kp_bytes);
            let outcome_result = {
                let mut guard = self.conversations.write();
                match guard.get_mut(&conv_id) {
                    Some(convo) => convo.add_members(vec![entry], now_ms),
                    None => Err(Error::UnknownConversation(conv_id.as_hex())),
                }
            };

            let outcome = match outcome_result {
                Ok(o) => o,
                Err(e) => {
                    outcomes.push(AdmitChatOutcome {
                        conversation_id: conv_id,
                        status: AdmitChatStatus::Failed {
                            error: e.to_string(),
                        },
                    });
                    continue;
                }
            };

            if let Err(e) = self.transport.send(outcome.commit).await {
                outcomes.push(AdmitChatOutcome {
                    conversation_id: conv_id,
                    status: AdmitChatStatus::Failed {
                        error: format!("commit send: {e}"),
                    },
                });
                continue;
            }
            if let Err(e) = self.transport.send(outcome.welcome).await {
                outcomes.push(AdmitChatOutcome {
                    conversation_id: conv_id,
                    status: AdmitChatStatus::Failed {
                        error: format!("welcome send: {e}"),
                    },
                });
                continue;
            }

            // Capture the snapshot under the read guard, drop it, then
            // flush async (never hold the lock across `.await`).
            let snap_result = self
                .conversations
                .read()
                .get(&conv_id)
                .map(|c| c.snapshot_inputs())
                .transpose();
            let flush_result = match snap_result {
                Ok(Some(snap)) => snap.flush().await,
                Ok(None) => Ok(()),
                Err(e) => Err(e),
            };
            if let Err(e) = flush_result {
                // Snapshot failure is non-fatal for the join — the MLS adds
                // already shipped — but record it so the host can decide
                // whether to retry. The next successful send/process will
                // re-snapshot anyway.
                outcomes.push(AdmitChatOutcome {
                    conversation_id: conv_id,
                    status: AdmitChatStatus::Failed {
                        error: format!("snapshot: {e}"),
                    },
                });
                continue;
            }

            outcomes.push(AdmitChatOutcome {
                conversation_id: conv_id,
                status: AdmitChatStatus::Admitted,
            });
        }
        Ok(outcomes)
    }

    pub async fn remove_members(
        &self,
        conv_id: ConversationId,
        leaf_indexes: Vec<u32>,
        now_ms: u64,
    ) -> Result<()> {
        // Send-then-merge — see `add_members` for the full rationale.
        let staged = {
            let mut guard = self.conversations.write();
            let convo = guard
                .get_mut(&conv_id)
                .ok_or_else(|| Error::UnknownConversation(conv_id.as_hex()))?;
            convo.stage_remove_members(leaf_indexes, now_ms)?
        };

        if let Err(send_err) = self.transport.send(staged.commit.clone()).await {
            let merged = {
                let mut guard = self.conversations.write();
                match guard.get_mut(&conv_id) {
                    Some(convo) if is_definite_rejection(&send_err) => {
                        let _ = convo.abort_staged();
                        false
                    }
                    Some(convo) => {
                        convo.confirm_staged(&staged, now_ms)?;
                        true
                    }
                    None => false,
                }
            };
            if merged {
                self.flush_conversation(&conv_id).await?;
            }
            return Err(send_err);
        }

        {
            let mut guard = self.conversations.write();
            let convo = guard
                .get_mut(&conv_id)
                .ok_or_else(|| Error::UnknownConversation(conv_id.as_hex()))?;
            convo.confirm_staged(&staged, now_ms)?;
        }
        self.flush_conversation(&conv_id).await?;
        Ok(())
    }

    /// Re-admit a device, first evicting any existing leaf that duplicates the
    /// new KeyPackage's signature key (the phrase-restore case — see
    /// [`Conversation::duplicate_signature_key_leaves`] and
    /// `docs/specs/re-admit-device.md`). Equivalent to [`Self::add_members`] when
    /// there is no duplicate, so it is a strict superset — safe to prefer on the
    /// recovery re-admission path.
    ///
    /// It composes the two conformance-tested membership primitives —
    /// `remove_members` (evict the dead duplicate leaf, freeing its signing key)
    /// then `add_members` (admit the fresh device + ship its Welcome) — each with
    /// its own send-then-merge rollback, so a server-rejected Commit never leaves
    /// the local epoch ahead of the server. Two Commits on the rare recovery path;
    /// folding them into a single combined Remove+Add commit is a possible future
    /// optimization (kept out of scope to reuse already-vetted primitives). If the
    /// remove succeeds but the add fails, the device is simply un-admitted (no
    /// worse than before) and the caller retries.
    pub async fn re_admit_device(
        &self,
        conv_id: ConversationId,
        entry: (DeviceId, Vec<u8>),
        now_ms: u64,
    ) -> Result<()> {
        let dup_leaves = {
            let guard = self.conversations.read();
            let convo = guard
                .get(&conv_id)
                .ok_or_else(|| Error::UnknownConversation(conv_id.as_hex()))?;
            convo.duplicate_signature_key_leaves(&entry.1)?
        };
        if !dup_leaves.is_empty() {
            self.remove_members(conv_id, dup_leaves, now_ms).await?;
        }
        self.add_members(conv_id, vec![entry], now_ms).await?;
        Ok(())
    }

    /// Leave a conversation. Broadcasts a self-Remove PROPOSAL (MLS doesn't
    /// allow committing your own removal — a remaining member commits it via
    /// [`Self::commit_pending_proposals`]). After this returns, the host should
    /// delete the conversation locally; the leaver remains a cryptographic
    /// member only until a peer commits the proposal, at which point the server
    /// stops delivering to this device.
    pub async fn leave_conversation(&self, conv_id: ConversationId, now_ms: u64) -> Result<()> {
        let proposal = {
            let mut guard = self.conversations.write();
            let convo = guard
                .get_mut(&conv_id)
                .ok_or_else(|| Error::UnknownConversation(conv_id.as_hex()))?;
            convo.leave_group(now_ms)?
        };
        // A proposal doesn't change the epoch, so there's nothing to roll back
        // on a send failure — surface the error and let the host retry.
        self.transport.send(proposal).await?;
        self.flush_conversation(&conv_id).await?;
        Ok(())
    }

    /// Drop a conversation's ENTIRE local state with **no** network side effect.
    ///
    /// Unlike [`Self::leave_conversation`] (which broadcasts a self-Remove
    /// proposal so a remaining member evicts you), this is a purely LOCAL
    /// teardown for a group the server does not back — e.g. an invite that minted
    /// the MLS group locally but never completed server-side, so the backend
    /// 404s `fetchSince` / 403s the member roster for it. The host detects that
    /// authoritative "not a member" verdict and calls this so the dead group
    /// stops rehydrating on every restart (and re-materialising as a ghost
    /// conversation). No envelope is sent — there is no live group to send to.
    ///
    /// Deletes the OpenMLS group state AND the host-side snapshot rows
    /// (`groups/{id}/…`, `cursors/{id}`, `device_leaves/{id}`) that
    /// [`Self::rehydrate_conversations`] would otherwise reload, then drops the
    /// in-memory handle + `stranded` marker. Idempotent: an unknown id still
    /// best-effort purges any orphan storage rows, and calling it twice is safe.
    pub async fn drop_conversation_local(&self, conv_id: ConversationId) -> Result<()> {
        // Take the handle OUT of the map first, then delete its OpenMLS state via
        // the owned value — mirrors the self-removal teardown in
        // `process_envelope` and keeps no lock across the awaits below.
        let existing = self.conversations.write().remove(&conv_id);
        if let Some(mut convo) = existing {
            if let Err(e) = convo.delete_group_state() {
                // Best-effort: dropping the in-memory handle + purging the
                // snapshot rows below already makes the group non-rehydratable.
                tracing::warn!(error = %e, "drop_conversation_local: delete group state failed");
            }
        }
        self.stranded.write().remove(&conv_id);

        // Purge the host-side snapshot rows so a restart's
        // `rehydrate_conversations` (which walks `groups/{id}/meta`) can't bring
        // the group back. Best-effort — deleting an absent key is a no-op on
        // every backend. Sweep the whole `groups/{id}/` prefix so any future
        // sub-key is covered, then the known `cursors` + `device_leaves` rows.
        let hex = conv_id.as_hex();
        if let Ok(keys) = self.storage.list_keys("groups", &format!("{hex}/")).await {
            for k in keys {
                let _ = self.storage.delete("groups", &k).await;
            }
        }
        let _ = self.storage.delete("cursors", &hex).await;
        let _ = self.storage.delete("device_leaves", &hex).await;
        Ok(())
    }

    /// Conversations with buffered pending proposals (e.g. a peer's leave
    /// proposal awaiting a Commit). The host polls this after a sync and, if it
    /// is the designated committer, calls [`Self::commit_pending_proposals`] to
    /// evict the leaver. Sorted for determinism.
    pub fn conversations_with_pending_proposals(&self) -> Vec<ConversationId> {
        let guard = self.conversations.read();
        let mut ids: Vec<ConversationId> = guard
            .iter()
            .filter(|(_, c)| c.has_pending_proposals())
            .map(|(id, _)| *id)
            .collect();
        ids.sort_by_key(|a| a.0);
        ids
    }

    /// Commit all buffered pending proposals for a conversation (evicts a peer
    /// who left). No-op (Ok) when nothing is pending. Send-then-merge with
    /// rollback like add/remove so a server-rejected Commit doesn't desync the
    /// epoch — on an `epoch_advanced` rejection the host should re-sync (another
    /// member already committed) and the pending proposal will have cleared.
    pub async fn commit_pending_proposals(
        &self,
        conv_id: ConversationId,
        now_ms: u64,
    ) -> Result<()> {
        let staged = {
            let mut guard = self.conversations.write();
            let convo = guard
                .get_mut(&conv_id)
                .ok_or_else(|| Error::UnknownConversation(conv_id.as_hex()))?;
            match convo.stage_commit_pending_proposals(now_ms)? {
                Some(s) => s,
                None => return Ok(()),
            }
        };

        if let Err(send_err) = self.transport.send(staged.commit.clone()).await {
            let merged = {
                let mut guard = self.conversations.write();
                match guard.get_mut(&conv_id) {
                    Some(convo) if is_definite_rejection(&send_err) => {
                        let _ = convo.abort_staged();
                        false
                    }
                    Some(convo) => {
                        convo.confirm_staged(&staged, now_ms)?;
                        true
                    }
                    None => false,
                }
            };
            if merged {
                self.flush_conversation(&conv_id).await?;
            }
            return Err(send_err);
        }

        {
            let mut guard = self.conversations.write();
            let convo = guard
                .get_mut(&conv_id)
                .ok_or_else(|| Error::UnknownConversation(conv_id.as_hex()))?;
            convo.confirm_staged(&staged, now_ms)?;
        }
        self.flush_conversation(&conv_id).await?;
        Ok(())
    }

    /// Process an inbound envelope coming from the transport's subscribe callback or a sync pull.
    /// Returns `Some` for application traffic, `None` for handshake messages (already merged).
    ///
    /// LIVE path: applies the envelope to the in-memory MLS working set and then
    /// durably flushes THIS envelope before returning (per-envelope crash-safety
    /// for streaming). The catch-up drain (`sync_conversations`) instead applies a
    /// whole page in-memory and flushes ONCE per page — see `apply_envelope_in_memory`.
    pub async fn process_envelope(
        &self,
        env: &MessageEnvelope,
        now_ms: u64,
    ) -> Result<Option<IncomingMessage>> {
        let (out, removed) = self.apply_envelope_in_memory(env, now_ms)?;
        // A self-removal already deleted the group's storage; nothing to flush.
        if !removed {
            self.flush_conversation(&env.conversation_id).await?;
        }
        Ok(out)
    }

    /// Apply an inbound envelope to the IN-MEMORY MLS working set and advance the
    /// in-memory cursor, WITHOUT the durable checkpoint flush. Returns the
    /// decrypted application message (if any) and whether the conversation was
    /// REMOVED (a Commit that removed our own leaf tears the group down and
    /// deletes its storage synchronously — a removed conversation needs no flush).
    ///
    /// The write guard is taken and dropped entirely within this synchronous
    /// helper — NO `.await` happens while it is held. That is the wasm
    /// parking-panic constraint: previously the write guard was held across
    /// `snapshot_to_storage().await`, and on the single-threaded wasm worker a
    /// concurrent reader that landed while a writer was waiting made `parking_lot`
    /// park → panic "Parking not supported". Callers `.await` the flush AFTER this
    /// returns (`process_envelope` per call; `sync_conversations` once per page).
    ///
    /// Batching the flush is the drain-wedge fix: `flush` writes a FULL-state
    /// checkpoint (the whole MLS store as one blob), so flushing per envelope made
    /// a backlog drain O(N × total_state) of synchronous serialization on the
    /// host's single serial queue — the stall the client saw as "messages
    /// delivered late". Crash-safety is unchanged: `flush` persists MLS state
    /// before the cursor (see `ConversationSnapshot::flush`), so a crash mid-page
    /// re-fetches the page and `SyncCursor::is_new` dedups the already-applied
    /// events.
    fn apply_envelope_in_memory(
        &self,
        env: &MessageEnvelope,
        now_ms: u64,
    ) -> Result<(Option<IncomingMessage>, bool)> {
        // Welcome envelopes for unknown conversations are routed to
        // `join_conversation` by the caller. Here we only handle traffic for
        // already-open groups.
        let mut guard = self.conversations.write();
        let convo = match guard.get_mut(&env.conversation_id) {
            Some(c) => c,
            None => return Err(Error::UnknownConversation(env.conversation_id.as_hex())),
        };
        let out = convo.process(env, now_ms)?;
        // SELF-REMOVAL: if this Commit removed our own leaf the group is now
        // Inactive. Tear it down — delete the OpenMLS state + drop the handle
        // — so the conversation becomes UNKNOWN to this client. A later
        // re-invite Welcome for the same group_id is then routed to
        // `join_conversation` (the host only joins UNKNOWN conversations) and
        // re-joins from clean storage, instead of being suppressed as an
        // "already-joined duplicate" (the cause of the stuck-un-joined re-invite).
        if !convo.is_active() {
            let conv_id = env.conversation_id;
            // Best-effort storage cleanup; even if it fails, dropping the
            // in-memory handle already makes the conversation re-joinable.
            if let Err(e) = convo.delete_group_state() {
                tracing::warn!(error = %e, "process_envelope: delete removed group state failed");
            }
            guard.remove(&conv_id);
            self.stranded.write().remove(&conv_id);
            return Ok((out, true));
        }
        Ok((out, false))
    }

    /// Catch-up sync: pull missing events for every open conversation since its cursor.
    /// Returns the list of newly-decrypted application messages, in apply order.
    pub async fn sync_conversations(&self, now_ms: u64) -> Result<Vec<IncomingMessage>> {
        // Snapshot the conversation IDs ONLY — not their cursors. The cursor is
        // re-read fresh per fetch below. Two bugs this avoids:
        //   1. Stale-cursor pagination: `process_envelope` advances the
        //      conversation's cursor as it applies each page, but the OLD code
        //      kept fetching from the cursor captured up-front — so a chat with
        //      more than one page (256+) of backlog re-fetched the SAME first
        //      page forever and never paged past it (catch-up silently truncated
        //      a freshly-linked device's history, incl. group-avatar/name
        //      hydration that lands after the first page).
        //   2. Join-during-sync: reading IDs fresh here (and re-reading the map
        //      each iteration) means a conversation the host joins via a live
        //      Welcome right before/around this call is still covered.
        // Iterate a sorted, de-duplicated ID set for deterministic order.
        let conversation_ids: Vec<ConversationId> = {
            let guard = self.conversations.read();
            let mut ids: Vec<ConversationId> = guard.keys().copied().collect();
            ids.sort_by_key(|a| a.0);
            ids
        };

        let mut delivered = Vec::new();
        for conv_id in conversation_ids {
            loop {
                // Re-read the LIVE cursor each iteration so pagination advances
                // as `process_envelope` consumes pages. If the conversation was
                // removed mid-sync (e.g. a wipe), stop cleanly.
                let cursor = match self.conversations.read().get(&conv_id) {
                    Some(c) => c.cursor.clone(),
                    None => break,
                };
                // PER-CONVERSATION ISOLATION: a transport error on ONE
                // conversation (a transient 5xx, a non-404 fetch failure) must
                // not abort catch-up for ALL the others. Previously a single
                // erroring conversation propagated `?` and failed the entire
                // `sync_conversations` — so on a freshly-linked device, one bad
                // conversation (e.g. the device group, or a group mid-churn)
                // blocked every chat from syncing, and group name/avatar
                // hydration broadcasts never arrived. Log + skip this
                // conversation instead.
                let batch = match self.transport.fetch_since(conv_id, cursor, 256).await {
                    Ok(b) => b,
                    Err(e) => {
                        tracing::warn!(error = %e, "sync_conversations: fetch_since failed; skipping conversation");
                        break;
                    }
                };
                if batch.is_empty() {
                    break;
                }
                let mut advanced = false;
                let mut removed = false;
                for env in &batch {
                    // PER-ENVELOPE ISOLATION: a single undecryptable / malformed
                    // / wrong-epoch envelope must not drop the rest of the page
                    // (or fail the sync). Skip it; the live stream / a later
                    // epoch advance can still deliver retriable ones.
                    //
                    // BATCHED CHECKPOINT (drain-wedge fix): apply each envelope to
                    // the in-memory working set WITHOUT flushing, then persist the
                    // whole page with ONE full-state checkpoint below. This turns a
                    // backlog drain from O(N × total_state) synchronous
                    // serializations into O(1) per page (~100-1000× less work on
                    // the host's single serial queue) — the stall the client saw
                    // as "messages delivered late".
                    match self.apply_envelope_in_memory(env, now_ms) {
                        Ok((msg, rm)) => {
                            // Commits / handshakes advance the in-memory cursor too
                            // (no app message surfaced) — treat them as progress so
                            // we keep paging instead of re-fetching the same page.
                            advanced = true;
                            if let Some(msg) = msg {
                                delivered.push(msg);
                            }
                            if rm {
                                // A self-removal Commit deleted the conversation
                                // mid-page; stop applying further envelopes to it.
                                removed = true;
                                break;
                            }
                        }
                        Err(e) => {
                            tracing::warn!(error = %e, "sync_conversations: process_envelope failed; skipping envelope");
                        }
                    }
                }
                if advanced {
                    // Progress was made — this conversation is not stranded
                    // (clear any prior stranded mark).
                    self.stranded.write().remove(&conv_id);
                }
                // ONE durable checkpoint for the whole page. Skipped when the
                // conversation was self-removed (its storage was already deleted).
                // Crash-safety is preserved by `flush`'s state-before-cursor
                // ordering: a crash before/inside this flush re-fetches the page on
                // restart and `SyncCursor::is_new` dedups the already-applied events.
                if advanced && !removed {
                    if let Err(e) = self.flush_conversation(&conv_id).await {
                        tracing::warn!(error = %e, "sync_conversations: batch checkpoint failed; will re-fetch on next sync");
                        break;
                    }
                }
                if removed {
                    break; // conversation gone; nothing more to page
                }
                if batch.len() < 256 {
                    break; // partial page → caught up
                }
                if !advanced {
                    // Full page but nothing advanced the cursor (every envelope
                    // errored / was wrong-epoch). The group can no longer advance
                    // from local state — a Commit was missed. Mark it STRANDED so
                    // the host can recover it (re-Welcome / snapshot import) via
                    // `stranded_conversations()`, and bail to avoid an infinite
                    // re-fetch loop.
                    tracing::warn!(
                        "sync_conversations: full page made no progress; marking conversation stranded"
                    );
                    self.stranded.write().insert(conv_id);
                    break;
                }
            }
        }
        Ok(delivered)
    }

    /// Rehydrate conversations from storage on startup ([CR-4]).
    ///
    /// Walks the host-side `groups` namespace for meta records, pairs each with its
    /// cursor + device→leaf map, and asks `Conversation::load` to re-attach to the
    /// underlying OpenMLS group state. The MLS state itself was persisted by the
    /// SQLite-backed `PersistentMlsProvider` on the previous run; this method
    /// reconciles the SDK-side caches with what's on disk.
    async fn rehydrate_conversations(self: &Arc<Self>, now_ms: u64) -> Result<()> {
        let metas = self.storage.list_keys("groups", "").await?;
        for path in metas {
            // path looks like "{convId}/meta"
            let Some((id_hex, suffix)) = path.split_once('/') else {
                continue;
            };
            if suffix != "meta" {
                continue;
            }
            let Some(meta_bytes) = self.storage.get("groups", &path).await? else {
                continue;
            };
            let meta: ConversationMeta = match codec::decode(&meta_bytes) {
                Ok(m) => m,
                Err(_) => continue,
            };
            let cursor_bytes = self
                .storage
                .get("cursors", id_hex)
                .await?
                .unwrap_or_default();
            let cursor = if cursor_bytes.is_empty() {
                SyncCursor::default()
            } else {
                SyncCursor::decode(&cursor_bytes).unwrap_or_default()
            };

            // [CR-2] device→leaf map was persisted alongside meta + cursor.
            let device_leaves_bytes = self
                .storage
                .get("device_leaves", id_hex)
                .await?
                .unwrap_or_default();
            let device_leaves: std::collections::BTreeMap<DeviceId, u32> =
                if device_leaves_bytes.is_empty() {
                    std::collections::BTreeMap::new()
                } else {
                    let pairs: Vec<(DeviceId, u32)> =
                        codec::decode(&device_leaves_bytes).unwrap_or_default();
                    pairs.into_iter().collect()
                };

            match Conversation::load(
                meta.id,
                meta.clone(),
                cursor,
                device_leaves,
                self.local_device.device_id.clone(),
                self.crypto.clone(),
                self.signing.clone(),
                self.storage.clone(),
                now_ms,
            ) {
                Ok(Some(convo)) => {
                    tracing::debug!(
                        target: "ping_core::client",
                        convo = %id_hex,
                        epoch = meta.epoch,
                        "rehydrated conversation from disk"
                    );
                    self.conversations.write().insert(meta.id, convo);
                }
                Ok(None) => {
                    tracing::warn!(
                        target: "ping_core::client",
                        convo = %id_hex,
                        "host-side meta present but OpenMLS state missing — skipping"
                    );
                }
                Err(e) => {
                    tracing::warn!(
                        target: "ping_core::client",
                        convo = %id_hex,
                        error = %e,
                        "Conversation::load failed — skipping"
                    );
                }
            }
        }
        Ok(())
    }

    // ------------------- Multi-device API -------------------

    /// Build a [`LinkingTicket`] for a new device. The caller obtains `new_device_kp` from the
    /// new device (e.g., via QR-encoded handshake) and is responsible for sealing the returned
    /// ticket against the new device's ephemeral X25519 pubkey before transmission via
    /// [`ping_link::seal_ticket`].
    ///
    /// [CR-13] `last_app_events` is a host-supplied list of `(conversation_id, app_event_bytes)`
    /// for the new device's "what you missed" UI. The SDK adds its own metas + (currently-
    /// empty) per-conversation MLS state and bundles everything into
    /// [`device::CatchupSnapshot`], CBOR-encoded into the ticket's `catchup_snapshot` field.
    /// Pass an empty `Vec` to suppress catchup data (the new device sees an empty
    /// conversation list until normal sync runs).
    pub async fn build_linking_ticket(
        self: &Arc<Self>,
        new_device_id: DeviceId,
        new_device_kp: Vec<u8>,
        last_app_events: Vec<(ConversationId, Vec<u8>)>,
        now_ms: u64,
    ) -> Result<LinkingTicket> {
        let device_binding_sig = self.identity.sign_device_binding(&new_device_id.0);
        let dg_id = device_group_id_for(self.identity.user_id());

        // [CR-10] DG is eagerly created at init now, but call ensure here too so
        // hosts that bypass `MessagingClient::init` (mocked tests, legacy upgrade
        // paths) keep working.
        self.ensure_device_group(now_ms).await?;

        // Admit the new device to the DeviceGroup.
        let outcome = {
            let mut conversations = self.conversations.write();
            // `ensure_device_group` above creates it, but return a typed error
            // instead of panicking (which would unwind across FFI/wasm) on the
            // pathological case where it's still missing.
            let dg = conversations.get_mut(&dg_id).ok_or_else(|| {
                Error::Invalid("device group missing after ensure_device_group".into())
            })?;
            // [CR-2] Record the new device's leaf in the DG so future `revoke_device`
            // can find it. The new_device_id we got as a parameter is the inviter's
            // own assertion — same trust model as the rest of `add_members`.
            dg.add_members(vec![(new_device_id.clone(), new_device_kp)], now_ms)?
        };

        // [CR-13] Assemble the catchup snapshot: SDK-known conversation metadata + host-
        // supplied last-known plaintext per conversation. [CR-7] now populates
        // `group_state_bytes` with each group's MLS state so the new device can decrypt
        // historical traffic without re-Welcoming. An empty `group_state_bytes` would
        // mean either a group with no exportable state (shouldn't happen) or an
        // encoder failure (we let those propagate as errors below).
        let catchup_snapshot = if last_app_events.is_empty() && self.conversations.read().is_empty()
        {
            // Cheap path: nothing to snapshot, skip the encode round-trip.
            Vec::new()
        } else {
            // CR-7 per-group state export is O(conversations) expensive MLS
            // serialization AND unconsumed by current hosts (both re-admit the new
            // device via the post-link re-Welcome reconciler rather than importing
            // this state). Exporting every group's state for an account with hundreds
            // or thousands of conversations is wasted CPU that also blows the 256 KB
            // ticket cap. So budget the EXPORT itself: fill `group_state_bytes` only
            // while under a small byte budget, in iteration order; once exhausted, ship
            // the (cheap) meta with empty `group_state_bytes` and skip the expensive
            // export entirely — the receiver falls back to the normal re-Welcome path
            // for those. `encode_within_cap` below is the final guarantee the whole
            // snapshot fits regardless. This keeps the per-link cost bounded to a
            // CONSTANT (~budget) instead of growing with conversation count.
            const GROUP_STATE_EXPORT_BUDGET: usize = CATCHUP_SNAPSHOT_SOFT_CAP; // 64 KiB
            let mut group_state_budget = GROUP_STATE_EXPORT_BUDGET;
            let conversation_metas: Vec<CatchupConversationEntry> = {
                let guard = self.conversations.read();
                let mut metas = Vec::with_capacity(guard.len());
                for c in guard.values() {
                    let group_state_bytes = if group_state_budget > 0 {
                        let bytes = c.export_state_snapshot(now_ms)?.to_vec();
                        if bytes.len() <= group_state_budget {
                            group_state_budget -= bytes.len();
                            bytes
                        } else {
                            // This one would overflow the budget — drop it and stop
                            // exporting further (later conversations skip the export).
                            group_state_budget = 0;
                            Vec::new()
                        }
                    } else {
                        Vec::new()
                    };
                    metas.push(CatchupConversationEntry {
                        conversation_id: c.id(),
                        meta: c.meta().clone(),
                        group_state_bytes,
                    });
                }
                metas
            };
            let last_app_events_per_conv: Vec<CatchupAppEventEntry> = last_app_events
                .into_iter()
                .map(|(conversation_id, app_event_bytes)| CatchupAppEventEntry {
                    conversation_id,
                    app_event_bytes,
                })
                .collect();
            // `encode_within_cap` (not `encode`) so a user with many/large groups can
            // still link: the snapshot snapshots EVERY conversation, so a big account
            // would otherwise blow the 256 KB ticket cap and hard-fail linking outright.
            // It sheds per-group MLS state (largest first) to fit — those conversations
            // catch up via the post-link re-Welcome reconciler the hosts already run.
            CatchupSnapshot {
                v: CATCHUP_SNAPSHOT_VERSION,
                conversation_metas,
                last_app_events_per_conv,
            }
            .encode_within_cap()?
        };

        Ok(LinkingTicket {
            v: 1,
            user_id: self.identity.user_id().clone(),
            user_pubkey: self.identity.public_key().to_bytes().to_vec(),
            new_device_id,
            device_binding_sig,
            device_group_welcome: outcome.welcome.payload,
            catchup_snapshot,
        })
    }

    /// Apply a received linking ticket. Joins the user's DeviceGroup; the catch-up snapshot
    /// (if any) is decrypted by the host using the standard per-conversation channel afterwards.
    pub async fn consume_linking_ticket(
        self: &Arc<Self>,
        ticket: &LinkingTicket,
        now_ms: u64,
    ) -> Result<()> {
        // Verify the binding the existing device made for us. (Ed25519 public keys are 32 bytes.)
        let pk_bytes: [u8; 32] = ticket
            .user_pubkey
            .as_slice()
            .try_into()
            .map_err(|_| Error::Identity("user_pubkey must be 32 bytes".into()))?;
        let user_pk = ed25519_dalek::VerifyingKey::from_bytes(&pk_bytes)
            .map_err(|e| Error::Identity(format!("bad user pubkey: {e}")))?;
        Identity::verify_device_binding(
            &user_pk,
            &ticket.user_id,
            &ticket.new_device_id.0,
            &ticket.device_binding_sig,
        )?;
        if ticket.new_device_id != self.local_device.device_id {
            return Err(Error::Invalid(
                "ticket addressed to a different device".into(),
            ));
        }

        let dummy_env = MessageEnvelope::new(
            ConversationId(device_group_id_for(&ticket.user_id).0),
            0,
            MessageKind::Welcome,
            self.local_device.device_id.clone(),
            0,
            crate::clock::Hlc::ZERO,
            ticket.device_group_welcome.clone(),
        );
        self.join_conversation(&dummy_env, now_ms).await?;
        Ok(())
    }

    /// [CR-7] Export the MLS state snapshot for one open conversation.
    ///
    /// Thin pass-through to [`Conversation::export_state_snapshot`]. Returned bytes
    /// are wrapped in `Zeroizing` because they contain past epoch secrets.
    pub fn export_conversation_state_snapshot(
        &self,
        conv_id: ConversationId,
        now_ms: u64,
    ) -> Result<zeroize::Zeroizing<Vec<u8>>> {
        let guard = self.conversations.read();
        let convo = guard
            .get(&conv_id)
            .ok_or_else(|| Error::UnknownConversation(conv_id.as_hex()))?;
        convo.export_state_snapshot(now_ms)
    }

    /// [CR-7] Import a `GroupStateSnapshot` produced by another device's
    /// [`Conversation::export_state_snapshot`].
    ///
    /// Replays the snapshot's entries into this client's OpenMLS provider, then
    /// reconstructs the `Conversation` handle via `MlsGroup::load`. After return,
    /// the conversation is in `list_conversations()` and `send`/`process_envelope`
    /// work against it normally.
    ///
    /// **Scope.** This is for the *same-user* hand-off (linking, recovery). The
    /// snapshot exposes the exporter's view of past epoch secrets for the target
    /// group; only call this when the receiving device has been authenticated to
    /// the same user identity (mnemonic, QR-handshake). Cross-user history transfer
    /// uses HPKE-sealed AppEvent re-shares (umbrella §15.6), not this method.
    ///
    /// **Sanity.** Refuses snapshots whose `group_id` doesn't match the bytes the
    /// receiver intends to claim — guards against host bugs that shuffle snapshots
    /// between groups. Refuses mismatched OpenMLS storage versions outright; no
    /// silent forward/back compatibility.
    pub async fn import_state_snapshot(
        self: &Arc<Self>,
        snapshot_bytes: &[u8],
        now_ms: u64,
    ) -> Result<ConversationId> {
        use crate::device::GroupStateSnapshot;
        let snap = GroupStateSnapshot::decode(snapshot_bytes)
            .map_err(|e| Error::Invalid(format!("snapshot decode: {e}")))?;

        if snap.openmls_storage_version != openmls_traits::storage::CURRENT_VERSION {
            return Err(Error::Invalid(format!(
                "snapshot openmls_storage_version={} not supported (this SDK supports v={})",
                snap.openmls_storage_version,
                openmls_traits::storage::CURRENT_VERSION
            )));
        }

        let conv_id = snap.group_id;

        // Refuse if we already have an active handle for this conv — the host should
        // close it first, otherwise import silently overwrites in-memory state and
        // the existing handle becomes stale.
        if self.conversations.read().contains_key(&conv_id) {
            return Err(Error::Invalid(format!(
                "conversation {} already open; close before importing snapshot",
                conv_id.as_hex()
            )));
        }

        // Replay raw KV pairs into the provider's working set.
        let entries: Vec<(Vec<u8>, Vec<u8>)> =
            snap.entries.into_iter().map(|e| (e.key, e.value)).collect();
        self.crypto
            .import_entries(entries)
            .map_err(|e| Error::Storage(format!("import entries: {e}")))?;

        // Reconstruct the Conversation handle. `Conversation::load` will return
        // `Ok(None)` if OpenMLS still can't find the group — i.e. our snapshot was
        // incomplete or for a different storage version.
        let meta = ConversationMeta {
            id: conv_id,
            name: None,
            epoch: 0, // will be overwritten from the loaded group state in process()
            member_count: 0,
            is_device_group: false, // host can flip this via meta update if needed
            created_at_ms: now_ms,
        };
        let convo = Conversation::load(
            conv_id,
            meta,
            SyncCursor::default(),
            std::collections::BTreeMap::new(),
            self.local_device.device_id.clone(),
            self.crypto.clone(),
            self.signing.clone(),
            self.storage.clone(),
            now_ms,
        )?
        .ok_or_else(|| {
            Error::Invalid(
                "snapshot imported but OpenMLS could not load the group — snapshot may be incomplete or storage version mismatched"
                    .into(),
            )
        })?;

        // Pull the live epoch + member count from the loaded group so the meta we
        // just stubbed is consistent with what we'll observe on subsequent process_envelope.
        let live_epoch = convo.epoch();
        let live_members = convo.group.members().count() as u32;
        let live_name = convo.name_from_group_state();
        let mut convo = convo;
        convo.meta.epoch = live_epoch;
        convo.meta.member_count = live_members;
        // Recover the name from the loaded GroupContext state (a snapshot import
        // is join-equivalent; the stubbed `name: None` would otherwise stick).
        convo.meta.name = live_name;
        convo.snapshot_to_storage().await?;

        self.conversations.write().insert(conv_id, convo);
        Ok(conv_id)
    }

    /// Export a derived secret from one conversation's MLS exporter ([CR-8]).
    ///
    /// Thin pass-through to [`Conversation::export_secret`]. See that method's doc comment
    /// for the contract on `label`, `context`, length validation, and zeroization. The
    /// returned `Zeroizing<Vec<u8>>` is automatically wiped when dropped.
    pub fn export_conversation_secret(
        &self,
        conv_id: ConversationId,
        label: &str,
        context: &[u8],
        length: usize,
    ) -> Result<Zeroizing<Vec<u8>>> {
        let guard = self.conversations.read();
        let convo = guard
            .get(&conv_id)
            .ok_or_else(|| Error::UnknownConversation(conv_id.as_hex()))?;
        convo.export_secret(label, context, length)
    }

    /// Revoke a device by removing its leaf from every conversation where we know its
    /// position ([CR-2]).
    ///
    /// Returns one Commit envelope per conversation the device was a leaf in. The host
    /// broadcasts each envelope to the affected conversation; the SDK has also already
    /// handed them to the transport via `transport.send` (idempotent broadcast is the
    /// host's call).
    ///
    /// **Scope.** The SDK can only resolve leaves it recorded itself — either when it
    /// admitted the device via [`Self::add_members`] or when this device joined as the
    /// target via Welcome. For peer-admitted devices the leaf index isn't locally known;
    /// those conversations are silently skipped. The host can fall back to
    /// `remove_members(leaf_index)` directly using a transport-side directory lookup if
    /// it needs to revoke from those conversations too. See
    /// `docs/architecture/multi-device.md §Device removal` for the broader flow.
    ///
    /// Conversations with no entry for `device_id` produce no envelope; an empty `Vec`
    /// return is a valid outcome (e.g. the device was already revoked, or was never
    /// added by this client).
    #[allow(clippy::await_holding_lock)] // see add_members for rationale
    pub async fn revoke_device(
        &self,
        device_id: DeviceId,
        now_ms: u64,
    ) -> Result<Vec<MessageEnvelope>> {
        // 1. Walk every open conversation and gather (conv_id, leaf_index) pairs where
        //    we know `device_id` controls a leaf. Done under a read lock so we don't hold
        //    the write lock across the per-conversation remove path.
        let targets: Vec<(ConversationId, u32)> = self
            .conversations
            .read()
            .iter()
            .filter_map(|(id, c)| c.leaf_index_of(&device_id).map(|leaf| (*id, leaf)))
            .collect();

        // 2. For each target, emit a remove_members commit. We do this sequentially: each
        //    one is a separate MLS epoch advance on its own group, and they don't share
        //    state, so parallel issuance is safe but adds complexity we don't need for v1.
        let mut envelopes = Vec::with_capacity(targets.len());
        for (conv_id, leaf_index) in targets {
            let envelope = {
                let mut guard = self.conversations.write();
                let convo = guard
                    .get_mut(&conv_id)
                    .ok_or_else(|| Error::UnknownConversation(conv_id.as_hex()))?;
                convo.remove_members(vec![leaf_index], now_ms)?
            };
            self.transport.send(envelope.clone()).await?;
            if let Some(c) = self.conversations.read().get(&conv_id) {
                c.snapshot_to_storage().await?;
            }
            envelopes.push(envelope);
        }

        // 3. Notify the auth-layer server so it can invalidate the
        //    revoked device's KeyPackage pool, mark `auth.devices.revoked_at`,
        //    and refuse any future envelope signed by the revoked device's
        //    JWT. Done AFTER the MLS Commits so peers learn via MLS first
        //    (the canonical path) and the auth layer is the eventual-
        //    consistency cleanup. Transport failures bubble up so callers
        //    can retry — but the MLS-side work has already shipped, so
        //    the device is functionally revoked in every group; only the
        //    auth-layer KeyPackage purge is pending.
        self.transport.revoke_device_remote(device_id).await?;
        Ok(envelopes)
    }
}

fn device_group_id_for(user_id: &UserId) -> ConversationId {
    // Deterministic 16-byte ID derived from the user's id, prefixed so it cannot collide with
    // a randomly-generated ULID in normal use (ULIDs start with a millisecond timestamp).
    let mut bytes = [0u8; 16];
    bytes[0] = 0xFF;
    bytes[1] = 0xDC; // "DeviCe" group sentinel
    let h = codec::sha256(&user_id.0);
    bytes[2..].copy_from_slice(&h[..14]);
    ConversationId(bytes)
}

fn encode_local_device(d: &LocalDevice) -> Result<Vec<u8>> {
    use serde::Serialize;
    #[derive(Serialize)]
    struct Persisted<'a> {
        device_id: &'a DeviceId,
        label: &'a str,
        created_at_ms: u64,
        #[serde(with = "serde_bytes")]
        signing_seed: &'a [u8],
    }
    codec::encode(&Persisted {
        device_id: &d.device_id,
        label: &d.label,
        created_at_ms: d.created_at_ms,
        signing_seed: d.signing.as_bytes(),
    })
}

fn decode_local_device(bytes: &[u8], user_id: UserId) -> Result<LocalDevice> {
    use serde::Deserialize;
    #[derive(Deserialize)]
    struct Persisted {
        device_id: DeviceId,
        label: String,
        created_at_ms: u64,
        #[serde(with = "serde_bytes")]
        signing_seed: Vec<u8>,
    }
    let p: Persisted = codec::decode(bytes)?;
    let seed: [u8; 32] = p
        .signing_seed
        .as_slice()
        .try_into()
        .map_err(|_| Error::Invalid("device signing seed must be 32 bytes".into()))?;
    let signing = ed25519_dalek::SigningKey::from_bytes(&seed);
    Ok(LocalDevice {
        device_id: p.device_id,
        user_id,
        label: p.label,
        signing,
        created_at_ms: p.created_at_ms,
    })
}