//! Automerge backend adapter for trait abstraction
//!
//! This module provides an adapter between the StorageBackend trait
//! and the AutomergeStore implementation. It enables backend-agnostic business
//! logic with fully open-source CRDT storage.
//!
//! # Architecture
//!
//! ```text
//! Business Logic (Coordinators)
//!         ↓
//! StorageBackend trait (backend-agnostic)
//!         ↓
//! AutomergeBackend (adapter) ← This module
//!         ↓
//! AutomergeStore (Automerge + RocksDB)
//!         ↓
//! ┌────────────────┐  ┌────────────┐
//! │ Automerge 0.7  │  │ RocksDB    │
//! │ (CRDT engine)  │  │ (persist)  │
//! └────────────────┘  └────────────┘
//! ```
//!
//! # Phase 1 Limitations
//!
//! **Current**: Phase 1 stores raw bytes in Automerge documents (simple blob storage).
//! - Documents stored as: `Automerge { "data": bytes }`
//! - No field-level CRDT semantics yet
//! - Provides Collection trait interface for backend-agnostic code
//!
//! **Future** (Phase 2): Protobuf → JSON → Automerge conversion for CRDT benefits.
//! - Field-level merging (OR-Set for arrays, LWW-Register for scalars)
//! - Delta sync (only changed fields transmitted)
//! - See `automerge_conversion.rs` for conversion utilities
//!
//! # Usage Examples
//!
//! ## Create backend with persistence
//!
//! ```ignore
//! use peat_protocol::storage::{AutomergeBackend, AutomergeStore};
//! use std::sync::Arc;
//!
//! let store = Arc::new(AutomergeStore::open("./data/automerge")?);
//! let backend = AutomergeBackend::new(store);
//!
//! // Use via StorageBackend trait
//! let cells = backend.collection("cells");
//! cells.upsert("cell-1", cell_state.encode_to_vec())?;
//! ```
//!
//! ## Backend features
//!
//! | Feature              | AutomergeBackend (Phase 1) | AutomergeBackend (Phase 2) |
//! |----------------------|----------------------------|----------------------------|
//! | CRDT Support         | ❌ (blob storage)          | ✅ (field-level)           |
//! | Persistence          | ✅ (RocksDB)               | ✅ (RocksDB)               |
//! | Network Sync         | ⏭ (Phase 4: Iroh)         | ⏭ (Phase 4: Iroh)         |
//! | License              | MIT/Apache 2.0             | MIT/Apache 2.0             |
//! | Backend-agnostic API | ✅                         | ✅                         |

#[cfg(feature = "automerge-backend")]
use super::automerge_command_storage::AutomergeCommandStorage;
#[cfg(feature = "automerge-backend")]
use super::automerge_conversion::{automerge_to_message, message_to_automerge_into};
#[cfg(feature = "automerge-backend")]
use super::automerge_store::AutomergeStore;
#[cfg(feature = "automerge-backend")]
use super::automerge_summary_storage::AutomergeSummaryStorage;
#[cfg(feature = "automerge-backend")]
use super::automerge_sync::AutomergeSyncCoordinator;
#[cfg(feature = "automerge-backend")]
use super::capabilities::{
    CrdtCapable, HierarchicalStorageCapable, SyncCapable, SyncStats, TypedCollection,
};
#[cfg(feature = "automerge-backend")]
use super::traits::{Collection, StorageBackend};
#[cfg(feature = "automerge-backend")]
use crate::command::CommandStorage;
#[cfg(feature = "automerge-backend")]
use crate::hierarchy::SummaryStorage;
#[cfg(feature = "automerge-backend")]
use crate::network::iroh_transport::IrohTransport;
#[cfg(feature = "automerge-backend")]
use anyhow::Result;
#[cfg(feature = "automerge-backend")]
use peat_mesh::network::EndpointId;
#[cfg(feature = "automerge-backend")]
use peat_mesh::storage::sync_transport::SyncTransport;
#[cfg(feature = "automerge-backend")]
use prost::Message as ProstMessage;
#[cfg(feature = "automerge-backend")]
use serde::{de::DeserializeOwned, Serialize};
#[cfg(feature = "automerge-backend")]
use std::collections::{HashMap, HashSet};
#[cfg(feature = "automerge-backend")]
use std::marker::PhantomData;
#[cfg(feature = "automerge-backend")]
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
#[cfg(feature = "automerge-backend")]
use std::sync::{Arc, RwLock};
#[cfg(feature = "automerge-backend")]
use tokio::task::JoinHandle;

/// Automerge backend adapter implementing StorageBackend trait
///
/// Wraps AutomergeStore to provide trait-based interface for backend-agnostic code.
///
/// # Complete Implementation (Phases 1-6)
///
/// - ✅ RocksDB persistence with LRU cache
/// - ✅ CRDT field-level semantics via Automerge
/// - ✅ P2P sync via Iroh QUIC transport
/// - ✅ Background sync coordination
/// - ✅ SyncCapable trait for lifecycle management
/// - ✅ Incoming sync handler (Phase 6.2)
#[cfg(feature = "automerge-backend")]
pub struct AutomergeBackend {
    /// Underlying AutomergeStore instance
    store: Arc<AutomergeStore>,
    /// Cache of known collection names
    collections: Arc<RwLock<HashMap<String, Arc<dyn Collection>>>>,
    /// Optional Iroh transport for P2P sync (Phase 5) — concrete type for protocol-specific features
    iroh_transport: Option<Arc<IrohTransport>>,
    /// Transport as trait object for sync coordinator
    transport: Option<Arc<dyn SyncTransport>>,
    /// Optional sync coordinator (Phase 5)
    sync_coordinator: Option<Arc<AutomergeSyncCoordinator>>,
    /// Sync state tracking
    sync_active: Arc<AtomicBool>,
    /// Bytes sent counter
    bytes_sent: Arc<AtomicU64>,
    /// Bytes received counter
    bytes_received: Arc<AtomicU64>,
    /// Incoming sync handler task handle (Phase 6.2)
    incoming_handler_task: Arc<RwLock<Option<JoinHandle<()>>>>,
    /// Automatic sync task handle (Phase 6.3)
    auto_sync_task: Arc<RwLock<Option<JoinHandle<()>>>>,
    /// Heartbeat sender task handle (Phase 6.4)
    heartbeat_task: Arc<RwLock<Option<JoinHandle<()>>>>,
    /// Heartbeat receiver task handle (Phase 6.4)
    heartbeat_receiver_task: Arc<RwLock<Option<JoinHandle<()>>>>,
    /// Active sync handlers per connection (to avoid spawning duplicates)
    active_sync_handlers: Arc<RwLock<HashSet<EndpointId>>>,
    /// Active heartbeat handlers per connection
    active_heartbeat_handlers: Arc<RwLock<HashSet<EndpointId>>>,
    /// Sync channel manager for persistent channels (Issue #438 Phase 2)
    channel_manager: Arc<RwLock<Option<Arc<super::sync_channel::SyncChannelManager>>>>,
}

/// Dedup window for duplicate `Connected` events on the
/// event-driven sync spawner (see [`AutomergeBackend::start_sync`]).
/// Calibrated to the 2026-05-10 sync diagnostic — connect-path /
/// accept-path race observed at 2.6s. 5s is comfortably wider than
/// the race without suppressing legitimate reconnect-after-recycle
/// (those are >50s apart by design of `CONNECTION_RECYCLE_INTERVAL_SECS`).
///
/// If iroh's connect/accept race window widens past this in a
/// future release, OR `CONNECTION_RECYCLE_INTERVAL_SECS` is tuned
/// down toward this value, regression tests in
/// `connected_dedup_tests` will fail their boundary assertions —
/// which is the signal to recalibrate.
#[cfg(feature = "automerge-backend")]
const CONNECTED_DEDUP_WINDOW: std::time::Duration = std::time::Duration::from_secs(5);

#[cfg(feature = "automerge-backend")]
impl AutomergeBackend {
    /// Create a new Automerge backend from an existing AutomergeStore
    ///
    /// This creates a backend without P2P sync capabilities.
    /// For sync support, use `with_transport()` instead.
    ///
    /// # Arguments
    ///
    /// * `store` - Configured AutomergeStore instance
    ///
    /// # Example
    ///
    /// ```ignore
    /// use peat_protocol::storage::{AutomergeBackend, AutomergeStore};
    /// use std::sync::Arc;
    ///
    /// let store = Arc::new(AutomergeStore::open("./data/automerge")?);
    /// let backend = AutomergeBackend::new(store);
    /// ```
    pub fn new(store: Arc<AutomergeStore>) -> Self {
        Self {
            store,
            collections: Arc::new(RwLock::new(HashMap::new())),
            iroh_transport: None,
            transport: None,
            sync_coordinator: None,
            sync_active: Arc::new(AtomicBool::new(false)),
            bytes_sent: Arc::new(AtomicU64::new(0)),
            bytes_received: Arc::new(AtomicU64::new(0)),
            incoming_handler_task: Arc::new(RwLock::new(None)),
            auto_sync_task: Arc::new(RwLock::new(None)),
            heartbeat_task: Arc::new(RwLock::new(None)),
            heartbeat_receiver_task: Arc::new(RwLock::new(None)),
            active_sync_handlers: Arc::new(RwLock::new(HashSet::new())),
            active_heartbeat_handlers: Arc::new(RwLock::new(HashSet::new())),
            channel_manager: Arc::new(RwLock::new(None)),
        }
    }

