peat-btle 0.4.0

// Copyright (c) 2025-2026 (r)evolve - Revolve Team LLC
// SPDX-License-Identifier: Apache-2.0
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

//! PeatMesh - Unified mesh management facade
//!
//! This module provides the main entry point for Peat BLE mesh operations.
//! It composes peer management, document sync, and observer notifications
//! into a single interface that platform implementations can use.
//!
//! ## Usage
//!
//! ```ignore
//! use peat_btle::peat_mesh::{PeatMesh, PeatMeshConfig};
//! use peat_btle::observer::{PeatEvent, PeatObserver};
//! use peat_btle::NodeId;
//! use std::sync::Arc;
//!
//! // Create mesh configuration
//! let config = PeatMeshConfig::new(NodeId::new(0x12345678), "ALPHA-1", "DEMO");
//!
//! // Create mesh instance
//! let mesh = PeatMesh::new(config);
//!
//! // Add observer for events
//! struct MyObserver;
//! impl PeatObserver for MyObserver {
//!     fn on_event(&self, event: PeatEvent) {
//!         println!("Event: {:?}", event);
//!     }
//! }
//! mesh.add_observer(Arc::new(MyObserver));
//!
//! // Platform BLE callbacks
//! mesh.on_ble_discovered("device-uuid", Some("PEAT_DEMO-AABBCCDD"), -65, Some("DEMO"), now_ms);
//! mesh.on_ble_connected("device-uuid", now_ms);
//! mesh.on_ble_data_received("device-uuid", &data, now_ms);
//!
//! // Periodic maintenance
//! if let Some(sync_data) = mesh.tick(now_ms) {
//!     // Broadcast sync_data to connected peers
//! }
//! ```

#[cfg(not(feature = "std"))]
use alloc::{string::String, sync::Arc, vec::Vec};
#[cfg(feature = "std")]
use std::collections::HashMap;
#[cfg(feature = "std")]
use std::sync::Arc;

use crate::document::{ENCRYPTED_MARKER, KEY_EXCHANGE_MARKER, PEER_E2EE_MARKER};
#[cfg(feature = "translator-codec")]
use crate::document::{
    TRANSLATOR_FRAME_MARKER, TRANSLATOR_RESERVED_MARKER_END, TRANSLATOR_RESERVED_MARKER_START,
};
use crate::document_sync::DocumentSync;
use crate::gossip::{GossipStrategy, RandomFanout};
use crate::observer::{DisconnectReason, PeatEvent, PeatObserver, SecurityViolationKind};
use crate::peer::{
    ConnectionStateGraph, FullStateCountSummary, IndirectPeer, PeatPeer, PeerConnectionState,
    PeerDegree, PeerManagerConfig, StateCountSummary,
};
use crate::peer_manager::PeerManager;
use crate::relay::{
    MessageId, RelayEnvelope, SeenMessageCache, DEFAULT_MAX_HOPS, DEFAULT_SEEN_TTL_MS,
    RELAY_ENVELOPE_MARKER,
};
use crate::security::{
    DeviceIdentity, IdentityAttestation, IdentityRegistry, KeyExchangeMessage, MeshEncryptionKey,
    PeerEncryptedMessage, PeerSessionManager, RegistryResult, SessionState,
};
use crate::sync::crdt::{EventType, Peripheral, PeripheralType};
use crate::sync::delta::{DeltaEncoder, DeltaStats};
use crate::sync::delta_document::{DeltaDocument, Operation};
use crate::NodeId;

#[cfg(feature = "std")]
use crate::observer::ObserverManager;

use crate::registry::DocumentRegistry;

/// Configuration for PeatMesh
#[derive(Debug, Clone)]
pub struct PeatMeshConfig {
    /// Our node ID
    pub node_id: NodeId,

    /// Our callsign (e.g., "ALPHA-1")
    pub callsign: String,

    /// Mesh ID to filter peers (e.g., "DEMO")
    pub mesh_id: String,

    /// Peripheral type for this device
    pub peripheral_type: PeripheralType,

    /// Peer management configuration
    pub peer_config: PeerManagerConfig,

    /// Sync interval in milliseconds (how often to broadcast state)
    pub sync_interval_ms: u64,

    /// Whether to auto-broadcast on emergency/ack
    pub auto_broadcast_events: bool,

    /// Optional shared secret for mesh-wide encryption (32 bytes)
    ///
    /// When set, all documents are encrypted using ChaCha20-Poly1305 before
    /// transmission and decrypted upon receipt. All nodes in the mesh must
    /// share the same secret to communicate.
    pub encryption_secret: Option<[u8; 32]>,

    /// Strict encryption mode - reject unencrypted documents when encryption is enabled
    ///
    /// When true and encryption is enabled, any unencrypted documents received
    /// will be rejected and trigger a SecurityViolation event. This prevents
    /// downgrade attacks where an adversary sends unencrypted malicious documents.
    ///
    /// Default: false (backward compatible - accepts unencrypted for gradual rollout)
    pub strict_encryption: bool,

    /// Enable multi-hop relay
    ///
    /// When enabled, received messages will be forwarded to other peers based
    /// on the gossip strategy. Requires message deduplication to prevent loops.
    ///
    /// Default: false
    pub enable_relay: bool,

    /// Maximum hops for relay messages (TTL)
    ///
    /// Messages will not be relayed beyond this many hops from the origin.
    /// Default: 7
    pub max_relay_hops: u8,

    /// Gossip fanout for relay
    ///
    /// Number of peers to forward each message to. Higher values increase
    /// convergence speed but also bandwidth usage.
    /// Default: 2
    pub relay_fanout: usize,

    /// TTL for seen message cache (milliseconds)
    ///
    /// How long to remember message IDs for deduplication.
    /// Default: 300_000 (5 minutes)
    pub seen_cache_ttl_ms: u64,
}

impl PeatMeshConfig {
    /// Create a new configuration with required fields
    pub fn new(node_id: NodeId, callsign: &str, mesh_id: &str) -> Self {
        Self {
            node_id,
            callsign: callsign.into(),
            mesh_id: mesh_id.into(),
            peripheral_type: PeripheralType::SoldierSensor,
            peer_config: PeerManagerConfig::with_mesh_id(mesh_id),
            sync_interval_ms: 5000,
            auto_broadcast_events: true,
            encryption_secret: None,
            strict_encryption: false,
            enable_relay: false,
            max_relay_hops: DEFAULT_MAX_HOPS,
            relay_fanout: 2,
            seen_cache_ttl_ms: DEFAULT_SEEN_TTL_MS,
        }
    }

    /// Enable mesh-wide encryption with a shared secret
    ///
    /// All documents will be encrypted using ChaCha20-Poly1305 before
    /// transmission. All mesh participants must use the same secret.
    pub fn with_encryption(mut self, secret: [u8; 32]) -> Self {
        self.encryption_secret = Some(secret);
        self
    }

    /// Set peripheral type
    pub fn with_peripheral_type(mut self, ptype: PeripheralType) -> Self {
        self.peripheral_type = ptype;
        self
    }

    /// Set sync interval
    pub fn with_sync_interval(mut self, interval_ms: u64) -> Self {
        self.sync_interval_ms = interval_ms;
        self
    }

    /// Set peer timeout
    pub fn with_peer_timeout(mut self, timeout_ms: u64) -> Self {
        self.peer_config.peer_timeout_ms = timeout_ms;
        self
    }

    /// Set max peers (for embedded systems)
    pub fn with_max_peers(mut self, max: usize) -> Self {
        self.peer_config.max_peers = max;
        self
    }

    /// Enable strict encryption mode
    ///
    /// When enabled (and encryption is also enabled), any unencrypted documents
    /// received will be rejected and trigger a `SecurityViolation` event.
    /// This prevents downgrade attacks.
    ///
    /// Note: This only has effect when encryption is enabled via `with_encryption()`.
    pub fn with_strict_encryption(mut self) -> Self {
        self.strict_encryption = true;
        self
    }

    /// Enable multi-hop relay
    ///
    /// When enabled, received messages will be forwarded to other connected peers
    /// based on the gossip strategy. This enables mesh-wide message propagation.
    pub fn with_relay(mut self) -> Self {
        self.enable_relay = true;
        self
    }

    /// Set maximum relay hops (TTL)
    ///
    /// Messages will not be relayed beyond this many hops from the origin.
    pub fn with_max_relay_hops(mut self, max_hops: u8) -> Self {
        self.max_relay_hops = max_hops;
        self
    }

    /// Set gossip fanout for relay
    ///
    /// Number of peers to forward each message to.
    pub fn with_relay_fanout(mut self, fanout: usize) -> Self {
        self.relay_fanout = fanout.max(1);
        self
    }

    /// Set TTL for seen message cache
    ///
    /// How long to remember message IDs for deduplication (milliseconds).
    pub fn with_seen_cache_ttl(mut self, ttl_ms: u64) -> Self {
        self.seen_cache_ttl_ms = ttl_ms;
        self
    }
}

/// Type alias for app document storage to reduce type complexity
#[cfg(feature = "std")]
type AppDocumentStore =
    std::sync::RwLock<HashMap<(u8, u32, u64), Box<dyn core::any::Any + Send + Sync>>>;

/// Main facade for Peat BLE mesh operations
///
/// Composes peer management, document sync, and observer notifications.
/// Platform implementations call into this from their BLE callbacks.
#[cfg(feature = "std")]
pub struct PeatMesh {
    /// Configuration
    config: PeatMeshConfig,

    /// Peer manager
    peer_manager: PeerManager,

    /// Document sync
    document_sync: DocumentSync,

    /// Observer manager
    observers: ObserverManager,

    /// Last sync broadcast time (u32 wraps every ~49 days, sufficient for intervals)
    last_sync_ms: std::sync::atomic::AtomicU32,

    /// Last cleanup time
    last_cleanup_ms: std::sync::atomic::AtomicU32,

    /// Monotonic outbound counter for peat-lite Document frames
    /// (`MessageType::Document`). Threaded into the wire header so
    /// polled receivers can dedup on `(source_node_id, seq_num)` —
    /// pre-fix callers passed 0 by default and silently disabled
    /// receive-side dedup. Wraps every ~4B publishes (u32), well
    /// past any realistic per-session frame rate.
    #[cfg(feature = "peat-lite-frame")]
    peat_lite_seq: std::sync::atomic::AtomicU32,

    /// Optional mesh-wide encryption key (derived from shared secret)
    encryption_key: Option<MeshEncryptionKey>,

    /// Optional per-peer E2EE session manager
    peer_sessions: std::sync::Mutex<Option<PeerSessionManager>>,

    /// Connection state graph for tracking peer connection lifecycle
    connection_graph: std::sync::Mutex<ConnectionStateGraph>,

    /// Seen message cache for relay deduplication
    seen_cache: std::sync::Mutex<SeenMessageCache>,

    /// Gossip strategy for relay peer selection
    gossip_strategy: Box<dyn GossipStrategy>,

    /// Delta encoder for per-peer sync state tracking
    ///
    /// Tracks what data has been sent to each peer to enable delta sync
    /// (sending only changes instead of full documents).
    delta_encoder: std::sync::Mutex<DeltaEncoder>,

    /// This node's cryptographic identity (Ed25519 keypair)
    ///
    /// When set, the node_id is derived from the public key and documents
    /// can be signed for authenticity verification.
    identity: Option<DeviceIdentity>,

    /// TOFU identity registry for tracking peer identities
    ///
    /// Maps node_id to public key on first contact, rejects mismatches.
    identity_registry: std::sync::Mutex<IdentityRegistry>,

    /// Peripheral state received from peers
    ///
    /// Stores the most recent peripheral data (callsign, location, etc.)
    /// received from each peer via document sync.
    peer_peripherals: std::sync::RwLock<HashMap<NodeId, Peripheral>>,

    /// Document registry for app-layer CRDT types.
    ///
    /// Enables external crates to register custom document types that sync
    /// through the mesh using the extensible registry pattern.
    document_registry: DocumentRegistry,

    /// Storage for app-layer documents.
    ///
    /// Keyed by (type_id, source_node, timestamp) to uniquely identify each
    /// document instance. Values are type-erased boxes that can be downcast
    /// using the document registry handlers.
    app_documents: AppDocumentStore,

    /// ADR-059 BleTranslator instance for inbound translator-frame decoding.
    ///
    /// One translator per PeatMesh, constructed with `TranslationConfig::default()`.
    /// Operators wanting custom collection names can swap via
    /// `set_translator_config`. Always present when the
    /// `translator-codec` feature is enabled — the receive-dispatch
    /// branch in `on_ble_data_received_anonymous` (and siblings) calls
    /// `decode_inbound_sync` on it for every 0xB6 frame.
    #[cfg(feature = "translator-codec")]
    ble_translator: std::sync::RwLock<crate::translator::BleTranslator>,

    /// Optional UniFFI-exported `DecodedDocumentJsonCallback` set by
    /// host bindings (Kotlin / Swift via peat-btle's UniFFI surface).
    /// ADR-059 Amendment 2 specifies this as the production wiring
    /// point for cross-transport receive — peat-atak-plugin's
    /// `BleDecodedDocumentBridge` implements the trait and forwards
    /// each decoded doc into peat-ffi's `publishDocument` with
    /// `origin="ble"`.
    ///
    /// **ADR-059 Amendment 4 Slice 4.b note.** The Rust-native
    /// `DecodedDocumentCallback` trait that previously lived alongside
    /// this one (taking the typed `peat_mesh::sync::Document`) is
    /// deleted: peat-btle has no peat-mesh dep at all post-Slice-4.b,
    /// so the only payload shape that crosses the FFI boundary now is
    /// the JSON projection. Rust-native consumers either consume the
    /// polled `DataReceivedResult::decoded_translator_frame` field
    /// (Amendment 3) or wire a `Translator` impl on the peat-mesh
    /// side (Slice 4.a's `peat_mesh::transport::btle_translator`).
    #[cfg(all(feature = "translator-codec", feature = "uniffi"))]
    decoded_document_json_callback:
        std::sync::RwLock<Option<Arc<dyn crate::DecodedDocumentJsonCallback>>>,

    /// ADR-059 Amendment 3 — set when a host calls
    /// [`acknowledge_polled_translator_consumer`](Self::acknowledge_polled_translator_consumer)
    /// to attest that it consumes
    /// [`DataReceivedResult::decoded_translator_frame`] per receive.
    /// Read by the receive dispatch as the third suppression branch
    /// for `PeatEvent::TranslatorNoCallback` (the event fires only
    /// when all three are absent — no Rust-trait callback, no UniFFI
    /// JSON callback, no polled attestation).
    ///
    /// **Defined unconditionally** (not `cfg`-gated) so the binding
    /// shape is stable across feature combos — UniFFI hosts call the
    /// same Kotlin signature against any peat-btle build. When
    /// `translator-codec` is off the read site (suppression check) is
    /// gated alongside the rest of the dispatch logic; the flag still
    /// exists, it just has no effect.
    polled_translator_consumer_attested: std::sync::atomic::AtomicBool,
}

#[cfg(feature = "std")]
impl PeatMesh {
    /// Create a new PeatMesh instance
    pub fn new(config: PeatMeshConfig) -> Self {
        let peer_manager = PeerManager::new(config.node_id, config.peer_config.clone());
        let document_sync = DocumentSync::with_peripheral_type(
            config.node_id,
            &config.callsign,
            config.peripheral_type,
        );

        // Derive encryption key from shared secret if configured
        let encryption_key = config
            .encryption_secret
            .map(|secret| MeshEncryptionKey::from_shared_secret(&config.mesh_id, &secret));

        // Create connection state graph with config thresholds
        let connection_graph = ConnectionStateGraph::with_config(
            config.peer_config.rssi_degraded_threshold,
            config.peer_config.lost_timeout_ms,
        );

        // Create seen message cache for relay deduplication
        let seen_cache = SeenMessageCache::with_ttl(config.seen_cache_ttl_ms);

        // Create gossip strategy for relay
        let gossip_strategy: Box<dyn GossipStrategy> =
            Box::new(RandomFanout::new(config.relay_fanout));

        // Create delta encoder for per-peer sync state tracking
        let delta_encoder = DeltaEncoder::new(config.node_id);

        let document_registry = DocumentRegistry::new();

        Self {
            config,
            peer_manager,
            document_sync,
            observers: ObserverManager::new(),
            last_sync_ms: std::sync::atomic::AtomicU32::new(0),
            last_cleanup_ms: std::sync::atomic::AtomicU32::new(0),
            #[cfg(feature = "peat-lite-frame")]
            peat_lite_seq: std::sync::atomic::AtomicU32::new(0),
            encryption_key,
            peer_sessions: std::sync::Mutex::new(None),
            connection_graph: std::sync::Mutex::new(connection_graph),
            seen_cache: std::sync::Mutex::new(seen_cache),
            gossip_strategy,
            delta_encoder: std::sync::Mutex::new(delta_encoder),
            identity: None,
            identity_registry: std::sync::Mutex::new(IdentityRegistry::new()),
            peer_peripherals: std::sync::RwLock::new(HashMap::new()),
            document_registry,
            app_documents: std::sync::RwLock::new(HashMap::new()),
            #[cfg(feature = "translator-codec")]
            ble_translator: std::sync::RwLock::new(
                crate::translator::BleTranslator::with_defaults(),
            ),
            #[cfg(all(feature = "translator-codec", feature = "uniffi"))]
            decoded_document_json_callback: std::sync::RwLock::new(None),
            polled_translator_consumer_attested: std::sync::atomic::AtomicBool::new(false),
        }
    }

    /// Create a new PeatMesh with a cryptographic identity
    ///
    /// The node_id will be derived from the identity's public key, overriding
    /// any node_id specified in the config. This ensures cryptographic binding
    /// between node_id and identity.
    pub fn with_identity(config: PeatMeshConfig, identity: DeviceIdentity) -> Self {
        // Override node_id with identity-derived value
        let mut config = config;
        config.node_id = identity.node_id();

        let peer_manager = PeerManager::new(config.node_id, config.peer_config.clone());
        let document_sync = DocumentSync::with_peripheral_type(
            config.node_id,
            &config.callsign,
            config.peripheral_type,
        );

        let encryption_key = config
            .encryption_secret
            .map(|secret| MeshEncryptionKey::from_shared_secret(&config.mesh_id, &secret));

        let connection_graph = ConnectionStateGraph::with_config(
            config.peer_config.rssi_degraded_threshold,
            config.peer_config.lost_timeout_ms,
        );

        let seen_cache = SeenMessageCache::with_ttl(config.seen_cache_ttl_ms);
        let gossip_strategy: Box<dyn GossipStrategy> =
            Box::new(RandomFanout::new(config.relay_fanout));
        let delta_encoder = DeltaEncoder::new(config.node_id);

        let document_registry = DocumentRegistry::new();

        Self {
            config,
            peer_manager,
            document_sync,
            observers: ObserverManager::new(),
            last_sync_ms: std::sync::atomic::AtomicU32::new(0),
            last_cleanup_ms: std::sync::atomic::AtomicU32::new(0),
            #[cfg(feature = "peat-lite-frame")]
            peat_lite_seq: std::sync::atomic::AtomicU32::new(0),
            encryption_key,
            peer_sessions: std::sync::Mutex::new(None),
            connection_graph: std::sync::Mutex::new(connection_graph),
            seen_cache: std::sync::Mutex::new(seen_cache),
            gossip_strategy,
            delta_encoder: std::sync::Mutex::new(delta_encoder),
            identity: Some(identity),
            identity_registry: std::sync::Mutex::new(IdentityRegistry::new()),
            peer_peripherals: std::sync::RwLock::new(HashMap::new()),
            document_registry,
            app_documents: std::sync::RwLock::new(HashMap::new()),
            #[cfg(feature = "translator-codec")]
            ble_translator: std::sync::RwLock::new(
                crate::translator::BleTranslator::with_defaults(),
            ),
            #[cfg(all(feature = "translator-codec", feature = "uniffi"))]
            decoded_document_json_callback: std::sync::RwLock::new(None),
            polled_translator_consumer_attested: std::sync::atomic::AtomicBool::new(false),
        }
    }

    /// Create a new PeatMesh from genesis data
    ///
    /// This is the recommended way to create a mesh for production use.
    /// The mesh will be configured with:
    /// - node_id derived from identity
    /// - mesh_id from genesis
    /// - encryption enabled using genesis-derived secret
    pub fn from_genesis(
        genesis: &crate::security::MeshGenesis,
        identity: DeviceIdentity,
        callsign: &str,
    ) -> Self {
        let config = PeatMeshConfig::new(identity.node_id(), callsign, &genesis.mesh_id())
            .with_encryption(genesis.encryption_secret());

        Self::with_identity(config, identity)
    }

    /// Create a PeatMesh from persisted state.
    ///
    /// Restores mesh configuration from previously saved state, including:
    /// - Device identity (Ed25519 keypair)
    /// - Mesh genesis (if present)
    /// - Identity registry (TOFU cache)
    ///
    /// Use this on device boot to restore mesh membership without re-provisioning.
    ///
    /// # Arguments
    ///
    /// * `state` - Previously persisted state
    /// * `callsign` - Human-readable identifier (may differ from original)
    ///
    /// # Errors
    ///
    /// Returns `PersistenceError` if the identity cannot be restored.
    ///
    /// # Example
    ///
    /// ```ignore
    /// // On boot, restore from secure storage
    /// let state = PersistedState::load(&storage)?;
    /// let mesh = PeatMesh::from_persisted(state, "SENSOR-01")?;
    /// ```
    #[cfg(feature = "std")]
    pub fn from_persisted(
        state: crate::security::PersistedState,
        callsign: &str,
    ) -> Result<Self, crate::security::PersistenceError> {
        // Restore identity
        let identity = state.restore_identity()?;

        // Restore genesis (if present)
        let genesis = state.restore_genesis();

        // Create mesh with or without genesis
        let mesh = if let Some(ref gen) = genesis {
            Self::from_genesis(gen, identity, callsign)
        } else {
            let config = PeatMeshConfig::new(identity.node_id(), callsign, "RESTORED");
            Self::with_identity(config, identity)
        };

        // Restore identity registry
        let restored_registry = state.restore_registry();
        if let Ok(mut registry) = mesh.identity_registry.lock() {
            *registry = restored_registry;
        }

        log::info!(
            "PeatMesh restored from persisted state: node_id={:08X}, known_peers={}",
            mesh.config.node_id.as_u32(),
            mesh.known_identity_count()
        );

        Ok(mesh)
    }

    /// Create persisted state from current mesh state.
    ///
    /// Captures the current identity, genesis, and registry for persistence.
    /// Call this periodically or before shutdown to save state.
    ///
    /// # Arguments
    ///
    /// * `genesis` - Optional genesis to include (if mesh was created from genesis)
    ///
    /// # Returns
    ///
    /// `None` if the mesh has no identity bound.
    #[cfg(feature = "std")]
    pub fn to_persisted_state(
        &self,
        genesis: Option<&crate::security::MeshGenesis>,
    ) -> Option<crate::security::PersistedState> {
        let identity = self.identity.as_ref()?;
        let registry = self.identity_registry.lock().ok()?;

        Some(crate::security::PersistedState::with_registry(
            identity, genesis, &registry,
        ))
    }

    // ==================== ADR-059 translator-frame receive ====================

