offline_protocol/protocol/

mod.rs

//! Main protocol engine.

mod blocking;
mod config_accessors;
mod custodian;
pub(crate) mod custody;
#[cfg(feature = "data")]
pub(crate) mod data;
#[cfg(feature = "data")]
mod data_sync;
mod decryption_queue;
pub(crate) mod mesh_relay;
mod message_dispatch;
mod nostr_discovery;
mod nostr_publication;
mod observability;
mod pending_queue;
mod prefixes;
pub(crate) mod reachability;
mod receive;
mod security;
mod send;
mod service_seal;
mod session;
pub(crate) mod state_crypto;
mod storage;
mod types;

pub use custody::{CustodyRefusal, CustodyStats};
pub(crate) use decryption_queue::PendingDecryptionQueue;
pub use decryption_queue::PendingQueueMetrics;
pub(crate) use prefixes::*;
#[cfg(feature = "file-store")]
pub(crate) use storage::{sealed_state_key_types, stored_state_record_key, StoredRecordKey};
pub(crate) use storage::{PruneAllowance, RestorableRecord};
pub(crate) use types::*;
pub use types::{GatewayCarrier, MediaSendOptions, ProtocolState, SendMessageOptions};

use crate::file_transfer::{FileTransferManager, OutboundTransferState};
use crate::group_mesh::{MAX_RELAY_CAPABILITIES, MAX_RELAY_CAPABILITY_TOKEN_BYTES};
use crate::mls_observability::{MlsEventEmitter, MlsEventRateLimiter, NoopMlsEventEmitter};
use crate::telemetry::aggregator::{
    build_metrics_frame, device_battery_from_available, diff_device_capability,
    diff_transport_state, DeviceSnap,
};
#[cfg(feature = "telemetry-pipe")]
use crate::telemetry::pipe::{
    store::Backend, AppState, PipeParts, TelemetryHost, TelemetryPipe, TelemetryStats,
    FINAL_FLUSH_BUDGET,
};
use crate::telemetry::{
    dispatch_record, Scrubber, TelemetryConfig, TelemetryContext, TelemetryRecord, TelemetrySink,
};
use crate::{
    Error, EstablishmentState, Event, ProtocolConfig, ProtocolStateStorage, Result,
    TransportManager,
};
use chrono::{DateTime, Utc};
use mesh_relay::{MeshRelayGovernor, RelayActivity};
use offline_protocol_core::{LamportClock, Message, MessageId, MutexExt, Username};
use offline_protocol_mls::{EncryptedMessage, MlsManager, MlsStorage, WelcomeMessage};
use offline_protocol_reliability::{AckManager, Deduplicator, RetryQueue};
use offline_protocol_router::{RelayPriority, RelayRole, TransportSelector};
use offline_protocol_services::MeshServices;
use offline_protocol_transport::{BleTransport, TransportStatus, TransportType};
use std::collections::HashMap;
use std::collections::HashSet;
use std::collections::VecDeque;
use std::sync::{Arc, Mutex, RwLock};
use std::time::Instant;
use tracing::{debug, error, info, warn};
use zeroize::Zeroizing;

/// Returns whether `timestamp` is at least `lifetime_ms` old without relying
/// on Chrono's panicking `DateTime - Duration` operator.
///
/// A lifetime too large to represent, or whose cutoff predates Chrono's
/// calendar range, cannot expire a representable timestamp.
fn lifetime_expired(now: DateTime<Utc>, timestamp: DateTime<Utc>, lifetime_ms: u64) -> bool {
    let Ok(lifetime_ms) = i64::try_from(lifetime_ms) else {
        return false;
    };
    now.checked_sub_signed(chrono::Duration::milliseconds(lifetime_ms))
        .is_some_and(|cutoff| timestamp <= cutoff)
}

/// Whether a relay-pushed outbox entry has outlived its fixed redelivery
/// window ([`offline_protocol_reliability::constants::REDELIVERY_WINDOW_MS`]
/// from `first_sent_at`). Neither the sliding window nor the restore refresh
/// moves this: every re-send past it would earn a second visible push, and
/// the recipient no longer recognises the id as a duplicate.
fn relay_push_window_closed(now: DateTime<Utc>, entry: &OutboxEntry) -> bool {
    entry.relay_pushed
        && lifetime_expired(
            now,
            entry.first_sent_at,
            offline_protocol_reliability::constants::REDELIVERY_WINDOW_MS,
        )
}

/// Main entry point for the Offline Protocol SDK.
///
/// This struct combines all protocol components and provides a unified API
/// for sending/receiving messages with automatic transport selection and
/// reliable delivery.
pub struct OfflineProtocol {
    /// Configuration.
    pub(crate) config: ProtocolConfig,

    /// This device's identity on the wire: the address derived from the
    /// identity key in this profile's storage.
    ///
    /// Every outbound frame is stamped with it, every self-filter compares
    /// against it, and it is one half of every 1:1 session slot. It is set
    /// once, by `initialize_mls`, from the key that storage holds — the
    /// protocol cannot know it before then, because storage is what carries
    /// the key and storage does not arrive until that call.
    ///
    /// Until then it holds [`ProtocolConfig::profile`], which is *not* an
    /// address and is never a peer's real identity. That fallback keeps the
    /// pre-MLS surface (transport wiring, local bookkeeping) working on an
    /// instance that has not been initialized.
    ///
    /// One configuration can put the fallback on the wire: the explicit
    /// `require_encryption = false` opt-out lets a send go out in plaintext
    /// with MLS never initialized, and such a frame is stamped with the
    /// profile. It is the same fall-through that already emits
    /// [`crate::events::SecurityWarningCode::PlaintextSend`], so the condition
    /// is reported rather than silent. The default config forbids it —
    /// `require_encryption` is `true`, which fails the send closed.
    pub(crate) local_id: String,

    /// Whether [`Self::local_id`] is the derived address rather than the
    /// profile fallback.
    pub(crate) identity_established: bool,

    /// Transport manager (manages all transports with DORS).
    pub(crate) transport_manager: TransportManager,

    /// ACK manager for tracking acknowledgments.
    ack_manager: AckManager,

    /// Retry queue for failed messages.
    retry_queue: RetryQueue,

    /// Deduplicator for preventing duplicates.
    pub(crate) deduplicator: Deduplicator,

    /// Decides which frames this device carries for other people, when they
    /// go out, and how fast. Kept separate from [`Self::deduplicator`], whose
    /// entries track messages addressed to us and are removed again when
    /// delivery has to be retried.
    pub(crate) mesh_relay: MeshRelayGovernor,

    /// Frames held in custody for a neighbour, with the quotas and the
    /// counters (`docs/spec/custody.md`). Disjoint from
    /// [`Self::mesh_relay`]'s suppression cache by construction: accepting a
    /// frame never records its id there, and redelivering one never consults
    /// it, so a custodian never blanks the route it is holding.
    pub(crate) custody: custody::CustodyStore,

    /// Receipts this device holds as a *depositor*: for each outbox entry,
    /// the custodians holding it and until when a further deposit request
    /// toward each is suppressed, in Unix milliseconds from the receipt's own
    /// timestamp. In memory only, bounded by the outbox and
    /// by [`custody::MAX_CUSTODY_RECEIPTS_PER_MESSAGE`]; losing it at a
    /// restart costs one duplicate deposit, which the custodian absorbs.
    pub(crate) custody_receipts: HashMap<MessageId, HashMap<String, i64>>,

    /// When the custody store was last swept for expired records.
    custody_last_sweep: Instant,

    /// Shared mutable state.
    shared_state: Arc<Mutex<SharedState>>,

    /// Messages awaiting delivery/acknowledgment (store-and-forward outbox).
    outbox: HashMap<MessageId, OutboxEntry>,

    /// Transient staging for outbox re-seal provenance, keyed by message id.
    /// Populated at send time (where the plaintext is in scope) and consumed
    /// when the outbox entry is first created (which happens deep in the send/
    /// retry machinery, from the already-sealed `Message`). Once consumed it
    /// lives on the `OutboxEntry` itself. See [`OutboxReseal`].
    pending_reseal: HashMap<MessageId, OutboxReseal>,

    /// Outbox ids already re-sealed and resent once in answer to an
    /// `undecryptable` ACK. A second such answer leaves the entry to its retry
    /// ladder, so two devices cannot trade undecryptable ACKs and resends at
    /// network speed. Cleared with the outbox entry.
    undecryptable_resent: HashSet<MessageId>,

    /// Dedicated outbox for file chunk messages, separate from the main outbox
    /// to prevent large file transfers from evicting regular messages.
    media_outbox: HashMap<MessageId, OutboxEntry>,

    /// Consecutive unreachable-park counter per recipient for plain DMs,
    /// driving the escalating reachability-probe interval on every carrier
    /// (see `handle_recipient_unreachable_for_message` —
    /// mirrors `WelcomeLifecycleRecord::unreachable_parks`). Reset on every
    /// reachability edge (`flush_outbox_for_peer_via` / `flush_outbox_all`) and
    /// on delivery; pruned alongside `cleanup_outbox` so only peers that
    /// still hold outbox entries can retain a counter. While live, ACK
    /// exhaustion for the recipient's plain DMs re-parks instead of
    /// settling terminally (`try_repark_exhausted_dm`). In-memory only: a
    /// restart re-drives the outbox anyway, which is itself a fresh probe.
    dm_unreachable_parks: HashMap<String, u32>,

    /// Delivery acknowledgements no carrier could take when they were owed,
    /// with the transport the acknowledged frame arrived on and when they
    /// were held. Re-routed on the next carrier-level edge
    /// (`flush_unrouted_acks` from `flush_outbox_all`); bounded by
    /// `MAX_UNROUTED_ACKS` and `UNROUTED_ACK_TTL`. In-memory only.
    unrouted_acks: VecDeque<(Message, TransportType, Instant)>,

    /// Per-recipient reachability claims, keyed by recipient and carrier.
    ///
    /// The generalisation of the counter above: that one records *how often*
    /// a recipient has been parked, this one records *what a carrier said*
    /// about reaching them, so a decision can be made before the send rather
    /// than only after it failed. Written by the same producers (a gateway
    /// verdict, a gateway presence answer), read at the send and carrying
    /// seams. In-memory only, and absent facts mean today's behaviour.
    pub(crate) reachability: reachability::ReachabilityFacts,

    /// Capability tokens the attached gateway daemon advertised.
    ///
    /// Delivered at attach, before the bridge reports the carrier available,
    /// and cleared when the carrier drops: a stale advertisement outlives the
    /// gateway that made it, and a reconnect may land on a different one.
    ///
    /// Stored and never read on a decision. The backbone behind the daemon
    /// is the gateway's own property and reaches this device only as a
    /// `backbone_<kind>_v1` token in this set; the set is for an application
    /// to show, and the first production reader would turn a string set by
    /// whoever holds the socket into a routing input (ADR 0026).
    ///
    /// Kept apart from `group_mesh.relay_capabilities` deliberately, though
    /// both are capability sets from a gateway. That one gates the relay
    /// group-broadcast path, and a token advertised by whatever box is
    /// plugged into the venue Wi-Fi must not be able to open it.
    gateway_capabilities: HashSet<String>,

    /// MLS manager for end-to-end encryption.
    mls_manager: Option<Arc<RwLock<MlsManager>>>,

    /// Pending messages waiting for session establishment (recipient -> messages).
    pending_encrypted_messages: HashMap<String, Vec<PendingMessage>>,

    /// Terminal message settlements produced while restoring, held until the
    /// event pipeline is live.
    ///
    /// Restore runs inside `initialize_mls`, which apps routinely call before
    /// installing an event callback — so a `message_failed` emitted there would
    /// be dropped, and the app would keep an id that never resolves. `start()`
    /// drains this, mirroring how restored media descriptors already wait to be
    /// announced.
    ///
    /// Explicitly capped at [`MAX_DEFERRED_RESTORE_SETTLEMENTS`] rather than
    /// left to the restore caps that feed it. Those caps do bound it, but they
    /// bound it as a *sum* across every category, and nothing drains this until
    /// `start()` — which an app that only ever calls `initialize_mls`, or that
    /// retries it against a failing store, may never reach.
    deferred_restore_settlements: Vec<Event>,

    /// Number of settlements suppressed by the cap above, so the count is
    /// reported even though the individual events are not.
    suppressed_restore_settlements: usize,

    /// Earliest wall-clock expiry in `pending_encrypted_messages`.
    ///
    /// `process()` consults this before scanning the bounded queue, avoiding an
    /// O(N) walk on every 100 ms tick while preserving exact configured expiry.
    /// A stale early deadline is harmless: it causes one extra scan, which then
    /// recomputes the real minimum.
    next_pending_message_expiry: Option<DateTime<Utc>>,

    /// Key packages received but not yet used (sender_id -> package).
    pub(crate) pending_key_packages: HashMap<String, ReceivedKeyPackage>,

    /// Peers we have pushed our key package to, with when we last did and how
    /// many times the push has been re-armed since.
    ///
    /// The instant is what lets a push that produced no session be re-armed
    /// (`Self::rearm_key_package_for_peer`) instead of standing in for a
    /// delivery it never proved; the count is that re-arm's backoff.
    pub(crate) key_package_sent_to: HashMap<String, (Instant, u32)>,

    /// All discovered/connected peers, tracked independently of encryption.
    /// Used by service discovery to know who to broadcast queries to.
    /// Values are last-seen instants: refreshed on every discovery signal,
    /// swept by [`Self::prune_stale_known_peers`] after `KNOWN_PEER_TTL_SECS`,
    /// and used to pick the least-recently-seen victim when an insert hits
    /// `MAX_KNOWN_PEERS`.
    known_peers: HashMap<String, Instant>,

    /// Sessions confirmed established (received Welcome or successful decrypt).
    /// Only encrypt messages when the session is confirmed to avoid race conditions.
    confirmed_sessions: std::collections::HashSet<String>,

    /// Peers that have demonstrated they run MLS, used by
    /// [`Self::accept_plaintext_content`] to reject cleartext claiming to come
    /// from an encryption-capable peer.
    ///
    /// # Why this is not `confirmed_sessions`, and not a storage read
    ///
    /// The gate used to ask `is_session_confirmed`, which reads the
    /// `session_states` protocol-state record. That record lives in the
    /// install-scoped app container, where an attacker with write access can
    /// simply **delete** it: restore then treats absence as `Pending` (it must
    /// — see `bootstrap_missing_session_state`), the peer drops out of
    /// `confirmed_sessions`, and the gate opens for unauthenticated cleartext
    /// carrying an attacker-chosen `sender`. Sealing the record does not help;
    /// a seal authenticates bytes that are present, and this attack removes
    /// them. So the signal has to come from somewhere the app container cannot
    /// reach.
    ///
    /// Both sources here are credential-store-resident: the MLS session list
    /// and the durable capability records
    /// ([`storage_keys::ENCRYPTION_CAPABLE_PEERS`]). Neither is sufficient
    /// alone. A session can exist for a peer that has never signed a control
    /// frame we verified (it is established from key packages, and the Welcome
    /// path marks capability without a signature), and a record can outlive the
    /// session it was learned alongside. The union is what actually tracks
    /// "this peer speaks MLS".
    ///
    /// # Monotone within a run — deliberately, and this is load-bearing
    ///
    /// Entries are added and **never removed on session teardown**. That looks
    /// like a leak and is in fact the whole security property. Teardown is
    /// reachable from an *unauthenticated* frame: an injected `__MLS_ENC__`
    /// classifies as `SessionDesync`, `schedule_session_rekey` tears down our
    /// session and advertises a `session_reset` key package, and the peer's
    /// `handle_key_package_message` then deletes its own. A set that shrank on
    /// teardown would therefore let one injected frame re-open the very gate
    /// this field exists to hold shut — moving the attack from "needs
    /// app-container write access" to "needs one forged packet", which is
    /// strictly worse than the bug it fixes.
    ///
    /// Keeping a peer after teardown costs nothing real: a peer that spoke MLS
    /// a moment ago is still encryption-capable, so its cleartext is still a
    /// downgrade. The invariant to preserve is one-directional — this set may
    /// **over**-approximate (which fails closed) but must never
    /// **under**-approximate (which fails open). Anything that populates it may
    /// be lossy; nothing may prune it.
    ///
    /// There is no longer any way out at all, and that is now *correct* rather
    /// than merely safe. The old escape hatch (`reset_tofu_for_peer`) meant
    /// "this identity has re-keyed, treat it as new" — a thing that could
    /// happen when an identity was a name. Under self-certifying addressing an
    /// address is the hash of its key, so a peer that re-keys becomes a
    /// *different address*: nothing about the old one ever stops being true,
    /// and forgetting it could only ever be wrong.
    ///
    /// Bounded by [`MAX_ENCRYPTION_CAPABLE_PEERS`], enforced as a refusal rather
    /// than an eviction so the monotonicity above survives the cap — see
    /// [`Self::mark_encryption_capable`].
    encryption_capable_peers: std::collections::HashSet<String>,

    /// Peers whose key package advertised the compact MLS envelope
    /// ([`MLS_ENVELOPE_COMPACT_V1`] in `env_versions`), so
    /// `seal_encrypted_content` may emit it instead of the legacy JSON form.
    /// Learned from key-package exchange, persisted per peer as
    /// [`PeerCapabilities`], and repopulated by `restore_peer_capabilities`
    /// on `initialize_mls` — unlike the transport manager's binary-wire
    /// registry, which stays in-memory because direct connections re-exchange
    /// key packages anyway. Bounded like `key_package_sent_to` (it is keyed
    /// by the wire-claimed sender).
    peer_compact_envelope: std::collections::HashSet<String>,

    /// Peers whose key package advertised that they verify the
    /// freshness-bound control payload ([`CTRL_SIGN_V2`] in `ctrl_versions`),
    /// so control frames addressed to them are signed under
    /// `offline-ctrl-v2`. Same lifecycle as `peer_compact_envelope` above:
    /// learned from key-package exchange, persisted as [`PeerCapabilities`],
    /// restored on `initialize_mls`, bounded like `key_package_sent_to`.
    ///
    /// This is the **send** half, and it is a hint: forgetting a peer costs
    /// them the older payload, which they still verify. The **accept** half is
    /// [`Self::control_freshness_peers`], which is a ratchet and is not
    /// allowed to forget.
    ///
    /// [`CTRL_SIGN_V2`]: offline_protocol_sealed::CTRL_SIGN_V2
    peer_ctrl_freshness: std::collections::HashSet<String>,

    /// Peers that read an `undecryptable` ACK as "not delivered"
    /// ([`ACK_UNDECRYPTABLE_V1`] in `ack_versions`). Only these are answered
    /// when a frame of theirs can never decrypt: an older SDK reads any ACK as
    /// a delivery. Same lifecycle as `peer_ctrl_freshness` above.
    ///
    /// [`ACK_UNDECRYPTABLE_V1`]: offline_protocol_sealed::ACK_UNDECRYPTABLE_V1
    pub(crate) peer_undecryptable_ack: std::collections::HashSet<String>,

    /// Peers that have presented a control-frame signature over the
    /// freshness-bound payload, so their frames under the older one are
    /// refused.
    ///
    /// A **ratchet**, and the difference from every capability set above is
    /// the whole point: those record what a peer said and may be forgotten
    /// harmlessly, this records what a peer *proved* and forgetting it
    /// re-opens the replay it exists to close. It is durable (in the peer's
    /// [`EncryptionCapableEntry`], not in [`PeerCapabilities`], which is
    /// overwritten wholesale by every key package) and has no removal path
    /// short of forgetting the peer entirely.
    ///
    /// A strict subset of [`Self::encryption_capable_peers`] by construction,
    /// so `MAX_ENCRYPTION_CAPABLE_PEERS` bounds it too. See
    /// `record_control_freshness_proved` for why the ordering that makes that
    /// true is load-bearing.
    pub(crate) control_freshness_peers: std::collections::HashSet<String>,

    /// Per-peer high-water mark: the signed timestamp of the most recent
    /// session reset acted on from that peer.
    ///
    /// A reset is honoured only when its signed timestamp is strictly newer
    /// than this, which is what stops one captured reset frame from being
    /// spent again and again inside the freshness window. Chosen over a set of
    /// spent frame ids because it is one integer per peer instead of a list
    /// that grows with traffic, and because it is strictly stronger: it
    /// refuses *every* older reset, including ones this node has no memory of.
    ///
    /// Durable in the peer's [`EncryptionCapableEntry`], and bounded by
    /// construction like [`Self::control_freshness_peers`]: entries are only
    /// written for peers the capped encryption-capable set accepted.
    pub(crate) control_reset_watermark: std::collections::HashMap<String, i64>,

    /// Peers whose key package advertised the sealed rich payload
    /// ([`RICH_PAYLOAD_V1`] in `rich_versions`), so the send path may seal
    /// rich extras (reply context, rich media metadata, forward attribution)
    /// inside the `__RICH_V1__` body. Same lifecycle as
    /// `peer_compact_envelope` above: learned from key-package exchange,
    /// persisted, restored on `initialize_mls`, bounded like
    /// `key_package_sent_to`. Forgetting a peer only costs silently dropped
    /// rich extras — never a cleartext fallback.
    peer_rich_payload: std::collections::HashSet<String>,

    /// Peers that route a sealed service frame ([`SVC_SEALED_V1`] in
    /// `svc_versions`), so a service request or response addressed to them
    /// may travel inside `__MLS_ENC__` rather than as signed plaintext.
    ///
    /// The **advertised** half: learned only from a **signed** key package,
    /// like the sets above (persisted as [`PeerCapabilities`], restored on
    /// `initialize_mls`), and rewritten by every key package, so a fresh one
    /// that stops advertising removes the peer. A strict subset of
    /// [`Self::encryption_capable_peers`], refused at its cap rather than
    /// cleared when full: clearing would send every established peer's
    /// service bodies in plaintext on a flood. The proved half is
    /// [`Self::svc_sealed_proved_peers`], and either one is enough to seal.
    ///
    /// [`SVC_SEALED_V1`]: offline_protocol_sealed::SVC_SEALED_V1
    pub(crate) peer_svc_sealed: std::collections::HashSet<String>,

    /// Peers that have sent us a service frame sealed inside the 1:1 session,
    /// which proves they route the sealed form.
    ///
    /// A **ratchet**, unlike [`Self::peer_svc_sealed`]: no key package can
    /// remove a peer from it. The failure it prevents is a replayed key
    /// package from before the peer upgraded, which would otherwise drop the
    /// peer from the advertised set and send its service bodies in signed
    /// plaintext again. A key package signed under `offline-ctrl-v1` is
    /// admitted at any age, so nothing else bounds that replay. Durable in
    /// the peer's [`EncryptionCapableEntry`] beside `ctrl_freshness_proved`,
    /// and a strict subset of [`Self::encryption_capable_peers`] by
    /// construction, so `MAX_ENCRYPTION_CAPABLE_PEERS` bounds it too.
    ///
    /// Knowledge, not policy: recorded whatever our own switch says, and the
    /// switch gates only whether we seal.
    pub(crate) svc_sealed_proved_peers: std::collections::HashSet<String>,