    /// Create a new Automerge backend with P2P sync capabilities
    ///
    /// # Arguments
    ///
    /// * `store` - Configured AutomergeStore instance
    /// * `transport` - IrohTransport for P2P networking
    ///
    /// # Example
    ///
    /// ```ignore
    /// use peat_protocol::storage::{AutomergeBackend, AutomergeStore};
    /// use peat_protocol::network::IrohTransport;
    /// use std::sync::Arc;
    ///
    /// let store = Arc::new(AutomergeStore::open("./data/automerge")?);
    /// let transport = Arc::new(IrohTransport::new().await?);
    /// let backend = AutomergeBackend::with_transport(store, transport);
    /// ```
    pub fn with_transport(store: Arc<AutomergeStore>, transport: Arc<IrohTransport>) -> Self {
        let transport_trait: Arc<dyn SyncTransport> =
            Arc::clone(&transport) as Arc<dyn SyncTransport>;
        let coordinator = Arc::new(AutomergeSyncCoordinator::new(
            Arc::clone(&store),
            Arc::clone(&transport_trait),
        ));

        Self {
            store,
            collections: Arc::new(RwLock::new(HashMap::new())),
            iroh_transport: Some(transport),
            transport: Some(transport_trait),
            sync_coordinator: Some(coordinator),
            sync_active: Arc::new(AtomicBool::new(false)),
            bytes_sent: Arc::new(AtomicU64::new(0)),
            bytes_received: Arc::new(AtomicU64::new(0)),
            incoming_handler_task: Arc::new(RwLock::new(None)),
            auto_sync_task: Arc::new(RwLock::new(None)),
            heartbeat_task: Arc::new(RwLock::new(None)),
            heartbeat_receiver_task: Arc::new(RwLock::new(None)),
            active_sync_handlers: Arc::new(RwLock::new(HashSet::new())),
            active_heartbeat_handlers: Arc::new(RwLock::new(HashSet::new())),
            channel_manager: Arc::new(RwLock::new(None)),
        }
    }

    /// Get access to underlying AutomergeStore for store-specific operations
    ///
    /// This provides an escape hatch for features not yet abstracted by the trait.
    pub fn automerge_store(&self) -> &AutomergeStore {
        &self.store
    }

    /// Get reference to sync coordinator for tombstone propagation (Issue #668)
    pub fn sync_coordinator(&self) -> Option<&Arc<AutomergeSyncCoordinator>> {
        self.sync_coordinator.as_ref()
    }

    /// Get reference to Iroh transport for peer enumeration (Issue #668)
    pub fn iroh_transport(&self) -> Option<&Arc<IrohTransport>> {
        self.iroh_transport.as_ref()
    }

    /// Manually trigger sync for a specific document with all connected peers
    ///
    /// This is useful for testing or for explicit sync triggering.
    /// In production, the background sync task will handle this automatically.
    ///
    /// # Arguments
    ///
    /// * `doc_key` - The full document key (e.g., "nodes:node-1")
    pub async fn sync_document(&self, doc_key: &str) -> Result<()> {
        if let Some(coordinator) = &self.sync_coordinator {
            coordinator.sync_document_with_all_peers(doc_key).await
        } else {
            anyhow::bail!("Cannot sync: backend created without transport")
        }
    }

    /// Spawn a sync handler for a specific peer (Issue #346)
    ///
    /// Suppress duplicate `Connected` events for the same peer that
    /// arrive within `window`. Returns `true` if `peer` is currently
    /// inside its dedup window (caller should `continue` the loop),
    /// `false` if this is the first observation in the window
    /// (caller proceeds and registers handlers).
    ///
    /// The connect-path and accept-path can both emit `Connected`
    /// for the same peer when racing on a recycle reconnect. Without
    /// this filter, both spawned `spawn_sync_handler_for_peer` AND
    /// the proactive `sync_all_documents_with_peer` push run twice
    /// per peer per cycle — wasted bandwidth, redundant log noise,
    /// and risk of duplicate sync stream construction.
    ///
    /// On every call this also reaps entries whose timestamps fall
    /// outside the window. O(n) per event is fine — peer count
    /// stays in the low double digits in practice and the dedup
    /// window itself bounds map size.
    ///
    /// Extracted from the inline async loop so the dedup contract
    /// has unit-test coverage. See `connected_dedup_tests`.
    fn check_and_record_connect(
        recent_connects: &Arc<RwLock<HashMap<EndpointId, std::time::Instant>>>,
        peer: EndpointId,
        now: std::time::Instant,
        window: std::time::Duration,
    ) -> bool {
        let mut map = recent_connects.write().expect("recent_connects poisoned");
        map.retain(|_, t| now.duration_since(*t) < window);
        match map.get(&peer) {
            Some(_) => true,
            None => {
                map.insert(peer, now);
                false
            }
        }
    }

    /// This is called both by the event-based handler spawner (for immediate response)
    /// and by the polling-based fallback (for any connections that might be missed).
    ///
    /// The function is idempotent - if a handler already exists for this peer, it does nothing.
    fn spawn_sync_handler_for_peer(
        peer_id: EndpointId,
        transport: &Arc<dyn SyncTransport>,
        coordinator: &Arc<AutomergeSyncCoordinator>,
        sync_active: &Arc<AtomicBool>,
        active_handlers: &Arc<RwLock<HashSet<EndpointId>>>,
    ) {
        // Skip if we already have a handler for this connection
        {
            let handlers = active_handlers.read().unwrap();
            if handlers.contains(&peer_id) {
                return;
            }
        }

        // Get connection and spawn continuous handler
        if let Some(conn) = transport.get_connection(&peer_id) {
            // Mark as having active handler
            active_handlers.write().unwrap().insert(peer_id);

            // Trigger sync of all existing documents with new peer (Issue #235)
            // Issue #346: Brief delay to allow conflict resolution to settle before syncing.
            let coord_for_initial_sync = Arc::clone(coordinator);
            let initial_sync_peer_id = peer_id;
            let conn_for_initial_check = conn.clone();
            tokio::spawn(async move {
                // Wait for conflict resolution to settle
                tokio::time::sleep(tokio::time::Duration::from_millis(50)).await;

                // Check if connection was closed by conflict resolution
                if conn_for_initial_check.close_reason().is_some() {
                    tracing::debug!(
                        "Skipping initial sync for {:?}: connection was superseded",
                        initial_sync_peer_id
                    );
                    return;
                }

                if let Err(e) = coord_for_initial_sync
                    .sync_all_documents_with_peer(initial_sync_peer_id)
                    .await
                {
                    tracing::warn!(
                        "Failed to sync existing documents with new peer {:?}: {}",
                        initial_sync_peer_id,
                        e
                    );
                }
            });

            let coordinator_clone = Arc::clone(coordinator);
            let sync_active_clone = Arc::clone(sync_active);
            let active_handlers_clone = Arc::clone(active_handlers);
            let handler_peer_id = peer_id;

            // Store the connection's stable_id to detect if it gets replaced by conflict resolution
            let conn_stable_id = conn.stable_id();

            // Spawn continuous handler that loops accepting streams
            tokio::spawn(async move {
                tracing::debug!(
                    "Started continuous sync handler for peer {:?} (conn_id={})",
                    handler_peer_id,
                    conn_stable_id
                );

                // Issue #346: Brief delay to allow conflict resolution to settle.
                // If both nodes connect simultaneously, one connection will be closed.
                // Give time for that to happen before we start using the connection.
                tokio::time::sleep(tokio::time::Duration::from_millis(50)).await;

                // Check if connection was closed by conflict resolution
                if conn.close_reason().is_some() {
                    tracing::debug!(
                        "Sync handler for {:?} exiting: connection was superseded by conflict resolution (conn_id={})",
                        handler_peer_id,
                        conn_stable_id
                    );
                    // Don't remove from active_handlers - a new handler will be spawned
                    // when the correct connection's Connected event fires
                    active_handlers_clone
                        .write()
                        .unwrap()
                        .remove(&handler_peer_id);
                    return;
                }

                // Loop accepting streams until connection closes or sync stops
                while sync_active_clone.load(Ordering::Relaxed) {
                    match conn.accept_bi().await {
                        Ok((send, recv)) => {
                            // Handle this stream in a separate task for parallelism
                            let coord = Arc::clone(&coordinator_clone);
                            let stream_peer_id = handler_peer_id;
                            tokio::spawn(async move {
                                if let Err(e) = coord
                                    .handle_incoming_sync_stream(stream_peer_id, send, recv)
                                    .await
                                {
                                    tracing::debug!("Error handling sync stream: {}", e);
                                }
                            });
                        }
                        Err(e) => {
                            // Connection closed or error - exit handler
                            tracing::debug!(
                                "Sync handler for {:?} exiting: {} (conn_id={})",
                                handler_peer_id,
                                e,
                                conn_stable_id
                            );
                            break;
                        }
                    }
                }

                // Clear sync state for this peer on disconnect
                coordinator_clone.clear_peer_sync_state(handler_peer_id);

                // Remove from active handlers on exit
                active_handlers_clone
                    .write()
                    .unwrap()
                    .remove(&handler_peer_id);
                tracing::debug!(
                    "Stopped continuous sync handler for peer {:?}",
                    handler_peer_id
                );
            });
        }
    }
}