    /// Install the UniFFI-exported
    /// [`DecodedDocumentJsonCallback`](crate::DecodedDocumentJsonCallback)
    /// invoked when a 0xB6 translator frame decodes successfully (ADR-059
    /// Amendment 2 — host-side wiring point for cross-transport receive).
    ///
    /// Idempotent — calling again replaces the previous callback. When
    /// no callback is installed and no polled-consumer attestation has
    /// fired (see [`acknowledge_polled_translator_consumer`](Self::acknowledge_polled_translator_consumer)),
    /// `PeatEvent::TranslatorNoCallback` fires on the observer channel
    /// so the gap is operator-observable.
    ///
    /// **ADR-059 Amendment 4 Slice 4.b note.** The previously parallel
    /// Rust-trait `set_decoded_document_callback` (taking
    /// `peat_mesh::sync::Document`) is deleted along with the trait
    /// itself; peat-btle has no peat-mesh dep at all post-Slice-4.b, so
    /// the only payload shape is the JSON projection. Rust-native
    /// consumers that need typed `Document` should wire a `Translator`
    /// impl on the peat-mesh side (Slice 4.a's
    /// `peat_mesh::transport::btle_translator`).
    #[cfg(all(feature = "translator-codec", feature = "uniffi"))]
    pub fn set_decoded_document_json_callback(
        &self,
        cb: Box<dyn crate::DecodedDocumentJsonCallback>,
    ) {
        // UniFFI 0.31 callback_interface generates a `Box<dyn Trait>` on
        // the FFI surface. We store as `Arc<dyn Trait>` internally so the
        // receive dispatch can snapshot-clone the handle out of the
        // RwLock without holding the lock during user-supplied dispatch.
        let cb_arc: Arc<dyn crate::DecodedDocumentJsonCallback> = Arc::from(cb);
        if let Ok(mut slot) = self.decoded_document_json_callback.write() {
            *slot = Some(cb_arc);
        }
    }

    /// Mark this `PeatMesh` as having a polled-field consumer (ADR-059
    /// Amendment 3). Hosts that read
    /// [`DataReceivedResult::decoded_translator_frame`] per receive and
    /// forward through their own publish path SHOULD call this once at
    /// startup, after constructing `PeatMesh` and before any GATT
    /// receive arrives. Without this attestation, the receive dispatch
    /// has no way to distinguish a polled-consumer host from one that
    /// decoded the frame and dropped it on the floor, and
    /// `PeatEvent::TranslatorNoCallback` will fire for every frame.
    ///
    /// Idempotent — calling again is a no-op. Defined unconditionally
    /// so the binding shape is stable across feature combos; when
    /// `translator-codec` is off the suppression-check read site is
    /// gated alongside the rest of the dispatch logic, so the flag has
    /// no effect (still safe to call).
    pub fn acknowledge_polled_translator_consumer(&self) {
        self.polled_translator_consumer_attested
            .store(true, std::sync::atomic::Ordering::Release);
    }

    /// Replace the active `BleTranslator` configuration. Optional; the
    /// default (operator collection names = "tracks"/"platforms"/
    /// "alerts"/"canned_messages") covers most deployments.
    #[cfg(feature = "translator-codec")]
    pub fn set_translator_config(&self, config: crate::translator::TranslationConfig) {
        if let Ok(mut t) = self.ble_translator.write() {
            *t = crate::translator::BleTranslator::new(config);
        }
    }

    /// Translator-frame / reserved-marker dispatcher. Returns `true` if the
    /// first byte of `decrypted` fell in `0xB6..=0xBF` (handled or silently
    /// dropped); the caller MUST stop processing and return `None`. Returns
    /// `false` if the byte was something else; the caller continues with
    /// the existing legacy / delta-document flow.
    ///
    /// `peer` is the BLE peer's identifier (typically a peripheral address).
    /// `source_node` is the peer's `NodeId` when known — used to populate
    /// the `peripheral_id` field on `DecodeInboundCtx`, which the `tracks`
    /// collection's decoder requires per the codec contract. Anonymous
    /// receive paths pass `None`; tracks decoding will then return an
    /// `Err` (logged + dropped) but other collections succeed.
    #[cfg(feature = "translator-codec")]
    fn try_handle_translator_marker(
        &self,
        decrypted: &[u8],
        peer: Option<&str>,
        source_node: Option<NodeId>,
    ) -> TranslatorMarkerOutcome {
        if decrypted.is_empty() {
            return TranslatorMarkerOutcome::NotTranslatorMarker;
        }
        let marker = decrypted[0];

        // Reserved range — silent drop for forward-compat. Receivers running
        // *this* peat-btle release MUST drop unknown future-translator
        // markers without falling through to merge_document, otherwise a
        // 0xB7..=0xBF frame would corrupt the GCounter (ADR-059 Amendment 1
        // §"Backwards compatibility… 2").
        if (TRANSLATOR_RESERVED_MARKER_START..=TRANSLATOR_RESERVED_MARKER_END).contains(&marker) {
            log::warn!(
                "ble: dropping reserved translator-marker frame (marker=0x{marker:02X}, len={})",
                decrypted.len()
            );
            return TranslatorMarkerOutcome::Handled;
        }

        if marker != TRANSLATOR_FRAME_MARKER {
            return TranslatorMarkerOutcome::NotTranslatorMarker;
        }

        // 0xB6 translator dispatch.
        if decrypted.len() < 2 {
            log::warn!(
                "ble: dropping truncated translator frame (len={}, missing collection code)",
                decrypted.len()
            );
            return TranslatorMarkerOutcome::Handled;
        }

        let code = decrypted[1];
        let payload = &decrypted[2..];

        // Resolve the collection name + build a DecodeInboundCtx under
        // the read-lock, drop the lock before invoking decode_inbound_sync.
        let (collection, decode_result) = {
            let translator = match self.ble_translator.read() {
                Ok(g) => g,
                Err(_) => {
                    log::warn!("ble: translator RwLock poisoned; dropping frame");
                    return TranslatorMarkerOutcome::Handled;
                }
            };
            let collection = match translator.code_to_collection(code) {
                Some(c) => c.to_string(),
                None => {
                    log::warn!(
                        "ble: dropping translator frame with unknown collection code 0x{code:02X}"
                    );
                    return TranslatorMarkerOutcome::Handled;
                }
            };

            let ctx = crate::translator::DecodeInboundCtx {
                collection: &collection,
                callsign: None,
                cell_id: None,
                peripheral_id: source_node.map(|n| n.as_u32()),
            };

            let result = translator.decode_inbound_sync(payload, &ctx);
            (collection, result)
        };

        match decode_result {
            Ok(Some(value)) => {
                // ADR-059 Amendment 2: UniFFI JSON callback. Amendment 3:
                // polled `decoded_translator_frame` struct field. Slice 4.b
                // deleted the parallel Rust-trait callback (it took
                // peat_mesh::sync::Document; peat-btle no longer depends
                // on peat-mesh). Suppression rule for
                // `PeatEvent::TranslatorNoCallback` fires only when both
                // remaining sinks are absent — no UniFFI JSON callback,
                // no polled-consumer attestation via
                // `acknowledge_polled_translator_consumer`.
                #[cfg(feature = "uniffi")]
                let json_cb = self
                    .decoded_document_json_callback
                    .read()
                    .ok()
                    .and_then(|g| g.as_ref().cloned());
                #[cfg(not(feature = "uniffi"))]
                let json_cb: Option<()> = None;

                let polled_attested = self
                    .polled_translator_consumer_attested
                    .load(std::sync::atomic::Ordering::Acquire);

                let any_consumer = json_cb.is_some() || polled_attested;

                // Serialize the Value projection to a JSON string. Same
                // wire shape Amendment 2 specifies — peat-mesh's
                // `Translator` impl on the receive side projects the
                // string back into `peat_mesh::sync::Document`; consumers
                // forwarding through peat-ffi pass the string through.
                let doc_json_result = serde_json::to_string(&value);

                #[cfg(feature = "uniffi")]
                if let (Some(json_cb), Ok(ref doc_json)) = (json_cb, &doc_json_result) {
                    json_cb.on_document(
                        collection.clone(),
                        doc_json.clone(),
                        peer.map(str::to_string),
                    );
                }

                if let Err(ref e) = doc_json_result {
                    log::warn!(
                        "ble: failed to serialize decoded {} value to JSON: {}",
                        collection,
                        e
                    );
                }

                if !any_consumer {
                    // No callback installed AND host hasn't attested to
                    // polling. Decoded successfully but nobody's
                    // listening. Per ADR-059 Amendment 1/2/3, surface
                    // through the operator-observable PeatEvent channel
                    // so staged-rollout deployments see, in real time,
                    // how many translator-frame payloads the bridge is
                    // decoding into the void. `log::debug!` alone is
                    // not a monitoring signal — disabled at default
                    // log levels.
                    log::debug!(
                        "ble: decoded {} frame but no consumer registered (no callback, no polled attestation)",
                        collection
                    );
                    self.notify(crate::observer::PeatEvent::TranslatorNoCallback {
                        collection: collection.clone(),
                        peer: peer.map(str::to_string),
                    });
                }

                // Surface the decoded frame to the caller as the
                // polled-consumer output. Caller hoists into
                // `DataReceivedResult.decoded_translator_frame`.
                // Falls through to `Handled` (no frame to surface) if
                // JSON serialization failed above.
                if let Ok(doc_json) = doc_json_result {
                    return TranslatorMarkerOutcome::Decoded(DecodedTranslatorFrame {
                        collection,
                        doc_json,
                        peer: peer.map(str::to_string),
                    });
                }
            }
            Ok(None) => {
                log::debug!(
                    "ble: codec declined translator frame for collection {}",
                    collection
                );
            }
            Err(e) => {
                log::warn!(
                    "ble: translator frame decode error (collection={}): {:#}",
                    collection,
                    e
                );
            }
        }

        TranslatorMarkerOutcome::Handled
    }

    // ==================== Encryption ====================

    /// Check if mesh-wide encryption is enabled
    pub fn is_encryption_enabled(&self) -> bool {
        self.encryption_key.is_some()
    }

    /// Check if strict encryption mode is enabled
    ///
    /// Returns true only if both encryption and strict_encryption are enabled.
    pub fn is_strict_encryption_enabled(&self) -> bool {
        self.config.strict_encryption && self.encryption_key.is_some()
    }

    /// Enable mesh-wide encryption with a shared secret
    ///
    /// Derives a ChaCha20-Poly1305 key from the secret using HKDF-SHA256.
    /// All mesh participants must use the same secret to communicate.
    pub fn enable_encryption(&mut self, secret: &[u8; 32]) {
        self.encryption_key = Some(MeshEncryptionKey::from_shared_secret(
            &self.config.mesh_id,
            secret,
        ));
    }

    /// Disable mesh-wide encryption
    pub fn disable_encryption(&mut self) {
        self.encryption_key = None;
    }

    /// Encrypt document bytes for transmission
    ///
    /// Returns the encrypted bytes with ENCRYPTED_MARKER prefix, or the
    /// original bytes if encryption is disabled.
    fn encrypt_document(&self, plaintext: &[u8]) -> Vec<u8> {
        match &self.encryption_key {
            Some(key) => {
                // Encrypt and prepend marker
                match key.encrypt_to_bytes(plaintext) {
                    Ok(ciphertext) => {
                        let mut buf = Vec::with_capacity(2 + ciphertext.len());
                        buf.push(ENCRYPTED_MARKER);
                        buf.push(0x00); // reserved
                        buf.extend_from_slice(&ciphertext);
                        buf
                    }
                    Err(e) => {
                        log::error!("Encryption failed: {}", e);
                        // Fall back to unencrypted on error (shouldn't happen)
                        plaintext.to_vec()
                    }
                }
            }
            None => plaintext.to_vec(),
        }
    }