    /// Peers whose key package advertised replicated-document sync
    /// ([`DATA_SYNC_V1`] in `data_versions`), so the send path may push
    /// `__DATA_V1__` frames to them. Same lifecycle as the two capability
    /// sets above: learned from key-package exchange, persisted, restored on
    /// `initialize_mls`, bounded like `key_package_sent_to`.
    ///
    /// Forgetting a peer costs replication with them until the next live
    /// exchange, which is worse than it sounds: documents stay editable on
    /// both sides and diverge quietly, so the symptom is not an error but
    /// two people looking at different text. It is why this set is restored
    /// rather than relearned.
    ///
    /// [`DATA_SYNC_V1`]: crate::protocol::types::DATA_SYNC_V1
    peer_data_sync: std::collections::HashSet<String>,

    /// Peers whose sealed-rich-payload support we learned *indirectly*: a
    /// group inviter attested it on the Add commit (to existing members) or
    /// the Welcome (to the joining member), because the members of a group
    /// never directly exchange key packages with everyone else. Kept
    /// separate from `peer_rich_payload` (direct self-advertisement) so
    /// direct knowledge can stay authoritative: any directly received key
    /// package from a peer evicts its attested entry. Consulted only by
    /// `group_rich_seal_active` — never DM sealing (which always has direct
    /// knowledge: session establishment exchanges key packages) and never
    /// envelope selection (a stale attestation there could corrupt
    /// decoding; here the worst case is one member rendering a literal
    /// `__RICH_V1__` body). Persisted inside `PeerCapabilities`, restored
    /// on `initialize_mls`, bounded like `key_package_sent_to`.
    peer_rich_attested: std::collections::HashSet<String>,

    /// Peers whose key package advertised *group* document replication
    /// ([`DATA_GROUP_V1`] in `data_versions`), so a group they share with
    /// this device may carry `__DATA_V1__` frames.
    ///
    /// Separate from `peer_data_sync` because the two entries mean
    /// different things and a peer can advertise the first without the
    /// second: an install predating group spaces replicates 1:1 perfectly
    /// well and would render a group sync frame as chat text. Same
    /// lifecycle as every capability set: learned from key-package
    /// exchange, persisted, restored on `initialize_mls`, bounded like
    /// `key_package_sent_to`.
    ///
    /// [`DATA_GROUP_V1`]: crate::protocol::types::DATA_GROUP_V1
    peer_data_group: std::collections::HashSet<String>,

    /// Peers whose *group* replication support we learned indirectly, from
    /// an inviter's attestation on the Add commit or Welcome.
    ///
    /// The group analogue of `peer_rich_attested`, and load-bearing for the
    /// same structural reason: members of a group never exchange key
    /// packages with each other, so without attestation a group would
    /// replicate only among whichever members happened to have met
    /// directly. Direct knowledge stays authoritative — a key package from
    /// the peer evicts its attested entry. Consulted only by
    /// `group_data_sync_active`, never by 1:1 sync.
    peer_data_group_attested: std::collections::HashSet<String>,

    /// Peers whose key package advertised attachment and snapshot carriage
    /// over the media path ([`DATA_MEDIA_V1`] in `data_versions`), so this
    /// device may ask them for a blob and may hand them a document too large
    /// to travel inside a sync frame.
    ///
    /// A third set rather than a flag on the first, for the same reason
    /// `peer_data_group` is separate: a peer can advertise 1:1 replication
    /// without this, and toward such a peer a data-purposed media transfer
    /// arrives as an ordinary received file. Same lifecycle as every
    /// capability set: learned from key-package exchange, persisted inside
    /// `PeerCapabilities`, restored on `initialize_mls`, bounded like
    /// `key_package_sent_to`.
    ///
    /// [`DATA_MEDIA_V1`]: crate::protocol::types::DATA_MEDIA_V1
    peer_data_media: std::collections::HashSet<String>,

    /// Peers whose key package advertised document removals
    /// ([`DATA_TOMBSTONE_V1`] in `data_versions`), so a version offer toward
    /// them carries the `gone` field.
    ///
    /// A fourth set for the same reason as the third, with one difference
    /// worth keeping in mind: this one gates traffic, not safety. A peer
    /// missing from it ignores the field rather than mishandling it, and
    /// this device still refuses that peer's offers of removed content
    /// against its own floor.
    ///
    /// [`DATA_TOMBSTONE_V1`]: crate::protocol::types::DATA_TOMBSTONE_V1
    peer_data_tombstones: std::collections::HashSet<String>,

    /// Peers whose key package advertised interest-scoped replication
    /// ([`DATA_INTEREST_V1`] in `data_versions`), so a version offer toward
    /// them carries the `want` field.
    ///
    /// A fifth set, and a traffic gate like the fourth: a peer missing from
    /// it answers with everything it holds, and this device refuses what it
    /// did not ask for on arrival.
    ///
    /// [`DATA_INTEREST_V1`]: crate::protocol::types::DATA_INTEREST_V1
    peer_data_interest: std::collections::HashSet<String>,

    /// Peers whose key package advertised attachment bytes inside a group
    /// ([`DATA_GROUP_BLOB_V1`] in `data_versions`).
    ///
    /// Direct knowledge only. Members of a group do not exchange key
    /// packages with each other, so absence from this set is "we have not
    /// heard" far more often than "they cannot", which is why the attested
    /// sibling below exists and why the refusal reads both.
    ///
    /// [`DATA_GROUP_BLOB_V1`]: crate::protocol::types::DATA_GROUP_BLOB_V1
    peer_data_group_blob: std::collections::HashSet<String>,

    /// Peers an inviter attested carry attachment bytes inside a group.
    ///
    /// The sibling of `peer_data_group_attested`, and it exists for the same
    /// reason: this is the only way to learn anything about a member whose
    /// key package this device has never seen. Evicted by any directly
    /// received key package, in either direction.
    peer_data_group_blob_attested: std::collections::HashSet<String>,

    /// Peers whose key package advertised the custody receipt
    /// ([`DATA_CUSTODY_V1`] in `data_versions`), so this device may answer
    /// their deposits with one. Gates the receipt and nothing else: a deposit
    /// is judged by the quotas. Persisted inside `PeerCapabilities`, restored
    /// on `initialize_mls`, bounded like `key_package_sent_to`.
    ///
    /// [`DATA_CUSTODY_V1`]: crate::protocol::types::DATA_CUSTODY_V1
    peer_data_custody: std::collections::HashSet<String>,

    /// Peers already flagged with a `PlaintextSend` security warning, so the
    /// explicit-opt-out plaintext path warns once per peer instead of once
    /// per message.
    pub(crate) plaintext_send_warned: std::collections::HashSet<String>,

    /// Peers already flagged with a `PlaintextReceiveRejected` security
    /// warning, so a chatty legacy or malicious peer warns once per peer
    /// instead of once per rejected message. Keys are wire-claimed
    /// (attacker-controllable) sender ids, so the set is bounded: it resets
    /// at `MAX_PLAINTEXT_RECEIVE_WARNED_PEERS` instead of growing without
    /// limit.
    pub(crate) plaintext_receive_warned: std::collections::HashSet<String>,

    /// Peers whose durable `encryption_capable_peers` record is known to be on
    /// disk — either restored from it at boot or written successfully this run.
    ///
    /// Purely a write-elision cache for
    /// [`OfflineProtocol::record_encryption_capable`]: a verified control frame
    /// arrives on chatty prefixes (`__TYPING__`, `__PRESENCE__`), and the record
    /// it would rewrite carries no field any decision reads. Tracking *landed
    /// writes* rather than set membership is what keeps a best-effort storage
    /// failure retryable on the peer's next frame instead of stranded for the
    /// process.
    ///
    /// A subset of [`Self::encryption_capable_peers`] by construction, so
    /// `MAX_ENCRYPTION_CAPABLE_PEERS` bounds it too.
    pub(crate) encryption_capable_persisted: HashSet<String>,

    /// Which control-gate rejection codes have already been reported per peer,
    /// as a bitmask (see `OfflineProtocol::control_gate_warning_bit`), so a
    /// peer whose control traffic keeps being refused warns once per code
    /// instead of once per frame.
    ///
    /// Keys are wire-claimed (attacker-controllable) sender ids, so the map is
    /// bounded: it resets at `MAX_CONTROL_GATE_WARNED_PEERS` instead of growing
    /// without limit.
    pub(crate) control_gate_warned: HashMap<String, u8>,

    /// Bounded pending decryption queue for encrypted messages received before
    /// the MLS session is ready.
    pub(crate) pending_queue: PendingDecryptionQueue,

    /// Secure storage for MLS/key material and SDK-owned secrets.
    secure_storage: Option<Arc<dyn MlsStorage>>,

    /// Install-scoped storage for restartable message-plane and protocol state.
    protocol_state_storage: Option<Arc<dyn ProtocolStateStorage>>,

    /// Seals sensitive protocol-state records before they reach the
    /// install-scoped store, with a per-install key held in `secure_storage`.
    ///
    /// `None` until that key is available (and if it never becomes available,
    /// for the whole session). Sealed categories then fail *closed*: they are
    /// not persisted at all rather than written in the clear, since losing
    /// crash recovery is recoverable and losing at-rest confidentiality is not.
    /// See `state_crypto` and `restore_or_init_state_record_key`.
    ///
    /// Always belongs to the currently attached `secure_storage` — it is
    /// re-derived, never carried across a storage swap.
    state_record_cipher: Option<state_crypto::StateRecordCipher>,

    /// Open replicated documents and their per-space indexes.
    #[cfg(feature = "data")]
    pub(crate) data: data::DataLayer,

    /// When each peer was last offered our document versions.
    ///
    /// Neighbour rediscovery is not a rare event: a peer walking in and out
    /// of Bluetooth range fires it repeatedly, and an offer per discovery
    /// would put a frame on the mesh every time. The offer is not urgent
    /// (anything committed locally is pushed as it happens) and it is not
    /// load-bearing (the next trigger repeats it), so it is exactly the kind
    /// of traffic that should be rate limited rather than merely small.
    #[cfg(feature = "data")]
    pub(crate) last_data_sync_offer: HashMap<String, Instant>,

    /// Settle offers waiting for local commits to a space to go quiet, keyed
    /// by space: when the offer is due, and the documents committed since
    /// the last one. Pushed out by every local commit and removed when it
    /// fires. See [`DATA_SYNC_SETTLE_DELAY`].
    ///
    /// [`DATA_SYNC_SETTLE_DELAY`]: crate::protocol::data_sync::DATA_SYNC_SETTLE_DELAY
    #[cfg(feature = "data")]
    pub(crate) data_sync_settle: HashMap<String, (Instant, std::collections::BTreeSet<String>)>,

    /// Attachment fetches asked for and not yet answered, keyed by space and
    /// blob hash, valued by the last sign that an answer is coming: the
    /// moment the question went out, or the arrival of a chunk of its
    /// answer.
    ///
    /// The refresh is what keeps this record's two jobs consistent. It bounds
    /// how long a question is remembered, and it is also what admits the
    /// bytes when they finally land, so a record that expired while its own
    /// answer was arriving would discard a blob that fully crossed.
    ///
    /// Load-bearing for more than bookkeeping: arriving blob bytes are
    /// admitted only against an entry here, checked when the transfer's
    /// chunk 0 identifies it and again before the bytes reach the
    /// application. Without it a peer could push blobs this device never
    /// asked for and the application would be handed files nobody wanted.
    ///
    /// The bound it gives is on delivery and on the bulk of the buffering,
    /// not on every byte: chunks need not arrive in order, so any that land
    /// before chunk 0 are buffered before there is anything to judge, and
    /// are released when it arrives. What remains is bounded by the media
    /// layer's own resource caps, the same ones every unsolicited file
    /// transfer has always been subject to.
    ///
    /// Gated on the data feature, which the record's own type now is: what
    /// a fetch is waiting on, and what it has assembled so far, are facts
    /// about the data layer and have no meaning without it.
    #[cfg(feature = "data")]
    pub(crate) pending_attachment_fetches:
        HashMap<String, crate::protocol::data_sync::PendingFetch>,

    /// Blob requests recently acted on, keyed by asking peer and blob hash.
    ///
    /// Its own map rather than a corner of `last_data_sync_offer`, because
    /// this is the only rate-limit key a peer chooses freely: every other one
    /// is a peer or a peer-and-group, bounded by a roster. Bounded here by
    /// [`MAX_BLOB_REQUESTS_PER_WINDOW`] per peer and
    /// [`MAX_BLOB_REQUEST_WINDOWS`] overall, and swept on every read.
    ///
    /// [`MAX_BLOB_REQUESTS_PER_WINDOW`]: crate::protocol::data_sync::MAX_BLOB_REQUESTS_PER_WINDOW
    /// [`MAX_BLOB_REQUEST_WINDOWS`]: crate::protocol::data_sync::MAX_BLOB_REQUEST_WINDOWS
    pub(crate) blob_request_windows: HashMap<String, Instant>,

    /// Whether the MLS group list has been read into `group_mesh.members`
    /// once this session, so the shared-group rediscovery sweep can see the
    /// groups this device is in.
    ///
    /// That cache is filled on demand by group traffic and is empty after a
    /// restart, so a sweep that only read it would find nothing on exactly
    /// the launch where two members have drifted apart and neither is
    /// editing. Enumerating is one key listing plus, per group not already
    /// cached, the group-state read the next group send would have paid
    /// anyway.
    ///
    /// One shot rather than per sweep because the cache stays complete once
    /// filled: creating a group inserts, a Welcome inserts, and leaving
    /// removes. Deliberately not a cold-start sweep, which was rejected for
    /// putting an offer per member per group on the mesh at every launch;
    /// this is per peer, only when that peer is actually in front of us, and
    /// still behind the offer window.
    #[cfg(feature = "data")]
    pub(crate) group_spaces_enumerated: bool,

    /// Lamport logical clock for causal message ordering.
    /// Ticked on send, merged on receive.
    lamport_clock: LamportClock,

    /// Retry schedule for peers whose confirmation persistence failed.
    confirmation_retry_due_at: HashMap<String, DateTime<Utc>>,

    /// Probe schedule for pending sessions to guarantee post-restart convergence.
    confirmation_probe_due_at: HashMap<String, DateTime<Utc>>,

    /// Consecutive relay "recipient unreachable" verdicts per peer for the
    /// session-confirmation probe. Drives an escalating backoff of
    /// `confirmation_probe_due_at` (15s -> 600s cap, the same ladder the welcome
    /// lifecycle uses) so a peer that established a session then vanished is not
    /// probed every `CONFIRMATION_PROBE_INTERVAL_SECS` forever; a fleet of such
    /// peers would otherwise exceed the relay's per-connection rate limit and
    /// flap the connection. Reset on a reachability edge (`on_peer_presence`).
    confirmation_probe_unreachable_parks: HashMap<String, u32>,

    /// The confirmation probe each pending peer last sent. At most one probe
    /// per peer lives in the outbox: a new one supersedes the last. Stacking a
    /// fresh probe (new id, full retry ladder) every
    /// `CONFIRMATION_PROBE_INTERVAL_SECS` on top of the unanswered ones filled
    /// the outbox whenever no relay verdict (`unreachable` or
    /// `relay_pushed`) arrived to back the schedule off, as on a mesh-only
    /// carrier. Capacity eviction then failed real messages.
    confirmation_probe_outstanding: HashMap<String, MessageId>,

    /// The probe each pending peer's current one superseded. A relay verdict
    /// finds its peer through the outbox entry, which the supersede removed,
    /// so a verdict slower than `CONFIRMATION_PROBE_INTERVAL_SECS` would
    /// otherwise be dropped and never back the schedule off, leaving the
    /// peer probed every 5s against a relay that has said it is not there.
    /// One id per peer, so this covers a verdict up to two intervals late;
    /// one later than that is dropped, and the next probe's verdict backs
    /// the schedule off instead.
    confirmation_probe_superseded: HashMap<String, MessageId>,

    /// Token bucket that caps how fast the RETRY/PROBE path
    /// (`process_retry_queue`) re-sends, so a large backlog of resends to
    /// unreachable peers cannot burst past the relay's per-connection rate limit
    /// and trigger the disconnect->reconnect->reflood loop ("flapping"). Only
    /// consulted when `RetryConfig::edge_driven_unreachable_dm` is opted in. It
    /// bounds ONLY resends — first-attempt sends, ACKs, control frames and
    /// handshakes take other paths and are never throttled, so live traffic
    /// keeps full throughput/latency. `retry_drain_last` is None until first use.
    retry_drain_tokens: f64,
    retry_drain_last: Option<Instant>,

    /// Rate-limit schedule for 1:1 session re-keys triggered by an epoch-desync
    /// decrypt failure. Bounds the re-key to at most one per peer per
    /// `REKEY_INTERVAL_SECS` so a peer replaying stale-epoch ciphertext (or an
    /// injected wrong-epoch frame) cannot drive a re-key storm. The floor is
    /// **never reset early** — not even by a successful decrypt on the healed
    /// session — because a genuine re-fork and a replay are indistinguishable
    /// here; it lapses only by the interval elapsing (see
    /// `schedule_session_rekey`). In-memory only: a re-key is a live-connectivity
    /// action, and a fresh desync after restart simply re-arms it.
    rekey_due_at: HashMap<String, DateTime<Utc>>,

    /// Outbound welcome lifecycle records keyed by peer id.
    welcome_lifecycles: HashMap<String, WelcomeLifecycleRecord>,

    /// Outbound connection requests still awaiting a transport outcome,
    /// keyed by message id → recipient. Lets `on_transport_send_failed`
    /// turn the relay's authoritative "recipient offline" DeliveryError
    /// into a typed `ConnectionRequestUndeliverable` event instead of
    /// silently discarding it (the welcome-only path never matches a
    /// connection request's message id). Entries are dropped on emission
    /// and on proof of delivery (the request's delivery ack, or an inbound
    /// accept/reject from the recipient), and TTL-pruned on insert and at
    /// read; bounded, so an app spamming requests cannot grow it
    /// unboundedly.
    pending_connection_requests: HashMap<String, PendingConnectionRequest>,

    /// Backoff state for presence-driven welcome rescue, keyed by peer id.
    /// Bounds how often an online-but-never-confirming peer triggers a
    /// welcome re-arm/re-send (see `on_peer_presence`). Entries are dropped
    /// lazily once the peer has no unconfirmed welcome left.
    welcome_presence_rescue: HashMap<String, PresenceRescueThrottle>,

    /// Peers for which we are the both-create "owner" (kept our own session
    /// group on the lexicographic tiebreak) and are still awaiting a
    /// *group-aware* proof that the peer adopted our group. While a peer is in
    /// this set, only `decrypt_success` may confirm the session — a plaintext
    /// confirmation probe/ack is NOT group-aware (it only proves the peer holds
    /// *some* session, possibly its own pre-adoption group) and must not be
    /// allowed to stop our Welcome retransmission, or the peer could be left
    /// stranded on a divergent group. Persisted (see
    /// `restore_both_create_awaiting_decrypt`) so an owner restart mid-convergence
    /// keeps the gate, rather than letting a stale plaintext probe/ack confirm
    /// prematurely. Mutate only via `mark_both_create_awaiting_decrypt` /
    /// `clear_both_create_awaiting_decrypt` to keep memory and storage in sync.
    both_create_awaiting_decrypt: std::collections::HashSet<String>,

    /// Sink for MLS lifecycle telemetry.
    mls_event_emitter: Arc<dyn MlsEventEmitter>,

    /// Rate limiting policy for MLS failure event floods.
    mls_event_rate_limiter: MlsEventRateLimiter,

    /// Pre-install scrubber used by MLS emit sites before a sink is
    /// installed (`enable_telemetry`, or `install_sink` beneath it). Once one
    /// is installed, emit sites read `self.telemetry.scrubber` instead via
    /// `current_scrubber()`.
    /// Both scrubbers share `telemetry_fallback_secret` so opaque identifiers
    /// observed by the legacy `MlsEventEmitter` stay consistent across the
    /// install boundary.
    telemetry_scrubber: Scrubber,

    /// Per-instance fallback secret for identifier scrubbing. Random at
    /// construction. Reused by both the pre-install `telemetry_scrubber`
    /// and by any scrubber built inside `install_sink` (unless the
    /// installed `TelemetryConfig` carries its own `scrub_secret`). Keeping
    /// the fallback stable across installs means the legacy
    /// `MlsEventEmitter` path observes consistent opaque IDs before and
    /// after a sink is installed on the same protocol instance.
    telemetry_fallback_secret: [u8; 16],

    /// Whether the persistent per-install scrub secret has been loaded from
    /// (or initialized into) secure storage yet. Starts `false`; set `true`
    /// the first time `restore_or_init_scrub_secret` succeeds so the load is
    /// idempotent across the two storage-entry paths
    /// (`initialize_mls` also enables persistence, which would otherwise
    /// re-load and rebuild the scrubber a second time).
    telemetry_secret_persisted: bool,

    /// Whether the persistent per-install Nostr signing secret has been
    /// loaded from (or initialized into) secure storage and installed into
    /// the Nostr transport yet. Same idempotency role across the two
    /// storage-entry paths as `telemetry_secret_persisted`; see
    /// `restore_or_init_nostr_signing_secret`.
    nostr_secret_persisted: bool,

    /// A Nostr signing secret that was installed into the transport but
    /// could not be persisted (storage `store` failed). Kept so the next
    /// `restore_or_init_nostr_signing_secret` attempt retries persisting
    /// this same secret instead of rotating the install's relay-visible
    /// identity again mid-session.
    nostr_unpersisted_secret: Option<Zeroizing<[u8; 32]>>,

    /// Installed telemetry context (sink + config + scrubber). `None` until
    /// `enable_telemetry` installs one; thereafter shared with
    /// `SharedState` via `Arc` clone so both emit paths dispatch through the
    /// same configuration.
    ///
    /// The duplication with [`SharedState::telemetry`] is deliberate: MLS
    /// lifecycle emission is driven from `&self` on `OfflineProtocol` and
    /// does not hold the shared-state lock, while protocol-event emission
    /// happens inside `SharedState::emit_event` under the lock. Each path
    /// reads the context from whichever side it already has in hand, and
    /// `install_sink` is the single writer that keeps both copies in sync
    /// (guaranteed atomic from the caller's perspective because `&mut self`
    /// excludes concurrent calls), and `detach_telemetry_pipe` the single
    /// clearer.
    pub(crate) telemetry: Option<Arc<TelemetryContext>>,

    /// The running telemetry pipe, when `enable_telemetry` built one. Its
    /// sink is what `telemetry` dispatches into; the pipe itself is kept so
    /// the tick can wake it and `initialize_mls` can hand it storage.
    #[cfg(feature = "telemetry-pipe")]
    telemetry_pipe: Option<Arc<TelemetryPipe>>,

    /// File transfer manager for chunking outbound and reassembling inbound media.
    file_transfer_manager: FileTransferManager,