#[cfg(feature = "automerge-backend")]
impl StorageBackend for AutomergeBackend {
    fn collection(&self, name: &str) -> Arc<dyn Collection> {
        // Check cache first
        {
            let collections = self.collections.read().unwrap();
            if let Some(collection) = collections.get(name) {
                return Arc::clone(collection);
            }
        }

        // Create new collection and cache it
        let collection = self.store.collection(name);
        self.collections
            .write()
            .unwrap()
            .insert(name.to_string(), Arc::clone(&collection));

        collection
    }

    fn list_collections(&self) -> Vec<String> {
        // Return known collections from cache
        let collections = self.collections.read().unwrap();
        collections.keys().cloned().collect()
    }

    fn flush(&self) -> Result<()> {
        // RocksDB handles durability automatically via write-ahead log
        // No explicit flush needed for Phase 1
        Ok(())
    }

    fn close(self) -> Result<()> {
        // RocksDB will be closed when AutomergeStore is dropped
        // No explicit cleanup needed for Phase 1
        // Phase 4 will need to stop sync here
        Ok(())
    }
}

/// Typed collection for Automerge backend with CRDT semantics
///
/// Stores protobuf messages as Automerge CRDT documents with field-level merging.
#[cfg(feature = "automerge-backend")]
pub struct AutomergeTypedCollection<M> {
    store: Arc<AutomergeStore>,
    prefix: String,
    _phantom: PhantomData<M>,
}

#[cfg(feature = "automerge-backend")]
impl<M> AutomergeTypedCollection<M>
where
    M: ProstMessage + Serialize + DeserializeOwned + Default + Clone,
{
    fn new(store: Arc<AutomergeStore>, collection_name: &str) -> Self {
        Self {
            store,
            prefix: format!("{}:", collection_name),
            _phantom: PhantomData,
        }
    }

    fn prefixed_key(&self, doc_id: &str) -> String {
        format!("{}{}", self.prefix, doc_id)
    }

    fn strip_prefix<'a>(&self, key: &'a str) -> Option<&'a str> {
        key.strip_prefix(&self.prefix)
    }
}

#[cfg(feature = "automerge-backend")]
impl<M> TypedCollection<M> for AutomergeTypedCollection<M>
where
    M: ProstMessage + Serialize + DeserializeOwned + Default + Clone,
{
    fn upsert(&self, doc_id: &str, message: &M) -> Result<()> {
        // Read existing doc (if any) so the new write forks from it,
        // preserving CRDT history. Closes peat#896: prior code passed
        // a brand-new `Automerge::new()` through every upsert, so each
        // write had a fresh actor with no causal link to earlier
        // writes — sync peers received byte-identical snapshots and
        // the CRDT merge picked stale values deterministically.
        let key = self.prefixed_key(doc_id);
        let existing = self.store.get(&key)?;
        let doc = message_to_automerge_into(message, existing.as_ref())?;
        self.store.put(&key, &doc)
    }

    fn get(&self, doc_id: &str) -> Result<Option<M>> {
        match self.store.get(&self.prefixed_key(doc_id))? {
            Some(doc) => {
                let message = automerge_to_message(&doc)?;
                Ok(Some(message))
            }
            None => Ok(None),
        }
    }

    fn delete(&self, doc_id: &str) -> Result<()> {
        self.store.delete(&self.prefixed_key(doc_id))
    }

    fn scan(&self) -> Result<Vec<(String, M)>> {
        let docs = self.store.scan_prefix(&self.prefix)?;
        let mut results = Vec::new();

        for (key, doc) in docs {
            if let Some(doc_id) = self.strip_prefix(&key) {
                let message = automerge_to_message(&doc)?;
                results.push((doc_id.to_string(), message));
            }
        }

        Ok(results)
    }

    fn find(&self, predicate: Box<dyn Fn(&M) -> bool + Send>) -> Result<Vec<(String, M)>> {
        let all_docs = self.scan()?;
        Ok(all_docs
            .into_iter()
            .filter(|(_, msg)| predicate(msg))
            .collect())
    }

    fn count(&self) -> Result<usize> {
        Ok(self.scan()?.len())
    }
}

/// Implement CrdtCapable trait to provide typed collections with CRDT semantics
#[cfg(feature = "automerge-backend")]
impl CrdtCapable for AutomergeBackend {
    fn typed_collection<M>(&self, name: &str) -> Arc<dyn TypedCollection<M>>
    where
        M: ProstMessage + Serialize + DeserializeOwned + Default + Clone + 'static,
    {
        Arc::new(AutomergeTypedCollection::new(Arc::clone(&self.store), name))
    }
}