    /// Decrypt document bytes received from peer
    ///
    /// Returns the decrypted bytes if encrypted and valid, or the original
    /// bytes if not encrypted. Returns None if decryption fails.
    ///
    /// In strict encryption mode (when both encryption and strict_encryption are enabled),
    /// unencrypted documents are rejected and trigger a SecurityViolation event.
    fn decrypt_document<'a>(
        &self,
        data: &'a [u8],
        source_hint: Option<&str>,
    ) -> Option<std::borrow::Cow<'a, [u8]>> {
        log::debug!(
            "decrypt_document: len={}, first_byte=0x{:02X}, source={:?}",
            data.len(),
            data.first().copied().unwrap_or(0),
            source_hint
        );

        // Check for encrypted marker
        if data.len() >= 2 && data[0] == ENCRYPTED_MARKER {
            // Encrypted document
            let _reserved = data[1];
            let encrypted_payload = &data[2..];

            log::debug!(
                "decrypt_document: encrypted payload len={}, nonce+ciphertext",
                encrypted_payload.len()
            );

            match &self.encryption_key {
                Some(key) => match key.decrypt_from_bytes(encrypted_payload) {
                    Ok(plaintext) => {
                        log::debug!(
                            "decrypt_document: SUCCESS, plaintext len={}",
                            plaintext.len()
                        );
                        Some(std::borrow::Cow::Owned(plaintext))
                    }
                    Err(e) => {
                        log::warn!(
                            "decrypt_document: FAILED (wrong key or corrupted): {} [payload_len={}, source={:?}]",
                            e,
                            encrypted_payload.len(),
                            source_hint
                        );
                        self.notify(PeatEvent::SecurityViolation {
                            kind: SecurityViolationKind::DecryptionFailed,
                            source: source_hint.map(String::from),
                        });
                        None
                    }
                },
                None => {
                    log::warn!(
                        "decrypt_document: encryption not enabled but received encrypted doc"
                    );
                    None
                }
            }
        } else {
            // Unencrypted document
            // Check strict encryption mode
            if self.config.strict_encryption && self.encryption_key.is_some() {
                log::warn!(
                    "Rejected unencrypted document in strict encryption mode (source: {:?})",
                    source_hint
                );
                self.notify(PeatEvent::SecurityViolation {
                    kind: SecurityViolationKind::UnencryptedInStrictMode,
                    source: source_hint.map(String::from),
                });
                None
            } else {
                // Permissive mode: accept unencrypted for backward compatibility
                Some(std::borrow::Cow::Borrowed(data))
            }
        }
    }

    // ==================== Transport Layer API ====================

    /// Decrypt data without parsing (transport-only operation)
    ///
    /// This method provides raw decrypted bytes for apps that want to handle
    /// message parsing themselves (using peat-lite or other libraries).
    ///
    /// # Arguments
    /// * `data` - Potentially encrypted data (0xAE marker indicates encryption)
    ///
    /// # Returns
    /// * `Some(plaintext)` - Decrypted bytes if successful, or original bytes if unencrypted
    /// * `None` - If decryption failed (wrong key, corrupted, or strict mode violation)
    pub fn decrypt_only(&self, data: &[u8]) -> Option<Vec<u8>> {
        self.decrypt_document(data, None)
            .map(|cow| cow.into_owned())
    }

    // ==================== Identity ====================

    /// Check if this mesh has a cryptographic identity
    pub fn has_identity(&self) -> bool {
        self.identity.is_some()
    }

    /// Get this node's public key (if identity is configured)
    pub fn public_key(&self) -> Option<[u8; 32]> {
        self.identity.as_ref().map(|id| id.public_key())
    }

    /// Create an identity attestation for this node
    ///
    /// Returns None if no identity is configured.
    pub fn create_attestation(&self, now_ms: u64) -> Option<IdentityAttestation> {
        self.identity
            .as_ref()
            .map(|id| id.create_attestation(now_ms))
    }

    /// Verify and register a peer's identity attestation
    ///
    /// Implements TOFU (Trust On First Use):
    /// - On first contact, registers the node_id → public_key binding
    /// - On subsequent contacts, verifies the public key matches
    ///
    /// Returns the verification result. Security violations should be handled
    /// by the caller (e.g., disconnect, alert).
    pub fn verify_peer_identity(&self, attestation: &IdentityAttestation) -> RegistryResult {
        self.identity_registry
            .lock()
            .unwrap()
            .verify_or_register(attestation)
    }

    /// Check if a peer's identity is known (has been registered)
    pub fn is_peer_identity_known(&self, node_id: NodeId) -> bool {
        self.identity_registry.lock().unwrap().is_known(node_id)
    }

    /// Get a peer's public key if known
    pub fn peer_public_key(&self, node_id: NodeId) -> Option<[u8; 32]> {
        self.identity_registry
            .lock()
            .unwrap()
            .get_public_key(node_id)
            .copied()
    }

    /// Get the number of known peer identities
    pub fn known_identity_count(&self) -> usize {
        self.identity_registry.lock().unwrap().len()
    }

    /// Pre-register a peer's identity (for out-of-band key exchange)
    ///
    /// Use this when keys are exchanged through a secure side channel
    /// (e.g., QR code, NFC tap, or provisioning server).
    pub fn pre_register_peer_identity(&self, node_id: NodeId, public_key: [u8; 32], now_ms: u64) {
        self.identity_registry
            .lock()
            .unwrap()
            .pre_register(node_id, public_key, now_ms);
    }

    /// Remove a peer's identity from the registry
    ///
    /// Use with caution - this allows re-registration with a different key.
    pub fn forget_peer_identity(&self, node_id: NodeId) {
        self.identity_registry.lock().unwrap().remove(node_id);
    }

    /// Sign arbitrary data with this node's identity
    ///
    /// Returns None if no identity is configured.
    pub fn sign(&self, data: &[u8]) -> Option<[u8; 64]> {
        self.identity.as_ref().map(|id| id.sign(data))
    }

    /// Verify a signature from a peer
    ///
    /// Uses the peer's public key from the identity registry.
    /// Returns false if peer is unknown or signature is invalid.
    pub fn verify_peer_signature(
        &self,
        node_id: NodeId,
        data: &[u8],
        signature: &[u8; 64],
    ) -> bool {
        if let Some(public_key) = self.peer_public_key(node_id) {
            crate::security::verify_signature(&public_key, data, signature)
        } else {
            false
        }
    }

    // ==================== Multi-Hop Relay ====================

    /// Check if multi-hop relay is enabled
    pub fn is_relay_enabled(&self) -> bool {
        self.config.enable_relay
    }

    /// Enable multi-hop relay
    pub fn enable_relay(&mut self) {
        self.config.enable_relay = true;
    }

    /// Disable multi-hop relay
    pub fn disable_relay(&mut self) {
        self.config.enable_relay = false;
    }

    /// Check if a message has been seen before (for deduplication)
    ///
    /// Returns true if the message was already seen (duplicate).
    pub fn has_seen_message(&self, message_id: &MessageId) -> bool {
        self.seen_cache.lock().unwrap().has_seen(message_id)
    }

    /// Mark a message as seen
    ///
    /// Returns true if this is a new message (first time seen).
    pub fn mark_message_seen(&self, message_id: MessageId, origin: NodeId, now_ms: u64) -> bool {
        self.seen_cache
            .lock()
            .unwrap()
            .check_and_mark(message_id, origin, now_ms)
    }

    /// Get the number of entries in the seen message cache
    pub fn seen_cache_size(&self) -> usize {
        self.seen_cache.lock().unwrap().len()
    }

    /// Clear the seen message cache
    pub fn clear_seen_cache(&self) {
        self.seen_cache.lock().unwrap().clear();
    }

    /// Wrap a document in a relay envelope for multi-hop transmission
    ///
    /// The returned bytes can be sent to peers and will be automatically
    /// relayed through the mesh if relay is enabled on receiving nodes.
    pub fn wrap_for_relay(&self, payload: Vec<u8>) -> Vec<u8> {
        let envelope = RelayEnvelope::broadcast(self.config.node_id, payload)
            .with_max_hops(self.config.max_relay_hops);
        envelope.encode()
    }

    /// Get peers to relay a message to
    ///
    /// Uses the configured gossip strategy to select relay targets.
    /// Excludes the source peer (if provided) to avoid sending back to sender.
    pub fn get_relay_targets(&self, exclude_peer: Option<NodeId>) -> Vec<PeatPeer> {
        let connected = self.peer_manager.get_connected_peers();
        let filtered: Vec<_> = if let Some(exclude) = exclude_peer {
            connected
                .into_iter()
                .filter(|p| p.node_id != exclude)
                .collect()
        } else {
            connected
        };

        self.gossip_strategy
            .select_peers(&filtered)
            .into_iter()
            .cloned()
            .collect()
    }

    /// Process an incoming relay envelope
    ///
    /// Handles deduplication, TTL checking, and determines if the message
    /// should be processed and/or relayed.
    ///
    /// Returns:
    /// - `Ok(Some(RelayDecision))` if message should be processed/relayed
    /// - `Ok(None)` if message was a duplicate or TTL expired
    /// - `Err` if parsing failed
    pub fn process_relay_envelope(
        &self,
        data: &[u8],
        source_peer: NodeId,
        now_ms: u64,
    ) -> Option<RelayDecision> {
        // Parse envelope
        let envelope = RelayEnvelope::decode(data)?;

        // Update indirect peer graph if origin differs from source
        // This means the message was relayed through source_peer from origin_node
        if envelope.origin_node != source_peer && envelope.origin_node != self.node_id() {
            let is_new = self.connection_graph.lock().unwrap().on_relay_received(
                source_peer,
                envelope.origin_node,
                envelope.hop_count,
                now_ms,
            );

            if is_new {
                log::debug!(
                    "Discovered indirect peer {:08X} via {:08X} ({} hops)",
                    envelope.origin_node.as_u32(),
                    source_peer.as_u32(),
                    envelope.hop_count
                );
            }
        }

        // Check deduplication
        if !self.mark_message_seen(envelope.message_id, envelope.origin_node, now_ms) {
            // Duplicate message
            let stats = self
                .seen_cache
                .lock()
                .unwrap()
                .get_stats(&envelope.message_id);
            let seen_count = stats.map(|(_, count, _)| count).unwrap_or(1);

            self.notify(PeatEvent::DuplicateMessageDropped {
                origin_node: envelope.origin_node,
                seen_count,
            });

            log::debug!(
                "Dropped duplicate message {} from {:08X} (seen {} times)",
                envelope.message_id,
                envelope.origin_node.as_u32(),
                seen_count
            );
            return None;
        }

        // Check TTL
        if !envelope.can_relay() {
            self.notify(PeatEvent::MessageTtlExpired {
                origin_node: envelope.origin_node,
                hop_count: envelope.hop_count,
            });

            log::debug!(
                "Message {} from {:08X} TTL expired at hop {}",
                envelope.message_id,
                envelope.origin_node.as_u32(),
                envelope.hop_count
            );

            // Still process locally even if TTL expired
            return Some(RelayDecision {
                payload: envelope.payload,
                origin_node: envelope.origin_node,
                hop_count: envelope.hop_count,
                should_relay: false,
                relay_envelope: None,
            });
        }

        // Determine if we should relay
        let should_relay = self.config.enable_relay;
        let relay_envelope = if should_relay {
            envelope.relay() // Increments hop count
        } else {
            None
        };

        Some(RelayDecision {
            payload: envelope.payload,
            origin_node: envelope.origin_node,
            hop_count: envelope.hop_count,
            should_relay,
            relay_envelope,
        })
    }

    /// Build a document wrapped in a relay envelope
    ///
    /// Convenience method that builds the document, encrypts it (if enabled),
    /// and wraps it in a relay envelope for multi-hop transmission.
    pub fn build_relay_document(&self) -> Vec<u8> {
        let doc = self.build_document(); // Already encrypted if encryption enabled
        self.wrap_for_relay(doc)
    }

    // ==================== Delta Sync ====================

    /// Register a peer for delta sync tracking
    ///
    /// Call this when a peer connects to start tracking what data has been
    /// sent to them. This enables future delta sync (sending only changes).
    pub fn register_peer_for_delta(&self, peer_id: &NodeId) {
        let mut encoder = self.delta_encoder.lock().unwrap();
        encoder.add_peer(peer_id);
        log::debug!(
            "Registered peer {:08X} for delta sync tracking",
            peer_id.as_u32()
        );
    }

    /// Unregister a peer from delta sync tracking
    ///
    /// Call this when a peer disconnects to clean up tracking state.
    pub fn unregister_peer_for_delta(&self, peer_id: &NodeId) {
        let mut encoder = self.delta_encoder.lock().unwrap();
        encoder.remove_peer(peer_id);
        log::debug!(
            "Unregistered peer {:08X} from delta sync tracking",
            peer_id.as_u32()
        );
    }

    /// Reset delta sync state for a peer
    ///
    /// Call this when a peer reconnects to force a full sync on next
    /// communication. This clears the record of what was previously sent.
    pub fn reset_peer_delta_state(&self, peer_id: &NodeId) {
        let mut encoder = self.delta_encoder.lock().unwrap();
        encoder.reset_peer(peer_id);
        log::debug!("Reset delta sync state for peer {:08X}", peer_id.as_u32());
    }

    /// Record bytes sent to a peer (for delta statistics)
    pub fn record_delta_sent(&self, peer_id: &NodeId, bytes: usize) {
        let mut encoder = self.delta_encoder.lock().unwrap();
        encoder.record_sent(peer_id, bytes);
    }

    /// Record bytes received from a peer (for delta statistics)
    pub fn record_delta_received(&self, peer_id: &NodeId, bytes: usize, timestamp: u64) {
        let mut encoder = self.delta_encoder.lock().unwrap();
        encoder.record_received(peer_id, bytes, timestamp);
    }

    /// Get delta sync statistics
    ///
    /// Returns aggregate statistics about delta sync across all peers,
    /// including bytes sent/received and sync counts.
    pub fn delta_stats(&self) -> DeltaStats {
        self.delta_encoder.lock().unwrap().stats()
    }

    /// Get delta sync statistics for a specific peer
    ///
    /// Returns the bytes sent/received and sync count for a single peer.
    pub fn peer_delta_stats(&self, peer_id: &NodeId) -> Option<(u64, u64, u32)> {
        let encoder = self.delta_encoder.lock().unwrap();
        encoder
            .get_peer_state(peer_id)
            .map(|state| (state.bytes_sent, state.bytes_received, state.sync_count))
    }

    /// Build a delta document for a specific peer
    ///
    /// This only includes operations that have changed since the last sync
    /// with this peer. Uses the delta encoder to track per-peer state.
    ///
    /// Returns the encoded delta document bytes, or None if there's nothing
    /// new to send to this peer.
    pub fn build_delta_document_for_peer(&self, peer_id: &NodeId, now_ms: u64) -> Option<Vec<u8>> {
        // Collect all current operations
        let mut all_operations: Vec<Operation> = Vec::new();

        // Add counter operations (one per node that has contributed)
        // Use the count value as the "timestamp" for tracking - only send if count increased
        for (node_id_u32, count) in self.document_sync.counter_entries() {
            all_operations.push(Operation::IncrementCounter {
                counter_id: 0, // Default mesh counter
                node_id: NodeId::new(node_id_u32),
                amount: count,
                timestamp: count, // Use count as timestamp for delta tracking
            });
        }

        // Add peripheral update
        // Use event timestamp if available, otherwise use 1 for initial send
        let peripheral = self.document_sync.peripheral_snapshot();
        let peripheral_timestamp = peripheral
            .last_event
            .as_ref()
            .map(|e| e.timestamp)
            .unwrap_or(1); // Use 1 (not 0) so it's sent initially
        all_operations.push(Operation::UpdatePeripheral {
            peripheral,
            timestamp: peripheral_timestamp,
        });

        // Add emergency operations if active
        if let Some(emergency) = self.document_sync.emergency_snapshot() {
            let source_node = NodeId::new(emergency.source_node());
            let timestamp = emergency.timestamp();

            // Add SetEmergency operation
            all_operations.push(Operation::SetEmergency {
                source_node,
                timestamp,
                known_peers: emergency.all_nodes(),
            });

            // Add AckEmergency for each node that has acked
            for acked_node in emergency.acked_nodes() {
                all_operations.push(Operation::AckEmergency {
                    node_id: NodeId::new(acked_node),
                    emergency_timestamp: timestamp,
                });
            }
        }

        // Add app document operations
        for app_op in self.app_document_delta_ops() {
            all_operations.push(Operation::App(app_op));
        }

        // Filter operations for this peer (only send what's new)
        let filtered_operations: Vec<Operation> = {
            let encoder = self.delta_encoder.lock().unwrap();
            if let Some(peer_state) = encoder.get_peer_state(peer_id) {
                all_operations
                    .into_iter()
                    .filter(|op| peer_state.needs_send(&op.key(), op.timestamp()))
                    .collect()
            } else {
                // Unknown peer, send all operations
                all_operations
            }
        };

        // If nothing new to send, return None
        if filtered_operations.is_empty() {
            return None;
        }

        // Mark operations as sent
        {
            let mut encoder = self.delta_encoder.lock().unwrap();
            if let Some(peer_state) = encoder.get_peer_state_mut(peer_id) {
                for op in &filtered_operations {
                    peer_state.mark_sent(&op.key(), op.timestamp());
                }
            }
        }

        // Build the delta document
        let mut delta = DeltaDocument::new(self.config.node_id, now_ms);
        for op in filtered_operations {
            delta.add_operation(op);
        }

        // Encode and optionally encrypt
        let encoded = delta.encode();
        let result = self.encrypt_document(&encoded);

        // Record stats
        {
            let mut encoder = self.delta_encoder.lock().unwrap();
            encoder.record_sent(peer_id, result.len());
        }

        Some(result)
    }

    /// Build a full delta document (for broadcast or new peers)
    ///
    /// Unlike `build_delta_document_for_peer`, this includes all state
    /// regardless of what has been sent before. Use this for broadcasts.
    pub fn build_full_delta_document(&self, now_ms: u64) -> Vec<u8> {
        let mut delta = DeltaDocument::new(self.config.node_id, now_ms);

        // Add all counter operations
        for (node_id_u32, count) in self.document_sync.counter_entries() {
            delta.add_operation(Operation::IncrementCounter {
                counter_id: 0,
                node_id: NodeId::new(node_id_u32),
                amount: count,
                timestamp: now_ms,
            });
        }

        // Add peripheral
        let peripheral = self.document_sync.peripheral_snapshot();
        let peripheral_timestamp = peripheral
            .last_event
            .as_ref()
            .map(|e| e.timestamp)
            .unwrap_or(now_ms);
        delta.add_operation(Operation::UpdatePeripheral {
            peripheral,
            timestamp: peripheral_timestamp,
        });

        // Add emergency if active
        if let Some(emergency) = self.document_sync.emergency_snapshot() {
            let source_node = NodeId::new(emergency.source_node());
            let timestamp = emergency.timestamp();

            delta.add_operation(Operation::SetEmergency {
                source_node,
                timestamp,
                known_peers: emergency.all_nodes(),
            });

            for acked_node in emergency.acked_nodes() {
                delta.add_operation(Operation::AckEmergency {
                    node_id: NodeId::new(acked_node),
                    emergency_timestamp: timestamp,
                });
            }
        }

        // Add app document operations
        for app_op in self.app_document_delta_ops() {
            delta.add_operation(Operation::App(app_op));
        }

        let encoded = delta.encode();
        self.encrypt_document(&encoded)
    }

    /// Internal: Process a received delta document
    ///
    /// Applies operations from a delta document to local state.
    fn process_delta_document_internal(
        &self,
        source_node: NodeId,
        data: &[u8],
        now_ms: u64,
        relay_data: Option<Vec<u8>>,
        origin_node: Option<NodeId>,
        hop_count: u8,
    ) -> Option<DataReceivedResult> {
        // Decode the delta document
        let delta = DeltaDocument::decode(data)?;

        // Don't process our own documents
        if delta.origin_node == self.config.node_id {
            return None;
        }

        // Apply operations to local state
        let mut counter_changed = false;
        let mut emergency_changed = false;
        let mut is_emergency = false;
        let mut is_ack = false;
        let mut event_timestamp = 0u64;
        let mut peer_peripheral: Option<crate::sync::crdt::Peripheral> = None;

        log::info!(
            "Delta document from {:08X}: {} operations, data_len={}",
            delta.origin_node.as_u32(),
            delta.operations.len(),
            data.len()
        );
        for op in &delta.operations {
            log::info!("  Operation: {}", op.key());
            match op {
                Operation::IncrementCounter {
                    node_id, amount, ..
                } => {
                    // Merge counter value (take max)
                    let current = self.document_sync.counter_entries();
                    let current_value = current
                        .iter()
                        .find(|(id, _)| *id == node_id.as_u32())
                        .map(|(_, v)| *v)
                        .unwrap_or(0);

                    if *amount > current_value {
                        // Need to merge - this is handled by the counter merge logic
                        // For now, we record that counter changed
                        counter_changed = true;
                    }
                }
                Operation::UpdatePeripheral {
                    peripheral,
                    timestamp,
                } => {
                    // Store peer peripheral for callsign lookup
                    if let Ok(mut peripherals) = self.peer_peripherals.write() {
                        peripherals.insert(delta.origin_node, peripheral.clone());
                    }
                    // Track the peripheral for the result
                    peer_peripheral = Some(peripheral.clone());
                    // Track the timestamp for the result
                    if *timestamp > event_timestamp {
                        event_timestamp = *timestamp;
                    }
                }
                Operation::SetEmergency { timestamp, .. } => {
                    is_emergency = true;
                    emergency_changed = true;
                    event_timestamp = *timestamp;
                }
                Operation::AckEmergency {
                    emergency_timestamp,
                    ..
                } => {
                    is_ack = true;
                    emergency_changed = true;
                    if *emergency_timestamp > event_timestamp {
                        event_timestamp = *emergency_timestamp;
                    }
                }
                Operation::ClearEmergency {
                    emergency_timestamp,
                } => {
                    emergency_changed = true;
                    if *emergency_timestamp > event_timestamp {
                        event_timestamp = *emergency_timestamp;
                    }
                }
                Operation::App(app_op) => {
                    // Handle app-layer document operation
                    //
                    // The timestamp field may contain version info in the upper bits
                    // (used by delta encoder for change detection). Extract the
                    // original document timestamp from the lower 48 bits.
                    let doc_timestamp = app_op.timestamp & 0x0000_FFFF_FFFF_FFFF;

                    log::info!(
                        "App operation received: type={:02X} op_code={:02X} from {:08X} ts={} payload_len={}",
                        app_op.type_id,
                        app_op.op_code,
                        app_op.source_node,
                        doc_timestamp,
                        app_op.payload.len()
                    );

                    // Try to find/create document and apply the delta operation
                    let doc_key = (app_op.type_id, app_op.source_node, doc_timestamp);
                    let changed = {
                        let mut docs = self.app_documents.write().unwrap();

                        if let Some(existing) = docs.get_mut(&doc_key) {
                            // Apply delta to existing document (merge via CRDT)
                            self.document_registry.apply_delta_op(
                                app_op.type_id,
                                existing.as_mut(),
                                app_op,
                            )
                        } else {
                            // Try to decode from payload (for FULL_STATE operations)
                            if let Some(decoded) = self
                                .document_registry
                                .decode(app_op.type_id, &app_op.payload)
                            {
                                docs.insert(doc_key, decoded);
                                true
                            } else {
                                // For delta ops on unknown documents, we can't apply
                                // The full state will be sent later
                                log::debug!(
                                    "Received delta for unknown doc {:?}, waiting for full state",
                                    doc_key
                                );
                                false
                            }
                        }
                    };

                    // Emit observer event with the real document timestamp
                    self.observers.notify(PeatEvent::app_document_received(
                        app_op.type_id,
                        NodeId::new(app_op.source_node),
                        doc_timestamp,
                        changed,
                    ));
                }
            }
        }

        // Record sync
        self.peer_manager.record_sync(source_node, now_ms);

        // Record delta received
        {
            let mut encoder = self.delta_encoder.lock().unwrap();
            encoder.record_received(&source_node, data.len(), now_ms);
        }

        // Generate events based on what was received
        if is_emergency {
            self.notify(PeatEvent::EmergencyReceived {
                from_node: delta.origin_node,
            });
        } else if is_ack {
            self.notify(PeatEvent::AckReceived {
                from_node: delta.origin_node,
            });
        }

        if counter_changed {
            let total_count = self.document_sync.total_count();
            self.notify(PeatEvent::DocumentSynced {
                from_node: delta.origin_node,
                total_count,
            });
        }

        // Emit relay event if we're relaying
        if relay_data.is_some() {
            let relay_targets = self.get_relay_targets(Some(source_node));
            self.notify(PeatEvent::MessageRelayed {
                origin_node: origin_node.unwrap_or(delta.origin_node),
                relay_count: relay_targets.len(),
                hop_count,
            });
        }

        let PeripheralFields {
            callsign,
            battery_percent,
            heart_rate,
            event_type,
            latitude,
            longitude,
            altitude,
            activity_level,
            alerts,
        } = DataReceivedResult::peripheral_fields(&peer_peripheral);

        Some(DataReceivedResult {
            source_node: delta.origin_node,
            is_emergency,
            is_ack,
            counter_changed,
            emergency_changed,
            total_count: self.document_sync.total_count(),
            event_timestamp,
            relay_data,
            origin_node,
            hop_count,
            callsign,
            battery_percent,
            heart_rate,
            event_type,
            latitude,
            longitude,
            altitude,
            activity_level,
            alerts,
            ..Default::default()
        })
    }

    // ==================== Per-Peer E2EE ====================

    /// Enable per-peer E2EE capability
    ///
    /// Creates a new identity key for this node. This allows establishing
    /// encrypted sessions with specific peers where only the sender and
    /// recipient can read messages (other mesh members cannot).
    pub fn enable_peer_e2ee(&self) {
        let mut sessions = self.peer_sessions.lock().unwrap();
        if sessions.is_none() {
            *sessions = Some(PeerSessionManager::new(self.config.node_id));
            log::info!(
                "Per-peer E2EE enabled for node {:08X}",
                self.config.node_id.as_u32()
            );
        }
    }

    /// Disable per-peer E2EE capability
    ///
    /// Clears all peer sessions and disables E2EE.
    pub fn disable_peer_e2ee(&self) {
        let mut sessions = self.peer_sessions.lock().unwrap();
        *sessions = None;
        log::info!("Per-peer E2EE disabled");
    }

    /// Check if per-peer E2EE is enabled
    pub fn is_peer_e2ee_enabled(&self) -> bool {
        self.peer_sessions.lock().unwrap().is_some()
    }

    /// Get our E2EE public key (for sharing with peers)
    ///
    /// Returns None if per-peer E2EE is not enabled.
    pub fn peer_e2ee_public_key(&self) -> Option<[u8; 32]> {
        self.peer_sessions
            .lock()
            .unwrap()
            .as_ref()
            .map(|s| s.our_public_key())
    }

    /// Initiate E2EE session with a specific peer
    ///
    /// Returns the key exchange message bytes to send to the peer.
    /// The message should be broadcast/sent to the peer.
    /// Returns None if per-peer E2EE is not enabled.
    pub fn initiate_peer_e2ee(&self, peer_node_id: NodeId, now_ms: u64) -> Option<Vec<u8>> {
        let mut sessions = self.peer_sessions.lock().unwrap();
        let session_mgr = sessions.as_mut()?;

        let key_exchange = session_mgr.initiate_session(peer_node_id, now_ms);
        let mut buf = Vec::with_capacity(2 + 37);
        buf.push(KEY_EXCHANGE_MARKER);
        buf.push(0x00); // reserved
        buf.extend_from_slice(&key_exchange.encode());

        log::info!(
            "Initiated E2EE session with peer {:08X}",
            peer_node_id.as_u32()
        );
        Some(buf)
    }

    /// Check if we have an established E2EE session with a peer
    pub fn has_peer_e2ee_session(&self, peer_node_id: NodeId) -> bool {
        self.peer_sessions
            .lock()
            .unwrap()
            .as_ref()
            .is_some_and(|s| s.has_session(peer_node_id))
    }

    /// Get E2EE session state with a peer
    pub fn peer_e2ee_session_state(&self, peer_node_id: NodeId) -> Option<SessionState> {
        self.peer_sessions
            .lock()
            .unwrap()
            .as_ref()
            .and_then(|s| s.session_state(peer_node_id))
    }

    /// Send an E2EE encrypted message to a specific peer
    ///
    /// Returns the encrypted message bytes to send, or None if no session exists.
    /// The message should be sent directly to the peer (not broadcast).
    pub fn send_peer_e2ee(
        &self,
        peer_node_id: NodeId,
        plaintext: &[u8],
        now_ms: u64,
    ) -> Option<Vec<u8>> {
        let mut sessions = self.peer_sessions.lock().unwrap();
        let session_mgr = sessions.as_mut()?;

        match session_mgr.encrypt_for_peer(peer_node_id, plaintext, now_ms) {
            Ok(encrypted) => {
                let mut buf = Vec::with_capacity(2 + encrypted.encode().len());
                buf.push(PEER_E2EE_MARKER);
                buf.push(0x00); // reserved
                buf.extend_from_slice(&encrypted.encode());
                Some(buf)
            }
            Err(e) => {
                log::warn!(
                    "Failed to encrypt for peer {:08X}: {:?}",
                    peer_node_id.as_u32(),
                    e
                );
                None
            }
        }
    }

    /// Close E2EE session with a peer
    pub fn close_peer_e2ee(&self, peer_node_id: NodeId) {
        let mut sessions = self.peer_sessions.lock().unwrap();
        if let Some(session_mgr) = sessions.as_mut() {
            session_mgr.close_session(peer_node_id);
            self.notify(PeatEvent::PeerE2eeClosed { peer_node_id });
            log::info!(
                "Closed E2EE session with peer {:08X}",
                peer_node_id.as_u32()
            );
        }
    }

    /// Get count of active E2EE sessions
    pub fn peer_e2ee_session_count(&self) -> usize {
        self.peer_sessions
            .lock()
            .unwrap()
            .as_ref()
            .map(|s| s.session_count())
            .unwrap_or(0)
    }

    /// Get count of established E2EE sessions
    pub fn peer_e2ee_established_count(&self) -> usize {
        self.peer_sessions
            .lock()
            .unwrap()
            .as_ref()
            .map(|s| s.established_count())
            .unwrap_or(0)
    }

    /// Handle incoming key exchange message
    ///
    /// Called internally when we receive a KEY_EXCHANGE_MARKER message.
    /// Returns the response key exchange bytes to send back, or None if invalid.
    fn handle_key_exchange(&self, data: &[u8], now_ms: u64) -> Option<Vec<u8>> {
        if data.len() < 2 || data[0] != KEY_EXCHANGE_MARKER {
            return None;
        }

        let payload = &data[2..];
        let msg = KeyExchangeMessage::decode(payload)?;

        let mut sessions = self.peer_sessions.lock().unwrap();
        let session_mgr = sessions.as_mut()?;

        let (response, established) = session_mgr.handle_key_exchange(&msg, now_ms)?;

        if established {
            self.notify(PeatEvent::PeerE2eeEstablished {
                peer_node_id: msg.sender_node_id,
            });
            log::info!(
                "E2EE session established with peer {:08X}",
                msg.sender_node_id.as_u32()
            );
        }

        // Return response key exchange
        let mut buf = Vec::with_capacity(2 + 37);
        buf.push(KEY_EXCHANGE_MARKER);
        buf.push(0x00);
        buf.extend_from_slice(&response.encode());
        Some(buf)
    }

    /// Handle incoming E2EE encrypted message
    ///
    /// Called internally when we receive a PEER_E2EE_MARKER message.
    /// Decrypts and notifies observers of the received message.
    fn handle_peer_e2ee_message(&self, data: &[u8], now_ms: u64) -> Option<Vec<u8>> {
        if data.len() < 2 || data[0] != PEER_E2EE_MARKER {
            return None;
        }

        let payload = &data[2..];
        let msg = PeerEncryptedMessage::decode(payload)?;

        let mut sessions = self.peer_sessions.lock().unwrap();
        let session_mgr = sessions.as_mut()?;

        match session_mgr.decrypt_from_peer(&msg, now_ms) {
            Ok(plaintext) => {
                // Notify observers of the decrypted message
                self.notify(PeatEvent::PeerE2eeMessageReceived {
                    from_node: msg.sender_node_id,
                    data: plaintext.clone(),
                });
                Some(plaintext)
            }
            Err(e) => {
                log::warn!(
                    "Failed to decrypt E2EE message from {:08X}: {:?}",
                    msg.sender_node_id.as_u32(),
                    e
                );
                None
            }
        }
    }

    // ==================== Configuration ====================

    /// Get our node ID
    pub fn node_id(&self) -> NodeId {
        self.config.node_id
    }

    /// Get our callsign
    pub fn callsign(&self) -> &str {
        &self.config.callsign
    }

    /// Get the mesh ID
    pub fn mesh_id(&self) -> &str {
        &self.config.mesh_id
    }

    /// Get the device name for BLE advertising
    pub fn device_name(&self) -> String {
        format!(
            "PEAT_{}-{:08X}",
            self.config.mesh_id,
            self.config.node_id.as_u32()
        )
    }

    /// Get a peer's callsign by node ID
    ///
    /// Returns the callsign from the peer's most recently received peripheral data,
    /// or None if no peripheral data has been received from this peer.
    pub fn get_peer_callsign(&self, node_id: NodeId) -> Option<String> {
        self.peer_peripherals.read().ok().and_then(|peripherals| {
            peripherals
                .get(&node_id)
                .map(|p| p.callsign_str().to_string())
        })
    }

    /// Get a peer's full peripheral data by node ID
    ///
    /// Returns a clone of the peripheral data from the peer's most recently received
    /// document, or None if no peripheral data has been received from this peer.
    pub fn get_peer_peripheral(&self, node_id: NodeId) -> Option<Peripheral> {
        self.peer_peripherals
            .read()
            .ok()
            .and_then(|peripherals| peripherals.get(&node_id).cloned())
    }

    // ==================== Document Registry ====================

    /// Get a reference to the document registry.
    ///
    /// Use this to register custom document types for mesh synchronization.
    ///
    /// # Example
    ///
    /// ```ignore
    /// use peat_btle::registry::DocumentType;
    ///
    /// // Register a custom document type
    /// mesh.document_registry().register::<MyCustomDocument>();
    /// ```
    pub fn document_registry(&self) -> &DocumentRegistry {
        &self.document_registry
    }

    /// Store an app-layer document.
    ///
    /// If a document with the same identity (type_id, source_node, timestamp)
    /// already exists, it will be merged using CRDT semantics.
    ///
    /// Returns true if the document was newly added or changed via merge.
    pub fn store_app_document<T: crate::registry::DocumentType>(&self, doc: T) -> bool {
        let type_id = T::TYPE_ID;
        let (source_node, timestamp) = doc.identity();
        let key = (type_id, source_node, timestamp);

        let mut docs = self.app_documents.write().unwrap();

        if let Some(existing) = docs.get_mut(&key) {
            // Merge with existing document
            self.document_registry
                .merge(type_id, existing.as_mut(), &doc)
        } else {
            // Insert new document
            docs.insert(key, Box::new(doc));
            true
        }
    }

    /// Store a type-erased app-layer document.
    ///
    /// Used when receiving documents from the network where the type is
    /// determined at runtime.
    ///
    /// Returns true if the document was newly added or changed via merge.
    pub fn store_app_document_boxed(
        &self,
        type_id: u8,
        source_node: u32,
        timestamp: u64,
        doc: Box<dyn core::any::Any + Send + Sync>,
    ) -> bool {
        let key = (type_id, source_node, timestamp);

        let mut docs = self.app_documents.write().unwrap();

        if let Some(existing) = docs.get_mut(&key) {
            // Merge with existing document
            self.document_registry
                .merge(type_id, existing.as_mut(), doc.as_ref())
        } else {
            // Insert new document
            docs.insert(key, doc);
            true
        }
    }

    /// Get a stored app-layer document by identity.
    ///
    /// Returns None if not found or if downcast to T fails.
    pub fn get_app_document<T: crate::registry::DocumentType>(
        &self,
        source_node: u32,
        timestamp: u64,
    ) -> Option<T> {
        let key = (T::TYPE_ID, source_node, timestamp);

        let docs = self.app_documents.read().unwrap();
        docs.get(&key).and_then(|d| d.downcast_ref::<T>()).cloned()
    }

    /// Get all stored app-layer documents of a specific type.
    ///
    /// Returns a vector of all documents of type T currently stored.
    pub fn get_all_app_documents_of_type<T: crate::registry::DocumentType>(&self) -> Vec<T> {
        let docs = self.app_documents.read().unwrap();
        docs.iter()
            .filter(|((type_id, _, _), _)| *type_id == T::TYPE_ID)
            .filter_map(|(_, doc)| doc.downcast_ref::<T>().cloned())
            .collect()
    }

    /// Get delta operations for all stored app documents.
    ///
    /// Used during sync to include app documents in the delta stream.
    pub fn app_document_delta_ops(&self) -> Vec<crate::registry::AppOperation> {
        let docs = self.app_documents.read().unwrap();
        let mut ops = Vec::new();

        for ((type_id, _source, _ts), doc) in docs.iter() {
            if let Some(op) = self.document_registry.to_delta_op(*type_id, doc.as_ref()) {
                ops.push(op);
            }
        }

        ops
    }

    /// Get all document keys for a given type.
    ///
    /// Returns (source_node, timestamp) pairs for all documents of the specified type.
    pub fn app_document_keys(&self, type_id: u8) -> Vec<(u32, u64)> {
        let docs = self.app_documents.read().unwrap();
        docs.keys()
            .filter(|(tid, _, _)| *tid == type_id)
            .map(|(_, source, ts)| (*source, *ts))
            .collect()
    }

    /// Get the number of stored app documents.
    pub fn app_document_count(&self) -> usize {
        self.app_documents.read().unwrap().len()
    }

    // ==================== Observer Management ====================

    /// Add an observer for mesh events
    pub fn add_observer(&self, observer: Arc<dyn PeatObserver>) {
        self.observers.add(observer);
    }

    /// Remove an observer
    pub fn remove_observer(&self, observer: &Arc<dyn PeatObserver>) {
        self.observers.remove(observer);
    }

    // ==================== User Actions ====================

    /// Send an emergency alert
    ///
    /// Returns the document bytes to broadcast to all peers.
    /// If encryption is enabled, the document is encrypted.
    pub fn send_emergency(&self, timestamp: u64) -> Vec<u8> {
        let data = self.document_sync.send_emergency(timestamp);
        self.notify(PeatEvent::MeshStateChanged {
            peer_count: self.peer_manager.peer_count(),
            connected_count: self.peer_manager.connected_count(),
        });
        self.encrypt_document(&data)
    }

    /// Send an ACK response
    ///
    /// Returns the document bytes to broadcast to all peers.
    /// If encryption is enabled, the document is encrypted.
    pub fn send_ack(&self, timestamp: u64) -> Vec<u8> {
        let data = self.document_sync.send_ack(timestamp);
        self.notify(PeatEvent::MeshStateChanged {
            peer_count: self.peer_manager.peer_count(),
            connected_count: self.peer_manager.connected_count(),
        });
        self.encrypt_document(&data)
    }

    /// Broadcast arbitrary bytes over the mesh.
    ///
    /// Takes raw payload bytes, encrypts them (if encryption is enabled),
    /// and returns bytes ready to send to all connected peers.
    ///
    /// This is useful for sending extension data like CannedMessages from peat-lite.
    pub fn broadcast_bytes(&self, payload: &[u8]) -> Vec<u8> {
        self.encrypt_document(payload)
    }

    /// Clear the current event (emergency or ack)
    pub fn clear_event(&self) {
        self.document_sync.clear_event();
    }

    /// Check if emergency is active
    pub fn is_emergency_active(&self) -> bool {
        self.document_sync.is_emergency_active()
    }

    /// Check if ACK is active
    pub fn is_ack_active(&self) -> bool {
        self.document_sync.is_ack_active()
    }

    /// Get current event type
    pub fn current_event(&self) -> Option<EventType> {
        self.document_sync.current_event()
    }

    // ==================== Emergency Management (Document-Based) ====================

    /// Start a new emergency event with ACK tracking
    ///
    /// Creates an emergency event that tracks ACKs from all known peers.
    /// Pass the list of known peer node IDs to track.
    /// Returns the document bytes to broadcast.
    /// If encryption is enabled, the document is encrypted.
    pub fn start_emergency(&self, timestamp: u64, known_peers: &[u32]) -> Vec<u8> {
        let data = self.document_sync.start_emergency(timestamp, known_peers);
        self.notify(PeatEvent::MeshStateChanged {
            peer_count: self.peer_manager.peer_count(),
            connected_count: self.peer_manager.connected_count(),
        });
        self.encrypt_document(&data)
    }

    /// Start a new emergency using all currently known peers
    ///
    /// Convenience method that automatically includes all discovered peers.
    pub fn start_emergency_with_known_peers(&self, timestamp: u64) -> Vec<u8> {
        let peers: Vec<u32> = self
            .peer_manager
            .get_peers()
            .iter()
            .map(|p| p.node_id.as_u32())
            .collect();
        self.start_emergency(timestamp, &peers)
    }

    /// Record our ACK for the current emergency
    ///
    /// Returns the document bytes to broadcast, or None if no emergency is active.
    /// If encryption is enabled, the document is encrypted.
    pub fn ack_emergency(&self, timestamp: u64) -> Option<Vec<u8>> {
        let result = self.document_sync.ack_emergency(timestamp);
        if result.is_some() {
            self.notify(PeatEvent::MeshStateChanged {
                peer_count: self.peer_manager.peer_count(),
                connected_count: self.peer_manager.connected_count(),
            });
        }
        result.map(|data| self.encrypt_document(&data))
    }

    /// Clear the current emergency event
    pub fn clear_emergency(&self) {
        self.document_sync.clear_emergency();
    }

    /// Check if there's an active emergency
    pub fn has_active_emergency(&self) -> bool {
        self.document_sync.has_active_emergency()
    }

    /// Get emergency status info
    ///
    /// Returns (source_node, timestamp, acked_count, pending_count) if emergency is active.
    pub fn get_emergency_status(&self) -> Option<(u32, u64, usize, usize)> {
        self.document_sync.get_emergency_status()
    }

    /// Check if a specific peer has ACKed the current emergency
    pub fn has_peer_acked(&self, peer_id: u32) -> bool {
        self.document_sync.has_peer_acked(peer_id)
    }

    /// Check if all peers have ACKed the current emergency
    pub fn all_peers_acked(&self) -> bool {
        self.document_sync.all_peers_acked()
    }

    // ==================== Chat Methods (requires `legacy-chat` feature) ====================

    /// Send a chat message
    ///
    /// Adds the message to the local CRDT and returns the document bytes
    /// to broadcast to all peers. If encryption is enabled, the document is encrypted.
    ///
    /// Returns the encrypted document bytes if the message was new,
    /// or None if it was a duplicate.
    #[cfg(feature = "legacy-chat")]
    pub fn send_chat(&self, sender: &str, text: &str, timestamp: u64) -> Option<Vec<u8>> {
        if self.document_sync.add_chat_message(sender, text, timestamp) {
            Some(self.encrypt_document(&self.build_document()))
        } else {
            None
        }
    }

    /// Send a chat reply
    ///
    /// Adds the reply to the local CRDT with reply-to information and returns
    /// the document bytes to broadcast. If encryption is enabled, the document is encrypted.
    ///
    /// Returns the encrypted document bytes if the message was new,
    /// or None if it was a duplicate.
    #[cfg(feature = "legacy-chat")]
    pub fn send_chat_reply(
        &self,
        sender: &str,
        text: &str,
        reply_to_node: u32,
        reply_to_timestamp: u64,
        timestamp: u64,
    ) -> Option<Vec<u8>> {
        if self.document_sync.add_chat_reply(
            sender,
            text,
            reply_to_node,
            reply_to_timestamp,
            timestamp,
        ) {
            Some(self.encrypt_document(&self.build_document()))
        } else {
            None
        }
    }

    /// Get the number of chat messages in the local CRDT
    #[cfg(feature = "legacy-chat")]
    pub fn chat_count(&self) -> usize {
        self.document_sync.chat_count()
    }

    /// Get chat messages newer than a timestamp
    ///
    /// Returns a vector of (origin_node, timestamp, sender, text, reply_to_node, reply_to_timestamp) tuples.
    #[cfg(feature = "legacy-chat")]
    pub fn chat_messages_since(
        &self,
        since_timestamp: u64,
    ) -> Vec<(u32, u64, String, String, u32, u64)> {
        self.document_sync.chat_messages_since(since_timestamp)
    }

    /// Get all chat messages
    ///
    /// Returns a vector of (origin_node, timestamp, sender, text, reply_to_node, reply_to_timestamp) tuples.
    #[cfg(feature = "legacy-chat")]
    pub fn all_chat_messages(&self) -> Vec<(u32, u64, String, String, u32, u64)> {
        self.document_sync.all_chat_messages()
    }

    // ==================== BLE Callbacks (Platform -> Mesh) ====================

    /// Called when a BLE device is discovered
    ///
    /// Returns `Some(PeatPeer)` if this is a new Peat peer on our mesh.
    pub fn on_ble_discovered(
        &self,
        identifier: &str,
        name: Option<&str>,
        rssi: i8,
        mesh_id: Option<&str>,
        now_ms: u64,
    ) -> Option<PeatPeer> {
        let (node_id, is_new) = self
            .peer_manager
            .on_discovered(identifier, name, rssi, mesh_id, now_ms)?;

        let peer = self.peer_manager.get_peer(node_id)?;

        // Update connection graph
        {
            let mut graph = self.connection_graph.lock().unwrap();
            graph.on_discovered(
                node_id,
                identifier.to_string(),
                name.map(|s| s.to_string()),
                mesh_id.map(|s| s.to_string()),
                rssi,
                now_ms,
            );
        }

        if is_new {
            self.notify(PeatEvent::PeerDiscovered { peer: peer.clone() });
            self.notify_mesh_state_changed();
        }

        Some(peer)
    }

    /// Called when a BLE connection is established (outgoing)
    ///
    /// Returns the NodeId if this identifier is known.
    pub fn on_ble_connected(&self, identifier: &str, now_ms: u64) -> Option<NodeId> {
        let node_id = match self.peer_manager.on_connected(identifier, now_ms) {
            Some(id) => id,
            None => {
                log::warn!(
                    "on_ble_connected: identifier {:?} not in peer map — \
                     use on_incoming_connection() for peripheral connections",
                    identifier
                );
                return None;
            }
        };

        // Update connection graph
        {
            let mut graph = self.connection_graph.lock().unwrap();
            graph.on_connected(node_id, now_ms);
        }

        // Register peer for delta sync tracking
        self.register_peer_for_delta(&node_id);

        self.notify(PeatEvent::PeerConnected { node_id });
        self.notify_mesh_state_changed();
        Some(node_id)
    }

    /// Called when a BLE connection is lost
    pub fn on_ble_disconnected(
        &self,
        identifier: &str,
        reason: DisconnectReason,
    ) -> Option<NodeId> {
        let (node_id, observer_reason) = self.peer_manager.on_disconnected(identifier, reason)?;

        // Update connection graph (convert observer reason to platform reason)
        {
            let mut graph = self.connection_graph.lock().unwrap();
            let platform_reason = match observer_reason {
                DisconnectReason::LocalRequest => crate::platform::DisconnectReason::LocalRequest,
                DisconnectReason::RemoteRequest => crate::platform::DisconnectReason::RemoteRequest,
                DisconnectReason::Timeout => crate::platform::DisconnectReason::Timeout,
                DisconnectReason::LinkLoss => crate::platform::DisconnectReason::LinkLoss,
                DisconnectReason::ConnectionFailed => {
                    crate::platform::DisconnectReason::ConnectionFailed
                }
                DisconnectReason::Unknown => crate::platform::DisconnectReason::Unknown,
            };
            let now_ms = std::time::SystemTime::now()
                .duration_since(std::time::UNIX_EPOCH)
                .map(|d| d.as_millis() as u64)
                .unwrap_or(0);
            graph.on_disconnected(node_id, platform_reason, now_ms);

            // Remove indirect peer paths that went through this peer
            // These paths may no longer be valid since the direct connection is lost
            graph.remove_via_peer(node_id);
        }

        // Unregister peer from delta sync tracking
        self.unregister_peer_for_delta(&node_id);

        self.notify(PeatEvent::PeerDisconnected {
            node_id,
            reason: observer_reason,
        });
        self.notify_mesh_state_changed();
        Some(node_id)
    }

    /// Called when a BLE connection is lost, using NodeId directly
    ///
    /// Alternative to on_ble_disconnected() when only NodeId is known (e.g., ESP32).
    pub fn on_peer_disconnected(&self, node_id: NodeId, reason: DisconnectReason) {
        if self
            .peer_manager
            .on_disconnected_by_node_id(node_id, reason)
        {
            // Update connection graph
            {
                let mut graph = self.connection_graph.lock().unwrap();
                let platform_reason = match reason {
                    DisconnectReason::LocalRequest => {
                        crate::platform::DisconnectReason::LocalRequest
                    }
                    DisconnectReason::RemoteRequest => {
                        crate::platform::DisconnectReason::RemoteRequest
                    }
                    DisconnectReason::Timeout => crate::platform::DisconnectReason::Timeout,
                    DisconnectReason::LinkLoss => crate::platform::DisconnectReason::LinkLoss,
                    DisconnectReason::ConnectionFailed => {
                        crate::platform::DisconnectReason::ConnectionFailed
                    }
                    DisconnectReason::Unknown => crate::platform::DisconnectReason::Unknown,
                };
                let now_ms = std::time::SystemTime::now()
                    .duration_since(std::time::UNIX_EPOCH)
                    .map(|d| d.as_millis() as u64)
                    .unwrap_or(0);
                graph.on_disconnected(node_id, platform_reason, now_ms);

                // Remove indirect peer paths that went through this peer
                graph.remove_via_peer(node_id);
            }

            // Unregister peer from delta sync tracking
            self.unregister_peer_for_delta(&node_id);

            self.notify(PeatEvent::PeerDisconnected { node_id, reason });
            self.notify_mesh_state_changed();
        }
    }

    /// Called when a remote device connects to us (incoming connection)
    ///
    /// Use this when we're acting as a peripheral and a central connects to us.
    pub fn on_incoming_connection(&self, identifier: &str, node_id: NodeId, now_ms: u64) -> bool {
        let is_new = self
            .peer_manager
            .on_incoming_connection(identifier, node_id, now_ms);

        // Update connection graph
        {
            let mut graph = self.connection_graph.lock().unwrap();
            if is_new {
                graph.on_discovered(
                    node_id,
                    identifier.to_string(),
                    None,
                    Some(self.config.mesh_id.clone()),
                    -50, // Default good RSSI for incoming connections
                    now_ms,
                );
            }
            graph.on_connected(node_id, now_ms);
        }

        // Register peer for delta sync tracking
        self.register_peer_for_delta(&node_id);

        if is_new {
            if let Some(peer) = self.peer_manager.get_peer(node_id) {
                self.notify(PeatEvent::PeerDiscovered { peer });
            }
        }

        self.notify(PeatEvent::PeerConnected { node_id });
        self.notify_mesh_state_changed();

        is_new
    }

    /// Called when data is received from a peer
    ///
    /// Parses the document, merges it, and generates appropriate events.
    /// If encryption is enabled, decrypts the document first.
    /// Handles per-peer E2EE messages (KEY_EXCHANGE and PEER_E2EE markers).
    /// Returns the source NodeId and whether the document contained an event.
    pub fn on_ble_data_received(
        &self,
        identifier: &str,
        data: &[u8],
        now_ms: u64,
    ) -> Option<DataReceivedResult> {
        // Get node ID from identifier
        let node_id = self.peer_manager.get_node_id(identifier)?;

        // Check for special message types first
        if data.len() >= 2 {
            match data[0] {
                KEY_EXCHANGE_MARKER => {
                    // Handle key exchange - returns response to send back
                    let _response = self.handle_key_exchange(data, now_ms);
                    // Return None as this isn't a document sync
                    return None;
                }
                PEER_E2EE_MARKER => {
                    // Handle encrypted peer message
                    let _plaintext = self.handle_peer_e2ee_message(data, now_ms);
                    // Return None as this isn't a document sync
                    return None;
                }
                RELAY_ENVELOPE_MARKER => {
                    // Handle relay envelope for multi-hop
                    return self
                        .handle_relay_envelope_with_identifier(node_id, identifier, data, now_ms);
                }
                _ => {}
            }
        }

        // Direct document (not relay envelope)
        self.process_document_data_with_identifier(node_id, identifier, data, now_ms, None, None, 0)
    }

    /// Internal: Process document data with identifier as source hint
    #[allow(clippy::too_many_arguments)]
    fn process_document_data_with_identifier(
        &self,
        source_node: NodeId,
        identifier: &str,
        data: &[u8],
        now_ms: u64,
        relay_data: Option<Vec<u8>>,
        origin_node: Option<NodeId>,
        hop_count: u8,
    ) -> Option<DataReceivedResult> {
        // Decrypt if encrypted (mesh-wide encryption) - use identifier as source hint
        let decrypted = self.decrypt_document(data, Some(identifier))?;

        // Universal Document transport (peat-lite envelope, magic
        // prefix b"PEAT" / 0x50). Checked BEFORE 0xB6 translator
        // dispatch so a peat-lite frame doesn't get mis-decoded as a
        // translator frame (the magic byte 0x50 is outside the
        // 0xB6-0xBF translator range, but the routing fall-through
        // would otherwise reach delta-document handling and fail
        // there). Population is feature-gated; the field
        // `peat_lite_document` on `DataReceivedResult` is in the
        // binding shape unconditionally.
        #[cfg(feature = "peat-lite-frame")]
        match crate::peat_lite_frame::try_handle_peat_lite_frame(&decrypted) {
            crate::peat_lite_frame::PeatLiteFrameOutcome::NotPeatLiteFrame => {}
            crate::peat_lite_frame::PeatLiteFrameOutcome::Handled => return None,
            crate::peat_lite_frame::PeatLiteFrameOutcome::Decoded(frame) => {
                return Some(DataReceivedResult {
                    source_node,
                    peat_lite_document: Some(frame),
                    ..Default::default()
                });
            }
        }

        // ADR-059 Amendment 1 §"Receive-side dispatch": route 0xB6 frames
        // through BleTranslator + silent-drop reserved 0xB7..=0xBF.
        // Amendment 3: surface decoded frames via the polled
        // `decoded_translator_frame` field on the returned result so
        // hosts (peat-atak-plugin et al.) don't depend on the JNA-broken
        // UniFFI callback path in ATAK.
        #[cfg(feature = "translator-codec")]
        match self.try_handle_translator_marker(&decrypted, Some(identifier), Some(source_node)) {
            TranslatorMarkerOutcome::NotTranslatorMarker => {}
            TranslatorMarkerOutcome::Handled => return None,
            TranslatorMarkerOutcome::Decoded(frame) => {
                return Some(DataReceivedResult::translator_frame(source_node, frame));
            }
        }

        // Check if this is a delta document (wire format v2)
        if DeltaDocument::is_delta_document(&decrypted) {
            return self.process_delta_document_internal(
                source_node,
                &decrypted,
                now_ms,
                relay_data,
                origin_node,
                hop_count,
            );
        }

        // Merge the document (legacy wire format v1)
        let result = self.document_sync.merge_document(&decrypted)?;

        // Store peer peripheral if present (for callsign lookup)
        if let Some(ref peripheral) = result.peer_peripheral {
            if let Ok(mut peripherals) = self.peer_peripherals.write() {
                peripherals.insert(result.source_node, peripheral.clone());
            }
        }

        // Record sync
        self.peer_manager.record_sync(source_node, now_ms);

        // Generate events based on what was received
        if result.is_emergency() {
            self.notify(PeatEvent::EmergencyReceived {
                from_node: result.source_node,
            });
        } else if result.is_ack() {
            self.notify(PeatEvent::AckReceived {
                from_node: result.source_node,
            });
        }

        if result.counter_changed {
            self.notify(PeatEvent::DocumentSynced {
                from_node: result.source_node,
                total_count: result.total_count,
            });
        }

        // Emit relay event if we're relaying
        if relay_data.is_some() {
            let relay_targets = self.get_relay_targets(Some(source_node));
            self.notify(PeatEvent::MessageRelayed {
                origin_node: origin_node.unwrap_or(result.source_node),
                relay_count: relay_targets.len(),
                hop_count,
            });
        }

        let PeripheralFields {
            callsign,
            battery_percent,
            heart_rate,
            event_type,
            latitude,
            longitude,
            altitude,
            activity_level,
            alerts,
        } = DataReceivedResult::peripheral_fields(&result.peer_peripheral);

        Some(DataReceivedResult {
            source_node: result.source_node,
            is_emergency: result.is_emergency(),
            is_ack: result.is_ack(),
            counter_changed: result.counter_changed,
            emergency_changed: result.emergency_changed,
            total_count: result.total_count,
            event_timestamp: result.event.as_ref().map(|e| e.timestamp).unwrap_or(0),
            relay_data,
            origin_node,
            hop_count,
            callsign,
            battery_percent,
            heart_rate,
            event_type,
            latitude,
            longitude,
            altitude,
            activity_level,
            alerts,
            ..Default::default()
        })
    }

    /// Internal: Handle relay envelope with identifier as source hint
    fn handle_relay_envelope_with_identifier(
        &self,
        source_node: NodeId,
        identifier: &str,
        data: &[u8],
        now_ms: u64,
    ) -> Option<DataReceivedResult> {
        // Process the relay envelope
        let envelope = RelayEnvelope::decode(data)?;

        // Check deduplication
        if !self.mark_message_seen(envelope.message_id, envelope.origin_node, now_ms) {
            let stats = self
                .seen_cache
                .lock()
                .unwrap()
                .get_stats(&envelope.message_id);
            let seen_count = stats.map(|(_, count, _)| count).unwrap_or(1);

            self.notify(PeatEvent::DuplicateMessageDropped {
                origin_node: envelope.origin_node,
                seen_count,
            });
            return None;
        }

        // Check TTL and get relay data
        let relay_data = if envelope.can_relay() && self.config.enable_relay {
            envelope.relay().map(|e| e.encode())
        } else {
            if !envelope.can_relay() {
                self.notify(PeatEvent::MessageTtlExpired {
                    origin_node: envelope.origin_node,
                    hop_count: envelope.hop_count,
                });
            }
            None
        };

        // Process the inner payload
        self.process_document_data_with_identifier(
            source_node,
            identifier,
            &envelope.payload,
            now_ms,
            relay_data,
            Some(envelope.origin_node),
            envelope.hop_count,
        )
    }

    /// Called when data is received but we don't have the identifier mapped
    ///
    /// Use this when receiving data from a peripheral we discovered.
    /// If encryption is enabled, decrypts the document first.
    /// Handles per-peer E2EE messages (KEY_EXCHANGE and PEER_E2EE markers).
    /// Handles relay envelopes for multi-hop mesh operation.
    pub fn on_ble_data_received_from_node(
        &self,
        node_id: NodeId,
        data: &[u8],
        now_ms: u64,
    ) -> Option<DataReceivedResult> {
        // Check for special message types first
        if data.len() >= 2 {
            match data[0] {
                KEY_EXCHANGE_MARKER => {
                    let _response = self.handle_key_exchange(data, now_ms);
                    return None;
                }
                PEER_E2EE_MARKER => {
                    let _plaintext = self.handle_peer_e2ee_message(data, now_ms);
                    return None;
                }
                RELAY_ENVELOPE_MARKER => {
                    // Handle relay envelope for multi-hop
                    return self.handle_relay_envelope(node_id, data, now_ms);
                }
                _ => {}
            }
        }

        // Direct document (not relay envelope)
        self.process_document_data(node_id, data, now_ms, None, None, 0)
    }

    /// Called when encrypted data is received from an unknown peer
    ///
    /// This handles the case where we receive an encrypted document from a BLE address
    /// that isn't registered in our peer manager (e.g., due to BLE address rotation).
    /// The function decrypts first using the mesh key, then extracts the source_node
    /// from the decrypted document header and registers the peer.
    ///
    /// Returns `Some(DataReceivedResult)` if decryption and processing succeed.
    /// Returns `None` if decryption fails or the document is invalid.
    pub fn on_ble_data_received_anonymous(
        &self,
        identifier: &str,
        data: &[u8],
        now_ms: u64,
    ) -> Option<DataReceivedResult> {
        log::debug!(
            "on_ble_data_received_anonymous: identifier={}, len={}, marker=0x{:02X}",
            identifier,
            data.len(),
            data.first().copied().unwrap_or(0)
        );

        // Try to decrypt (handles both encrypted and unencrypted documents)
        let decrypted = match self.decrypt_document(data, Some(identifier)) {
            Some(d) => d,
            None => {
                log::warn!(
                    "on_ble_data_received_anonymous: decrypt/parse FAILED for {} byte doc from {}",
                    data.len(),
                    identifier
                );
                return None;
            }
        };

        // Universal Document transport (peat-lite envelope) — checked
        // before 0xB6 translator dispatch. Anonymous-path: the
        // peat-lite header itself carries source_node_id, so we can
        // surface it on the returned result even though the outer
        // identifier didn't yield a NodeId.
        #[cfg(feature = "peat-lite-frame")]
        match crate::peat_lite_frame::try_handle_peat_lite_frame(&decrypted) {
            crate::peat_lite_frame::PeatLiteFrameOutcome::NotPeatLiteFrame => {}
            crate::peat_lite_frame::PeatLiteFrameOutcome::Handled => return None,
            crate::peat_lite_frame::PeatLiteFrameOutcome::Decoded(frame) => {
                let source = NodeId::new(frame.source_node_id);
                return Some(DataReceivedResult {
                    source_node: source,
                    peat_lite_document: Some(frame),
                    ..Default::default()
                });
            }
        }

        // ADR-059 Amendment 1 §"Receive-side dispatch": route 0xB6
        // translator frames through BleTranslator::decode_inbound, and
        // silent-drop reserved 0xB7..=0xBF frames before they reach
        // PeatDocument::decode (where a 0xB6+ payload could hit the
        // GCounter-pollution hazard for inputs whose bytes 8..11 fall
        // <= MAX_COUNTER_ENTRIES). Anonymous receive: source_node is
        // unknown until the legacy path extracts it; for translator
        // frames there's no PeatDocument header to extract from, so we
        // pass None and accept that `tracks` decoding will Err for lack
        // of `local_wire_id` (other collections succeed).
        #[cfg(feature = "translator-codec")]
        match self.try_handle_translator_marker(&decrypted, Some(identifier), None) {
            TranslatorMarkerOutcome::NotTranslatorMarker => {}
            TranslatorMarkerOutcome::Handled => return None,
            TranslatorMarkerOutcome::Decoded(frame) => {
                // Anonymous path has no source_node yet — translator
                // frames carry no PeatDocument header. Use a placeholder
                // NodeId; the meaningful payload rides
                // `decoded_translator_frame` per Amendment 3.
                return Some(DataReceivedResult::translator_frame(NodeId::new(0), frame));
            }
        }

        // Extract source_node from decrypted document header
        // Header format: [version: 4 bytes (LE)][node_id: 4 bytes (LE)]
        if decrypted.len() < 8 {
            log::warn!("Decrypted document too short to extract source_node");
            return None;
        }

        let source_node_u32 =
            u32::from_le_bytes([decrypted[4], decrypted[5], decrypted[6], decrypted[7]]);
        let source_node = NodeId::new(source_node_u32);

        log::info!(
            "Anonymous document from {}: source_node={:08X}, len={}",
            identifier,
            source_node_u32,
            decrypted.len()
        );

        // Register the peer with this identifier so future lookups work
        // This handles BLE address rotation
        self.peer_manager
            .register_identifier(identifier, source_node);

        // Check if this is a delta document
        let is_delta = DeltaDocument::is_delta_document(&decrypted);
        log::info!(
            "Document format: delta={}, first_byte=0x{:02X}, len={}",
            is_delta,
            decrypted.first().copied().unwrap_or(0),
            decrypted.len()
        );

        if is_delta {
            return self.process_delta_document_internal(
                source_node,
                &decrypted,
                now_ms,
                None,
                None,
                0,
            );
        }

        // Handle app-layer message (0xAF marker from peat-lite CannedMessages)
        // These are handled by consumers who register DocumentType implementations
        // via the DocumentRegistry. Pass through as relay_data for platform layer.
        const APP_LAYER_MARKER: u8 = 0xAF;
        if decrypted.first().copied() == Some(APP_LAYER_MARKER) {
            log::debug!(
                "App-layer message (0xAF) from {:08X}, {} bytes - passing to relay",
                source_node.as_u32(),
                decrypted.len()
            );
            return Some(DataReceivedResult {
                source_node,
                is_emergency: false,
                is_ack: false,
                counter_changed: false,
                emergency_changed: false,
                total_count: 0,
                event_timestamp: now_ms,
                relay_data: Some(decrypted.to_vec()),
                origin_node: None,
                hop_count: 0,
                callsign: None,
                battery_percent: None,
                heart_rate: None,
                event_type: None,
                latitude: None,
                longitude: None,
                altitude: None,
                ..Default::default()
            });
        }

        // Merge the document (legacy wire format v1)
        log::info!(
            "Processing legacy document from {:08X}",
            source_node.as_u32()
        );
        let result = self.document_sync.merge_document(&decrypted)?;

        // Log what we got from the merge
        log::info!(
            "Merge result: peer_peripheral={}, counter_changed={}",
            result.peer_peripheral.is_some(),
            result.counter_changed
        );
        if let Some(ref p) = result.peer_peripheral {
            log::info!("Peripheral callsign: '{}'", p.callsign_str());
        }

        // Record sync
        self.peer_manager.record_sync(source_node, now_ms);

        // Generate events
        if result.is_emergency() {
            self.notify(PeatEvent::EmergencyReceived {
                from_node: result.source_node,
            });
        } else if result.is_ack() {
            self.notify(PeatEvent::AckReceived {
                from_node: result.source_node,
            });
        }

        if result.counter_changed {
            self.notify(PeatEvent::DocumentSynced {
                from_node: result.source_node,
                total_count: result.total_count,
            });
        }

        let PeripheralFields {
            callsign,
            battery_percent,
            heart_rate,
            event_type,
            latitude,
            longitude,
            altitude,
            activity_level,
            alerts,
        } = DataReceivedResult::peripheral_fields(&result.peer_peripheral);

        Some(DataReceivedResult {
            source_node: result.source_node,
            is_emergency: result.is_emergency(),
            is_ack: result.is_ack(),
            counter_changed: result.counter_changed,
            emergency_changed: result.emergency_changed,
            total_count: result.total_count,
            event_timestamp: result.event.as_ref().map(|e| e.timestamp).unwrap_or(0),
            relay_data: None,
            origin_node: None,
            hop_count: 0,
            callsign,
            battery_percent,
            heart_rate,
            event_type,
            latitude,
            longitude,
            altitude,
            activity_level,
            alerts,
            ..Default::default()
        })
    }

    /// Internal: Process document data (shared by direct and relay paths)
    fn process_document_data(
        &self,
        source_node: NodeId,
        data: &[u8],
        now_ms: u64,
        relay_data: Option<Vec<u8>>,
        origin_node: Option<NodeId>,
        hop_count: u8,
    ) -> Option<DataReceivedResult> {
        // Decrypt if encrypted (mesh-wide encryption)
        let source_hint = format!("node:{:08X}", source_node.as_u32());
        let decrypted = self.decrypt_document(data, Some(&source_hint))?;

        // Universal Document transport (peat-lite envelope) — checked
        // before 0xB6 translator dispatch.
        #[cfg(feature = "peat-lite-frame")]
        match crate::peat_lite_frame::try_handle_peat_lite_frame(&decrypted) {
            crate::peat_lite_frame::PeatLiteFrameOutcome::NotPeatLiteFrame => {}
            crate::peat_lite_frame::PeatLiteFrameOutcome::Handled => return None,
            crate::peat_lite_frame::PeatLiteFrameOutcome::Decoded(frame) => {
                return Some(DataReceivedResult {
                    source_node,
                    relay_data,
                    origin_node,
                    hop_count,
                    peat_lite_document: Some(frame),
                    ..Default::default()
                });
            }
        }

        // ADR-059 Amendment 1 §"Receive-side dispatch": route 0xB6 frames
        // through BleTranslator + silent-drop reserved 0xB7..=0xBF.
        // Amendment 3: surface decoded frames via `decoded_translator_frame`.
        #[cfg(feature = "translator-codec")]
        match self.try_handle_translator_marker(&decrypted, None, Some(source_node)) {
            TranslatorMarkerOutcome::NotTranslatorMarker => {}
            TranslatorMarkerOutcome::Handled => return None,
            TranslatorMarkerOutcome::Decoded(frame) => {
                return Some(DataReceivedResult::translator_frame(source_node, frame));
            }
        }

        // Check if this is a delta document (wire format v2)
        if DeltaDocument::is_delta_document(&decrypted) {
            return self.process_delta_document_internal(
                source_node,
                &decrypted,
                now_ms,
                relay_data,
                origin_node,
                hop_count,
            );
        }

        // Merge the document (legacy wire format v1)
        let result = self.document_sync.merge_document(&decrypted)?;

        // Store peer peripheral if present (for callsign lookup)
        if let Some(ref peripheral) = result.peer_peripheral {
            if let Ok(mut peripherals) = self.peer_peripherals.write() {
                peripherals.insert(result.source_node, peripheral.clone());
            }
        }

        // Record sync
        self.peer_manager.record_sync(source_node, now_ms);

        // Generate events based on what was received
        if result.is_emergency() {
            self.notify(PeatEvent::EmergencyReceived {
                from_node: result.source_node,
            });
        } else if result.is_ack() {
            self.notify(PeatEvent::AckReceived {
                from_node: result.source_node,
            });
        }

        if result.counter_changed {
            self.notify(PeatEvent::DocumentSynced {
                from_node: result.source_node,
                total_count: result.total_count,
            });
        }

        // Emit relay event if we're relaying
        if relay_data.is_some() {
            let relay_targets = self.get_relay_targets(Some(source_node));
            self.notify(PeatEvent::MessageRelayed {
                origin_node: origin_node.unwrap_or(result.source_node),
                relay_count: relay_targets.len(),
                hop_count,
            });
        }

        let PeripheralFields {
            callsign,
            battery_percent,
            heart_rate,
            event_type,
            latitude,
            longitude,
            altitude,
            activity_level,
            alerts,
        } = DataReceivedResult::peripheral_fields(&result.peer_peripheral);

        Some(DataReceivedResult {
            source_node: result.source_node,
            is_emergency: result.is_emergency(),
            is_ack: result.is_ack(),
            counter_changed: result.counter_changed,
            emergency_changed: result.emergency_changed,
            total_count: result.total_count,
            event_timestamp: result.event.as_ref().map(|e| e.timestamp).unwrap_or(0),
            relay_data,
            origin_node,
            hop_count,
            callsign,
            battery_percent,
            heart_rate,
            event_type,
            latitude,
            longitude,
            altitude,
            activity_level,
            alerts,
            ..Default::default()
        })
    }

    /// Internal: Handle relay envelope
    fn handle_relay_envelope(
        &self,
        source_node: NodeId,
        data: &[u8],
        now_ms: u64,
    ) -> Option<DataReceivedResult> {
        // Process the relay envelope
        let decision = self.process_relay_envelope(data, source_node, now_ms)?;

        // Get relay data if we should relay
        let relay_data = if decision.should_relay {
            decision.relay_data()
        } else {
            None
        };

        // Process the inner payload
        self.process_document_data(
            source_node,
            &decision.payload,
            now_ms,
            relay_data,
            Some(decision.origin_node),
            decision.hop_count,
        )
    }

    /// Called when data is received without a known identifier
    ///
    /// This is the simplest data receive method - it extracts the source node_id
    /// from the document itself. Use this when you don't track identifiers
    /// (e.g., ESP32 NimBLE).
    /// If encryption is enabled, decrypts the document first.
    /// Handles per-peer E2EE messages (KEY_EXCHANGE and PEER_E2EE markers).
    /// Handles relay envelopes for multi-hop mesh operation.
    pub fn on_ble_data(
        &self,
        identifier: &str,
        data: &[u8],
        now_ms: u64,
    ) -> Option<DataReceivedResult> {
        // Check for special message types first
        if data.len() >= 2 {
            match data[0] {
                KEY_EXCHANGE_MARKER => {
                    let _response = self.handle_key_exchange(data, now_ms);
                    return None;
                }
                PEER_E2EE_MARKER => {
                    let _plaintext = self.handle_peer_e2ee_message(data, now_ms);
                    return None;
                }
                RELAY_ENVELOPE_MARKER => {
                    // Handle relay envelope - extract origin from envelope
                    return self.handle_relay_envelope_with_incoming(identifier, data, now_ms);
                }
                _ => {}
            }
        }

        // Direct document - process normally
        self.process_incoming_document(identifier, data, now_ms, None, None, 0)
    }

    /// Internal: Process incoming document (handles peer registration)
    fn process_incoming_document(
        &self,
        identifier: &str,
        data: &[u8],
        now_ms: u64,
        relay_data: Option<Vec<u8>>,
        origin_node: Option<NodeId>,
        hop_count: u8,
    ) -> Option<DataReceivedResult> {
        // Decrypt if encrypted (mesh-wide encryption)
        let decrypted = self.decrypt_document(data, Some(identifier))?;

        // Merge the document (extracts node_id internally)
        let result = self.document_sync.merge_document(&decrypted)?;

        // Record sync using the source_node from the merged document
        self.peer_manager.record_sync(result.source_node, now_ms);

        // Only register the identifier mapping for direct messages (not relayed)
        // For relayed messages, the identifier belongs to the relay source (who forwarded it),
        // not the origin node (who created the document). The relay source is already registered
        // via handle_relay_envelope_with_incoming when the relay envelope is first processed.
        if origin_node.is_none() {
            // Direct message - register the peer with this identifier
            let is_new =
                self.peer_manager
                    .on_incoming_connection(identifier, result.source_node, now_ms);

            // Update connection graph to track connection state
            {
                let mut graph = self.connection_graph.lock().unwrap();
                if is_new {
                    graph.on_discovered(
                        result.source_node,
                        identifier.to_string(),
                        None,
                        Some(self.config.mesh_id.clone()),
                        -50, // Default RSSI for data-based discovery
                        now_ms,
                    );
                }
                graph.on_connected(result.source_node, now_ms);
            }
        }

        // Generate events based on what was received
        if result.is_emergency() {
            self.notify(PeatEvent::EmergencyReceived {
                from_node: result.source_node,
            });
        } else if result.is_ack() {
            self.notify(PeatEvent::AckReceived {
                from_node: result.source_node,
            });
        }

        if result.counter_changed {
            self.notify(PeatEvent::DocumentSynced {
                from_node: result.source_node,
                total_count: result.total_count,
            });
        }

        // Emit relay event if we're relaying
        if relay_data.is_some() {
            let relay_targets = self.get_relay_targets(Some(result.source_node));
            self.notify(PeatEvent::MessageRelayed {
                origin_node: origin_node.unwrap_or(result.source_node),
                relay_count: relay_targets.len(),
                hop_count,
            });
        }

        let PeripheralFields {
            callsign,
            battery_percent,
            heart_rate,
            event_type,
            latitude,
            longitude,
            altitude,
            activity_level,
            alerts,
        } = DataReceivedResult::peripheral_fields(&result.peer_peripheral);

        Some(DataReceivedResult {
            source_node: result.source_node,
            is_emergency: result.is_emergency(),
            is_ack: result.is_ack(),
            counter_changed: result.counter_changed,
            emergency_changed: result.emergency_changed,
            total_count: result.total_count,
            event_timestamp: result.event.as_ref().map(|e| e.timestamp).unwrap_or(0),
            relay_data,
            origin_node,
            hop_count,
            callsign,
            battery_percent,
            heart_rate,
            event_type,
            latitude,
            longitude,
            altitude,
            activity_level,
            alerts,
            ..Default::default()
        })
    }

    /// Internal: Handle relay envelope with incoming connection registration
    fn handle_relay_envelope_with_incoming(
        &self,
        identifier: &str,
        data: &[u8],
        now_ms: u64,
    ) -> Option<DataReceivedResult> {
        // Parse envelope to get origin
        let envelope = RelayEnvelope::decode(data)?;

        // Try to look up the source peer from identifier to register indirect path
        // If we know who sent this relay, we can track indirect peers via them
        if let Some(source_peer) = self.peer_manager.get_node_id(identifier) {
            if envelope.origin_node != source_peer && envelope.origin_node != self.node_id() {
                let is_new = self.connection_graph.lock().unwrap().on_relay_received(
                    source_peer,
                    envelope.origin_node,
                    envelope.hop_count,
                    now_ms,
                );

                if is_new {
                    log::debug!(
                        "Discovered indirect peer {:08X} via {:08X} ({} hops)",
                        envelope.origin_node.as_u32(),
                        source_peer.as_u32(),
                        envelope.hop_count
                    );
                }
            }
        }

        // Check deduplication
        if !self.mark_message_seen(envelope.message_id, envelope.origin_node, now_ms) {
            // Duplicate - get stats for event
            let stats = self
                .seen_cache
                .lock()
                .unwrap()
                .get_stats(&envelope.message_id);
            let seen_count = stats.map(|(_, count, _)| count).unwrap_or(1);

            self.notify(PeatEvent::DuplicateMessageDropped {
                origin_node: envelope.origin_node,
                seen_count,
            });
            return None;
        }

        // Check TTL
        let (should_relay, relay_data) = if envelope.can_relay() && self.config.enable_relay {
            let relay_env = envelope.relay();
            (true, relay_env.map(|e| e.encode()))
        } else {
            if !envelope.can_relay() {
                self.notify(PeatEvent::MessageTtlExpired {
                    origin_node: envelope.origin_node,
                    hop_count: envelope.hop_count,
                });
            }
            (false, None)
        };

        // Process the inner payload
        self.process_incoming_document(
            identifier,
            &envelope.payload,
            now_ms,
            if should_relay { relay_data } else { None },
            Some(envelope.origin_node),
            envelope.hop_count,
        )
    }

    // ==================== Periodic Maintenance ====================

    /// Periodic tick - call this regularly (e.g., every second)
    ///
    /// Performs:
    /// - Stale peer cleanup
    /// - Periodic sync broadcast (if interval elapsed)
    ///
    /// Returns `Some(data)` if a sync broadcast is needed.
    pub fn tick(&self, now_ms: u64) -> Option<Vec<u8>> {
        use std::sync::atomic::Ordering;

        // Use u32 for atomic storage (wraps every ~49 days, intervals still work)
        let now_ms_32 = now_ms as u32;

        // Cleanup stale peers
        let last_cleanup = self.last_cleanup_ms.load(Ordering::Relaxed);
        let cleanup_elapsed = now_ms_32.wrapping_sub(last_cleanup);
        if cleanup_elapsed >= self.config.peer_config.cleanup_interval_ms as u32 {
            self.last_cleanup_ms.store(now_ms_32, Ordering::Relaxed);
            let removed = self.peer_manager.cleanup_stale(now_ms);
            for node_id in &removed {
                self.notify(PeatEvent::PeerLost { node_id: *node_id });
            }
            if !removed.is_empty() {
                self.notify_mesh_state_changed();
            }

            // Run connection graph maintenance (transition Disconnected -> Lost)
            {
                let mut graph = self.connection_graph.lock().unwrap();
                let newly_lost = graph.tick(now_ms);
                // Also cleanup peers lost for more than peer_timeout
                graph.cleanup_lost(self.config.peer_config.peer_timeout_ms, now_ms);
                drop(graph);

                // Emit PeerLost events for newly lost peers from graph
                // (these may differ from peer_manager removals)
                for node_id in newly_lost {
                    // Only notify if not already notified by peer_manager
                    if !removed.contains(&node_id) {
                        self.notify(PeatEvent::PeerLost { node_id });
                    }
                }
            }
        }

        // Check if sync broadcast is needed
        let last_sync = self.last_sync_ms.load(Ordering::Relaxed);
        let sync_elapsed = now_ms_32.wrapping_sub(last_sync);
        if sync_elapsed >= self.config.sync_interval_ms as u32 {
            self.last_sync_ms.store(now_ms_32, Ordering::Relaxed);
            // Only broadcast if we have connected peers
            if self.peer_manager.connected_count() > 0 {
                let doc = self.document_sync.build_document();
                return Some(self.encrypt_document(&doc));
            }
        }

        None
    }

    /// Periodic tick returning per-peer delta documents
    ///
    /// Unlike `tick()` which broadcasts a single document to all peers,
    /// this returns targeted deltas that only include changes each peer
    /// hasn't seen. Use this for platforms that support per-peer transmission.
    ///
    /// Returns a list of (NodeId, encrypted_delta) tuples, one per connected peer.
    /// Empty vector if no sync is needed (interval not elapsed or no connected peers).
    pub fn tick_with_peer_deltas(&self, now_ms: u64) -> Vec<(NodeId, Vec<u8>)> {
        use std::sync::atomic::Ordering;
        let now_ms_32 = now_ms as u32;

        // Cleanup stale peers (same as tick())
        let last_cleanup = self.last_cleanup_ms.load(Ordering::Relaxed);
        let cleanup_elapsed = now_ms_32.wrapping_sub(last_cleanup);
        if cleanup_elapsed >= self.config.peer_config.cleanup_interval_ms as u32 {
            self.last_cleanup_ms.store(now_ms_32, Ordering::Relaxed);
            let removed = self.peer_manager.cleanup_stale(now_ms);
            for node_id in &removed {
                self.notify(PeatEvent::PeerLost { node_id: *node_id });
            }
            if !removed.is_empty() {
                self.notify_mesh_state_changed();
            }

            // Run connection graph maintenance
            {
                let mut graph = self.connection_graph.lock().unwrap();
                let newly_lost = graph.tick(now_ms);
                graph.cleanup_lost(self.config.peer_config.peer_timeout_ms, now_ms);
                drop(graph);

                for node_id in newly_lost {
                    if !removed.contains(&node_id) {
                        self.notify(PeatEvent::PeerLost { node_id });
                    }
                }
            }
        }

        // Check if sync is needed
        let last_sync = self.last_sync_ms.load(Ordering::Relaxed);
        let sync_elapsed = now_ms_32.wrapping_sub(last_sync);
        if sync_elapsed >= self.config.sync_interval_ms as u32 {
            self.last_sync_ms.store(now_ms_32, Ordering::Relaxed);

            // Build document for each connected peer
            let doc = self.document_sync.build_document();
            let encrypted = self.encrypt_document(&doc);
            let mut results = Vec::new();
            for peer in self.get_connected_peers() {
                results.push((peer.node_id, encrypted.clone()));
            }
            return results;
        }

        Vec::new()
    }

    // ==================== State Queries ====================

    /// Get all known peers
    pub fn get_peers(&self) -> Vec<PeatPeer> {
        self.peer_manager.get_peers()
    }

    /// Get connected peers only
    pub fn get_connected_peers(&self) -> Vec<PeatPeer> {
        self.peer_manager.get_connected_peers()
    }

    /// Get a specific peer by NodeId
    pub fn get_peer(&self, node_id: NodeId) -> Option<PeatPeer> {
        self.peer_manager.get_peer(node_id)
    }

    /// Get peer count
    pub fn peer_count(&self) -> usize {
        self.peer_manager.peer_count()
    }

    /// Get connected peer count
    pub fn connected_count(&self) -> usize {
        self.peer_manager.connected_count()
    }

    /// Check if a device mesh ID matches our mesh
    pub fn matches_mesh(&self, device_mesh_id: Option<&str>) -> bool {
        self.peer_manager.matches_mesh(device_mesh_id)
    }

    // ==================== Connection State Graph ====================

    /// Get the connection state graph with all peer states
    ///
    /// Returns a snapshot of all tracked peers and their connection lifecycle state.
    /// Apps can use this to display appropriate UI indicators:
    /// - Green for Connected peers
    /// - Yellow for Degraded or RecentlyDisconnected peers
    /// - Gray for Lost peers
    ///
    /// # Example
    /// ```ignore
    /// let states = mesh.get_connection_graph();
    /// for peer in states {
    ///     match peer.state {
    ///         ConnectionState::Connected => show_green_indicator(&peer),
    ///         ConnectionState::Degraded => show_yellow_indicator(&peer),
    ///         ConnectionState::Disconnected => show_stale_indicator(&peer),
    ///         ConnectionState::Lost => show_gray_indicator(&peer),
    ///         _ => {}
    ///     }
    /// }
    /// ```
    pub fn get_connection_graph(&self) -> Vec<PeerConnectionState> {
        self.connection_graph.lock().unwrap().get_all_owned()
    }

    /// Get a specific peer's connection state
    pub fn get_peer_connection_state(&self, node_id: NodeId) -> Option<PeerConnectionState> {
        self.connection_graph
            .lock()
            .unwrap()
            .get_peer(node_id)
            .cloned()
    }

    /// Get all currently connected peers from the connection graph
    pub fn get_connected_states(&self) -> Vec<PeerConnectionState> {
        self.connection_graph
            .lock()
            .unwrap()
            .get_connected()
            .into_iter()
            .cloned()
            .collect()
    }

    /// Get peers in degraded state (connected but poor signal quality)
    pub fn get_degraded_peers(&self) -> Vec<PeerConnectionState> {
        self.connection_graph
            .lock()
            .unwrap()
            .get_degraded()
            .into_iter()
            .cloned()
            .collect()
    }

    /// Get peers that disconnected within the specified time window
    ///
    /// Useful for showing "stale" peers that were recently connected.
    pub fn get_recently_disconnected(
        &self,
        within_ms: u64,
        now_ms: u64,
    ) -> Vec<PeerConnectionState> {
        self.connection_graph
            .lock()
            .unwrap()
            .get_recently_disconnected(within_ms, now_ms)
            .into_iter()
            .cloned()
            .collect()
    }

    /// Get peers in Lost state (disconnected and no longer advertising)
    pub fn get_lost_peers(&self) -> Vec<PeerConnectionState> {
        self.connection_graph
            .lock()
            .unwrap()
            .get_lost()
            .into_iter()
            .cloned()
            .collect()
    }

    /// Get summary counts of peers in each connection state
    pub fn get_connection_state_counts(&self) -> StateCountSummary {
        self.connection_graph.lock().unwrap().state_counts()
    }

    // ==================== Indirect Peer Methods ====================

    /// Get all indirect (multi-hop) peers
    ///
    /// Returns peers discovered via relay messages that are not directly
    /// connected via BLE. Each indirect peer includes the minimum hop count
    /// and the direct peers through which they can be reached.
    pub fn get_indirect_peers(&self) -> Vec<IndirectPeer> {
        self.connection_graph
            .lock()
            .unwrap()
            .get_indirect_peers_owned()
    }

    /// Get the degree (hop count) for a specific peer
    ///
    /// Returns:
    /// - `Some(PeerDegree::Direct)` for directly connected BLE peers
    /// - `Some(PeerDegree::OneHop/TwoHop/ThreeHop)` for indirect peers
    /// - `None` if peer is not known
    pub fn get_peer_degree(&self, node_id: NodeId) -> Option<PeerDegree> {
        self.connection_graph.lock().unwrap().peer_degree(node_id)
    }

    /// Get full state counts including indirect peers
    ///
    /// Returns counts of direct peers by connection state plus counts
    /// of indirect peers by hop count (1-hop, 2-hop, 3-hop).
    pub fn get_full_state_counts(&self) -> FullStateCountSummary {
        self.connection_graph.lock().unwrap().full_state_counts()
    }

    /// Get all paths to reach an indirect peer
    ///
    /// Returns a list of (via_peer_id, hop_count) pairs showing all
    /// known routes to the specified peer.
    pub fn get_paths_to_peer(&self, node_id: NodeId) -> Vec<(NodeId, u8)> {
        self.connection_graph.lock().unwrap().get_paths_to(node_id)
    }

    /// Check if a node is known (either direct or indirect)
    pub fn is_peer_known(&self, node_id: NodeId) -> bool {
        self.connection_graph.lock().unwrap().is_known(node_id)
    }

    /// Get number of indirect peers
    pub fn indirect_peer_count(&self) -> usize {
        self.connection_graph.lock().unwrap().indirect_peer_count()
    }

    /// Cleanup stale indirect peers
    ///
    /// Removes indirect peers that haven't been seen within the timeout.
    /// Returns the list of removed peer IDs.
    pub fn cleanup_indirect_peers(&self, now_ms: u64) -> Vec<NodeId> {
        self.connection_graph
            .lock()
            .unwrap()
            .cleanup_indirect(now_ms)
    }

    /// Get total counter value
    pub fn total_count(&self) -> u64 {
        self.document_sync.total_count()
    }

    /// Get document version
    pub fn document_version(&self) -> u32 {
        self.document_sync.version()
    }

    /// Get document version (alias)
    pub fn version(&self) -> u32 {
        self.document_sync.version()
    }

    /// Update health status (battery percentage)
    pub fn update_health(&self, battery_percent: u8) {
        self.document_sync.update_health(battery_percent);
    }

    /// Update activity level. The wire field is a non-validated `u8`
    /// render hint (see `HealthStatus::decode`); receivers map known
    /// values and silently ignore unknown ones. Current emitters use:
    /// 0=still/standing, 3=PossibleFall, 4+=reserved for sender-side
    /// extensions.
    pub fn update_activity(&self, activity: u8) {
        self.document_sync.update_activity(activity);
    }

    /// Update full health status (battery and activity)
    pub fn update_health_full(&self, battery_percent: u8, activity: u8) {
        self.document_sync
            .update_health_full(battery_percent, activity);
    }

    /// Update heart rate
    pub fn update_heart_rate(&self, heart_rate: u8) {
        self.document_sync.update_heart_rate(heart_rate);
    }

    /// Replace the alert flags bitfield. Callers OR together
    /// `HealthStatus::ALERT_*` constants (e.g. `ALERT_MAN_DOWN`) to
    /// express which alerts are currently active. Passing `0` clears all.
    pub fn update_alerts(&self, alerts: u8) {
        self.document_sync.update_alerts(alerts);
    }

    /// Update location
    pub fn update_location(&self, latitude: f32, longitude: f32, altitude: Option<f32>) {
        self.document_sync
            .update_location(latitude, longitude, altitude);
    }

    /// Clear location
    pub fn clear_location(&self) {
        self.document_sync.clear_location();
    }

    /// Update callsign
    pub fn update_callsign(&self, callsign: &str) {
        self.document_sync.update_callsign(callsign);
    }

    /// Set peripheral event type
    pub fn set_peripheral_event(&self, event_type: EventType, timestamp: u64) {
        self.document_sync
            .set_peripheral_event(event_type, timestamp);
    }

    /// Clear peripheral event
    pub fn clear_peripheral_event(&self) {
        self.document_sync.clear_peripheral_event();
    }

    /// Update full peripheral state in one call
    ///
    /// This is the most efficient way to update all peripheral data before
    /// calling `build_document()` for encrypted transmission.
    #[allow(clippy::too_many_arguments)]
    pub fn update_peripheral_state(
        &self,
        callsign: &str,
        battery_percent: u8,
        heart_rate: Option<u8>,
        latitude: Option<f32>,
        longitude: Option<f32>,
        altitude: Option<f32>,
        event_type: Option<EventType>,
        timestamp: u64,
    ) {
        self.document_sync.update_peripheral_state(
            callsign,
            battery_percent,
            heart_rate,
            latitude,
            longitude,
            altitude,
            event_type,
            timestamp,
        );
    }

    /// Build current document for transmission
    ///
    /// If encryption is enabled, the document is encrypted.
    pub fn build_document(&self) -> Vec<u8> {
        let doc = self.document_sync.build_document();
        self.encrypt_document(&doc)
    }

    /// Build a translator-framed (`0xB6`) wire payload for the given
    /// collection + document, encrypted under the mesh secret if
    /// encryption is configured. Returns ready-to-broadcast bytes for
    /// the caller's transport (BLE notification / GATT write / nimble
    /// gossip), or `None` if the translator declines the collection.
    ///
    /// Symmetric with [`Self::build_document`] (legacy peripheral
    /// path). Hosts that already integrate with peat-mesh use this
    /// shape from peat-mesh's own `Translator` impl (Slice 4.a
    /// `peat_mesh::transport::btle_translator::BleTranslator`).
    /// Embedded callers without peat-mesh in their dep graph (M5Stack
    /// Core2 et al.) use the slim siblings
    /// `publish_platform_advertisement` / future per-collection
    /// methods, which take peat-btle-native typed input directly. The
    /// wire shape (encrypt-wrap of `[0xB6, collection_code,
    /// postcard...]`) is identical across both — receivers can't
    /// distinguish publishers.
    ///
    /// **ADR-059 Amendment 4 Slice 4.b note.** Pre-Slice-4.b this
    /// method took `&peat_mesh::sync::Document`. peat-btle no longer
    /// depends on peat-mesh, so the parameter is now
    /// `&serde_json::Value` (the same JSON projection peat-mesh's
    /// `Translator` impl produces internally before invoking the
    /// peat-btle codec). peat-mesh-using callers serialize their
    /// `Document` to a `Value` once and pass it through; standalone
    /// callers use the slim siblings instead.
    #[cfg(feature = "translator-codec")]
    pub fn publish_translator_frame(
        &self,
        collection: &str,
        value: &serde_json::Value,
    ) -> Option<Vec<u8>> {
        let translator = self.ble_translator.read().ok()?;
        let framed = translator.encode_outbound_sync(value, collection)?;
        drop(translator);
        Some(self.encrypt_document(&framed))
    }

    /// Build a translator-framed platform advertisement from a
    /// [`crate::translator::BlePeripheral`] directly — codec-only,
    /// no peat-mesh dep needed. This is the publish surface embedded
    /// firmware uses (M5Stack Core2 emits one of these on every
    /// activity transition, encoding ALERT_MAN_DOWN into
    /// `peripheral.health.alerts` to drive the receiver-side
    /// "unavailable" status semantic).
    ///
    /// Wire format is identical to [`Self::publish_translator_frame`]
    /// invoked with `collection = "platforms"` and a peat-mesh
    /// `Document` whose fields came from
    /// `peripheral_to_platform_in_cell` — the receiver dispatcher
    /// (`try_handle_translator_marker`) can't distinguish the two.
    /// Skipping the `BlePeripheral → JSON Value → BlePeripheral`
    /// round-trip the generic path takes saves an allocation pair on
    /// the embedded side and lets the M5Stack avoid pulling
    /// peat-mesh's tokio + signal-hook chain (which doesn't build for
    /// xtensa-esp32).
    #[cfg(feature = "translator-codec")]
    pub fn publish_platform_advertisement(
        &self,
        peripheral: &crate::translator::BlePeripheral,
    ) -> Option<Vec<u8>> {
        let payload = crate::translator::postcard_encode(peripheral)?;
        let mut framed = Vec::with_capacity(2 + payload.len());
        framed.push(crate::document::TRANSLATOR_FRAME_MARKER);
        framed.push(crate::translator::COLLECTION_CODE_PLATFORMS);
        framed.extend_from_slice(&payload);
        Some(self.encrypt_document(&framed))
    }

    /// Build a wire-ready peat-lite Document frame for transmission.
    ///
    /// Wraps the supplied envelope payload (the bytes
    /// `peat_mesh::transport::document_codec::encode_document`
    /// produced) with the 16-byte peat-lite header
    /// (`MessageType::Document`, this node's id + sequence number),
    /// then encrypts via the same `encrypt_document` path
    /// translator-frame and platform-advertisement publishers use, so
    /// the wire-level encryption stays uniform across frame types.
    ///
    /// Caller hands the returned `Vec<u8>` to whatever GATT-write +
    /// chunked-transport path the platform adapter uses for outbound
    /// frames. peat-lite envelopes that exceed BLE MTU pass through
    /// `crate::sync::protocol::chunk_data` /
    /// `ChunkReassembler` transparently — no caller-side
    /// fragmentation logic.
    ///
    /// The wire `seq_num` is sourced from an internal monotonic
    /// counter (`peat_lite_seq`) so the polled-host receive-side dedup
    /// on `(source_node_id, seq_num)` always sees a unique pair per
    /// outbound frame. Pre-dedup-fix the caller threaded their own
    /// counter and the API documented "pass 0 if dedup happens
    /// elsewhere," which silently disabled dedup whenever the caller
    /// took the escape hatch — fixed per the QA review on the initial
    /// PR. The counter is `Relaxed`-ordered: only the per-emission
    /// monotonicity matters; cross-thread visibility is irrelevant.
    #[cfg(feature = "peat-lite-frame")]
    pub fn publish_peat_lite_document(&self, envelope_payload: &[u8]) -> Option<Vec<u8>> {
        let seq_num = self
            .peat_lite_seq
            .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
        let framed = crate::peat_lite_frame::encode_peat_lite_document(
            self.node_id().as_u32(),
            seq_num,
            envelope_payload,
        )
        .ok()?;
        Some(self.encrypt_document(&framed))
    }

    /// Get peers that should be synced with
    pub fn peers_needing_sync(&self, now_ms: u64) -> Vec<PeatPeer> {
        self.peer_manager.peers_needing_sync(now_ms)
    }

    // ==================== Internal Helpers ====================

    fn notify(&self, event: PeatEvent) {
        self.observers.notify(event);
    }

    fn notify_mesh_state_changed(&self) {
        self.notify(PeatEvent::MeshStateChanged {
            peer_count: self.peer_manager.peer_count(),
            connected_count: self.peer_manager.connected_count(),
        });
    }

    // ==================== CannedMessage Integration ====================
    //
    // These methods provide deduplication support for peat-lite CannedMessages.
    // They use document identity (source_node + timestamp) instead of content hash,
    // because CRDT merge can change byte ordering.

    /// Check if a CannedMessage should be processed.
    ///
    /// Uses document identity (source_node + timestamp) for deduplication.
    /// This prevents broadcast storms when relaying CannedMessages across the mesh.
    ///
    /// # Arguments
    /// * `source_node` - The source node ID from the CannedMessage
    /// * `timestamp` - The timestamp from the CannedMessage
    /// * `_ttl_ms` - TTL parameter (currently unused, uses cache's default TTL)
    ///
    /// # Returns
    /// `true` if this message is new and should be processed,
    /// `false` if it was seen recently and should be skipped.
    pub fn check_canned_message(&self, source_node: u32, timestamp: u64, _ttl_ms: u64) -> bool {
        // Create a unique key from source_node and timestamp
        // MessageId is 16 bytes: [source_node: 4B][timestamp: 8B][padding: 4B]
        let mut id_bytes = [0u8; 16];
        id_bytes[0..4].copy_from_slice(&source_node.to_le_bytes());
        id_bytes[4..12].copy_from_slice(&timestamp.to_le_bytes());
        let message_id = crate::relay::MessageId::from_bytes(id_bytes);

        // Check the seen cache
        let seen = self.seen_cache.lock().unwrap();
        !seen.has_seen(&message_id)
    }

    /// Mark a CannedMessage as seen (for deduplication).
    ///
    /// Call this after processing a CannedMessage to prevent reprocessing
    /// the same message from other relay paths.
    pub fn mark_canned_message_seen(&self, source_node: u32, timestamp: u64) {
        let now = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .map(|d| d.as_millis() as u64)
            .unwrap_or(0);

        // MessageId is 16 bytes: [source_node: 4B][timestamp: 8B][padding: 4B]
        let mut id_bytes = [0u8; 16];
        id_bytes[0..4].copy_from_slice(&source_node.to_le_bytes());
        id_bytes[4..12].copy_from_slice(&timestamp.to_le_bytes());
        let message_id = crate::relay::MessageId::from_bytes(id_bytes);
        let origin = NodeId::new(source_node);

        let mut seen = self.seen_cache.lock().unwrap();
        seen.mark_seen(message_id, origin, now);
    }

    /// Get list of connected peer identifiers for relay.
    ///
    /// Used by the platform layer (Kotlin/Swift) to relay CannedMessages
    /// to other peers after deduplication check.
    pub fn get_connected_peer_identifiers(&self) -> Vec<String> {
        self.peer_manager.get_connected_identifiers()
    }
}