    /// Stashed (content_type, media_metadata) from chunk-0 of incoming file transfers,
    /// used to populate the FileReceived event once all chunks arrive.
    pending_media_metadata: HashMap<String, PendingMediaMetadataEntry>,
    /// Tracks outbound media transfer delivery progress keyed by file_id.
    outbound_media_transfers: HashMap<String, OutboundMediaTransfer>,
    /// Maps each outbound media chunk message ID to (file_id, chunk_index).
    outbound_media_chunks: HashMap<MessageId, (String, u32)>,
    /// Sliding-window state for outbound file transfers, keyed by file_id.
    /// Chunks are only sent when the window has capacity (previous chunks ACKed).
    outbound_media_windows: HashMap<String, OutboundTransferState>,
    /// Descriptors of transfers that were in flight when the previous process
    /// died, restored from storage. `start()` emits one `MediaResendRequired`
    /// event each but deliberately does NOT drain them: entries stay parked
    /// (and persisted) until a same-`file_id` resend consumes them — after
    /// `send_media_with` checksum-validates the re-supplied bytes against
    /// the descriptor — or the restore TTL prunes them, so an app that
    /// misses the signal gets it again next restart.
    restored_media_descriptors: HashMap<String, MediaTransferDescriptor>,

    /// Mesh service registry and handler (extracted crate).
    mesh_services: MeshServices,

    /// Bundled state for mesh group messaging (member cache, dedup, pending commits).
    pub(crate) group_mesh: crate::group_mesh::GroupMeshState,

    /// Set of blocked user IDs. Messages from blocked users are silently
    /// dropped (no ACK, no event). Persisted via `ProtocolStateStorage`.
    blocked_users: HashSet<String>,

    /// Timestamp of the last `kick_pending_session_reconciliation` execution.
    /// Used to throttle reconciliation to avoid expensive storage I/O
    /// (list_sessions → Keychain/Keystore) on every process tick / receive poll.
    last_reconciliation_at: Option<Instant>,

    /// Instant of the last periodic `MetricsFrame` emission, used to rate
    /// the telemetry aggregator to `TelemetryConfig::metrics_cadence`.
    last_metrics_emit_at: Option<Instant>,

    /// Last observed per-transport status map, used by the telemetry
    /// aggregator to diff against the current map and emit one
    /// `TransportStateEvent` per transition.
    transport_status_snapshot: HashMap<TransportType, TransportStatus>,

    /// Last observed (battery, charging, relay_role) snapshot, used by the
    /// telemetry aggregator to fire `DeviceCapabilitySnapshot` on change.
    device_capability_snapshot: Option<DeviceSnap>,

    /// The Lamport clock value at the time of the last storage write.
    /// Used to debounce `persist_lamport_clock()` — only write when the
    /// in-memory value has advanced past this by `LAMPORT_PERSIST_INTERVAL`
    /// ticks. On crash, at most `LAMPORT_PERSIST_INTERVAL` ticks are lost,
    /// which is harmless for a logical clock (the gap is absorbed by the
    /// next merge with any peer).
    last_persisted_lamport: u64,

    /// Receive-side changes to the deduplicator's seen set since it was last
    /// written. Drives the batched write in `persist_dedup_seen_if_due`;
    /// see `DedupSeenRecord` for why the set is persisted at all.
    dedup_dirty: u32,

    /// When the seen set was last written (or restored), for the time half of
    /// the batching rule. Starts at construction so the first write waits for
    /// either the interval or the dirty threshold like every later one.
    dedup_last_persist: Instant,

    /// The Nostr receive watermark last written to storage, or `None` if this
    /// session has neither written nor restored one. Debounces
    /// `persist_nostr_watermark()` the same way `last_persisted_lamport`
    /// debounces the clock.
    ///
    /// Deliberately *not* captured for `initialize_mls` rollback, matching
    /// where the restore itself runs: the mark lives in the transport, which
    /// the rollback cannot reach, and re-installing a durable value that only
    /// ever moves forward is not something a failed MLS restore needs undone.
    last_persisted_nostr_watermark: Option<i64>,

    /// Which MLS key package stands in each published Nostr slot.
    ///
    /// Persisted, because the alternative to remembering it is republishing
    /// under fresh slot ids on every launch — leaving the old records standing
    /// with packages we would no longer track for consumption.
    nostr_publication_slots: Vec<nostr_publication::NostrPublicationSlot>,

    /// Slot ids published during *this* process.
    ///
    /// Deliberately not persisted: an addressable record lives on the relays,
    /// so a launch cannot know which of them still hold it. Publishing each
    /// slot once per process is the cheap way to be sure — it replaces rather
    /// than accumulates, so the cost of being wrong is one redundant event.
    nostr_published_slots: HashSet<String>,

    /// When the publication slots were last re-scanned, throttling the MLS
    /// storage reads that scan costs. See `nostr_slot_refresh_due`.
    last_nostr_slot_refresh: Option<Instant>,

    /// Per-slot republish backoff after publications that never reached a
    /// relay. Memory-only, and pruned to the live slot ids each refresh.
    nostr_publication_backoff: HashMap<String, nostr_publication::PublicationBackoff>,

    /// When the slot-exhaustion security warning was last emitted, suppressing
    /// repeats for a cause that by nature persists.
    last_nostr_slot_warning: Option<Instant>,

    /// The username this install has published a discovery claim for, as last
    /// persisted. `None` means no claim stands.
    ///
    /// Persisted, because it is the only thing that knows *which* name to
    /// retract when the profile changes or the feature is turned off. See
    /// `storage_keys::NOSTR_DISCOVERY_CLAIM`.
    nostr_discovery_claim: Option<Username>,

    /// Names whose tombstone has not yet reached a relay, oldest first.
    ///
    /// Persisted beside the claim, and for a stronger reason than the claim
    /// is: the claim can be rebuilt from `config.profile`, while a retracted
    /// name has no other record anywhere once the profile has moved on.
    /// Dropping it before the tombstone lands strands a human-readable name in
    /// a public directory pointing at an address whose key packages are still
    /// being published, so nothing downstream expires it.
    nostr_discovery_retracting: Vec<Username>,

    /// Whether the standing claim has been published during *this* process.
    ///
    /// Not persisted, for the same reason `nostr_published_slots` is not: an
    /// addressable record lives on the relays and a launch cannot know which
    /// of them still hold it. Republishing once per process replaces rather
    /// than accumulates.
    nostr_discovery_published: bool,

    /// Republish backoff for the discovery claim, keyed by its tag.
    ///
    /// Memory-only and structurally identical to
    /// `nostr_publication_backoff`: a relay that rejects kind 30777 outright
    /// would otherwise be retried once per refresh forever.
    nostr_discovery_backoff: HashMap<String, nostr_publication::PublicationBackoff>,

    /// When the discovery claim was last re-examined, throttling the tick.
    last_nostr_discovery_refresh: Option<Instant>,

    /// In-flight username resolutions, keyed by query id.
    ///
    /// A resolution accumulates every verified claim for its username and
    /// emits **one** event carrying the whole set. See
    /// `nostr_discovery::PendingResolution` for why the set is the only thing
    /// that is ever emitted.
    nostr_resolutions: HashMap<String, nostr_discovery::PendingResolution>,

    /// Username lookups requested but not yet minted into a relay query, with
    /// the time each was asked for.
    ///
    /// The transport holds the queue; this holds the *deadline*. They are
    /// separate because minting is what may never happen: the platform pumps
    /// queries only while the relay socket is up, so a lookup requested offline
    /// would otherwise have no clock running behind it and no event to arrive.
    /// Bounded by the transport's own resolve-queue ceiling, since an entry is
    /// recorded only when the transport accepts the lookup.
    nostr_resolution_requests: HashMap<Username, Instant>,

    /// When the push key-package pool-exhaustion warning was last emitted.
    ///
    /// Suppressed for the same reason the slot warning is: the condition is a
    /// pool at its ceiling, which persists until packages are consumed or
    /// expire, and every push while it holds would otherwise emit.
    last_push_key_package_warning: Option<Instant>,

    /// When the last `GatewayAddressDeclarationRefused` and
    /// `GatewayAddressBindingMismatch` warnings were emitted, for the same
    /// suppression the push-key-package warning uses.
    last_gateway_refusal_warning: Option<Instant>,
    last_gateway_binding_mismatch_warning: Option<Instant>,
}

impl Drop for OfflineProtocol {
    fn drop(&mut self) {
        // Flush debounced Lamport clock so no ticks are lost when the
        // protocol is dropped without an explicit stop() call.
        self.flush_lamport_clock();
        self.flush_nostr_watermark();
        // And the deduplicator's seen set, whose write is batched the same
        // way: a copy of a message received in the last few seconds before
        // the drop would otherwise be re-processed by the next launch.
        self.flush_dedup_seen();
        // Same reason: edits batch before they reach a record, so without a
        // flush here the debounce window between an edit and its delta is a
        // window in which work is lost.
        #[cfg(feature = "data")]
        let _ = self.data_flush_all();
    }
}

#[cfg(test)]
struct TestProtocolStateStorage {
    storage: Arc<dyn MlsStorage>,
}

/// Maps the fixture's MLS-storage failures onto the protocol-state contract,
/// mirroring what the UniFFI adapter does for real providers.
#[cfg(test)]
pub(crate) fn map_test_storage_error(
    error: offline_protocol_mls::StorageError,
) -> crate::ProtocolStateError {
    use crate::ProtocolStateError as P;
    use offline_protocol_mls::StorageError as S;
    match error {
        S::KeyNotFound(detail) => P::NotFound(detail),
        S::CorruptedData(detail) => P::Corrupted(detail),
        S::StoreFailed(detail) => P::StoreFailed(detail),
        S::LoadFailed(detail) => P::LoadFailed(detail),
        S::DeleteFailed(detail) => P::DeleteFailed(detail),
        S::Unavailable(detail) => P::LoadFailed(detail),
        // Required by `#[non_exhaustive]` on StorageError. This fixture mirrors
        // the UniFFI adapter, whose own fallback also degrades to `LoadFailed`.
        ref other => P::LoadFailed(other.to_string()),
    }
}

#[cfg(test)]
impl ProtocolStateStorage for TestProtocolStateStorage {
    fn store(&self, key_type: &str, key_id: &str, data: &[u8]) -> crate::ProtocolStateResult<()> {
        self.storage
            .store(key_type, key_id, data)
            .map_err(map_test_storage_error)
    }

    fn load(&self, key_type: &str, key_id: &str) -> crate::ProtocolStateResult<Option<Vec<u8>>> {
        self.storage
            .load(key_type, key_id)
            .map_err(map_test_storage_error)
    }

    fn delete(&self, key_type: &str, key_id: &str) -> crate::ProtocolStateResult<()> {
        self.storage
            .delete(key_type, key_id)
            .map_err(map_test_storage_error)
    }

    fn list_keys(&self, key_type: &str) -> crate::ProtocolStateResult<Vec<String>> {
        self.storage
            .list_keys(key_type)
            .map_err(map_test_storage_error)
    }
}

impl OfflineProtocol {
    /// Creates a new protocol instance.
    ///
    /// # Arguments
    ///
    /// * `config` - Protocol configuration
    ///
    /// # Returns
    ///
    /// Returns `Ok(OfflineProtocol)` if successful, `Err` if configuration is invalid.
    pub fn new(config: ProtocolConfig) -> Result<Self> {
        // Validate configuration
        config.validate()?;

        // Create transport selector for DORS
        let transport_selector = TransportSelector::with_config(config.dors.clone());

        // Create transport manager
        let transport_manager = TransportManager::new(transport_selector);

        // Per-instance fallback secret for identifier scrubbing. Stable for
        // the life of this protocol instance so opaque IDs stay consistent
        // across any later `install_telemetry_sink` call that does not
        // supply its own `scrub_secret`.
        let telemetry_fallback_secret = *uuid::Uuid::new_v4().as_bytes();

        Ok(Self {
            transport_manager,
            ack_manager: AckManager::with_config(config.reliability.ack.clone()),
            retry_queue: RetryQueue::with_config(config.reliability.retry.clone()),
            deduplicator: Deduplicator::with_config(config.reliability.dedup.clone()),
            mesh_relay: MeshRelayGovernor::with_config(
                config.profile.clone(),
                config.mesh_relay.clone(),
            ),
            custody: custody::CustodyStore::new(config.custody.clone()),
            custody_receipts: HashMap::new(),
            custody_last_sweep: Instant::now(),
            local_id: config.profile.clone(),
            identity_established: false,
            shared_state: Arc::new(Mutex::new(SharedState::new())),
            outbox: HashMap::new(),
            pending_reseal: HashMap::new(),
            undecryptable_resent: HashSet::new(),
            media_outbox: HashMap::new(),
            dm_unreachable_parks: HashMap::new(),
            unrouted_acks: VecDeque::new(),
            reachability: reachability::ReachabilityFacts::default(),
            gateway_capabilities: HashSet::new(),
            mls_manager: None,
            pending_encrypted_messages: HashMap::new(),
            deferred_restore_settlements: Vec::new(),
            suppressed_restore_settlements: 0,
            next_pending_message_expiry: None,
            pending_key_packages: HashMap::new(),
            key_package_sent_to: HashMap::new(),
            known_peers: HashMap::new(),
            confirmed_sessions: std::collections::HashSet::new(),
            encryption_capable_peers: std::collections::HashSet::new(),
            peer_compact_envelope: std::collections::HashSet::new(),
            peer_ctrl_freshness: std::collections::HashSet::new(),
            peer_undecryptable_ack: std::collections::HashSet::new(),
            control_freshness_peers: std::collections::HashSet::new(),
            control_reset_watermark: std::collections::HashMap::new(),
            peer_rich_payload: std::collections::HashSet::new(),
            peer_svc_sealed: std::collections::HashSet::new(),
            svc_sealed_proved_peers: std::collections::HashSet::new(),
            peer_data_sync: std::collections::HashSet::new(),
            peer_data_group: std::collections::HashSet::new(),
            peer_data_group_attested: std::collections::HashSet::new(),
            peer_data_media: std::collections::HashSet::new(),
            peer_data_tombstones: std::collections::HashSet::new(),
            peer_data_interest: std::collections::HashSet::new(),
            peer_data_group_blob: std::collections::HashSet::new(),
            peer_data_group_blob_attested: std::collections::HashSet::new(),
            peer_data_custody: std::collections::HashSet::new(),
            peer_rich_attested: std::collections::HashSet::new(),
            plaintext_send_warned: std::collections::HashSet::new(),
            plaintext_receive_warned: std::collections::HashSet::new(),
            encryption_capable_persisted: HashSet::new(),
            control_gate_warned: HashMap::new(),
            pending_queue: PendingDecryptionQueue::default(),
            secure_storage: None,
            protocol_state_storage: None,
            state_record_cipher: None,
            #[cfg(feature = "data")]
            data: data::DataLayer::default(),
            #[cfg(feature = "data")]
            last_data_sync_offer: HashMap::new(),
            #[cfg(feature = "data")]
            data_sync_settle: HashMap::new(),
            #[cfg(feature = "data")]
            pending_attachment_fetches: HashMap::new(),
            blob_request_windows: HashMap::new(),
            #[cfg(feature = "data")]
            group_spaces_enumerated: false,
            lamport_clock: LamportClock::new(),
            confirmation_retry_due_at: HashMap::new(),
            confirmation_probe_due_at: HashMap::new(),
            confirmation_probe_unreachable_parks: HashMap::new(),
            confirmation_probe_outstanding: HashMap::new(),
            confirmation_probe_superseded: HashMap::new(),
            retry_drain_tokens: 0.0,
            retry_drain_last: None,
            rekey_due_at: HashMap::new(),
            welcome_lifecycles: HashMap::new(),
            pending_connection_requests: HashMap::new(),
            welcome_presence_rescue: HashMap::new(),
            both_create_awaiting_decrypt: std::collections::HashSet::new(),
            mls_event_emitter: Arc::new(NoopMlsEventEmitter),
            mls_event_rate_limiter: MlsEventRateLimiter::default(),
            // The pre-install scrubber uses `TelemetryConfig::default()` —
            // `scrub_ids=true`, no explicit `scrub_secret`. Identifier hashing
            // is therefore on from the moment the protocol is constructed,
            // matching today's always-scrub MLS observability semantics. When
            // `install_telemetry_sink` later supplies a config with
            // `scrub_ids(false)`, a new scrubber is derived from that config
            // but reuses `telemetry_fallback_secret` so opaque IDs remain
            // stable across the install boundary.
            telemetry_scrubber: Scrubber::from_config(
                &TelemetryConfig::default(),
                telemetry_fallback_secret,
            ),
            telemetry_fallback_secret,
            telemetry_secret_persisted: false,
            nostr_secret_persisted: false,
            nostr_unpersisted_secret: None,
            telemetry: None,
            #[cfg(feature = "telemetry-pipe")]
            telemetry_pipe: None,
            file_transfer_manager: FileTransferManager::new(),
            pending_media_metadata: HashMap::new(),
            outbound_media_transfers: HashMap::new(),
            outbound_media_chunks: HashMap::new(),
            outbound_media_windows: HashMap::new(),
            restored_media_descriptors: HashMap::new(),
            mesh_services: MeshServices::new(),
            group_mesh: crate::group_mesh::GroupMeshState::default(),
            blocked_users: HashSet::new(),
            last_reconciliation_at: None,
            last_metrics_emit_at: None,
            transport_status_snapshot: HashMap::new(),
            device_capability_snapshot: None,
            last_persisted_lamport: 0,
            dedup_dirty: 0,
            dedup_last_persist: Instant::now(),
            last_persisted_nostr_watermark: None,
            nostr_publication_slots: Vec::new(),
            nostr_published_slots: HashSet::new(),
            last_nostr_slot_refresh: None,
            nostr_publication_backoff: HashMap::new(),
            last_nostr_slot_warning: None,
            nostr_discovery_claim: None,
            nostr_discovery_retracting: Vec::new(),
            nostr_discovery_published: false,
            nostr_discovery_backoff: HashMap::new(),
            last_nostr_discovery_refresh: None,
            nostr_resolutions: HashMap::new(),
            nostr_resolution_requests: HashMap::new(),
            last_push_key_package_warning: None,
            last_gateway_refusal_warning: None,
            last_gateway_binding_mismatch_warning: None,
            config,
        })
    }

    /// Initializes MLS encryption and protocol persistence.
    ///
    /// This must be called before encryption can be used. The storage
    /// `secure_storage` must be a platform-native credential store.
    /// `protocol_state_storage` must be scoped to the app container and is
    /// removed when that container is deleted.
    ///
    /// Ownership model:
    /// - `OfflineProtocol` is the single authoritative owner of `MlsManager`
    /// - initialization is idempotent per protocol instance
    /// - subsequent calls return without replacing the existing manager
    /// - manager publication is transactional: restore must succeed before
    ///   `mls_manager` becomes visible to callers
    ///
    /// The two storage domains are intentionally different trait objects. This
    /// prevents message-plane state from inheriting the lifecycle of Keychain
    /// or another credential store.
    pub fn initialize_mls(
        &mut self,
        secure_storage: Arc<dyn MlsStorage>,
        protocol_state_storage: Arc<dyn ProtocolStateStorage>,
    ) -> Result<()> {
        self.initialize_mls_inner(secure_storage, protocol_state_storage, true)
    }

    /// Returns this device's address, once it has one.
    ///
    /// `None` before `initialize_mls`: the address is the hash of an identity
    /// key that lives in this profile's storage, and storage does not arrive
    /// until then. It never changes afterwards, and it is stable across
    /// restarts of the same profile because the key is.
    pub fn local_address(&self) -> Option<&str> {
        self.identity_established.then_some(self.local_id.as_str())
    }