/// Implement SyncCapable trait for background synchronization
///
/// Phase 5: Provides lifecycle management for P2P sync with Iroh transport.
#[cfg(feature = "automerge-backend")]
impl SyncCapable for AutomergeBackend {
    fn start_sync(&self) -> Result<()> {
        // Check if transport is available
        if self.transport.is_none() || self.sync_coordinator.is_none() {
            anyhow::bail!(
                "Cannot start sync: backend created without transport (use with_transport())"
            );
        }

        // Check if already syncing
        if self
            .sync_active
            .compare_exchange(false, true, Ordering::SeqCst, Ordering::Relaxed)
            .is_err()
        {
            anyhow::bail!("Sync already active");
        }

        // Phase 6.1: Start accept loop to receive incoming connections
        if let Some(transport) = &self.iroh_transport {
            // Only start if not already running (may have been started by initialize())
            if !transport.is_accept_loop_running() {
                transport.start_accept_loop()?;
            }
        }

        // Issue #438 Phase 2: Initialize sync channel manager for persistent channels
        // Issue #435: Wire up coordinator to use channel manager for all sends
        {
            let transport = self.transport.clone().unwrap();
            let coordinator = self.sync_coordinator.clone().unwrap();
            let manager = Arc::new(super::sync_channel::SyncChannelManager::new(
                transport,
                Arc::clone(&coordinator),
            ));
            // Enable coordinator to use persistent channels for sending
            coordinator.set_channel_manager(Arc::clone(&manager));
            *self.channel_manager.write().unwrap() = Some(manager);
            tracing::debug!("SyncChannelManager initialized with bidirectional wiring");
        }

        // Phase 6.2: Spawn incoming sync handler task
        //
        // Note: For Phase 6.2, we rely on manual sync triggering via sync_document().
        // The incoming handler is invoked per-stream when a peer sends us a sync message.
        // The accept loop in IrohTransport accepts connections, and we need a stream
        // accept loop for each connection to handle incoming sync messages.
        //
        // Issue #346: Use event-based handler spawning to avoid race conditions.
        // Previously, we only polled every 100ms which could miss sync messages
        // sent immediately after connection establishment.
        let iroh_for_events = self.iroh_transport.clone().unwrap();
        let transport = self.transport.clone().unwrap();
        let coordinator = self.sync_coordinator.clone().unwrap();
        let sync_active = Arc::clone(&self.sync_active);
        let active_handlers = Arc::clone(&self.active_sync_handlers);

        // Issue #346: Event-based handler spawning
        // Subscribe to connection events and spawn handlers IMMEDIATELY when peers connect.
        // This eliminates the race condition where sync messages arrive before the
        // polling-based handler has a chance to run.
        let transport_events = iroh_for_events.subscribe_peer_events();
        let transport_for_events = Arc::clone(&transport);
        let coordinator_for_events = Arc::clone(&coordinator);
        let sync_active_for_events = Arc::clone(&sync_active);
        let active_handlers_for_events = Arc::clone(&active_handlers);

        // De-dup window for duplicate Connected events per peer.
        // The connect-path and accept-path can both emit Connected
        // for the same peer when racing on a recycle reconnect (seen
        // in 2026-05-10 sync diagnostic: a single node observed two
        // Connected events for the same peer ID within 2.6s, each
        // spawning its own proactive-push task — wasted bandwidth
        // and a confusing log trail). 5s is comfortably wider than
        // the observed race window without suppressing legitimate
        // reconnect-after-recycle events (those are >50s apart).
        let recent_connects: Arc<
            std::sync::RwLock<std::collections::HashMap<EndpointId, std::time::Instant>>,
        > = Arc::new(std::sync::RwLock::new(std::collections::HashMap::new()));

        tokio::spawn(async move {
            let mut events = transport_events;
            while let Some(event) = events.recv().await {
                if !sync_active_for_events.load(Ordering::Relaxed) {
                    break;
                }
                // Disconnected events clear the dedup entry so a
                // legitimate reconnect-after-flap (BLE link drop,
                // network blip, peer-side force-stop+restart, iroh
                // transport reset) gets the full proactive-push
                // treatment even if it lands inside the 5s window.
                // Without this, the recycle-driven dedup wrongly
                // suppresses externally-driven reconnects too —
                // which is exactly the path the marker-tombstone
                // sync this PR enables depends on.
                if let crate::network::iroh_transport::TransportPeerEvent::Disconnected {
                    endpoint_id,
                    ..
                } = event
                {
                    let mut map = recent_connects.write().expect("recent_connects poisoned");
                    if map.remove(&endpoint_id).is_some() {
                        tracing::debug!(
                            peer = ?endpoint_id,
                            "Cleared dedup entry on Disconnected — next Connected will be treated as fresh"
                        );
                    }
                    continue;
                }
                if let crate::network::iroh_transport::TransportPeerEvent::Connected {
                    endpoint_id,
                    ..
                } = event
                {
                    // Skip duplicate Connected within the dedup window
                    // — otherwise both the connect and accept paths
                    // double-spawn proactive pushes for the same peer.
                    // A real disconnect+reconnect cycle is NOT suppressed
                    // because the Disconnected arm above cleared the
                    // dedup entry.
                    let now = std::time::Instant::now();
                    let is_duplicate = Self::check_and_record_connect(
                        &recent_connects,
                        endpoint_id,
                        now,
                        CONNECTED_DEDUP_WINDOW,
                    );
                    if is_duplicate {
                        tracing::debug!(
                            peer = ?endpoint_id,
                            "Suppressing duplicate Connected event within dedup window"
                        );
                        continue;
                    }

                    // Spawn handler immediately for new connection
                    Self::spawn_sync_handler_for_peer(
                        endpoint_id,
                        &transport_for_events,
                        &coordinator_for_events,
                        &sync_active_for_events,
                        &active_handlers_for_events,
                    );

                    // Explicitly push all local documents to the new peer.
                    // The sync handler above handles incoming streams, but we also
                    // need to proactively push our documents to ensure bidirectional sync.
                    let coordinator_for_push = Arc::clone(&coordinator_for_events);
                    let push_peer_id = endpoint_id;
                    tokio::spawn(async move {
                        tokio::time::sleep(tokio::time::Duration::from_millis(500)).await;
                        match coordinator_for_push
                            .sync_all_documents_with_peer(push_peer_id)
                            .await
                        {
                            Ok(()) => {
                                // Success-side tracing — without this,
                                // a silently-failing proactive push is
                                // indistinguishable from a successful
                                // one in logs. Diagnostic log spammed
                                // ~50 connect events with zero info on
                                // whether any push actually ran.
                                tracing::info!(
                                    peer = ?push_peer_id,
                                    "Proactive document push to peer succeeded"
                                );
                            }
                            Err(e) => {
                                tracing::warn!(
                                    "Proactive document push to peer {:?} failed: {}",
                                    push_peer_id,
                                    e
                                );
                            }
                        }
                    });

                    // ADR-034 Phase 2: Exchange tombstones with new peer
                    // This ensures deletions are synchronized when peers connect
                    let coordinator_for_tombstones = Arc::clone(&coordinator_for_events);
                    let peer_id_for_tombstones = endpoint_id;
                    tokio::spawn(async move {
                        // Small delay to let connection fully establish
                        tokio::time::sleep(tokio::time::Duration::from_millis(100)).await;
                        if let Err(e) = coordinator_for_tombstones
                            .sync_tombstones_with_peer(peer_id_for_tombstones)
                            .await
                        {
                            tracing::debug!(
                                "Tombstone exchange with peer {:?} failed: {}",
                                peer_id_for_tombstones,
                                e
                            );
                        }
                    });
                }
            }
            tracing::debug!("Event-based sync handler spawner stopped");
        });

        // Issue #346: Polling fallback - runs infrequently since event-based spawning is primary.
        // The longer interval (5s) ensures handshakes complete before we try to sync.
        // This is just a safety net in case Connected events are missed.
        let task = tokio::spawn(async move {
            // Initial delay to allow handshakes to complete
            tokio::time::sleep(tokio::time::Duration::from_secs(2)).await;

            while sync_active.load(Ordering::Relaxed) {
                let peer_ids = transport.connected_peers();

                for peer_id in peer_ids {
                    // Use the same helper function as event-based spawning
                    Self::spawn_sync_handler_for_peer(
                        peer_id,
                        &transport,
                        &coordinator,
                        &sync_active,
                        &active_handlers,
                    );
                }

                // Issue #346: Increased interval to 5s to ensure handshakes complete
                // Primary sync handler spawning is event-based (Connected events)
                tokio::time::sleep(tokio::time::Duration::from_secs(5)).await;
            }

            tracing::debug!("Incoming sync handler manager stopped");
        });

        *self.incoming_handler_task.write().unwrap() = Some(task);

        // Phase 6.5: Spawn change propagation + transitive-gossip task
        //
        // Subscribes to the origin-tagged change broadcast (peat-mesh#151 /
        // peat#891) so the pusher fires on BOTH local writes AND sync-
        // received remote applies — the latter is what makes transitive
        // gossip work for hub-and-spoke topologies. The origin attribution
        // carried on each event drives the per-peer filter:
        //
        // - ChangeOrigin::Local: push to every connected peer.
        // - ChangeOrigin::Remote(src): push to every connected peer EXCEPT
        //   `src` (the source peer already has the doc; pushing back forms
        //   the same #115 ping-pong that the original suppression closed,
        //   except now we close it at the consumer's edge with full
        //   attribution rather than suppressing the broadcast).
        {
            let store_for_changes = Arc::clone(&self.store);
            let coordinator_for_changes: Arc<peat_mesh::storage::AutomergeSyncCoordinator> =
                Arc::clone(self.sync_coordinator.as_ref().unwrap());
            let transport_for_changes = self.transport.clone().unwrap();
            let sync_active_for_changes = Arc::clone(&self.sync_active);
            tokio::spawn(async move {
                use peat_mesh::storage::ChangeOrigin;
                let mut change_rx = store_for_changes.subscribe_to_changes_with_origin();
                // Debounce is keyed by (doc_key, peer_id) — different peers
                // are independent destinations. A doc that was just pushed
                // to bravo should still be pushable to charlie without
                // waiting the debounce window. Pre-gossip the key was just
                // doc_key, which was fine for single-direction local writes
                // but would over-suppress under gossip.
                //
                // Bounded LRU with lazy TTL sweep (peat#909 QA): every
                // unique (doc, peer) pair that passes through the
                // propagation task leaves a `last_push` entry, so on a
                // long-running hub node with document or peer churn the
                // map would otherwise grow without bound. Cap by entry
                // count and sweep entries older than the retention
                // window on each insert. Sized for ~32 documents × ~32
                // peers steady-state with headroom; well above any
                // realistic working set, low enough that map ops stay
                // O(small).
                const LAST_PUSH_CAP: usize = 4096;
                let debounce = std::time::Duration::from_secs(2);
                let retain_for = debounce * 5; // any entry older than this
                                               // can't influence debounce
                let mut last_push: std::collections::HashMap<(String, String), std::time::Instant> =
                    std::collections::HashMap::with_capacity(64);
                while sync_active_for_changes.load(Ordering::Relaxed) {
                    match change_rx.recv().await {
                        Ok(evt) => {
                            let doc_key = evt.key;
                            let exclude: Option<String> = match evt.origin {
                                ChangeOrigin::Local => None,
                                ChangeOrigin::Remote(src) => Some(src),
                            };

                            let peer_ids = transport_for_changes.connected_peers();
                            let now = std::time::Instant::now();

                            // Lazy TTL sweep — drop entries older than the
                            // retention window so the map stays bounded under
                            // long-running document/peer churn.
                            last_push.retain(|_, last| now.duration_since(*last) < retain_for);
                            // Hard cap as a backstop: if the sweep didn't free
                            // enough (e.g. all entries are fresh because of a
                            // burst), evict ~25% of the oldest to keep ops
                            // O(small) and avoid an unbounded steady-state.
                            if last_push.len() >= LAST_PUSH_CAP {
                                let target_drop = last_push.len() / 4;
                                let mut entries: Vec<_> = last_push.iter().collect();
                                entries.sort_by_key(|(_, t)| **t);
                                let drop_keys: Vec<_> = entries
                                    .into_iter()
                                    .take(target_drop)
                                    .map(|(k, _)| k.clone())
                                    .collect();
                                for k in drop_keys {
                                    last_push.remove(&k);
                                }
                            }

                            for peer_id in peer_ids {
                                let peer_str = peer_id.to_string();
                                // Gossip filter: don't echo back to the source
                                // peer. This is the peat#891 fix — without it,
                                // alpha would push bravo's just-received doc
                                // back to bravo (wasted bandwidth at best,
                                // bucket-saturation at worst per peat-mesh#115).
                                if exclude.as_ref() == Some(&peer_str) {
                                    continue;
                                }
                                // Per-(doc, peer) debounce to suppress
                                // intra-window resends without blocking
                                // cross-peer fan-out.
                                let entry_key = (doc_key.clone(), peer_str);
                                if let Some(last) = last_push.get(&entry_key) {
                                    if now.duration_since(*last) < debounce {
                                        continue;
                                    }
                                }
                                last_push.insert(entry_key, now);

                                if let Err(e) = coordinator_for_changes
                                    .sync_document_with_peer(&doc_key, peer_id)
                                    .await
                                {
                                    tracing::trace!(
                                        "Change propagation failed for '{}' → {:?}: {}",
                                        doc_key,
                                        peer_id,
                                        e
                                    );
                                }
                            }
                        }
                        Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => {
                            // Acceptable — we'll catch up on next change
                        }
                        Err(_) => break,
                    }
                }
                tracing::debug!("Change propagation / gossip task stopped");
            });
        }

        // Phase 6.4: Spawn incoming heartbeat handler task
        //
        // Accept incoming heartbeat messages on unidirectional streams
        // Uses continuous per-connection handlers for low latency
        let transport_heartbeat_rx = self.transport.clone().unwrap();
        let coordinator_heartbeat_rx = self.sync_coordinator.clone().unwrap();
        let sync_active_heartbeat_rx = Arc::clone(&self.sync_active);
        let active_heartbeat_handlers = Arc::clone(&self.active_heartbeat_handlers);

        let heartbeat_rx_task = tokio::spawn(async move {
            // Issue #346: Initial delay to allow handshakes to complete
            tokio::time::sleep(tokio::time::Duration::from_secs(2)).await;

            while sync_active_heartbeat_rx.load(Ordering::Relaxed) {
                let peer_ids = transport_heartbeat_rx.connected_peers();

                for peer_id in peer_ids {
                    // Skip if we already have a handler for this connection
                    {
                        let handlers = active_heartbeat_handlers.read().unwrap();
                        if handlers.contains(&peer_id) {
                            continue;
                        }
                    }

                    // Get connection and spawn continuous handler
                    if let Some(conn) = transport_heartbeat_rx.get_connection(&peer_id) {
                        // Mark as having active handler
                        active_heartbeat_handlers.write().unwrap().insert(peer_id);

                        let coordinator_clone = Arc::clone(&coordinator_heartbeat_rx);
                        let sync_active_clone = Arc::clone(&sync_active_heartbeat_rx);
                        let active_handlers_clone = Arc::clone(&active_heartbeat_handlers);
                        let handler_peer_id = peer_id;

                        // Spawn continuous handler that loops accepting heartbeat streams
                        tokio::spawn(async move {
                            tracing::debug!(
                                "Started continuous heartbeat handler for peer {:?}",
                                handler_peer_id
                            );

                            while sync_active_clone.load(Ordering::Relaxed) {
                                match conn.accept_uni().await {
                                    Ok(recv) => {
                                        let coord = Arc::clone(&coordinator_clone);
                                        let stream_peer_id = handler_peer_id;
                                        tokio::spawn(async move {
                                            if let Err(e) = coord
                                                .handle_incoming_heartbeat_stream(
                                                    stream_peer_id,
                                                    recv,
                                                )
                                                .await
                                            {
                                                tracing::trace!(
                                                    "Error handling heartbeat stream: {}",
                                                    e
                                                );
                                            }
                                        });
                                    }
                                    Err(e) => {
                                        tracing::debug!(
                                            "Heartbeat handler for {:?} exiting: {}",
                                            handler_peer_id,
                                            e
                                        );
                                        break;
                                    }
                                }
                            }

                            active_handlers_clone
                                .write()
                                .unwrap()
                                .remove(&handler_peer_id);
                            tracing::debug!(
                                "Stopped continuous heartbeat handler for peer {:?}",
                                handler_peer_id
                            );
                        });
                    }
                }

                // Issue #346: Increased interval to ensure handshakes complete first
                tokio::time::sleep(tokio::time::Duration::from_secs(5)).await;
            }

            tracing::debug!("Incoming heartbeat handler manager stopped");
        });

        *self.heartbeat_receiver_task.write().unwrap() = Some(heartbeat_rx_task);

        // Phase 6.3: Spawn automatic sync task for outgoing sync
        //
        // Subscribe to document change notifications and automatically sync
        // changed documents with all connected peers.
        let mut change_rx = self.store.subscribe_to_changes();
        let coordinator = self.sync_coordinator.clone().unwrap();
        let sync_active = Arc::clone(&self.sync_active);
        let store_for_resync = Arc::clone(&self.store);
        // Issue #438 Phase 2: Use persistent channels for batch sync
        let channel_manager = self.channel_manager.read().unwrap().clone().unwrap();

        let auto_task = tokio::spawn(async move {
            use std::time::{Duration, Instant};

            tracing::debug!("Automatic sync task started (batch mode with persistent channels)");

            // Issue #346: Track last resync time to prevent thundering herd
            let mut last_resync: Option<Instant> = None;
            const RESYNC_COOLDOWN: Duration = Duration::from_secs(5);

            // Issue #438: Batch sync parameters
            const BATCH_WINDOW: Duration = Duration::from_millis(50);
            const MAX_BATCH_SIZE: usize = 20;

            // Pending documents for batch sync
            let mut pending_docs: Vec<String> = Vec::new();
            let mut window_start = Instant::now();

            while sync_active.load(Ordering::Relaxed) {
                // Use timeout to implement batch window
                let timeout = if pending_docs.is_empty() {
                    // No pending docs, wait indefinitely for next change
                    Duration::from_secs(3600) // 1 hour (effectively infinite)
                } else {
                    // Have pending docs, wait for remaining window time
                    BATCH_WINDOW.saturating_sub(window_start.elapsed())
                };

                match tokio::time::timeout(timeout, change_rx.recv()).await {
                    Ok(Ok(doc_key)) => {
                        // Document changed, add to pending batch
                        if pending_docs.is_empty() {
                            window_start = Instant::now();
                        }

                        // Avoid duplicates in the same batch
                        if !pending_docs.contains(&doc_key) {
                            pending_docs.push(doc_key);
                        }

                        // Flush if batch is full
                        if pending_docs.len() >= MAX_BATCH_SIZE {
                            tracing::debug!(
                                "Batch full ({} docs), flushing via persistent channels",
                                pending_docs.len()
                            );
                            let doc_refs: Vec<&str> =
                                pending_docs.iter().map(|s| s.as_str()).collect();
                            // Issue #438 Phase 2: Use persistent channels
                            match coordinator.create_batch_for_documents(&doc_refs) {
                                Ok(batch) => {
                                    if let Err(e) = channel_manager.broadcast(&batch).await {
                                        tracing::warn!("Batch broadcast failed: {}", e);
                                    }
                                }
                                Err(e) => {
                                    tracing::warn!("Batch creation failed: {}", e);
                                }
                            }
                            pending_docs.clear();
                        }
                    }
                    Ok(Err(tokio::sync::broadcast::error::RecvError::Lagged(n))) => {
                        // Issue #346: When lagged, we MUST resync all documents
                        tracing::warn!("Change notification lagged, skipped {} messages", n);

                        // Flush any pending docs first
                        if !pending_docs.is_empty() {
                            let doc_refs: Vec<&str> =
                                pending_docs.iter().map(|s| s.as_str()).collect();
                            // Issue #438 Phase 2: Use persistent channels
                            if let Ok(batch) = coordinator.create_batch_for_documents(&doc_refs) {
                                let _ = channel_manager.broadcast(&batch).await;
                            }
                            pending_docs.clear();
                        }

                        // Back-pressure: Check if we recently resynced
                        let should_resync = match last_resync {
                            Some(last) if last.elapsed() < RESYNC_COOLDOWN => {
                                tracing::debug!(
                                    "Skipping resync - cooldown active ({:?} remaining)",
                                    RESYNC_COOLDOWN - last.elapsed()
                                );
                                false
                            }
                            _ => true,
                        };

                        if should_resync {
                            // Add jitter (0-500ms) to spread load across nodes
                            let jitter_ms = rand::random::<u64>() % 500;
                            tokio::time::sleep(Duration::from_millis(jitter_ms)).await;

                            last_resync = Some(Instant::now());

                            // Issue #438 Phase 2: Use persistent channels for resync
                            let store_clone = Arc::clone(&store_for_resync);
                            let coordinator_clone = coordinator.clone();
                            let channel_manager_clone = Arc::clone(&channel_manager);
                            tokio::spawn(async move {
                                if let Ok(all_docs) = store_clone.scan_prefix("") {
                                    tracing::info!(
                                        "Batch resyncing {} documents via persistent channels",
                                        all_docs.len()
                                    );
                                    // Collect all doc keys
                                    let doc_keys: Vec<String> =
                                        all_docs.into_iter().map(|(k, _)| k).collect();
                                    let doc_refs: Vec<&str> =
                                        doc_keys.iter().map(|s| s.as_str()).collect();

                                    // Send as batch(es) via persistent channels
                                    for chunk in doc_refs.chunks(MAX_BATCH_SIZE) {
                                        if let Ok(batch) =
                                            coordinator_clone.create_batch_for_documents(chunk)
                                        {
                                            if let Err(e) =
                                                channel_manager_clone.broadcast(&batch).await
                                            {
                                                tracing::debug!(
                                                    "Batch resync broadcast failed: {}",
                                                    e
                                                );
                                            }
                                        }
                                        // Yield to prevent starving other tasks
                                        tokio::task::yield_now().await;
                                    }
                                }
                            });
                        }
                    }
                    Ok(Err(tokio::sync::broadcast::error::RecvError::Closed)) => {
                        // Channel closed
                        tracing::debug!("Change notification channel closed");
                        break;
                    }
                    Err(_elapsed) => {
                        // Batch window expired, flush pending docs
                        if !pending_docs.is_empty() {
                            tracing::debug!(
                                "Batch window expired ({} docs), flushing via persistent channels",
                                pending_docs.len()
                            );
                            let doc_refs: Vec<&str> =
                                pending_docs.iter().map(|s| s.as_str()).collect();
                            // Issue #438 Phase 2: Use persistent channels
                            match coordinator.create_batch_for_documents(&doc_refs) {
                                Ok(batch) => {
                                    if let Err(e) = channel_manager.broadcast(&batch).await {
                                        tracing::warn!("Batch broadcast failed: {}", e);
                                    }
                                }
                                Err(e) => {
                                    tracing::warn!("Batch creation failed: {}", e);
                                }
                            }
                            pending_docs.clear();
                        }
                    }
                }
            }

            // Flush any remaining docs on shutdown
            if !pending_docs.is_empty() {
                tracing::debug!(
                    "Flushing {} pending docs on shutdown via persistent channels",
                    pending_docs.len()
                );
                let doc_refs: Vec<&str> = pending_docs.iter().map(|s| s.as_str()).collect();
                // Issue #438 Phase 2: Use persistent channels
                if let Ok(batch) = coordinator.create_batch_for_documents(&doc_refs) {
                    let _ = channel_manager.broadcast(&batch).await;
                }
            }

            tracing::debug!("Automatic sync task stopped (persistent channels)");
        });

        *self.auto_sync_task.write().unwrap() = Some(auto_task);

        // Phase 6.4: Spawn heartbeat task for partition detection
        //
        // Periodically send heartbeats to all connected peers to detect partitions
        let coordinator_heartbeat = self.sync_coordinator.clone().unwrap();
        let sync_active_heartbeat = Arc::clone(&self.sync_active);

        let heartbeat_task = tokio::spawn(async move {
            tracing::debug!("Heartbeat task started");

            // Get heartbeat interval from partition detector config
            let heartbeat_interval = coordinator_heartbeat
                .partition_detector()
                .config()
                .heartbeat_interval;

            while sync_active_heartbeat.load(Ordering::Relaxed) {
                // Send heartbeats to all connected peers
                if let Err(e) = coordinator_heartbeat.send_heartbeats_to_all_peers().await {
                    tracing::debug!("Error sending heartbeats: {}", e);
                }

                // Check for partition timeouts
                let partitioned_peers = coordinator_heartbeat.check_partition_timeouts();
                if !partitioned_peers.is_empty() {
                    tracing::warn!("Detected {} partitioned peers", partitioned_peers.len());
                }

                // Sleep until next heartbeat interval
                tokio::time::sleep(heartbeat_interval).await;
            }

            tracing::debug!("Heartbeat task stopped");
        });

        *self.heartbeat_task.write().unwrap() = Some(heartbeat_task);

        // Issue #435: Connection recycling task to mitigate upstream iroh memory leak
        //
        // The iroh library has a memory leak (iroh#3565) where WeakConnectionHandle
        // references accumulate in RttActor's MergeUnbounded stream. This causes
        // ~0.875 MB/sec memory growth during active sync operations.
        //
        // Workaround: Periodically disconnect and reconnect peers to clear accumulated
        // state. The reconnection manager handles automatic reconnection.
        let recycle_interval = crate::network::iroh_transport::connection_recycle_interval_secs();
        if recycle_interval > 0 {
            let transport_for_recycle = self.iroh_transport.clone().unwrap();
            let sync_active_recycle = Arc::clone(&self.sync_active);

            tokio::spawn(async move {
                tracing::info!(
                    interval_secs = recycle_interval,
                    "Starting connection recycling task (Issue #435 memory leak workaround)"
                );

                let recycle_duration = std::time::Duration::from_secs(recycle_interval);

                // Initial delay - don't recycle immediately after startup
                tokio::time::sleep(recycle_duration).await;

                while sync_active_recycle.load(Ordering::Relaxed) {
                    let recycled = transport_for_recycle.recycle_old_connections(recycle_duration);
                    if recycled > 0 {
                        tracing::debug!(
                            recycled = recycled,
                            "Connection recycling complete, reconnection will happen automatically"
                        );
                    }

                    // Check every 10 seconds but only recycle connections older than recycle_interval
                    tokio::time::sleep(std::time::Duration::from_secs(10)).await;
                }

                tracing::debug!("Connection recycling task stopped");
            });
        }

        Ok(())
    }