/// Result from receiving BLE data
#[derive(Debug, Clone, Default)]
pub struct DataReceivedResult {
    /// Node that sent this data
    pub source_node: NodeId,

    /// Whether this contained an emergency event
    pub is_emergency: bool,

    /// Whether this contained an ACK event
    pub is_ack: bool,

    /// Whether the counter changed (new data)
    pub counter_changed: bool,

    /// Whether emergency state changed (new emergency or ACK updates)
    pub emergency_changed: bool,

    /// Updated total count
    pub total_count: u64,

    /// Event timestamp (if event present) - use to detect duplicate events
    pub event_timestamp: u64,

    /// Data to relay to other peers (if multi-hop relay is enabled)
    ///
    /// When present, the platform adapter should send this data to peers
    /// returned by `get_relay_targets(Some(source_node))`.
    pub relay_data: Option<Vec<u8>>,

    /// Origin node for relay (may differ from source_node for relayed messages)
    pub origin_node: Option<NodeId>,

    /// Current hop count (for relayed messages)
    pub hop_count: u8,

    // ========== Peripheral data from sender ==========
    /// Sender's callsign (up to 12 chars)
    pub callsign: Option<String>,

    /// Sender's battery percentage (0-100)
    pub battery_percent: Option<u8>,

    /// Sender's heart rate (BPM)
    pub heart_rate: Option<u8>,