    /// `adopt_legacy_state` exists only so test fixtures can point both handles
    /// at one backend. Adoption moves records *between* the two stores, which
    /// on a shared backend would mean deleting a record through the same store
    /// it was just read from.
    ///
    /// There is one identity path and every caller takes it — a fixture that
    /// wants a particular identity seeds the key into `secure_storage` first
    /// (see [`crate::test_identity`]) rather than naming an id here, because an
    /// id nothing holds a key for is exactly what `MlsManager::new` refuses.
    fn initialize_mls_inner(
        &mut self,
        secure_storage: Arc<dyn MlsStorage>,
        protocol_state_storage: Arc<dyn ProtocolStateStorage>,
        adopt_legacy_state: bool,
    ) -> Result<()> {
        if self.mls_manager.is_some() {
            return Ok(());
        }

        // The bootstrap inversion: load (or mint) the identity key first, and
        // take this device's wire identity from it. Before this point the
        // protocol has no address — `local_id` is still the profile — which is
        // exactly why the key has to be resolved here, ahead of the manager
        // that embeds the id in its credential and in every session slot.
        let (_, address) = MlsManager::load_or_create_identity(&secure_storage)?;
        let local_id = address.to_string();

        let mut manager = MlsManager::new(&local_id, secure_storage.clone())?;
        // Installed before the manager becomes shared: the enforcement flag is
        // read on every group decrypt, and this is the one point where the
        // manager is still uniquely owned.
        manager.set_enforce_admin_commits(self.config.group.enforce_admin_commits);
        let manager = Arc::new(RwLock::new(manager));

        // Keep initialization transactional so a restore failure cannot leave
        // partially-initialized MLS state visible and then permanently block retries.
        let previous_secure_storage = self.secure_storage.clone();
        let previous_protocol_state_storage = self.protocol_state_storage.clone();
        // The record cipher belongs to the secure store it was loaded from, so
        // it is part of the same transaction as the storage handles: taken here
        // so a failed init cannot leave the new store's key installed next to
        // the rolled-back handles, and restored below alongside them.
        let previous_state_record_cipher = self.state_record_cipher.take();
        let previous_pending_messages = self.pending_encrypted_messages.clone();
        let previous_pending_message_expiry = self.next_pending_message_expiry;
        // The inbound half of the same transaction: the restore re-admits
        // parked frames from the store being attached, so a rollback has to put
        // back what the queue held before, not an empty queue.
        let previous_pending_queue = self.pending_queue.clone();
        // Populated by restore steps that run before other fallible ones, so
        // they belong in the transaction like everything else: a failed init
        // must not leave a key-package cache, parked media descriptors, or an
        // owner gate sourced from a store the rollback has just detached.
        let previous_pending_key_packages = self.pending_key_packages.clone();
        let previous_restored_media_descriptors = self.restored_media_descriptors.clone();
        let previous_both_create_awaiting_decrypt = self.both_create_awaiting_decrypt.clone();
        let previous_confirmed_sessions = self.confirmed_sessions.clone();
        let previous_encryption_capable_peers = self.encryption_capable_peers.clone();
        let previous_welcome_lifecycles = self.welcome_lifecycles.clone();
        let previous_lamport_clock = self.lamport_clock.value();
        // The debounce watermark travels with the clock: `restore_lamport_clock`
        // sets both, so rolling back only the clock leaves a watermark ahead of
        // it and `persist_lamport_clock`'s `wrapping_sub` reads that as an
        // enormous delta — writing on every tick instead of every interval.
        let previous_last_persisted_lamport = self.last_persisted_lamport;
        let previous_blocked_users = self.blocked_users.clone();
        let previous_outbox = self.outbox.clone();
        let previous_peer_compact_envelope = self.peer_compact_envelope.clone();
        let previous_peer_ctrl_freshness = self.peer_ctrl_freshness.clone();
        let previous_peer_undecryptable_ack = self.peer_undecryptable_ack.clone();
        // Snapshotted with its siblings so a failed re-initialize restores it
        // too. It is a ratchet, so the direction that matters is that a
        // rollback must not *lose* it: a peer we hold to the newer payload has
        // to still be held to it if the re-initialize is abandoned.
        let previous_control_freshness_peers = self.control_freshness_peers.clone();
        let previous_control_reset_watermark = self.control_reset_watermark.clone();
        let previous_peer_rich_payload = self.peer_rich_payload.clone();
        let previous_peer_svc_sealed = self.peer_svc_sealed.clone();
        let previous_svc_sealed_proved_peers = self.svc_sealed_proved_peers.clone();
        let previous_peer_data_sync = self.peer_data_sync.clone();
        let previous_peer_rich_attested = self.peer_rich_attested.clone();
        let previous_peer_data_group = self.peer_data_group.clone();
        let previous_peer_data_group_attested = self.peer_data_group_attested.clone();
        let previous_peer_data_media = self.peer_data_media.clone();
        let previous_peer_data_tombstones = self.peer_data_tombstones.clone();
        let previous_peer_data_interest = self.peer_data_interest.clone();
        let previous_peer_data_group_blob = self.peer_data_group_blob.clone();
        let previous_peer_data_group_blob_attested = self.peer_data_group_blob_attested.clone();

        let previous_local_id = std::mem::replace(&mut self.local_id, local_id.clone());
        let previous_identity_established = self.identity_established;
        self.identity_established = true;
        // The governor took the profile at construction because no address
        // existed yet; it keys relay fan-out and jitter off the local id, so it
        // has to move to the real one here.
        self.mesh_relay.set_local_id(local_id);

        self.secure_storage = Some(secure_storage);
        self.protocol_state_storage = Some(protocol_state_storage);

        // Must precede every restore below: sealed protocol-state records
        // (pending messages, outbox, media descriptors) are unreadable without
        // this key, so restoring first would silently start from empty and then
        // overwrite durable state with that empty view.
        self.restore_or_init_state_record_key();

        // Then adopt anything the pre-split build left in secure storage, so
        // the restores below see one complete view. Must follow the record key
        // (adoption seals what it moves) and precede every restore (a restore
        // that ran first would find nothing and then overwrite the legacy state
        // with that empty view). Best-effort and resumable: a failure here
        // leaves the legacy records in place for the next launch.
        if adopt_legacy_state {
            self.adopt_legacy_protocol_state();
        }

        // The settlement queue is deliberately NOT captured for rollback, and
        // that is the whole point rather than an omission.
        //
        // The rollback below restores *in-memory* state. It restores no storage
        // state, because restore has none to give back: by the time a later
        // step fails, an earlier one has already deleted an unaddressable
        // pending queue, dropped a record its store reported corrupt, and
        // rewritten a peer's snapshot without the entries it evicted for
        // capacity. Those records are gone. So the invariant adoption is
        // documented with — nothing can re-derive a settlement for a record
        // that no longer exists — is not special to adoption; it holds for
        // every settlement produced under this transaction, and rolling any of
        // them back means the application keeps an id that resolves to nothing,
        // on this launch and on every launch after it.
        //
        // The cost of not rolling back is that a retry which re-examines a
        // still-present record can settle the same id twice. `message_failed`
        // is terminal and idempotent for any reasonable consumer, and a
        // duplicate terminal event is a far smaller lie than silence.
        //
        // Load the persistent scrub secret outside the transactional restore
        // below: it is independent of MLS state, and a later MLS-restore
        // rollback must not undo it (the secret is idempotent and reused on
        // the next attempt).
        self.restore_or_init_scrub_secret();

        // Same reasoning for the Nostr signing secret: independent of MLS
        // state, idempotent, and must survive an MLS-restore rollback.
        self.restore_or_init_nostr_signing_secret();

        // And for the Nostr receive watermark, which sits outside the
        // transaction below for a third reason on top of those two: it does
        // not live in `self` at all. The mark is transport state, so the
        // rollback — which restores in-memory fields of `OfflineProtocol` —
        // could not put it back even if it wanted to. Installing it under a
        // later failure is harmless: it only ever moves forward, and the worst
        // a stale-but-installed mark does is narrow one replay window.
        self.restore_nostr_watermark();
        self.restore_nostr_publication_slots();
        self.restore_nostr_discovery_claim();

        // The launch's whole durable-delete allowance, in one place because the
        // bound is on the launch. A device-barrier storm kills *this call*, not
        // any single walk in it, so no walk may hand itself a pool — see
        // `PruneAllowance::pool`. Four of them, and the ceiling is their sum:
        //
        // - `advisory_prunes` is shared by the `ADVISORY_PRUNE_WALKS` walks
        //   whose prunes are caches or advisory signals. They draw in the order
        //   written below, and that order is part of the contract: each leaves
        //   the ones after it a `MIN_ADVISORY_PRUNE_DELETES` floor and may take
        //   the rest, so an earlier walk gets first call on whatever the later
        //   ones do not reserve. Reordering them, or adding a sixth without
        //   bumping `ADVISORY_PRUNE_WALKS`, changes who gets pruned under load.
        // - `pending_prunes` and `outbox_prunes` are private to the two
        //   settlement-paired walks, because being starved there defers a
        //   diagnostic or a *delivery* rather than a cache eviction, and
        //   neither may be held hostage to a key-package flood in a category it
        //   has nothing to do with.
        // - `inbound_prunes` is private to the two receive-side walks for the
        //   same reason: the pending-decryption walk holds ciphertext the app
        //   is told about when it is lost. It draws first, and the seen-set
        //   restore takes whatever is left for its one possible delete.
        //
        // See `storage::MAX_RESTORE_PRUNE_DELETES`.
        let mut advisory_prunes = PruneAllowance::pool();
        let mut pending_prunes = PruneAllowance::pool();
        let mut outbox_prunes = PruneAllowance::pool();
        let mut inbound_prunes = PruneAllowance::pool();

        // Restore state from previous session
        let restore_result = (|| {
            self.restore_pending_messages(&mut pending_prunes)?;
            self.restore_pending_decrypt_entries(&mut inbound_prunes);
            self.restore_custody(&mut inbound_prunes);
            self.restore_lamport_clock();
            self.restore_dedup_seen(&mut inbound_prunes);
            self.restore_encryption_capable_peers();
            self.restore_blocked_users()?;
            self.restore_session_states_from_manager(manager.clone(), &mut advisory_prunes)?;
            self.restore_peer_key_packages(&manager, &mut advisory_prunes)?;
            // Must precede start(): flush_restored_confirmed_pending_messages
            // re-makes the rich seal decision against these sets, and an
            // empty set there silently drops queued rich extras.
            self.restore_peer_capabilities(&manager, &mut advisory_prunes);
            self.restore_welcome_lifecycles(&mut advisory_prunes)?;
            self.restore_outbox(&mut outbox_prunes)?;
            self.restore_media_descriptors(&mut advisory_prunes)?;
            self.restore_both_create_awaiting_decrypt();
            Ok(())
        })();

        if let Err(err) = restore_result {
            self.mesh_relay.set_local_id(previous_local_id.clone());
            self.local_id = previous_local_id;
            self.identity_established = previous_identity_established;
            self.secure_storage = previous_secure_storage;
            self.protocol_state_storage = previous_protocol_state_storage;
            self.state_record_cipher = previous_state_record_cipher;
            self.pending_encrypted_messages = previous_pending_messages;
            self.next_pending_message_expiry = previous_pending_message_expiry;
            // Back to what the queue held before this call. Emptying it lost
            // frames that were never persisted (no store was attached, or the
            // best-effort write failed). Keeping what the restore added would
            // leave entries sourced from the store the rollback just detached:
            // draining one deletes nothing on disk, so the next launch restores
            // it and drains it again into a spent ratchet generation.
            self.pending_queue = previous_pending_queue;
            // `deferred_restore_settlements` is deliberately left alone — see
            // the comment where the other baselines are captured.
            self.pending_key_packages = previous_pending_key_packages;
            self.restored_media_descriptors = previous_restored_media_descriptors;
            self.both_create_awaiting_decrypt = previous_both_create_awaiting_decrypt;
            self.confirmed_sessions = previous_confirmed_sessions;
            // Restored for consistency with the rest of the transaction, not
            // for safety: this set only ever grows, so keeping the wider view
            // would also have been sound. Rolling it back keeps "a failed
            // initialize_mls leaves no trace" true without exception.
            self.encryption_capable_peers = previous_encryption_capable_peers;
            self.welcome_lifecycles = previous_welcome_lifecycles;
            self.lamport_clock = LamportClock::from_value(previous_lamport_clock);
            self.last_persisted_lamport = previous_last_persisted_lamport;
            self.blocked_users = previous_blocked_users;
            self.outbox = previous_outbox;
            self.peer_compact_envelope = previous_peer_compact_envelope;
            self.peer_ctrl_freshness = previous_peer_ctrl_freshness;
            self.peer_undecryptable_ack = previous_peer_undecryptable_ack;
            self.control_freshness_peers = previous_control_freshness_peers;
            self.control_reset_watermark = previous_control_reset_watermark;
            self.peer_rich_payload = previous_peer_rich_payload;
            self.peer_svc_sealed = previous_peer_svc_sealed;
            self.svc_sealed_proved_peers = previous_svc_sealed_proved_peers;
            self.peer_data_sync = previous_peer_data_sync;
            self.peer_rich_attested = previous_peer_rich_attested;
            self.peer_data_group = previous_peer_data_group;
            self.peer_data_group_attested = previous_peer_data_group_attested;
            self.peer_data_media = previous_peer_data_media;
            self.peer_data_tombstones = previous_peer_data_tombstones;
            self.peer_data_interest = previous_peer_data_interest;
            self.peer_data_group_blob = previous_peer_data_group_blob;
            self.peer_data_group_blob_attested = previous_peer_data_group_blob_attested;
            return Err(err);
        }

        self.mls_manager = Some(manager);
        #[cfg(feature = "telemetry-pipe")]
        self.attach_telemetry_storage();
        // A re-init can swap the identity, and with it which groups exist, so
        // the one-shot group enumeration has to run again. Past the last
        // `return Err` above, so a failed init leaves the flag alone.
        #[cfg(feature = "data")]
        {
            self.group_spaces_enumerated = false;
        }
        self.emit_mls_initialized();

        // Announced only here, past the last `return Err` — a restore failure
        // rolls the address back, and announcing an identity the instance then
        // abandoned would leave the app addressing a device that is not this
        // one.
        self.emit_event(Event::identity_ready(self.local_id.clone()));

        info!(profile = %self.config.profile, "MLS encryption initialized with split secure and protocol-state storage");
        Ok(())
    }

    /// Test-only adapter for fixtures that intentionally use one
    /// in-memory/fault-injection backend to observe both storage domains.
    #[cfg(test)]
    pub(crate) fn initialize_mls_for_test(&mut self, storage: Arc<dyn MlsStorage>) -> Result<()> {
        let protocol_state_storage = Arc::new(TestProtocolStateStorage {
            storage: storage.clone(),
        });
        // Give the storage the identity this fixture's profile label stands
        // for, then take the production path over it. The instance comes up as
        // a genuine derived address, and it is the *same* address every time a
        // fixture says `"alice"` — which is what lets tests keep readable
        // labels while still exercising derived identity.
        crate::test_identity::seed_identity(&storage, &self.config.profile);
        self.initialize_mls_inner(storage, protocol_state_storage, false)
    }

    /// Test-only init that takes the **production** identity path: the address
    /// is derived from the identity key in `storage` rather than forced.
    #[cfg(test)]
    pub(crate) fn initialize_mls_for_test_derived(
        &mut self,
        storage: Arc<dyn MlsStorage>,
    ) -> Result<()> {
        let protocol_state_storage = Arc::new(TestProtocolStateStorage {
            storage: storage.clone(),
        });
        self.initialize_mls_inner(storage, protocol_state_storage, false)
    }

    /// Test-only storage initialization for persistence-focused fixtures that
    /// do not need an MLS manager.
    #[cfg(test)]
    pub(crate) fn enable_message_persistence_for_test(
        &mut self,
        storage: Arc<dyn MlsStorage>,
    ) -> Result<()> {
        self.secure_storage = Some(storage.clone());
        self.protocol_state_storage = Some(Arc::new(TestProtocolStateStorage { storage }));
        self.restore_or_init_state_record_key();
        self.restore_or_init_scrub_secret();
        self.restore_or_init_nostr_signing_secret();
        self.restore_nostr_watermark();
        self.restore_nostr_publication_slots();
        self.restore_nostr_discovery_claim();
        // Same four pools as `initialize_mls_inner`, for the same reason.
        let mut advisory_prunes = PruneAllowance::pool();
        let mut pending_prunes = PruneAllowance::pool();
        let mut outbox_prunes = PruneAllowance::pool();
        let mut inbound_prunes = PruneAllowance::pool();
        self.restore_pending_messages(&mut pending_prunes)?;
        self.restore_pending_decrypt_entries(&mut inbound_prunes);
        self.restore_custody(&mut inbound_prunes);
        self.restore_lamport_clock();
        self.restore_dedup_seen(&mut inbound_prunes);
        self.restore_encryption_capable_peers();
        self.restore_blocked_users()?;
        self.restore_outbox(&mut outbox_prunes)?;
        self.restore_media_descriptors(&mut advisory_prunes)?;
        Ok(())
    }

    /// Checks if MLS encryption is initialized.
    pub fn is_mls_initialized(&self) -> bool {
        self.mls_manager.is_some()
    }

    /// Records that `peer_id` has demonstrated it runs MLS.
    ///
    /// The only way into [`Self::encryption_capable_peers`]. Call it from
    /// anywhere a peer proves encryption capability — creating or joining a 1:1
    /// session with them, or pinning their signing key. Calling it redundantly
    /// is free and calling it too often is safe; the failure that matters is
    /// *not* calling it, which leaves the plaintext gate open for that peer.
    ///
    /// # Bounded by refusal, never by eviction
    ///
    /// The keys are wire-claimed ids and need their own cap, like every other
    /// map keyed that way — but not for the reason this comment used to give.
    /// It cited an unsigned `__MLS_WELCOME__` reaching the marking path; the
    /// control gate now rejects unsigned control frames before dispatch, and
    /// `__MLS_WELCOME__` is in neither exemption list, so `handle_welcome_message`
    /// only ever runs on a verified frame.
    ///
    /// The cap survives that because a signature was never the bound. A
    /// signature proves the signer holds the key its address derives from; it
    /// does not prove the address belongs to anyone real, and minting a fresh
    /// one costs a keygen. So an attacker willing to spend keygens can still
    /// present an unbounded supply of distinct, perfectly well-signed peers —
    /// which is precisely what a cap on a wire-keyed map is for.
    ///
    /// It has to be a refusal. Evicting would un-mark a peer that really is
    /// encryption-capable — the fail-open direction this field exists to
    /// prevent — and would hand an attacker a way to unprotect a *chosen*
    /// victim by flooding. Refusing only declines to add knowledge: that peer
    /// falls through to the `is_session_confirmed` check in
    /// [`Self::accept_plaintext_content`], which is exactly what shipped before
    /// this set existed. So a flood costs later peers the improvement and never
    /// costs anyone the protection they already had.
    ///
    /// Legitimate peers are also first in line by construction: restore seeds
    /// the set from the session list and the durable capability records during
    /// `initialize_mls`, before `start()` admits any traffic.
    ///
    /// # Return value
    ///
    /// `true` when `peer_id` is in the set on return — whether this call added
    /// it or a previous one did. This is what bounds the *durable* record:
    /// [`Self::record_encryption_capable`](crate::OfflineProtocol) persists only
    /// when this returns `true`, so the storage category can never hold more
    /// peers than the cap below. Without that coupling a Sybil flood — cheap,
    /// since each identity is one Ed25519 keygen that signs honestly as itself
    /// — would write unbounded records, and the restore walk (bounded by
    /// `MAX_RESTORE_KEYS_PER_CATEGORY`) could then miss a legitimate peer's
    /// record and re-open the plaintext gate for them on the next launch.
    pub(crate) fn mark_encryption_capable(&mut self, peer_id: &str) -> bool {
        if peer_id.is_empty() || peer_id == self.local_id {
            return false;
        }
        if self.encryption_capable_peers.contains(peer_id) {
            return true;
        }
        if self.encryption_capable_peers.len() >= MAX_ENCRYPTION_CAPABLE_PEERS {
            debug!(
                peer_id = %peer_id,
                cap = MAX_ENCRYPTION_CAPABLE_PEERS,
                "Encryption-capability set at capacity; peer falls back to the session-state check"
            );
            return false;
        }
        self.encryption_capable_peers.insert(peer_id.to_string());
        debug!(peer_id = %peer_id, "Peer marked encryption-capable");
        if self.encryption_capable_peers.len() == MAX_ENCRYPTION_CAPABLE_PEERS {
            // Emitted on the transition into full, so it cannot become a
            // per-frame flood: the set only grows and has no removal path, so
            // this line is reachable exactly once per process.
            warn!(
                cap = MAX_ENCRYPTION_CAPABLE_PEERS,
                "Encryption-capability set reached capacity; further peers fall back to the \
                 session-state check. Expected only under a forged-sender flood."
            );
        }
        true
    }

    /// Whether `peer_id` is known to run MLS. See
    /// [`Self::encryption_capable_peers`] for why this is not a storage read.
    pub(crate) fn is_encryption_capable(&self, peer_id: &str) -> bool {
        self.encryption_capable_peers.contains(peer_id)
    }

    /// Drops `peer_id` from the in-memory capability set **only**, leaving its
    /// durable record and write-elision cache entry in place.
    ///
    /// Test-only, and deliberately not a general "forget" — the set has no
    /// removal path by design (see [`Self::encryption_capable_peers`]). It
    /// reproduces one specific state that production does reach: the rollback in
    /// `initialize_mls` restores the pre-restore snapshot of the set while
    /// [`Self::encryption_capable_persisted`] keeps what the restore walk read,
    /// so the two disagree until the peer's next verified frame re-marks them.
    #[cfg(test)]
    pub(crate) fn forget_encryption_capable_for_test(&mut self, peer_id: &str) {
        self.encryption_capable_peers.remove(peer_id);
    }

    /// Returns whether auto-encryption should be applied.
    fn should_auto_encrypt(&self) -> bool {
        self.config.encryption.enabled && self.mls_manager.is_some()
    }

    /// Configures the MLS lifecycle event emitter.
    pub fn set_mls_event_emitter(&mut self, emitter: Arc<dyn MlsEventEmitter>) {
        self.mls_event_emitter = emitter;
    }

    /// Installs the unified telemetry sink for this protocol instance.
    ///
    /// Every subsequent protocol [`Event`] and MLS lifecycle event is
    /// additionally forwarded to `sink` as a
    /// [`crate::telemetry::TelemetryRecord`], gated by the verbosity tier
    /// and identifier-scrubbing preferences in `config`. Protocol events
    /// flow through as [`TelemetryRecord::Protocol`], MLS lifecycle events
    /// as [`TelemetryRecord::Mls`].
    ///
    /// This does not replace the legacy [`crate::EventCallback`] and
    /// [`MlsEventEmitter`] paths — they continue to fire independently for
    /// backward compatibility.
    ///
    /// Events emitted before this call (for example
    /// [`crate::mls_observability::MlsLifecycleEvent::Initialized`] fired
    /// during `initialize_mls`) are not replayed to a sink installed later.
    ///
    /// # Re-install semantics
    ///
    /// Calling this method a second time *replaces* the previously installed
    /// sink and config — the old sink stops receiving records as soon as
    /// this call returns. The per-instance fallback secret is preserved so
    /// opaque identifiers stay stable across the swap when the new config
    /// does not carry its own `scrub_secret`.
    ///
    /// The per-tick diff snapshots (transport status, device capability,
    /// metrics cadence) are **rearmed** on every install: the next
    /// `process()` tick observes the current state as its baseline and emits
    /// no synthetic transitions for transports that were already available
    /// at install time. Apps that need the current transport statuses at
    /// install time should pull them explicitly via
    /// [`TransportManager::get_all_transport_statuses`].
    ///
    /// The escalation-trigger dedupe window (`ESCALATION_TRIGGER_DEDUPE_SECS`
    /// inside [`TransportManager`]) is **not** rearmed. A sink installed
    /// within that window after an escalation event fired to a previous
    /// sink may miss the next occurrence of the same reason until the
    /// window elapses. This is intentional: the dedupe is a property of
    /// the routing engine, not of the observer.
    ///
    /// # Lifecycle
    ///
    /// Installing a sink wires the structured routing-decision callback
    /// on the underlying [`TransportManager`]. The callback lives for the
    /// life of the protocol instance or until another
    /// `install_telemetry_sink` call replaces it — `start()` and `stop()`
    /// do not toggle it. This means apps can install a sink before
    /// `start()` and see routing records from the very first `send()`,
    /// and a `stop() → start()` cycle preserves the wiring so no
    /// re-install is needed. Conversely, apps that never install a sink
    /// pay no per-`send()` routing-emission overhead.
    ///
    /// # Errors
    ///
    /// Returns `Err` if the shared-state mutex is poisoned (indicating an
    /// earlier panic in another thread while holding the lock). On error no
    /// sink is installed, and the legacy emission paths are unaffected.
    ///
    /// [`TelemetryRecord::Protocol`]: crate::telemetry::TelemetryRecord::Protocol
    /// [`TelemetryRecord::Mls`]: crate::telemetry::TelemetryRecord::Mls
    // Crate-private since 0.26: the pipe is the one production sink, and the
    // engine's own tests are the remaining callers.
    #[cfg_attr(not(test), allow(dead_code))]
    pub(crate) fn install_telemetry_sink(
        &mut self,
        sink: Arc<dyn TelemetrySink>,
        config: TelemetryConfig,
    ) -> Result<()> {
        self.install_sink(
            sink,
            config,
            Arc::new(std::sync::atomic::AtomicBool::new(true)),
        )
    }