    fn stop_sync(&self) -> Result<()> {
        // Mark as inactive
        if !self.sync_active.swap(false, Ordering::SeqCst) {
            anyhow::bail!("Sync is not active");
        }

        // Phase 6.1: Stop accept loop
        if let Some(transport) = &self.iroh_transport {
            // Ignore error if accept loop already stopped
            let _ = transport.stop_accept_loop();
        }

        // Issue #438 Phase 2: Clear channel manager (channels close when Arc drops)
        if let Some(manager) = self.channel_manager.write().unwrap().take() {
            // Spawn async cleanup task
            tokio::spawn(async move {
                manager.shutdown().await;
            });
        }

        // TODO Phase 6.2: Signal background sync task to stop and wait for completion

        Ok(())
    }

    fn sync_stats(&self) -> Result<SyncStats> {
        let peer_count = self
            .iroh_transport
            .as_ref()
            .map(|t| t.peer_count())
            .unwrap_or(0);

        // Get statistics from sync coordinator if available
        let (bytes_sent, bytes_received, last_sync) =
            if let Some(coordinator) = &self.sync_coordinator {
                let bytes_sent = coordinator.total_bytes_sent();
                let bytes_received = coordinator.total_bytes_received();

                // Find the most recent sync timestamp across all peers
                let last_sync = coordinator
                    .all_peer_stats()
                    .values()
                    .filter_map(|stats| stats.last_sync)
                    .max();

                (bytes_sent, bytes_received, last_sync)
            } else {
                // Fallback to local counters if no coordinator
                (
                    self.bytes_sent.load(Ordering::Relaxed),
                    self.bytes_received.load(Ordering::Relaxed),
                    None,
                )
            };

        Ok(SyncStats {
            peer_count,
            bytes_sent,
            bytes_received,
            last_sync,
        })
    }
}