    /// Sender's event type (from PeripheralEvent)
    pub event_type: Option<u8>,

    /// Sender's latitude
    pub latitude: Option<f32>,

    /// Sender's longitude
    pub longitude: Option<f32>,

    /// Sender's altitude (meters)
    pub altitude: Option<f32>,

    /// Sender's activity level (0=Standing, 3=PossibleFall, 4=Prone, …).
    /// Mirrors `HealthStatus::activity` from the merged peripheral so
    /// hosts can render posture state without re-decoding the wire
    /// bytes. None when no peripheral data was attached to the merge.
    pub activity_level: Option<u8>,

    /// Sender's health alerts bitfield (e.g. `ALERT_MAN_DOWN = 0x01`).
    /// Mirrors `HealthStatus::alerts` from the merged peripheral. The
    /// receive-side surfaces this so consumers (the ATAK plugin's
    /// `_platformOfflinePeers` toggle) can flip OFFLINE state directly
    /// from peripheral broadcasts, without needing the higher-level
    /// 0xB6 platforms translator-frame to fire.
    pub alerts: Option<u8>,

    /// ADR-059 Amendment 3 — set when the receive dispatch decoded a
    /// 0xB6 translator frame on this call. `None` for legacy/delta/
    /// reserved-marker paths, for 0xB6 frames that declined or errored,
    /// and for builds compiled without the `translator-codec` feature
    /// (the field stays in the binding shape unconditionally so hosts
    /// don't see binding drift across feature combos; only the
    /// population logic is feature-gated).
    pub decoded_translator_frame: Option<DecodedTranslatorFrame>,