    /// Installs `sink` behind `gate`, the flag every emit site reads first.
    fn install_sink(
        &mut self,
        sink: Arc<dyn TelemetrySink>,
        config: TelemetryConfig,
        gate: Arc<std::sync::atomic::AtomicBool>,
    ) -> Result<()> {
        // Forward the routing-diagnostic preference to the TransportManager
        // before wiring any callbacks so the very first routing decision
        // post-install already reflects the requested tier.
        self.transport_manager
            .set_routing_diagnostic(config.routing_diagnostic());

        let ctx = TelemetryContext::with_gate(sink, config, self.telemetry_fallback_secret, gate);
        let mut state = lock_shared_state(&self.shared_state).map_err(|err| {
            error!(
                error = %err,
                "install_telemetry_sink: shared-state mutex poisoned; sink NOT installed",
            );
            err
        })?;
        state.telemetry = Some(ctx.clone());
        drop(state);
        self.telemetry = Some(ctx);

        // Wire the structured routing-decision callback here so the
        // `TransportManager` pays the per-`send()` emission cost only when
        // a sink is actually installed, and so the wiring persists across
        // `stop() → start()` cycles (the callback reads `s.telemetry` on
        // every invocation and that field stays set). A subsequent
        // `install_telemetry_sink` replaces the closure with one capturing
        // the fresh `shared_routing` clone; both closures read the same
        // `SharedState::telemetry` slot, so even a concurrent in-flight
        // emission from the pre-replace closure would still resolve to
        // the currently-installed sink.
        let shared_routing = self.shared_state.clone();
        self.transport_manager
            .set_routing_decision_callback(Some(Arc::new(move |decision| {
                let s = shared_routing.lock_or_recover();
                if let Some(ctx) = s.telemetry.as_ref().filter(|ctx| ctx.enabled()) {
                    let record = TelemetryRecord::Routing(Box::new(decision));
                    // Dispatch is panic-isolated so a sink that panics
                    // here cannot unwind through the live `MutexGuard`
                    // and poison `SharedState`.
                    dispatch_record(&ctx.sink, &record);
                }
            })));

        // Rearm the per-tick diff snapshots so the first tick after install
        // reports honest transitions only. Without this, a sink installed
        // after start() would observe a synthetic `Unavailable → Available`
        // transition for every already-running transport on the next tick.
        let (statuses, available) = self.transport_manager.snapshot_status_and_available();
        self.transport_status_snapshot = statuses;
        let (battery_level, is_charging) =
            device_battery_from_available(self.transport_manager.current_transport(), &available);
        let relay_role = self.relay_role();
        self.device_capability_snapshot = Some(DeviceSnap::from_parts(
            battery_level,
            is_charging,
            relay_role,
        ));
        // Leave `last_metrics_emit_at` at None so the first tick after
        // install fires a fresh metrics snapshot — that is the bootstrap
        // payload a new sink actually wants (full counter state in one
        // record).
        self.last_metrics_emit_at = None;

        Ok(())
    }

    /// Enables telemetry: the SDK collects, batches and uploads accepted
    /// events to the ingest itself.
    ///
    /// `config` must carry the `api_key` and `app_id` the developer portal
    /// issued; [`TelemetryConfig::validate`] names any field that is refused,
    /// and the refusal is [`Error::TelemetryConfigInvalid`]. `host` names the
    /// platform stamped on every batch and the application state right now.
    ///
    /// Nothing waits on I/O here: the pipe's storage load and first upload
    /// run on its own thread. Batches are durable across a restart once a
    /// protocol-state store is attached, which on the mobile bindings
    /// happens inside `initialize_mls`; before that they are held in memory.
    /// Calling this again replaces the running pipe after its final flush.
    ///
    /// What leaves the device, when, and how to switch it off are in
    /// `docs/telemetry.md`.
    #[cfg(feature = "telemetry-pipe")]
    pub fn enable_telemetry(&mut self, config: TelemetryConfig, host: TelemetryHost) -> Result<()> {
        if let Err(field) = config.validate() {
            return Err(Error::TelemetryConfigInvalid(field.to_string()));
        }
        if let Some(previous) = self.detach_telemetry_pipe() {
            previous.stop(FINAL_FLUSH_BUDGET);
        }
        // Always started in memory, even when a store is already attached.
        // Adopting a backend loads the durable queue: an index read, up to
        // `MAX_PENDING_BATCHES` sealed record reads with a decrypt and a parse
        // each, a `list_keys` and the orphan sweep. Doing that here would run
        // it on the caller, holding `&mut self`, which on the mobile bindings
        // is the common order: `initialize_mls` has already attached storage
        // by the time an app enables telemetry. `attach_telemetry_storage`
        // hands the backend over instead, and the pipe's own thread adopts it.
        let device_id = self.telemetry_install_id();
        let pipe = TelemetryPipe::start(
            &config,
            PipeParts::production(host, Backend::Memory, device_id),
        )?;
        self.install_sink(pipe.sink(), config, pipe.enabled_flag())?;
        self.telemetry_pipe = Some(pipe);
        self.attach_telemetry_storage();
        Ok(())
    }

    /// Disables telemetry: one final flush within [`FINAL_FLUSH_BUDGET`], then
    /// the uploader thread stops and every emit site goes back to its
    /// telemetry-off cost. Callers holding an outer lock should prefer
    /// [`Self::detach_telemetry_pipe`] and stop the pipe outside it.
    #[cfg(feature = "telemetry-pipe")]
    pub fn disable_telemetry(&mut self) {
        if let Some(pipe) = self.detach_telemetry_pipe() {
            pipe.stop(FINAL_FLUSH_BUDGET);
        }
    }

    /// Detaches the running pipe without stopping it: the sink is
    /// uninstalled, the routing callback cleared, and the pipe handed back
    /// for the caller to stop once it holds no engine lock. Its final flush
    /// may block for up to [`FINAL_FLUSH_BUDGET`], which is why the two steps
    /// are separable.
    #[cfg(feature = "telemetry-pipe")]
    pub fn detach_telemetry_pipe(&mut self) -> Option<Arc<TelemetryPipe>> {
        let pipe = self.telemetry_pipe.take()?;
        self.telemetry = None;
        self.shared_state.lock_or_recover().telemetry = None;
        self.transport_manager.set_routing_decision_callback(None);
        self.transport_manager.set_routing_diagnostic(false);
        Some(pipe)
    }

    /// The running telemetry pipe, if any.
    #[cfg(feature = "telemetry-pipe")]
    pub fn telemetry_pipe(&self) -> Option<Arc<TelemetryPipe>> {
        self.telemetry_pipe.clone()
    }

    /// Stops or resumes collection without tearing the pipe down. Off, every
    /// emit site costs one atomic load; what is already queued still drains.
    #[cfg(feature = "telemetry-pipe")]
    pub fn set_telemetry_enabled(&self, enabled: bool) {
        if let Some(pipe) = &self.telemetry_pipe {
            pipe.set_enabled(enabled);
        }
    }

    /// Asks the uploader to flush now. Returns immediately.
    #[cfg(feature = "telemetry-pipe")]
    pub fn flush_telemetry(&self) {
        if let Some(pipe) = &self.telemetry_pipe {
            pipe.flush();
        }
    }

    /// Flushes and waits up to `budget` for the uploader to finish.
    #[cfg(feature = "telemetry-pipe")]
    pub fn flush_telemetry_blocking(&self, budget: std::time::Duration) -> bool {
        self.telemetry_pipe
            .as_ref()
            .is_some_and(|pipe| pipe.flush_blocking(budget))
    }

    /// The pipe's counters, or `None` while telemetry is not enabled.
    #[cfg(feature = "telemetry-pipe")]
    pub fn telemetry_stats(&self) -> Option<TelemetryStats> {
        self.telemetry_pipe.as_ref().map(|pipe| pipe.stats())
    }

    /// Closes the current telemetry session: summary, a blocking flush,
    /// then a fresh session id.
    #[cfg(feature = "telemetry-pipe")]
    pub fn end_telemetry_session(&self) {
        if let Some(pipe) = &self.telemetry_pipe {
            pipe.end_session();
        }
    }

    /// Reports an application lifecycle transition to the pipe.
    #[cfg(feature = "telemetry-pipe")]
    pub fn notify_app_state(&self, state: AppState) {
        if let Some(pipe) = &self.telemetry_pipe {
            pipe.notify_app_state(state);
        }
    }

    /// Hands the pipe the protocol-state store and the per-install id once
    /// `initialize_mls` has them. Adoption happens on the pipe's own thread.
    #[cfg(feature = "telemetry-pipe")]
    fn attach_telemetry_storage(&self) {
        let Some(pipe) = &self.telemetry_pipe else {
            return;
        };
        if let (Some(storage), Some(cipher)) =
            (&self.protocol_state_storage, &self.state_record_cipher)
        {
            pipe.attach_storage(storage.clone(), cipher.clone());
        }
        pipe.set_device_id(self.telemetry_install_id());
    }

    /// Starts the protocol.
    ///
    /// # Returns
    ///
    /// Returns `Ok(())` if started successfully, `Err` if already started.
    pub fn start(&mut self) -> Result<()> {
        let state = lock_shared_state(&self.shared_state)?;

        if state.state != ProtocolState::Stopped {
            return Err(Error::AlreadyStarted);
        }

        // Start all transports
        drop(state);
        self.transport_manager.start()?;

        // Apply the Nostr sealing kill switch here rather than at transport
        // construction: the transport is built by the bindings layer, so this
        // is the one place every caller — FFI, RN, and a pure-Rust embedder
        // that installed its own `NostrTransport` — passes through with the
        // config in hand. A no-op when Nostr is not installed.
        self.transport_manager
            .set_nostr_sealing_enabled(self.config.transport.nostr_sealing_enabled);
        self.transport_manager
            .set_nostr_cold_contact_enabled(self.config.transport.nostr_cold_contact_enabled);
        self.transport_manager.set_nostr_username_discovery_enabled(
            self.config.transport.nostr_username_discovery_enabled,
        );

        // Wire DORS event callback so app receives dors_score_updated, dors_transport_selected,
        let shared = self.shared_state.clone();
        self.transport_manager
            .set_dors_event_callback(Some(Arc::new(move |event| {
                shared.lock_or_recover().emit_event(event);
            })));

        // NOTE: the structured routing-decision callback is wired by
        // `install_telemetry_sink`, not here. Its lifetime is tied to sink
        // presence rather than protocol start/stop — see the docstring on
        // `install_telemetry_sink` for the rationale.

        // Wire BLE fragment eviction callback so app receives FragmentAssemblyEvicted.
        if let Some(ble_arc) = self.transport_manager.get_transport(TransportType::BLE) {
            let shared = self.shared_state.clone();
            if let Some(ble) = ble_arc.as_any().downcast_ref::<BleTransport>() {
                ble.set_fragment_eviction_callback(Some(Arc::new(move |info| {
                    shared
                        .lock_or_recover()
                        .emit_event(Event::fragment_assembly_evicted(
                            info.message_id,
                            info.completion_percent,
                            "capacity".to_string(),
                        ));
                })));
            }
        }

        let mut state = lock_shared_state(&self.shared_state)?;

        state.state = ProtocolState::Running;
        drop(state);

        // Settle what restore could not recover, now that the event pipeline is
        // live. These ids were handed to the app by `send_message*` before the
        // process died; without this they would never resolve to anything.
        // Drained before the flush below so an app sees the failures first and
        // cannot mistake a restored send for the settlement of a lost one.
        self.drain_deferred_restore_settlements();

        self.flush_restored_confirmed_pending_messages();

        // Cold start: sessions restored from storage are already confirmed,
        // so no confirmation transition will fire for them and nothing else
        // would offer versions until the peer happened to be rediscovered.
        // Frames toward an unreachable peer queue in the outbox, which is the
        // ladder doing its job rather than a reason not to try.
        #[cfg(feature = "data")]
        {
            // The space list is derived from the records themselves rather
            // than from the space index, because the index is a cache: a
            // document written without ever being explicitly created leaves
            // records but no index entry, and those are precisely the
            // documents no peer has seen yet.
            // Group spaces appear in this list and are deliberately not
            // swept: `kick_data_sync` answers only for a space named after
            // a peer, so a group space falls through it. A cold-start sweep
            // for groups would mean an offer per member per group at every
            // launch, and what it would recover is narrow — a local commit
            // is already pushed to the roster when it happens, and the
            // per-member outbox carries it across a restart. Rediscovery
            // and joins reconcile the rest.
            for space in self.data_list_spaces().unwrap_or_default() {
                self.kick_data_sync(&space, "start");
            }
        }

        self.kick_pending_session_reconciliation("start");
        self.process_welcome_retry_queue()?;

        // Re-drive delivery of any outbox entries restored from a previous
        // session. They land here in the "stranded" state (not in the retry
        // queue, not awaiting an ACK), which flush_outbox_all already recovers:
        // it sends immediately where a transport is available and re-enqueues
        // the rest with backoff. Runs at start() rather than at restore time
        // because transports aren't up yet during initialize_mls.
        self.flush_outbox_all();

        // Media transfers that died with the previous process cannot be
        // re-driven — only their descriptors survive (never chunk bytes).
        // Tell the app which transfers to re-initiate, now that the event
        // pipeline is live. The descriptors stay parked (and persisted)
        // until the resend consumes them or the restore TTL prunes them,
        // so an app that misses this signal gets it again next restart.
        //
        // Except the layer's own. An application cannot answer one of those:
        // the only surface that takes a caller-supplied `file_id` is
        // `send_media_with`, which forces the purpose to `None`, so a
        // well-behaved app re-supplying the bytes would deliver a document
        // snapshot to the peer's user as a downloaded file. They are dropped
        // instead, and nothing is stranded: the requester's fetch expires and
        // can be asked again, and a snapshot is re-offered by the next
        // anti-entropy exchange.
        let (recoverable, internal): (Vec<_>, Vec<_>) = self
            .restored_media_descriptors
            .values()
            .cloned()
            .partition(|descriptor| descriptor.data_purpose.is_none());
        for descriptor in &internal {
            debug!(
                file_id = %descriptor.file_id,
                "Dropping an interrupted document-layer transfer; the asking side will retry"
            );
            self.restored_media_descriptors.remove(&descriptor.file_id);
            self.remove_media_descriptor(&descriptor.file_id);
        }
        for descriptor in recoverable {
            self.emit_event(Event::media_resend_required(
                descriptor.file_id.clone(),
                descriptor.recipient.clone(),
                descriptor.file_name.clone(),
                descriptor.file_size,
                descriptor.app_id.clone(),
            ));
        }

        Ok(())
    }

    /// Stops the protocol gracefully.
    ///
    /// # Returns
    ///
    /// Returns `Ok(())` if stopped successfully, `Err` if not started.
    pub fn stop(&mut self) -> Result<()> {
        let state = lock_shared_state(&self.shared_state)?;

        if state.state == ProtocolState::Stopped {
            return Ok(()); // Already stopped
        }

        // Flush debounced Lamport clock before stopping so no ticks are lost.
        drop(state);
        self.flush_lamport_clock();
        // Same for the Nostr receive watermark: an un-flushed tail costs the
        // next launch a wider replay window.
        self.flush_nostr_watermark();
        // And the deduplicator's seen set: an un-flushed tail is a message
        // whose second copy the next launch fails to recognise.
        self.flush_dedup_seen();
        // Answer any lookup still in flight while the transport is still up to
        // take the cancellations. Nothing pumps relays or sweeps deadlines once
        // stopped, so a resolution left here is an event that never comes.
        self.flush_username_resolutions_for_stop();

        // Clear event callbacks to release shared_state references.
        // NOTE: the routing-decision callback is deliberately NOT cleared
        // here. Its lifetime is sink-scoped (installed by
        // `install_telemetry_sink`, replaced by a subsequent install),
        // not protocol-running-scoped — so a `stop() → start()` cycle
        // preserves the wiring without requiring the app to re-install.
        self.transport_manager.set_dors_event_callback(None);
        if let Some(ble_arc) = self.transport_manager.get_transport(TransportType::BLE) {
            if let Some(ble) = ble_arc.as_any().downcast_ref::<BleTransport>() {
                ble.set_fragment_eviction_callback(None);
            }
        }

        self.transport_manager.stop()?;
        let mut state = lock_shared_state(&self.shared_state)?;

        state.state = ProtocolState::Stopped;

        Ok(())
    }

    /// Pauses the protocol (for background mode).
    pub fn pause(&mut self) -> Result<()> {
        self.flush_lamport_clock();
        self.flush_nostr_watermark();
        let mut state = lock_shared_state(&self.shared_state)?;

        if state.state != ProtocolState::Running {
            return Err(Error::NotStarted);
        }

        state.state = ProtocolState::Paused;
        Ok(())
    }

    /// Resumes the protocol from pause.
    pub fn resume(&mut self) -> Result<()> {
        {
            let mut state = lock_shared_state(&self.shared_state)?;

            if state.state != ProtocolState::Paused {
                return Err(Error::InvalidConfiguration(
                    "Protocol is not paused".to_string(),
                ));
            }

            state.state = ProtocolState::Running;
        }

        // A pause is the other edge back into a live event pipeline, so it owes
        // the same drain `start()` does. `settle_restored_message_failure` parks
        // anything it produces while the protocol is not `Running`, and
        // `update_retry_config` reaches it at runtime: shortening
        // `pending_message_max_lifetime_ms` in the background expires queued
        // messages and parks their terminal `message_failed`. Without this the
        // app would hold those ids until a `start()` that may never come again.
        self.drain_deferred_restore_settlements();
        Ok(())
    }

    /// Emits every terminal settlement parked while the event pipeline was not
    /// live, and reports the count of any the cap dropped.
    ///
    /// Called from both edges into `Running` ([`Self::start`] and
    /// [`Self::resume`]) so a parked settlement has no state to be stranded in.
    /// Idempotent: draining an empty queue emits nothing.
    fn drain_deferred_restore_settlements(&mut self) {
        for event in std::mem::take(&mut self.deferred_restore_settlements) {
            self.emit_event(event);
        }
        let suppressed = std::mem::take(&mut self.suppressed_restore_settlements);
        if suppressed > 0 {
            warn!(
                suppressed,
                cap = MAX_DEFERRED_RESTORE_SETTLEMENTS,
                "Restore produced more terminal settlements than any legitimate run can produce; \
                 the ids past the cap were not reported individually"
            );
        }
    }

    /// Forget that a peer replicates documents.
    ///
    /// The capability and the offer window are two halves of one fact, and
    /// they have to be dropped together: a window left behind outlives the
    /// capability and then suppresses the first offer to the same peer once
    /// it is relearned, which is a document that silently does not sync.
    pub(crate) fn forget_data_sync_peer(&mut self, peer: &str) {
        self.peer_data_sync.remove(peer);
        self.peer_data_group.remove(peer);
        self.peer_data_group_attested.remove(peer);
        self.peer_data_media.remove(peer);
        self.peer_data_tombstones.remove(peer);
        self.peer_data_interest.remove(peer);
        self.peer_data_group_blob.remove(peer);
        self.peer_data_group_blob_attested.remove(peer);
        self.peer_data_custody.remove(peer);
        // A peer we have stopped replicating with cannot answer anything we
        // asked them for, so the questions go too. Left behind they would
        // hold slots against the fetch bound until they timed out.
        #[cfg(feature = "data")]
        self.forget_attachment_state(peer);
        #[cfg(feature = "data")]
        self.forget_data_sync_offer_windows(peer);
    }

    /// [`Self::forget_data_sync_peer`] for every peer at once.
    pub(crate) fn forget_every_data_sync_peer(&mut self) {
        self.peer_data_sync.clear();
        self.peer_data_group.clear();
        self.peer_data_group_attested.clear();
        self.peer_data_media.clear();
        self.peer_data_tombstones.clear();
        self.peer_data_interest.clear();
        self.peer_data_group_blob.clear();
        self.peer_data_group_blob_attested.clear();
        // Reported, not merely dropped, exactly as the single-peer road
        // reports. Nothing the application did reaches this one: the bound
        // on remembered peers is hit, a stranger's key package forgets every
        // capability at once, and a fetch outstanding at that moment ends
        // for a reason its asker has no way to see. Dropping the record
        // silently also forecloses the expiry that would have reported it
        // later, so the spinner this event exists to end never ends at all.
        #[cfg(feature = "data")]
        self.end_every_pending_fetch();
        self.blob_request_windows.clear();
        #[cfg(feature = "data")]
        self.last_data_sync_offer.clear();
    }

    /// Called when a new neighbor is discovered.
    ///
    /// When auto key exchange is enabled, this method sends our key package
    /// to the newly discovered peer to enable encrypted communication.
    ///
    /// # Arguments
    ///
    /// * `peer_id` - The ID of the discovered peer
    pub fn on_neighbor_discovered(&mut self, peer_id: &str) {
        self.on_neighbor_discovered_via(peer_id, None);
    }

    /// [`Self::on_neighbor_discovered`] with an optional transport override
    /// for the outbox flush (see [`Self::flush_outbox_for_peer_via`]).
    /// `unpark_via` is `Some(Internet)` when the relay presence-online edge
    /// re-enters discovery through its rescue branch: a message whose
    /// forced-internet re-drive just failed sits in the retry queue with no
    /// pending ACK, so this inner flush would pick it up again — and
    /// without the override DORS could route it into the mesh void with the
    /// park counter already cleared, past re-park's reach.
    fn on_neighbor_discovered_via(&mut self, peer_id: &str, unpark_via: Option<TransportType>) {
        // Don't track ourselves
        if peer_id == self.local_id {
            return;
        }

        // Don't track or auto-exchange keys with blocked users
        if self.is_user_blocked(peer_id) {
            debug!(peer_id = %peer_id, "Ignoring neighbor discovery for blocked user");
            return;
        }

        // Track discovered peers for service discovery and routing. Existing
        // peers get their last-seen refreshed; a new peer at capacity evicts
        // the least-recently-seen entry so a genuinely present neighbor is
        // never locked out by stale message-path senders (issue #140).
        if let Some(last_seen) = self.known_peers.get_mut(peer_id) {
            *last_seen = Instant::now();
        } else {
            if self.known_peers.len() >= MAX_KNOWN_PEERS {
                if let Some(victim) = self
                    .known_peers
                    .iter()
                    .min_by_key(|(_, seen)| **seen)
                    .map(|(id, _)| id.clone())
                {
                    debug!(peer_id = %peer_id, evicted = %victim, cap = MAX_KNOWN_PEERS, "Known peers at capacity, evicting least-recently-seen");
                    self.evict_known_peer(&victim);
                }
            }
            self.known_peers.insert(peer_id.to_string(), Instant::now());
        }

        // Flush any pending outbox messages destined for this peer
        self.flush_outbox_for_peer_via(peer_id, unpark_via);

        // Held frames for this neighbour, or to try through it
        // (`docs/spec/custody.md`, "Redelivery").
        self.redeliver_custody_to(peer_id);

        // A Welcome that stalled or expired while this peer was unreachable now
        // has a fresh delivery opportunity over the carrier that surfaced this
        // peer — re-arm it. No-op when there is no pending Welcome for the peer.
        self.rearm_welcome_for_peer(peer_id, "peer_rediscovered");

        // Offer document versions while the peer is in front of us. This has
        // to sit above the key-package guard below, which returns early for
        // any peer we have already exchanged with — that is to say, for
        // exactly the established peers whose documents drifted apart during
        // the partition this discovery just ended.
        #[cfg(feature = "data")]
        self.kick_data_sync(peer_id, "peer_rediscovered");

        // The same reconciliation for every group space shared with this
        // peer. A group space cannot ride the 1:1 trigger above: two
        // members of a group need not have a session with each other at
        // all, so for them this discovery is the only moment anything knows
        // they are both reachable.
        #[cfg(feature = "data")]
        self.kick_shared_group_data_sync(peer_id, "peer_rediscovered");

        // Only send key package if encryption is enabled and auto key exchange is on
        if !self.config.encryption.enabled || !self.config.encryption.auto_key_exchange {
            return;
        }

        // The first push is discovery's. A later discovery re-arms it only
        // when that push produced no session, at most once per interval.
        if self.key_package_sent_to.contains_key(peer_id) {
            self.rearm_key_package_for_peer(peer_id);
            return;
        }

        // Only if MLS is initialized
        if self.mls_manager.is_none() {
            return;
        }

        if let Err(e) = self.send_key_package_to(peer_id, false) {
            warn!(error = %e, peer_id = %peer_id, "Failed to send key package on discovery");
        }
    }