/// Implement HierarchicalStorageCapable trait for hierarchical aggregation mode
///
/// This enables backend-agnostic access to SummaryStorage and CommandStorage,
/// eliminating the need for downcast patterns in peat-sim.
#[cfg(feature = "automerge-backend")]
impl HierarchicalStorageCapable for AutomergeBackend {
    fn summary_storage(&self) -> Arc<dyn SummaryStorage> {
        Arc::new(AutomergeSummaryStorage::new(Arc::clone(&self.store)))
    }

    fn command_storage(&self) -> Arc<dyn CommandStorage> {
        Arc::new(AutomergeCommandStorage::new(Arc::clone(&self.store)))
    }
}

#[cfg(all(test, feature = "automerge-backend"))]
mod tests {
    use super::*;
    use tempfile::TempDir;

    fn create_test_backend() -> (AutomergeBackend, TempDir) {
        let temp_dir = TempDir::new().unwrap();
        let store = Arc::new(AutomergeStore::open(temp_dir.path()).unwrap());
        let backend = AutomergeBackend::new(store);
        (backend, temp_dir)
    }

    #[test]
    fn test_backend_collection_creation() {
        let (backend, _temp) = create_test_backend();

        let collection = backend.collection("test");
        assert!(collection.count().unwrap() == 0);
    }