    /// Universal-Document-transport amendment — set when the receive
    /// dispatch decoded a peat-lite Document frame
    /// (`MessageType::Document`, magic-prefix `b"PEAT"`) on this
    /// call. Polled-consumer hosts (peat-mesh, peat-atak-plugin's
    /// marker bridge, …) forward populated entries through their
    /// existing publish-with-origin FFI surface to plug the body
    /// into peat-mesh's doc store.
    ///
    /// `None` for typed-translator (0xB6) / delta / legacy-peripheral
    /// paths, for peat-lite frames that declined or errored, and for
    /// builds compiled without the `peat-lite-frame` feature (the
    /// field stays in the binding shape unconditionally so hosts
    /// don't see binding drift across feature combos; only the
    /// population logic is feature-gated).
    pub peat_lite_document: Option<PeatLiteDocumentFrame>,
}

/// Owned snapshot of a decoded peat-lite Document frame, surfaced via
/// [`DataReceivedResult::peat_lite_document`] when the receive
/// dispatch successfully decodes a `MessageType::Document` envelope.
///
/// Mirrors the [`DecodedTranslatorFrame`] design pattern: defined
/// unconditionally so the UniFFI binding shape stays stable across
/// feature combos. Population only happens when the
/// `peat-lite-frame` feature is on (the codec dispatcher in
/// [`crate::peat_lite_frame`] is feature-gated).
///
/// Hosts forward populated entries through their existing
/// publish-with-origin FFI surface to plug into peat-mesh's doc
/// store. The `body` is opaque bytes — peat-btle does not interpret;
/// consumers (peat-mesh's `transport::document_codec`,
/// peat-atak-plugin's marker bridge, M5Stack firmware) own the body
/// schema.
#[derive(Debug, Clone)]
pub struct PeatLiteDocumentFrame {
    /// Source node id from the peat-lite header. Polled hosts use
    /// this for loop-prevention and dedup tracking.
    pub source_node_id: u32,
    /// Sequence number from the peat-lite header. Polled hosts use
    /// this to detect duplicate deliveries.
    pub seq_num: u32,
    /// Flags byte from the Document envelope (bit 0 = tombstone).
    pub flags: u8,
    /// Collection name (UTF-8). Doc-store routing key alongside
    /// `doc_id`.
    pub collection: String,
    /// Document id (UTF-8). Empty signals
    /// "publisher-delegated id assignment."
    pub doc_id: String,
    /// Unix epoch milliseconds.
    pub timestamp_ms: i64,
    /// Opaque body bytes (typically JSON-encoded
    /// `Document.fields` from peat-mesh's `transport::document_codec`).
    pub body: Vec<u8>,
}

impl PeatLiteDocumentFrame {
    /// Returns true if the deletion-tombstone flag is set.
    #[inline]
    pub fn is_tombstone(&self) -> bool {
        // peat-lite bit 0 — duplicated as a const here to keep the
        // type defined without the optional peat-lite dep.
        self.flags & 0x01 != 0
    }
}

/// ADR-059 Amendment 3 — payload returned in
/// [`DataReceivedResult::decoded_translator_frame`] when a 0xB6
/// translator frame decodes successfully on the receive dispatch.
/// Hosts (peat-atak-plugin and equivalents) forward populated entries
/// through their existing publish-with-origin FFI surface (e.g.
/// peat-ffi's `publishDocumentWithOriginJni(collection, doc_json,
/// "ble")`). Defined unconditionally — no `#[cfg(feature = "translator-codec")]`
/// — so the UniFFI binding shape is stable across feature combos.
/// Population only happens when `translator-codec` is on.
#[derive(Debug, Clone)]
pub struct DecodedTranslatorFrame {
    /// `BleTranslator` collection name, e.g. `"tracks"` / `"platforms"`.
    pub collection: String,
    /// serde-JSON serialization of the decoded `Document` — same shape
    /// as the JSON callback variant from 0.3.1, just delivered via
    /// struct return rather than callback invocation.
    pub doc_json: String,
    /// BLE peer identifier from the receive context (when known).
    pub peer: Option<String>,
}