    /// Immediately attempts to send all pending outbox messages destined
    /// for a specific peer, bypassing backoff timers. Called on per-peer
    /// reachability edges (discovery, relay presence-online) to flush
    /// messages that were queued while the peer was unreachable.
    ///
    /// `unpark_via` is an optional transport override for the re-driven
    /// sends: `Some(Internet)` on the relay presence-online edge
    /// (`on_peer_presence`, including its rescue branch's re-entry through
    /// `on_neighbor_discovered_via`): the reachability
    /// proof is relay-scoped, so the re-drive must go out over the internet
    /// transport — DORS could otherwise route it back into the mesh, where
    /// the send locally succeeds against a peer that is not there,
    /// re-registering an unanswerable ACK that re-strands the message (and,
    /// with the platform bridge unwatching the peer on the online answer,
    /// no further presence edge would arrive to save it). Pinned media
    /// transports still win over the override.
    fn flush_outbox_for_peer_via(&mut self, peer_id: &str, unpark_via: Option<TransportType>) {
        // Probe ACKs first: with a live park counter, the peer's DMs may be
        // mid-probe — awaiting a probe ACK that may never arrive (never, over
        // a mesh carrier) — and the awaiting-ACK filter below would skip
        // exactly the messages this edge exists to re-drive.
        self.cancel_probe_state_for_parked_peer(peer_id);
        // A per-peer reachability edge: the escalating unreachable-probe
        // interval starts over (mirrors rearm_welcome_for_peer resetting
        // unreachable_parks). Snapshotted, not dropped: an edge whose
        // re-drives all fail didn't actually take, and restores it below.
        let prior_parks = self.dm_unreachable_parks.remove(peer_id);

        // Collect matching messages from outbox + media_outbox
        let mut to_send: Vec<(Message, u32)> = Vec::new();

        for entry in self.outbox.values() {
            if entry.message.recipient.as_str() == peer_id
                && !self.ack_manager.is_waiting_for_ack(&entry.message.id)
            {
                to_send.push((entry.message.clone(), entry.attempt_count));
            }
        }
        for entry in self.media_outbox.values() {
            if entry.message.recipient.as_str() == peer_id
                && !self.ack_manager.is_waiting_for_ack(&entry.message.id)
            {
                to_send.push((entry.message.clone(), entry.attempt_count));
            }
        }

        if to_send.is_empty() {
            return;
        }

        let count = to_send.len().min(crate::constants::FLUSH_BATCH_LIMIT);
        debug!(peer_id = %peer_id, count = count, "Flushing outbox for discovered peer");

        let mut iter = to_send.into_iter();
        let mut parked_dm_seen = false;
        let mut parked_dm_redriven = false;

        // Only process up to FLUSH_BATCH_LIMIT messages per edge.
        for (message, attempt_count) in iter.by_ref().take(crate::constants::FLUSH_BATCH_LIMIT) {
            let parkable = self.is_parkable_plain_dm(&message.id);
            // Remove from retry queue since we're sending immediately
            self.retry_queue.remove(&message.id.as_str());
            let sent = self.try_flush_send_via(message, attempt_count, unpark_via);
            if parkable {
                parked_dm_seen = true;
                parked_dm_redriven |= sent.is_some();
            }
        }

        // Re-enqueue the overflow (mirrors flush_outbox_all): the probe
        // cancel above may have stripped a parked message's retry entry, so
        // "still holds its backoff timer" cannot be assumed past the batch
        // limit. Enqueue dedupes by id — entries still scheduled keep their
        // existing timer, canceled ones get a fresh backoff slot instead of
        // stranding edge-only with the park counter cleared.
        for (message, attempt_count) in iter {
            parked_dm_seen = parked_dm_seen || self.is_parkable_plain_dm(&message.id);
            let _ = self.retry_queue.enqueue(message, attempt_count);
        }

        // An edge that re-drove no parkable DM at all didn't actually take:
        // the DMs now sit in the retry queue, whose later sends are
        // DORS-routed (no transport override there), and a send that succeeds
        // locally without reaching the peer would re-register an ACK nothing
        // will answer. Restore the counter so exhaustion stays within
        // `try_repark_exhausted_dm`'s reach instead of settling terminally
        // with the counter cleared.
        //
        // Restore granularity is per-peer, matching the counter: one
        // successful re-drive clears it even when a sibling's send failed on
        // the same edge, leaving that sibling on DORS-routed backoff exposed
        // to terminal exhaustion. Accepted gap — it requires one transport
        // to both succeed and fail within a single flush loop.
        if let Some(parks) = prior_parks {
            if parked_dm_seen && !parked_dm_redriven {
                self.dm_unreachable_parks.insert(peer_id.to_string(), parks);
            }
        }
    }

    /// Immediately attempts to send all pending outbox messages across all peers.
    ///
    /// Called when a transport becomes available (e.g. internet reconnects) to
    /// flush all queued messages, bypassing backoff timers.
    pub fn flush_outbox_all(&mut self) {
        // Acknowledgements we owe first: they are what stop peers re-sending
        // to us, and they have no outbox entry for the walk below to find.
        self.flush_unrouted_acks();
        // Probe ACKs first (see `flush_outbox_for_peer_via`): mid-probe DMs
        // are awaiting unanswerable mesh ACKs, and the awaiting-ACK guard
        // below would strand them past this edge with their counters
        // cleared. Canceled entries surface through the stranded-outbox
        // collection.
        let parked_peers: Vec<String> = self.dm_unreachable_parks.keys().cloned().collect();
        for peer_id in parked_peers {
            self.cancel_probe_state_for_parked_peer(&peer_id);
        }
        // A carrier-level reachability edge (reconnect / start): every
        // escalating unreachable-probe interval starts over. Snapshotted,
        // not dropped: a peer whose re-drives all fail gets its counter
        // restored below (see `flush_outbox_for_peer_via`).
        let prior_parks = std::mem::take(&mut self.dm_unreachable_parks);

        // Drain all entries from the retry queue (ignores timing)
        let retry_entries = self.retry_queue.drain_all();

        // Also collect outbox entries NOT in the retry queue (stranded after
        // previous max_retries rejection)
        let mut all_messages: Vec<(Message, u32)> = retry_entries
            .into_iter()
            .map(|e| (e.message, e.retry_count))
            .collect();

        // Add outbox entries that weren't in the retry queue (stranded) AND
        // aren't already waiting for an ACK (successfully sent, just awaiting
        // confirmation). Without the ACK check, a prior flush that succeeded
        // would cause these messages to be re-sent unnecessarily.
        let retry_ids: std::collections::HashSet<String> =
            all_messages.iter().map(|(m, _)| m.id.as_str()).collect();

        for entry in self.outbox.values() {
            if !retry_ids.contains(&entry.message.id.as_str())
                && !self.ack_manager.is_waiting_for_ack(&entry.message.id)
            {
                all_messages.push((entry.message.clone(), entry.attempt_count));
            }
        }
        for entry in self.media_outbox.values() {
            if !retry_ids.contains(&entry.message.id.as_str())
                && !self.ack_manager.is_waiting_for_ack(&entry.message.id)
            {
                all_messages.push((entry.message.clone(), entry.attempt_count));
            }
        }

        // Opt-in (`edge_driven_unreachable_dm`): on the start/reconnect re-drive,
        // leave durably-failing messages (first sent longer ago than
        // OUTBOX_DURABLE_UNREACHABLE_AGE_SECS, still undelivered) edge-driven
        // rather than re-driving them, so a restart does not re-ramp a large
        // dead-peer backlog. They rest in the outbox and flush the instant their
        // peer next proves reachable. Default-off preserves the documented
        // "reconnect re-drives, ignoring timing" behaviour for every existing
        // app. drain_all() already removed these from the retry queue and the
        // outbox entry is untouched, so "not collected" == "resting edge-driven".
        if self.config.reliability.retry.edge_driven_unreachable_dm {
            let now = Utc::now();
            let durable_age = chrono::Duration::seconds(
                crate::protocol::types::OUTBOX_DURABLE_UNREACHABLE_AGE_SECS,
            );
            all_messages.retain(|(m, _)| {
                match self
                    .outbox
                    .get(&m.id)
                    .or_else(|| self.media_outbox.get(&m.id))
                    .map(|e| e.first_sent_at)
                {
                    Some(first_sent) => now.signed_duration_since(first_sent) < durable_age,
                    None => true,
                }
            });
        }

        if all_messages.is_empty() {
            return;
        }

        let count = all_messages.len().min(crate::constants::FLUSH_BATCH_LIMIT);
        debug!(
            count = count,
            total = all_messages.len(),
            "Flushing all outbox messages"
        );

        let mut iter = all_messages.into_iter();
        let mut parked_dm_seen: std::collections::HashSet<String> =
            std::collections::HashSet::new();
        let mut parked_dm_redriven: std::collections::HashSet<String> =
            std::collections::HashSet::new();

        for (message, attempt_count) in iter.by_ref().take(crate::constants::FLUSH_BATCH_LIMIT) {
            let recipient = message.recipient.as_str().to_string();
            let parkable =
                prior_parks.contains_key(&recipient) && self.is_parkable_plain_dm(&message.id);
            let sent_via = self.try_flush_send(message, attempt_count);
            if parkable {
                // Unlike the per-peer edges (discovery carries mesh-scoped
                // proof, presence-online forces the answering carrier), this
                // carrier-level edge re-drives via DORS with no per-peer
                // reachability proof: a mesh-routed send "locally succeeds"
                // into the mesh whether or not the peer is there, so only a
                // send over a carrier that answers verdicts counts as
                // re-driven for counter-clearing — that is the send whose
                // failure would say something. A mesh-routed success keeps
                // the counter restore-eligible; if its ACK genuinely
                // arrives, delivery prunes the counter.
                if sent_via.and_then(GatewayCarrier::from_transport).is_some() {
                    parked_dm_redriven.insert(recipient);
                } else {
                    parked_dm_seen.insert(recipient);
                }
            }
        }

        // Re-enqueue any messages beyond the batch limit so they aren't lost
        for (message, attempt_count) in iter {
            let recipient = message.recipient.as_str();
            if prior_parks.contains_key(recipient) && self.is_parkable_plain_dm(&message.id) {
                parked_dm_seen.insert(recipient.to_string());
            }
            let _ = self.retry_queue.enqueue(message, attempt_count);
        }

        // Restore the counter of every parked peer none of whose DMs were
        // re-driven on this edge (see `flush_outbox_for_peer_via`): their
        // retry-queue sends are DORS-routed, and a send that succeeds locally
        // without reaching the peer must leave exhaustion re-parkable rather
        // than terminal.
        for (peer_id, parks) in prior_parks {
            if parked_dm_seen.contains(&peer_id) && !parked_dm_redriven.contains(&peer_id) {
                self.dm_unreachable_parks.insert(peer_id, parks);
            }
        }
    }

    /// Records what the attached gateway daemon says it can do.
    ///
    /// Wholesale replace, like the relay's: each attach describes the gateway
    /// actually connected now. The bridge calls this **before** it reports the
    /// carrier available, which is the contract's own ordering — a device must
    /// not drain into a gateway whose features it has not been told — and it
    /// clears the set when the carrier drops.
    ///
    /// The list is wire-supplied, so it is bounded exactly as the relay's is,
    /// which is what the contract points at rather than inventing a second
    /// pair of numbers: at most 64 tokens of at most 128 bytes each, the same
    /// `MAX_RELAY_CAPABILITIES` and `MAX_RELAY_CAPABILITY_TOKEN_BYTES` the
    /// relay's advertisement is bounded by. Oversized tokens are
    /// dropped *before* the count is applied, so padding cannot evict real
    /// ones.
    ///
    /// Nothing here gates a security decision. Tokens gate features, and an
    /// attacker who can set them is already the gateway.
    pub fn set_gateway_capabilities(&mut self, capabilities: Vec<String>) {
        let caps: HashSet<String> = capabilities
            .into_iter()
            .filter(|c| !c.is_empty() && c.len() <= MAX_RELAY_CAPABILITY_TOKEN_BYTES)
            .take(MAX_RELAY_CAPABILITIES)
            .collect();
        debug!(count = caps.len(), "Gateway capabilities updated");
        self.gateway_capabilities = caps;
    }

    /// Forgets the attached gateway's advertisement, on carrier drop.
    pub fn clear_gateway_capabilities(&mut self) {
        if !self.gateway_capabilities.is_empty() {
            debug!("Gateway capabilities cleared on carrier drop");
            self.gateway_capabilities.clear();
        }
    }

    /// The attached gateway's advertised capability tokens, sorted so the
    /// order a caller sees does not depend on hashing.
    pub fn gateway_capabilities(&self) -> Vec<String> {
        let mut tokens: Vec<String> = self.gateway_capabilities.iter().cloned().collect();
        tokens.sort();
        tokens
    }

    /// Peers the platform layer should watch via a gateway `CheckPresence`
    /// query: every peer with an undelivered or session-unproven MLS welcome
    /// ([`Self::welcome_pending_peers`]) plus every recipient of an outbox
    /// message not currently awaiting an ACK — which includes DMs parked on a
    /// `recipient_unreachable` verdict. A gateway's presence-online answer
    /// lands in [`Self::on_peer_presence_via`], whose
    /// `flush_outbox_for_peer_via` is what re-drives (un-parks) those
    /// messages, so the SDK — not the app — owns the "watch my DeliveryError
    /// recipients" duty. Bounded by the outbox cap (500 entries).
    ///
    /// Carrier-agnostic on purpose: the set is "who this device is waiting to
    /// hear about", which does not depend on who is asked. Every attached
    /// gateway watches the same peers.
    pub fn presence_watch_peers(&self) -> Vec<String> {
        let mut peers = self.welcome_pending_peers();
        let mut seen: std::collections::HashSet<String> = peers.iter().cloned().collect();
        for entry in self.outbox.values() {
            if self.ack_manager.is_waiting_for_ack(&entry.message.id) {
                continue;
            }
            let recipient = entry.message.recipient.as_str();
            if seen.insert(recipient.to_string()) {
                peers.push(recipient.to_string());
            }
        }
        peers
    }

    /// Attempts to send a single message as part of a flush operation.
    ///
    /// Ensures the outbox entry exists before sending. A message reaching here
    /// normally has one: every path that gives a message up retires its retry
    /// entry with it (`retire_undeliverable_message`), so an eviction no longer
    /// leaves one behind to resurrect here, and re-creating a settled message
    /// would evict a live one at capacity. On success,
    /// registers ACK tracking and updates the outbox entry. On failure,
    /// re-enqueues the message to the retry queue with its current attempt count
    /// so backoff resumes. Returns the transport the send went out over, or
    /// `None` when it failed (or, defensively, when the transport could not
    /// be attributed — the park restore logic treats both as not-redriven,
    /// the safe direction; in practice a successful send always has one).
    fn try_flush_send(&mut self, message: Message, attempt_count: u32) -> Option<TransportType> {
        self.try_flush_send_via(message, attempt_count, None)
    }

    /// [`Self::try_flush_send`] with an optional transport override (see
    /// [`Self::flush_outbox_for_peer_via`]). A pinned media transport takes
    /// precedence over the override; without either, DORS picks.
    fn try_flush_send_via(
        &mut self,
        mut message: Message,
        attempt_count: u32,
        transport_override: Option<TransportType>,
    ) -> Option<TransportType> {
        self.ensure_outbox_entry(&message);
        // Tier 2: re-seal against the peer's current session before flushing, so
        // a post-re-key epoch change no longer leaves the bytes undecryptable.
        // No-op for media/plaintext/unconfirmed sessions.
        self.reseal_for_resend_in_place(&mut message);
        let forced_transport = self
            .pinned_media_transport_for_message(&message.id)
            .or(transport_override);
        let send_result = if let Some(transport) = forced_transport {
            self.transport_manager
                .send_via_transport(&message, transport)
        } else {
            self.transport_manager.send(&message)
        };
        let current_transport =
            forced_transport.or_else(|| self.transport_manager.current_transport());

        match send_result {
            Ok(()) => {
                if let Err(e) = self.ensure_ack_registration(&message) {
                    warn!(message_id = %message.id, error = %e, "ACK registration failed during flush");
                }
                self.mark_message_sent(
                    &message,
                    current_transport,
                    Some(attempt_count.saturating_add(1)),
                );
                self.offer_resend_to_mesh(&message, true);
                debug!(message_id = %message.id, "Flush send succeeded");
                current_transport
            }
            Err(e) => {
                self.offer_resend_to_mesh(&message, false);
                let _ = self.retry_queue.enqueue(message.clone(), attempt_count);
                debug!(message_id = %message.id, error = %e, "Flush send failed, re-enqueued");
                None
            }
        }
    }

    /// Called when a neighbor is lost.
    ///
    /// Cleans up tracking state for the lost peer.
    ///
    /// # Arguments
    ///
    /// * `peer_id` - The ID of the lost peer
    pub fn on_neighbor_lost(&mut self, peer_id: &str) {
        self.evict_known_peer(peer_id);
    }

    /// Drops a peer from discovery tracking. Shared by explicit neighbor
    /// loss, the TTL sweep, and least-recently-seen eviction at capacity —
    /// all three mean "treat this peer as gone until re-seen", so all three
    /// also clear the key-package marker, letting a re-appearing peer
    /// receive a fresh key package exactly as on a BLE reconnect.
    fn evict_known_peer(&mut self, peer_id: &str) {
        self.key_package_sent_to.remove(peer_id);
        self.known_peers.remove(peer_id);
        // Release the forwarding allowance held for this link. A peer that
        // reconnects starts with a full allowance, which is correct: the limit
        // exists to pace a live link, not to punish one that came back.
        self.mesh_relay.forget_peer(peer_id);
    }

    /// Tears down the 1:1 session with `peer_id` and rebuilds it, advancing
    /// post-compromise security.
    ///
    /// # Why an application has to ask
    ///
    /// Post-compromise security arrives when a commit rotates a member's leaf
    /// in the ratchet tree. This engine originates one on a re-key, and until
    /// now the only thing that drove a re-key was an **epoch desync**: a
    /// healthy pair that never forked never rotated, so the window an attacker
    /// holding old key material keeps open was bounded by nothing.
    ///
    /// That is felt hardest against a leaf node, which never commits at all, so
    /// every rotation in such a pair is the phone's to originate. It is not
    /// exclusive to one: two phones that simply never desync are in the same
    /// position.
    ///
    /// The cadence is the application's because this engine cannot pick one.
    /// A rotation costs a teardown, a key-package exchange and a re-establish,
    /// and what that is worth depends on the deployment: a door lock on mains
    /// power and a phone on a metered link do not want the same interval, and
    /// nothing on the wire says which one is on the other end.
    ///
    /// # What the peer sees
    ///
    /// Exactly what a desync-driven re-key sends, because it is the same code:
    /// a `__MLS_KEY_PKG__` frame carrying `session_reset`. A peer running this
    /// engine discards its stale session and converges in one round trip; a
    /// leaf node discards its session, mints a fresh key package and re-pairs.
    /// Queued plaintext survives, because it is sealed at flush time against
    /// whatever session is current then.
    ///
    /// # Rate limit
    ///
    /// Shared with the desync path, one re-key per peer per window. A caller
    /// looping on this cannot do to a pair what that bound exists to stop an
    /// attacker doing.
    ///
    /// # Returns
    ///
    /// * `Ok(true)` - the reset was advertised and the session torn down.
    /// * `Ok(false)` - the rate-limit window has not lapsed; nothing was done,
    ///   and a later call succeeds. Not an error: a caller on a fixed schedule
    ///   that briefly runs faster than the floor is behaving correctly.
    ///
    /// # A rotation that fails changes nothing
    ///
    /// The reset is advertised **before** the local session is torn down, so
    /// an `Err` leaves the pair exactly as it was: the session is still there,
    /// still usable, and the rate-limit window is not spent. Rotate while the
    /// peer is reachable, and treat a failure as "try again later" rather than
    /// as a session to rebuild.
    ///
    /// The order matters most against a leaf node, which speaks only when
    /// spoken to. Tearing down first and failing to send would leave this side
    /// with no session, the device holding one it will never be told to drop,
    /// and nothing on either side that re-opens the exchange.
    ///
    /// # Errors
    ///
    /// * [`Error::MlsNotInitialized`] - encryption is not set up.
    /// * [`Error::UserBlocked`] - the peer is blocked.
    /// * [`Error::Mls`] with `SessionNotFound` - there is no session to rebuild.
    ///   Establishing a first session is [`Self::establish_secure_session`];
    ///   this heals one that exists.
    /// * [`Error::Transport`] - no transport accepted the reset frame. The
    ///   session is untouched, as above.
    pub fn rekey_session(&mut self, peer_id: &str) -> Result<bool> {
        if self.is_user_blocked(peer_id) {
            return Err(Error::UserBlocked(peer_id.to_string()));
        }
        if self.mls_manager.is_none() {
            return Err(Error::MlsNotInitialized);
        }
        // A session has to exist for there to be one to heal. Checked before
        // the slot is claimed, so a refused call does not spend the window and
        // leave a real re-key waiting on it.
        if !self.has_mls_session(peer_id)? {
            return Err(Error::Mls(offline_protocol_mls::MlsError::SessionNotFound(
                peer_id.to_string(),
            )));
        }
        if !self.claim_rekey_slot(peer_id) {
            debug!(
                peer_id = %peer_id,
                "rekey_session: the rate-limit window has not lapsed"
            );
            return Ok(false);
        }
        if let Err(err) = self.drive_session_rekey(peer_id, "requested by the application") {
            // Nothing was sent and nothing was torn down, so nothing was
            // spent: a caller back in range a second from now is asking for a
            // rotation this pair has not had.
            self.release_rekey_slot(peer_id);
            return Err(err);
        }
        Ok(true)
    }