    #[test]
    fn test_backend_collection_caching() {
        let (backend, _temp) = create_test_backend();

        let coll1 = backend.collection("test");
        let coll2 = backend.collection("test");

        // Both should point to the same cached collection
        assert_eq!(Arc::as_ptr(&coll1), Arc::as_ptr(&coll2));
    }

    #[test]
    fn test_backend_list_collections() {
        let (backend, _temp) = create_test_backend();

        assert_eq!(backend.list_collections().len(), 0);

        backend.collection("coll1");
        backend.collection("coll2");

        let collections = backend.list_collections();
        assert_eq!(collections.len(), 2);
        assert!(collections.contains(&"coll1".to_string()));
        assert!(collections.contains(&"coll2".to_string()));
    }

    #[test]
    fn test_backend_operations_via_trait() {
        let (backend, _temp) = create_test_backend();

        let collection = backend.collection("test");

        // Test CRUD via trait interface
        collection.upsert("doc1", b"data1".to_vec()).unwrap();
        let retrieved = collection.get("doc1").unwrap().unwrap();
        assert_eq!(retrieved, b"data1");

        collection.delete("doc1").unwrap();
        assert!(collection.get("doc1").unwrap().is_none());
    }

    #[test]
    fn test_backend_flush_and_close() {
        let (backend, _temp) = create_test_backend();

        // Flush should succeed (no-op in Phase 1)
        assert!(backend.flush().is_ok());

        // Close should succeed
        assert!(backend.close().is_ok());
    }

    // Phase 2: CRDT Integration Tests
    use peat_schema::common::v1::Position;
    use peat_schema::node::v1::NodeState;

    #[test]
    fn test_typed_collection_crdt_upsert_get() {
        use crate::storage::capabilities::CrdtCapable;

        let (backend, _temp) = create_test_backend();
        let nodes: Arc<dyn TypedCollection<NodeState>> = backend.typed_collection("nodes");

        let node = NodeState {
            position: Some(Position {
                latitude: 37.7749,
                longitude: -122.4194,
                altitude: 100.0,
            }),
            fuel_minutes: 60,
            health: 1,
            phase: 1,
            cell_id: Some("cell-1".to_string()),
            zone_id: None,
            timestamp: None,
        };

        nodes.upsert("node-1", &node).unwrap();
        let retrieved = nodes.get("node-1").unwrap().unwrap();

        assert_eq!(retrieved.fuel_minutes, 60);
        assert_eq!(retrieved.cell_id, Some("cell-1".to_string()));
        assert!(retrieved.position.is_some());
    }

    /// Reproducer for peat#896 — sequential upserts to the same key must
    /// accumulate Automerge history. The sync layer at
    /// `peat_mesh::storage::automerge_sync::initiate_sync_inner` reads
    /// the stored doc via `store.get(key)?.save()` and sends those bytes
    /// to peers. If history is dropped, every write produces a
    /// byte-identical (or near-identical) snapshot and the receiver's
    /// CRDT merge picks one actor's value deterministically — which can
    /// be the wrong/stale one (see peat#896 trace logs showing
    /// `initiate_sync_inner: got doc, len=164` constant across writes
    /// while alpha's view of bravo's counter stays frozen for 15-20 s
    /// at a time).
    ///
    /// Expected post-fix behavior: the saved byte count is **strictly
    /// increasing** across sequential writes (history accumulates) and
    /// the most recent write's value is the one a fresh `get` returns.
    #[test]
    fn test_typed_collection_upsert_preserves_history_peat896() {
        use crate::storage::capabilities::CrdtCapable;

        let (backend, _temp) = create_test_backend();
        let nodes: Arc<dyn TypedCollection<NodeState>> = backend.typed_collection("nodes");
        let store = backend.automerge_store();

        let mut saved_lens = vec![];
        for counter in (90u32..=100).rev() {
            let node = NodeState {
                fuel_minutes: counter,
                cell_id: Some("alpha".to_string()),
                ..Default::default()
            };
            nodes.upsert("alpha", &node).unwrap();

            // Probe the same byte sequence the sync layer would send.
            let doc = store.get("nodes:alpha").unwrap().expect("doc must exist");
            saved_lens.push(doc.save().len());
        }

        // 11 sequential writes must produce 11 distinct snapshots, with
        // monotonically non-decreasing length (history grows). The bug
        // (peat#896) produced byte-identical snapshots — that's the
        // failure mode this assertion catches.
        let unique: std::collections::BTreeSet<_> = saved_lens.iter().copied().collect();
        assert!(
            unique.len() > 1,
            "all {} sequential upserts produced byte-identical snapshots ({:?}); \
             peat#896 says history is being dropped",
            saved_lens.len(),
            saved_lens.first()
        );
        for w in saved_lens.windows(2) {
            assert!(
                w[1] >= w[0],
                "saved bytes shrunk across sequential writes: {w:?}"
            );
        }

        // Final read-back must return the LAST value we wrote, not an
        // earlier one resolved via CRDT tiebreaker.
        let final_state = nodes.get("alpha").unwrap().expect("must read back");
        assert_eq!(
            final_state.fuel_minutes, 90,
            "expected most recent write (90), got {}",
            final_state.fuel_minutes
        );
    }

    #[test]
    fn test_typed_collection_crdt_scan() {
        use crate::storage::capabilities::CrdtCapable;

        let (backend, _temp) = create_test_backend();
        let nodes: Arc<dyn TypedCollection<NodeState>> = backend.typed_collection("nodes");

        let node1 = NodeState {
            fuel_minutes: 60,
            health: 1,
            phase: 1,
            cell_id: Some("cell-1".to_string()),
            ..Default::default()
        };

        let node2 = NodeState {
            fuel_minutes: 45,
            health: 1,
            phase: 2,
            cell_id: Some("cell-2".to_string()),
            ..Default::default()
        };

        nodes.upsert("node-1", &node1).unwrap();
        nodes.upsert("node-2", &node2).unwrap();

        let results = nodes.scan().unwrap();
        assert_eq!(results.len(), 2);

        let ids: Vec<String> = results.iter().map(|(id, _)| id.clone()).collect();
        assert!(ids.contains(&"node-1".to_string()));
        assert!(ids.contains(&"node-2".to_string()));
    }

    #[test]
    fn test_typed_collection_crdt_find_with_predicate() {
        use crate::storage::capabilities::CrdtCapable;

        let (backend, _temp) = create_test_backend();
        let nodes: Arc<dyn TypedCollection<NodeState>> = backend.typed_collection("nodes");

        let node1 = NodeState {
            fuel_minutes: 60,
            health: 1,
            phase: 1,
            cell_id: Some("cell-1".to_string()),
            ..Default::default()
        };

        let node2 = NodeState {
            fuel_minutes: 30,
            health: 1,
            phase: 1,
            cell_id: Some("cell-1".to_string()),
            ..Default::default()
        };

        let node3 = NodeState {
            fuel_minutes: 45,
            health: 1,
            phase: 1,
            cell_id: Some("cell-2".to_string()),
            ..Default::default()
        };

        nodes.upsert("node-1", &node1).unwrap();
        nodes.upsert("node-2", &node2).unwrap();
        nodes.upsert("node-3", &node3).unwrap();

        // Find nodes with low fuel
        let low_fuel_nodes = nodes.find(Box::new(|node| node.fuel_minutes < 40)).unwrap();

        assert_eq!(low_fuel_nodes.len(), 1);
        assert_eq!(low_fuel_nodes[0].1.fuel_minutes, 30);
    }

    #[test]
    fn test_typed_collection_delete() {
        use crate::storage::capabilities::CrdtCapable;

        let (backend, _temp) = create_test_backend();
        let nodes: Arc<dyn TypedCollection<NodeState>> = backend.typed_collection("nodes");

        let node = NodeState {
            fuel_minutes: 60,
            ..Default::default()
        };

        nodes.upsert("node-1", &node).unwrap();
        assert!(nodes.get("node-1").unwrap().is_some());

        nodes.delete("node-1").unwrap();
        assert!(nodes.get("node-1").unwrap().is_none());
    }