/// ADR-059 Amendment 3 — per-call outcome of the
/// `try_handle_translator_marker` dispatcher. Per-call return-threaded;
/// no shared mutable state, so concurrent receives on different
/// threads can't race on a slot.
#[cfg(feature = "translator-codec")]
#[derive(Debug, Clone)]
enum TranslatorMarkerOutcome {
    /// First byte didn't match the translator-frame range; caller
    /// continues the legacy / delta-document dispatch.
    NotTranslatorMarker,
    /// 0xB6 frame decoded successfully. Caller hoists the payload into
    /// `DataReceivedResult.decoded_translator_frame` before returning.
    Decoded(DecodedTranslatorFrame),
    /// 0xB6 reserved-range / decode-error / codec-decline / unknown-code
    /// — caller stops processing but no frame to surface.
    Handled,
}

impl DataReceivedResult {
    /// ADR-059 Amendment 3 — construct a `DataReceivedResult` whose
    /// only meaningful payload is a successfully-decoded 0xB6 translator
    /// frame. Other fields default (no event, no counter change,
    /// no peripheral data); the host reads `decoded_translator_frame`
    /// and forwards through its publish-with-origin FFI surface.
    /// Used by the receive entry points when
    /// `try_handle_translator_marker` returns `Decoded(frame)`.
    #[cfg(feature = "translator-codec")]
    pub(crate) fn translator_frame(source_node: NodeId, frame: DecodedTranslatorFrame) -> Self {
        Self {
            source_node,
            decoded_translator_frame: Some(frame),
            ..Default::default()
        }
    }

    /// Extract peripheral fields from an `Option<Peripheral>`.
    ///
    /// Returns a [`PeripheralFields`] struct rather than a positional
    /// tuple so future additions don't require lockstep updates at
    /// every destructure site (the prior 9-tuple had 5 callers and
    /// adding the next field would have meant 6 mechanical edits with
    /// the type system unable to catch a misordered destructure).
    fn peripheral_fields(peripheral: &Option<crate::sync::crdt::Peripheral>) -> PeripheralFields {
        match peripheral {
            Some(p) => {
                let callsign = {
                    let s = p.callsign_str();
                    if s.is_empty() {
                        None
                    } else {
                        Some(s.to_string())
                    }
                };
                let battery_percent = if p.health.battery_percent > 0 {
                    Some(p.health.battery_percent)
                } else {
                    None
                };
                let (latitude, longitude, altitude) = match &p.location {
                    Some(loc) => (Some(loc.latitude), Some(loc.longitude), loc.altitude),
                    None => (None, None, None),
                };
                PeripheralFields {
                    callsign,
                    battery_percent,
                    heart_rate: p.health.heart_rate,
                    event_type: p.last_event.as_ref().map(|e| e.event_type as u8),
                    latitude,
                    longitude,
                    altitude,
                    activity_level: Some(p.health.activity),
                    alerts: Some(p.health.alerts),
                }
            }
            None => PeripheralFields::default(),
        }
    }
}

/// Subset of merged-peripheral fields surfaced on
/// [`DataReceivedResult`]. Returned by
/// [`DataReceivedResult::peripheral_fields`] so callers don't have to
/// destructure a positional tuple — adding a new field is a struct
/// update with the type system enforcing every consumer either reads
/// or ignores it explicitly, instead of silently re-indexing existing
/// destructures.
#[derive(Debug, Clone, Default)]
pub struct PeripheralFields {
    /// Sender's callsign (None when the peripheral has an empty callsign).
    pub callsign: Option<String>,
    /// Sender's battery percentage (None when the peripheral reports 0,
    /// which the wire convention treats as "absent").
    pub battery_percent: Option<u8>,
    /// Sender's heart rate (None when the wire byte is 0 — the
    /// `unwrap_or(0)` convention from the Rust encode side).
    pub heart_rate: Option<u8>,
    /// Sender's PeripheralEvent type (None when no event is attached).
    pub event_type: Option<u8>,
    /// Sender's latitude (None when no location record is attached).
    pub latitude: Option<f32>,
    /// Sender's longitude (None when no location record is attached).
    pub longitude: Option<f32>,
    /// Sender's altitude in meters (None when no location record is
    /// attached, or when the location's altitude field is itself absent).
    pub altitude: Option<f32>,
    /// Sender's activity level (0=Standing, 3=PossibleFall, 4=Prone …).
    pub activity_level: Option<u8>,
    /// Sender's health-alerts bitfield (e.g. `ALERT_MAN_DOWN = 0x01`).
    pub alerts: Option<u8>,
}

/// Decision from processing a relay envelope
#[derive(Debug, Clone)]
pub struct RelayDecision {
    /// The payload (document) to process locally
    pub payload: Vec<u8>,

    /// Original sender of the message
    pub origin_node: NodeId,

    /// Current hop count
    pub hop_count: u8,

    /// Whether this message should be relayed to other peers
    pub should_relay: bool,

    /// The relay envelope to forward (with incremented hop count)
    ///
    /// Only present if `should_relay` is true and TTL not expired.
    pub relay_envelope: Option<RelayEnvelope>,
}

impl RelayDecision {
    /// Get the relay data to send to peers
    ///
    /// Returns None if relay is not needed.
    pub fn relay_data(&self) -> Option<Vec<u8>> {
        self.relay_envelope.as_ref().map(|e| e.encode())
    }
}

#[cfg(all(test, feature = "std"))]
mod tests {
    use super::*;
    use crate::observer::CollectingObserver;

    // Valid timestamp for testing (2024-01-15 00:00:00 UTC)
    const TEST_TIMESTAMP: u64 = 1705276800000;

    fn create_mesh(node_id: u32, callsign: &str) -> PeatMesh {
        let config = PeatMeshConfig::new(NodeId::new(node_id), callsign, "TEST");
        PeatMesh::new(config)
    }

    #[test]
    fn test_mesh_creation() {
        let mesh = create_mesh(0x12345678, "ALPHA-1");

        assert_eq!(mesh.node_id().as_u32(), 0x12345678);
        assert_eq!(mesh.callsign(), "ALPHA-1");
        assert_eq!(mesh.mesh_id(), "TEST");
        assert_eq!(mesh.device_name(), "PEAT_TEST-12345678");
    }

    #[test]
    fn test_peer_discovery() {
        let mesh = create_mesh(0x11111111, "ALPHA-1");
        let observer = Arc::new(CollectingObserver::new());
        mesh.add_observer(observer.clone());

        // Discover a peer
        let peer = mesh.on_ble_discovered(
            "device-uuid",
            Some("PEAT_TEST-22222222"),
            -65,
            Some("TEST"),
            1000,
        );

        assert!(peer.is_some());
        let peer = peer.unwrap();
        assert_eq!(peer.node_id.as_u32(), 0x22222222);

        // Check events were generated
        let events = observer.events();
        assert!(events
            .iter()
            .any(|e| matches!(e, PeatEvent::PeerDiscovered { .. })));
        assert!(events
            .iter()
            .any(|e| matches!(e, PeatEvent::MeshStateChanged { .. })));
    }

    #[test]
    fn test_connection_lifecycle() {
        let mesh = create_mesh(0x11111111, "ALPHA-1");
        let observer = Arc::new(CollectingObserver::new());
        mesh.add_observer(observer.clone());

        // Discover and connect
        mesh.on_ble_discovered(
            "device-uuid",
            Some("PEAT_TEST-22222222"),
            -65,
            Some("TEST"),
            1000,
        );

        let node_id = mesh.on_ble_connected("device-uuid", 2000);
        assert_eq!(node_id, Some(NodeId::new(0x22222222)));
        assert_eq!(mesh.connected_count(), 1);

        // Disconnect
        let node_id = mesh.on_ble_disconnected("device-uuid", DisconnectReason::RemoteRequest);
        assert_eq!(node_id, Some(NodeId::new(0x22222222)));
        assert_eq!(mesh.connected_count(), 0);

        // Check events
        let events = observer.events();
        assert!(events
            .iter()
            .any(|e| matches!(e, PeatEvent::PeerConnected { .. })));
        assert!(events
            .iter()
            .any(|e| matches!(e, PeatEvent::PeerDisconnected { .. })));
    }

    #[test]
    fn test_emergency_flow() {
        let mesh1 = create_mesh(0x11111111, "ALPHA-1");
        let mesh2 = create_mesh(0x22222222, "BRAVO-1");

        let observer2 = Arc::new(CollectingObserver::new());
        mesh2.add_observer(observer2.clone());

        // mesh1 sends emergency
        let doc = mesh1.send_emergency(TEST_TIMESTAMP);
        assert!(mesh1.is_emergency_active());

        // mesh2 receives it
        let result =
            mesh2.on_ble_data_received_from_node(NodeId::new(0x11111111), &doc, TEST_TIMESTAMP);

        assert!(result.is_some());
        let result = result.unwrap();
        assert!(result.is_emergency);
        assert_eq!(result.source_node.as_u32(), 0x11111111);

        // Check events on mesh2
        let events = observer2.events();
        assert!(events
            .iter()
            .any(|e| matches!(e, PeatEvent::EmergencyReceived { .. })));
    }

    #[test]
    fn test_ack_flow() {
        let mesh1 = create_mesh(0x11111111, "ALPHA-1");
        let mesh2 = create_mesh(0x22222222, "BRAVO-1");

        let observer2 = Arc::new(CollectingObserver::new());
        mesh2.add_observer(observer2.clone());

        // mesh1 sends ACK
        let doc = mesh1.send_ack(TEST_TIMESTAMP);
        assert!(mesh1.is_ack_active());

        // mesh2 receives it
        let result =
            mesh2.on_ble_data_received_from_node(NodeId::new(0x11111111), &doc, TEST_TIMESTAMP);

        assert!(result.is_some());
        let result = result.unwrap();
        assert!(result.is_ack);

        // Check events on mesh2
        let events = observer2.events();
        assert!(events
            .iter()
            .any(|e| matches!(e, PeatEvent::AckReceived { .. })));
    }

    #[test]
    fn test_tick_cleanup() {
        let config = PeatMeshConfig::new(NodeId::new(0x11111111), "ALPHA-1", "TEST")
            .with_peer_timeout(10_000);
        let mesh = PeatMesh::new(config);

        let observer = Arc::new(CollectingObserver::new());
        mesh.add_observer(observer.clone());

        // Discover a peer
        mesh.on_ble_discovered(
            "device-uuid",
            Some("PEAT_TEST-22222222"),
            -65,
            Some("TEST"),
            1000,
        );
        assert_eq!(mesh.peer_count(), 1);

        // Tick at t=5000 - not stale yet
        mesh.tick(5000);
        assert_eq!(mesh.peer_count(), 1);

        // Tick at t=20000 - peer is stale (10s timeout exceeded)
        mesh.tick(20000);
        assert_eq!(mesh.peer_count(), 0);

        // Check PeerLost event
        let events = observer.events();
        assert!(events
            .iter()
            .any(|e| matches!(e, PeatEvent::PeerLost { .. })));
    }

    #[test]
    fn test_tick_sync_broadcast() {
        let config = PeatMeshConfig::new(NodeId::new(0x11111111), "ALPHA-1", "TEST")
            .with_sync_interval(5000);
        let mesh = PeatMesh::new(config);

        // Discover and connect a peer first
        mesh.on_ble_discovered(
            "device-uuid",
            Some("PEAT_TEST-22222222"),
            -65,
            Some("TEST"),
            1000,
        );
        mesh.on_ble_connected("device-uuid", 1000);

        // First tick at t=0 sets last_sync
        let _result = mesh.tick(0);
        // May or may not broadcast depending on initial state

        // Tick before interval - no broadcast
        let result = mesh.tick(3000);
        assert!(result.is_none());

        // After interval - should broadcast
        let result = mesh.tick(6000);
        assert!(result.is_some());

        // Immediate second tick - no broadcast (interval not elapsed)
        let result = mesh.tick(6100);
        assert!(result.is_none());

        // After another interval - should broadcast again
        let result = mesh.tick(12000);
        assert!(result.is_some());
    }

    #[test]
    fn test_incoming_connection() {
        let mesh = create_mesh(0x11111111, "ALPHA-1");
        let observer = Arc::new(CollectingObserver::new());
        mesh.add_observer(observer.clone());

        // Incoming connection from unknown peer
        let is_new = mesh.on_incoming_connection("central-uuid", NodeId::new(0x22222222), 1000);

        assert!(is_new);
        assert_eq!(mesh.peer_count(), 1);
        assert_eq!(mesh.connected_count(), 1);

        // Check events
        let events = observer.events();
        assert!(events
            .iter()
            .any(|e| matches!(e, PeatEvent::PeerDiscovered { .. })));
        assert!(events
            .iter()
            .any(|e| matches!(e, PeatEvent::PeerConnected { .. })));
    }

    #[test]
    fn test_mesh_filtering() {
        let mesh = create_mesh(0x11111111, "ALPHA-1");

        // Wrong mesh - ignored
        let peer = mesh.on_ble_discovered(
            "device-uuid-1",
            Some("PEAT_OTHER-22222222"),
            -65,
            Some("OTHER"),
            1000,
        );
        assert!(peer.is_none());
        assert_eq!(mesh.peer_count(), 0);

        // Correct mesh - accepted
        let peer = mesh.on_ble_discovered(
            "device-uuid-2",
            Some("PEAT_TEST-33333333"),
            -65,
            Some("TEST"),
            1000,
        );
        assert!(peer.is_some());
        assert_eq!(mesh.peer_count(), 1);
    }

    // ==================== Encryption Tests ====================

    fn create_encrypted_mesh(node_id: u32, callsign: &str, secret: [u8; 32]) -> PeatMesh {
        let config =
            PeatMeshConfig::new(NodeId::new(node_id), callsign, "TEST").with_encryption(secret);
        PeatMesh::new(config)
    }

    #[test]
    fn test_encryption_enabled() {
        let secret = [0x42u8; 32];
        let mesh = create_encrypted_mesh(0x11111111, "ALPHA-1", secret);

        assert!(mesh.is_encryption_enabled());
    }

    #[test]
    fn test_encryption_disabled_by_default() {
        let mesh = create_mesh(0x11111111, "ALPHA-1");

        assert!(!mesh.is_encryption_enabled());
    }

    #[test]
    fn test_encrypted_document_exchange() {
        let secret = [0x42u8; 32];
        let mesh1 = create_encrypted_mesh(0x11111111, "ALPHA-1", secret);
        let mesh2 = create_encrypted_mesh(0x22222222, "BRAVO-1", secret);

        // mesh1 sends document
        let doc = mesh1.build_document();

        // Document should be encrypted (starts with ENCRYPTED_MARKER)
        assert!(doc.len() >= 2);
        assert_eq!(doc[0], crate::document::ENCRYPTED_MARKER);

        // mesh2 receives and decrypts
        let result = mesh2.on_ble_data_received_from_node(NodeId::new(0x11111111), &doc, 1000);

        assert!(result.is_some());
        let result = result.unwrap();
        assert_eq!(result.source_node.as_u32(), 0x11111111);
    }

    #[test]
    fn test_encrypted_emergency_exchange() {
        let secret = [0x42u8; 32];
        let mesh1 = create_encrypted_mesh(0x11111111, "ALPHA-1", secret);
        let mesh2 = create_encrypted_mesh(0x22222222, "BRAVO-1", secret);

        let observer = Arc::new(CollectingObserver::new());
        mesh2.add_observer(observer.clone());

        // mesh1 sends emergency
        let doc = mesh1.send_emergency(TEST_TIMESTAMP);

        // mesh2 receives and decrypts
        let result =
            mesh2.on_ble_data_received_from_node(NodeId::new(0x11111111), &doc, TEST_TIMESTAMP);

        assert!(result.is_some());
        let result = result.unwrap();
        assert!(result.is_emergency);

        // Check EmergencyReceived event was fired
        let events = observer.events();
        assert!(events
            .iter()
            .any(|e| matches!(e, PeatEvent::EmergencyReceived { .. })));
    }

    #[test]
    fn test_wrong_key_fails_decrypt() {
        let secret1 = [0x42u8; 32];
        let secret2 = [0x43u8; 32]; // Different key
        let mesh1 = create_encrypted_mesh(0x11111111, "ALPHA-1", secret1);
        let mesh2 = create_encrypted_mesh(0x22222222, "BRAVO-1", secret2);

        // mesh1 sends document
        let doc = mesh1.build_document();

        // mesh2 cannot decrypt (wrong key)
        let result = mesh2.on_ble_data_received_from_node(NodeId::new(0x11111111), &doc, 1000);

        assert!(result.is_none());
    }

    #[test]
    fn test_unencrypted_mesh_can_read_unencrypted() {
        let mesh1 = create_mesh(0x11111111, "ALPHA-1");
        let mesh2 = create_mesh(0x22222222, "BRAVO-1");

        // mesh1 sends document (unencrypted)
        let doc = mesh1.build_document();

        // mesh2 receives (also unencrypted)
        let result = mesh2.on_ble_data_received_from_node(NodeId::new(0x11111111), &doc, 1000);

        assert!(result.is_some());
    }

    #[test]
    fn test_encrypted_mesh_can_receive_unencrypted() {
        // Backward compatibility: encrypted mesh can receive unencrypted docs
        let secret = [0x42u8; 32];
        let mesh1 = create_mesh(0x11111111, "ALPHA-1"); // unencrypted
        let mesh2 = create_encrypted_mesh(0x22222222, "BRAVO-1", secret); // encrypted

        // mesh1 sends unencrypted document
        let doc = mesh1.build_document();

        // mesh2 can receive unencrypted (backward compat)
        let result = mesh2.on_ble_data_received_from_node(NodeId::new(0x11111111), &doc, 1000);

        assert!(result.is_some());
    }

    #[test]
    fn test_unencrypted_mesh_cannot_receive_encrypted() {
        let secret = [0x42u8; 32];
        let mesh1 = create_encrypted_mesh(0x11111111, "ALPHA-1", secret); // encrypted
        let mesh2 = create_mesh(0x22222222, "BRAVO-1"); // unencrypted

        // mesh1 sends encrypted document
        let doc = mesh1.build_document();

        // mesh2 cannot decrypt (no key)
        let result = mesh2.on_ble_data_received_from_node(NodeId::new(0x11111111), &doc, 1000);

        assert!(result.is_none());
    }

    #[test]
    fn test_enable_disable_encryption() {
        let mut mesh = create_mesh(0x11111111, "ALPHA-1");

        assert!(!mesh.is_encryption_enabled());

        // Enable encryption
        let secret = [0x42u8; 32];
        mesh.enable_encryption(&secret);
        assert!(mesh.is_encryption_enabled());

        // Build document should now be encrypted
        let doc = mesh.build_document();
        assert_eq!(doc[0], crate::document::ENCRYPTED_MARKER);

        // Disable encryption
        mesh.disable_encryption();
        assert!(!mesh.is_encryption_enabled());

        // Build document should now be unencrypted
        let doc = mesh.build_document();
        assert_ne!(doc[0], crate::document::ENCRYPTED_MARKER);
    }

    #[test]
    fn test_encryption_overhead() {
        let secret = [0x42u8; 32];
        let mesh_encrypted = create_encrypted_mesh(0x11111111, "ALPHA-1", secret);
        let mesh_unencrypted = create_mesh(0x22222222, "BRAVO-1");

        let doc_encrypted = mesh_encrypted.build_document();
        let doc_unencrypted = mesh_unencrypted.build_document();

        // Encrypted doc should be larger by:
        // - 2 bytes marker header (0xAE + reserved)
        // - 12 bytes nonce
        // - 16 bytes auth tag
        // Total: 30 bytes overhead
        let overhead = doc_encrypted.len() - doc_unencrypted.len();
        assert_eq!(overhead, 30); // 2 (marker) + 12 (nonce) + 16 (tag)
    }

    // ==================== Per-Peer E2EE Tests ====================

    #[test]
    fn test_peer_e2ee_enable_disable() {
        let mesh = create_mesh(0x11111111, "ALPHA-1");

        assert!(!mesh.is_peer_e2ee_enabled());
        assert!(mesh.peer_e2ee_public_key().is_none());

        mesh.enable_peer_e2ee();
        assert!(mesh.is_peer_e2ee_enabled());
        assert!(mesh.peer_e2ee_public_key().is_some());

        mesh.disable_peer_e2ee();
        assert!(!mesh.is_peer_e2ee_enabled());
    }

    #[test]
    fn test_peer_e2ee_initiate_session() {
        let mesh = create_mesh(0x11111111, "ALPHA-1");
        mesh.enable_peer_e2ee();

        let key_exchange = mesh.initiate_peer_e2ee(NodeId::new(0x22222222), 1000);
        assert!(key_exchange.is_some());

        let key_exchange = key_exchange.unwrap();
        // Should start with KEY_EXCHANGE_MARKER
        assert_eq!(key_exchange[0], crate::document::KEY_EXCHANGE_MARKER);

        // Should have a pending session
        assert_eq!(mesh.peer_e2ee_session_count(), 1);
        assert_eq!(mesh.peer_e2ee_established_count(), 0);
    }

    #[test]
    fn test_peer_e2ee_full_handshake() {
        let mesh1 = create_mesh(0x11111111, "ALPHA-1");
        let mesh2 = create_mesh(0x22222222, "BRAVO-1");

        mesh1.enable_peer_e2ee();
        mesh2.enable_peer_e2ee();

        let observer1 = Arc::new(CollectingObserver::new());
        let observer2 = Arc::new(CollectingObserver::new());
        mesh1.add_observer(observer1.clone());
        mesh2.add_observer(observer2.clone());

        // mesh1 initiates to mesh2
        let key_exchange1 = mesh1
            .initiate_peer_e2ee(NodeId::new(0x22222222), 1000)
            .unwrap();

        // mesh2 receives and responds
        let response = mesh2.handle_key_exchange(&key_exchange1, 1000);
        assert!(response.is_some());

        // Check mesh2 has established session
        assert!(mesh2.has_peer_e2ee_session(NodeId::new(0x11111111)));

        // mesh1 receives mesh2's response
        let key_exchange2 = response.unwrap();
        let _ = mesh1.handle_key_exchange(&key_exchange2, 1000);

        // Check mesh1 has established session
        assert!(mesh1.has_peer_e2ee_session(NodeId::new(0x22222222)));

        // Both should have E2EE established events
        let events1 = observer1.events();
        assert!(events1
            .iter()
            .any(|e| matches!(e, PeatEvent::PeerE2eeEstablished { .. })));

        let events2 = observer2.events();
        assert!(events2
            .iter()
            .any(|e| matches!(e, PeatEvent::PeerE2eeEstablished { .. })));
    }

    #[test]
    fn test_peer_e2ee_encrypt_decrypt() {
        let mesh1 = create_mesh(0x11111111, "ALPHA-1");
        let mesh2 = create_mesh(0x22222222, "BRAVO-1");

        mesh1.enable_peer_e2ee();
        mesh2.enable_peer_e2ee();

        // Establish session via key exchange
        let key_exchange1 = mesh1
            .initiate_peer_e2ee(NodeId::new(0x22222222), 1000)
            .unwrap();
        let key_exchange2 = mesh2.handle_key_exchange(&key_exchange1, 1000).unwrap();
        mesh1.handle_key_exchange(&key_exchange2, 1000);

        // mesh1 sends encrypted message to mesh2
        let plaintext = b"Secret message from mesh1";
        let encrypted = mesh1.send_peer_e2ee(NodeId::new(0x22222222), plaintext, 2000);
        assert!(encrypted.is_some());

        let encrypted = encrypted.unwrap();
        // Should start with PEER_E2EE_MARKER
        assert_eq!(encrypted[0], crate::document::PEER_E2EE_MARKER);

        // mesh2 receives and decrypts
        let observer2 = Arc::new(CollectingObserver::new());
        mesh2.add_observer(observer2.clone());

        let decrypted = mesh2.handle_peer_e2ee_message(&encrypted, 2000);
        assert!(decrypted.is_some());
        assert_eq!(decrypted.unwrap(), plaintext);

        // Should have received message event
        let events = observer2.events();
        assert!(events.iter().any(|e| matches!(
            e,
            PeatEvent::PeerE2eeMessageReceived { from_node, data }
            if from_node.as_u32() == 0x11111111 && data == plaintext
        )));
    }