    /// Establishes a secure MLS session with a peer.
    ///
    /// This high-level method handles the complete session establishment flow:
    /// 1. Checks if a session already exists (returns Ok(None) if so)
    /// 2. Checks for a pending key package from the peer
    /// 3. If found, imports the key package, creates the session, and sends the Welcome message
    /// 4. If no key package is available, returns an error
    ///
    /// This method is designed for use by application code that needs explicit control
    /// over session establishment, as opposed to the automatic encryption flow.
    ///
    /// # Arguments
    ///
    /// * `peer_id` - The ID of the peer to establish a session with
    ///
    /// # Returns
    ///
    /// * `Ok(Some(WelcomeMessage))` - Session created, Welcome message returned (and sent to peer)
    /// * `Ok(None)` - Session already exists
    /// * `Err(SessionNotReady(state))` - Establishment in progress; caller can retry or show "Establishing…"
    pub fn establish_secure_session(&mut self, peer_id: &str) -> Result<Option<WelcomeMessage>> {
        if self.is_user_blocked(peer_id) {
            return Err(Error::UserBlocked(peer_id.to_string()));
        }

        let mls = self.mls_manager.clone().ok_or(Error::MlsNotInitialized)?;

        // Check if session already exists
        let has_session = {
            let manager = mls
                .read()
                .map_err(|_| Error::Other("MLS lock poisoned".to_string()))?;
            manager.has_session(peer_id)?
        };

        if has_session {
            debug!(peer_id = %peer_id, "Session already exists");
            return Ok(None);
        }

        // Try loading key package from storage (e.g. after restart) before giving up
        self.try_load_key_package_from_storage_into_memory(peer_id);

        // Check for pending key package (memory, possibly just restored from storage)
        // Clone first, only remove after all operations succeed to avoid losing the key package on failure
        if let Some(received_pkg) = self.pending_key_packages.get(peer_id).cloned() {
            // Check if key package has expired (using local clock)
            let now_ms = Utc::now().timestamp_millis() as u64;
            if now_ms >= received_pkg.local_expires_at_ms {
                warn!(peer_id = %peer_id, "Received key package has expired, discarding");
                self.pending_key_packages.remove(peer_id);
                self.delete_peer_key_package_from_storage(peer_id);
            } else {
                self.import_pending_key_package(&mls, peer_id, &received_pkg.key_package_data)?;

                // Create session and get welcome message
                let welcome = {
                    let manager = mls
                        .read()
                        .map_err(|_| Error::Other("MLS lock poisoned".to_string()))?;
                    manager.create_session(peer_id)?
                };
                // The handshake starts here. This is the auto-establish path
                // taken the moment a peer's key package arrives, which is how
                // an ordinary session begins and which raises no
                // `session_missing` at all; see `emit_mls_session_establishing`.
                self.emit_mls_session_establishing(
                    peer_id,
                    welcome.group_id.as_str(),
                    crate::mls_observability::MlsOperationContext::Welcome,
                );
                // We only ever build a session from a key package the peer
                // published, so reaching here proves they run MLS.
                self.mark_encryption_capable(peer_id);

                // Send welcome message to peer
                let welcome_sent = self.send_welcome_message(peer_id, &welcome, false)?;

                // All operations succeeded, now safe to remove the key package
                self.pending_key_packages.remove(peer_id);
                self.delete_peer_key_package_from_storage(peer_id);

                let group_id = welcome.group_id.as_str().to_string();
                let is_session = group_id.starts_with("session:");
                if let Err(err) = self.ensure_session_state_entry(peer_id, "session_created_local")
                {
                    warn!(
                        peer_id = %peer_id,
                        error = %err,
                        "Failed to persist pending session state"
                    );
                }

                info!(peer_id = %peer_id, group_id = %group_id, "Established secure session");

                // Send a fresh key package so the peer has one available for
                // group invites (the original was consumed to create this session).
                if self.config.encryption.enabled {
                    let _ = self.send_key_package_to(peer_id, false);
                }

                if welcome_sent {
                    debug!(peer_id = %peer_id, group_id = %group_id, is_session, "Welcome synchronously sent");
                }

                return Ok(Some(welcome));
            }
        }

        // No key package available (memory nor storage) — return non-terminal state so caller can retry
        Err(Error::SessionNotReady(self.establishment_state(peer_id)?))
    }

    /// Imports `peer_id`'s pending key package into MLS, reporting a refusal
    /// by this device's clock.
    ///
    /// The one place the engine admits a pending package, so that the refusal
    /// is reported whichever route reached it: `establish_secure_session`,
    /// taken when a package arrives, and the send path's
    /// `ensure_session_establishment`, taken on the first send after a
    /// restart or with `auto_key_exchange` off. Reported on only the first,
    /// the restart case left a message queued "anyway" with nothing to say why,
    /// which is the failure the code exists for. The throttle makes a second
    /// route free: one event per peer either way.
    ///
    /// A window that has not started keeps the package, since it becomes valid
    /// once the clocks agree. A window that has closed discards it, exactly as
    /// an expired cached package is discarded, because it may be a package a
    /// relay held for days rather than a clock, and kept it would fail every
    /// attempt until its cached expiry. The next package the peer sends
    /// replaces it.
    pub(super) fn import_pending_key_package(
        &mut self,
        mls: &Arc<RwLock<MlsManager>>,
        peer_id: &str,
        key_package_data: &[u8],
    ) -> Result<()> {
        let imported = {
            let manager = mls
                .read()
                .map_err(|_| Error::Other("MLS lock poisoned".to_string()))?;
            manager.import_key_package(peer_id, key_package_data)
        };
        if let Err(offline_protocol_mls::MlsError::KeyPackageOutsideValidityWindow { expired }) =
            &imported
        {
            let reason = if *expired {
                warn!(
                    peer_id = %peer_id,
                    "Peer key package's validity window has closed by this device's \
                     clock; discarding it until the peer sends a fresh one"
                );
                self.pending_key_packages.remove(peer_id);
                self.delete_peer_key_package_from_storage(peer_id);
                "The peer's key package has expired by this device's clock. Either \
                 this device's date is ahead, or the package is older than its 30-day \
                 window; it is discarded and the next one the peer sends replaces it."
            } else {
                warn!(
                    peer_id = %peer_id,
                    "Peer key package's validity window has not started by this \
                     device's clock; this device cannot start a session with the peer \
                     until the clocks agree"
                );
                "The peer's key package is not valid yet at this device's time. \
                 Check this device's date and time, then the peer's."
            };
            self.warn_control_gate_rejection(
                peer_id,
                crate::events::SecurityWarningCode::KeyPackageOutsideValidityWindow,
                reason,
            );
        }
        Ok(imported?)
    }

    /// Checks if a pending key package is available for a peer.
    ///
    /// This can be used to check if session establishment is possible
    /// before calling `establish_secure_session`.
    ///
    /// # Arguments
    ///
    /// * `peer_id` - The ID of the peer to check
    ///
    /// # Returns
    ///
    /// `true` if a key package is available, `false` otherwise
    pub fn has_pending_key_package(&self, peer_id: &str) -> bool {
        self.pending_key_packages.contains_key(peer_id)
    }

    /// Returns the current establishment state for a peer.
    ///
    /// Use this to show "Establishing…" or drive retries without calling send/establish.
    pub fn get_establishment_state(&self, peer_id: &str) -> Result<EstablishmentState> {
        self.establishment_state(peer_id)
    }

    /// Whether rich extras (reply context, rich media metadata, forward
    /// attribution) on a send toward `peer_id` would travel in the sealed
    /// rich body — i.e. the peer's last key package advertised the
    /// capability (surviving restarts) and our own `rich_payload_enabled`
    /// kill switch is on. When `false` the extras are silently dropped and
    /// the message degrades to plain text, so apps can use this to render
    /// honest degraded-reply UX instead of discovering the drop after the
    /// fact.
    pub fn peer_supports_rich_payload(&self, peer_id: &str) -> bool {
        self.rich_seal_active(peer_id)
    }

    /// Creates a session using manually imported key material.
    ///
    /// This entrypoint is for bindings that expose low-level MLS APIs. It must
    /// still keep protocol-level session lifecycle state in sync with MLS state.
    pub fn manual_mls_create_session(&mut self, peer_id: &str) -> Result<WelcomeMessage> {
        let mls = self.mls_manager.clone().ok_or(Error::MlsNotInitialized)?;
        let welcome = {
            let manager = mls
                .read()
                .map_err(|_| Error::Other("MLS lock poisoned".to_string()))?;
            manager.create_session(peer_id)?
        };
        self.mark_encryption_capable(peer_id);
        if let Err(err) = self.ensure_session_state_entry(peer_id, "manual_session_created") {
            warn!(
                peer_id = %peer_id,
                error = %err,
                "Failed to persist pending session state after manual session create"
            );
        }
        Ok(welcome)
    }

    /// Imports a contact's key package and records them encryption-capable.
    ///
    /// The entry point for bindings that expose low-level MLS APIs. It exists
    /// because the FFI wrapper used to reach straight for the `MlsManager` and
    /// pass the since-deleted `KeyPackageTrust::FirstUse` — so an app could
    /// import a key package
    /// under any peer id with no correlation to that peer, bypassing the check
    /// `establish_secure_session` performs. The identity claim no longer needs
    /// the detour: `import_key_package` re-derives the address from the leaf
    /// signature key and refuses a mismatch on its own. The routing stays
    /// because the capability record lives here, not on the manager.
    pub fn manual_mls_import_key_package(
        &mut self,
        peer_id: &str,
        key_package_data: &[u8],
    ) -> Result<()> {
        let mls = self.mls_manager.clone().ok_or(Error::MlsNotInitialized)?;
        {
            let manager = mls
                .read()
                .map_err(|_| Error::Other("MLS lock poisoned".to_string()))?;
            manager.import_key_package(peer_id, key_package_data)?;
        }
        // A key package we accepted for this peer is proof they run MLS.
        self.mark_encryption_capable(peer_id);
        Ok(())
    }

    /// Deletes a 1:1 session and clears protocol-level lifecycle state.
    pub fn manual_mls_delete_session(&mut self, peer_id: &str) -> Result<()> {
        let mls = self.mls_manager.clone().ok_or(Error::MlsNotInitialized)?;
        let manager = mls
            .read()
            .map_err(|_| Error::Other("MLS lock poisoned".to_string()))?;
        manager.delete_session(peer_id)?;

        // Apply protocol-state cleanup only after MLS deletion succeeds.
        self.confirmed_sessions.remove(peer_id);
        self.clear_confirmation_recovery_tracking(peer_id);
        self.welcome_lifecycles.remove(peer_id);
        // Drop the both-create owner gate too (memory + storage). A leaked gate
        // entry would make `can_confirm_from_source` reject every non-decrypt
        // confirmation source — including `welcome_received` — on the NEXT
        // session with this peer, re-stranding it in Pending (and surviving
        // restart via `restore_both_create_awaiting_decrypt`).
        self.clear_both_create_awaiting_decrypt(peer_id);
        self.clear_session_state_entry(peer_id)?;
        self.clear_welcome_lifecycle_entry(peer_id)?;
        Ok(())
    }

    /// Joins a session from a Welcome message and synchronizes confirmation state.
    pub fn manual_mls_join_session(
        &mut self,
        welcome: &WelcomeMessage,
    ) -> Result<offline_protocol_mls::GroupInfo> {
        let mls = self.mls_manager.clone().ok_or(Error::MlsNotInitialized)?;
        let group_info = {
            let manager = mls
                .read()
                .map_err(|_| Error::Other("MLS lock poisoned".to_string()))?;
            manager.join_session(welcome)?
        };
        self.handle_manual_welcome_confirmation(&welcome.inviter_id);
        Ok(group_info)
    }

    /// Processes an MLS Welcome message and synchronizes confirmation state.
    pub fn manual_mls_process_welcome(
        &mut self,
        welcome: &WelcomeMessage,
    ) -> Result<offline_protocol_mls::GroupInfo> {
        let mls = self.mls_manager.clone().ok_or(Error::MlsNotInitialized)?;
        let group_info = {
            let manager = mls
                .read()
                .map_err(|_| Error::Other("MLS lock poisoned".to_string()))?;
            manager.process_welcome(welcome)?
        };
        self.handle_manual_welcome_confirmation(&welcome.inviter_id);
        Ok(group_info)
    }

    /// Decrypts a user-scoped MLS message and synchronizes confirmation state.
    pub fn manual_mls_decrypt_from_user(
        &mut self,
        encrypted: &EncryptedMessage,
    ) -> Result<Option<Vec<u8>>> {
        let mls = self.mls_manager.clone().ok_or(Error::MlsNotInitialized)?;
        let plaintext = {
            let manager = mls
                .read()
                .map_err(|_| Error::Other("MLS lock poisoned".to_string()))?;
            manager.decrypt_from_user(encrypted, &encrypted.sender_id)?
        };
        if plaintext.is_some() {
            self.handle_manual_decrypt_confirmation(&encrypted.sender_id);
        }
        Ok(plaintext)
    }

    /// Decrypts any MLS message and synchronizes confirmation state for 1:1 flows.
    pub fn manual_mls_decrypt(&mut self, encrypted: &EncryptedMessage) -> Result<Option<Vec<u8>>> {
        let mls = self.mls_manager.clone().ok_or(Error::MlsNotInitialized)?;
        let plaintext = {
            let manager = mls
                .read()
                .map_err(|_| Error::Other("MLS lock poisoned".to_string()))?;
            manager.decrypt(encrypted, &encrypted.sender_id)?
        };
        if plaintext.is_some() && Self::is_session_group_id(encrypted.group_id.as_str()) {
            self.handle_manual_decrypt_confirmation(&encrypted.sender_id);
        }
        Ok(plaintext)
    }

    pub(crate) fn is_session_group_id(group_id: &str) -> bool {
        group_id.starts_with("session:")
    }

    fn handle_manual_welcome_confirmation(&mut self, peer_id: &str) {
        // Shared by `manual_mls_join_session` and `manual_mls_process_welcome`:
        // in both cases we just joined a session this peer built for us.
        self.mark_encryption_capable(peer_id);
        match self.confirm_session_state(peer_id, "welcome_received") {
            Ok(true) => {
                let _ = self.flush_pending_messages(peer_id);
                self.process_pending_decryption(peer_id);
            }
            Ok(false) => {}
            Err(err) => {
                warn!(
                    peer_id = %peer_id,
                    error = %err,
                    "Failed to persist session confirmation after manual welcome processing"
                );
            }
        }
    }

    fn handle_manual_decrypt_confirmation(&mut self, peer_id: &str) {
        if !self.can_confirm_from_source(peer_id, "decrypt_success") {
            return;
        }
        match self.confirm_session_state(peer_id, "decrypt_success") {
            Ok(true) => {
                let _ = self.flush_pending_messages(peer_id);
            }
            Ok(false) => {}
            Err(err) => {
                warn!(
                    peer_id = %peer_id,
                    error = %err,
                    "Failed to persist session confirmation after manual decrypt"
                );
            }
        }
    }

    /// Processes internal MLS protocol messages.
    ///
    /// Returns `Some(InternalMessageResult::Consumed)` if the message was an internal
    /// protocol message that should not be surfaced to the application.
    /// Returns `Some(InternalMessageResult::Decrypted(plaintext))` if the message was
    /// encrypted and successfully decrypted.
    /// Returns `None` if the message is not an internal message.
    pub(crate) fn process_internal_message(
        &mut self,
        message: &Message,
    ) -> Option<InternalMessageResult> {
        self.process_internal_message_via(message, None)
    }

    /// [`Self::process_internal_message`] with the transport the frame
    /// arrived on, when the caller knows it. Handlers that trust relay-server
    /// answers (`__GROUP_CREATED__` setting `relay_synced`) require the
    /// Internet transport; with `None` they treat the frame as untrusted.
    pub(crate) fn process_internal_message_via(
        &mut self,
        message: &Message,
        arrival_transport: Option<TransportType>,
    ) -> Option<InternalMessageResult> {
        let content = &message.content;

        // Run the security gate for control messages (transport identity +
        // signature verification + freshness). `Rejected` drops the message;
        // `Proceed` carries whether the frame was actually signed, for the one
        // handler that consumes a payload field as authenticated data, and
        // whether its age was established, for the one that acts on a
        // directive worth replaying.
        let (signed, freshness_bound) =
            match self.security_gate_control_message(message, arrival_transport) {
                ControlGateOutcome::Rejected(result) => return Some(result),
                ControlGateOutcome::Proceed {
                    signed,
                    freshness_bound,
                } => (signed, freshness_bound),
            };

        let sender = message.sender.as_str();

        // Handle key package messages
        if let Some(data) = content.strip_prefix(internal_prefixes::KEY_PACKAGE) {
            // The stamp travels only when it was signed *and* checked, so a
            // handler cannot reach for it in the case where it means nothing.
            let freshness = freshness_bound.then(|| message.timestamp.as_millis());
            self.handle_key_package_message(sender, data, signed, freshness);
            return Some(InternalMessageResult::Consumed);
        }

        if content.starts_with(internal_prefixes::SESSION_CONFIRM_PROBE) {
            self.handle_session_confirm_probe(sender, content);
            return Some(InternalMessageResult::Consumed);
        }

        if content.starts_with(internal_prefixes::SESSION_CONFIRM_ACK) {
            self.handle_session_confirm_ack(sender, content);
            return Some(InternalMessageResult::Consumed);
        }

        // Handle welcome messages (session invitation)
        if let Some(data) = content.strip_prefix(internal_prefixes::WELCOME) {
            self.handle_welcome_message(sender, data);
            return Some(InternalMessageResult::Consumed);
        }

        // Handle encrypted messages
        if let Some(data) = content.strip_prefix(internal_prefixes::ENCRYPTED) {
            if let Some(result) =
                self.handle_encrypted_message(sender, data, message, arrival_transport)
            {
                return Some(result);
            }
            return Some(InternalMessageResult::Consumed);
        }

        // Handle connection request messages
        if let Some(data) = content.strip_prefix(internal_prefixes::CONN_REQUEST) {
            self.handle_connection_request(sender, data);
            return Some(InternalMessageResult::Consumed);
        }

        // Handle connection accepted messages
        if let Some(data) = content.strip_prefix(internal_prefixes::CONN_ACCEPT) {
            self.handle_connection_accepted(sender, data);
            return Some(InternalMessageResult::Consumed);
        }

        // Handle connection rejected messages
        if content
            .strip_prefix(internal_prefixes::CONN_REJECT)
            .is_some()
        {
            self.handle_connection_rejected(sender);
            return Some(InternalMessageResult::Consumed);
        }

        // Handle connection cancelled messages
        if content
            .strip_prefix(internal_prefixes::CONN_CANCEL)
            .is_some()
        {
            self.handle_connection_cancelled(sender);
            return Some(InternalMessageResult::Consumed);
        }

        // --- Presence, typing, and read receipt messages ---

        if let Some(data) = content.strip_prefix(internal_prefixes::PRESENCE) {
            self.handle_presence_message(sender, data);
            return Some(InternalMessageResult::Consumed);
        }

        if let Some(data) = content.strip_prefix(internal_prefixes::TYPING_INDICATOR) {
            self.handle_typing_indicator(sender, data);
            return Some(InternalMessageResult::Consumed);
        }

        if let Some(data) = content.strip_prefix(internal_prefixes::READ_RECEIPT) {
            self.handle_read_receipt(sender, data);
            return Some(InternalMessageResult::Consumed);
        }

        // A custody receipt from a neighbour holding one of our frames. Past
        // the control gate like every signed frame; it suppresses re-deposit
        // toward that custodian and settles nothing (`docs/spec/custody.md`).
        // Consumed whatever the body says: a malformed one is refused
        // silently, and a receipt never requests an acknowledgement.
        if let Some(data) = content.strip_prefix(internal_prefixes::CUSTODY_RECEIPT) {
            self.handle_custody_receipt(sender, data, message.timestamp.as_millis());
            return Some(InternalMessageResult::Consumed);
        }

        // --- Group (mesh/MLS) messages ---

        if let Some(data) = content.strip_prefix(internal_prefixes::GROUP_MLS_MSG) {
            return Some(self.handle_group_mls_msg_via(message, sender, data, arrival_transport));
        }

        if let Some(data) = content.strip_prefix(internal_prefixes::GROUP_MLS_WELCOME) {
            let mid = message.id.as_str();
            self.handle_group_mls_welcome(&mid, sender, data);
            return Some(InternalMessageResult::Consumed);
        }

        if let Some(data) = content.strip_prefix(internal_prefixes::GROUP_MLS_COMMIT) {
            let mid = message.id.as_str();
            self.handle_group_mls_commit(&mid, sender, data);
            return Some(InternalMessageResult::Consumed);
        }

        if let Some(data) = content.strip_prefix(internal_prefixes::GROUP_MLS_LEAVE) {
            let mid = message.id.as_str();
            self.handle_group_mls_leave(&mid, sender, data);
            return Some(InternalMessageResult::Consumed);
        }

        if let Some(data) = content.strip_prefix(internal_prefixes::GROUP_ROLE_CHANGE) {
            let mid = message.id.as_str();
            self.handle_group_role_change(&mid, sender, data);
            return Some(InternalMessageResult::Consumed);
        }

        if let Some(data) = content.strip_prefix(internal_prefixes::GROUP_RENAME) {
            let mid = message.id.as_str();
            self.handle_group_rename(&mid, sender, data);
            return Some(InternalMessageResult::Consumed);
        }

        // --- Group (relay) and service messages ---

        if content.starts_with(internal_prefixes::GROUP_CREATED)
            || content.starts_with(internal_prefixes::GROUP_MSG)
            || content.starts_with(internal_prefixes::GROUP_MEMBER_ADDED)
            || content.starts_with(internal_prefixes::GROUP_MEMBER_REMOVED)
            || content.starts_with(internal_prefixes::GROUP_INFO)
            || content.starts_with(internal_prefixes::USER_GROUPS)
            || content.starts_with(internal_prefixes::GROUP_ERROR)
        {
            self.handle_group_relay_message(sender, content, arrival_transport);
            return Some(InternalMessageResult::Consumed);
        }

        // --- Service discovery & request/response ---
        if content.starts_with(offline_protocol_services::SVC_MESSAGE_PREFIX) {
            self.handle_service_message(sender, content, message, false);
            return Some(InternalMessageResult::Consumed);
        }

        None // Not an internal message
    }

    /// Registers an event handler.
    ///
    /// # Arguments
    ///
    /// * `handler` - Callback function that will be called for each event
    pub fn on_event<F>(&mut self, handler: F)
    where
        F: Fn(Event) + Send + Sync + 'static,
    {
        let Ok(mut state) = lock_shared_state(&self.shared_state) else {
            error!("Failed to lock shared state in on_event");
            return;
        };
        state.event_handlers.push(Arc::new(handler));
    }