    // Phase 5: SyncCapable Trait Tests

    #[tokio::test]
    async fn test_backend_without_transport_cannot_sync() {
        use crate::storage::capabilities::SyncCapable;

        let (backend, _temp) = create_test_backend();

        // Should fail - no transport configured
        let result = backend.start_sync();
        assert!(result.is_err());
        assert!(result
            .unwrap_err()
            .to_string()
            .contains("without transport"));
    }

    #[tokio::test]
    async fn test_backend_with_transport_sync_lifecycle() {
        use crate::network::IrohTransport;
        use crate::storage::capabilities::SyncCapable;

        let temp_dir = TempDir::new().unwrap();
        let store = Arc::new(AutomergeStore::open(temp_dir.path()).unwrap());
        let transport = Arc::new(IrohTransport::new().await.unwrap());
        let backend = AutomergeBackend::with_transport(store, transport);

        // Start sync should succeed
        assert!(backend.start_sync().is_ok());

        // Starting again should fail
        let result = backend.start_sync();
        assert!(result.is_err());
        assert!(result.unwrap_err().to_string().contains("already active"));

        // Stop should succeed
        assert!(backend.stop_sync().is_ok());

        // Stopping again should fail
        let result = backend.stop_sync();
        assert!(result.is_err());
        assert!(result.unwrap_err().to_string().contains("not active"));
    }

    #[tokio::test]
    async fn test_sync_stats_without_transport() {
        use crate::storage::capabilities::SyncCapable;

        let (backend, _temp) = create_test_backend();

        let stats = backend.sync_stats().unwrap();
        assert_eq!(stats.peer_count, 0);
        assert_eq!(stats.bytes_sent, 0);
        assert_eq!(stats.bytes_received, 0);
        assert!(stats.last_sync.is_none());
    }

    #[tokio::test]
    async fn test_sync_stats_with_transport() {
        use crate::network::IrohTransport;
        use crate::storage::capabilities::SyncCapable;

        let temp_dir = TempDir::new().unwrap();
        let store = Arc::new(AutomergeStore::open(temp_dir.path()).unwrap());
        let transport = Arc::new(IrohTransport::new().await.unwrap());
        let backend = AutomergeBackend::with_transport(store, transport);

        let stats = backend.sync_stats().unwrap();
        assert_eq!(stats.peer_count, 0); // No peers connected yet
        assert_eq!(stats.bytes_sent, 0);
        assert_eq!(stats.bytes_received, 0);
    }

    /// Regression coverage for `check_and_record_connect` —
    /// the dedup helper that the event-driven sync spawner calls on
    /// every `TransportPeerEvent::Connected`. Exercises four
    /// behaviors the production loop relies on:
    ///   - first observation within the window passes
    ///   - second observation for the same peer within the window suppresses
    ///   - second observation outside the window passes (legitimate reconnect)
    ///   - second observation for a *different* peer within the window passes
    ///     (no accidental cross-peer dedup)
    ///
    /// If `CONNECTED_DEDUP_WINDOW` is tuned, or the helper's
    /// reaping logic regresses (e.g. `retain` swapped for a no-op),
    /// one of these assertions fails before the bug ships.
    mod connected_dedup_tests {
        use super::*;
        use std::time::{Duration, Instant};

        fn fresh_map() -> Arc<RwLock<HashMap<EndpointId, Instant>>> {
            Arc::new(RwLock::new(HashMap::new()))
        }

        fn peer(seed: u8) -> EndpointId {
            // EndpointId is iroh's Ed25519 public key — `from_bytes`
            // requires a valid Edwards curve point, not arbitrary
            // 32-byte patterns. iroh 0.98 dropped the caller-side `&mut R`
            // argument on `SecretKey::generate()`, which used to consume the
            // seeded `StdRng` here. Tests that need a deterministic NodeId
            // per `seed` now go through `IrohTransport::endpoint_id_from_seed`
            // (HKDF-SHA-256 KDF). The `seed: u64` is stringified so the
            // helper signature stays compatible.
            peat_mesh::network::iroh_transport::IrohTransport::endpoint_id_from_seed(&format!(
                "test-peer-{seed}"
            ))
        }

        #[test]
        fn first_connect_passes() {
            let map = fresh_map();
            let now = Instant::now();
            let is_dup = AutomergeBackend::check_and_record_connect(
                &map,
                peer(1),
                now,
                Duration::from_secs(5),
            );
            assert!(!is_dup, "first observation must not be marked duplicate");
        }

        #[test]
        fn second_connect_within_window_suppresses() {
            let map = fresh_map();
            let window = Duration::from_secs(5);
            let t0 = Instant::now();
            AutomergeBackend::check_and_record_connect(&map, peer(1), t0, window);
            let t1 = t0 + Duration::from_secs(2);
            let is_dup = AutomergeBackend::check_and_record_connect(&map, peer(1), t1, window);
            assert!(
                is_dup,
                "second observation within window MUST be marked duplicate"
            );
        }

        #[test]
        fn second_connect_outside_window_passes() {
            let map = fresh_map();
            let window = Duration::from_secs(5);
            let t0 = Instant::now();
            AutomergeBackend::check_and_record_connect(&map, peer(1), t0, window);
            // Slightly past the window — should re-register as fresh.
            let t1 = t0 + Duration::from_secs(6);
            let is_dup = AutomergeBackend::check_and_record_connect(&map, peer(1), t1, window);
            assert!(
                !is_dup,
                "legitimate reconnect after window must NOT be suppressed"
            );
        }

        #[test]
        fn different_peers_within_window_both_pass() {
            let map = fresh_map();
            let window = Duration::from_secs(5);
            let t0 = Instant::now();
            AutomergeBackend::check_and_record_connect(&map, peer(1), t0, window);
            let t1 = t0 + Duration::from_secs(1);
            let is_dup = AutomergeBackend::check_and_record_connect(&map, peer(2), t1, window);
            assert!(
                !is_dup,
                "dedup must be per-peer; different peer must not be suppressed"
            );
        }

        /// Reaping behavior: when a stale entry falls outside the
        /// window, the next call for that same peer treats it as
        /// fresh. Verifies the `retain` pass in
        /// `check_and_record_connect` is doing its job.
        #[test]
        fn stale_entries_get_reaped_on_subsequent_call() {
            let map = fresh_map();
            let window = Duration::from_secs(5);
            let t0 = Instant::now();
            AutomergeBackend::check_and_record_connect(&map, peer(1), t0, window);

            // Way past the window — entry should be reaped on next call.
            let t1 = t0 + Duration::from_secs(60);
            // Touch with a DIFFERENT peer so the reap runs without
            // implicitly insert-and-update for peer(1).
            AutomergeBackend::check_and_record_connect(&map, peer(2), t1, window);

            // peer(1) should no longer be in the map; verify by
            // confirming the next call for peer(1) is not flagged duplicate.
            let t2 = t1 + Duration::from_millis(1);
            let is_dup = AutomergeBackend::check_and_record_connect(&map, peer(1), t2, window);
            assert!(
                !is_dup,
                "stale entry reaped — peer(1) must register as fresh"
            );
        }

        /// Regression lock for the round-3 review finding: a real
        /// disconnect+reconnect (BLE flap, force-stop+restart) must
        /// NOT be suppressed by the dedup window. The Disconnected
        /// branch of the event loop clears the peer's map entry so
        /// the next Connected is treated as fresh. This test models
        /// that lifecycle directly against the helper: after a
        /// removal, a Connected within the dedup window passes.
        #[test]
        fn disconnect_then_reconnect_within_window_passes() {
            let map = fresh_map();
            let window = Duration::from_secs(5);
            let t0 = Instant::now();
            // Initial connect — records the peer.
            AutomergeBackend::check_and_record_connect(&map, peer(1), t0, window);
            // Simulate Disconnected → loop clears the peer's entry.
            map.write().unwrap().remove(&peer(1));
            // Real fast reconnect within the dedup window.
            let t1 = t0 + Duration::from_secs(2);
            let is_dup = AutomergeBackend::check_and_record_connect(&map, peer(1), t1, window);
            assert!(
                !is_dup,
                "real disconnect+reconnect within window must NOT be suppressed"
            );
        }

        /// The production constant matches the documented calibration.
        /// If a future tweak shifts the window, the assertion forces
        /// the change to flow through the comment + the regression
        /// rationale rather than silently slipping in.
        #[test]
        fn production_window_matches_documented_calibration() {
            assert_eq!(
                CONNECTED_DEDUP_WINDOW,
                Duration::from_secs(5),
                "CONNECTED_DEDUP_WINDOW changed — re-evaluate against the iroh \
                 connect/accept race window and the recycle interval, and update \
                 the comment at the constant's definition"
            );
        }
    }
}