    #[test]
    fn test_peer_e2ee_bidirectional() {
        let mesh1 = create_mesh(0x11111111, "ALPHA-1");
        let mesh2 = create_mesh(0x22222222, "BRAVO-1");

        mesh1.enable_peer_e2ee();
        mesh2.enable_peer_e2ee();

        // Establish session
        let key_exchange1 = mesh1
            .initiate_peer_e2ee(NodeId::new(0x22222222), 1000)
            .unwrap();
        let key_exchange2 = mesh2.handle_key_exchange(&key_exchange1, 1000).unwrap();
        mesh1.handle_key_exchange(&key_exchange2, 1000);

        // mesh1 -> mesh2
        let msg1 = mesh1
            .send_peer_e2ee(NodeId::new(0x22222222), b"Hello from mesh1", 2000)
            .unwrap();
        let dec1 = mesh2.handle_peer_e2ee_message(&msg1, 2000).unwrap();
        assert_eq!(dec1, b"Hello from mesh1");

        // mesh2 -> mesh1
        let msg2 = mesh2
            .send_peer_e2ee(NodeId::new(0x11111111), b"Hello from mesh2", 2000)
            .unwrap();
        let dec2 = mesh1.handle_peer_e2ee_message(&msg2, 2000).unwrap();
        assert_eq!(dec2, b"Hello from mesh2");
    }

    #[test]
    fn test_peer_e2ee_close_session() {
        let mesh = create_mesh(0x11111111, "ALPHA-1");
        mesh.enable_peer_e2ee();

        let observer = Arc::new(CollectingObserver::new());
        mesh.add_observer(observer.clone());

        // Initiate a session
        mesh.initiate_peer_e2ee(NodeId::new(0x22222222), 1000);
        assert_eq!(mesh.peer_e2ee_session_count(), 1);

        // Close session
        mesh.close_peer_e2ee(NodeId::new(0x22222222));

        // Check close event
        let events = observer.events();
        assert!(events
            .iter()
            .any(|e| matches!(e, PeatEvent::PeerE2eeClosed { .. })));
    }

    #[test]
    fn test_peer_e2ee_without_enabling() {
        let mesh = create_mesh(0x11111111, "ALPHA-1");

        // E2EE not enabled - should return None
        let result = mesh.initiate_peer_e2ee(NodeId::new(0x22222222), 1000);
        assert!(result.is_none());

        let result = mesh.send_peer_e2ee(NodeId::new(0x22222222), b"test", 1000);
        assert!(result.is_none());

        assert!(!mesh.has_peer_e2ee_session(NodeId::new(0x22222222)));
    }

    #[test]
    fn test_peer_e2ee_overhead() {
        let mesh1 = create_mesh(0x11111111, "ALPHA-1");
        let mesh2 = create_mesh(0x22222222, "BRAVO-1");

        mesh1.enable_peer_e2ee();
        mesh2.enable_peer_e2ee();

        // Establish session
        let key_exchange1 = mesh1
            .initiate_peer_e2ee(NodeId::new(0x22222222), 1000)
            .unwrap();
        let key_exchange2 = mesh2.handle_key_exchange(&key_exchange1, 1000).unwrap();
        mesh1.handle_key_exchange(&key_exchange2, 1000);

        // Encrypt a message
        let plaintext = b"Test message";
        let encrypted = mesh1
            .send_peer_e2ee(NodeId::new(0x22222222), plaintext, 2000)
            .unwrap();

        // Overhead should be:
        // - 2 bytes marker header
        // - 4 bytes recipient node ID
        // - 4 bytes sender node ID
        // - 8 bytes counter
        // - 12 bytes nonce
        // - 16 bytes auth tag
        // Total: 46 bytes overhead
        let overhead = encrypted.len() - plaintext.len();
        assert_eq!(overhead, 46);
    }

    // ==================== Strict Encryption Mode Tests ====================

    fn create_strict_encrypted_mesh(node_id: u32, callsign: &str, secret: [u8; 32]) -> PeatMesh {
        let config = PeatMeshConfig::new(NodeId::new(node_id), callsign, "TEST")
            .with_encryption(secret)
            .with_strict_encryption();
        PeatMesh::new(config)
    }

    #[test]
    fn test_strict_encryption_enabled() {
        let secret = [0x42u8; 32];
        let mesh = create_strict_encrypted_mesh(0x11111111, "ALPHA-1", secret);

        assert!(mesh.is_encryption_enabled());
        assert!(mesh.is_strict_encryption_enabled());
    }

    #[test]
    fn test_strict_encryption_disabled_by_default() {
        let secret = [0x42u8; 32];
        let mesh = create_encrypted_mesh(0x11111111, "ALPHA-1", secret);

        assert!(mesh.is_encryption_enabled());
        assert!(!mesh.is_strict_encryption_enabled());
    }

    #[test]
    fn test_strict_encryption_requires_encryption_enabled() {
        // strict_encryption without encryption should have no effect
        let config = PeatMeshConfig::new(NodeId::new(0x11111111), "ALPHA-1", "TEST")
            .with_strict_encryption(); // No encryption!
        let mesh = PeatMesh::new(config);

        assert!(!mesh.is_encryption_enabled());
        assert!(!mesh.is_strict_encryption_enabled());
    }

    #[test]
    fn test_strict_mode_accepts_encrypted_documents() {
        let secret = [0x42u8; 32];
        let mesh1 = create_encrypted_mesh(0x11111111, "ALPHA-1", secret);
        let mesh2 = create_strict_encrypted_mesh(0x22222222, "BRAVO-1", secret);

        // mesh1 sends encrypted document
        let doc = mesh1.build_document();
        assert_eq!(doc[0], crate::document::ENCRYPTED_MARKER);

        // mesh2 (strict mode) should accept encrypted documents
        let result = mesh2.on_ble_data_received_from_node(NodeId::new(0x11111111), &doc, 1000);
        assert!(result.is_some());
    }

    #[test]
    fn test_strict_mode_rejects_unencrypted_documents() {
        let secret = [0x42u8; 32];
        let mesh1 = create_mesh(0x11111111, "ALPHA-1"); // Unencrypted sender
        let mesh2 = create_strict_encrypted_mesh(0x22222222, "BRAVO-1", secret); // Strict receiver

        let observer = Arc::new(CollectingObserver::new());
        mesh2.add_observer(observer.clone());

        // mesh1 sends unencrypted document
        let doc = mesh1.build_document();
        assert_ne!(doc[0], crate::document::ENCRYPTED_MARKER);

        // mesh2 (strict mode) should reject unencrypted documents
        let result = mesh2.on_ble_data_received_from_node(NodeId::new(0x11111111), &doc, 1000);
        assert!(result.is_none());

        // Should have SecurityViolation event
        let events = observer.events();
        assert!(events.iter().any(|e| matches!(
            e,
            PeatEvent::SecurityViolation {
                kind: crate::observer::SecurityViolationKind::UnencryptedInStrictMode,
                ..
            }
        )));
    }

    #[test]
    fn test_non_strict_mode_accepts_unencrypted_documents() {
        let secret = [0x42u8; 32];
        let mesh1 = create_mesh(0x11111111, "ALPHA-1"); // Unencrypted sender
        let mesh2 = create_encrypted_mesh(0x22222222, "BRAVO-1", secret); // Non-strict receiver

        // mesh1 sends unencrypted document
        let doc = mesh1.build_document();

        // mesh2 (non-strict) should accept unencrypted documents (backward compat)
        let result = mesh2.on_ble_data_received_from_node(NodeId::new(0x11111111), &doc, 1000);
        assert!(result.is_some());
    }

    #[test]
    fn test_strict_mode_security_violation_event_includes_source() {
        let secret = [0x42u8; 32];
        let mesh1 = create_mesh(0x11111111, "ALPHA-1");
        let mesh2 = create_strict_encrypted_mesh(0x22222222, "BRAVO-1", secret);

        let observer = Arc::new(CollectingObserver::new());
        mesh2.add_observer(observer.clone());

        let doc = mesh1.build_document();

        // Use on_ble_data_received with identifier to test source is captured
        mesh2.on_ble_discovered(
            "test-device-uuid",
            Some("PEAT_TEST-11111111"),
            -65,
            Some("TEST"),
            500,
        );
        mesh2.on_ble_connected("test-device-uuid", 600);

        let _result = mesh2.on_ble_data_received("test-device-uuid", &doc, 1000);

        // Check SecurityViolation event has source
        let events = observer.events();
        let violation = events.iter().find(|e| {
            matches!(
                e,
                PeatEvent::SecurityViolation {
                    kind: crate::observer::SecurityViolationKind::UnencryptedInStrictMode,
                    ..
                }
            )
        });
        assert!(violation.is_some());

        if let Some(PeatEvent::SecurityViolation { source, .. }) = violation {
            assert!(source.is_some());
            assert_eq!(source.as_ref().unwrap(), "test-device-uuid");
        }
    }

    #[test]
    fn test_decryption_failure_emits_security_violation() {
        let secret1 = [0x42u8; 32];
        let secret2 = [0x43u8; 32]; // Different key
        let mesh1 = create_encrypted_mesh(0x11111111, "ALPHA-1", secret1);
        let mesh2 = create_encrypted_mesh(0x22222222, "BRAVO-1", secret2);

        let observer = Arc::new(CollectingObserver::new());
        mesh2.add_observer(observer.clone());

        // mesh1 sends encrypted document
        let doc = mesh1.build_document();

        // mesh2 cannot decrypt (wrong key)
        let result = mesh2.on_ble_data_received_from_node(NodeId::new(0x11111111), &doc, 1000);
        assert!(result.is_none());

        // Should have SecurityViolation event for decryption failure
        let events = observer.events();
        assert!(events.iter().any(|e| matches!(
            e,
            PeatEvent::SecurityViolation {
                kind: crate::observer::SecurityViolationKind::DecryptionFailed,
                ..
            }
        )));
    }

    #[test]
    fn test_strict_mode_builder_chain() {
        let secret = [0x42u8; 32];
        let config = PeatMeshConfig::new(NodeId::new(0x11111111), "ALPHA-1", "TEST")
            .with_encryption(secret)
            .with_strict_encryption()
            .with_sync_interval(10_000)
            .with_peer_timeout(60_000);

        let mesh = PeatMesh::new(config);

        assert!(mesh.is_encryption_enabled());
        assert!(mesh.is_strict_encryption_enabled());
    }

    // ==================== Multi-Hop Relay Tests ====================

    fn create_relay_mesh(node_id: u32, callsign: &str) -> PeatMesh {
        let config = PeatMeshConfig::new(NodeId::new(node_id), callsign, "TEST").with_relay();
        PeatMesh::new(config)
    }

    #[test]
    fn test_relay_disabled_by_default() {
        let mesh = create_mesh(0x11111111, "ALPHA-1");
        assert!(!mesh.is_relay_enabled());
    }

    #[test]
    fn test_relay_enabled() {
        let mesh = create_relay_mesh(0x11111111, "ALPHA-1");
        assert!(mesh.is_relay_enabled());
    }

    #[test]
    fn test_relay_config_builder() {
        let config = PeatMeshConfig::new(NodeId::new(0x11111111), "ALPHA-1", "TEST")
            .with_relay()
            .with_max_relay_hops(5)
            .with_relay_fanout(3)
            .with_seen_cache_ttl(60_000);

        assert!(config.enable_relay);
        assert_eq!(config.max_relay_hops, 5);
        assert_eq!(config.relay_fanout, 3);
        assert_eq!(config.seen_cache_ttl_ms, 60_000);
    }

    #[test]
    fn test_seen_message_deduplication() {
        let mesh = create_relay_mesh(0x11111111, "ALPHA-1");
        let origin = NodeId::new(0x22222222);
        let msg_id = crate::relay::MessageId::from_content(origin, 1000, 0xDEADBEEF);

        // First time - should be new
        assert!(mesh.mark_message_seen(msg_id, origin, 1000));

        // Second time - should be duplicate
        assert!(!mesh.mark_message_seen(msg_id, origin, 2000));

        assert_eq!(mesh.seen_cache_size(), 1);
    }

    #[test]
    fn test_wrap_for_relay() {
        let mesh = create_relay_mesh(0x11111111, "ALPHA-1");

        let payload = vec![1, 2, 3, 4, 5];
        let wrapped = mesh.wrap_for_relay(payload.clone());

        // Should start with relay envelope marker
        assert_eq!(wrapped[0], crate::relay::RELAY_ENVELOPE_MARKER);

        // Decode and verify
        let envelope = crate::relay::RelayEnvelope::decode(&wrapped).unwrap();
        assert_eq!(envelope.payload, payload);
        assert_eq!(envelope.origin_node, NodeId::new(0x11111111));
        assert_eq!(envelope.hop_count, 0);
    }

    #[test]
    fn test_process_relay_envelope_new_message() {
        let mesh = create_relay_mesh(0x11111111, "ALPHA-1");
        let observer = Arc::new(CollectingObserver::new());
        mesh.add_observer(observer.clone());

        // Create an envelope from another node
        let payload = vec![1, 2, 3, 4, 5];
        let envelope =
            crate::relay::RelayEnvelope::broadcast(NodeId::new(0x22222222), payload.clone())
                .with_max_hops(7);
        let data = envelope.encode();

        // Process it
        let decision = mesh.process_relay_envelope(&data, NodeId::new(0x33333333), 1000);

        assert!(decision.is_some());
        let decision = decision.unwrap();
        assert_eq!(decision.payload, payload);
        assert_eq!(decision.origin_node.as_u32(), 0x22222222);
        assert_eq!(decision.hop_count, 0);
        assert!(decision.should_relay);
        assert!(decision.relay_envelope.is_some());

        // Relay envelope should have incremented hop count
        let relay_env = decision.relay_envelope.unwrap();
        assert_eq!(relay_env.hop_count, 1);
    }

    #[test]
    fn test_process_relay_envelope_duplicate() {
        let mesh = create_relay_mesh(0x11111111, "ALPHA-1");
        let observer = Arc::new(CollectingObserver::new());
        mesh.add_observer(observer.clone());

        let payload = vec![1, 2, 3, 4, 5];
        let envelope = crate::relay::RelayEnvelope::broadcast(NodeId::new(0x22222222), payload);
        let data = envelope.encode();

        // First time - should succeed
        let decision = mesh.process_relay_envelope(&data, NodeId::new(0x33333333), 1000);
        assert!(decision.is_some());

        // Second time - should be duplicate
        let decision = mesh.process_relay_envelope(&data, NodeId::new(0x33333333), 2000);
        assert!(decision.is_none());

        // Should have DuplicateMessageDropped event
        let events = observer.events();
        assert!(events
            .iter()
            .any(|e| matches!(e, PeatEvent::DuplicateMessageDropped { .. })));
    }

    #[test]
    fn test_process_relay_envelope_ttl_expired() {
        let mesh = create_relay_mesh(0x11111111, "ALPHA-1");
        let observer = Arc::new(CollectingObserver::new());
        mesh.add_observer(observer.clone());

        // Create envelope at max hops (TTL expired)
        let payload = vec![1, 2, 3, 4, 5];
        let mut envelope =
            crate::relay::RelayEnvelope::broadcast(NodeId::new(0x22222222), payload.clone())
                .with_max_hops(3);

        // Simulate having been relayed 3 times already
        envelope = envelope.relay().unwrap(); // hop 1
        envelope = envelope.relay().unwrap(); // hop 2
        envelope = envelope.relay().unwrap(); // hop 3 - at max now

        let data = envelope.encode();

        // Process - should still process locally but not relay further
        let decision = mesh.process_relay_envelope(&data, NodeId::new(0x33333333), 1000);

        assert!(decision.is_some());
        let decision = decision.unwrap();
        assert_eq!(decision.payload, payload);
        assert!(!decision.should_relay); // Cannot relay further
        assert!(decision.relay_envelope.is_none());

        // Should have MessageTtlExpired event
        let events = observer.events();
        assert!(events
            .iter()
            .any(|e| matches!(e, PeatEvent::MessageTtlExpired { .. })));
    }

    #[test]
    fn test_build_relay_document() {
        let mesh = create_relay_mesh(0x11111111, "ALPHA-1");

        let relay_doc = mesh.build_relay_document();

        // Should be a valid relay envelope
        assert_eq!(relay_doc[0], crate::relay::RELAY_ENVELOPE_MARKER);

        // Decode and verify it contains a valid document
        let envelope = crate::relay::RelayEnvelope::decode(&relay_doc).unwrap();
        assert_eq!(envelope.origin_node.as_u32(), 0x11111111);

        // The payload should be a valid PeatDocument
        let doc = crate::document::PeatDocument::decode(&envelope.payload);
        assert!(doc.is_some());
    }

    #[test]
    fn test_relay_targets_excludes_source() {
        let mesh = create_relay_mesh(0x11111111, "ALPHA-1");

        // Add some peers
        mesh.on_ble_discovered(
            "peer-1",
            Some("PEAT_TEST-22222222"),
            -60,
            Some("TEST"),
            1000,
        );
        mesh.on_ble_connected("peer-1", 1000);

        mesh.on_ble_discovered(
            "peer-2",
            Some("PEAT_TEST-33333333"),
            -65,
            Some("TEST"),
            1000,
        );
        mesh.on_ble_connected("peer-2", 1000);

        mesh.on_ble_discovered(
            "peer-3",
            Some("PEAT_TEST-44444444"),
            -70,
            Some("TEST"),
            1000,
        );
        mesh.on_ble_connected("peer-3", 1000);

        // Get relay targets excluding peer-2
        let targets = mesh.get_relay_targets(Some(NodeId::new(0x33333333)));

        // Should not include peer-2 in targets
        assert!(targets.iter().all(|p| p.node_id.as_u32() != 0x33333333));
    }

    #[test]
    fn test_clear_seen_cache() {
        let mesh = create_relay_mesh(0x11111111, "ALPHA-1");
        let origin = NodeId::new(0x22222222);

        // Add some messages
        mesh.mark_message_seen(
            crate::relay::MessageId::from_content(origin, 1000, 0x11111111),
            origin,
            1000,
        );
        mesh.mark_message_seen(
            crate::relay::MessageId::from_content(origin, 2000, 0x22222222),
            origin,
            2000,
        );

        assert_eq!(mesh.seen_cache_size(), 2);

        // Clear
        mesh.clear_seen_cache();
        assert_eq!(mesh.seen_cache_size(), 0);
    }

    // ====================================================================
    // Universal Document transport (peat-lite envelope) integration tests
    // ====================================================================

    /// **Outbound encode end-to-end**: build a peat-lite envelope payload,
    /// run it through `publish_peat_lite_document`, peel the encryption,
    /// verify the wire bytes start with peat-lite's MAGIC + carry the
    /// expected sender node id + seq num + envelope contents. Locks the
    /// outbound contract that the marker-bridge demo will rely on.
    #[cfg(feature = "peat-lite-frame")]
    #[test]
    fn publish_peat_lite_document_emits_well_formed_frame() {
        use peat_lite::protocol::document as pl_doc;

        let mesh = create_mesh(0xCAFEBABE, "ALPHA-1");

        // Build a Document envelope payload (mirrors what
        // `peat_mesh::transport::document_codec::encode_document`
        // produces).
        let mut env = vec![0u8; 256];
        let env_n = pl_doc::encode(
            0,
            "markers",
            "test-id",
            1_700_000_000_000,
            b"body",
            &mut env,
        )
        .expect("envelope encode");
        env.truncate(env_n);

        // Publish via the peat-mesh outbound API. `seq_num` is now
        // sourced from the internal monotonic counter — first publish
        // starts at 0 and increments per emission.
        let wire = mesh
            .publish_peat_lite_document(&env)
            .expect("publish should succeed");

        // Peel the wire-level encryption (matches what the receive
        // path does on inbound).
        let decrypted = mesh
            .decrypt_document(&wire, None)
            .expect("decrypt own outbound");

        // Decrypted bytes are the peat-lite frame; verify shape.
        assert!(
            crate::peat_lite_frame::is_peat_lite_frame(&decrypted),
            "decrypted outbound must start with peat-lite MAGIC"
        );

        match crate::peat_lite_frame::try_handle_peat_lite_frame(&decrypted) {
            crate::peat_lite_frame::PeatLiteFrameOutcome::Decoded(doc) => {
                assert_eq!(doc.source_node_id, 0xCAFEBABE);
                assert_eq!(doc.seq_num, 0, "first publish must use seq=0");
                assert_eq!(doc.collection, "markers");
                assert_eq!(doc.doc_id, "test-id");
                assert_eq!(doc.body.as_slice(), b"body");
            }
            other => panic!("decoded outbound frame mis-routed: {:?}", other),
        }

        // Second publish must increment — locks the dedup contract
        // that the QA review on the initial PR flagged. If a future
        // refactor drops the internal counter back to a per-call
        // parameter with a 0 escape hatch, this assertion catches it.
        let wire2 = mesh
            .publish_peat_lite_document(&env)
            .expect("second publish should succeed");
        let decrypted2 = mesh
            .decrypt_document(&wire2, None)
            .expect("decrypt second outbound");
        match crate::peat_lite_frame::try_handle_peat_lite_frame(&decrypted2) {
            crate::peat_lite_frame::PeatLiteFrameOutcome::Decoded(doc) => {
                assert_eq!(
                    doc.seq_num, 1,
                    "second publish must increment seq, not collide with first"
                );
            }
            other => panic!("second outbound frame mis-routed: {:?}", other),
        }
    }

    /// **Routing-safety end-to-end**: a peat-lite frame produced by
    /// `publish_peat_lite_document` must NOT be classified as a
    /// translator (0xB6) frame on the receive side. peat-lite's
    /// magic byte 0x50 is outside translator range; this test pins
    /// the routing invariant so a future dispatcher refactor can't
    /// silently re-route.
    #[cfg(feature = "peat-lite-frame")]
    #[test]
    fn peat_lite_frames_do_not_collide_with_translator_routing() {
        use peat_lite::protocol::document as pl_doc;

        let mesh = create_mesh(0x11111111, "ALPHA-1");
        let mut env = vec![0u8; 64];
        let n = pl_doc::encode(0, "tracks", "id", 0, b"x", &mut env).expect("envelope encode");
        env.truncate(n);

        let wire = mesh.publish_peat_lite_document(&env).expect("publish");
        let decrypted = mesh.decrypt_document(&wire, None).expect("decrypt");

        // First byte MUST be peat-lite's 'P' (0x50), not in the
        // translator range 0xB6..=0xBF.
        assert_eq!(
            decrypted[0], 0x50,
            "peat-lite frame's first byte must be 0x50, not in translator range"
        );
        assert!(
            !(0xB6..=0xBF).contains(&decrypted[0]),
            "peat-lite frame must not collide with translator marker range"
        );
    }

    /// **DataReceivedResult binding-shape stability**: the
    /// `peat_lite_document` field is unconditional in the struct
    /// definition, so consumers compiled without the
    /// `peat-lite-frame` feature still get the field (always `None`).
    /// This locks the binding shape contract so UniFFI consumers
    /// don't see drift across feature combos.
    #[test]
    fn data_received_result_default_has_peat_lite_field() {
        let result = DataReceivedResult::default();
        assert!(
            result.peat_lite_document.is_none(),
            "default DataReceivedResult must have peat_lite_document = None"
        );
    }
}