    /// Processes pending operations (retries, timeouts, etc.).
    ///
    /// This should be called periodically to handle background tasks.
    pub fn process(&mut self) -> Result<()> {
        {
            let state = lock_shared_state(&self.shared_state)?;
            if state.state != ProtocolState::Running {
                return Ok(()); // Don't process if not running
            }
        }

        // Transmit forwards whose hold has elapsed. Runs early and before the
        // retry ladders: other people's traffic is time-sensitive in a way our
        // own retries are not — a frame held too long is dropped rather than
        // sent late.
        self.flush_mesh_relays();
        // Held frames past their hold, and receipt suppressions past their
        // end. Throttled inside; hours-long holds need no per-tick walk.
        self.sweep_custody();

        self.process_retry_queue()?;
        self.process_welcome_retry_queue()?;
        self.process_relay_register_retries();
        self.process_relay_broadcast_report_timeouts();
        self.process_timed_out_acks()?;

        // Throttle session reconciliation to avoid expensive storage I/O
        // (list_sessions → Keychain/Keystore) on every tick. Only run when
        // there's pending work AND enough time has elapsed since the last run.
        self.run_throttled_reconciliation("process_tick");

        let _ = self.prune_expired_pending_global_front(Instant::now(), 256);
        // Batched like the Lamport clock: the seen set reaches disk on a
        // change threshold or a time cadence, never per message.
        self.persist_dedup_seen_if_due();
        self.pump_media_transfers();
        #[cfg(feature = "data")]
        self.send_due_data_settle_offers(Instant::now());
        self.refresh_nostr_key_package_slots();
        self.refresh_nostr_discovery_claim();
        // A relay is free never to send end-of-stored-events, and that is the
        // only natural completion signal a resolution has. Without this sweep
        // a quiet relay turns a lookup into a hang rather than an empty answer.
        self.sweep_username_resolutions();
        self.cleanup_expired_entries();
        self.evaluate_relay_role();
        self.tick_telemetry_categories();

        Ok(())
    }

    /// Reports this device's condition to the forwarding governor and announces
    /// any change in whether it is acting as a relay (`RelayPromoted`,
    /// `RelayDemoted`, `RelayDemotedBattery`).
    ///
    /// Runs every tick independently of telemetry: these are app-facing events
    /// delivered through the `EventCallback` channel, which is live whether or
    /// not a telemetry sink is installed. It also keeps the
    /// `DeviceCapabilitySnapshot.relay_role` signal honest, since
    /// `tick_telemetry_categories` reads the standing immediately after.
    ///
    /// **The relay standing is observed, not predicted.** It says this device
    /// has been carrying other people's traffic, not that its battery and
    /// neighbor count suggest it could — those only bias *how much* it carries
    /// (`set_conditions`), continuously and with no threshold to cross. So the
    /// promotion here is a report about the mesh, and an app can trust
    /// `is_relay()` to mean the device is doing the work.
    ///
    /// Two things still transition on policy rather than on observation, and
    /// both are gates rather than predictions: relaying switched off in
    /// configuration, and the battery falling through the forwarding floor.
    /// Each stops forwarding outright, so the standing is surrendered at once
    /// instead of decaying over the activity windows — the window measures
    /// quiet, and this is refusal.
    ///
    /// Unlike the threshold policy this replaces, a device with no battery
    /// reading still participates: forwarding is what is being observed, and
    /// it happens whether or not the host has reported a level.
    fn evaluate_relay_role(&mut self) {
        let (_statuses, available) = self.transport_manager.snapshot_status_and_available();
        let (battery_level, is_charging) =
            device_battery_from_available(self.transport_manager.current_transport(), &available);

        let eager = matches!(self.config.relay.relay_priority, RelayPriority::Always);
        // The bias ramp is measured from the level at which forwarding actually
        // stops, which is the configured minimum raised to the hard critical
        // floor — not the raw configured value, which may be set below it.
        // Passing the unclamped one would start the ramp under the gate and
        // scale a device that is about to stop forwarding altogether.
        let forwarding_floor = self.relay_battery_floor(is_charging);
        self.mesh_relay
            .set_conditions(battery_level, is_charging, eager, forwarding_floor);
        let relay = &self.config.relay;

        // Relaying switched off in configuration.
        if !relay.allow_relay || matches!(relay.relay_priority, RelayPriority::Never) {
            if self.mesh_relay.force_inactive() {
                self.emit_event(Event::relay_demoted("relaying disabled by configuration"));
            }
            return;
        }

        // Battery through the forwarding floor. Same gate `try_relay_message`
        // applies to each frame, evaluated from the reading already in hand.
        if !self.battery_allows_relaying_with(battery_level, is_charging) {
            if self.mesh_relay.force_inactive() {
                self.emit_event(Event::relay_demoted_battery(
                    battery_level.unwrap_or(0),
                    self.relay_battery_floor(is_charging),
                ));
            }
            return;
        }

        match self.mesh_relay.observe_activity(Instant::now()) {
            Some(RelayActivity::Began { .. }) => {
                let connection_count = self.known_peers.len();
                self.emit_event(Event::relay_promoted(connection_count, battery_level));
            }
            Some(RelayActivity::Ceased) => {
                self.emit_event(Event::relay_demoted(
                    "no traffic carried for other devices recently",
                ));
            }
            None => {}
        }
    }

    /// The relay standing this device currently holds, for telemetry.
    fn relay_role(&self) -> RelayRole {
        if self.mesh_relay.is_active_relay() {
            RelayRole::Relay
        } else {
            RelayRole::Regular
        }
    }

    /// Per-tick telemetry work: diff transport statuses, diff device
    /// capability, and emit a `MetricsFrame` when `metrics_cadence` has
    /// elapsed. No-ops unless a sink has been installed.
    ///
    /// Runs at the end of `process()` so reliability/queue state observed
    /// here is the same state the rest of the tick has already advanced.
    /// Snapshots `get_available_transports()` exactly once per tick and
    /// reuses the map across every downstream helper that needs it.
    ///
    /// If an earlier step in `process()` returns `Err(...)`, this method is
    /// skipped and telemetry emission for that interval is deferred to the
    /// next successful tick — the cadence guarantee relaxes under sustained
    /// partial failure.
    ///
    /// # Sink-panic semantics
    ///
    /// Each per-tick cursor is advanced *before* the emit call that
    /// observes it, so a panicking sink does not cause re-delivery of the
    /// record it panicked on. The transport-status path advances
    /// **per-transport**: when multiple transitions fall in a single tick,
    /// a panic on transition K commits K's entry (at-most-once for K) but
    /// leaves entries L..N in `transport_status_snapshot` at their previous
    /// values, so the next tick re-diffs and emits them fresh. This avoids
    /// silent data loss for transitions that never reached the sink.
    ///
    /// The device-capability and metrics-frame paths emit at most one
    /// record per tick each, so a simpler "advance then emit" discipline
    /// suffices there: a panic advances the cursor and the same record is
    /// not re-emitted on the next tick.
    fn tick_telemetry_categories(&mut self) {
        let Some(ctx) = self.telemetry.clone().filter(|ctx| ctx.enabled()) else {
            return;
        };
        let now_ms = Utc::now().timestamp_millis();
        let now = Instant::now();

        // One lock-per-transport pass: get statuses (for the transition
        // diff) and the available-only metrics map (for the metrics frame
        // and the device-capability diff). Reused across every helper
        // below.
        let (current_statuses, available) = self.transport_manager.snapshot_status_and_available();

        // Transport-status diff: one record per changed transport. Commit
        // each transport's new snapshot entry *before* emitting its event
        // so a panic on event K is at-most-once for K; untouched entries
        // L..N stay at their previous values and get re-diffed next tick.
        let transitions =
            diff_transport_state(now_ms, &self.transport_status_snapshot, &current_statuses);
        for event in transitions {
            let transport = event.transport;
            match current_statuses.get(&transport) {
                Some(status) => {
                    self.transport_status_snapshot.insert(transport, *status);
                }
                None => {
                    self.transport_status_snapshot.remove(&transport);
                }
            }
            dispatch_record(&ctx.sink, &TelemetryRecord::TransportState(event));
            // An internet-capable transport coming up is the pipe's cue to
            // send what an offline stretch queued; the pipe decides which
            // transports count.
            #[cfg(feature = "telemetry-pipe")]
            if let Some(pipe) = &self.telemetry_pipe {
                pipe.notify_transport(event.transport, event.current);
            }
        }

        // Device capability diff. At-most-one emission per tick, so the
        // simple advance-before-emit pattern preserves at-most-once.
        let (battery_level, is_charging) =
            device_battery_from_available(self.transport_manager.current_transport(), &available);
        let relay_role = self.relay_role();
        let device_now = DeviceSnap::from_parts(battery_level, is_charging, relay_role);
        let device_change =
            diff_device_capability(now_ms, self.device_capability_snapshot, device_now);
        self.device_capability_snapshot = Some(device_now);
        if let Some(snapshot) = device_change {
            dispatch_record(&ctx.sink, &TelemetryRecord::Device(snapshot));
        }

        // Periodic metrics snapshot. Rearm cadence before emit so a
        // panicking sink doesn't cause an immediate retry next tick.
        if let Some(cadence) = ctx.config.metrics_cadence() {
            let due = match self.last_metrics_emit_at {
                None => true,
                Some(prev) => now.saturating_duration_since(prev) >= cadence,
            };
            if due {
                let is_local_relay = matches!(relay_role, RelayRole::Relay);
                let frame = build_metrics_frame(
                    now_ms,
                    self.transport_manager.current_transport(),
                    &available,
                    &self.retry_queue,
                    &self.deduplicator,
                    &self.ack_manager,
                    self.known_peers.len(),
                    is_local_relay,
                );
                self.last_metrics_emit_at = Some(now);
                dispatch_record(
                    &ctx.sink,
                    &TelemetryRecord::MetricsSnapshot(Box::new(frame)),
                );
            }
        }
    }

    /// Processes messages ready for retry from the retry queue.
    ///
    /// EDGE CASE HANDLING:
    /// - Checks transport availability before each retry attempt
    /// - Handles transport switch mid-retry
    /// - Properly tracks retry counts and transport failures
    fn process_retry_queue(&mut self) -> Result<()> {
        // Limit batch size to prevent blocking on large queues
        let mut max_batch_size = crate::constants::FLUSH_BATCH_LIMIT;

        // Opt-in resend rate cap: bound how fast the RETRY/PROBE path re-sends,
        // so a backlog of resends to unreachable peers cannot burst past the
        // relay's per-connection rate limit and flap the connection. Refills a
        // token bucket by elapsed wall-time and caps this call's batch to the
        // whole tokens available. Only resends are bounded; first-attempt sends,
        // ACKs, control frames and handshakes take other paths and are never
        // throttled, so live traffic keeps full throughput and latency. Fully
        // off by default (unchanged retry behaviour for every existing app).
        if self.config.reliability.retry.edge_driven_unreachable_dm {
            let now = Instant::now();
            match self.retry_drain_last {
                Some(last) => {
                    let elapsed = now.duration_since(last).as_secs_f64();
                    self.retry_drain_tokens = (self.retry_drain_tokens
                        + elapsed * crate::protocol::types::RETRY_DRAIN_RATE_PER_SEC)
                        .min(crate::protocol::types::RETRY_DRAIN_BURST);
                }
                None => {
                    // First drain: start with a full burst budget.
                    self.retry_drain_tokens = crate::protocol::types::RETRY_DRAIN_BURST;
                }
            }
            self.retry_drain_last = Some(now);
            let budget = self.retry_drain_tokens.floor() as usize;
            max_batch_size = max_batch_size.min(budget);
        }

        let mut processed = 0;

        while processed < max_batch_size {
            let mut entry = match self.retry_queue.dequeue_ready() {
                Some(e) => e,
                None => break,
            };

            processed += 1;
            let previous_transport = self.transport_manager.current_transport();
            // Normally finds the entry already there: a retry entry outlives
            // its outbox entry only if a give-up path skipped
            // `retire_undeliverable_message`, and at capacity this call would
            // then evict a live message to bring a settled one back.
            self.ensure_outbox_entry(&entry.message);

            // Tier 2: re-seal against the peer's current session before this
            // resend, so a post-re-key epoch change no longer leaves the bytes
            // undecryptable. No-op for media/plaintext/unconfirmed sessions.
            self.reseal_for_resend_in_place(&mut entry.message);

            let forced_transport = self.pinned_media_transport_for_message(&entry.message.id);
            let send_result = if let Some(transport) = forced_transport {
                self.transport_manager
                    .send_via_transport(&entry.message, transport)
            } else {
                self.transport_manager.send(&entry.message)
            };
            let current_transport =
                forced_transport.or_else(|| self.transport_manager.current_transport());

            match send_result {
                Ok(()) => {
                    let ack_registered_now = self.ensure_ack_registration(&entry.message)?;

                    if ack_registered_now {
                        // The resend registered a brand-new ACK (the prior one
                        // was cleared, e.g. by an ACK timeout that re-queued
                        // this message). A fresh ACK starts at retry_count 0, so
                        // carry the count the retry-queue entry already holds —
                        // advanced by this attempt — onto it. Without this the
                        // backoff ladder resets to 0 on every relay-accepted
                        // resend and `delay_for_retry` is pinned at its 1s floor
                        // forever for a peer that never ACKs (an offline
                        // recipient), flooding the relay ~once per second.
                        self.ack_manager.set_retry_count(
                            &entry.message.id,
                            entry.retry_count.saturating_add(1),
                        );
                    } else {
                        self.ack_manager.increment_retry_count(&entry.message.id);
                    }
                    self.mark_message_sent(
                        &entry.message,
                        current_transport,
                        Some(entry.retry_count.saturating_add(1)),
                    );

                    if let Some(transport) = current_transport {
                        self.transport_manager.reset_retry_count(transport);
                    }
                    self.offer_resend_to_mesh(&entry.message, true);

                    debug!(
                        message_id = %entry.message.id,
                        retry_count = entry.retry_count,
                        transport = ?current_transport,
                        "Retry send succeeded"
                    );
                }
                Err(e) => {
                    self.offer_resend_to_mesh(&entry.message, false);
                    // Re-enqueue with incremented retry count for backoff
                    let next_retry_at = self
                        .retry_queue
                        .enqueue(entry.message.clone(), entry.retry_count + 1);

                    if let Some(transport) = forced_transport.or(previous_transport) {
                        self.transport_manager.record_retry_failure(transport);
                    }

                    // Surface the schedule so apps can show retry state
                    // instead of inferring it from silence. None = the id
                    // was already queued; the earlier emission stands.
                    if let Some(retry_at) = next_retry_at {
                        self.emit_event(Event::message_retrying(
                            entry.message.id.clone(),
                            entry.message.recipient.as_str().to_string(),
                            entry.retry_count + 1,
                            retry_at.timestamp_millis(),
                        ));
                    }

                    debug!(
                        message_id = %entry.message.id,
                        retry_count = entry.retry_count,
                        transport = ?current_transport,
                        error = %e,
                        "Retry send failed, will retry later"
                    );
                }
            }
        }

        if processed > 0 {
            debug!(processed = processed, "Processed retry queue entries");
            // Spend the resend-rate tokens actually used (opt-in cap only).
            if self.config.reliability.retry.edge_driven_unreachable_dm {
                self.retry_drain_tokens = (self.retry_drain_tokens - processed as f64).max(0.0);
            }
        }

        Ok(())
    }

    /// Processes timed out ACKs and handles retry or failure.
    fn process_timed_out_acks(&mut self) -> Result<()> {
        let timed_out = self.ack_manager.drain_timed_out();
        for pending in timed_out {
            let message_id = pending.message_id.clone();

            if pending.retry_count >= self.config.reliability.retry.max_retries {
                self.handle_max_retries_exceeded(&message_id, pending.retry_count)?;
                continue;
            }

            self.handle_ack_timeout_retry(&message_id, pending.retry_count)?;
        }
        Ok(())
    }

    /// Handles a message that has exceeded maximum retries.
    fn handle_max_retries_exceeded(
        &mut self,
        message_id: &MessageId,
        retry_count: u32,
    ) -> Result<()> {
        // A plain DM whose recipient still holds a live unreachable-park
        // counter re-parks instead of settling: the exhausted budget was burnt
        // by reachability probes — sends that succeed locally without proving
        // the peer is back — not by a peer believed reachable (see
        // try_repark_exhausted_dm).
        if self.try_repark_exhausted_dm(message_id) {
            return Ok(());
        }

        // Retry exhaustion is terminal for a connection request: settle the
        // pending entry and surface the typed undeliverable signal alongside
        // the generic message_failed, so apps keep connection-request
        // context without parsing reason strings.
        let undeliverable_recipient = self.take_undeliverable_connection_request(message_id);

        let state = lock_shared_state(&self.shared_state).map_err(|e| {
            error!(
                "Failed to lock shared state for message failed event: {}",
                e
            );
            e
        })?;
        state.emit_event(Event::message_failed(
            message_id.clone(),
            "Max retries exceeded",
            retry_count,
        ));
        if let Some(recipient) = undeliverable_recipient {
            warn!(
                recipient = %recipient,
                message_id = %message_id,
                "Connection request undeliverable: max retries exceeded"
            );
            state.emit_event(Event::connection_request_undeliverable(
                recipient,
                message_id.as_str(),
                "max_retries_exceeded",
            ));
        }
        drop(state);

        self.retire_undeliverable_message(message_id, "max retries exceeded");
        Ok(())
    }

    /// Handles retry logic for a timed out ACK.
    fn handle_ack_timeout_retry(&mut self, message_id: &MessageId, retry_count: u32) -> Result<()> {
        if let Some(entry) = self
            .outbox
            .get(message_id)
            .or_else(|| self.media_outbox.get(message_id))
        {
            let message_clone = entry.message.clone();
            let recipient = entry.message.recipient.as_str().to_string();
            let last_transport = entry.last_transport;

            // enqueue is infallible (retry queue has no attempt limit)
            let next_retry_at = self.retry_queue.enqueue(message_clone, retry_count);
            if let Some(transport) = last_transport {
                self.transport_manager.record_retry_failure(transport);
            }
            if let Some(retry_at) = next_retry_at {
                self.emit_event(Event::message_retrying(
                    message_id.clone(),
                    recipient,
                    retry_count,
                    retry_at.timestamp_millis(),
                ));
            }
        } else {
            self.handle_missing_outbox_entry(message_id, retry_count)?;
        }
        Ok(())
    }

    /// Emits a protocol event to all registered handlers.
    ///
    /// Silently no-ops if the shared state lock is poisoned.
    pub(crate) fn emit_event(&self, event: Event) {
        if let Ok(state) = lock_shared_state(&self.shared_state) {
            state.emit_event(event);
        }
    }

    /// Emits a terminal settlement, or parks it until `start()` if the event
    /// pipeline is not live yet.
    ///
    /// Restore paths settle messages the app is still holding ids for. They run
    /// from `initialize_mls`, before `start()` and often before the app has
    /// installed its event callback, so emitting directly there would silently
    /// discard exactly the signals that exist to stop an id from hanging
    /// forever. Once running, this is a plain emit — the same call is used from
    /// `process()`-driven expiry, where no deferral is wanted.
    pub(crate) fn settle_restored_message_failure(&mut self, event: Event) {
        let running = self.protocol_is_running();
        self.settle_one_restored_message_failure(event, running);
    }

    /// Settles a batch, taking the shared-state lock once rather than once per
    /// event. Restore emits these in loops that a tampered or pre-split store
    /// can drive into the thousands.
    pub(crate) fn settle_restored_message_failures(
        &mut self,
        events: impl IntoIterator<Item = Event>,
    ) {
        let running = self.protocol_is_running();
        for event in events {
            self.settle_one_restored_message_failure(event, running);
        }
    }

    /// Whether [`Self::start`] has run and nothing has stopped it since.
    ///
    /// Two callers want different things from the same fact. The restore
    /// paths above want to know whether an emit reaches an application at
    /// all. The document layer wants to know whether a frame it is about to
    /// build can leave the device, because a send attempted before the
    /// transports are up stamps a rate-limit window against an offer that
    /// was never carried.
    pub(crate) fn protocol_is_running(&self) -> bool {
        lock_shared_state(&self.shared_state)
            .map(|state| state.state == ProtocolState::Running)
            .unwrap_or(false)
    }

    fn settle_one_restored_message_failure(&mut self, event: Event, running: bool) {
        if running {
            self.emit_event(event);
            return;
        }
        // Every other accumulation on the restore path has an explicit ceiling
        // and logs what it dropped; this one is retained until `start()`, which
        // may never come, so it gets the same treatment. Keeping the *oldest*
        // is deliberate: the settlements a restore produces first are the ones
        // for records it examined first, and dropping those in favour of later
        // ones would bias the survivors by backend listing order.
        if self.deferred_restore_settlements.len() >= MAX_DEFERRED_RESTORE_SETTLEMENTS {
            self.suppressed_restore_settlements += 1;
            return;
        }
        self.deferred_restore_settlements.push(event);
    }

    /// Returns an iterator over outbox messages (test-only).
    #[cfg(test)]
    pub(crate) fn outbox_messages(&self) -> impl Iterator<Item = &Message> {
        self.outbox.values().map(|e| &e.message)
    }

    /// Returns the total number of entries across outbox and media_outbox (test-only).
    #[cfg(test)]
    pub(crate) fn outbox_entry_count(&self) -> usize {
        self.outbox.len() + self.media_outbox.len()
    }

    /// Clears all outbox entries (test-only).
    #[cfg(test)]
    pub(crate) fn clear_outbox(&mut self) {
        self.outbox.clear();
    }

    /// Returns `true` when `message_id` is anywhere in the reliability
    /// machinery — awaiting an ACK, held in the outbox, or queued for retry
    /// (test-only).
    ///
    /// Exists for the fire-and-forget frames (relay hints, delivery ACKs),
    /// whose correctness is defined by leaving *no* trace here: an un-ACKable
    /// frame that registers a pending ACK is retransmitted until the retry cap
    /// and then reported failed.
    #[cfg(test)]
    pub(crate) fn is_tracked_for_delivery(&self, message_id: &MessageId) -> bool {
        self.ack_manager.is_waiting_for_ack(message_id)
            || self.outbox.contains_key(message_id)
            || self.media_outbox.contains_key(message_id)
            || self.retry_queue.contains(&message_id.as_str())
    }

    /// Returns a reference to the MLS manager Arc (test-only).
    #[cfg(test)]
    pub(crate) fn mls_manager_for_testing(&self) -> &Arc<RwLock<offline_protocol_mls::MlsManager>> {
        self.mls_manager
            .as_ref()
            .expect("MLS not initialized in test")
    }

    /// Acquires a read guard on the MLS manager.
    ///
    /// Returns an error if MLS is not initialized or the lock is poisoned.
    pub(crate) fn read_mls_guard(
        &self,
    ) -> Result<std::sync::RwLockReadGuard<'_, offline_protocol_mls::MlsManager>> {
        let mls = self
            .mls_manager
            .as_ref()
            .ok_or_else(|| Error::Other("MLS not initialized".to_string()))?;
        mls.read()
            .map_err(|_| Error::Other("MLS lock poisoned".to_string()))
    }
}

#[cfg(test)]
pub(crate) mod tests;