//! Automerge sync protocol implementation
//!
//! This module provides the sync coordinator that manages Automerge document
//! synchronization over Iroh QUIC streams.
//!
//! # Phase 4 Implementation
//!
//! Implements the Automerge sync protocol (https://arxiv.org/abs/2012.00472) over
//! Iroh P2P connections to enable CRDT document synchronization.
//!
//! ## Sync Flow
//!
//! ```text
//! Node A                          Node B
//!   │                               │
//!   ├─ Document updated             │
//!   ├─ generate_sync_message() ────→│
//!   │                               ├─ receive_sync_message()
//!   │                               ├─ apply changes
//!   │                               ├─ generate_sync_message()
//!   │←────────────────────────────┤
//!   ├─ receive_sync_message()       │
//!   ├─ apply changes                │
//!   │                               │
//!   ├─ Synced! ✅                   ├─ Synced! ✅
//! ```
//!
//! ## Wire Format
//!
//! Sync messages are sent over Iroh bidirectional streams with length prefixing:
//! ```text
//! [4 bytes: message length (u32, big-endian)][N bytes: serialized sync::Message]
//! ```

#[cfg(feature = "automerge-backend")]
use super::automerge_store::{AutomergeStore, ChangeOrigin};
#[cfg(feature = "automerge-backend")]
use super::flow_control::{FlowControlConfig, FlowControlStats, FlowController};
#[cfg(feature = "automerge-backend")]
use super::negentropy_sync::{NegentropySync, SyncItem};
#[cfg(feature = "automerge-backend")]
use super::partition_detection::PartitionDetector;
#[cfg(feature = "automerge-backend")]
use super::sync_errors::{SyncError, SyncErrorHandler};
#[cfg(feature = "automerge-backend")]
use super::sync_transport::{SyncRouter, SyncTransport};
#[cfg(feature = "automerge-backend")]
use crate::qos::{SyncMode, SyncModeRegistry};
#[cfg(feature = "automerge-backend")]
use anyhow::{Context, Result};
#[cfg(feature = "automerge-backend")]
use automerge::sync::{Message as SyncMessage, State as SyncState, SyncDoc};
#[cfg(feature = "automerge-backend")]
use automerge::Automerge;
#[cfg(feature = "automerge-backend")]
use iroh::endpoint::Connection;
#[cfg(feature = "automerge-backend")]
use iroh::EndpointId;
#[cfg(feature = "automerge-backend")]
use std::collections::{BTreeSet, HashMap};
#[cfg(feature = "automerge-backend")]
use std::sync::atomic::{AtomicU64, Ordering};
#[cfg(feature = "automerge-backend")]
use std::sync::{Arc, Mutex, RwLock, Weak};
#[cfg(feature = "automerge-backend")]
use std::time::SystemTime;
#[cfg(feature = "automerge-backend")]
#[allow(unused_imports)] // Used in sync message send/receive methods
use tokio::io::{AsyncReadExt, AsyncWriteExt};
#[cfg(feature = "automerge-backend")]
use tokio::sync::Notify;

/// Wire format message type prefix (Issue #355, ADR-034)
///
/// Used to distinguish between delta-based sync messages, state snapshots,
/// and tombstone sync messages.
///
/// # Wire Format v3 (ADR-034 Phase 2)
///
/// ```text
/// 0x00 = DeltaSync (original Automerge protocol)
/// 0x01 = StateSnapshot (LatestOnly mode)
/// 0x02 = WindowedHistory (Phase 2)
/// 0x03 = Reserved
/// 0x04 = Tombstone (ADR-034)
/// 0x05 = TombstoneBatch (ADR-034)
/// 0x06 = TombstoneAck (ADR-034)
/// ```
#[cfg(feature = "automerge-backend")]
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[repr(u8)]
pub enum SyncMessageType {
    /// Delta-based sync message (Automerge sync protocol)
    DeltaSync = 0x00,
    /// Full state snapshot (doc.save() bytes)
    StateSnapshot = 0x01,
    /// Windowed history sync message (Phase 2)
    WindowedHistory = 0x02,
    /// Single tombstone message (ADR-034 Phase 2)
    Tombstone = 0x04,
    /// Batch of tombstones for initial exchange (ADR-034 Phase 2)
    TombstoneBatch = 0x05,
    /// Acknowledgement of received tombstones (ADR-034 Phase 2)
    TombstoneAck = 0x06,
    /// Batch of sync messages for multiple documents (Issue #438)
    SyncBatch = 0x07,
    /// Negentropy set reconciliation initiate (ADR-040, Issue #435)
    NegentropyInit = 0x08,
    /// Negentropy set reconciliation response (ADR-040, Issue #435)
    NegentropyResponse = 0x09,
    /// Negentropy reconciliation complete - request missing docs (ADR-040, Issue #435)
    NegentropyRequest = 0x0A,
}

/// Sync direction for hierarchical routing (Issue #438 Phase 3)
///
/// Determines how sync messages should be routed based on document type:
/// - Upward: Sync to parent/leader only (nodes, beacons, platforms)
/// - Downward: Sync from leader to members (commands)
/// - Lateral: Sync to peers at same level (cells)
/// - Broadcast: Sync to all connected peers (alerts, contact-reports)
#[cfg(feature = "automerge-backend")]
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum SyncDirection {
    /// Sync upward to parent/leader only
    /// Used for: nodes, beacons, platforms
    Upward,
    /// Sync downward from leader to members
    /// Used for: commands
    Downward,
    /// Sync laterally to peers at same hierarchy level
    /// Used for: cells
    Lateral,
    /// Broadcast to all connected peers (existing behavior)
    /// Used for: alerts, contact-reports, and unknown collections
    Broadcast,
}

#[cfg(feature = "automerge-backend")]
impl SyncDirection {
    /// Determine sync direction from document key
    ///
    /// Parses the collection name from the document key (format: "collection:id")
    /// and returns the appropriate sync direction.
    pub fn from_doc_key(doc_key: &str) -> Self {
        // Extract collection name from "collection:id" format
        let collection = doc_key.split(':').next().unwrap_or(doc_key);

        match collection {
            // Upward: State aggregation data flows up the hierarchy
            "nodes" | "beacons" | "platforms" | "summaries" => SyncDirection::Upward,
            // Downward: Commands flow down from coordinators to executors
            "commands" => SyncDirection::Downward,
            // Lateral: Cell state syncs between peers in same cell
            "cells" => SyncDirection::Lateral,
            // Broadcast: Alerts and reports go everywhere
            "alerts" | "contact-reports" | "contact_reports" | "events" => SyncDirection::Broadcast,
            // Default to broadcast for unknown collections (safe fallback)
            _ => SyncDirection::Broadcast,
        }
    }
}

/// A single document sync entry within a batch (Issue #438)
///
/// Represents one document's sync data within a `SyncBatch`. Can be a delta sync,
/// state snapshot, or tombstone.
#[cfg(feature = "automerge-backend")]
#[derive(Debug, Clone)]
pub struct SyncEntry {
    /// Document key (e.g., "nodes:node-1")
    pub doc_key: String,
    /// Type of sync for this document
    pub sync_type: SyncMessageType,
    /// Payload bytes (encoded SyncMessage, state bytes, or tombstone)
    pub payload: Vec<u8>,
}

#[cfg(feature = "automerge-backend")]
impl SyncEntry {
    /// Create a new sync entry
    pub fn new(doc_key: String, sync_type: SyncMessageType, payload: Vec<u8>) -> Self {
        Self {
            doc_key,
            sync_type,
            payload,
        }
    }

    /// Encode to wire format
    ///
    /// Wire format:
    /// ```text
    /// [2 bytes: doc_key_len][doc_key][1 byte: sync_type][4 bytes: payload_len][payload]
    /// ```
    pub fn encode(&self) -> Vec<u8> {
        let doc_key_bytes = self.doc_key.as_bytes();
        let doc_key_len = doc_key_bytes.len() as u16;

        let mut buf = Vec::with_capacity(2 + doc_key_bytes.len() + 1 + 4 + self.payload.len());
        buf.extend_from_slice(&doc_key_len.to_be_bytes());
        buf.extend_from_slice(doc_key_bytes);
        buf.push(self.sync_type as u8);
        buf.extend_from_slice(&(self.payload.len() as u32).to_be_bytes());
        buf.extend_from_slice(&self.payload);
        buf
    }

    /// Decode from wire format, returns (entry, bytes_consumed)
    pub fn decode(bytes: &[u8]) -> anyhow::Result<(Self, usize)> {
        use anyhow::Context;

        if bytes.len() < 7 {
            anyhow::bail!("SyncEntry too short: {} bytes", bytes.len());
        }

        let doc_key_len = u16::from_be_bytes([bytes[0], bytes[1]]) as usize;
        let mut offset = 2;

        if bytes.len() < offset + doc_key_len + 1 + 4 {
            anyhow::bail!("SyncEntry truncated at doc_key");
        }

        let doc_key = String::from_utf8(bytes[offset..offset + doc_key_len].to_vec())
            .context("Invalid UTF-8 in doc_key")?;
        offset += doc_key_len;

        let sync_type = match bytes[offset] {
            0x00 => SyncMessageType::DeltaSync,
            0x01 => SyncMessageType::StateSnapshot,
            0x02 => SyncMessageType::WindowedHistory,
            0x04 => SyncMessageType::Tombstone,
            0x05 => SyncMessageType::TombstoneBatch,
            0x06 => SyncMessageType::TombstoneAck,
            0x07 => SyncMessageType::SyncBatch,
            other => anyhow::bail!("Unknown sync type: 0x{:02x}", other),
        };
        offset += 1;

        let payload_len = u32::from_be_bytes([
            bytes[offset],
            bytes[offset + 1],
            bytes[offset + 2],
            bytes[offset + 3],
        ]) as usize;
        offset += 4;

        if bytes.len() < offset + payload_len {
            anyhow::bail!("SyncEntry truncated at payload");
        }

        let payload = bytes[offset..offset + payload_len].to_vec();
        offset += payload_len;

        Ok((
            Self {
                doc_key,
                sync_type,
                payload,
            },
            offset,
        ))
    }
}

/// Batch of sync messages for multiple documents (Issue #438)
///
/// Enables sending multiple document syncs in a single QUIC stream,
/// reducing stream-per-message overhead from O(N×M) to O(N) where
/// N = peers and M = documents.
///
/// # Wire Format
///
/// ```text
/// [8 bytes: batch_id][8 bytes: created_at][1 byte: ttl][4 bytes: entry_count][entries...]
/// ```
#[cfg(feature = "automerge-backend")]
#[derive(Debug, Clone)]
pub struct SyncBatch {
    /// Unique batch ID for tracking/acknowledgement
    pub batch_id: u64,
    /// Timestamp when batch was created (millis since UNIX epoch)
    pub created_at: u64,
    /// Time-to-live (hop count) for multi-hop forwarding
    /// Decremented at each hop; batch is dropped when TTL reaches 0
    pub ttl: u8,
    /// Entries in this batch
    pub entries: Vec<SyncEntry>,
}

/// Default TTL for sync batches (max 5 hops)
#[cfg(feature = "automerge-backend")]
pub const DEFAULT_SYNC_BATCH_TTL: u8 = 5;

/// Minimum bandwidth-permit size used for FullHistory syncs, where the actual
/// delta size is unknown at prepare time. Using 0 would bypass QoS gating
/// entirely; this floor ensures FullHistory syncs are still subject to back-
/// pressure, sized conservatively at one typical CRDT delta (peat-mesh#236).
#[cfg(feature = "automerge-backend")]
const FULL_HISTORY_BANDWIDTH_FLOOR: usize = 64 * 1024; // 64 KiB

#[cfg(feature = "automerge-backend")]
impl SyncBatch {
    /// Create a new empty batch
    pub fn new() -> Self {
        Self {
            batch_id: 0,
            created_at: std::time::SystemTime::now()
                .duration_since(std::time::UNIX_EPOCH)
                .unwrap()
                .as_millis() as u64,
            ttl: DEFAULT_SYNC_BATCH_TTL,
            entries: Vec::new(),
        }
    }

    /// Create a batch with a specific ID
    pub fn with_id(batch_id: u64) -> Self {
        Self {
            batch_id,
            created_at: std::time::SystemTime::now()
                .duration_since(std::time::UNIX_EPOCH)
                .unwrap()
                .as_millis() as u64,
            ttl: DEFAULT_SYNC_BATCH_TTL,
            entries: Vec::new(),
        }
    }

    /// Create a batch with entries (Issue #438 Phase 3)
    pub fn with_entries(entries: Vec<SyncEntry>) -> Self {
        Self {
            batch_id: 0, // Will be assigned when sent
            created_at: std::time::SystemTime::now()
                .duration_since(std::time::UNIX_EPOCH)
                .unwrap()
                .as_millis() as u64,
            ttl: DEFAULT_SYNC_BATCH_TTL,
            entries,
        }
    }

    /// Set TTL for this batch
    pub fn with_ttl(mut self, ttl: u8) -> Self {
        self.ttl = ttl;
        self
    }

    /// Decrement TTL and return true if batch should still be forwarded
    pub fn decrement_ttl(&mut self) -> bool {
        if self.ttl > 0 {
            self.ttl -= 1;
            true
        } else {
            false
        }
    }

    /// Add a delta sync entry
    pub fn add_delta(&mut self, doc_key: &str, message: &automerge::sync::Message) {
        let payload = message.clone().encode();
        self.entries.push(SyncEntry::new(
            doc_key.to_string(),
            SyncMessageType::DeltaSync,
            payload,
        ));
    }

    /// Add a state snapshot entry
    pub fn add_snapshot(&mut self, doc_key: &str, state_bytes: Vec<u8>) {
        self.entries.push(SyncEntry::new(
            doc_key.to_string(),
            SyncMessageType::StateSnapshot,
            state_bytes,
        ));
    }

    /// Add a tombstone entry
    pub fn add_tombstone(&mut self, tombstone: &crate::qos::TombstoneSyncMessage) {
        self.entries.push(SyncEntry::new(
            format!(
                "tombstones:{}:{}",
                tombstone.tombstone.collection, tombstone.tombstone.document_id
            ),
            SyncMessageType::Tombstone,
            tombstone.encode(),
        ));
    }

    /// Check if batch is empty
    pub fn is_empty(&self) -> bool {
        self.entries.is_empty()
    }

    /// Get number of entries
    pub fn len(&self) -> usize {
        self.entries.len()
    }

    /// Get total payload size in bytes
    pub fn payload_size(&self) -> usize {
        self.entries.iter().map(|e| e.payload.len()).sum()
    }

    /// Encode to wire format
    /// Format: [8: batch_id][8: created_at][1: ttl][4: entry_count][entries...]
    pub fn encode(&self) -> Vec<u8> {
        let mut buf = Vec::with_capacity(21 + self.payload_size());
        buf.extend_from_slice(&self.batch_id.to_be_bytes());
        buf.extend_from_slice(&self.created_at.to_be_bytes());
        buf.push(self.ttl);
        buf.extend_from_slice(&(self.entries.len() as u32).to_be_bytes());
        for entry in &self.entries {
            buf.extend_from_slice(&entry.encode());
        }
        buf
    }

    /// Decode from wire format
    pub fn decode(bytes: &[u8]) -> anyhow::Result<Self> {
        if bytes.len() < 21 {
            anyhow::bail!("SyncBatch too short: {} bytes", bytes.len());
        }

        let batch_id = u64::from_be_bytes([
            bytes[0], bytes[1], bytes[2], bytes[3], bytes[4], bytes[5], bytes[6], bytes[7],
        ]);
        let created_at = u64::from_be_bytes([
            bytes[8], bytes[9], bytes[10], bytes[11], bytes[12], bytes[13], bytes[14], bytes[15],
        ]);
        let ttl = bytes[16];
        let entry_count = u32::from_be_bytes([bytes[17], bytes[18], bytes[19], bytes[20]]) as usize;

        let mut offset = 21;
        let mut entries = Vec::with_capacity(entry_count);

        for _ in 0..entry_count {
            let (entry, consumed) = SyncEntry::decode(&bytes[offset..])?;
            entries.push(entry);
            offset += consumed;
        }

        Ok(Self {
            batch_id,
            created_at,
            ttl,
            entries,
        })
    }
}

#[cfg(feature = "automerge-backend")]
impl Default for SyncBatch {
    fn default() -> Self {
        Self::new()
    }
}

/// Received sync payload (Issue #355, ADR-034, Issue #438)
///
/// Can be a delta-based sync message, state snapshot, tombstone, or batch of syncs.
#[cfg(feature = "automerge-backend")]
#[derive(Debug)]
pub enum ReceivedSyncPayload {
    /// Delta-based sync message from Automerge protocol
    Delta(SyncMessage),
    /// Full document state snapshot (from LatestOnly mode)
    StateSnapshot(Vec<u8>),
    /// Single tombstone (ADR-034 Phase 2)
    Tombstone(crate::qos::TombstoneSyncMessage),
    /// Batch of tombstones for initial exchange (ADR-034 Phase 2)
    TombstoneBatch(crate::qos::TombstoneBatch),
    /// Batch of sync messages for multiple documents (Issue #438)
    Batch(SyncBatch),
    /// Negentropy init message - initiates set reconciliation (ADR-040)
    NegentropyInit(Vec<u8>),
    /// Negentropy response message - reconciliation round (ADR-040)
    NegentropyResponse(Vec<u8>),
}

/// Per-peer sync statistics
#[cfg(feature = "automerge-backend")]
#[derive(Debug, Clone, Default)]
pub struct PeerSyncStats {
    /// Total bytes sent to this peer
    pub bytes_sent: u64,
    /// Total bytes received from this peer
    pub bytes_received: u64,
    /// Number of successful syncs
    pub sync_count: u64,
    /// Last successful sync timestamp
    pub last_sync: Option<SystemTime>,
    /// Number of sync failures
    pub failure_count: u64,
    /// Count of LatestOnly mode syncs
    pub latest_only_count: u64,
    /// Count of FullHistory mode syncs
    pub full_history_count: u64,
    /// Count of WindowedHistory mode syncs
    pub windowed_count: u64,
}

/// Peer-independent sync payload for a single document, computed once per
/// fanout and shared across every target peer (peat-mesh#247).
///
/// Before this existed, `sync_document_with_all_peers` fanned out via a
/// per-peer `join_all` and each peer's `initiate_sync` re-ran `store.get`
/// (an Automerge clone out of the cache) **and** `doc.save()` (snapshot
/// serialization). For an N-peer fanout that was N redundant clones +
/// serializations of the same document on the caller's serial change loop —
/// the dominant per-document cost behind the under-load convergence tails.
/// Hoisting that work here makes it O(1) per document instead of
/// O(N_peers), shrinking the head-of-line stall a backlog imposes on a
/// probe write.
#[cfg(feature = "automerge-backend")]
struct PreparedDoc {
    /// Sync mode for the document's collection (drives snapshot vs delta).
    sync_mode: SyncMode,
    /// The loaded document, shared read-only with the FullHistory delta
    /// path (`generate_sync_message` borrows it; it never mutates the doc).
    doc: Arc<Automerge>,
    /// `doc.save()` snapshot bytes for LatestOnly / WindowedHistory paths
    /// and bandwidth-permit accounting. `None` for FullHistory: `doc.save()`
    /// serialises the full change graph which is O(edit-history depth), but
    /// FullHistory sync uses `initiate_delta_sync` (delta only — the doc
    /// object above) and never reads the snapshot bytes. Skipping the save
    /// eliminates the per-sync O(N) cost that caused convergence to degrade
    /// linearly as a FullHistory doc accumulated edits (peat-mesh#236).
    doc_bytes: Option<Arc<Vec<u8>>>,
}

/// QoS-priority fanout queue + its drain-worker wakeup (peat-mesh#247,
/// ADR-0013).
///
/// Local-write notifications enqueue a document key here (non-blocking)
/// instead of awaiting the fanout inline; a single worker drains the queue
/// highest-QoS-first. This reorders *which document is fanned out next* — the
/// level at which the under-load head-of-line stall actually lives — while a
/// single drainer preserves the per-(peer, channel) ordering and implicit
/// backpressure the old serial change loop provided (ADR-0013 rejects the
/// spawn-everything shape that regressed convergence).
#[cfg(feature = "automerge-backend")]
struct FanoutState {
    inner: Mutex<FanoutQueue>,
    /// Wakes the drain worker when an entry is enqueued.
    notify: Notify,
}

/// Inner state of the fanout queue, guarded by `FanoutState::inner`.
#[cfg(feature = "automerge-backend")]
#[derive(Default)]
struct FanoutQueue {
    /// Pending fanout keys ordered `(qos_rank ASC, seq ASC)`. `first()` is the
    /// highest-priority, earliest-enqueued entry (what the worker pops);
    /// `last()` is the lowest-priority, latest entry (the eviction victim).
    /// `qos_rank = QoSClass as u8` — `Critical = 1` (highest) … `Bulk = 5`
    /// (lowest) — so numeric ascending == priority descending.
    pending: BTreeSet<(u8, u64, String)>,
    /// `doc_key -> (qos_rank, seq)` for O(log n) coalescing and removal.
    index: HashMap<String, (u8, u64)>,
    /// Monotonic enqueue sequence — FIFO tiebreak within a QoS class.
    next_seq: u64,
    /// Per-QoS-class shed counters (index = `QoSClass as u8`; slot 0 unused).
    /// Observability for the bounded-queue drop policy (ADR-0013).
    dropped_by_class: [u64; 6],
}

/// Coordinator for Automerge document synchronization over Iroh
///
/// Manages sync state for each peer and coordinates message exchange.
///
/// # Phase 4-5 Enhancements
///
/// - ✅ Per-peer sync state management
/// - ✅ Sync statistics tracking (bytes, counts, timestamps)
/// - ✅ Error handling with retry logic and circuit breaker (Phase 5)
/// - ✅ Partition detection with heartbeat mechanism (Phase 6.3)
/// - ✅ Flow control and backpressure (Issue #97)
/// - ✅ Sync modes: LatestOnly vs FullHistory (Issue #355)
/// - ✅ Batch sync to reduce stream-per-message overhead (Issue #438)
#[cfg(feature = "automerge-backend")]
pub struct AutomergeSyncCoordinator {
    /// Reference to the AutomergeStore
    store: Arc<AutomergeStore>,
    /// Reference to the transport (trait object)
    transport: Arc<dyn SyncTransport>,
    /// Sync state for each peer (per document)
    /// Map: document_key -> peer_id -> SyncState
    peer_states: Arc<RwLock<HashMap<String, HashMap<EndpointId, SyncState>>>>,
    /// Per-peer sync statistics
    /// Map: peer_id -> PeerSyncStats
    peer_stats: Arc<RwLock<HashMap<EndpointId, PeerSyncStats>>>,
    /// Total bytes sent (across all peers)
    total_bytes_sent: Arc<AtomicU64>,
    /// Total bytes received (across all peers)
    total_bytes_received: Arc<AtomicU64>,
    /// Error handler with retry logic and circuit breaker
    error_handler: Arc<SyncErrorHandler>,
    /// Partition detector for heartbeat tracking
    partition_detector: Arc<PartitionDetector>,
    /// Flow controller for rate limiting and backpressure
    flow_controller: Arc<FlowController>,
    /// Sync mode registry for per-collection sync mode configuration (Issue #355)
    sync_mode_registry: Arc<SyncModeRegistry>,
    /// Next batch ID counter (Issue #438)
    next_batch_id: Arc<AtomicU64>,
    /// Optional sync router for direction-based sync routing (Issue #438 Phase 3)
    sync_router: Option<Arc<dyn SyncRouter>>,
    /// Optional channel manager for persistent stream sync (Issue #435)
    /// Uses Weak to avoid circular reference with SyncChannelManager
    channel_manager: Arc<RwLock<Option<Weak<super::sync_channel::SyncChannelManager>>>>,
    /// Negentropy set reconciliation for efficient document discovery (ADR-040, Issue #435)
    negentropy_sync: Arc<NegentropySync>,
    /// Optional TTL manager for automatic document expiration
    ttl_manager: Arc<RwLock<Option<Arc<super::ttl_manager::TtlManager>>>>,
    /// Optional bandwidth allocation for QoS-aware sync (PRD-004)
    bandwidth_allocation: Arc<RwLock<Option<Arc<crate::qos::BandwidthAllocation>>>>,
    /// Optional handle to the node-local Lamport clock (peat-mesh#196).
    ///
    /// When set (by `AutomergeBackend::with_iroh` after construction
    /// via [`Self::set_lamport_clock`]), inbound `Tombstone.lamport`
    /// values from `handle_incoming_tombstone` and
    /// `handle_incoming_tombstone_batch` advance the clock via
    /// `crate::sync::automerge_backend::observe_atomic` before the
    /// tombstone is applied — so a locally-authored operation that
    /// follows the receive sorts strictly after the observed Lamport
    /// in the causal order Lamport's partial order captures.
    ///
    /// `None` when the coordinator is constructed standalone (the
    /// `peat-mesh-node` binary's manual wire-up, unit tests, or any
    /// consumer that doesn't yet thread a clock through). In that
    /// case the receive-side wire-up is a no-op and the consumer
    /// must call `observe_lamport` itself if it cares about cross-
    /// node ordering.
    lamport_clock: Arc<RwLock<Option<Arc<AtomicU64>>>>,
    /// QoS-priority fanout queue, drained by the worker spawned in
    /// `AutomergeBackend::start_sync` (peat-mesh#247, ADR-0013). Empty and
    /// idle until something calls [`Self::enqueue_fanout`].
    fanout: Arc<FanoutState>,
}

#[cfg(feature = "automerge-backend")]
impl AutomergeSyncCoordinator {
    /// Create a new sync coordinator
    ///
    /// # Arguments
    ///
    /// * `store` - The AutomergeStore managing documents
    /// * `transport` - Transport implementing `SyncTransport` for P2P connections
    pub fn new(store: Arc<AutomergeStore>, transport: Arc<dyn SyncTransport>) -> Self {
        Self::with_flow_control(store, transport, FlowControlConfig::default())
    }

    /// Create a new sync coordinator with custom flow control configuration
    ///
    /// # Arguments
    ///
    /// * `store` - The AutomergeStore managing documents
    /// * `transport` - Transport implementing `SyncTransport` for P2P connections
    /// * `flow_config` - Custom flow control configuration
    pub fn with_flow_control(
        store: Arc<AutomergeStore>,
        transport: Arc<dyn SyncTransport>,
        flow_config: FlowControlConfig,
    ) -> Self {
        Self {
            store,
            transport,
            peer_states: Arc::new(RwLock::new(HashMap::new())),
            peer_stats: Arc::new(RwLock::new(HashMap::new())),
            total_bytes_sent: Arc::new(AtomicU64::new(0)),
            total_bytes_received: Arc::new(AtomicU64::new(0)),
            error_handler: Arc::new(SyncErrorHandler::new()),
            partition_detector: Arc::new(PartitionDetector::new()),
            flow_controller: Arc::new(FlowController::with_config(flow_config)),
            sync_mode_registry: Arc::new(SyncModeRegistry::with_defaults()),
            next_batch_id: Arc::new(AtomicU64::new(1)),
            sync_router: None,
            channel_manager: Arc::new(RwLock::new(None)),
            negentropy_sync: Arc::new(NegentropySync::new()),
            ttl_manager: Arc::new(RwLock::new(None)),
            bandwidth_allocation: Arc::new(RwLock::new(None)),
            lamport_clock: Arc::new(RwLock::new(None)),
            fanout: Arc::new(FanoutState {
                inner: Mutex::new(FanoutQueue::default()),
                notify: Notify::new(),
            }),
        }
    }

    /// Create a new sync coordinator with custom sync mode registry
    ///
    /// # Arguments
    ///
    /// * `store` - The AutomergeStore managing documents
    /// * `transport` - Transport implementing `SyncTransport` for P2P connections
    /// * `sync_mode_registry` - Custom sync mode configuration
    pub fn with_sync_modes(
        store: Arc<AutomergeStore>,
        transport: Arc<dyn SyncTransport>,
        sync_mode_registry: Arc<SyncModeRegistry>,
    ) -> Self {
        Self {
            store,
            transport,
            peer_states: Arc::new(RwLock::new(HashMap::new())),
            peer_stats: Arc::new(RwLock::new(HashMap::new())),
            total_bytes_sent: Arc::new(AtomicU64::new(0)),
            total_bytes_received: Arc::new(AtomicU64::new(0)),
            error_handler: Arc::new(SyncErrorHandler::new()),
            partition_detector: Arc::new(PartitionDetector::new()),
            flow_controller: Arc::new(FlowController::with_config(FlowControlConfig::default())),
            sync_mode_registry,
            next_batch_id: Arc::new(AtomicU64::new(1)),
            sync_router: None,
            channel_manager: Arc::new(RwLock::new(None)),
            negentropy_sync: Arc::new(NegentropySync::new()),
            ttl_manager: Arc::new(RwLock::new(None)),
            bandwidth_allocation: Arc::new(RwLock::new(None)),
            lamport_clock: Arc::new(RwLock::new(None)),
            fanout: Arc::new(FanoutState {
                inner: Mutex::new(FanoutQueue::default()),
                notify: Notify::new(),
            }),
        }
    }

    /// Absolute bound on pending fanout entries (ADR-0013). Coalescing
    /// collapses same-document churn, so this tracks *distinct in-flight
    /// documents*, not write rate — 4096 distinct hot documents is generous
    /// for tactical meshes and bounds memory at roughly the doc-key strings.
    const MAX_PENDING_FANOUT: usize = 4096;

    /// Enqueue a document for QoS-priority fanout (peat-mesh#247, ADR-0013).
    ///
    /// Non-blocking and infallible by contract: it never blocks the caller
    /// (the change loop) and never returns an error. Returning `Err` to the
    /// change loop is the peat#864 shape — the auto-sync pusher swallows it and
    /// the change is silently lost from the outbound pipeline with no retry.
    ///
    /// - **Coalescing:** if `doc_key` is already pending, this is a no-op — the
    ///   worker fans out the document's *current* store state when it reaches
    ///   it, so one pending entry already covers any number of intervening
    ///   writes (complements the per-(peer, doc) sync cooldown, #239).
    /// - **Priority:** the entry's QoS class
    ///   ([`QoSClass::for_collection`](crate::qos::QoSClass::for_collection))
    ///   determines drain order; FIFO within a class.
    /// - **Bounded:** at [`Self::MAX_PENDING_FANOUT`] the lowest-priority
    ///   pending entry is evicted (possibly this one, if it is the lowest),
    ///   counted per-class and `warn`-logged. Safe because shedding only
    ///   happens under sustained overload, sheds Bulk/Low first, and an
    ///   evicted fanout is recovered by the next change to that document or the
    ///   periodic / reconnect re-sync.
    pub fn enqueue_fanout(&self, doc_key: &str) {
        let collection = Self::collection_from_doc_key(doc_key);
        let rank = crate::qos::QoSClass::for_collection(collection) as u8;
        {
            let mut q = self.fanout.inner.lock().unwrap_or_else(|e| e.into_inner());
            if q.index.contains_key(doc_key) {
                return; // coalesced — already pending
            }
            let seq = q.next_seq;
            q.next_seq += 1;
            q.pending.insert((rank, seq, doc_key.to_string()));
            q.index.insert(doc_key.to_string(), (rank, seq));

            // Enforce the bound by shedding the lowest-priority pending entry
            // (`last()`), which may be the one just inserted.
            while q.pending.len() > Self::MAX_PENDING_FANOUT {
                let Some(victim) = q.pending.iter().next_back().cloned() else {
                    break;
                };
                q.pending.remove(&victim);
                q.index.remove(&victim.2);
                if let Some(slot) = q.dropped_by_class.get_mut(victim.0 as usize) {
                    *slot += 1;
                }
                tracing::warn!(
                    doc_key = victim.2,
                    qos_rank = victim.0,
                    max = Self::MAX_PENDING_FANOUT,
                    "Fanout queue full; shed lowest-priority pending fanout (peat-mesh#247)"
                );
            }
        }
        // Wake the worker. `notify_one` stores a permit if the worker isn't
        // currently parked, so an enqueue that races the worker's empty-check
        // is not lost.
        self.fanout.notify.notify_one();
    }

    /// Drain loop for the QoS-priority fanout queue (peat-mesh#247).
    ///
    /// A **single** worker: pops the highest-priority pending document, runs
    /// the full per-document fanout ([`Self::sync_document_with_all_peers`]),
    /// and repeats; parks on the notify when the queue is empty. Single-worker
    /// (not a pool) keeps the per-(peer, channel) ordering and the implicit
    /// backpressure the serial change loop gave us — it is that loop, reordered
    /// by priority. Spawned by `AutomergeBackend::start_sync`; runs until the
    /// task is aborted (shutdown). The queue lock is never held across the
    /// `.await`.
    pub async fn run_fanout_worker(self: Arc<Self>) {
        loop {
            let next = {
                let mut q = self.fanout.inner.lock().unwrap_or_else(|e| e.into_inner());
                if let Some(item) = q.pending.iter().next().cloned() {
                    q.pending.remove(&item);
                    q.index.remove(&item.2);
                    Some(item.2)
                } else {
                    None
                }
            };
            match next {
                Some(doc_key) => {
                    if let Err(e) = self.sync_document_with_all_peers(&doc_key).await {
                        tracing::debug!(
                            doc_key = doc_key,
                            error = %e,
                            "fanout worker: sync_document_with_all_peers errored"
                        );
                    }
                }
                None => self.fanout.notify.notified().await,
            }
        }
    }

    /// Per-QoS-class counts of fanout entries shed under the bounded-queue
    /// policy (peat-mesh#247 / ADR-0013). Indexed by `QoSClass as u8`
    /// (`Critical = 1` … `Bulk = 5`; slot 0 is unused). This is the
    /// observability surface the ADR commits to for the drop policy: shedding
    /// fires only under sustained overload, and a counter reconstructs what
    /// happened directly rather than by scraping `warn`-lines.
    pub fn fanout_dropped_by_class(&self) -> [u64; 6] {
        self.fanout
            .inner
            .lock()
            .unwrap_or_else(|e| e.into_inner())
            .dropped_by_class
    }

    /// Set the channel manager for persistent stream sync (Issue #435)
    ///
    /// This enables routing all sync operations through persistent channels
    /// instead of opening new streams for each operation.
    pub fn set_channel_manager(&self, manager: Arc<super::sync_channel::SyncChannelManager>) {
        *self
            .channel_manager
            .write()
            .unwrap_or_else(|e| e.into_inner()) = Some(Arc::downgrade(&manager));
    }

    /// Set the TTL manager for automatic document expiration on synced documents
    pub fn set_ttl_manager(&self, manager: Arc<super::ttl_manager::TtlManager>) {
        *self.ttl_manager.write().unwrap_or_else(|e| e.into_inner()) = Some(manager);
    }

    /// Set the bandwidth allocation for QoS-aware sync (PRD-004)
    pub fn set_bandwidth_allocation(&self, allocation: Arc<crate::qos::BandwidthAllocation>) {
        *self
            .bandwidth_allocation
            .write()
            .unwrap_or_else(|e| e.into_inner()) = Some(allocation);
    }

    /// Set the node-local Lamport clock for the receive-side wire-up
    /// (peat-mesh#196).
    ///
    /// After this is called, every inbound `Tombstone.lamport` value
    /// (single `Tombstone` messages and every entry in a
    /// `TombstoneBatch`) advances the clock via
    /// `crate::sync::automerge_backend::observe_atomic` before the
    /// tombstone is applied — so a subsequent locally-authored
    /// `next_lamport()` is strictly greater than any previously-
    /// observed remote Lamport.
    ///
    /// Called by `AutomergeBackend::with_iroh` immediately after the
    /// coordinator is constructed, threading in the same
    /// `Arc<AtomicU64>` the backend uses for `next_lamport`. Standalone
    /// coordinator consumers (the `peat-mesh-node` binary, unit tests,
    /// any consumer that doesn't yet care about cross-node Lamport
    /// semantics) can simply skip this call; the receive-side path
    /// then leaves the clock alone.
    pub fn set_lamport_clock(&self, clock: Arc<AtomicU64>) {
        *self
            .lamport_clock
            .write()
            .unwrap_or_else(|e| e.into_inner()) = Some(clock);
    }

    /// Advance the wired Lamport clock past `observed`, if a clock is
    /// wired (peat-mesh#196). No-op when no clock has been attached
    /// via [`Self::set_lamport_clock`].
    fn observe_inbound_lamport(&self, observed: u64) {
        let clock = self
            .lamport_clock
            .read()
            .unwrap_or_else(|e| e.into_inner())
            .clone();
        if let Some(clock) = clock {
            crate::sync::automerge_backend::observe_atomic(&clock, observed);
        }
    }

    /// Persist a sync-received document with origin attribution (peat#891).
    ///
    /// Why this is not `put_without_notify`: the original Issue #115 fix
    /// suppressed `change_tx` entirely on receive paths to prevent the
    /// outbound auto-sync pusher from re-broadcasting back to the sender.
    /// That fix held the ping-pong invariant but also killed transitive
    /// gossip — a doc received from peer X never propagated onward to
    /// peer Y. Hub-and-spoke topologies (alpha as the hub, bravo/charlie
    /// as spokes) could not converge because alpha never re-pushed bravo's
    /// state to charlie. See peat#891 (the field report) and peat#907
    /// (the tracking issue for the architectural fix).
    ///
    /// The new contract: receive paths call [`AutomergeStore::put_with_origin`]
    /// with `ChangeOrigin::Remote(peer_id)`. The legacy `change_tx`
    /// (subscribed by the existing outbound pusher) still does NOT fire —
    /// `put_with_origin` skips it whenever origin is `Remote(_)`, so the
    /// #115 ping-pong invariant is preserved for that consumer. But the
    /// origin-tagged `gossip_tx` (subscribed by
    /// `subscribe_to_changes_with_origin`) DOES fire with the peer
    /// attribution, so a gossip-aware consumer can push the doc to every
    /// peer except the source.
    fn put_received(
        &self,
        key: &str,
        doc: &automerge::Automerge,
        peer_id: EndpointId,
    ) -> anyhow::Result<()> {
        let origin = ChangeOrigin::Remote(peer_id.to_string());
        let ttl_mgr = self
            .ttl_manager
            .read()
            .unwrap_or_else(|e| e.into_inner())
            .clone();
        // Route through the single combined `put_with_origin_and_ttl`
        // entry point so the put + ttl pairing is restored at the API
        // surface and the key-prefix → collection convention stays
        // defined in one place (`AutomergeStore::apply_ttl`). Without
        // this the inline `key.find(['/', ':'])` logic would be
        // duplicated across two crates — peat-mesh#151 QA caught this.
        if let Some(ref mgr) = ttl_mgr {
            self.store.put_with_origin_and_ttl(key, doc, origin, mgr)
        } else {
            self.store.put_with_origin(key, doc, origin)
        }
    }

    /// Whether the persistent sync-channel manager is currently wired.
    ///
    /// Returns `true` if `set_channel_manager` was called AND the
    /// corresponding `Arc<SyncChannelManager>` is still alive somewhere
    /// (the coordinator only holds a `Weak`, so this upgrades the
    /// `Weak` to test liveness). Returns `false` if either:
    /// - `set_channel_manager` was never called, OR
    /// - the strong reference was dropped (the BLOCKER class that
    ///   peat-mesh#178 fixed: AutomergeBackend's old code dropped the
    ///   `Arc<SyncChannelManager>` at the end of `with_iroh`, the
    ///   `Weak` immediately dangled, every consumer fell through to
    ///   the legacy per-message-stream fallback).
    ///
    /// Used by the `two_backend_persistent_sync_e2e` integration test
    /// to pin the wire-up directly rather than only via downstream
    /// delivery (legacy fallback also delivers — without this
    /// accessor the regression test can't distinguish the two paths).
    pub fn has_persistent_sync_channel(&self) -> bool {
        self.get_channel_manager().is_some()
    }

    /// Get the channel manager if available
    fn get_channel_manager(&self) -> Option<Arc<super::sync_channel::SyncChannelManager>> {
        self.channel_manager
            .read()
            .unwrap()
            .as_ref()
            .and_then(|weak| weak.upgrade())
    }

    /// Create a new sync coordinator with a sync router (Issue #438 Phase 3)
    ///
    /// When a router is provided, sync operations will route messages
    /// based on document type direction:
    /// - Upward (nodes, beacons, platforms): Only sync to cell leader
    /// - Downward (commands): Only sync from leader to cell members
    /// - Lateral (cells): Sync to peers in same cell
    /// - Broadcast (alerts, contact_reports): Sync to all (existing behavior)
    ///
    /// # Arguments
    ///
    /// * `store` - The AutomergeStore managing documents
    /// * `transport` - Transport implementing `SyncTransport` for P2P connections
    /// * `router` - Router implementing `SyncRouter` for direction-based routing
    pub fn with_sync_router(
        store: Arc<AutomergeStore>,
        transport: Arc<dyn SyncTransport>,
        router: Arc<dyn SyncRouter>,
    ) -> Self {
        Self {
            store,
            transport,
            peer_states: Arc::new(RwLock::new(HashMap::new())),
            peer_stats: Arc::new(RwLock::new(HashMap::new())),
            total_bytes_sent: Arc::new(AtomicU64::new(0)),
            total_bytes_received: Arc::new(AtomicU64::new(0)),
            error_handler: Arc::new(SyncErrorHandler::new()),
            partition_detector: Arc::new(PartitionDetector::new()),
            flow_controller: Arc::new(FlowController::with_config(FlowControlConfig::default())),
            sync_mode_registry: Arc::new(SyncModeRegistry::with_defaults()),
            next_batch_id: Arc::new(AtomicU64::new(1)),
            sync_router: Some(router),
            channel_manager: Arc::new(RwLock::new(None)),
            negentropy_sync: Arc::new(NegentropySync::new()),
            ttl_manager: Arc::new(RwLock::new(None)),
            bandwidth_allocation: Arc::new(RwLock::new(None)),
            lamport_clock: Arc::new(RwLock::new(None)),
            fanout: Arc::new(FanoutState {
                inner: Mutex::new(FanoutQueue::default()),
                notify: Notify::new(),
            }),
        }
    }

    /// Set the sync router after construction
    pub fn set_sync_router(&self, router: Arc<dyn SyncRouter>) {
        // Note: This requires interior mutability. For now we only support
        // setting the router at construction time via with_sync_router().
        // TODO: If runtime router changes are needed, add RwLock wrapper.
        tracing::debug!("Sync router set on coordinator");
        let _ = router; // Router is set at construction time
    }

    /// Get the sync mode registry for runtime configuration
    pub fn sync_mode_registry(&self) -> &Arc<SyncModeRegistry> {
        &self.sync_mode_registry
    }

    /// Extract collection name from document key
    ///
    /// Document keys are formatted as "collection:doc_id" (e.g., "beacons:beacon-1")
    fn collection_from_doc_key(doc_key: &str) -> &str {
        doc_key.split(':').next().unwrap_or(doc_key)
    }

    /// Get sync mode for a document key
    fn sync_mode_for_doc(&self, doc_key: &str) -> SyncMode {
        let collection = Self::collection_from_doc_key(doc_key);
        self.sync_mode_registry.get(collection)
    }

    /// Resolve a peer to a live iroh `Connection`, returning a retryable
    /// Network error if the connection has already closed.
    ///
    /// peat#850 follow-up to the categorizer fix in #107. The
    /// categorizer correctly classifies "Failed to write doc_key length"
    /// as a Network error (retryable), but the underlying write was
    /// failing because the cached `Connection` returned by
    /// `SyncTransport::get_or_connect` had already entered the closing
    /// state — `open_bi()` returned a stream against a doomed
    /// connection, and the first `write_all` then failed. Repeating
    /// this dance N times triggered the circuit breaker even with the
    /// correct retryable classification, because `consecutive_failures`
    /// climbed regardless of which error class.
    ///
    /// This helper checks `Connection::close_reason()` immediately
    /// after `get_or_connect` returns. If the connection is already
    /// closed, we bail with a phrase the categorizer recognizes as
    /// Network ("connection lost…") rather than opening a doomed
    /// stream and racking up `consecutive_failures` from a write that
    /// was never going to succeed. The next sync tick can retry, and
    /// the underlying transport will either re-establish the
    /// connection or stay disconnected — but either way we don't
    /// burn breaker budget on a known-dead connection.
    async fn get_alive_connection(&self, peer_id: EndpointId) -> Result<Connection> {
        let conn = self.transport.get_or_connect(&peer_id).await?;
        // Fast-path liveness check, NOT a guarantee. A connection can
        // transition to closing between this check returning Ok and
        // the caller's `open_bi()` — that's a real TOCTOU window.
        // The categorizer recovers cleanly in that case: the
        // subsequent stream-I/O failure phrases (`stream closed`,
        // `ApplicationClosed`, `Failed to write …`) all classify as
        // Network, so the breaker treats them as retryable. The
        // value this check adds is short-circuiting the dominant
        // case where the cache returned an already-closed connection
        // — saving the breaker budget on a write that was never
        // going to succeed. peat#850.
        if let Some(reason) = conn.close_reason() {
            anyhow::bail!(
                "{} (peer={:?}, close_reason: {:?})",
                BAIL_CONNECTION_LOST,
                peer_id,
                reason
            );
        }
        Ok(conn)
    }

    /// Initiate sync for a document with a peer.
    ///
    /// Generates an initial sync message and sends it to the peer. This is
    /// the single-peer entry point: it loads the document once (see
    /// [`PreparedDoc`]) and delegates to [`Self::initiate_sync_prepared`],
    /// which owns the circuit-breaker / cooldown / rate-limit gates and the
    /// error-handling accounting. The all-peers fanout
    /// ([`Self::sync_document_with_all_peers`]) shares one `PreparedDoc`
    /// across peers instead of re-loading per peer (peat-mesh#247).
    ///
    /// # Arguments
    ///
    /// * `doc_key` - The document identifier (e.g., "cells:cell-1")
    /// * `peer_id` - The EndpointId of the peer to sync with
    pub async fn initiate_sync(&self, doc_key: &str, peer_id: EndpointId) -> Result<()> {
        let prepared = self.prepare_doc_for_sync(doc_key)?;
        self.initiate_sync_prepared(doc_key, peer_id, prepared.as_ref())
            .await
    }

    /// Per-peer sync against an already-[`PreparedDoc`] (or `None` when the
    /// document is tombstoned/absent and there is nothing to send).
    ///
    /// Owns the per-peer gates and breaker accounting so that both the
    /// single-peer [`Self::initiate_sync`] and the shared-payload all-peers
    /// fanout run the identical circuit-breaker / defer-on-cooldown /
    /// rate-limit logic. Only the document load/serialize differs: it
    /// happens once per fanout (in [`Self::prepare_doc_for_sync`]) rather
    /// than once per peer (peat-mesh#247).
    async fn initiate_sync_prepared(
        &self,
        doc_key: &str,
        peer_id: EndpointId,
        prepared: Option<&PreparedDoc>,
    ) -> Result<()> {
        // Check circuit breaker before attempting sync
        if self.error_handler.should_block_sync(&peer_id) {
            tracing::warn!("Sync blocked by circuit breaker for peer {:?}", peer_id);
            // Preserve the SyncError type through anyhow so callers
            // (and regression tests) can downcast back to the typed
            // variant. peat-mesh#109 QA — substring-matching the
            // Display string was brittle against cosmetic rewordings.
            return Err(anyhow::Error::from(SyncError::CircuitBreakerOpen));
        }

        // Defer-on-cooldown: if the sync cooldown for (peer, doc) is still
        // active, sleep it out before the flow-control check rather than
        // returning Err. The auto-sync push task (`AutomergeBackend::start_sync`)
        // is driven by `change_tx` and silently swallows sync errors; without
        // this wait, two local writes to the same doc that fire within the
        // 100ms cooldown window result in the second sync being dropped
        // entirely — the change exists in the local store but never reaches
        // peers. peat-node's `attachments::inbox` Transferring→Completed
        // back-to-back writes hit exactly this: receiver wrote Completed
        // ~60ms after Transferring, sync attempt 2 returned CooldownActive,
        // the change_tx event was consumed, the Completed update sat
        // unpushed forever. Refs: defenseunicorns/peat#864.
        //
        // The cooldown's intent (rate-limit per (peer, doc) to prevent
        // sync storms) is preserved — we just defer instead of drop.
        // Rate-limit bucket and memory checks below still gate the request.
        //
        // Looping closes the TOCTOU window between `remaining_cooldown` and
        // the post-loop check: a concurrent caller (e.g.
        // `sync_all_documents_with_peer` on a fresh connect racing with
        // `change_rx`-driven `sync_document_with_all_peers`) can call
        // `record_sync` between our sleep ending and our check running,
        // which would re-arm the cooldown and put us right back where we
        // started — silent-drop. With the loop the second iteration sees
        // the fresh remaining, sleeps again, and retries. The
        // `MAX_COOLDOWN_DEFERRALS` cap bounds worst-case latency on a
        // pathologically contended (peer, doc) so this can't pin the
        // task indefinitely; a 5-deferral cap on a 100 ms cooldown is
        // half a second worst case.
        //
        // After the loop, we skip the cooldown branch of the flow-control
        // check entirely (via `check_rate_limit_only`) — re-checking it
        // would be both redundant (we just waited the cooldown out) and a
        // correctness regression on budget exhaustion (a still-armed
        // cooldown would surface CooldownActive to a caller that drops it
        // silently, re-introducing peat#864's failure mode under lower
        // probability). The cooldown's rate-limit intent is fully honored
        // by the sleep above; losing a change would be strictly worse than
        // syncing one extra time after the deferral budget runs out.
        const MAX_COOLDOWN_DEFERRALS: usize = 5;
        for deferral in 0..MAX_COOLDOWN_DEFERRALS {
            let cooldown_remaining = self.flow_controller.remaining_cooldown(&peer_id, doc_key);
            if cooldown_remaining.is_zero() {
                break;
            }
            tracing::debug!(
                peer = ?peer_id,
                doc_key = doc_key,
                wait_ms = cooldown_remaining.as_millis() as u64,
                deferral = deferral,
                "Sync cooldown active; deferring (not dropping) until elapsed"
            );
            tokio::time::sleep(cooldown_remaining).await;
        }

        // Rate-limit + resource checks only (cooldown bypassed — see comment
        // above for the failure mode this avoids).
        if let Err(flow_err) = self
            .flow_controller
            .check_rate_limit_only(&peer_id, doc_key)
        {
            tracing::debug!(
                "Sync blocked by flow control (non-cooldown) for peer {:?}, doc {}: {}",
                peer_id,
                doc_key,
                flow_err
            );
            return Err(anyhow::anyhow!("{}", flow_err));
        }

        // Attempt sync operation against the shared, pre-loaded payload.
        // `None` => tombstoned/absent document: nothing to send, but this
        // is a clean success (the deletion propagates via the tombstone
        // path) — never an error that would feed the peer breaker
        // (peat-mesh#243).
        let result = match prepared {
            Some(prepared) => self.send_prepared_to_peer(doc_key, peer_id, prepared).await,
            None => Ok(()),
        };

        // Handle the result through error handler
        match &result {
            Ok(_) => {
                self.error_handler.record_success(&peer_id);
                // Record sync for cooldown tracking
                self.flow_controller.record_sync(&peer_id, doc_key);
                tracing::debug!("Sync initiated successfully with peer {:?}", peer_id);
            }
            Err(e) => {
                // Convert error to SyncError via the extracted categorizer
                // so the categorization patterns are unit-testable. peat#850.
                let sync_error = categorize_sync_error(&e.to_string());

                // Process error through handler
                match self.error_handler.handle_error(&peer_id, sync_error) {
                    Ok(Some(retry_delay)) => {
                        tracing::warn!(
                            "Sync failed for peer {:?}, will retry after {:?}",
                            peer_id,
                            retry_delay
                        );
                    }
                    Ok(None) => {
                        tracing::error!("Sync failed for peer {:?}, max retries exceeded", peer_id);
                    }
                    Err(SyncError::CircuitBreakerOpen) => {
                        tracing::error!("Circuit breaker opened for peer {:?}", peer_id);
                    }
                    Err(e) => {
                        tracing::error!(
                            "Error handling sync failure for peer {:?}: {}",
                            peer_id,
                            e
                        );
                    }
                }
            }
        }

        result
    }

    /// Load and serialize a document once for a fanout, independent of any
    /// target peer (peat-mesh#247).
    ///
    /// Runs the peer-independent half of the old `initiate_sync_inner`: the
    /// send-side tombstone guard (ADR-034), the sync-mode lookup, the
    /// `store.get` Automerge clone, and the `doc.save()` snapshot
    /// serialization. The result is shared across every peer in an all-peers
    /// fanout instead of being recomputed per peer.
    ///
    /// Returns `Ok(None)` when there is nothing to send — the document is
    /// tombstoned locally, or absent from the store (deleted / never
    /// existed). Both are clean skips, not errors: the deletion propagates
    /// via the tombstone sync path, and an absent document must never feed
    /// the peer circuit breaker (peat-mesh#243).
    fn prepare_doc_for_sync(&self, doc_key: &str) -> Result<Option<PreparedDoc>> {
        // Tombstone guard (ADR-034) — never send a live snapshot/delta for a
        // document we've tombstoned locally. `store.get` below still returns
        // the pre-delete bytes (the tombstone is separate metadata), so without
        // this guard a reconnect's "sync existing documents" pass re-sends the
        // deleted doc as a StateSnapshot, racing the tombstone and resurrecting
        // it on the peer when the snapshot is applied before the tombstone is
        // recorded. The deletion propagates via the tombstone sync path; this
        // is the send-side mirror of the guards in `receive_sync_message` and
        // `apply_state_snapshot`.
        if let Some(colon_pos) = doc_key.find(':') {
            let collection = &doc_key[..colon_pos];
            let doc_id = &doc_key[colon_pos + 1..];
            if self
                .store
                .has_tombstone(collection, doc_id)
                .unwrap_or(false)
            {
                tracing::debug!(
                    doc_key = doc_key,
                    "Skipping sync send for tombstoned document"
                );
                return Ok(None);
            }
        }

        // Check sync mode for this collection (Issue #355)
        let sync_mode = self.sync_mode_for_doc(doc_key);
        tracing::debug!(
            "prepare_doc_for_sync: sync_mode={} for {}",
            sync_mode,
            doc_key
        );

        // Get the document. An absent document (store returns None) is NOT a
        // failure: it has been deleted (or never existed locally), so there is
        // simply nothing to send on this path — the deletion itself propagates
        // via the tombstone sync path (see the tombstone guard above). Returning
        // Err here surfaces as `SyncError::Document`, which is Fatal/non-retryable
        // and increments the *peer* circuit breaker; a burst of >= failure_threshold
        // (default 5) deletes then opens the breaker and blocks ALL subsequent
        // syncs to that peer — silently dropping unrelated, healthy writes until
        // open_timeout (default 5s) elapses (peat-mesh#243). A local
        // document-state condition must never trip a peer-health breaker, so we
        // skip cleanly instead.
        let Some(doc) = self.store.get(doc_key)? else {
            tracing::debug!(
                doc_key = doc_key,
                "Skipping sync send for absent document (deleted/not present)"
            );
            return Ok(None);
        };

        // For FullHistory, `doc.save()` is O(edit-history depth): it serialises
        // the entire change graph. But FullHistory sync uses `initiate_delta_sync`
        // which only needs the live doc object — `doc_bytes` would only be used
        // for the bandwidth-permit size, which we conservatively treat as 0 for
        // delta syncs (deltas are always ≪ the full snapshot). Skipping the save
        // here eliminates the O(N) cost that caused per-edit convergence to
        // degrade linearly as edit history grew (peat-mesh#236).
        let doc_bytes = match sync_mode {
            SyncMode::FullHistory => {
                tracing::debug!(
                    doc_key,
                    "prepare_doc_for_sync: FullHistory — skipping doc.save() (peat-mesh#236)"
                );
                None
            }
            _ => {
                let bytes = doc.save();
                tracing::debug!("prepare_doc_for_sync: got doc, len={}", bytes.len());
                Some(Arc::new(bytes))
            }
        };

        Ok(Some(PreparedDoc {
            sync_mode,
            doc: Arc::new(doc),
            doc_bytes,
        }))
    }

    /// Send an already-[`PreparedDoc`] to a single peer.
    ///
    /// The peer-specific half of the old `initiate_sync_inner`: acquire the
    /// per-peer bandwidth permit, then dispatch on sync mode. The expensive
    /// `store.get` + `doc.save()` already happened once in
    /// [`Self::prepare_doc_for_sync`]; here we only borrow the shared
    /// snapshot bytes (LatestOnly / WindowedHistory) or the shared loaded
    /// document (FullHistory delta) — no per-peer reload (peat-mesh#247).
    async fn send_prepared_to_peer(
        &self,
        doc_key: &str,
        peer_id: EndpointId,
        prepared: &PreparedDoc,
    ) -> Result<()> {
        // doc_bytes is Some for LatestOnly/WindowedHistory (snapshot send),
        // None for FullHistory (delta sync — never needs snapshot bytes).
        // For FullHistory the actual delta size is unknown at prepare time;
        // use FULL_HISTORY_BANDWIDTH_FLOOR (64 KiB) rather than 0 so that
        // FullHistory syncs are still subject to QoS back-pressure.
        // Using 0 would cause can_transmit(0) to always return true,
        // bypassing the bandwidth budget entirely (peat-mesh#236).
        let permit_size = prepared
            .doc_bytes
            .as_ref()
            .map(|b| b.len())
            .unwrap_or(FULL_HISTORY_BANDWIDTH_FLOOR);

        // Acquire bandwidth permit if allocation is configured (PRD-004).
        // Per-peer: each send to a peer consumes its own bandwidth budget,
        // sized by the snapshot length on every mode (unchanged from the
        // pre-#247 per-peer behaviour).
        if let Some(alloc) = self
            .bandwidth_allocation
            .read()
            .unwrap_or_else(|e| e.into_inner())
            .as_ref()
        {
            let collection = Self::collection_from_doc_key(doc_key);
            let qos_class = crate::qos::QoSClass::for_collection(collection);
            if alloc.acquire(qos_class, permit_size).is_none() {
                tracing::debug!(
                    doc_key = doc_key,
                    class = %qos_class,
                    size = permit_size,
                    "Bandwidth exhausted, deferring sync"
                );
                return Err(anyhow::anyhow!("Bandwidth exhausted for {}", qos_class));
            }
        }

        // Use appropriate sync method based on mode and track per-mode metrics
        match prepared.sync_mode {
            SyncMode::LatestOnly => {
                let doc_bytes = prepared
                    .doc_bytes
                    .as_deref()
                    .expect("LatestOnly PreparedDoc always has doc_bytes");
                tracing::debug!(
                    "send_prepared_to_peer: using LatestOnly mode, sending {} bytes state snapshot",
                    doc_bytes.len()
                );
                self.send_state_snapshot(peer_id, doc_key, doc_bytes)
                    .await?;
                {
                    let mut stats = self.peer_stats.write().unwrap_or_else(|e| e.into_inner());
                    stats.entry(peer_id).or_default().latest_only_count += 1;
                }
                Ok(())
            }
            SyncMode::WindowedHistory { .. } => {
                let doc_bytes = prepared
                    .doc_bytes
                    .as_deref()
                    .expect("WindowedHistory PreparedDoc always has doc_bytes");
                tracing::debug!(
                    "send_prepared_to_peer: using WindowedHistory mode, sending {} bytes state snapshot",
                    doc_bytes.len()
                );
                self.send_state_snapshot(peer_id, doc_key, doc_bytes)
                    .await?;
                {
                    let mut stats = self.peer_stats.write().unwrap_or_else(|e| e.into_inner());
                    stats.entry(peer_id).or_default().windowed_count += 1;
                }
                Ok(())
            }
            SyncMode::FullHistory => {
                // Traditional delta-based sync (per-peer sync state; the loaded
                // doc is shared read-only across peers).
                let result = self
                    .initiate_delta_sync(doc_key, peer_id, prepared.doc.as_ref())
                    .await;
                {
                    let mut stats = self.peer_stats.write().unwrap_or_else(|e| e.into_inner());
                    stats.entry(peer_id).or_default().full_history_count += 1;
                }
                result
            }
        }
    }

    /// Initiate delta-based sync (FullHistory mode)
    ///
    /// Uses Automerge's sync protocol to exchange deltas.
    async fn initiate_delta_sync(
        &self,
        doc_key: &str,
        peer_id: EndpointId,
        doc: &Automerge,
    ) -> Result<()> {
        // Get or create sync state for this peer
        let mut sync_state = self.get_or_create_sync_state(doc_key, peer_id);

        // Generate initial sync message using SyncDoc trait
        // NOTE: generate_sync_message mutates sync_state internally to track which heads
        // have been "prepared for sending". We must only persist this state AFTER
        // successful send, otherwise retries will fail with "nothing to send".
        let message = match SyncDoc::generate_sync_message(doc, &mut sync_state) {
            Some(msg) => {
                tracing::debug!(
                    "initiate_delta_sync: generated sync message, encoded_len={}",
                    msg.clone().encode().len()
                );
                msg
            }
            None => {
                // Per Automerge's sync-protocol contract,
                // `generate_sync_message` returns `None` when the local
                // document is **already in sync** with the peer (per
                // the cached `sync_state`) — there is nothing new to
                // send. That's a desired terminal state, not a
                // failure. Returning `Err` here previously caused
                // `sync_all_documents_with_peer` to log "Failed to
                // sync document X" and feed the circuit breaker — so
                // documents that were fully converged across the mesh
                // showed up as continuous sync failures, drove
                // unnecessary retries on the Peat ATAK consumer, and
                // skewed the breaker into Open against peers that
                // were actually healthy (observed during the Galaxy
                // Tab A9+ field test: hundreds of "No initial sync
                // message to send" warnings per minute against peers
                // whose docs were entirely in sync; cross-references
                // defenseunicorns/peat#873).
                //
                // No `update_sync_state` call is needed: on `None`
                // there are no new heads "prepared for sending"
                // (the mutation comment on `generate_sync_message`
                // applies only when it actually returns a message).
                tracing::debug!(
                    "initiate_delta_sync: doc {} already in sync with peer {:?}; nothing to send",
                    doc_key,
                    peer_id
                );
                return Ok(());
            }
        };

        // Send message to peer with document key BEFORE updating sync state
        // This ensures that if send fails, we can retry with the same state
        tracing::debug!(
            "initiate_delta_sync: sending sync message to peer {:?}",
            peer_id
        );
        self.send_sync_message_for_doc(peer_id, doc_key, &message)
            .await?;
        tracing::debug!("initiate_delta_sync: sync message sent successfully");

        // Only update sync state AFTER successful send
        self.update_sync_state(doc_key, peer_id, sync_state);

        Ok(())
    }

    /// Send a state snapshot for LatestOnly sync mode (Issue #355, #435)
    ///
    /// Instead of delta-based sync, sends the full document state.
    /// Uses persistent channels when available.
    async fn send_state_snapshot(
        &self,
        peer_id: EndpointId,
        doc_key: &str,
        state_bytes: &[u8],
    ) -> Result<()> {
        // Issue #435: Use persistent channel if available
        if let Some(channel_manager) = self.get_channel_manager() {
            let total_bytes = channel_manager
                .send_state_snapshot(peer_id, doc_key, state_bytes.to_vec())
                .await?;

            self.total_bytes_sent
                .fetch_add(total_bytes as u64, Ordering::Relaxed);

            {
                let mut stats = self.peer_stats.write().unwrap_or_else(|e| e.into_inner());
                let peer_stat = stats.entry(peer_id).or_default();
                peer_stat.bytes_sent += total_bytes as u64;
                peer_stat.sync_count += 1;
                peer_stat.last_sync = Some(SystemTime::now());
            }

            tracing::debug!(
                "Sent state snapshot for {} to {:?} via persistent channel ({} bytes)",
                doc_key,
                peer_id,
                total_bytes
            );
            return Ok(());
        }

        // Fallback: Open a new stream (legacy path)
        let conn = self.get_alive_connection(peer_id).await?;

        let (mut send, mut recv) = conn
            .open_bi()
            .await
            .context("Failed to open bidirectional stream")?;

        let doc_key_bytes = doc_key.as_bytes();
        let doc_key_len = doc_key_bytes.len() as u16;

        send.write_all(&doc_key_len.to_be_bytes())
            .await
            .context("Failed to write doc_key length")?;
        send.write_all(doc_key_bytes)
            .await
            .context("Failed to write doc_key")?;
        send.write_all(&[SyncMessageType::StateSnapshot as u8])
            .await
            .context("Failed to write message type")?;

        let state_len = state_bytes.len() as u32;
        send.write_all(&state_len.to_be_bytes())
            .await
            .context("Failed to write state length")?;
        send.write_all(state_bytes)
            .await
            .context("Failed to write state bytes")?;

        send.finish().context("Failed to finish stream")?;
        let _ = recv.stop(0u32.into());

        let total_bytes = 2 + doc_key_bytes.len() + 1 + 4 + state_bytes.len();
        self.total_bytes_sent
            .fetch_add(total_bytes as u64, Ordering::Relaxed);

        {
            let mut stats = self.peer_stats.write().unwrap_or_else(|e| e.into_inner());
            let peer_stat = stats.entry(peer_id).or_default();
            peer_stat.bytes_sent += total_bytes as u64;
            peer_stat.sync_count += 1;
            peer_stat.last_sync = Some(SystemTime::now());
        }

        tracing::debug!(
            "Sent state snapshot for {} to {:?}: {} bytes",
            doc_key,
            peer_id,
            total_bytes
        );

        Ok(())
    }

    /// Receive and process a sync message from a peer
    ///
    /// Applies the changes to the document and generates a response message if needed.
    ///
    /// # Arguments
    ///
    /// * `doc_key` - The document identifier
    /// * `peer_id` - The EndpointId of the sending peer
    /// * `message` - The received sync message
    /// * `message_size` - Size of the received message in bytes (for statistics)
    pub async fn receive_sync_message(
        &self,
        doc_key: &str,
        peer_id: EndpointId,
        message: SyncMessage,
        message_size: usize,
    ) -> Result<()> {
        // Tombstone guard — reject sync for deleted documents (ADR-034).
        // The peer will learn about the tombstone via periodic tombstone sync.
        if let Some(colon_pos) = doc_key.find(':') {
            let collection = &doc_key[..colon_pos];
            let doc_id = &doc_key[colon_pos + 1..];
            if self
                .store
                .has_tombstone(collection, doc_id)
                .unwrap_or(false)
            {
                tracing::debug!(
                    doc_key = doc_key,
                    peer = %peer_id.fmt_short(),
                    "Rejecting sync for tombstoned document"
                );
                return Ok(());
            }
        }

        // Track statistics first
        self.total_bytes_received
            .fetch_add(message_size as u64, Ordering::Relaxed);

        // Update per-peer statistics
        {
            let mut stats = self.peer_stats.write().unwrap_or_else(|e| e.into_inner());
            let peer_stat = stats.entry(peer_id).or_default();
            peer_stat.bytes_received += message_size as u64;
        }

        tracing::debug!(
            "Received sync message for {} from {:?}: {} bytes",
            doc_key,
            peer_id,
            message_size
        );

        // Hold per-document lock across the get→apply→put sequence to prevent
        // compact() from overwriting concurrent sync changes (Issue #74).
        // Lock is released before the async send to avoid holding across .await.
        let response = {
            let _guard = self.store.lock_doc(doc_key);

            let mut doc = self.store.get(doc_key)?.unwrap_or_else(Automerge::new);
            let doc_len_before = doc.save().len();

            let mut sync_state = self.get_or_create_sync_state(doc_key, peer_id);

            SyncDoc::receive_sync_message(&mut doc, &mut sync_state, message)?;

            let doc_len_after = doc.save().len();
            tracing::debug!(
                "receive_sync_message: doc {} size changed from {} to {} bytes",
                doc_key,
                doc_len_before,
                doc_len_after
            );

            // Issue #115 / peat#891: persist with origin attribution.
            // `change_tx` is still suppressed (the #115 ping-pong fix for
            // the existing local-only pusher), but the origin-tagged
            // `gossip_tx` fires so the new gossip pusher can re-propagate
            // this change to every peer except the source.
            self.put_received(doc_key, &doc, peer_id)?;

            let response = SyncDoc::generate_sync_message(&doc, &mut sync_state);
            if response.is_none() {
                // Sync converged: bound the SyncState's memory WITHOUT losing
                // its quiescence. See `prune_converged_sync_state`
                // (peat-mesh#239). The pre-#239 code reset to a fresh
                // `SyncState` here, which made a converged doc re-emit a sync
                // message → gossip-amplification storm.
                Self::prune_converged_sync_state(&mut sync_state);
            }
            self.update_sync_state(doc_key, peer_id, sync_state);
            response
        };
        // _guard dropped here — lock released before network I/O

        if let Some(response) = response {
            self.send_sync_message_for_doc(peer_id, doc_key, &response)
                .await?;
        }

        Ok(())
    }

    /// Send a sync message to a peer through persistent channel (Issue #435)
    ///
    /// Uses SyncChannelManager for persistent stream multiplexing instead of
    /// opening a new stream for each message.
    async fn send_sync_message_for_doc(
        &self,
        peer_id: EndpointId,
        doc_key: &str,
        message: &SyncMessage,
    ) -> Result<()> {
        // Issue #435: Use persistent channel if available
        if let Some(channel_manager) = self.get_channel_manager() {
            let total_bytes = channel_manager
                .send_delta_sync(peer_id, doc_key, message)
                .await?;

            // Track statistics
            self.total_bytes_sent
                .fetch_add(total_bytes as u64, Ordering::Relaxed);

            {
                let mut stats = self.peer_stats.write().unwrap_or_else(|e| e.into_inner());
                let peer_stat = stats.entry(peer_id).or_default();
                peer_stat.bytes_sent += total_bytes as u64;
                peer_stat.sync_count += 1;
                peer_stat.last_sync = Some(SystemTime::now());
            }

            tracing::debug!(
                "Sent delta sync for {} to peer {:?} via persistent channel ({} bytes)",
                doc_key,
                peer_id,
                total_bytes
            );
            return Ok(());
        }

        // Fallback: Open a new stream (legacy path)
        let conn = self.get_alive_connection(peer_id).await?;

        let (mut send, mut recv) = conn
            .open_bi()
            .await
            .context("Failed to open bidirectional stream")?;

        let doc_key_bytes = doc_key.as_bytes();
        let doc_key_len = doc_key_bytes.len() as u16;

        send.write_all(&doc_key_len.to_be_bytes())
            .await
            .context("Failed to write doc_key length")?;

        send.write_all(doc_key_bytes)
            .await
            .context("Failed to write doc_key")?;

        send.write_all(&[SyncMessageType::DeltaSync as u8])
            .await
            .context("Failed to write message type")?;

        let encoded = message.clone().encode();

        let message_len = encoded.len() as u32;
        send.write_all(&message_len.to_be_bytes())
            .await
            .context("Failed to write message length")?;

        send.write_all(&encoded)
            .await
            .context("Failed to write message")?;

        send.finish().context("Failed to finish stream")?;
        let _ = recv.stop(0u32.into());

        let total_bytes = 2 + doc_key_bytes.len() + 1 + 4 + encoded.len();
        self.total_bytes_sent
            .fetch_add(total_bytes as u64, Ordering::Relaxed);

        {
            let mut stats = self.peer_stats.write().unwrap_or_else(|e| e.into_inner());
            let peer_stat = stats.entry(peer_id).or_default();
            peer_stat.bytes_sent += total_bytes as u64;
            peer_stat.sync_count += 1;
            peer_stat.last_sync = Some(SystemTime::now());
        }

        tracing::debug!(
            "Sent delta sync message for {} to {:?}: {} bytes",
            doc_key,
            peer_id,
            total_bytes
        );

        Ok(())
    }

    /// Receive a sync payload from a peer over Iroh stream (Issue #355)
    ///
    /// Wire format (v2 with message type):
    /// ```text
    /// [2 bytes: doc_key_len][N bytes: doc_key][1 byte: msg_type][4 bytes: payload_len][M bytes: payload]
    /// ```
    ///
    /// Takes the stream by mutable reference so callers can read multiple
    /// frames per stream (peat-mesh#175 / ADR-063 persistent sync streams).
    /// A graceful end-of-stream at the next frame boundary surfaces as
    /// `ReadExactError::FinishedEarly(0)` wrapped in the error chain — callers
    /// that loop over this function should detect that case and exit cleanly.
    ///
    /// Returns (doc_key, payload, total_bytes_received)
    pub(crate) async fn receive_sync_payload_from_stream(
        &self,
        recv: &mut iroh::endpoint::RecvStream,
    ) -> Result<(String, ReceivedSyncPayload, usize)> {
        // Read doc_key length prefix (2 bytes, big-endian)
        let mut doc_key_len_bytes = [0u8; 2];
        recv.read_exact(&mut doc_key_len_bytes)
            .await
            .context("Failed to read doc_key length")?;
        let doc_key_len = u16::from_be_bytes(doc_key_len_bytes) as usize;

        // Read doc_key
        let mut doc_key_bytes = vec![0u8; doc_key_len];
        recv.read_exact(&mut doc_key_bytes)
            .await
            .context("Failed to read doc_key")?;
        let doc_key =
            String::from_utf8(doc_key_bytes).context("Failed to parse doc_key as UTF-8")?;

        // Read message type (1 byte) - Issue #355
        let mut msg_type_byte = [0u8; 1];
        recv.read_exact(&mut msg_type_byte)
            .await
            .context("Failed to read message type")?;

        // Read payload length prefix (4 bytes, big-endian)
        let mut payload_len_bytes = [0u8; 4];
        recv.read_exact(&mut payload_len_bytes)
            .await
            .context("Failed to read payload length")?;
        let payload_len = u32::from_be_bytes(payload_len_bytes) as usize;

        // Read the payload
        let mut buffer = vec![0u8; payload_len];
        recv.read_exact(&mut buffer)
            .await
            .context("Failed to read payload")?;

        // Calculate total bytes: doc_key overhead + type + payload size
        let total_bytes = 2 + doc_key_len + 1 + 4 + payload_len;

        // Parse based on message type
        let payload = match msg_type_byte[0] {
            0x00 => {
                // DeltaSync - decode as Automerge sync message
                let message =
                    SyncMessage::decode(&buffer).context("Failed to decode sync message")?;
                ReceivedSyncPayload::Delta(message)
            }
            0x01 => {
                // StateSnapshot - raw Automerge document bytes
                tracing::debug!(
                    "Received state snapshot for {}: {} bytes",
                    doc_key,
                    buffer.len()
                );
                ReceivedSyncPayload::StateSnapshot(buffer)
            }
            0x04 => {
                // Tombstone - single tombstone message (ADR-034 Phase 2)
                tracing::debug!(
                    "Received single tombstone for {}: {} bytes",
                    doc_key,
                    buffer.len()
                );
                let tombstone = crate::qos::TombstoneSyncMessage::decode(&buffer)
                    .map_err(|e| anyhow::anyhow!("Failed to decode tombstone: {}", e))?;
                ReceivedSyncPayload::Tombstone(tombstone)
            }
            0x05 => {
                // TombstoneBatch - batch of tombstones (ADR-034 Phase 2)
                tracing::debug!(
                    "Received tombstone batch for {}: {} bytes",
                    doc_key,
                    buffer.len()
                );
                let batch = crate::qos::TombstoneBatch::decode(&buffer)
                    .map_err(|e| anyhow::anyhow!("Failed to decode tombstone batch: {}", e))?;
                ReceivedSyncPayload::TombstoneBatch(batch)
            }
            0x06 => {
                // TombstoneAck - acknowledgement (ADR-034 Phase 2)
                // For now, just log and ignore - acks are informational
                tracing::debug!(
                    "Received tombstone ack for {}: {} bytes",
                    doc_key,
                    buffer.len()
                );
                // Return as a batch with no tombstones to indicate ack
                ReceivedSyncPayload::TombstoneBatch(crate::qos::TombstoneBatch::new())
            }
            0x07 => {
                // SyncBatch - batch of sync messages for multiple documents (Issue #438)
                tracing::debug!("Received sync batch: {} bytes", buffer.len());
                let batch = SyncBatch::decode(&buffer)
                    .map_err(|e| anyhow::anyhow!("Failed to decode sync batch: {}", e))?;
                ReceivedSyncPayload::Batch(batch)
            }
            0x08 => {
                // NegentropyInit - Negentropy set reconciliation init (ADR-040, Issue #435)
                tracing::debug!("Received Negentropy init from peer: {} bytes", buffer.len());
                ReceivedSyncPayload::NegentropyInit(buffer)
            }
            0x09 => {
                // NegentropyResponse - Negentropy reconciliation response (ADR-040, Issue #435)
                tracing::debug!(
                    "Received Negentropy response from peer: {} bytes",
                    buffer.len()
                );
                ReceivedSyncPayload::NegentropyResponse(buffer)
            }
            other => {
                return Err(anyhow::anyhow!(
                    "Unknown sync message type: 0x{:02x}",
                    other
                ));
            }
        };

        Ok((doc_key, payload, total_bytes))
    }

    /// Legacy receive function for backwards compatibility
    ///
    /// Calls the new payload receiver and extracts delta sync message.
    /// Returns error if a state snapshot is received (caller should use new API).
    async fn receive_sync_message_from_stream(
        &self,
        mut recv: iroh::endpoint::RecvStream,
    ) -> Result<(String, SyncMessage, usize)> {
        let (doc_key, payload, total_bytes) =
            self.receive_sync_payload_from_stream(&mut recv).await?;

        match payload {
            ReceivedSyncPayload::Delta(message) => Ok((doc_key, message, total_bytes)),
            ReceivedSyncPayload::StateSnapshot(_) => Err(anyhow::anyhow!(
                "Received state snapshot but expected delta sync message for {}",
                doc_key
            )),
            ReceivedSyncPayload::Tombstone(_) | ReceivedSyncPayload::TombstoneBatch(_) => {
                Err(anyhow::anyhow!(
                    "Received tombstone but expected delta sync message for {}",
                    doc_key
                ))
            }
            ReceivedSyncPayload::Batch(_) => Err(anyhow::anyhow!(
                "Received sync batch but expected delta sync message for {}",
                doc_key
            )),
            ReceivedSyncPayload::NegentropyInit(_) | ReceivedSyncPayload::NegentropyResponse(_) => {
                Err(anyhow::anyhow!(
                    "Received Negentropy message but expected delta sync message for {}",
                    doc_key
                ))
            }
        }
    }

    /// Get or create sync state for a peer
    fn get_or_create_sync_state(&self, doc_key: &str, peer_id: EndpointId) -> SyncState {
        let mut states = self.peer_states.write().unwrap_or_else(|e| e.into_inner());
        states
            .entry(doc_key.to_string())
            .or_default()
            .entry(peer_id)
            .or_default()
            .clone()
    }

    /// Update sync state for a peer
    fn update_sync_state(&self, doc_key: &str, peer_id: EndpointId, state: SyncState) {
        let mut states = self.peer_states.write().unwrap_or_else(|e| e.into_inner());
        states
            .entry(doc_key.to_string())
            .or_default()
            .insert(peer_id, state);
    }

    /// Bound a converged peer `SyncState`'s memory without breaking quiescence
    /// (peat-mesh#239).
    ///
    /// Issue #435 added a convergence reset to stop the `SyncState` growing
    /// unboundedly over a document's edit history. That reset replaced the
    /// whole state with a fresh `SyncState` keeping only `shared_heads` — which
    /// discards the negotiated quiescence (`their_heads` / `last_sent_heads` /
    /// `have_responded`). The consequence: the next `generate_sync_message` for
    /// an already-converged doc re-negotiates from scratch and returns `Some`,
    /// so every transitive-gossip re-sync (peat#891) of a converged doc emits a
    /// no-op delta → the peer re-applies (no change) → re-fires `gossip_tx` →
    /// re-broadcasts → an unbounded gossip-amplification storm that starves new
    /// writes after a couple transactions.
    ///
    /// Fix: RETAIN the converged state — it is quiescent
    /// (`generate_sync_message` → `None`) — and clear only the field that
    /// actually accumulates, `sent_hashes` (a `BTreeSet<ChangeHash>` appended to
    /// per change sent). Quiescence is governed by the heads fields, not
    /// `sent_hashes`, so this is safe; on the next genuine change the peer
    /// no-ops any redundant re-send (bounded, never a cascade).
    fn prune_converged_sync_state(state: &mut SyncState) {
        state.sent_hashes.clear();
    }

    /// Clear sync state for a document (for all peers)
    ///
    /// This should be called when a document is modified locally, to ensure
    /// the next sync attempt will generate a fresh sync message with the new
    /// document heads rather than thinking peers are already up-to-date.
    pub fn clear_sync_state_for_document(&self, doc_key: &str) {
        let mut states = self.peer_states.write().unwrap_or_else(|e| e.into_inner());
        if states.remove(doc_key).is_some() {
            tracing::debug!("Cleared sync state for document {}", doc_key);
        }
    }

    /// Clear all sync state for a peer (call on disconnect/reconnect)
    ///
    /// This removes sync state for a peer across ALL documents. Call this when:
    /// - A peer disconnects (to allow fresh sync on reconnect)
    /// - A peer reconnects (to ensure sync starts from scratch)
    ///
    /// Without this, reconnecting peers may fail to sync because the stale
    /// sync state thinks "I already sent those changes" even though the peer
    /// never received them.
    pub fn clear_peer_sync_state(&self, peer_id: EndpointId) {
        let mut states = self.peer_states.write().unwrap_or_else(|e| e.into_inner());
        let mut cleared_count = 0;
        for (_doc_key, peer_map) in states.iter_mut() {
            if peer_map.remove(&peer_id).is_some() {
                cleared_count += 1;
            }
        }
        if cleared_count > 0 {
            tracing::debug!(
                "Cleared sync state for peer {:?} ({} document(s))",
                peer_id,
                cleared_count
            );
        }
    }

    /// Sync a specific document with a peer
    ///
    /// This initiates sync for a single document with a peer.
    /// Use this when a document has been created or modified.
    ///
    /// # Arguments
    ///
    /// * `doc_key` - The document identifier (e.g., "nodes:node-1")
    /// * `peer_id` - The EndpointId of the peer to sync with
    pub async fn sync_document_with_peer(&self, doc_key: &str, peer_id: EndpointId) -> Result<()> {
        self.initiate_sync(doc_key, peer_id).await
    }

    /// Sync a document with all connected peers
    ///
    /// This initiates sync for a single document with all currently connected peers.
    /// Clears existing sync state first to ensure fresh sync messages are generated
    /// even if the document was recently synced but has been modified locally.
    ///
    /// # Arguments
    ///
    /// * `doc_key` - The document identifier (e.g., "nodes:node-1")
    pub async fn sync_document_with_all_peers(&self, doc_key: &str) -> Result<()> {
        let peer_ids = self.transport.connected_peers();
        tracing::info!(
            "sync_document_with_all_peers: syncing {} with {} peers",
            doc_key,
            peer_ids.len()
        );

        // No connected peers: skip the load entirely (peat-mesh#247 — avoid
        // an Automerge clone + `doc.save()` when there's nothing to fan to).
        if peer_ids.is_empty() {
            return Ok(());
        }

        // Clear sync state to ensure we generate fresh sync messages
        // This is important after local document modifications
        self.clear_sync_state_for_document(doc_key);

        // Load + serialize the document ONCE for the whole fanout
        // (peat-mesh#247). The snapshot bytes / loaded doc are
        // peer-independent, so doing this here instead of inside each peer's
        // `initiate_sync` collapses the per-document cost from O(N_peers)
        // Automerge clones + `doc.save()` serializations down to O(1). On the
        // serial change-driven push loop (`AutomergeBackend::start_sync`),
        // that per-document cost is exactly the head-of-line stall a backlog
        // of writes imposes on a probe write. `None` => the document is
        // tombstoned/absent; nothing to send, every peer is a clean skip.
        let prepared = self.prepare_doc_for_sync(doc_key)?;
        let prepared = prepared.as_ref();

        // Run per-peer syncs concurrently so a cooldown wait on one peer
        // does not head-of-line stall the others. With the prior serial
        // `.await` loop, the worst-case latency to drain a single
        // `change_rx` event after the defer-on-cooldown change in
        // `initiate_sync` was O(N_peers * sync_cooldown) — at 5 peers, half
        // a second; at 20 peers, two seconds — meaningfully regressing
        // convergence on larger meshes. `join_all` runs the per-peer
        // futures cooperatively on the current task (no `Arc<Self>`
        // clone), each one waits its own cooldown independently, and
        // per-peer errors stay logged-and-swallowed here. Each future
        // borrows the single shared `PreparedDoc`.
        let sync_tasks = peer_ids.into_iter().map(|peer_id| async move {
            tracing::debug!("Syncing {} with peer {:?}", doc_key, peer_id);
            if let Err(e) = self
                .initiate_sync_prepared(doc_key, peer_id, prepared)
                .await
            {
                tracing::warn!("Failed to sync {} with peer {:?}: {}", doc_key, peer_id, e);
            }
        });
        futures::future::join_all(sync_tasks).await;

        Ok(())
    }

    /// Sync all existing documents with a newly connected peer (Issue #235)
    ///
    /// This is called when a new peer connection is established to ensure
    /// documents created before the peer connected are synchronized.
    ///
    /// # Arguments
    ///
    /// * `peer_id` - The EndpointId of the newly connected peer
    pub async fn sync_all_documents_with_peer(&self, peer_id: EndpointId) -> Result<()> {
        // Peer-level circuit-breaker gate (peat-mesh#131). Without this,
        // a peer that's mDNS-discoverable but iroh-unreachable triggers
        // one full N-document iteration per rediscovery (~10s on Android),
        // even though every per-doc `initiate_sync()` call below would
        // immediately bail at the gate at line 800. Each iteration logs
        // an "Syncing N existing documents" line + N per-doc bail
        // warnings; under sustained rediscovery this fed the OOM chain
        // peat#873 tracked on Android. Same gate as the per-doc path
        // (`initiate_sync` at line 800); short-circuiting at the peer
        // level keeps the work proportional to actual progress, not to
        // rediscovery cadence.
        if self.error_handler.should_block_sync(&peer_id) {
            tracing::debug!(
                "sync_all_documents_with_peer blocked by circuit breaker for peer {:?} \
                 — skipping document iteration (peat-mesh#131)",
                peer_id
            );
            return Ok(());
        }

        // Get all document keys from the store.
        //
        // Use `keys_with_prefix` not `scan_prefix`: the per-doc Automerge
        // body returned by `scan_prefix` was discarded immediately on the
        // next line, but `Automerge::load` for every document still ran —
        // reconstructing the full op-set of each persisted document
        // simultaneously into a `Vec<(String, Automerge)>`. With even a
        // handful of long-history CRDTs that peak allocation is multi-GB
        // and lived for the entire sort + iterate sequence below, which
        // under sustained mDNS rediscovery on Android (every breaker
        // Open→HalfOpen→Open cycle slips one call through) fed the OOM
        // chain in defenseunicorns/peat#873. `sync_document_with_peer`
        // inside the loop loads each document on demand from the
        // store/cache when it's actually needed.
        let mut doc_keys = self.store.keys_with_prefix("")?;

        tracing::info!(
            "Syncing {} existing documents with new peer {:?}",
            doc_keys.len(),
            peer_id
        );

        // Sort documents by QoS priority (Critical first, Bulk last) then by sync mode
        // (LatestOnly before FullHistory within same priority). This ensures mission-critical
        // data syncs first, and fast state-snapshot syncs run before expensive delta syncs.
        doc_keys.sort_by_key(|key| {
            let collection = Self::collection_from_doc_key(key);
            let qos_class = crate::qos::QoSClass::for_collection(collection);
            let mode = self.sync_mode_for_doc(key);
            let mode_order = match mode {
                SyncMode::LatestOnly | SyncMode::WindowedHistory { .. } => 0u8,
                SyncMode::FullHistory => 1u8,
            };
            // Primary: QoS class (Critical=1 first), Secondary: sync mode (fast first)
            (qos_class.as_u8(), mode_order)
        });

        for doc_key in doc_keys {
            if let Err(e) = self.sync_document_with_peer(&doc_key, peer_id).await {
                tracing::warn!(
                    "Failed to sync document {} with new peer {:?}: {}",
                    doc_key,
                    peer_id,
                    e
                );
            }
        }

        Ok(())
    }

    // === Batch sync methods (Issue #438) ===
    //
    // These methods enable sending multiple document syncs in a single QUIC stream,
    // reducing stream-per-message overhead from O(N×M) to O(N).

    /// Generate the next batch ID
    fn next_batch_id(&self) -> u64 {
        self.next_batch_id.fetch_add(1, Ordering::Relaxed)
    }

    /// Create a batch for multiple documents without sending
    ///
    /// This is useful for channel-based sync where the batch is queued.
    ///
    /// # Arguments
    ///
    /// * `doc_keys` - Document keys to include in the batch
    pub fn create_batch_for_documents(&self, doc_keys: &[&str]) -> Result<SyncBatch> {
        let mut batch = SyncBatch::with_id(self.next_batch_id());

        for doc_key in doc_keys {
            // Get the document
            let doc = match self.store.get(doc_key)? {
                Some(doc) => doc,
                None => {
                    tracing::debug!("Document {} not found, skipping in batch", doc_key);
                    continue;
                }
            };

            // Check sync mode for this collection
            let sync_mode = self.sync_mode_for_doc(doc_key);

            match sync_mode {
                SyncMode::LatestOnly | SyncMode::WindowedHistory { .. } => {
                    // State snapshot — both LatestOnly and WindowedHistory send current
                    // state rather than computing expensive deltas
                    let state_bytes = doc.save();
                    batch.add_snapshot(doc_key, state_bytes);
                }
                SyncMode::FullHistory => {
                    // Delta sync - need to generate message
                    // Note: For batch sync we use a fresh sync state since we're
                    // sending to potentially multiple peers
                    let mut sync_state = SyncState::new();
                    if let Some(message) =
                        automerge::sync::SyncDoc::generate_sync_message(&doc, &mut sync_state)
                    {
                        batch.add_delta(doc_key, &message);
                    }
                }
            }
        }

        Ok(batch)
    }

    /// Send a batch of sync messages to a peer (Issue #438)
    ///
    /// Opens a single QUIC stream and sends all entries in the batch.
    ///
    /// # Wire Format
    ///
    /// ```text
    /// [2 bytes: doc_key_len="batch"][5 bytes: "batch"][1 byte: msg_type=0x07]
    /// [4 bytes: batch_len][batch_bytes...]
    /// ```
    pub async fn send_batch_message(&self, peer_id: EndpointId, batch: &SyncBatch) -> Result<()> {
        if batch.is_empty() {
            tracing::debug!("Empty batch, nothing to send to {:?}", peer_id);
            return Ok(());
        }

        // Get connection to peer
        let conn = self.get_alive_connection(peer_id).await?;

        // Open a bidirectional stream
        let (mut send, mut recv) = conn
            .open_bi()
            .await
            .context("Failed to open bidirectional stream for batch")?;

        // Use "batch" as the doc_key for batch messages
        let doc_key = "batch";
        let doc_key_bytes = doc_key.as_bytes();
        let doc_key_len = doc_key_bytes.len() as u16;

        // Encode the batch
        let batch_bytes = batch.encode();

        // Write doc_key length prefix (2 bytes, big-endian)
        send.write_all(&doc_key_len.to_be_bytes())
            .await
            .context("Failed to write doc_key length")?;

        // Write doc_key
        send.write_all(doc_key_bytes)
            .await
            .context("Failed to write doc_key")?;

        // Write message type (1 byte) - SyncBatch = 0x07
        send.write_all(&[SyncMessageType::SyncBatch as u8])
            .await
            .context("Failed to write message type")?;

        // Write batch length prefix (4 bytes, big-endian)
        let batch_len = batch_bytes.len() as u32;
        send.write_all(&batch_len.to_be_bytes())
            .await
            .context("Failed to write batch length")?;

        // Write the batch bytes
        send.write_all(&batch_bytes)
            .await
            .context("Failed to write batch bytes")?;

        // Finish the stream
        send.finish().context("Failed to finish batch stream")?;

        // Issue #435: Explicitly stop recv stream to prevent resource accumulation
        let _ = recv.stop(0u32.into());

        // Track statistics
        let total_bytes = 2 + doc_key_bytes.len() + 1 + 4 + batch_bytes.len();
        self.total_bytes_sent
            .fetch_add(total_bytes as u64, Ordering::Relaxed);

        // Update per-peer statistics
        {
            let mut stats = self.peer_stats.write().unwrap_or_else(|e| e.into_inner());
            let peer_stat = stats.entry(peer_id).or_default();
            peer_stat.bytes_sent += total_bytes as u64;
            peer_stat.sync_count += batch.len() as u64;
            peer_stat.last_sync = Some(SystemTime::now());
        }

        tracing::debug!(
            "Sent batch {} with {} entries ({} bytes) to {:?}",
            batch.batch_id,
            batch.len(),
            total_bytes,
            peer_id
        );

        Ok(())
    }

    /// Receive and process a batch of sync messages (Issue #438)
    ///
    /// Processes each entry in the batch, applying changes to local documents.
    pub async fn receive_batch_message(
        &self,
        peer_id: EndpointId,
        batch: SyncBatch,
        total_bytes: usize,
    ) -> Result<()> {
        // Track statistics first
        self.total_bytes_received
            .fetch_add(total_bytes as u64, Ordering::Relaxed);

        {
            let mut stats = self.peer_stats.write().unwrap_or_else(|e| e.into_inner());
            let peer_stat = stats.entry(peer_id).or_default();
            peer_stat.bytes_received += total_bytes as u64;
            peer_stat.sync_count += batch.len() as u64;
            peer_stat.last_sync = Some(SystemTime::now());
        }

        tracing::debug!(
            "Received batch {} with {} entries ({} bytes) from {:?}",
            batch.batch_id,
            batch.len(),
            total_bytes,
            peer_id
        );

        // Process each entry in the batch
        for entry in batch.entries {
            match entry.sync_type {
                SyncMessageType::DeltaSync => {
                    // Decode and apply delta sync message
                    match SyncMessage::decode(&entry.payload) {
                        Ok(message) => {
                            if let Err(e) = self
                                .receive_sync_message(
                                    &entry.doc_key,
                                    peer_id,
                                    message,
                                    entry.payload.len(),
                                )
                                .await
                            {
                                tracing::warn!(
                                    "Failed to apply delta sync for {} from batch: {}",
                                    entry.doc_key,
                                    e
                                );
                            }
                        }
                        Err(e) => {
                            tracing::warn!(
                                "Failed to decode delta sync message for {}: {}",
                                entry.doc_key,
                                e
                            );
                        }
                    }
                }
                SyncMessageType::StateSnapshot => {
                    // Apply state snapshot
                    let payload_len = entry.payload.len();
                    if let Err(e) = self
                        .apply_state_snapshot(&entry.doc_key, peer_id, entry.payload, payload_len)
                        .await
                    {
                        tracing::warn!(
                            "Failed to apply state snapshot for {} from batch: {}",
                            entry.doc_key,
                            e
                        );
                    }
                }
                SyncMessageType::Tombstone => {
                    // Handle tombstone
                    match crate::qos::TombstoneSyncMessage::decode(&entry.payload) {
                        Ok(tombstone_msg) => {
                            if let Err(e) = self
                                .handle_incoming_tombstone(
                                    &entry.doc_key,
                                    peer_id,
                                    tombstone_msg,
                                    entry.payload.len(),
                                )
                                .await
                            {
                                tracing::warn!(
                                    "Failed to apply tombstone for {} from batch: {}",
                                    entry.doc_key,
                                    e
                                );
                            }
                        }
                        Err(e) => {
                            tracing::warn!(
                                "Failed to decode tombstone for {}: {}",
                                entry.doc_key,
                                e
                            );
                        }
                    }
                }
                other => {
                    tracing::warn!(
                        "Unexpected sync type {:?} in batch entry for {}",
                        other,
                        entry.doc_key
                    );
                }
            }
        }

        Ok(())
    }

    /// Sync multiple documents with a single peer using batch (Issue #438)
    ///
    /// Creates a batch containing all specified documents and sends it in a single stream.
    /// This reduces stream overhead from O(M) to O(1) for M documents.
    ///
    /// # Arguments
    ///
    /// * `doc_keys` - Document keys to sync
    /// * `peer_id` - Target peer
    pub async fn sync_documents_batch(&self, doc_keys: &[&str], peer_id: EndpointId) -> Result<()> {
        // Check circuit breaker before attempting sync
        if self.error_handler.should_block_sync(&peer_id) {
            tracing::warn!(
                "Batch sync blocked by circuit breaker for peer {:?}",
                peer_id
            );
            // Symmetric with initiate_sync — preserve the typed
            // SyncError variant through anyhow so SDK callers /
            // regression tests can downcast back to the typed
            // variant uniformly across both public sync entry
            // points. peat-mesh#109 QA round-2.
            return Err(anyhow::Error::from(SyncError::CircuitBreakerOpen));
        }

        // Clear sync state for all documents in batch
        for doc_key in doc_keys {
            self.clear_sync_state_for_document(doc_key);
        }

        // Create the batch
        let batch = self.create_batch_for_documents(doc_keys)?;

        if batch.is_empty() {
            tracing::debug!("No documents to sync in batch with {:?}", peer_id);
            return Ok(());
        }

        // Send the batch
        let result = self.send_batch_message(peer_id, &batch).await;

        // Handle the result through error handler
        match &result {
            Ok(_) => {
                self.error_handler.record_success(&peer_id);
                // Record sync for each document's cooldown tracking
                for doc_key in doc_keys {
                    self.flow_controller.record_sync(&peer_id, doc_key);
                }
                tracing::debug!(
                    "Batch sync of {} docs with peer {:?} succeeded",
                    batch.len(),
                    peer_id
                );
            }
            Err(e) => {
                // Use the shared `categorize_sync_error` so the batch
                // sync path benefits from the same widened Network-class
                // pattern set as `initiate_sync`. The pre-fix inline
                // match here had the identical "only matches connection/
                // network substrings" gap — `send_batch_message`'s
                // "Failed to write doc_key length" / "Failed to write
                // batch length" failures fell through to Protocol
                // (non-retryable), tripped the circuit breaker, and
                // blocked further batch syncs to the peer. peat#850.
                let sync_error = categorize_sync_error(&e.to_string());

                if let Err(SyncError::CircuitBreakerOpen) =
                    self.error_handler.handle_error(&peer_id, sync_error)
                {
                    tracing::error!("Circuit breaker opened for peer {:?}", peer_id);
                }
            }
        }

        result
    }

    /// Sync multiple documents with all connected peers using batch (Issue #438)
    ///
    /// Sends a batch to each connected peer in parallel.
    pub async fn sync_documents_batch_with_all_peers(&self, doc_keys: &[&str]) -> Result<()> {
        let peer_ids = self.transport.connected_peers();

        tracing::info!(
            "Batch syncing {} documents with {} peers",
            doc_keys.len(),
            peer_ids.len()
        );

        // Clear sync state for all documents
        for doc_key in doc_keys {
            self.clear_sync_state_for_document(doc_key);
        }

        // Send to each peer (could be parallelized with futures::join_all)
        for peer_id in peer_ids {
            if let Err(e) = self.sync_documents_batch(doc_keys, peer_id).await {
                tracing::warn!(
                    "Failed to batch sync {} docs with peer {:?}: {}",
                    doc_keys.len(),
                    peer_id,
                    e
                );
            }
        }

        Ok(())
    }

    /// Get sync targets for a given sync direction (Issue #438 Phase 3)
    ///
    /// Returns the appropriate peer IDs to sync with based on the direction
    /// and the hierarchical router configuration.
    ///
    /// If no hierarchical router is configured, returns all connected peers (broadcast).
    pub async fn get_sync_targets(&self, direction: SyncDirection) -> Vec<EndpointId> {
        let all_peers = self.transport.connected_peers();

        // If no sync router, fall back to broadcast
        let router = match &self.sync_router {
            Some(r) => r,
            None => return all_peers,
        };

        router.get_targets(direction, &all_peers).await
    }

    /// Sync batch with hierarchical routing (Issue #438 Phase 3)
    ///
    /// Splits the batch entries by their sync direction and routes each
    /// sub-batch to the appropriate peers.
    ///
    /// This replaces `sync_documents_batch_with_all_peers` when hierarchical
    /// routing is enabled.
    pub async fn sync_batch_with_hierarchical_routing(&self, batch: &SyncBatch) -> Result<()> {
        // If no sync router, fall back to broadcast
        if self.sync_router.is_none() {
            return self.broadcast_batch(batch).await;
        }

        // Group entries by direction
        let mut upward_entries = Vec::new();
        let mut downward_entries = Vec::new();
        let mut lateral_entries = Vec::new();
        let mut broadcast_entries = Vec::new();

        for entry in &batch.entries {
            let direction = SyncDirection::from_doc_key(&entry.doc_key);
            match direction {
                SyncDirection::Upward => upward_entries.push(entry.clone()),
                SyncDirection::Downward => downward_entries.push(entry.clone()),
                SyncDirection::Lateral => lateral_entries.push(entry.clone()),
                SyncDirection::Broadcast => broadcast_entries.push(entry.clone()),
            }
        }

        tracing::debug!(
            "Hierarchical routing: {} upward, {} downward, {} lateral, {} broadcast",
            upward_entries.len(),
            downward_entries.len(),
            lateral_entries.len(),
            broadcast_entries.len()
        );

        // Route each direction separately
        if !upward_entries.is_empty() {
            let upward_batch = SyncBatch::with_entries(upward_entries);
            let targets = self.get_sync_targets(SyncDirection::Upward).await;
            self.send_batch_to_peers(&upward_batch, &targets).await?;
        }

        if !downward_entries.is_empty() {
            let downward_batch = SyncBatch::with_entries(downward_entries);
            let targets = self.get_sync_targets(SyncDirection::Downward).await;
            self.send_batch_to_peers(&downward_batch, &targets).await?;
        }

        if !lateral_entries.is_empty() {
            let lateral_batch = SyncBatch::with_entries(lateral_entries);
            let targets = self.get_sync_targets(SyncDirection::Lateral).await;
            self.send_batch_to_peers(&lateral_batch, &targets).await?;
        }

        if !broadcast_entries.is_empty() {
            let broadcast_batch = SyncBatch::with_entries(broadcast_entries);
            let targets = self.get_sync_targets(SyncDirection::Broadcast).await;
            self.send_batch_to_peers(&broadcast_batch, &targets).await?;
        }

        Ok(())
    }

    /// Send batch to specific peers
    async fn send_batch_to_peers(&self, batch: &SyncBatch, peers: &[EndpointId]) -> Result<()> {
        if peers.is_empty() {
            tracing::trace!("No peers to send batch to");
            return Ok(());
        }

        tracing::debug!("Sending batch {} to {} peers", batch.batch_id, peers.len());

        for peer_id in peers {
            if let Err(e) = self.send_batch_message(*peer_id, batch).await {
                tracing::warn!("Failed to send batch to peer {:?}: {}", peer_id, e);
            }
        }

        Ok(())
    }

    /// Broadcast batch to all connected peers
    async fn broadcast_batch(&self, batch: &SyncBatch) -> Result<()> {
        let peers = self.transport.connected_peers();
        self.send_batch_to_peers(batch, &peers).await
    }

    /// Check if hierarchical routing is enabled
    pub fn has_hierarchical_routing(&self) -> bool {
        self.sync_router.is_some()
    }

    /// Get sync router reference (if configured)
    pub fn sync_router(&self) -> Option<&Arc<dyn SyncRouter>> {
        self.sync_router.as_ref()
    }

    // === Tombstone sync methods (ADR-034 Phase 2) ===

    /// Send all tombstones to a peer as a batch
    ///
    /// Called when connecting to a new peer to exchange deletion markers.
    /// This ensures the peer knows about all documents we've deleted.
    ///
    /// # Wire Format
    ///
    /// ```text
    /// [2 bytes: doc_key_len][N bytes: doc_key][1 byte: msg_type=0x05][4 bytes: batch_len][M bytes: batch]
    /// ```
    pub async fn send_tombstones_to_peer(&self, peer_id: EndpointId) -> Result<()> {
        // Get all tombstones from storage
        let tombstones = self.store.get_all_tombstones()?;

        if tombstones.is_empty() {
            tracing::debug!("No tombstones to send to peer {:?}", peer_id);
            return Ok(());
        }

        tracing::info!(
            "Sending {} tombstones to peer {:?}",
            tombstones.len(),
            peer_id
        );

        // Convert to TombstoneSyncMessages with direction
        let sync_messages: Vec<crate::qos::TombstoneSyncMessage> = tombstones
            .into_iter()
            .map(crate::qos::TombstoneSyncMessage::from_tombstone)
            .collect();

        // Create batch
        let batch = crate::qos::TombstoneBatch::with_messages(sync_messages);

        // Encode the batch
        let payload = batch.encode();

        tracing::debug!(
            "Encoded tombstone batch ({} bytes) for peer {:?}",
            payload.len(),
            peer_id
        );

        // Get connection to peer
        let conn = self
            .transport
            .get_connection(&peer_id)
            .context("No connection to peer for tombstone exchange")?;

        // Open a bidirectional stream
        let (mut send, mut recv) = conn
            .open_bi()
            .await
            .context("Failed to open bidirectional stream for tombstone exchange")?;

        // Use "tombstones:batch" as the doc_key for tombstone batches
        let doc_key = "tombstones:batch";
        let doc_key_bytes = doc_key.as_bytes();
        let doc_key_len = doc_key_bytes.len() as u16;

        // Write doc_key length prefix (2 bytes, big-endian)
        send.write_all(&doc_key_len.to_be_bytes())
            .await
            .context("Failed to write doc_key length")?;

        // Write doc_key
        send.write_all(doc_key_bytes)
            .await
            .context("Failed to write doc_key")?;

        // Write message type (1 byte) - TombstoneBatch = 0x05
        send.write_all(&[SyncMessageType::TombstoneBatch as u8])
            .await
            .context("Failed to write message type")?;

        // Write payload length prefix (4 bytes, big-endian)
        let payload_len = payload.len() as u32;
        send.write_all(&payload_len.to_be_bytes())
            .await
            .context("Failed to write payload length")?;

        // Write the payload bytes
        send.write_all(&payload)
            .await
            .context("Failed to write tombstone batch payload")?;

        // Finish the stream
        send.finish().context("Failed to finish tombstone stream")?;

        // Issue #435: Explicitly stop recv stream to prevent resource accumulation
        let _ = recv.stop(0u32.into());

        // Track statistics: bytes sent = doc_key overhead + type + payload size
        let total_bytes = 2 + doc_key_bytes.len() + 1 + 4 + payload.len();
        self.total_bytes_sent
            .fetch_add(total_bytes as u64, Ordering::Relaxed);

        {
            let mut stats = self.peer_stats.write().unwrap_or_else(|e| e.into_inner());
            let peer_stat = stats.entry(peer_id).or_default();
            peer_stat.bytes_sent += total_bytes as u64;
            peer_stat.sync_count += 1;
            peer_stat.last_sync = Some(SystemTime::now());
        }

        tracing::debug!(
            "Successfully sent tombstone batch ({} bytes) to peer {:?}",
            total_bytes,
            peer_id
        );

        Ok(())
    }

    /// Exchange tombstones with a peer on connection
    ///
    /// Called when a new peer connection is established.
    /// Sends our tombstones and prepares to receive theirs.
    pub async fn sync_tombstones_with_peer(&self, peer_id: EndpointId) -> Result<()> {
        tracing::debug!("Initiating tombstone exchange with peer {:?}", peer_id);

        // Send our tombstones to the peer
        if let Err(e) = self.send_tombstones_to_peer(peer_id).await {
            tracing::warn!("Failed to send tombstones to peer {:?}: {}", peer_id, e);
            // Don't fail the whole sync just because tombstone exchange failed
        }

        Ok(())
    }

    /// Apply a tombstone received from a peer
    ///
    /// Stores the tombstone and optionally deletes the local document.
    pub async fn apply_tombstone(
        &self,
        tombstone: &crate::qos::Tombstone,
        peer_id: EndpointId,
    ) -> Result<bool> {
        // Check if we already have this tombstone
        if self
            .store
            .has_tombstone(&tombstone.collection, &tombstone.document_id)?
        {
            tracing::trace!(
                "Tombstone for {}:{} already exists, skipping",
                tombstone.collection,
                tombstone.document_id
            );
            return Ok(false);
        }

        // Store the tombstone
        self.store.put_tombstone(tombstone)?;

        // Delete the local document if it exists. Route through
        // `delete_with_origin(Remote(peer))` (peat-mesh#202) so the
        // `observer_tx` channel fires for CDC consumers (closing the
        // documented "fires for ALL document changes" contract) and
        // the origin-tagged `gossip_tx` carries peer attribution for
        // transitive-gossip drivers — while the local-only
        // `change_tx` stays suppressed to preserve the peat-mesh#115
        // ping-pong invariant.
        let doc_key = format!("{}:{}", tombstone.collection, tombstone.document_id);
        if self.store.get(&doc_key)?.is_some() {
            self.store
                .delete_with_origin(&doc_key, ChangeOrigin::Remote(peer_id.to_string()))?;
            tracing::info!(
                "Applied tombstone from peer {:?}: deleted document {}",
                peer_id,
                doc_key
            );
        }

        Ok(true)
    }

    /// Handle an incoming sync connection from a peer
    ///
    /// This is called when a peer initiates sync with us.
    pub async fn handle_incoming_sync(&self, conn: Connection) -> Result<()> {
        let peer_id = conn.remote_id();

        // Accept a bidirectional stream
        let (_send, recv) = conn
            .accept_bi()
            .await
            .context("Failed to accept bidirectional stream")?;

        // Receive the sync message (now includes doc_key and size in wire format)
        let (doc_key, message, message_size) = self.receive_sync_message_from_stream(recv).await?;

        // Process the message with statistics tracking
        self.receive_sync_message(&doc_key, peer_id, message, message_size)
            .await?;

        Ok(())
    }

    /// Handle an incoming sync stream (when streams are accepted externally)
    ///
    /// Persistent-stream variant (peat-mesh#175 / ADR-063): loops over frames
    /// on the same accepted bidirectional stream until the peer half-closes
    /// the send side. Each frame uses the v2 wire format
    /// (`receive_sync_payload_from_stream`) and dispatches via
    /// `dispatch_received_payload`. End-of-stream at a frame boundary
    /// (`ReadExactError::FinishedEarly(0)`) terminates the loop cleanly.
    ///
    /// # Arguments
    ///
    /// * `peer_id` - The EndpointId of the peer (for stats tracking)
    /// * `send` - The send half of the bidirectional stream (used for Negentropy responses)
    /// * `recv` - The receive half of the bidirectional stream
    pub async fn handle_incoming_sync_stream(
        &self,
        peer_id: EndpointId,
        mut send: iroh::endpoint::SendStream,
        mut recv: iroh::endpoint::RecvStream,
    ) -> Result<()> {
        loop {
            match self.receive_sync_payload_from_stream(&mut recv).await {
                Ok((doc_key, payload, payload_size)) => {
                    self.dispatch_received_payload(
                        peer_id,
                        doc_key,
                        payload,
                        payload_size,
                        &mut send,
                    )
                    .await?;
                }
                Err(e) if is_graceful_stream_end(&e) => {
                    tracing::debug!(
                        peer = %peer_id.fmt_short(),
                        "Sync stream gracefully ended at frame boundary"
                    );
                    return Ok(());
                }
                Err(e) => return Err(e),
            }
        }
    }

    /// Dispatch a parsed payload to its per-msg-type handler (peat-mesh#175).
    ///
    /// Shared between the accept-side persistent loop
    /// (`handle_incoming_sync_stream`) and the connect-side `SyncChannel`
    /// receiver. Keeps the dispatch table in one place so the two loops
    /// can never drift apart on supported variants.
    pub(crate) async fn dispatch_received_payload(
        &self,
        peer_id: EndpointId,
        doc_key: String,
        payload: ReceivedSyncPayload,
        payload_size: usize,
        send: &mut iroh::endpoint::SendStream,
    ) -> Result<()> {
        match payload {
            ReceivedSyncPayload::Delta(message) => {
                self.receive_sync_message(&doc_key, peer_id, message, payload_size)
                    .await?;
            }
            ReceivedSyncPayload::StateSnapshot(state_bytes) => {
                self.apply_state_snapshot(&doc_key, peer_id, state_bytes, payload_size)
                    .await?;
            }
            ReceivedSyncPayload::Tombstone(tombstone_msg) => {
                self.handle_incoming_tombstone(&doc_key, peer_id, tombstone_msg, payload_size)
                    .await?;
            }
            ReceivedSyncPayload::TombstoneBatch(batch) => {
                self.handle_incoming_tombstone_batch(&doc_key, peer_id, batch, payload_size)
                    .await?;
            }
            ReceivedSyncPayload::Batch(batch) => {
                self.receive_batch_message(peer_id, batch, payload_size)
                    .await?;
            }
            ReceivedSyncPayload::NegentropyInit(message) => {
                self.handle_negentropy_init(peer_id, message, send).await?;
            }
            ReceivedSyncPayload::NegentropyResponse(message) => {
                self.handle_negentropy_response(peer_id, message, send)
                    .await?;
            }
        }
        Ok(())
    }

    /// Apply a state snapshot to a document (Issue #355)
    ///
    /// Used for LatestOnly sync mode. Replaces the local document with the
    /// received state, or merges if the document already exists.
    async fn apply_state_snapshot(
        &self,
        doc_key: &str,
        peer_id: EndpointId,
        state_bytes: Vec<u8>,
        payload_size: usize,
    ) -> Result<()> {
        // Tombstone guard — reject state snapshots for deleted documents (ADR-034).
        // Prevents resurrection of documents deleted in this partition.
        if let Some(colon_pos) = doc_key.find(':') {
            let collection = &doc_key[..colon_pos];
            let doc_id = &doc_key[colon_pos + 1..];
            if self
                .store
                .has_tombstone(collection, doc_id)
                .unwrap_or(false)
            {
                tracing::debug!(
                    doc_key = doc_key,
                    peer = %peer_id.fmt_short(),
                    "Rejecting state snapshot for tombstoned document"
                );
                return Ok(());
            }
        }

        // Track statistics first
        self.total_bytes_received
            .fetch_add(payload_size as u64, Ordering::Relaxed);

        // Update per-peer statistics
        {
            let mut stats = self.peer_stats.write().unwrap_or_else(|e| e.into_inner());
            let peer_stat = stats.entry(peer_id).or_default();
            peer_stat.bytes_received += payload_size as u64;
            peer_stat.sync_count += 1;
            peer_stat.last_sync = Some(SystemTime::now());
        }

        tracing::debug!(
            "Applying state snapshot for {} from {:?}: {} bytes",
            doc_key,
            peer_id,
            state_bytes.len()
        );

        // Load the received document
        let received_doc =
            Automerge::load(&state_bytes).context("Failed to load state snapshot")?;

        // Hold per-document lock across the get→merge→put sequence (Issue #74).
        {
            let _guard = self.store.lock_doc(doc_key);

            let mut received_doc = received_doc;
            match self.store.get(doc_key) {
                Ok(Some(mut existing_doc)) => {
                    existing_doc
                        .merge(&mut received_doc)
                        .context("Failed to merge state snapshot")?;
                    // peat#891: origin-tagged put; see put_received doc-comment.
                    self.put_received(doc_key, &existing_doc, peer_id)?;
                    tracing::debug!("Merged state snapshot into existing document {}", doc_key);
                }
                Ok(None) => {
                    self.put_received(doc_key, &received_doc, peer_id)?;
                    tracing::debug!("Stored new document {} from state snapshot", doc_key);
                }
                Err(e) => {
                    tracing::warn!(
                        "Error checking existing document {}: {}, storing received state",
                        doc_key,
                        e
                    );
                    self.put_received(doc_key, &received_doc, peer_id)?;
                }
            }
        }

        Ok(())
    }

    /// Handle an incoming tombstone message (ADR-034 Phase 2)
    ///
    /// Processes a single tombstone received from a peer. This applies the
    /// deletion locally and may propagate it further based on direction policy.
    async fn handle_incoming_tombstone(
        &self,
        _doc_key: &str,
        peer_id: EndpointId,
        tombstone_msg: crate::qos::TombstoneSyncMessage,
        payload_size: usize,
    ) -> Result<()> {
        // Track statistics
        self.total_bytes_received
            .fetch_add(payload_size as u64, Ordering::Relaxed);

        {
            let mut stats = self.peer_stats.write().unwrap_or_else(|e| e.into_inner());
            let peer_stat = stats.entry(peer_id).or_default();
            peer_stat.bytes_received += payload_size as u64;
            peer_stat.sync_count += 1;
            peer_stat.last_sync = Some(SystemTime::now());
        }

        // peat-mesh#196: Lamport receive-side rule — advance the node-
        // local clock past the inbound tombstone's Lamport so a
        // subsequent locally-authored `next_lamport()` causally
        // follows it. No-op if the coordinator was constructed
        // standalone (no clock wired).
        self.observe_inbound_lamport(tombstone_msg.tombstone.lamport);

        tracing::debug!(
            "Received tombstone for document {} in collection {} from peer {:?}, direction: {:?}",
            tombstone_msg.tombstone.document_id,
            tombstone_msg.tombstone.collection,
            peer_id,
            tombstone_msg.direction
        );

        // Apply tombstone to local store
        let applied = self
            .apply_tombstone(&tombstone_msg.tombstone, peer_id)
            .await?;

        if applied {
            tracing::info!(
                "Applied tombstone for {}:{} from peer {:?}",
                tombstone_msg.tombstone.collection,
                tombstone_msg.tombstone.document_id,
                peer_id
            );

            // Propagate to other peers based on direction policy (ADR-034 Phase 2)
            self.propagate_tombstone_to_peers(&tombstone_msg, peer_id)
                .await;
        }

        Ok(())
    }

    /// Propagate a tombstone to other connected peers based on direction policy
    ///
    /// Implements ADR-034 direction-aware propagation:
    /// - SystemWide: propagate to ALL peers (for security-critical deletions)
    /// - Bidirectional: propagate to all peers (mesh topology)
    /// - UpOnly: propagate only to parent peers (requires hierarchy info)
    /// - DownOnly: propagate only to child peers (requires hierarchy info)
    ///
    /// Note: UpOnly/DownOnly require hierarchy context from the PeatMesh layer.
    /// At this transport level, we can't distinguish parent vs child peers,
    /// so we conservatively propagate bidirectionally for now.
    async fn propagate_tombstone_to_peers(
        &self,
        tombstone_msg: &crate::qos::TombstoneSyncMessage,
        source_peer_id: EndpointId,
    ) {
        use crate::qos::PropagationDirection;

        let direction = tombstone_msg.direction;

        // Get peers to propagate to (excluding source)
        let all_peers = self.transport.connected_peers();
        let target_peers: Vec<EndpointId> = all_peers
            .into_iter()
            .filter(|p| *p != source_peer_id)
            .collect();

        if target_peers.is_empty() {
            tracing::debug!(
                "No other peers to propagate tombstone {}:{} to",
                tombstone_msg.tombstone.collection,
                tombstone_msg.tombstone.document_id
            );
            return;
        }

        // Determine which peers to propagate to based on direction
        let peers_to_propagate: Vec<EndpointId> = match direction {
            PropagationDirection::SystemWide | PropagationDirection::Bidirectional => {
                // Propagate to all connected peers
                tracing::debug!(
                    "Propagating tombstone {}:{} to {} peers ({:?} mode)",
                    tombstone_msg.tombstone.collection,
                    tombstone_msg.tombstone.document_id,
                    target_peers.len(),
                    direction
                );
                target_peers
            }
            PropagationDirection::UpOnly | PropagationDirection::DownOnly => {
                // Use SyncRouter for hierarchy-aware direction filtering
                if let Some(router) = &self.sync_router {
                    let sync_dir = match direction {
                        PropagationDirection::UpOnly => SyncDirection::Upward,
                        PropagationDirection::DownOnly => SyncDirection::Downward,
                        _ => unreachable!(),
                    };
                    let targets = router.get_targets(sync_dir, &target_peers).await;
                    tracing::debug!(
                        "Propagating tombstone {}:{} to {} peers ({:?} via SyncRouter)",
                        tombstone_msg.tombstone.collection,
                        tombstone_msg.tombstone.document_id,
                        targets.len(),
                        direction
                    );
                    targets
                } else {
                    // No router configured — fall back to bidirectional (safe default)
                    tracing::debug!(
                        "No SyncRouter configured for {:?} tombstone {}:{} — falling back to bidirectional",
                        direction,
                        tombstone_msg.tombstone.collection,
                        tombstone_msg.tombstone.document_id,
                    );
                    target_peers
                }
            }
        };

        // Send tombstone to each target peer
        for peer_id in peers_to_propagate {
            if let Err(e) = self
                .send_single_tombstone_to_peer(peer_id, tombstone_msg)
                .await
            {
                tracing::warn!(
                    "Failed to propagate tombstone {}:{} to peer {:?}: {}",
                    tombstone_msg.tombstone.collection,
                    tombstone_msg.tombstone.document_id,
                    peer_id,
                    e
                );
            }
        }
    }

    /// Propagate a locally-created tombstone to all connected peers (Issue #668)
    ///
    /// Unlike `propagate_tombstone_to_peers()` which excludes a source peer,
    /// this method sends to ALL connected peers. Used when a local `delete()`
    /// creates a tombstone that needs immediate propagation.
    pub async fn propagate_tombstone_to_all(
        &self,
        tombstone_msg: &crate::qos::TombstoneSyncMessage,
    ) {
        use crate::qos::PropagationDirection;

        let direction = tombstone_msg.direction;
        let all_peers = self.transport.connected_peers();

        if all_peers.is_empty() {
            return;
        }

        let target_peers: Vec<EndpointId> = match direction {
            PropagationDirection::SystemWide | PropagationDirection::Bidirectional => all_peers,
            PropagationDirection::UpOnly | PropagationDirection::DownOnly => {
                if let Some(router) = &self.sync_router {
                    let sync_dir = match direction {
                        PropagationDirection::UpOnly => SyncDirection::Upward,
                        PropagationDirection::DownOnly => SyncDirection::Downward,
                        _ => unreachable!(),
                    };
                    router.get_targets(sync_dir, &all_peers).await
                } else {
                    // No router — fall back to all peers (safe default)
                    all_peers
                }
            }
        };

        for peer_id in target_peers {
            if let Err(e) = self
                .send_single_tombstone_to_peer(peer_id, tombstone_msg)
                .await
            {
                tracing::warn!(
                    "Failed to propagate tombstone {}:{} to peer {:?}: {}",
                    tombstone_msg.tombstone.collection,
                    tombstone_msg.tombstone.document_id,
                    peer_id,
                    e
                );
            }
        }
    }

    /// Send a single tombstone to a peer
    ///
    /// # Wire Format
    ///
    /// ```text
    /// [2 bytes: doc_key_len][N bytes: doc_key][1 byte: msg_type=0x04][4 bytes: payload_len][M bytes: payload]
    /// ```
    async fn send_single_tombstone_to_peer(
        &self,
        peer_id: EndpointId,
        tombstone_msg: &crate::qos::TombstoneSyncMessage,
    ) -> Result<()> {
        // Get connection to peer
        let conn = self
            .transport
            .get_connection(&peer_id)
            .context("No connection to peer for single tombstone")?;

        // Open a bidirectional stream
        let (mut send, mut recv) = conn
            .open_bi()
            .await
            .context("Failed to open bidirectional stream for single tombstone")?;

        // Use "tombstones:single" as the doc_key for single tombstones
        let doc_key = format!(
            "tombstones:{}:{}",
            tombstone_msg.tombstone.collection, tombstone_msg.tombstone.document_id
        );
        let doc_key_bytes = doc_key.as_bytes();
        let doc_key_len = doc_key_bytes.len() as u16;

        // Encode the tombstone
        let payload = tombstone_msg.encode();

        // Write doc_key length prefix (2 bytes, big-endian)
        send.write_all(&doc_key_len.to_be_bytes())
            .await
            .context("Failed to write doc_key length")?;

        // Write doc_key
        send.write_all(doc_key_bytes)
            .await
            .context("Failed to write doc_key")?;

        // Write message type (1 byte) - Tombstone = 0x04
        send.write_all(&[SyncMessageType::Tombstone as u8])
            .await
            .context("Failed to write message type")?;

        // Write payload length prefix (4 bytes, big-endian)
        let payload_len = payload.len() as u32;
        send.write_all(&payload_len.to_be_bytes())
            .await
            .context("Failed to write payload length")?;

        // Write the payload bytes
        send.write_all(&payload)
            .await
            .context("Failed to write tombstone payload")?;

        // Finish the stream
        send.finish().context("Failed to finish tombstone stream")?;

        // Issue #435: Explicitly stop recv stream to prevent resource accumulation
        let _ = recv.stop(0u32.into());

        // Track statistics
        let total_bytes = 2 + doc_key_bytes.len() + 1 + 4 + payload.len();
        self.total_bytes_sent
            .fetch_add(total_bytes as u64, Ordering::Relaxed);

        {
            let mut stats = self.peer_stats.write().unwrap_or_else(|e| e.into_inner());
            let peer_stat = stats.entry(peer_id).or_default();
            peer_stat.bytes_sent += total_bytes as u64;
        }

        tracing::trace!(
            "Propagated tombstone {}:{} to peer {:?} ({} bytes)",
            tombstone_msg.tombstone.collection,
            tombstone_msg.tombstone.document_id,
            peer_id,
            total_bytes
        );

        Ok(())
    }

    /// Handle an incoming tombstone batch (ADR-034 Phase 2)
    ///
    /// Processes multiple tombstones received from a peer during initial sync
    /// or batch exchange. This is more efficient than sending individual tombstones.
    async fn handle_incoming_tombstone_batch(
        &self,
        _doc_key: &str,
        peer_id: EndpointId,
        batch: crate::qos::TombstoneBatch,
        payload_size: usize,
    ) -> Result<()> {
        // Track statistics
        self.total_bytes_received
            .fetch_add(payload_size as u64, Ordering::Relaxed);

        {
            let mut stats = self.peer_stats.write().unwrap_or_else(|e| e.into_inner());
            let peer_stat = stats.entry(peer_id).or_default();
            peer_stat.bytes_received += payload_size as u64;
            peer_stat.sync_count += 1;
            peer_stat.last_sync = Some(SystemTime::now());
        }

        let count = batch.tombstones.len();
        tracing::info!(
            "Received tombstone batch with {} tombstones from peer {:?}",
            count,
            peer_id
        );

        // peat-mesh#196: advance the node-local Lamport clock past the
        // maximum Lamport in the batch. `observe_atomic` is idempotent
        // and only takes the max, so a single observe of the max
        // value is equivalent to observing each tombstone individually
        // and saves N-1 CAS attempts. No-op if no clock wired.
        if let Some(max_lamport) = batch.tombstones.iter().map(|t| t.tombstone.lamport).max() {
            self.observe_inbound_lamport(max_lamport);
        }

        // Apply each tombstone to local store
        let mut applied_count = 0;
        for tombstone_msg in batch.tombstones {
            match self
                .apply_tombstone(&tombstone_msg.tombstone, peer_id)
                .await
            {
                Ok(true) => applied_count += 1,
                Ok(false) => {
                    // Tombstone already existed, skip
                }
                Err(e) => {
                    tracing::warn!(
                        "Failed to apply tombstone for {}:{}: {}",
                        tombstone_msg.tombstone.collection,
                        tombstone_msg.tombstone.document_id,
                        e
                    );
                }
            }
        }

        tracing::info!(
            "Applied {}/{} tombstones from peer {:?}",
            applied_count,
            count,
            peer_id
        );

        // TODO: Issue #367 - Propagate to other peers based on direction policies

        Ok(())
    }

    /// Get total bytes sent across all peers
    pub fn total_bytes_sent(&self) -> u64 {
        self.total_bytes_sent.load(Ordering::Relaxed)
    }

    /// Get total bytes received across all peers
    pub fn total_bytes_received(&self) -> u64 {
        self.total_bytes_received.load(Ordering::Relaxed)
    }

    /// Get statistics for a specific peer
    pub fn peer_stats(&self, peer_id: &EndpointId) -> Option<PeerSyncStats> {
        self.peer_stats
            .read()
            .unwrap_or_else(|e| e.into_inner())
            .get(peer_id)
            .cloned()
    }

    /// Get statistics for all peers
    pub fn all_peer_stats(&self) -> HashMap<EndpointId, PeerSyncStats> {
        self.peer_stats
            .read()
            .unwrap_or_else(|e| e.into_inner())
            .clone()
    }

    /// Get reference to the error handler for diagnostics
    pub fn error_handler(&self) -> &SyncErrorHandler {
        &self.error_handler
    }

    /// Get reference to the partition detector
    pub fn partition_detector(&self) -> &PartitionDetector {
        &self.partition_detector
    }

    /// Get reference to the flow controller
    pub fn flow_controller(&self) -> &FlowController {
        &self.flow_controller
    }

    /// Get flow control statistics
    pub fn flow_control_stats(&self) -> FlowControlStats {
        self.flow_controller.stats()
    }

    // ========================================================================
    // Negentropy Set Reconciliation (ADR-040, Issue #435)
    // ========================================================================

    /// Get local document inventory as SyncItems for Negentropy reconciliation
    ///
    /// Returns a list of all documents with their keys and timestamps,
    /// suitable for Negentropy set reconciliation.
    fn get_local_sync_items(&self) -> Vec<SyncItem> {
        // Negentropy only needs the doc keys + timestamps; the Automerge
        // body for each document is irrelevant here. Use `keys_with_prefix`
        // rather than `scan_prefix` so the full op-set isn't reconstructed
        // into memory for every persisted document on every Negentropy
        // probe. Same over-fetch class as `sync_all_documents_with_peer`
        // — see defenseunicorns/peat#873.
        let keys = self.store.keys_with_prefix("").unwrap_or_default();
        keys.into_iter()
            .map(|key| {
                // Use current timestamp - could be improved with actual doc timestamps
                let timestamp = SystemTime::now()
                    .duration_since(SystemTime::UNIX_EPOCH)
                    .map(|d| d.as_secs())
                    .unwrap_or(0);
                SyncItem::from_doc_key(&key, timestamp)
            })
            .collect()
    }

    /// Initiate Negentropy sync with a peer
    ///
    /// This discovers which documents need to be synced using O(log n) rounds
    /// instead of syncing all documents blindly.
    ///
    /// # Returns
    ///
    /// The initial Negentropy message to send to the peer.
    pub fn initiate_negentropy_sync(&self, peer_id: EndpointId) -> Result<Vec<u8>> {
        let items = self.get_local_sync_items();
        tracing::debug!(
            "Initiating Negentropy sync with peer {:?}, local_docs={}",
            peer_id,
            items.len()
        );
        self.negentropy_sync.initiate_sync(peer_id, items)
    }

    /// Handle incoming Negentropy message from a peer
    ///
    /// Processes the Negentropy reconciliation message and returns:
    /// - Response message to send back (if reconciliation not complete)
    /// - List of document keys we have that peer needs (have_keys)
    /// - List of document keys peer has that we need (need_keys)
    pub fn handle_negentropy_message(
        &self,
        peer_id: EndpointId,
        message: &[u8],
    ) -> Result<super::negentropy_sync::ReconcileResult> {
        let items = self.get_local_sync_items();
        self.negentropy_sync.handle_message(peer_id, message, items)
    }

    /// Send Negentropy initiation message to peer
    ///
    /// Opens a bidirectional stream and sends the initial Negentropy message.
    pub async fn send_negentropy_init(&self, peer_id: EndpointId) -> Result<()> {
        let init_msg = self.initiate_negentropy_sync(peer_id)?;

        let conn = self.get_alive_connection(peer_id).await?;

        let (mut send, mut recv) = conn
            .open_bi()
            .await
            .context("Failed to open bidirectional stream")?;

        // Wire format: [2 bytes: doc_key_len][doc_key][1 byte: msg_type][4 bytes: len][payload]
        let doc_key = "_negentropy";
        let doc_key_bytes = doc_key.as_bytes();

        send.write_all(&(doc_key_bytes.len() as u16).to_be_bytes())
            .await?;
        send.write_all(doc_key_bytes).await?;
        send.write_all(&[SyncMessageType::NegentropyInit as u8])
            .await?;
        send.write_all(&(init_msg.len() as u32).to_be_bytes())
            .await?;
        send.write_all(&init_msg).await?;
        send.finish()?;

        // Close recv side - we don't need it for this message
        recv.stop(0u32.into())?;

        self.total_bytes_sent.fetch_add(
            2 + doc_key_bytes.len() as u64 + 1 + 4 + init_msg.len() as u64,
            Ordering::Relaxed,
        );

        tracing::debug!(
            "Sent Negentropy init to peer {:?}, msg_len={}",
            peer_id,
            init_msg.len()
        );

        Ok(())
    }

    /// Perform full Negentropy-based sync with a peer
    ///
    /// This is the main entry point for efficient sync:
    /// 1. Negentropy reconciliation to discover differences
    /// 2. Send documents we have that peer needs
    /// 3. Request documents peer has that we need
    pub async fn sync_with_peer_negentropy(&self, peer_id: EndpointId) -> Result<()> {
        let conn = self.get_alive_connection(peer_id).await?;

        // Phase 1: Initiate Negentropy sync
        let init_msg = self.initiate_negentropy_sync(peer_id)?;

        let (mut send, mut recv) = conn
            .open_bi()
            .await
            .context("Failed to open bidirectional stream")?;

        // Wire format: [2 bytes: doc_key_len][doc_key][1 byte: msg_type][4 bytes: len][payload]
        let doc_key = "_negentropy";
        let doc_key_bytes = doc_key.as_bytes();

        // Send init message
        send.write_all(&(doc_key_bytes.len() as u16).to_be_bytes())
            .await?;
        send.write_all(doc_key_bytes).await?;
        send.write_all(&[SyncMessageType::NegentropyInit as u8])
            .await?;
        send.write_all(&(init_msg.len() as u32).to_be_bytes())
            .await?;
        send.write_all(&init_msg).await?;

        self.total_bytes_sent.fetch_add(
            2 + doc_key_bytes.len() as u64 + 1 + 4 + init_msg.len() as u64,
            Ordering::Relaxed,
        );

        // Phase 2: Reconciliation loop
        let mut have_keys: Vec<String> = Vec::new();
        let mut need_keys: Vec<String> = Vec::new();

        loop {
            // Read response with doc_key prefix
            let mut doc_key_len_bytes = [0u8; 2];
            recv.read_exact(&mut doc_key_len_bytes).await?;
            let resp_doc_key_len = u16::from_be_bytes(doc_key_len_bytes) as usize;

            let mut resp_doc_key_bytes = vec![0u8; resp_doc_key_len];
            recv.read_exact(&mut resp_doc_key_bytes).await?;

            let mut type_buf = [0u8; 1];
            recv.read_exact(&mut type_buf).await?;

            let mut len_buf = [0u8; 4];
            recv.read_exact(&mut len_buf).await?;
            let len = u32::from_be_bytes(len_buf) as usize;

            let mut payload = vec![0u8; len];
            recv.read_exact(&mut payload).await?;

            self.total_bytes_received.fetch_add(
                2 + resp_doc_key_len as u64 + 1 + 4 + len as u64,
                Ordering::Relaxed,
            );

            // Process response
            let result = self.handle_negentropy_message(peer_id, &payload)?;

            have_keys.extend(result.have_keys);
            need_keys.extend(result.need_keys);

            if result.is_complete {
                tracing::info!(
                    "Negentropy sync complete with {:?}: have={}, need={}",
                    peer_id,
                    have_keys.len(),
                    need_keys.len()
                );
                break;
            }

            // Send next message
            if let Some(next_msg) = result.next_message {
                send.write_all(&(doc_key_bytes.len() as u16).to_be_bytes())
                    .await?;
                send.write_all(doc_key_bytes).await?;
                send.write_all(&[SyncMessageType::NegentropyResponse as u8])
                    .await?;
                send.write_all(&(next_msg.len() as u32).to_be_bytes())
                    .await?;
                send.write_all(&next_msg).await?;

                self.total_bytes_sent.fetch_add(
                    2 + doc_key_bytes.len() as u64 + 1 + 4 + next_msg.len() as u64,
                    Ordering::Relaxed,
                );
            }
        }

        send.finish()?;

        // Phase 3: Sync documents based on discovery
        // Send documents we have that peer needs
        if !have_keys.is_empty() {
            tracing::debug!(
                "Sending {} documents to peer {:?}",
                have_keys.len(),
                peer_id
            );
            let doc_key_refs: Vec<&str> = have_keys.iter().map(|s| s.as_str()).collect();
            self.sync_documents_batch(&doc_key_refs, peer_id).await?;
        }

        // Request documents peer has that we need
        // (The peer will send these based on their have_keys from reconciliation)

        Ok(())
    }

    /// Get Negentropy sync statistics
    pub fn negentropy_stats(&self) -> super::negentropy_sync::NegentropyStats {
        self.negentropy_sync.stats()
    }

    /// Handle incoming Negentropy init message from a peer
    ///
    /// This is called when a peer initiates Negentropy sync with us.
    /// We process their init message and send back a response.
    async fn handle_negentropy_init(
        &self,
        peer_id: EndpointId,
        message: Vec<u8>,
        send: &mut iroh::endpoint::SendStream,
    ) -> Result<()> {
        tracing::debug!(
            "Handling Negentropy init from {:?}, msg_len={}",
            peer_id,
            message.len()
        );

        // Get our local documents to compare
        let items = self.get_local_sync_items();

        // Start a new session and process their init message
        // First initiate our side (to create session)
        let _init = self.negentropy_sync.initiate_sync(peer_id, items.clone())?;

        // Then handle their message
        let result = self
            .negentropy_sync
            .handle_message(peer_id, &message, items)?;

        // Send documents we have that peer needs
        if !result.have_keys.is_empty() {
            tracing::debug!(
                "Negentropy: we have {} docs peer {:?} needs",
                result.have_keys.len(),
                peer_id
            );
            // Will sync these after reconciliation completes
        }

        // Track documents peer has that we need
        if !result.need_keys.is_empty() {
            tracing::debug!(
                "Negentropy: peer {:?} has {} docs we need",
                peer_id,
                result.need_keys.len()
            );
        }

        // Send response if not complete
        if let Some(next_msg) = &result.next_message {
            // Wire format: [2 bytes: doc_key_len][doc_key][1 byte: msg_type][4 bytes: len][payload]
            // Use "_negentropy" as doc_key for Negentropy messages
            let doc_key = "_negentropy";
            let doc_key_bytes = doc_key.as_bytes();

            send.write_all(&(doc_key_bytes.len() as u16).to_be_bytes())
                .await?;
            send.write_all(doc_key_bytes).await?;
            send.write_all(&[SyncMessageType::NegentropyResponse as u8])
                .await?;
            send.write_all(&(next_msg.len() as u32).to_be_bytes())
                .await?;
            send.write_all(next_msg).await?;
            send.finish()?;

            self.total_bytes_sent.fetch_add(
                2 + doc_key_bytes.len() as u64 + 1 + 4 + next_msg.len() as u64,
                Ordering::Relaxed,
            );

            tracing::debug!(
                "Sent Negentropy response to {:?}, msg_len={}",
                peer_id,
                next_msg.len()
            );
        } else {
            // Reconciliation complete on first round
            tracing::info!(
                "Negentropy sync complete with {:?} on init (have={}, need={})",
                peer_id,
                result.have_keys.len(),
                result.need_keys.len()
            );

            // Send our documents that peer needs
            if !result.have_keys.is_empty() {
                let doc_key_refs: Vec<&str> = result.have_keys.iter().map(|s| s.as_str()).collect();
                self.sync_documents_batch(&doc_key_refs, peer_id).await?;
            }
        }

        Ok(())
    }

    /// Handle incoming Negentropy response message from a peer
    ///
    /// This is called during ongoing Negentropy reconciliation.
    async fn handle_negentropy_response(
        &self,
        peer_id: EndpointId,
        message: Vec<u8>,
        send: &mut iroh::endpoint::SendStream,
    ) -> Result<()> {
        tracing::debug!(
            "Handling Negentropy response from {:?}, msg_len={}",
            peer_id,
            message.len()
        );

        // Process the response
        let items = self.get_local_sync_items();
        let result = self
            .negentropy_sync
            .handle_message(peer_id, &message, items)?;

        if result.is_complete {
            tracing::info!(
                "Negentropy sync complete with {:?} (have={}, need={})",
                peer_id,
                result.have_keys.len(),
                result.need_keys.len()
            );

            // Send our documents that peer needs
            if !result.have_keys.is_empty() {
                let doc_key_refs: Vec<&str> = result.have_keys.iter().map(|s| s.as_str()).collect();
                self.sync_documents_batch(&doc_key_refs, peer_id).await?;
            }
        } else if let Some(next_msg) = &result.next_message {
            // Send next reconciliation message
            let doc_key = "_negentropy";
            let doc_key_bytes = doc_key.as_bytes();

            send.write_all(&(doc_key_bytes.len() as u16).to_be_bytes())
                .await?;
            send.write_all(doc_key_bytes).await?;
            send.write_all(&[SyncMessageType::NegentropyResponse as u8])
                .await?;
            send.write_all(&(next_msg.len() as u32).to_be_bytes())
                .await?;
            send.write_all(next_msg).await?;

            self.total_bytes_sent.fetch_add(
                2 + doc_key_bytes.len() as u64 + 1 + 4 + next_msg.len() as u64,
                Ordering::Relaxed,
            );

            tracing::debug!(
                "Sent next Negentropy message to {:?}, msg_len={}",
                peer_id,
                next_msg.len()
            );
        }

        Ok(())
    }

    /// Send a heartbeat to a peer
    ///
    /// Sends a minimal heartbeat message to verify the peer is reachable.
    /// Wire format: [1 byte: 0x01 (heartbeat marker)][8 bytes: timestamp (u64, big-endian)]
    ///
    /// # Arguments
    ///
    /// * `peer_id` - The EndpointId of the peer to send heartbeat to
    pub async fn send_heartbeat(&self, peer_id: EndpointId) -> Result<()> {
        // Get connection to peer
        let conn = self.get_alive_connection(peer_id).await?;

        // Open a unidirectional stream (heartbeats don't need response)
        let mut send = conn
            .open_uni()
            .await
            .context("Failed to open unidirectional stream")?;

        // Write heartbeat marker (1 byte: 0x01)
        send.write_all(&[0x01])
            .await
            .context("Failed to write heartbeat marker")?;

        // Write timestamp (8 bytes, big-endian)
        let timestamp = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .unwrap()
            .as_millis() as u64;
        send.write_all(&timestamp.to_be_bytes())
            .await
            .context("Failed to write timestamp")?;

        // Finish the stream
        send.finish().context("Failed to finish stream")?;

        tracing::trace!("Sent heartbeat to peer {:?}", peer_id);

        Ok(())
    }

    /// Handle an incoming heartbeat from a peer
    ///
    /// Called when a peer sends a heartbeat. Records the heartbeat
    /// success in the partition detector.
    ///
    /// # Arguments
    ///
    /// * `conn` - The connection the heartbeat arrived on
    pub async fn handle_incoming_heartbeat(&self, conn: Connection) -> Result<()> {
        let peer_id = conn.remote_id();

        // Accept a unidirectional stream
        let mut recv = conn
            .accept_uni()
            .await
            .context("Failed to accept unidirectional stream")?;

        // Read heartbeat marker (1 byte: 0x01)
        let mut marker = [0u8; 1];
        recv.read_exact(&mut marker)
            .await
            .context("Failed to read heartbeat marker")?;

        if marker[0] != 0x01 {
            anyhow::bail!(
                "Invalid heartbeat marker: expected 0x01, got {:#x}",
                marker[0]
            );
        }

        // Read timestamp (8 bytes, big-endian)
        let mut timestamp_bytes = [0u8; 8];
        recv.read_exact(&mut timestamp_bytes)
            .await
            .context("Failed to read timestamp")?;
        let _timestamp = u64::from_be_bytes(timestamp_bytes);

        // Record heartbeat success in partition detector
        self.partition_detector.record_heartbeat_success(&peer_id);

        tracing::trace!("Received heartbeat from peer {:?}", peer_id);

        Ok(())
    }

    /// Handle an incoming heartbeat stream (when streams are accepted externally)
    ///
    /// This is a more efficient variant for continuous accept loops.
    ///
    /// # Arguments
    ///
    /// * `peer_id` - The EndpointId of the peer (for partition detection)
    /// * `recv` - The unidirectional receive stream
    pub async fn handle_incoming_heartbeat_stream(
        &self,
        peer_id: EndpointId,
        mut recv: iroh::endpoint::RecvStream,
    ) -> Result<()> {
        // Read heartbeat marker (1 byte: 0x01)
        let mut marker = [0u8; 1];
        recv.read_exact(&mut marker)
            .await
            .context("Failed to read heartbeat marker")?;

        if marker[0] != 0x01 {
            anyhow::bail!(
                "Invalid heartbeat marker: expected 0x01, got {:#x}",
                marker[0]
            );
        }

        // Read timestamp (8 bytes, big-endian)
        let mut timestamp_bytes = [0u8; 8];
        recv.read_exact(&mut timestamp_bytes)
            .await
            .context("Failed to read timestamp")?;
        let _timestamp = u64::from_be_bytes(timestamp_bytes);

        // Record heartbeat success in partition detector
        self.partition_detector.record_heartbeat_success(&peer_id);

        tracing::trace!("Received heartbeat from peer {:?}", peer_id);

        Ok(())
    }

    /// Send heartbeats to all connected peers
    ///
    /// This is called periodically by the background heartbeat task.
    pub async fn send_heartbeats_to_all_peers(&self) -> Result<()> {
        let peer_ids = self.transport.connected_peers();

        for peer_id in peer_ids {
            // Register peer with partition detector if not already registered
            self.partition_detector.register_peer(peer_id);

            // Send heartbeat
            if let Err(e) = self.send_heartbeat(peer_id).await {
                tracing::debug!("Failed to send heartbeat to {:?}: {}", peer_id, e);
                // Record heartbeat failure - event already logged via tracing in partition_detector
                let _event = self.partition_detector.record_heartbeat_failure(&peer_id);
            }
        }

        Ok(())
    }

    /// Check all peers for partition timeouts
    ///
    /// This is called periodically to detect partitions based on elapsed time
    /// since last successful heartbeat.
    ///
    /// Returns partition events for newly detected partitions (events already logged via tracing).
    pub fn check_partition_timeouts(&self) -> Vec<super::partition_detection::PartitionEvent> {
        self.partition_detector.check_timeouts()
    }
}

/// Convert a sync error string into a `SyncError` variant. Categorization
/// matters because the error handler treats `Network` as retryable but
/// `Protocol` as non-retryable: Protocol failures, when repeated, open
/// the per-peer circuit breaker; once open, every subsequent
/// `initiate_sync` to that peer is blocked at the top of the function.
///
/// The pre-fix categorizer only recognized "connection" / "network" /
/// "document" / "CRDT" substrings — which missed the dominant
/// real-world QUIC stream-I/O failure mode: when the underlying
/// connection drops between `open_bi()` and the per-message
/// `write_all` calls (e.g. during the 60s
/// `CONNECTION_RECYCLE_INTERVAL_SECS` recycle in peat-protocol's
/// `IrohTransport`, or any transient peer disconnect), the writes
/// return errors like `Failed to write doc_key length` / `Failed to
/// write doc_key` / `Failed to write state length`. None contain
/// "connection" or "network", so the categorizer dropped them into
/// `SyncError::Protocol`, the circuit breaker opened after
/// `max_retries`, and every subsequent sync to that peer was blocked.
/// Observed on the 2026-05-11 bench: `request_sync` invocations
/// reported success at the outer FFI layer while ~100% of
/// `sync_document_with_peer` calls per inner doc failed silently.
///
/// The patterns added below cover the QUIC stream-I/O failure
/// surface: anything matching the "Failed to write" / "Failed to
/// read" sync-framing call sites in this file, plus the iroh /
/// transport-layer phrasing ("broken pipe", "reset by peer",
/// "stream closed", "ApplicationClosed", "disconnecting"). All
/// retryable. peat#850.
/// Bail phrase emitted by `AutomergeSyncCoordinator::get_alive_connection`
/// when `Connection::close_reason()` returns `Some(_)`. Pinned as a
/// const so the helper's `anyhow::bail!` and `categorize_sync_error`'s
/// pattern-matching both reference one source of truth — a future
/// refactor that drops "connection" from the phrase has to change
/// this const, and the regression test below catches the change.
#[cfg(feature = "automerge-backend")]
pub(crate) const BAIL_CONNECTION_LOST: &str = "connection lost before stream open";

/// Detect graceful end-of-stream at a frame boundary on a persistent sync
/// stream (peat-mesh#175 / ADR-063).
///
/// When the peer half-closes the send side of a bidirectional stream at a
/// frame boundary, the next `RecvStream::read_exact` for the leading two
/// `doc_key_len` bytes returns `ReadExactError::FinishedEarly(0)`. That is
/// the expected loop-exit signal — not an error. Mid-frame closure
/// (`FinishedEarly(n)` with `n > 0`) is a protocol error and propagates.
///
/// `pub(crate)` so the connect-side `SyncChannel::receive_loop` and the
/// accept-side `handle_incoming_sync_stream` share one classifier — if iroh
/// ever adds another graceful-close variant, only this function needs
/// updating.
#[cfg(feature = "automerge-backend")]
pub(crate) fn is_graceful_stream_end(err: &anyhow::Error) -> bool {
    for cause in err.chain() {
        if let Some(read_exact_err) = cause.downcast_ref::<iroh::endpoint::ReadExactError>() {
            return matches!(
                read_exact_err,
                iroh::endpoint::ReadExactError::FinishedEarly(0)
            );
        }
    }
    false
}

#[cfg(feature = "automerge-backend")]
pub(crate) fn categorize_sync_error(err: &str) -> SyncError {
    // Lowercase once at the top so every pattern check sees the same
    // case-normalized string. The pre-fix variant mixed raw `err.contains`
    // for the "Failed to write/read" patterns with lowercased
    // `lower.contains` for everything else — an asymmetry that could
    // silently drop Network classification if an upstream `anyhow`
    // wrap ever folded the casing of the "Failed to …" segment, or
    // a hand-rolled wrapper used lowercase phrasing. Single
    // case-fold removes the latent footgun.
    let lower = err.to_lowercase();
    let is_transport_io = lower.contains("failed to write")
        || lower.contains("failed to read")
        || lower.contains("connection")
        || lower.contains("network")
        || lower.contains("broken pipe")
        || lower.contains("reset by peer")
        || lower.contains("stream closed")
        || lower.contains("applicationclosed")
        || lower.contains("disconnecting");
    if is_transport_io {
        SyncError::Network(err.to_string())
    } else if lower.contains("document") || lower.contains("crdt") {
        SyncError::Document(err.to_string())
    } else {
        SyncError::Protocol(err.to_string())
    }
}

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

    // === Batch sync tests (Issue #438) ===

    #[test]
    fn test_sync_entry_encode_decode() {
        // Test with DeltaSync type
        let entry = SyncEntry::new(
            "nodes:test-node-1".to_string(),
            SyncMessageType::DeltaSync,
            vec![1, 2, 3, 4, 5],
        );

        let encoded = entry.encode();

        // Verify wire format
        // [2 bytes: doc_key_len][doc_key][1 byte: sync_type][4 bytes: payload_len][payload]
        assert_eq!(u16::from_be_bytes([encoded[0], encoded[1]]), 17); // "nodes:test-node-1" len
        assert_eq!(&encoded[2..19], b"nodes:test-node-1");
        assert_eq!(encoded[19], 0x00); // DeltaSync
        assert_eq!(
            u32::from_be_bytes([encoded[20], encoded[21], encoded[22], encoded[23]]),
            5
        );
        assert_eq!(&encoded[24..], &[1, 2, 3, 4, 5]);

        // Decode and verify roundtrip
        let (decoded, consumed) = SyncEntry::decode(&encoded).unwrap();
        assert_eq!(consumed, encoded.len());
        assert_eq!(decoded.doc_key, "nodes:test-node-1");
        assert_eq!(decoded.sync_type, SyncMessageType::DeltaSync);
        assert_eq!(decoded.payload, vec![1, 2, 3, 4, 5]);
    }

    #[test]
    fn test_sync_entry_encode_decode_state_snapshot() {
        let entry = SyncEntry::new(
            "beacons:beacon-42".to_string(),
            SyncMessageType::StateSnapshot,
            vec![10, 20, 30, 40, 50, 60],
        );

        let encoded = entry.encode();
        let (decoded, _) = SyncEntry::decode(&encoded).unwrap();

        assert_eq!(decoded.doc_key, "beacons:beacon-42");
        assert_eq!(decoded.sync_type, SyncMessageType::StateSnapshot);
        assert_eq!(decoded.payload, vec![10, 20, 30, 40, 50, 60]);
    }

    #[test]
    fn test_sync_batch_empty() {
        let batch = SyncBatch::new();
        assert!(batch.is_empty());
        assert_eq!(batch.len(), 0);
        assert_eq!(batch.payload_size(), 0);
    }

    #[test]
    fn test_sync_batch_encode_decode() {
        let mut batch = SyncBatch::with_id(12345);

        // Add entries
        batch.entries.push(SyncEntry::new(
            "nodes:node-1".to_string(),
            SyncMessageType::DeltaSync,
            vec![1, 2, 3],
        ));
        batch.entries.push(SyncEntry::new(
            "beacons:beacon-1".to_string(),
            SyncMessageType::StateSnapshot,
            vec![4, 5, 6, 7],
        ));

        assert_eq!(batch.len(), 2);
        assert_eq!(batch.payload_size(), 7); // 3 + 4

        let encoded = batch.encode();

        // Verify header
        // [8 bytes: batch_id][8 bytes: created_at][1 byte: ttl][4 bytes: entry_count][entries...]
        assert_eq!(u64::from_be_bytes(encoded[0..8].try_into().unwrap()), 12345);
        assert_eq!(encoded[16], DEFAULT_SYNC_BATCH_TTL); // TTL byte
        assert_eq!(u32::from_be_bytes(encoded[17..21].try_into().unwrap()), 2); // 2 entries

        // Decode and verify roundtrip
        let decoded = SyncBatch::decode(&encoded).unwrap();
        assert_eq!(decoded.batch_id, 12345);
        assert_eq!(decoded.entries.len(), 2);

        assert_eq!(decoded.entries[0].doc_key, "nodes:node-1");
        assert_eq!(decoded.entries[0].sync_type, SyncMessageType::DeltaSync);
        assert_eq!(decoded.entries[0].payload, vec![1, 2, 3]);

        assert_eq!(decoded.entries[1].doc_key, "beacons:beacon-1");
        assert_eq!(decoded.entries[1].sync_type, SyncMessageType::StateSnapshot);
        assert_eq!(decoded.entries[1].payload, vec![4, 5, 6, 7]);
    }

    #[test]
    fn test_sync_batch_with_many_entries() {
        let mut batch = SyncBatch::with_id(99999);

        // Add 100 entries
        for i in 0..100 {
            batch.entries.push(SyncEntry::new(
                format!("docs:doc-{}", i),
                SyncMessageType::DeltaSync,
                vec![i as u8; 10],
            ));
        }

        assert_eq!(batch.len(), 100);
        assert_eq!(batch.payload_size(), 1000); // 100 * 10

        let encoded = batch.encode();
        let decoded = SyncBatch::decode(&encoded).unwrap();

        assert_eq!(decoded.batch_id, 99999);
        assert_eq!(decoded.entries.len(), 100);

        // Verify a few entries
        assert_eq!(decoded.entries[0].doc_key, "docs:doc-0");
        assert_eq!(decoded.entries[0].payload, vec![0u8; 10]);

        assert_eq!(decoded.entries[50].doc_key, "docs:doc-50");
        assert_eq!(decoded.entries[50].payload, vec![50u8; 10]);

        assert_eq!(decoded.entries[99].doc_key, "docs:doc-99");
        assert_eq!(decoded.entries[99].payload, vec![99u8; 10]);
    }

    #[test]
    fn test_sync_entry_decode_error_too_short() {
        let result = SyncEntry::decode(&[0, 1, 2]);
        assert!(result.is_err());
        assert!(result.unwrap_err().to_string().contains("too short"));
    }

    #[test]
    fn test_sync_batch_decode_error_too_short() {
        let result = SyncBatch::decode(&[0, 1, 2, 3, 4, 5, 6, 7, 8, 9]);
        assert!(result.is_err());
        assert!(result.unwrap_err().to_string().contains("too short"));
    }

    // === End batch sync tests ===

    // === SyncDirection tests (Issue #438 Phase 3) ===

    #[test]
    fn test_sync_direction_upward() {
        // Node-related collections should sync upward
        assert_eq!(
            SyncDirection::from_doc_key("nodes:node-1"),
            SyncDirection::Upward
        );
        assert_eq!(
            SyncDirection::from_doc_key("beacons:beacon-42"),
            SyncDirection::Upward
        );
        assert_eq!(
            SyncDirection::from_doc_key("platforms:platform-a"),
            SyncDirection::Upward
        );
        assert_eq!(
            SyncDirection::from_doc_key("summaries:cell-1"),
            SyncDirection::Upward
        );
    }

    #[test]
    fn test_sync_direction_downward() {
        // Commands flow from coordinators to executors
        assert_eq!(
            SyncDirection::from_doc_key("commands:cmd-123"),
            SyncDirection::Downward
        );
        assert_eq!(
            SyncDirection::from_doc_key("commands:urgent-456"),
            SyncDirection::Downward
        );
    }

    #[test]
    fn test_sync_direction_lateral() {
        // Cell state syncs between peers
        assert_eq!(
            SyncDirection::from_doc_key("cells:cell-alpha"),
            SyncDirection::Lateral
        );
        assert_eq!(
            SyncDirection::from_doc_key("cells:formation-1"),
            SyncDirection::Lateral
        );
    }

    #[test]
    fn test_sync_direction_broadcast() {
        // Alerts and reports go everywhere
        assert_eq!(
            SyncDirection::from_doc_key("alerts:alert-1"),
            SyncDirection::Broadcast
        );
        // Both the canonical hyphenated and legacy underscored forms route to
        // Broadcast (peat-mesh#209 QA). The dual-name arm is behavior-neutral
        // today because the unknown-collection fallthrough is also Broadcast,
        // but pinning both forms ensures a future change to the default arm
        // can't silently change `contact-reports` routing.
        assert_eq!(
            SyncDirection::from_doc_key("contact-reports:cr-789"),
            SyncDirection::Broadcast
        );
        assert_eq!(
            SyncDirection::from_doc_key("contact_reports:cr-789"),
            SyncDirection::Broadcast
        );
        assert_eq!(
            SyncDirection::from_doc_key("events:event-1"),
            SyncDirection::Broadcast
        );
        // Unknown collections default to broadcast
        assert_eq!(
            SyncDirection::from_doc_key("unknown:item-1"),
            SyncDirection::Broadcast
        );
        assert_eq!(
            SyncDirection::from_doc_key("custom_collection:x"),
            SyncDirection::Broadcast
        );
    }

    #[test]
    fn test_sync_direction_edge_cases() {
        // Document key without colon - should use whole key as collection
        assert_eq!(SyncDirection::from_doc_key("nodes"), SyncDirection::Upward);
        assert_eq!(
            SyncDirection::from_doc_key("commands"),
            SyncDirection::Downward
        );
        // Empty key defaults to broadcast
        assert_eq!(SyncDirection::from_doc_key(""), SyncDirection::Broadcast);
    }

    #[test]
    fn test_sync_batch_with_entries() {
        let entries = vec![
            SyncEntry::new("nodes:n1".to_string(), SyncMessageType::DeltaSync, vec![1]),
            SyncEntry::new(
                "commands:c1".to_string(),
                SyncMessageType::StateSnapshot,
                vec![2],
            ),
        ];
        let batch = SyncBatch::with_entries(entries);
        assert_eq!(batch.len(), 2);
        assert_eq!(batch.entries[0].doc_key, "nodes:n1");
        assert_eq!(batch.entries[1].doc_key, "commands:c1");
    }

    // === End SyncDirection tests ===

    /// Regression coverage for the per-error `SyncError` categorizer used
    /// inside `initiate_sync`'s error branch. The mis-categorization
    /// surfaced on the 2026-05-11 bench: every QUIC stream-I/O failure
    /// ("Failed to write doc_key length" / "Failed to write doc_key" /
    /// "Failed to write state length") was bucketed into
    /// `SyncError::Protocol` (non-retryable), which opened the
    /// per-peer circuit breaker after max-retries and blocked every
    /// subsequent sync attempt. Markers never propagated cross-mesh
    /// despite `request_sync` reporting success at the outer FFI layer.
    /// peat#850.
    mod categorize_sync_error_tests {
        use super::*;

        /// Every `Failed to write` call site in this file's sync-stream
        /// framing maps to a Network (retryable) error — the underlying
        /// QUIC connection dropped between `open_bi()` and the per-message
        /// write, NOT a protocol violation.
        #[test]
        fn quic_stream_write_failures_are_network() {
            for msg in &[
                "Failed to write doc_key length",
                "Failed to write doc_key",
                "Failed to write message type",
                "Failed to write state length",
                "Failed to write state",
            ] {
                let err = categorize_sync_error(msg);
                assert!(
                    matches!(err, SyncError::Network(_)),
                    "{msg} must be SyncError::Network, got {err:?}"
                );
                assert!(
                    err.is_retryable(),
                    "{msg} must be retryable (else circuit breaker traps the peer)"
                );
            }
        }

        /// Mirror coverage for the receive-side errors.
        #[test]
        fn quic_stream_read_failures_are_network() {
            for msg in &[
                "Failed to read doc_key length",
                "Failed to read doc_key",
                "Failed to read message type",
            ] {
                let err = categorize_sync_error(msg);
                assert!(
                    matches!(err, SyncError::Network(_)),
                    "{msg} must be SyncError::Network, got {err:?}"
                );
                assert!(err.is_retryable());
            }
        }

        /// iroh's transport-layer phrasing for graceful peer shutdown
        /// and unexpected disconnect — also retryable.
        #[test]
        fn iroh_transport_phrases_are_network() {
            for msg in &[
                "ApplicationClosed(ApplicationClose { error_code: 0, reason: b\"disconnecting\" })",
                "connection lost",
                "Network unreachable",
                "broken pipe",
                "stream closed",
                "reset by peer",
            ] {
                let err = categorize_sync_error(msg);
                assert!(
                    matches!(err, SyncError::Network(_)),
                    "{msg} must be SyncError::Network, got {err:?}"
                );
            }
        }

        /// Document/CRDT semantic errors stay categorized as Document
        /// — these aren't recoverable by retry of the same content
        /// (e.g. malformed Automerge changes, divergent doc state).
        #[test]
        fn document_crdt_errors_stay_document() {
            for msg in &[
                "document corrupt",
                "CRDT merge failed",
                "Document does not exist",
            ] {
                let err = categorize_sync_error(msg);
                assert!(
                    matches!(err, SyncError::Document(_)),
                    "{msg} must be SyncError::Document, got {err:?}"
                );
            }
        }

        /// Symptom-level regression: a QUIC stream-I/O error string
        /// routed through `categorize_sync_error` →
        /// `SyncErrorHandler::handle_error` must produce a retryable
        /// outcome (`Ok(Some(retry_delay))`), NOT `Ok(None)`
        /// (non-retryable, give up). This is the actual semantic
        /// difference between the pre-fix Protocol bucketing and the
        /// post-fix Network bucketing. If a future categorizer
        /// refactor accidentally drops "Failed to write" from the
        /// Network set, this regression test fails before the bug
        /// reaches a bench again. peat#850.
        #[tokio::test]
        async fn quic_stream_error_through_handler_is_retryable() {
            use crate::storage::sync_errors::{CircuitBreakerConfig, RetryPolicy};
            use iroh::SecretKey;

            let handler = SyncErrorHandler::with_policies(
                RetryPolicy::default(),
                // Generous failure_threshold so we test retryability,
                // not breaker timing.
                CircuitBreakerConfig {
                    failure_threshold: 100,
                    ..Default::default()
                },
            );
            let peer_id = SecretKey::generate().public();

            let sync_err = categorize_sync_error("Failed to write doc_key length");
            assert!(matches!(sync_err, SyncError::Network(_)));

            let result = handler.handle_error(&peer_id, sync_err);
            match result {
                Ok(Some(_delay)) => { /* retryable, retry scheduled — correct */ }
                Ok(None) => panic!(
                    "QUIC stream-I/O failure must be retryable; got Ok(None) which means \
                     handler told caller to give up. This is the exact pre-fix behavior \
                     that opened the circuit breaker."
                ),
                Err(e) => panic!("unexpected error: {:?}", e),
            }
        }

        /// The `get_alive_connection` helper emits a bail when
        /// `Connection::close_reason()` returns Some. The bail phrase
        /// is pinned by `BAIL_CONNECTION_LOST`; this test feeds that
        /// exact const into the categorizer so the two ends of the
        /// contract are coupled at compile time. A future refactor
        /// that renames the const (and therefore the helper's bail
        /// output) without also updating the categorizer's pattern
        /// set will fail this test before the bug reaches a bench.
        /// End-to-end invariant: helper bail → categorizer → handler
        /// → retryable. peat#850.
        #[test]
        fn get_alive_connection_bail_phrase_is_network() {
            let err = categorize_sync_error(&format!(
                "{} (peer=PublicKey(...), close_reason: Some(ApplicationClosed(...)))",
                BAIL_CONNECTION_LOST
            ));
            assert!(
                matches!(err, SyncError::Network(_)),
                "the helper's bail phrase ({BAIL_CONNECTION_LOST:?}) must classify as Network; got {err:?}"
            );
        }

        /// Mirror: a genuine Protocol error (e.g. unknown sync-message
        /// type) is non-retryable and returns `Ok(None)`. Locks down
        /// the OTHER side of the categorizer contract — Protocol
        /// errors should remain non-retryable so the caller actually
        /// stops trying.
        #[tokio::test]
        async fn genuine_protocol_error_through_handler_is_non_retryable() {
            use crate::storage::sync_errors::{CircuitBreakerConfig, RetryPolicy};
            use iroh::SecretKey;

            let handler = SyncErrorHandler::with_policies(
                RetryPolicy::default(),
                CircuitBreakerConfig {
                    failure_threshold: 100,
                    ..Default::default()
                },
            );
            let peer_id = SecretKey::generate().public();

            let sync_err = categorize_sync_error("Sync protocol version mismatch");
            assert!(matches!(sync_err, SyncError::Protocol(_)));

            let result = handler.handle_error(&peer_id, sync_err);
            assert!(
                matches!(result, Ok(None)),
                "Protocol errors must be non-retryable; got {:?}",
                result
            );
        }

        /// Genuinely-protocol errors (sync-message version mismatch,
        /// unknown message type, signature failure) STAY Protocol —
        /// the categorizer's new pattern set must not over-broaden
        /// and accidentally pull these into Network.
        #[test]
        fn genuine_protocol_errors_stay_protocol() {
            for msg in &[
                "Unknown sync message type: 99",
                "Sync protocol version mismatch",
                "Signature verification failed",
            ] {
                let err = categorize_sync_error(msg);
                assert!(
                    matches!(err, SyncError::Protocol(_)),
                    "{msg} must be SyncError::Protocol, got {err:?}"
                );
            }
        }
    }

    /// Regression coverage for the circuit-breaker one-way trap fixed
    /// in peat-mesh#109 (peat#850).
    ///
    /// The `should_block_sync` unit tests in `sync_errors.rs` verify
    /// the gate's algorithm but say nothing about which gate
    /// `initiate_sync` actually calls. The original bug was precisely
    /// that the production call site used `is_circuit_open` (a
    /// pure-read view that never checks the open_timeout), so once
    /// the breaker tripped it stayed blocking indefinitely. If a
    /// future refactor reverts the call site to `is_circuit_open` —
    /// or introduces a new gate that ignores the timeout — those
    /// algorithm-level tests stay green and the regression returns
    /// silently. These tests pin the wiring.
    #[cfg(feature = "automerge-backend")]
    mod circuit_breaker_call_site_tests {
        use super::*;
        use crate::storage::automerge_store::AutomergeStore;
        use crate::storage::sync_errors::{CircuitBreakerConfig, RetryPolicy, SyncErrorHandler};
        use iroh::SecretKey;
        use std::time::Duration;

        /// A no-op `SyncTransport`: `get_connection` always returns
        /// `None`, so the default `get_or_connect` impl fails with
        /// "no connection to peer". Sufficient for tests that only
        /// care whether the breaker gate fires before or instead of
        /// the transport.
        struct NoOpTransport;

        #[async_trait::async_trait]
        impl SyncTransport for NoOpTransport {
            fn get_connection(&self, _peer_id: &EndpointId) -> Option<Connection> {
                None
            }
            fn connected_peers(&self) -> Vec<EndpointId> {
                Vec::new()
            }
        }

        /// Build a coordinator wired to a `SyncErrorHandler` with a
        /// short `open_timeout` so the test can observe the Open →
        /// HalfOpen transition without minutes of wall-clock sleep.
        ///
        /// We construct the struct directly (vs. going through
        /// `AutomergeSyncCoordinator::new`) because there is no public
        /// constructor that lets callers inject a custom error handler,
        /// and adding one purely for tests would be surface bloat.
        /// Private-field access is fine here — same module.
        fn make_coordinator(error_handler: Arc<SyncErrorHandler>) -> AutomergeSyncCoordinator {
            AutomergeSyncCoordinator {
                store: Arc::new(AutomergeStore::in_memory()),
                transport: Arc::new(NoOpTransport),
                peer_states: Arc::new(RwLock::new(HashMap::new())),
                peer_stats: Arc::new(RwLock::new(HashMap::new())),
                total_bytes_sent: Arc::new(AtomicU64::new(0)),
                total_bytes_received: Arc::new(AtomicU64::new(0)),
                error_handler,
                partition_detector: Arc::new(PartitionDetector::new()),
                flow_controller: Arc::new(
                    FlowController::with_config(FlowControlConfig::default()),
                ),
                sync_mode_registry: Arc::new(SyncModeRegistry::with_defaults()),
                next_batch_id: Arc::new(AtomicU64::new(1)),
                sync_router: None,
                channel_manager: Arc::new(RwLock::new(None)),
                negentropy_sync: Arc::new(NegentropySync::new()),
                ttl_manager: Arc::new(RwLock::new(None)),
                bandwidth_allocation: Arc::new(RwLock::new(None)),
                lamport_clock: Arc::new(RwLock::new(None)),
                fanout: Arc::new(FanoutState {
                    inner: Mutex::new(FanoutQueue::default()),
                    notify: Notify::new(),
                }),
            }
        }

        /// Pre-fix: `initiate_sync` early-bailed on `is_circuit_open`
        /// before any code path could transition Open → HalfOpen, so
        /// the breaker stayed blocking after `open_timeout` elapsed.
        /// Post-fix (`should_block_sync` wired at `initiate_sync:792`):
        /// the gate respects the timeout and lets the next attempt
        /// probe. This test fails on the pre-fix code and passes
        /// post-fix — the canonical regression for peat#850.
        #[tokio::test]
        async fn initiate_sync_clears_breaker_after_open_timeout() {
            // Open_timeout / sleep cushion: 200ms timeout + 300ms sleep
            // gives ~100ms slack to absorb CI scheduling jitter on the
            // self-hosted runner. Tighter values flaked in pre-merge
            // review; the test is single-iteration, so a few hundred ms
            // is cheap.
            let error_handler = Arc::new(SyncErrorHandler::with_policies(
                RetryPolicy::default(),
                CircuitBreakerConfig {
                    failure_threshold: 2,
                    open_timeout: Duration::from_millis(200),
                    ..Default::default()
                },
            ));
            let coordinator = make_coordinator(error_handler.clone());

            let peer_id = SecretKey::generate().public();

            // A present doc so the post-timeout probe fails at the *transport*
            // (NoOpTransport has no connection), not at the absent-doc skip
            // which now returns Ok (peat-mesh#243). The invariant under test is
            // the breaker gate, so we need a downstream failure past it.
            {
                use automerge::transaction::Transactable as _;
                let mut doc = automerge::Automerge::new();
                let mut tx = doc.transaction();
                tx.put(automerge::ROOT, "x", "1").unwrap();
                tx.commit();
                coordinator.store.put("test:doc", &doc).unwrap();
            }

            // Trip the breaker: two Network failures hit the threshold.
            let _ = error_handler.handle_error(&peer_id, SyncError::Network("seed".to_string()));
            let _ = error_handler.handle_error(&peer_id, SyncError::Network("seed".to_string()));
            assert!(
                error_handler.is_circuit_open(&peer_id),
                "precondition: breaker must be Open after threshold failures"
            );

            // Immediately: initiate_sync must be rejected BY THE BREAKER.
            // We downcast to the typed `SyncError` variant instead of
            // substring-matching the Display string — production now
            // builds the anyhow::Error via `Error::from(SyncError)`
            // so the type is preserved. A future cosmetic Display
            // change won't silently flip the assertion.
            let err = coordinator
                .initiate_sync("test:doc", peer_id)
                .await
                .expect_err("tripped breaker must reject sync");
            let sync_err = err
                .downcast_ref::<SyncError>()
                .expect("pre-timeout error must be a typed SyncError");
            assert!(
                matches!(sync_err, SyncError::CircuitBreakerOpen),
                "pre-timeout call must be rejected by the breaker; \
                 expected SyncError::CircuitBreakerOpen, got: {sync_err:?}"
            );

            // Sleep past open_timeout.
            tokio::time::sleep(Duration::from_millis(300)).await;

            // Now: initiate_sync must NOT be rejected by the breaker.
            // The downstream NoOpTransport will fail at get_or_connect
            // (or the store will fail at "Document not found") — either
            // is fine. The only invariant under test is that the gate
            // wired into `initiate_sync` respects open_timeout, i.e.
            // is `should_block_sync` and not `is_circuit_open`.
            //
            // The post-timeout error either (a) is not a typed
            // SyncError at all (anyhow wrapping a generic transport /
            // store error), or (b) downcasts to a different SyncError
            // variant — but NEVER CircuitBreakerOpen.
            let err = coordinator
                .initiate_sync("test:doc", peer_id)
                .await
                .expect_err("NoOpTransport / empty store will fail downstream");
            if let Some(sync_err) = err.downcast_ref::<SyncError>() {
                assert!(
                    !matches!(sync_err, SyncError::CircuitBreakerOpen),
                    "post-timeout call must NOT be rejected by the breaker; \
                     got SyncError::CircuitBreakerOpen"
                );
            }
            // If downcast yields None, the error came from a layer
            // below the breaker (transport / store) — that's exactly
            // what we want, the breaker stayed out of the way.
        }

        /// peat-mesh#131 — peer-level circuit-breaker gate on
        /// `sync_all_documents_with_peer`.
        ///
        /// Pre-fix: the function unconditionally scanned the store
        /// and iterated all N documents, calling `initiate_sync` per
        /// doc. Each per-doc call bailed at the gate at line 800,
        /// but the outer function still logged "Syncing N existing
        /// documents" + N per-doc "Failed to sync document X"
        /// warnings per mDNS rediscovery (~10s on Android), feeding
        /// the OOM chain tracked in peat#873.
        ///
        /// Post-fix: the function checks `should_block_sync` once at
        /// entry and returns `Ok(())` without scanning the store when
        /// the breaker is open for the target peer.
        ///
        /// This test pins the behavior; the structural pin (the gate
        /// is wired) is the
        /// `production_call_sites_use_should_block_sync_not_is_circuit_open`
        /// QA test below.
        #[tokio::test]
        async fn sync_all_documents_with_peer_short_circuits_when_breaker_open() {
            let error_handler = Arc::new(SyncErrorHandler::with_policies(
                RetryPolicy::default(),
                CircuitBreakerConfig {
                    failure_threshold: 2,
                    // Long enough to stay open across the call under test.
                    open_timeout: Duration::from_secs(60),
                    ..Default::default()
                },
            ));
            let coordinator = make_coordinator(error_handler.clone());

            let peer_id = SecretKey::generate().public();

            // Sentinel doc in the store so a non-gated iteration would
            // have something to attempt (and would log "Syncing 1
            // existing documents"). With the gate, the store is never
            // scanned.
            use automerge::transaction::Transactable;
            let mut doc = automerge::Automerge::new();
            doc.transact::<_, _, automerge::AutomergeError>(|tx| {
                tx.put(automerge::ROOT, "k", "v")?;
                Ok(())
            })
            .unwrap();
            coordinator.store.put("sentinel:doc", &doc).unwrap();

            // Trip the breaker (failure_threshold = 2).
            let _ = error_handler.handle_error(&peer_id, SyncError::Network("seed".to_string()));
            let _ = error_handler.handle_error(&peer_id, SyncError::Network("seed".to_string()));
            assert!(
                error_handler.is_circuit_open(&peer_id),
                "precondition: breaker must be Open after threshold failures"
            );

            // Baseline: no scan_prefix calls have happened yet.
            assert_eq!(
                coordinator.store.scan_prefix_call_count(),
                0,
                "precondition: scan_prefix should not have been called yet"
            );

            // Call sync_all_documents_with_peer with the breaker open.
            // The contract: returns Ok without doing per-doc work.
            let result = coordinator.sync_all_documents_with_peer(peer_id).await;
            assert!(
                result.is_ok(),
                "sync_all_documents_with_peer must return Ok when breaker is open; got {result:?}"
            );

            // Load-bearing assertion (peat-mesh#131): the gated path must
            // not scan the store. Without the peer-level gate, the
            // function would call `store.scan_prefix("")` before iterating
            // the per-doc gate. With the gate, scan_prefix is never
            // invoked. This counter (cfg(test) on AutomergeStore) is the
            // only differentiating signal between the peer-gated and
            // doc-gated paths — both return Ok, both leave the breaker
            // open, both leave peer_stats untouched. The QA review
            // (PR #132) called this out specifically: assert "no scan
            // happened" makes the test self-sufficient instead of
            // co-dependent on the structural pin below.
            assert_eq!(
                coordinator.store.scan_prefix_call_count(),
                0,
                "sync_all_documents_with_peer with breaker open must not scan the store; \
                 scan_prefix was called {} time(s) — peer-level gate is missing",
                coordinator.store.scan_prefix_call_count()
            );

            // Sanity: the gate reads the breaker state but does not
            // mutate it. The breaker remains open after the call.
            assert!(
                error_handler.is_circuit_open(&peer_id),
                "breaker must remain open after a gated call"
            );
        }

        /// Symmetric coverage for the second production call site:
        /// `sync_documents_batch` at `automerge_sync.rs:1869` had the
        /// same `is_circuit_open` bug. We can't easily drive that
        /// async path through a public method without spinning up a
        /// transport, but we can pin the gate choice with a static
        /// invariant.
        ///
        /// The check has two halves:
        /// - **Positive**: each production function body must contain
        ///   a call to `should_block_sync`. If a future refactor drops
        ///   the gate entirely, this fails.
        /// - **Negative**: neither body may call
        ///   `error_handler.is_circuit_open` as its gate. If a future
        ///   refactor reverts to the buggy gate, this fails.
        ///
        /// (Earlier versions of this test counted total `should_block_sync`
        /// occurrences in the whole file with a `>= 2` floor — that was
        /// tautologically true because the test's own source counts
        /// toward the total. Per-window counting is the honest version.)
        #[test]
        fn production_call_sites_use_should_block_sync_not_is_circuit_open() {
            let source = include_str!("automerge_sync.rs");
            for fn_name in &[
                "initiate_sync",
                "sync_documents_batch",
                // peat-mesh#131: the multi-doc per-peer entry that the
                // mDNS-rediscovery path uses. Without a peer-level gate
                // here, every rediscovery would log N "Syncing document
                // X" + bail-warn cycles even though every per-doc call
                // would bail at the per-doc gate at line 800. Pinning
                // the gate here so a future refactor that removes it
                // re-opens the rediscovery loop bug.
                "sync_all_documents_with_peer",
            ] {
                // Anchor the locator on the `(` after the name so
                // prefix collisions like `sync_documents_batch` vs.
                // `sync_documents_batch_with_all_peers` don't match.
                let needle = format!("pub async fn {fn_name}(");
                // Exactly-one-definition check: if either function
                // gets duplicated (trait helper, mock impl, doc
                // example), `source.find` would silently window the
                // wrong site. Fail loudly so the test author can
                // refine the locator instead of asserting on the
                // wrong slice.
                let def_count = source.matches(&needle).count();
                assert_eq!(
                    def_count, 1,
                    "expected exactly one `pub async fn {fn_name}(` definition, found {def_count} — \
                     refine the locator before trusting the windowed slice"
                );
                let fn_idx = source.find(&needle).unwrap();
                // Look at the first ~40 lines of the function body for
                // the gate call — the gate is always near the top.
                let snippet = &source[fn_idx..fn_idx.saturating_add(2000)];

                // Positive: gate must exist.
                assert!(
                    snippet.contains("should_block_sync"),
                    "{fn_name} must call should_block_sync as the sync gate \
                     (missing gate re-opens the unguarded-sync regression)"
                );
                // Negative: gate must not be the buggy one. Match on
                // the call form `is_circuit_open(` so a refactor like
                // `let eh = &self.error_handler; eh.is_circuit_open(`
                // can't slip through. No legitimate reference to
                // `is_circuit_open(` exists inside these gate windows.
                assert!(
                    !snippet.contains("is_circuit_open("),
                    "{fn_name} must not call is_circuit_open as the sync gate \
                     (re-introduces the peat#850 one-way trap)"
                );
            }
        }

        /// Structural pin for the headline fix in this PR (#136): when
        /// `SyncDoc::generate_sync_message` returns `None`,
        /// `initiate_delta_sync` must treat it as the Automerge
        /// sync-protocol terminal "already in sync" — i.e. `Ok(())` —
        /// not as a `"No initial sync message to send"` error.
        ///
        /// QA Review on commit 43ae369 noted: a revert of `Ok(())` back
        /// to `Err(...)` here would compile cleanly, all existing 1575
        /// tests would still pass, and the regression would only be
        /// visible on a converged-peer field deployment. This pin
        /// closes that gap at compile-test time by inspecting the
        /// function source directly (same shape as
        /// `production_call_sites_use_should_block_sync_not_is_circuit_open`
        /// above).
        ///
        /// Downstream consequences also covered transitively:
        /// - `initiate_sync` records success via
        ///   `error_handler.record_success` for converged docs, instead
        ///   of feeding the breaker via `handle_error`.
        /// - `sync_all_documents_with_peer` stops emitting "Failed to
        ///   sync document X" noise for already-converged docs.
        #[test]
        fn initiate_delta_sync_treats_no_sync_message_as_success() {
            let source = include_str!("automerge_sync.rs");
            // `initiate_delta_sync` is a private async fn (not `pub`),
            // so the locator drops the visibility modifier. Build the
            // needle via `format!` rather than a literal — `include_str!`
            // pulls the test source into `source`, so a literal locator
            // would match itself and break the exactly-one-definition
            // check (same workaround as the line-4307 pin above).
            let fn_name = "initiate_delta_sync";
            let needle = format!("async fn {fn_name}(");
            let def_count = source.matches(&needle).count();
            assert_eq!(
                def_count, 1,
                "expected exactly one `async fn {fn_name}(` definition, found {def_count} — \
                 refine the locator before trusting the windowed slice"
            );
            let fn_idx = source.find(&needle).unwrap();
            // Function body is ~70 lines; 4 KB window comfortably
            // covers the `None =>` branch + the trailing `Ok(())` that
            // is the actual subject of this pin.
            let snippet = &source[fn_idx..fn_idx.saturating_add(4000)];

            // Positive: the new "already in sync" debug log line must
            // be present in the body. If a refactor removes the
            // explicit None handling, this fails.
            assert!(
                snippet.contains("already in sync"),
                "initiate_delta_sync must log \"already in sync\" on the None branch \
                 (missing log indicates the None-is-success handling was removed)"
            );
            // Negative: the old anyhow! error message must not appear
            // anywhere in the function body. Asserting the literal
            // catches both the direct revert and any near-duplicate
            // copy/paste.
            assert!(
                !snippet.contains("No initial sync message to send"),
                "initiate_delta_sync must not return Err(\"No initial sync message to send\") \
                 — that classification feeds the circuit breaker against converged peers (#136)"
            );
        }

        /// Structural pin for the `scan_prefix → keys_with_prefix` swap
        /// in `sync_all_documents_with_peer` and `get_local_sync_items`
        /// (PR #136 + parent peat-mesh#873). Both call sites used to
        /// fan-out `Automerge::load` for every persisted document via
        /// `scan_prefix("")`, then discard the bodies — multi-GB peak
        /// allocation under sustained mDNS rediscovery on Android,
        /// per defenseunicorns/peat#873.
        ///
        /// The count() paths got `scan_prefix_call_count` runtime pins
        /// (commits 4296f8d, 518eec0). These two production call sites
        /// don't have analogous runtime coverage — a revert here would
        /// pass the existing peat-mesh#131 breaker-open test (it
        /// short-circuits before this line) and the count-pin tests
        /// (different code paths) without surfacing the OOM-class
        /// regression. Pin the swap statically.
        #[test]
        fn keys_only_scan_used_in_full_store_iteration_paths() {
            let source = include_str!("automerge_sync.rs");
            // (fn_name, fn_prefix). The prefix is `pub async fn` for the
            // public sync entry and the bare `fn` for the Negentropy
            // helper. Both built via `format!` to avoid the literal-
            // matches-itself trap that include_str! sets (see comment
            // on the sibling pin above).
            for (fn_name, fn_prefix) in &[
                ("sync_all_documents_with_peer", "pub async fn"),
                ("get_local_sync_items", "fn"),
            ] {
                let needle = format!("{fn_prefix} {fn_name}(");
                let def_count = source.matches(&needle).count();
                assert_eq!(
                    def_count, 1,
                    "expected exactly one `{fn_prefix} {fn_name}(` definition, found {def_count} — \
                     refine the locator before trusting the windowed slice"
                );
                let fn_idx = source.find(&needle).unwrap();
                // sync_all_documents_with_peer has a long block comment
                // explaining the peat#873 allocation chain before the
                // `keys_with_prefix(\"\")` call, so the window has to
                // reach past it. 4 KB covers both call sites without
                // straying into adjacent fns.
                let snippet = &source[fn_idx..fn_idx.saturating_add(4000)];

                // Positive: must call keys_with_prefix on the store.
                assert!(
                    snippet.contains("keys_with_prefix("),
                    "{fn_name} must use keys_with_prefix for the full-store iteration \
                     (missing call re-opens the peat#873 multi-GB allocation pattern)"
                );
                // Negative: must not call scan_prefix on the store.
                // Some legitimate doc comments in the file reference
                // `scan_prefix` by name, but the windowed snippet only
                // covers the call site, where any `scan_prefix(` token
                // would be the buggy revert form.
                assert!(
                    !snippet.contains("scan_prefix("),
                    "{fn_name} must not use scan_prefix for the full-store iteration \
                     (re-introduces the peat#873 Vec<(String, Automerge)> fan-out)"
                );
            }
        }

        /// Mirror of `NoOpTransport` but reports a fixed set of connected
        /// peers, so `sync_document_with_all_peers` actually drives
        /// per-peer logic instead of short-circuiting on an empty list.
        /// `get_connection` still returns `None`, so each per-peer call
        /// fails downstream of the flow-control gate — we only want to
        /// observe timing here, not push real bytes.
        struct TwoPeerTransport {
            peers: Vec<EndpointId>,
        }

        #[async_trait::async_trait]
        impl SyncTransport for TwoPeerTransport {
            fn get_connection(&self, _peer_id: &EndpointId) -> Option<Connection> {
                None
            }
            fn connected_peers(&self) -> Vec<EndpointId> {
                self.peers.clone()
            }
        }

        fn make_coordinator_with_transport(
            transport: Arc<dyn SyncTransport>,
        ) -> AutomergeSyncCoordinator {
            AutomergeSyncCoordinator {
                store: Arc::new(AutomergeStore::in_memory()),
                transport,
                peer_states: Arc::new(RwLock::new(HashMap::new())),
                peer_stats: Arc::new(RwLock::new(HashMap::new())),
                total_bytes_sent: Arc::new(AtomicU64::new(0)),
                total_bytes_received: Arc::new(AtomicU64::new(0)),
                error_handler: Arc::new(SyncErrorHandler::new()),
                partition_detector: Arc::new(PartitionDetector::new()),
                flow_controller: Arc::new(
                    FlowController::with_config(FlowControlConfig::default()),
                ),
                sync_mode_registry: Arc::new(SyncModeRegistry::with_defaults()),
                next_batch_id: Arc::new(AtomicU64::new(1)),
                sync_router: None,
                channel_manager: Arc::new(RwLock::new(None)),
                negentropy_sync: Arc::new(NegentropySync::new()),
                ttl_manager: Arc::new(RwLock::new(None)),
                bandwidth_allocation: Arc::new(RwLock::new(None)),
                lamport_clock: Arc::new(RwLock::new(None)),
                fanout: Arc::new(FanoutState {
                    inner: Mutex::new(FanoutQueue::default()),
                    notify: Notify::new(),
                }),
            }
        }

        /// Regression for the defer-on-cooldown fix (peat#864 root cause).
        ///
        /// Pre-fix `initiate_sync` returned `CooldownActive` whenever the
        /// 100ms cooldown was active; the auto-sync push task silently
        /// discarded it and the change was lost from the outbound
        /// pipeline. Post-fix, `initiate_sync` waits the remaining
        /// cooldown out before calling `check_sync_allowed` so the call
        /// resolves cleanly (or hits a different, real error downstream)
        /// instead of dropping the request.
        ///
        /// We can't drive a full sync through `NoOpTransport` here (it
        /// returns `None` from `get_connection`), but the cooldown gate
        /// fires *before* any transport / store work. So:
        ///
        /// 1. Mark `(peer, doc)` as freshly synced via `record_sync`.
        /// 2. Call `initiate_sync` for that exact pair within the 100ms
        ///    cooldown window.
        /// 3. Time the call: it must take at least ~`sync_cooldown` -
        ///    epsilon to return, proving it waited (and didn't fast-fail).
        /// 4. Inspect the error: it must NOT be `CooldownActive`. Any
        ///    downstream error (transport, store, etc.) is fine — those
        ///    fire *after* the cooldown gate, so seeing one means the
        ///    cooldown stopped blocking and execution proceeded.
        ///
        /// A future refactor that drops the `tokio::time::sleep` from
        /// `initiate_sync` and reverts to fast-fail would fail this test
        /// on both the timing and the error-shape assertions.
        #[tokio::test]
        async fn initiate_sync_defers_instead_of_dropping_on_cooldown() {
            let error_handler = Arc::new(SyncErrorHandler::new());
            let coordinator = make_coordinator(error_handler);
            let peer_id = SecretKey::generate().public();
            let doc_key = "test:defer-on-cooldown";

            // A present doc so the post-defer call fails at the *transport*
            // (NoOpTransport), not at the absent-doc skip which now returns Ok
            // (peat-mesh#243). We only assert the failure isn't CooldownActive.
            {
                use automerge::transaction::Transactable as _;
                let mut doc = automerge::Automerge::new();
                let mut tx = doc.transaction();
                tx.put(automerge::ROOT, "x", "1").unwrap();
                tx.commit();
                coordinator.store.put(doc_key, &doc).unwrap();
            }

            // Default sync_cooldown is 100ms. Stamp a recent sync so the
            // next call hits the cooldown branch.
            coordinator.flow_controller.record_sync(&peer_id, doc_key);

            let start = std::time::Instant::now();
            let result = coordinator.initiate_sync(doc_key, peer_id).await;
            let elapsed = start.elapsed();

            // Timing: must have waited the cooldown. We allow a small
            // epsilon below 100ms for the recorded-sync timestamp having
            // moved a few microseconds before we hit `record_sync` — the
            // wait is `cooldown - elapsed_since_record` so it'll be a
            // hair under 100ms on a fast machine.
            assert!(
                elapsed >= std::time::Duration::from_millis(80),
                "initiate_sync within cooldown must defer (≥80ms wait); \
                 got {elapsed:?} — likely fast-failed with CooldownActive"
            );

            // Error shape: NoOpTransport will fail at the send step
            // (`send_prepared_to_peer`). That's fine — we only require that
            // the failure is *not* the cooldown-dropped variant.
            let err = result.expect_err(
                "NoOpTransport / empty-store path always fails downstream after the wait",
            );
            if let Some(sync_err) = err.downcast_ref::<SyncError>() {
                let msg = sync_err.to_string();
                assert!(
                    !msg.contains("Sync cooldown active"),
                    "post-defer error must not be CooldownActive; got: {msg}"
                );
            }
            let anyhow_msg = err.to_string();
            assert!(
                !anyhow_msg.contains("Sync cooldown active"),
                "post-defer error must not surface CooldownActive (any layer); got: {anyhow_msg}"
            );
        }

        /// Regression for the concurrent-peer-sync fix in
        /// `sync_document_with_all_peers`. Pre-fix the loop awaited each
        /// peer serially with `.await`, so two peers each in cooldown
        /// stalled the auto-sync task for `2 * sync_cooldown`. Post-fix
        /// it runs them concurrently via `futures::join_all`, so the
        /// two waits overlap and the call returns after roughly one
        /// cooldown.
        ///
        /// We stamp `record_sync` for both peers right before the call,
        /// then time the elapsed. Allowed window: `[80ms, 180ms]` —
        /// floor matches the single-peer test (timing jitter between
        /// `record_sync` and the call); ceiling rules out the serial
        /// path (which would land near 200 ms) while leaving slack for
        /// task scheduling on a loaded runner.
        ///
        /// A future revert to the serial loop fails the ceiling
        /// assertion; an accidental introduction of `Arc<Self>::clone`
        /// + `tokio::spawn` without joining would still allow the call
        /// to return before peers complete and fails differently —
        /// either way the regression surfaces here.
        #[tokio::test]
        async fn sync_document_with_all_peers_runs_peer_cooldowns_concurrently() {
            let peer_a = SecretKey::generate().public();
            let peer_b = SecretKey::generate().public();
            let transport = Arc::new(TwoPeerTransport {
                peers: vec![peer_a, peer_b],
            });
            let coordinator = make_coordinator_with_transport(transport);
            let doc_key = "test:concurrent-peer-cooldowns";

            // Stamp recent syncs for both peers. The next call will hit
            // the cooldown branch for each.
            coordinator.flow_controller.record_sync(&peer_a, doc_key);
            coordinator.flow_controller.record_sync(&peer_b, doc_key);

            let start = std::time::Instant::now();
            // sync_document_with_all_peers swallows per-peer errors, so
            // we ignore the (always-Ok) return — only timing is under test.
            let _ = coordinator.sync_document_with_all_peers(doc_key).await;
            let elapsed = start.elapsed();

            assert!(
                elapsed >= std::time::Duration::from_millis(80),
                "expected ≥80ms (one cooldown wait), got {elapsed:?}"
            );
            assert!(
                elapsed < std::time::Duration::from_millis(180),
                "expected <180ms (concurrent peer waits); got {elapsed:?} — \
                 likely reverted to serial `.await` which would be ~200ms"
            );
        }

        /// Pin the load-bearing behavior of the bounded retry loop: a
        /// concurrent `record_sync` mid-sleep must cause the loop to
        /// iterate again rather than fall through with a still-armed
        /// cooldown.
        ///
        /// We stamp `record_sync` once, then spawn a task that stamps it
        /// again ~30 ms in (well before the first 100 ms sleep
        /// completes). The expected behavior:
        ///
        /// 1. Iteration 1 reads `remaining_cooldown ≈ 100ms`, sleeps it.
        /// 2. The concurrent task re-stamps `record_sync` at ~30 ms.
        /// 3. Iteration 1 wakes at ~100 ms.
        /// 4. Iteration 2 reads `remaining_cooldown ≈ 30ms` (100 - 70),
        ///    sleeps it.
        /// 5. Iteration 2 wakes at ~130 ms total; cooldown is now clear.
        ///
        /// So the call should take ~130 ms, NOT ~100 ms (single iteration)
        /// and not ~200 ms (which would mean two full cooldowns — wrong).
        /// A future change to `MAX_COOLDOWN_DEFERRALS = 1` or removal of
        /// the loop fails the floor assertion (would land at ~100 ms with
        /// a still-armed cooldown after).
        #[tokio::test]
        async fn initiate_sync_loop_re_observes_concurrent_record_sync() {
            let error_handler = Arc::new(SyncErrorHandler::new());
            let coordinator = Arc::new(make_coordinator(error_handler));
            let peer_id = SecretKey::generate().public();
            let doc_key = "test:loop-re-observes-concurrent-record";

            coordinator.flow_controller.record_sync(&peer_id, doc_key);

            // Spawn a re-stamp at 30ms — well inside the first iteration's
            // sleep window. After the first iteration wakes at ~100ms,
            // the remaining cooldown will be ~30ms, triggering a second
            // iteration.
            let coordinator_clone = Arc::clone(&coordinator);
            let restamp = tokio::spawn(async move {
                tokio::time::sleep(std::time::Duration::from_millis(30)).await;
                coordinator_clone
                    .flow_controller
                    .record_sync(&peer_id, doc_key);
            });

            let start = std::time::Instant::now();
            let _ = coordinator.initiate_sync(doc_key, peer_id).await;
            let elapsed = start.elapsed();
            restamp.await.unwrap();

            assert!(
                elapsed >= std::time::Duration::from_millis(115),
                "expected ≥115ms (two iterations of defer loop after a \
                 mid-sleep re-stamp); got {elapsed:?} — likely \
                 MAX_COOLDOWN_DEFERRALS shrunk to 1 or the loop was \
                 removed entirely"
            );
            assert!(
                elapsed < std::time::Duration::from_millis(250),
                "expected <250ms (two short iterations, not five full \
                 cooldowns); got {elapsed:?}"
            );
        }
    }

    /// Regression coverage for peat-mesh#115 — receive-path notify loop.
    ///
    /// The original bug: `receive_sync_message` and `apply_state_snapshot`
    /// persisted received documents via `put_with_ttl`, which fires
    /// `AutomergeStore::change_tx`. That broadcast drives the outbound
    /// auto-sync pusher (`automerge_backend::start_sync`), so every received
    /// message caused the coordinator to re-broadcast the same doc back to
    /// the sender, forming a self-sustaining sync ping-pong that drained the
    /// per-peer flow-control bucket and starved first-sync of new documents.
    ///
    /// Fix: route receive-side writes through `put_received`, which uses
    /// `AutomergeStore::put_without_notify[_with_ttl]` (the Issue #346 escape
    /// hatch that was never wired into the delta-sync / state-snapshot paths).
    /// The observer channel (`observer_tx`) MUST still fire — hierarchical
    /// aggregation depends on seeing remotely-synced docs (Issue #377).
    ///
    /// These tests pin both invariants. They fail on the pre-fix code and
    /// pass post-fix.
    #[cfg(feature = "automerge-backend")]
    mod issue_115_receive_path_notify_tests {
        use super::*;
        use crate::storage::automerge_store::AutomergeStore;
        use automerge::sync::SyncDoc;
        use automerge::transaction::Transactable;
        use iroh::SecretKey;
        use tokio::sync::broadcast::error::TryRecvError;

        /// Minimal no-op transport: never returns a connection, so any send
        /// attempt fails after the put. That's fine — the put happens before
        /// the send, and the put-vs-notify invariant is what's under test.
        struct NoOpTransport;

        #[async_trait::async_trait]
        impl SyncTransport for NoOpTransport {
            fn get_connection(&self, _peer_id: &EndpointId) -> Option<Connection> {
                None
            }
            fn connected_peers(&self) -> Vec<EndpointId> {
                Vec::new()
            }
        }

        fn make_coordinator() -> AutomergeSyncCoordinator {
            AutomergeSyncCoordinator {
                store: Arc::new(AutomergeStore::in_memory()),
                transport: Arc::new(NoOpTransport),
                peer_states: Arc::new(RwLock::new(HashMap::new())),
                peer_stats: Arc::new(RwLock::new(HashMap::new())),
                total_bytes_sent: Arc::new(AtomicU64::new(0)),
                total_bytes_received: Arc::new(AtomicU64::new(0)),
                error_handler: Arc::new(SyncErrorHandler::default()),
                partition_detector: Arc::new(PartitionDetector::new()),
                flow_controller: Arc::new(
                    FlowController::with_config(FlowControlConfig::default()),
                ),
                sync_mode_registry: Arc::new(SyncModeRegistry::with_defaults()),
                next_batch_id: Arc::new(AtomicU64::new(1)),
                sync_router: None,
                channel_manager: Arc::new(RwLock::new(None)),
                negentropy_sync: Arc::new(NegentropySync::new()),
                ttl_manager: Arc::new(RwLock::new(None)),
                bandwidth_allocation: Arc::new(RwLock::new(None)),
                lamport_clock: Arc::new(RwLock::new(None)),
                fanout: Arc::new(FanoutState {
                    inner: Mutex::new(FanoutQueue::default()),
                    notify: Notify::new(),
                }),
            }
        }

        // ---- peat-mesh#247 QoS-priority fanout queue pins -------------------

        /// Coalescing: re-enqueuing a doc_key already pending is a no-op — one
        /// entry covers any number of intervening writes (the worker fans the
        /// current store state when it reaches it).
        #[test]
        fn enqueue_fanout_coalesces_duplicate_doc_keys() {
            let c = make_coordinator();
            c.enqueue_fanout("alerts:a");
            c.enqueue_fanout("alerts:a");
            c.enqueue_fanout("alerts:a");
            let q = c.fanout.inner.lock().unwrap();
            assert_eq!(
                q.pending.len(),
                1,
                "duplicate enqueues for the same doc_key must coalesce to one entry"
            );
            assert_eq!(q.index.len(), 1, "index must stay consistent with pending");
        }

        /// Priority ordering: the worker pops `pending.first()`, which must be
        /// the highest-QoS (lowest rank) entry regardless of enqueue order;
        /// the lowest-QoS entry sorts last (the eviction victim).
        #[test]
        fn enqueue_fanout_orders_highest_qos_first() {
            let c = make_coordinator();
            c.enqueue_fanout("sitreps:b"); // Bulk
            c.enqueue_fanout("alerts:a"); // Critical
            c.enqueue_fanout("beacons:n"); // Normal
            let q = c.fanout.inner.lock().unwrap();
            assert_eq!(
                q.pending.iter().next().unwrap().2,
                "alerts:a",
                "Critical-QoS doc must sort ahead of Normal/Bulk (worker pops it first)"
            );
            assert_eq!(
                q.pending.iter().next_back().unwrap().2,
                "sitreps:b",
                "Bulk-QoS doc must sort last (the eviction victim)"
            );
        }

        /// Bounded shed: at the bound, a higher-priority arrival is admitted by
        /// evicting the lowest-priority pending entry, and the eviction is
        /// counted on the victim's QoS-class slot (the ADR-0013 observability
        /// surface).
        #[test]
        fn enqueue_fanout_sheds_lowest_priority_when_full_and_counts_it() {
            use crate::qos::QoSClass;
            let c = make_coordinator();
            let max = AutomergeSyncCoordinator::MAX_PENDING_FANOUT;

            for i in 0..max {
                c.enqueue_fanout(&format!("sitreps:b{i}")); // Bulk
            }
            assert_eq!(c.fanout.inner.lock().unwrap().pending.len(), max);
            assert_eq!(
                c.fanout_dropped_by_class(),
                [0; 6],
                "no shedding below the bound"
            );

            // One Critical-QoS arrival over the bound: admitted, evicting a Bulk.
            c.enqueue_fanout("alerts:urgent");
            {
                let q = c.fanout.inner.lock().unwrap();
                assert_eq!(q.pending.len(), max, "queue stays at the bound");
                assert!(
                    q.index.contains_key("alerts:urgent"),
                    "the Critical-QoS arrival must be admitted, not dropped"
                );
            }
            let dropped = c.fanout_dropped_by_class();
            assert_eq!(
                dropped[QoSClass::Bulk as usize],
                1,
                "exactly one Bulk entry must be shed and counted on its class slot"
            );
            assert_eq!(
                dropped[QoSClass::Critical as usize],
                0,
                "a Critical arrival must never be the shed victim here"
            );
        }

        /// Like `make_coordinator()` but backed by an on-disk redb store.
        /// `AutomergeStore::in_memory()` makes `put_tombstone`/`has_tombstone`
        /// no-ops, so the send-side tombstone guard can only be exercised
        /// against a redb-backed store.
        fn make_coordinator_with_store(store: Arc<AutomergeStore>) -> AutomergeSyncCoordinator {
            let mut c = make_coordinator();
            c.store = store;
            c
        }

        /// peat-mesh#217: a tombstoned document must never be re-sent as a
        /// live snapshot. `store.get` still returns the pre-delete doc bytes
        /// (the tombstone is separate metadata), so without the send-side
        /// guard a reconnect's "sync existing documents" pass re-ships the
        /// deleted doc as a `StateSnapshot`, racing the tombstone and
        /// resurrecting it on the peer when the snapshot is applied before the
        /// tombstone is recorded. The guard lives in `prepare_doc_for_sync`
        /// (peat-mesh#247 hoisted it out of the per-peer path); pins: with an
        /// active tombstone, `prepare_doc_for_sync` returns `None` and the
        /// `initiate_sync` send path sends nothing (the LatestOnly
        /// `latest_only_count` stat stays 0). Fails on the pre-fix code (which
        /// would reach `send_state_snapshot`), passes post-fix.
        #[tokio::test]
        async fn initiate_sync_skips_tombstoned_document() {
            let tmp = tempfile::TempDir::new().unwrap();
            let store = Arc::new(AutomergeStore::open(tmp.path()).unwrap());

            // `capabilities` is a LatestOnly collection → state-snapshot path.
            let doc_key = "capabilities:cap-thermal";
            let mut doc = automerge::Automerge::new();
            {
                let mut tx = doc.transaction();
                tx.put(automerge::ROOT, "name", "thermal-sensor").unwrap();
                tx.commit();
            }
            store.put(doc_key, &doc).unwrap();

            // Tombstone it, mirroring a `peat delete`.
            let tombstone = crate::qos::Tombstone::new("cap-thermal", "capabilities", "node-a", 1);
            store.put_tombstone(&tombstone).unwrap();
            assert!(store.has_tombstone("capabilities", "cap-thermal").unwrap());

            let coordinator = make_coordinator_with_store(store);
            let peer = SecretKey::generate().public();

            // The send-side guard fires in the peer-independent prepare step.
            assert!(
                coordinator
                    .prepare_doc_for_sync(doc_key)
                    .expect("prepare must not error for a tombstoned doc")
                    .is_none(),
                "prepare_doc_for_sync must return None for a tombstoned doc (peat-mesh#217)"
            );

            // And end-to-end through the public send path: returns Ok, sends nothing.
            coordinator
                .initiate_sync(doc_key, peer)
                .await
                .expect("initiate_sync must return Ok for a tombstoned doc");

            let count = coordinator
                .peer_stats
                .read()
                .unwrap()
                .get(&peer)
                .map(|s| s.latest_only_count)
                .unwrap_or(0);
            assert_eq!(
                count, 0,
                "tombstoned doc must not be sent as a state snapshot (peat-mesh#217)"
            );
        }

        /// peat-mesh#243: an absent (deleted) document must not be counted as a
        /// *peer* failure. `store.get` returns None for a doc whose data is
        /// gone, and the pre-fix code turned that into `SyncError::Document`
        /// ("Document not found for sync") — Fatal/non-retryable — which
        /// incremented the peer circuit breaker. A burst of
        /// >= `failure_threshold` (default 5) deletes then opened the breaker
        /// and blocked ALL subsequent syncs to that peer, silently dropping
        /// unrelated healthy writes for `open_timeout` (default 5s). Observed
        /// as: bulk-delete then write -> the write never reaches the direct
        /// peer (c2-alpha -> c2-bravo), though it reaches others.
        ///
        /// Pins: syncing absent docs returns Ok (clean skip) and never opens
        /// the breaker. Fails on the pre-fix code (the first absent-doc sync
        /// returns Err, and by the 6th the breaker is open), passes post-fix.
        #[tokio::test]
        async fn absent_document_sync_does_not_trip_circuit_breaker() {
            let tmp = tempfile::TempDir::new().unwrap();
            let store = Arc::new(AutomergeStore::open(tmp.path()).unwrap());
            let coordinator = make_coordinator_with_store(store);
            let peer = SecretKey::generate().public();

            // A burst of 6 (> the default failure_threshold of 5) syncs for
            // documents absent from the store (store.get -> None), exactly what
            // a burst of deletes produces on the change-driven push path.
            for i in 0..6 {
                let doc_key = format!("sitrep:ghost-{i}");
                coordinator.initiate_sync(&doc_key, peer).await.expect(
                    "syncing an absent document must return Ok, not a Document \
                     error that feeds the peer circuit breaker (peat-mesh#243)",
                );
            }

            assert!(
                !coordinator.error_handler.should_block_sync(&peer),
                "absent-document syncs must not open the peer circuit breaker \
                 (peat-mesh#243) — a local document-state condition must never \
                 block healthy traffic to a peer"
            );
        }

        /// Build a synthetic `SyncMessage` from a fresh remote doc with one
        /// field. A fresh sync state against any non-empty doc always
        /// generates an initial sync message, so the `expect` is reliable.
        fn remote_initial_sync_message() -> (automerge::sync::Message, usize) {
            let mut remote_doc = automerge::Automerge::new();
            let mut tx = remote_doc.transaction();
            tx.put(automerge::ROOT, "field", "value")
                .expect("put on root must succeed");
            tx.commit();
            let mut remote_state = automerge::sync::State::new();
            let msg = SyncDoc::generate_sync_message(&remote_doc, &mut remote_state)
                .expect("fresh remote doc must produce an initial sync message");
            let size = msg.clone().encode().len();
            (msg, size)
        }

        /// peat-mesh#239 — gossip-amplification storm seed. A CRDT doc fully
        /// converged with a peer must be QUIESCENT: re-generating a sync message
        /// yields `None`. The receive path's Issue-#435 convergence reset
        /// (replace the `SyncState` with `SyncState::new()` keeping only
        /// `shared_heads`) must preserve that quiescence. If a post-reset
        /// converged doc still emits a sync message, every gossip re-sync of an
        /// already-converged doc sends a no-op delta → the peer re-applies
        /// (no change) → re-fires `gossip_tx` → re-broadcasts → unbounded
        /// amplification storm (observed ~285 initiate/s on one unchanged doc)
        /// that starves new writes after a couple transactions.
        ///
        /// Pure-logic guard (no network/gossip pusher needed): drives the
        /// automerge sync protocol to convergence, then checks both the
        /// pre-#239 full reset (the bug — re-negotiates) and the #239
        /// `prune_converged_sync_state` (the fix — stays quiescent).
        #[test]
        fn converged_resync_is_quiescent_after_issue_435_reset() {
            use automerge::sync::{State as PlainSyncState, SyncDoc as PlainSyncDoc};
            use automerge::transaction::Transactable;
            use automerge::Automerge;

            // peer A has a one-key doc; peer B starts empty.
            let mut doc_a = Automerge::new();
            {
                let mut tx = doc_a.transaction();
                tx.put(automerge::ROOT, "k", "v").unwrap();
                tx.commit();
            }
            let mut doc_b = Automerge::new();
            let mut state_a = PlainSyncState::new();
            let mut state_b = PlainSyncState::new();

            // Drive the automerge sync protocol A<->B to convergence.
            for _ in 0..20 {
                let a_msg = PlainSyncDoc::generate_sync_message(&doc_a, &mut state_a);
                if let Some(m) = a_msg.clone() {
                    PlainSyncDoc::receive_sync_message(&mut doc_b, &mut state_b, m).unwrap();
                }
                let b_msg = PlainSyncDoc::generate_sync_message(&doc_b, &mut state_b);
                if let Some(m) = b_msg.clone() {
                    PlainSyncDoc::receive_sync_message(&mut doc_a, &mut state_a, m).unwrap();
                }
                if a_msg.is_none() && b_msg.is_none() {
                    break;
                }
            }
            assert_eq!(doc_a.get_heads(), doc_b.get_heads(), "docs must converge");
            assert!(
                PlainSyncDoc::generate_sync_message(&doc_a, &mut state_a).is_none(),
                "converged state should be quiescent before the reset"
            );

            // (a) BUG documentation — the pre-#239 full reset
            //     (`SyncState::new()` keeping only `shared_heads`) is NOT
            //     quiescent: a converged doc re-emits a sync message, which is
            //     the gossip-amplification storm seed.
            let mut full_reset = PlainSyncState::new();
            full_reset.shared_heads = state_a.shared_heads.clone();
            assert!(
                PlainSyncDoc::generate_sync_message(&doc_a, &mut full_reset).is_some(),
                "pre-#239 full reset should re-negotiate (Some) — this is the bug being fixed"
            );

            // (b) FIX guard — the production prune (peat-mesh#239) retains the
            //     converged state and clears only `sent_hashes`, so a converged
            //     doc stays quiescent (`generate_sync_message` → `None`) and
            //     does not feed the gossip storm.
            let mut pruned = state_a.clone();
            super::super::AutomergeSyncCoordinator::prune_converged_sync_state(&mut pruned);
            assert!(
                PlainSyncDoc::generate_sync_message(&doc_a, &mut pruned).is_none(),
                "peat-mesh#239: converged state after prune_converged_sync_state must be \
                 quiescent (None) — else gossip re-syncs of converged docs amplify into a storm"
            );

            // The prune must actually bound memory: sent_hashes cleared.
            assert!(
                pruned.sent_hashes.is_empty(),
                "prune must clear sent_hashes (Issue #435)"
            );
        }

        /// THE peat-mesh#115 invariant, extended for peat#891 / peat#907:
        ///
        /// - `change_tx` (legacy local-only pusher) must NOT fire on receive.
        ///   That's the original ping-pong protection: the existing
        ///   sync-coordinator outbound pusher would otherwise re-broadcast
        ///   the just-received doc back to the sender.
        /// - `observer_tx` (untagged, fires on all changes) STILL fires.
        ///   Hierarchical aggregation (Issue #377) depends on this.
        /// - `gossip_tx` (origin-tagged) DOES fire, with
        ///   `ChangeOrigin::Remote(peer_id.to_string())`. The new gossip
        ///   pusher subscribes here and skips the source peer — same #115
        ///   protection, just lifted from "suppress" to "attribute + filter".
        #[tokio::test]
        async fn receive_sync_message_change_tx_silent_gossip_tx_attributed() {
            let coordinator = make_coordinator();
            let peer_id = SecretKey::generate().public();

            let mut change_rx = coordinator.store.subscribe_to_changes();
            let mut observer_rx = coordinator.store.subscribe_to_observer_changes();
            let mut gossip_rx = coordinator.store.subscribe_to_changes_with_origin();

            let (msg, size) = remote_initial_sync_message();
            // NoOpTransport will fail the response-send, but the put already
            // happened — that's the window the test inspects.
            let _ = coordinator
                .receive_sync_message("test:doc", peer_id, msg, size)
                .await;

            // INVARIANT 1: legacy change_tx silent (peat-mesh#115 ping-pong protection).
            match change_rx.try_recv() {
                Err(TryRecvError::Empty) => {}
                Ok(key) => panic!(
                    "receive_sync_message fired change_tx with {key:?} — would re-broadcast \
                     received doc to sender via the legacy pusher (peat-mesh#115)"
                ),
                Err(e) => panic!("unexpected change_rx state: {e:?}"),
            }

            // INVARIANT 2: observer_tx fires (Issue #377 — hierarchical aggregation).
            assert!(
                observer_rx.try_recv().is_ok(),
                "observer_tx must fire for sync-received docs (Issue #377)"
            );

            // INVARIANT 3: gossip_tx fires with the source peer attributed.
            // The new transitive-gossip pusher (peat#891 / peat#907) reads
            // this and pushes to every peer EXCEPT `peer_id`.
            let evt = gossip_rx
                .try_recv()
                .expect("gossip_tx must fire for sync-received docs (peat#891)");
            assert_eq!(evt.key, "test:doc");
            match evt.origin {
                ChangeOrigin::Remote(ref s) => assert_eq!(s, &peer_id.to_string()),
                ChangeOrigin::Local => panic!(
                    "gossip_tx fired with ChangeOrigin::Local for a sync-received doc — \
                     attribution is required to suppress the echo back to the source peer"
                ),
            }
        }

        /// peat-mesh#116 QA, extended for peat#891: exercise the
        /// TTL-applying branch of `put_received` end-to-end. We wire a
        /// real `TtlManager` with `TtlConfig::tactical()` (which has a
        /// TTL for the `beacons` collection) and assert ALL three
        /// invariants from `receive_sync_message_change_tx_silent_gossip_tx_attributed`
        /// still hold through the TTL-applying code path:
        /// - the TTL is registered for the synced doc,
        /// - `change_tx` still does not fire (the #115 invariant), AND
        /// - `gossip_tx` STILL fires with `Remote(peer_id)` attribution.
        #[tokio::test]
        async fn receive_sync_message_with_ttl_registers_ttl_and_skips_notify() {
            use crate::storage::ttl::TtlConfig;
            use crate::storage::ttl_manager::TtlManager;

            let coordinator = make_coordinator();
            let ttl_manager = Arc::new(TtlManager::new(
                Arc::clone(&coordinator.store),
                TtlConfig::tactical(),
            ));
            coordinator.set_ttl_manager(Arc::clone(&ttl_manager));

            let peer_id = SecretKey::generate().public();

            let mut change_rx = coordinator.store.subscribe_to_changes();
            let mut gossip_rx = coordinator.store.subscribe_to_changes_with_origin();

            // `beacons` collection has a TTL configured in the tactical
            // preset (see ttl.rs), so apply_ttl will register one.
            let (msg, size) = remote_initial_sync_message();
            let _ = coordinator
                .receive_sync_message("beacons:node-1", peer_id, msg, size)
                .await;

            // Invariant 1: TTL registered → put_received's with-TTL branch
            // called apply_ttl successfully.
            assert_eq!(
                ttl_manager.pending_count(),
                1,
                "synced doc in TTL'd collection must register a pending expiry — \
                 put_received's TTL branch is not wired up"
            );

            // Invariant 2: change_tx still empty — the #115 fix holds
            // through the TTL-applying code path too.
            assert!(
                matches!(
                    change_rx.try_recv(),
                    Err(tokio::sync::broadcast::error::TryRecvError::Empty)
                ),
                "TTL branch of put_received must also skip change_tx"
            );

            // Invariant 3 (peat#891): gossip_tx STILL fires with
            // `Remote(peer_id)` even on the TTL-applying path.
            let evt = gossip_rx
                .try_recv()
                .expect("gossip_tx must fire on TTL'd sync-received docs too (peat#891)");
            assert_eq!(evt.key, "beacons:node-1");
            assert!(
                matches!(evt.origin, ChangeOrigin::Remote(ref s) if s == &peer_id.to_string()),
                "TTL branch must still attribute origin = Remote(peer_id)"
            );
        }

        /// Positive control: a local `put` MUST fire `change_tx`, otherwise
        /// the fix has over-rotated and broken sync-on-local-change.
        #[test]
        fn local_put_still_fires_change_tx() {
            let coordinator = make_coordinator();
            let mut change_rx = coordinator.store.subscribe_to_changes();

            let mut local_doc = automerge::Automerge::new();
            let mut tx = local_doc.transaction();
            tx.put(automerge::ROOT, "field", "value").unwrap();
            tx.commit();

            coordinator.store.put("local:doc", &local_doc).unwrap();

            match change_rx.try_recv() {
                Ok(key) => assert_eq!(
                    key, "local:doc",
                    "change_tx must broadcast the key of the locally-written doc"
                ),
                Err(e) => panic!(
                    "local put did NOT fire change_tx ({e:?}) — sync-on-change \
                     for local mutations is broken; the fix has over-rotated"
                ),
            }
        }

        /// peat#891: a local `put` must fire `gossip_tx` with
        /// `ChangeOrigin::Local`, so the gossip pusher pushes to all
        /// peers (not skipping any).
        #[test]
        fn local_put_fires_gossip_tx_with_local_origin() {
            let coordinator = make_coordinator();
            let mut gossip_rx = coordinator.store.subscribe_to_changes_with_origin();

            let mut local_doc = automerge::Automerge::new();
            let mut tx = local_doc.transaction();
            tx.put(automerge::ROOT, "field", "value").unwrap();
            tx.commit();

            coordinator.store.put("local:doc", &local_doc).unwrap();

            let evt = gossip_rx
                .try_recv()
                .expect("local put must fire gossip_tx (peat#891)");
            assert_eq!(evt.key, "local:doc");
            assert_eq!(
                evt.origin,
                ChangeOrigin::Local,
                "local put must carry ChangeOrigin::Local — otherwise the \
                 gossip pusher would skip a peer that should receive this write"
            );
        }

        /// peat-mesh#151 QA finding: direct surface-tier test of
        /// `put_with_origin(.., ChangeOrigin::Local)`. The existing
        /// `local_put_fires_gossip_tx_with_local_origin` test covers
        /// the Local-variant behavior indirectly through `store.put()`;
        /// this one calls `put_with_origin` directly to guard against
        /// future drift in the `matches!(origin, ChangeOrigin::Local)`
        /// branch inside `put_with_origin` that decides whether to fire
        /// `change_tx`. If that match ever silently widens to
        /// suppress Local too, only this direct test would catch it.
        #[test]
        fn put_with_origin_local_fires_change_tx_and_gossip_tx() {
            let coordinator = make_coordinator();
            let mut change_rx = coordinator.store.subscribe_to_changes();
            let mut gossip_rx = coordinator.store.subscribe_to_changes_with_origin();

            let mut doc = automerge::Automerge::new();
            let mut tx = doc.transaction();
            tx.put(automerge::ROOT, "field", "value").unwrap();
            tx.commit();

            coordinator
                .store
                .put_with_origin("origin-local:doc", &doc, ChangeOrigin::Local)
                .unwrap();

            // change_tx fires because origin is Local — the legacy
            // local-only pusher contract is preserved.
            match change_rx.try_recv() {
                Ok(key) => assert_eq!(key, "origin-local:doc"),
                Err(e) => panic!(
                    "put_with_origin(Local) did NOT fire change_tx ({e:?}) — \
                     the `matches!(origin, ChangeOrigin::Local)` branch in \
                     put_with_origin has drifted and broken sync-on-local-change"
                ),
            }

            // gossip_tx fires with Local attribution.
            let evt = gossip_rx
                .try_recv()
                .expect("put_with_origin(Local) must fire gossip_tx");
            assert_eq!(evt.key, "origin-local:doc");
            assert_eq!(evt.origin, ChangeOrigin::Local);
        }

        /// peat#891 invariant for `put_without_notify`: legacy callers that
        /// don't supply an origin still get the gossip channel attribution
        /// they were expecting — `ChangeOrigin::Local`. This keeps the
        /// existing semantic ("no sync-back trigger") for the legacy
        /// `change_tx` consumer while making the new gossip consumer's
        /// behavior identical to a regular local put. It's the safe default
        /// for callers that haven't been ported to `put_with_origin` yet.
        #[test]
        fn put_without_notify_still_fires_gossip_tx_as_local() {
            let coordinator = make_coordinator();
            let mut change_rx = coordinator.store.subscribe_to_changes();
            let mut gossip_rx = coordinator.store.subscribe_to_changes_with_origin();

            let mut doc = automerge::Automerge::new();
            let mut tx = doc.transaction();
            tx.put(automerge::ROOT, "field", "value").unwrap();
            tx.commit();

            coordinator
                .store
                .put_without_notify("legacy:doc", &doc)
                .unwrap();

            // change_tx silent (legacy contract)
            assert!(matches!(change_rx.try_recv(), Err(TryRecvError::Empty)));
            // gossip_tx fires with Local origin (peat#891 default for
            // callers that haven't supplied a peer attribution)
            let evt = gossip_rx
                .try_recv()
                .expect("put_without_notify must still fire gossip_tx");
            assert_eq!(evt.origin, ChangeOrigin::Local);
        }
    }

    /// Persistent sync stream wire-format and accept-side loop regression
    /// tests (peat-mesh#175 / ADR-063).
    ///
    /// Two stacked defects were observed under the lab telemetry sweep:
    ///
    /// 1. `receive_sync_payload_from_stream` consumed its `RecvStream` by
    ///    value, so the accept side (`handle_incoming_sync_stream`) could
    ///    only ever read one frame per stream — every subsequent frame on
    ///    a persistent bidirectional stream was discarded when the function
    ///    returned and the stream dropped.
    ///
    /// 2. `SyncChannel::receive_loop` read `[1 marker][4 len][N data]`
    ///    directly, but the sender writes the v2 framing
    ///    `[2 doc_key_len][N doc_key][1 msg_type][4 payload_len][payload]`.
    ///    The receiver desynchronized on the very first byte
    ///    (`0x00` high byte of `doc_key_len=5`) and warned-and-skipped it
    ///    as `Unexpected message type`.
    ///
    /// These tests pin the post-fix behavior end-to-end against a real
    /// iroh bi-stream so a future refactor cannot silently re-open the
    /// gap.
    #[cfg(feature = "automerge-backend")]
    mod persistent_stream_wire_format_peat_mesh_175 {
        use super::*;
        use automerge::transaction::Transactable;
        use iroh::address_lookup::memory::MemoryLookup;
        use iroh::endpoint::{Connection, Endpoint};
        use std::sync::atomic::AtomicU64;
        use std::sync::Arc;
        use std::time::Duration;

        const TEST_ALPN: &[u8] = b"test/persistent-sync/1";

        /// Serialise QUIC-port-binding tests in this module — UDP port
        /// allocation collides if multiple tests bind in parallel.
        static TEST_LOCK: tokio::sync::Mutex<()> = tokio::sync::Mutex::const_new(());

        /// Spin up two iroh endpoints, B → A connect over a test ALPN.
        /// Returns (endpoint_a, endpoint_b, conn_b_to_a, conn_a_from_b).
        ///
        /// Uses `Endpoint::accept()` directly on A's side (with the ALPN
        /// registered via `Builder::alpns`) instead of the router-with-
        /// handler pattern. The router pattern wraps the inbound
        /// connection in a tokio task owned by the router; if the
        /// handler returns, the connection drops. For tests that need
        /// to drive bi-streams on A's side, accepting raw via
        /// `endpoint_a.accept()` is the path that gives us a
        /// `Connection` we can hold across the test body.
        async fn connected_pair() -> (Endpoint, Endpoint, Connection, Connection) {
            let timeout = Duration::from_secs(15);

            let lookup_a = MemoryLookup::new();
            let endpoint_a = tokio::time::timeout(
                timeout,
                Endpoint::builder(iroh::endpoint::presets::Empty)
                    .crypto_provider(crate::security::tls_provider::iroh_quic_provider())
                    .address_lookup(lookup_a.clone())
                    .secret_key(iroh::SecretKey::from_bytes(&[42u8; 32]))
                    .alpns(vec![TEST_ALPN.to_vec()])
                    .bind(),
            )
            .await
            .expect("endpoint_a bind timed out")
            .unwrap();

            let lookup_b = MemoryLookup::new();
            let endpoint_b = tokio::time::timeout(
                timeout,
                Endpoint::builder(iroh::endpoint::presets::Empty)
                    .crypto_provider(crate::security::tls_provider::iroh_quic_provider())
                    .address_lookup(lookup_b.clone())
                    .secret_key(iroh::SecretKey::from_bytes(&[43u8; 32]))
                    .bind(),
            )
            .await
            .expect("endpoint_b bind timed out")
            .unwrap();

            lookup_b.add_endpoint_info(endpoint_a.addr());

            // Concurrent connect (B side) and accept (A side).
            let endpoint_a_clone = endpoint_a.clone();
            let accept_task = tokio::spawn(async move {
                let incoming = endpoint_a_clone
                    .accept()
                    .await
                    .expect("endpoint_a yielded no incoming");
                let accepting = incoming.accept().expect("accept() on Incoming");
                accepting.await.expect("accepting handshake")
            });

            let conn_b_to_a =
                tokio::time::timeout(timeout, endpoint_b.connect(endpoint_a.id(), TEST_ALPN))
                    .await
                    .expect("B → A connect timed out")
                    .unwrap();

            let conn_a_from_b = tokio::time::timeout(timeout, accept_task)
                .await
                .expect("A side accept timed out")
                .expect("accept task panicked");

            (endpoint_a, endpoint_b, conn_b_to_a, conn_a_from_b)
        }

        /// Encode a v2 frame: `[2 doc_key_len][N doc_key][1 msg_type][4 payload_len][payload]`.
        /// Used by every per-variant pin to construct the wire bytes for a
        /// single frame matching the layout in
        /// `AutomergeSyncCoordinator::receive_sync_payload_from_stream`.
        fn encode_v2_frame(doc_key: &[u8], msg_type: SyncMessageType, payload: &[u8]) -> Vec<u8> {
            let mut frame = Vec::with_capacity(2 + doc_key.len() + 1 + 4 + payload.len());
            frame.extend_from_slice(&(doc_key.len() as u16).to_be_bytes());
            frame.extend_from_slice(doc_key);
            frame.push(msg_type as u8);
            frame.extend_from_slice(&(payload.len() as u32).to_be_bytes());
            frame.extend_from_slice(payload);
            frame
        }

        /// Encode a v2 frame for a `SyncBatch` (doc_key="batch",
        /// msg_type=0x07), matching `SyncChannel::send`'s framing.
        fn encode_batch_frame(batch: &SyncBatch) -> Vec<u8> {
            encode_v2_frame(b"batch", SyncMessageType::SyncBatch, &batch.encode())
        }

        /// `SyncTransport` that returns a single pinned iroh `Connection`
        /// for a single pinned peer id. Lets dispatch arms that try to send
        /// a response (e.g. `DeltaSync` calling `send_sync_message_for_doc`)
        /// reach a real open_bi + write_all rather than failing with
        /// "no connection to peer" under NoOpTransport. The peer-side recv
        /// is unread — the write completes into iroh's send buffer and the
        /// stream is closed with `send.finish()` + `recv.stop(0)` per the
        /// fallback path's normal teardown.
        struct PinnedConnTransport {
            peer_id: EndpointId,
            conn: Connection,
        }
        impl std::fmt::Debug for PinnedConnTransport {
            fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
                f.debug_struct("PinnedConnTransport").finish()
            }
        }
        impl SyncTransport for PinnedConnTransport {
            fn get_connection(&self, peer_id: &EndpointId) -> Option<Connection> {
                if *peer_id == self.peer_id {
                    Some(self.conn.clone())
                } else {
                    None
                }
            }
            fn connected_peers(&self) -> Vec<EndpointId> {
                vec![self.peer_id]
            }
        }

        /// Build a coordinator wired to the given transport. Used to swap
        /// between `NoOpTransport` (parser-only tests) and
        /// `PinnedConnTransport` (round-trip tests where the dispatch arm
        /// needs to send a response).
        fn make_coordinator_with_transport(
            transport: Arc<dyn SyncTransport>,
        ) -> AutomergeSyncCoordinator {
            use crate::storage::sync_errors::SyncErrorHandler;
            use std::collections::HashMap;
            use std::sync::RwLock;

            AutomergeSyncCoordinator {
                store: Arc::new(AutomergeStore::in_memory()),
                transport,
                peer_states: Arc::new(RwLock::new(HashMap::new())),
                peer_stats: Arc::new(RwLock::new(HashMap::new())),
                total_bytes_sent: Arc::new(AtomicU64::new(0)),
                total_bytes_received: Arc::new(AtomicU64::new(0)),
                error_handler: Arc::new(SyncErrorHandler::default()),
                partition_detector: Arc::new(PartitionDetector::new()),
                flow_controller: Arc::new(
                    FlowController::with_config(FlowControlConfig::default()),
                ),
                sync_mode_registry: Arc::new(SyncModeRegistry::with_defaults()),
                next_batch_id: Arc::new(AtomicU64::new(1)),
                sync_router: None,
                channel_manager: Arc::new(RwLock::new(None)),
                negentropy_sync: Arc::new(NegentropySync::new()),
                ttl_manager: Arc::new(RwLock::new(None)),
                bandwidth_allocation: Arc::new(RwLock::new(None)),
                lamport_clock: Arc::new(RwLock::new(None)),
                fanout: Arc::new(FanoutState {
                    inner: Mutex::new(FanoutQueue::default()),
                    notify: Notify::new(),
                }),
            }
        }

        /// Minimal coordinator suitable for parser-only tests that never
        /// reach the response-send tail of any dispatch arm.
        fn make_minimal_coordinator() -> AutomergeSyncCoordinator {
            #[derive(Debug)]
            struct NoOpTransport;
            impl SyncTransport for NoOpTransport {
                fn get_connection(&self, _peer_id: &EndpointId) -> Option<Connection> {
                    None
                }
                fn connected_peers(&self) -> Vec<EndpointId> {
                    Vec::new()
                }
            }
            make_coordinator_with_transport(Arc::new(NoOpTransport))
        }

        /// Await an atomic counter reaching a target within a bounded
        /// budget. Replaces the prior 10s poll-and-assert loop with a
        /// `tokio::time::timeout`-bounded async poll — failure surfaces
        /// after `budget` rather than hanging until a hard test-runner
        /// timeout, and the observed value is still readable for the
        /// caller's assertion. Polls every 5ms with `tokio::time::sleep`
        /// so the spawned `receive_loop` task gets executor budget between
        /// loads.
        async fn await_counter_at_least(
            counter: &std::sync::atomic::AtomicU64,
            target: u64,
            budget: Duration,
        ) -> u64 {
            let result = tokio::time::timeout(budget, async {
                loop {
                    let cur = counter.load(std::sync::atomic::Ordering::Relaxed);
                    if cur >= target {
                        return cur;
                    }
                    tokio::time::sleep(Duration::from_millis(5)).await;
                }
            })
            .await;
            match result {
                Ok(cur) => cur,
                Err(_) => counter.load(std::sync::atomic::Ordering::Relaxed),
            }
        }

        /// Pin: `receive_sync_payload_from_stream` can be called repeatedly
        /// on the same `RecvStream` and parses consecutive v2 frames.
        ///
        /// Before peat-mesh#175 the function took the stream by value, so
        /// a caller could not loop over it — the persistent-channel
        /// scaffolding silently never wired this up correctly.
        ///
        /// Asserts: two frames decoded with correct (doc_key, payload_size,
        /// variant), and a third read returns `ReadExactError::FinishedEarly(0)`
        /// — the graceful end-of-stream signal `is_graceful_stream_end`
        /// relies on.
        #[tokio::test]
        async fn receive_sync_payload_from_stream_decodes_consecutive_v2_frames() {
            let _lock = TEST_LOCK.lock().await;
            let (endpoint_a, endpoint_b, conn_b_to_a, conn_a_from_b) = connected_pair().await;

            // B opens a bi-stream and writes two SyncBatch frames + finish.
            let (mut send_b, _recv_b) = conn_b_to_a.open_bi().await.expect("open_bi on B side");

            let batch_one = SyncBatch::new();
            let mut batch_two = SyncBatch::new();
            batch_two.entries.push(SyncEntry::new(
                "nodes:peat-mesh-175".to_string(),
                SyncMessageType::DeltaSync,
                vec![0xDE, 0xAD, 0xBE, 0xEF],
            ));

            send_b
                .write_all(&encode_batch_frame(&batch_one))
                .await
                .expect("write frame 1");
            send_b
                .write_all(&encode_batch_frame(&batch_two))
                .await
                .expect("write frame 2");
            send_b.finish().expect("finish send_b");

            // A accepts the bi-stream and reads frames with the helper.
            let (_send_a, mut recv_a) = conn_a_from_b
                .accept_bi()
                .await
                .expect("accept_bi on A side");

            let coordinator = make_minimal_coordinator();

            // Frame 1: empty SyncBatch.
            let (doc_key_one, payload_one, size_one) = coordinator
                .receive_sync_payload_from_stream(&mut recv_a)
                .await
                .expect("frame 1 must parse");
            assert_eq!(doc_key_one, "batch", "doc_key for SyncBatch is reserved");
            assert!(
                matches!(payload_one, ReceivedSyncPayload::Batch(ref b) if b.entries.is_empty()),
                "frame 1 must decode as an empty SyncBatch"
            );
            assert!(
                size_one > 0,
                "size accounting must be non-zero for a frame with header bytes"
            );

            // Frame 2: SyncBatch with one DeltaSync entry. Pre-fix the
            // accept side never reached this read because the helper
            // dropped its stream after frame 1.
            let (doc_key_two, payload_two, _size_two) = coordinator
                .receive_sync_payload_from_stream(&mut recv_a)
                .await
                .expect(
                    "frame 2 must parse — pre-#175, the helper consumed the stream after frame 1",
                );
            assert_eq!(doc_key_two, "batch");
            let batch_two_decoded = match payload_two {
                ReceivedSyncPayload::Batch(b) => b,
                other => panic!("frame 2 must decode as a SyncBatch, got {:?}", other),
            };
            assert_eq!(batch_two_decoded.entries.len(), 1);
            assert_eq!(batch_two_decoded.entries[0].doc_key, "nodes:peat-mesh-175");
            assert_eq!(
                batch_two_decoded.entries[0].payload,
                vec![0xDE, 0xAD, 0xBE, 0xEF]
            );

            // Frame 3: peer half-closed the send side, so the next read
            // surfaces ReadExactError::FinishedEarly(0). This is the
            // signal `is_graceful_stream_end` matches on to terminate
            // `handle_incoming_sync_stream`'s loop cleanly.
            let err = coordinator
                .receive_sync_payload_from_stream(&mut recv_a)
                .await
                .expect_err("third read must hit end-of-stream");
            assert!(
                is_graceful_stream_end(&err),
                "end-of-stream at frame boundary must classify as graceful: {err:?}",
            );

            endpoint_a.close().await;
            endpoint_b.close().await;
        }

        /// Pin: `handle_incoming_sync_stream` loops over frames on a single
        /// accepted bi-stream until the peer half-closes.
        ///
        /// Before peat-mesh#175 the function returned `Ok(())` after the
        /// first frame, so every subsequent frame on a persistent stream
        /// was silently dropped.
        ///
        /// Approach: B writes three empty SyncBatch frames + finish; A
        /// runs `handle_incoming_sync_stream`. After it returns, A's
        /// `total_bytes_received` reflects three frames' worth of
        /// payload-counted bytes. `receive_batch_message` increments
        /// `total_bytes_received` on each call — that counter is the
        /// observable side-channel proving "ran three times, not once."
        #[tokio::test]
        async fn handle_incoming_sync_stream_processes_multiple_frames_per_stream() {
            let _lock = TEST_LOCK.lock().await;
            let (endpoint_a, endpoint_b, conn_b_to_a, conn_a_from_b) = connected_pair().await;

            let (mut send_b, _recv_b) = conn_b_to_a.open_bi().await.expect("open_bi on B side");

            let empty_batch = SyncBatch::new();
            let frame_bytes = encode_batch_frame(&empty_batch);
            let frame_len = frame_bytes.len();

            for _ in 0..3 {
                send_b.write_all(&frame_bytes).await.expect("write frame");
            }
            send_b.finish().expect("finish send_b");

            let (send_a, recv_a) = conn_a_from_b
                .accept_bi()
                .await
                .expect("accept_bi on A side");

            let coordinator = make_minimal_coordinator();
            let peer_id = conn_a_from_b.remote_id();

            // Before: counter is zero.
            assert_eq!(
                coordinator
                    .total_bytes_received
                    .load(std::sync::atomic::Ordering::Relaxed),
                0
            );

            // Loop must process all three frames and terminate cleanly on
            // FinishedEarly(0).
            coordinator
                .handle_incoming_sync_stream(peer_id, send_a, recv_a)
                .await
                .expect("loop must terminate Ok on graceful end-of-stream");

            // After: counter reflects three frame payloads. Pre-fix, only
            // one frame would have advanced the counter before the
            // function returned.
            let observed = coordinator
                .total_bytes_received
                .load(std::sync::atomic::Ordering::Relaxed);
            assert_eq!(
                observed,
                (frame_len as u64) * 3,
                "handle_incoming_sync_stream must loop over all 3 frames; \
                 pre-#175 it would return after frame 1 (observed {observed} bytes, \
                 expected {} = 3 × {frame_len})",
                (frame_len as u64) * 3,
            );

            endpoint_a.close().await;
            endpoint_b.close().await;
        }

        /// Per-variant dispatch pin: a `DeltaSync` frame routed through
        /// `handle_incoming_sync_stream`'s loop reaches the coordinator's
        /// `receive_sync_message` path.
        ///
        /// peat-mesh#176 QA finding 4 (partial): the multi-frame test
        /// above exercises only the `Batch` arm of
        /// `dispatch_received_payload`'s 7-way match. This pin demonstrates
        /// the dispatch table works for at least one non-Batch variant
        /// (`DeltaSync`, msg_type=0x00) — hardening the
        /// "single source of truth across both loops" claim from
        /// `dispatch_received_payload`'s doc. The remaining variants
        /// (`StateSnapshot`/`Tombstone`/`TombstoneBatch`/`NegentropyInit`/
        /// `NegentropyResponse`) are existing handlers reachable via the
        /// old code path tested elsewhere — not regression risks for this
        /// PR per the QA reviewer's own assessment — and are deferred to
        /// a follow-up for full per-variant pin coverage.
        #[tokio::test]
        async fn handle_incoming_sync_stream_dispatches_delta_sync_variant() {
            let _lock = TEST_LOCK.lock().await;
            let (endpoint_a, endpoint_b, conn_b_to_a, conn_a_from_b) = connected_pair().await;

            let (mut send_b, _recv_b) = conn_b_to_a.open_bi().await.expect("open_bi on B side");

            // Synthesise a real Automerge sync message from a fresh remote
            // doc with one field — `generate_sync_message` always produces
            // an initial message against an empty peer state.
            let mut remote_doc = automerge::Automerge::new();
            let mut tx = remote_doc.transaction();
            tx.put(automerge::ROOT, "field", "delta-sync-variant-pin")
                .expect("put on root");
            tx.commit();
            let mut remote_state = automerge::sync::State::new();
            let sync_msg =
                automerge::sync::SyncDoc::generate_sync_message(&remote_doc, &mut remote_state)
                    .expect("fresh remote doc must produce an initial sync message");
            let encoded = sync_msg.clone().encode();
            let encoded_len = encoded.len();

            // Encode a v2 frame with msg_type=DeltaSync (0x00), not a
            // SyncBatch wrapper.
            let doc_key = b"nodes:delta-sync-variant-pin";
            let mut frame = Vec::with_capacity(2 + doc_key.len() + 1 + 4 + encoded_len);
            frame.extend_from_slice(&(doc_key.len() as u16).to_be_bytes());
            frame.extend_from_slice(doc_key);
            frame.push(SyncMessageType::DeltaSync as u8);
            frame.extend_from_slice(&(encoded_len as u32).to_be_bytes());
            frame.extend_from_slice(&encoded);
            let frame_len = frame.len();

            send_b.write_all(&frame).await.expect("write delta frame");
            send_b.finish().expect("finish send_b");

            let (send_a, recv_a) = conn_a_from_b
                .accept_bi()
                .await
                .expect("accept_bi on A side");
            let peer_id = conn_a_from_b.remote_id();

            // `PinnedConnTransport` returns the real `conn_a_from_b` so
            // `receive_sync_message`'s response-send tail
            // (`send_sync_message_for_doc` → `get_alive_connection` →
            // `open_bi` → write the response frame) reaches a real iroh
            // bi-stream. The response goes into iroh's send buffer; nobody
            // reads it on the peer side, but `send.finish()` +
            // `recv.stop(0)` cleans up the response stream normally. With
            // this transport the full dispatch + response pipeline is
            // asserted via `.expect(Ok)` — addressing peat-mesh#176
            // round-2 QA suggestion 1.
            let coordinator = make_coordinator_with_transport(Arc::new(PinnedConnTransport {
                peer_id,
                conn: conn_a_from_b.clone(),
            }));

            coordinator
                .handle_incoming_sync_stream(peer_id, send_a, recv_a)
                .await
                .expect(
                    "handle_incoming_sync_stream must complete cleanly under PinnedConnTransport",
                );

            // `total_bytes_received` is the cross-variant counter advanced
            // by every dispatch path (Delta / StateSnapshot / Tombstone /
            // Batch). Advancing by exactly `frame_len` confirms the
            // `DeltaSync` arm fired and routed the payload via
            // `dispatch_received_payload`. A counter at zero would mean
            // the loop returned without reaching the dispatcher.
            let observed = coordinator
                .total_bytes_received
                .load(std::sync::atomic::Ordering::Relaxed);
            assert_eq!(
                observed, frame_len as u64,
                "DeltaSync variant must route via dispatch_received_payload — observed {observed} bytes, expected {frame_len}"
            );

            // Response was sent on a fresh bi-stream over `conn_a_from_b`.
            // We don't introspect it here (that's a follow-up under
            // peat-mesh#177 once a capturing-not-just-pinning transport
            // lands), but `total_bytes_sent` advances on the coordinator
            // — confirming the response side of the pipeline ran end-to-end.
            let bytes_sent = coordinator
                .total_bytes_sent
                .load(std::sync::atomic::Ordering::Relaxed);
            assert!(
                bytes_sent > 0,
                "DeltaSync dispatch must generate a response sync message — \
                 total_bytes_sent advanced from 0 to {bytes_sent}"
            );

            // Doc-level effect: the synced doc must now exist in the
            // receiver's store. Beyond the byte counter, this proves the
            // DeltaSync handler (`receive_sync_message`) actually applied
            // the change rather than just counting bytes.
            let doc_key_str = std::str::from_utf8(doc_key).unwrap();
            let stored = coordinator.store.get(doc_key_str).expect("get doc_key");
            assert!(
                stored.is_some(),
                "DeltaSync dispatch must materialise the doc in the store"
            );

            endpoint_a.close().await;
            endpoint_b.close().await;
        }

        /// Per-variant dispatch pin: `StateSnapshot` (msg_type=0x01).
        ///
        /// peat-mesh#177: closes the remaining-variant gap from peat-mesh#176
        /// finding 4. Writes a full Automerge document via `doc.save()`,
        /// routes it through `handle_incoming_sync_stream`, and asserts the
        /// snapshot materialised in the receiver's store. `apply_state_snapshot`
        /// merges the snapshot into any existing doc; we send to a fresh
        /// doc_key so the "no existing doc" branch fires.
        #[tokio::test]
        async fn handle_incoming_sync_stream_dispatches_state_snapshot_variant() {
            let _lock = TEST_LOCK.lock().await;
            let (endpoint_a, endpoint_b, conn_b_to_a, conn_a_from_b) = connected_pair().await;

            let (mut send_b, _recv_b) = conn_b_to_a.open_bi().await.expect("open_bi on B side");

            // Build a real Automerge doc and serialise it as a snapshot.
            let mut remote_doc = automerge::Automerge::new();
            let mut tx = remote_doc.transaction();
            tx.put(automerge::ROOT, "field", "state-snapshot-variant-pin")
                .expect("put on root");
            tx.commit();
            let snapshot_bytes = remote_doc.save();

            let doc_key = b"nodes:state-snapshot-variant-pin";
            let frame = encode_v2_frame(doc_key, SyncMessageType::StateSnapshot, &snapshot_bytes);
            let frame_len = frame.len();

            send_b
                .write_all(&frame)
                .await
                .expect("write snapshot frame");
            send_b.finish().expect("finish send_b");

            let (send_a, recv_a) = conn_a_from_b
                .accept_bi()
                .await
                .expect("accept_bi on A side");
            let peer_id = conn_a_from_b.remote_id();

            let coordinator = make_coordinator_with_transport(Arc::new(PinnedConnTransport {
                peer_id,
                conn: conn_a_from_b.clone(),
            }));

            coordinator
                .handle_incoming_sync_stream(peer_id, send_a, recv_a)
                .await
                .expect("handle_incoming_sync_stream must complete cleanly");

            let observed = coordinator
                .total_bytes_received
                .load(std::sync::atomic::Ordering::Relaxed);
            assert_eq!(
                observed, frame_len as u64,
                "StateSnapshot variant must route via dispatch_received_payload"
            );

            // Doc-level effect: apply_state_snapshot must have stored the
            // received document at doc_key.
            let doc_key_str = std::str::from_utf8(doc_key).unwrap();
            let stored = coordinator.store.get(doc_key_str).expect("get doc_key");
            assert!(
                stored.is_some(),
                "StateSnapshot dispatch must materialise the doc in the store"
            );

            endpoint_a.close().await;
            endpoint_b.close().await;
        }

        /// Per-variant dispatch pin: `Tombstone` (msg_type=0x04).
        ///
        /// peat-mesh#177: closes the remaining-variant gap. Writes a single
        /// `TombstoneSyncMessage` and asserts the dispatch arm fired by
        /// observing `total_bytes_received` advance (incremented by
        /// `handle_incoming_tombstone` at `automerge_sync.rs:2639` before
        /// any persistence call).
        ///
        /// State-effect assertion (`has_tombstone`) is **not** included:
        /// `AutomergeStore::put_tombstone` is a no-op in memory mode
        /// (`automerge_store.rs:765` — `if let Some(ref db) = self.db
        /// else { return Ok(()); }`) and `make_coordinator_with_transport`
        /// wires an in-memory store. A persistence-layer pin would
        /// require swapping in an on-disk redb store, which is outside
        /// the dispatch-arm regression scope. Persistence-layer tombstone
        /// behaviour has its own tests in this file.
        #[tokio::test]
        async fn handle_incoming_sync_stream_dispatches_tombstone_variant() {
            use crate::qos::{PropagationDirection, Tombstone, TombstoneSyncMessage};

            let _lock = TEST_LOCK.lock().await;
            let (endpoint_a, endpoint_b, conn_b_to_a, conn_a_from_b) = connected_pair().await;

            let (mut send_b, _recv_b) = conn_b_to_a.open_bi().await.expect("open_bi on B side");

            let tombstone = Tombstone::new(
                "doc-tombstone-variant-pin",
                "nodes",
                "deleter-id",
                42, // lamport
            );
            let tombstone_msg =
                TombstoneSyncMessage::new(tombstone, PropagationDirection::Bidirectional);
            let payload = tombstone_msg.encode();

            // doc_key is ignored by handle_incoming_tombstone (line 2633:
            // `_doc_key`) — it derives collection/doc_id from the
            // tombstone message itself. We still need a valid v2 frame.
            let doc_key = b"tombstones:nodes:doc-tombstone-variant-pin";
            let frame = encode_v2_frame(doc_key, SyncMessageType::Tombstone, &payload);
            let frame_len = frame.len();

            send_b
                .write_all(&frame)
                .await
                .expect("write tombstone frame");
            send_b.finish().expect("finish send_b");

            let (send_a, recv_a) = conn_a_from_b
                .accept_bi()
                .await
                .expect("accept_bi on A side");
            let peer_id = conn_a_from_b.remote_id();

            let coordinator = make_coordinator_with_transport(Arc::new(PinnedConnTransport {
                peer_id,
                conn: conn_a_from_b.clone(),
            }));

            coordinator
                .handle_incoming_sync_stream(peer_id, send_a, recv_a)
                .await
                .expect("handle_incoming_sync_stream must complete cleanly");

            let observed = coordinator
                .total_bytes_received
                .load(std::sync::atomic::Ordering::Relaxed);
            assert_eq!(
                observed, frame_len as u64,
                "Tombstone variant must route via dispatch_received_payload — \
                 observed {observed} bytes, expected {frame_len}"
            );

            endpoint_a.close().await;
            endpoint_b.close().await;
        }

        /// Per-variant dispatch pin: `TombstoneBatch` (msg_type=0x05).
        ///
        /// peat-mesh#177: closes the remaining-variant gap. Asserts the
        /// dispatch arm fired by observing `total_bytes_received` advance.
        /// Per-entry materialisation can't be pinned here for the same
        /// in-memory-store reason as the Tombstone variant pin above.
        #[tokio::test]
        async fn handle_incoming_sync_stream_dispatches_tombstone_batch_variant() {
            use crate::qos::{Tombstone, TombstoneBatch, TombstoneSyncMessage};

            let _lock = TEST_LOCK.lock().await;
            let (endpoint_a, endpoint_b, conn_b_to_a, conn_a_from_b) = connected_pair().await;

            let (mut send_b, _recv_b) = conn_b_to_a.open_bi().await.expect("open_bi on B side");

            let tombstones = vec![
                Tombstone::new("doc-batch-1", "nodes", "deleter-id", 100),
                Tombstone::new("doc-batch-2", "nodes", "deleter-id", 101),
                Tombstone::new("doc-batch-3", "beacons", "deleter-id", 102),
            ];
            let tombstone_msgs: Vec<TombstoneSyncMessage> = tombstones
                .into_iter()
                .map(TombstoneSyncMessage::from_tombstone)
                .collect();
            let batch = TombstoneBatch::with_messages(tombstone_msgs);
            let payload = batch.encode();

            let doc_key = b"tombstones:batch";
            let frame = encode_v2_frame(doc_key, SyncMessageType::TombstoneBatch, &payload);
            let frame_len = frame.len();

            send_b
                .write_all(&frame)
                .await
                .expect("write tombstone batch frame");
            send_b.finish().expect("finish send_b");

            let (send_a, recv_a) = conn_a_from_b
                .accept_bi()
                .await
                .expect("accept_bi on A side");
            let peer_id = conn_a_from_b.remote_id();

            let coordinator = make_coordinator_with_transport(Arc::new(PinnedConnTransport {
                peer_id,
                conn: conn_a_from_b.clone(),
            }));

            coordinator
                .handle_incoming_sync_stream(peer_id, send_a, recv_a)
                .await
                .expect("handle_incoming_sync_stream must complete cleanly");

            let observed = coordinator
                .total_bytes_received
                .load(std::sync::atomic::Ordering::Relaxed);
            assert_eq!(
                observed, frame_len as u64,
                "TombstoneBatch variant must route via dispatch_received_payload — \
                 observed {observed} bytes, expected {frame_len}"
            );

            endpoint_a.close().await;
            endpoint_b.close().await;
        }

        /// peat-mesh#196: inbound `Tombstone.lamport` advances the
        /// node-local Lamport clock wired into the coordinator via
        /// `set_lamport_clock`. Drives a single-tombstone frame through
        /// the full `handle_incoming_sync_stream` →
        /// `dispatch_received_payload` → `handle_incoming_tombstone`
        /// path with a clock attached, and asserts the clock advanced
        /// past the payload's Lamport.
        ///
        /// This is the dispatch-tier pin the issue acceptance criterion
        /// names: "construct a coordinator with a fake clock handle,
        /// dispatch a synthetic SyncEntry::Tombstone, assert the handle
        /// was advanced past the payload's lamport." The handle is a
        /// plain `Arc<AtomicU64>`; the coordinator's
        /// `observe_inbound_lamport` calls
        /// `crate::sync::automerge_backend::observe_atomic` against it.
        #[tokio::test]
        async fn tombstone_receive_advances_wired_lamport_clock() {
            use crate::qos::{PropagationDirection, Tombstone, TombstoneSyncMessage};

            let _lock = TEST_LOCK.lock().await;
            let (endpoint_a, endpoint_b, conn_b_to_a, conn_a_from_b) = connected_pair().await;

            let (mut send_b, _recv_b) = conn_b_to_a.open_bi().await.expect("open_bi on B side");

            // Pick a Lamport value the coordinator's clock cannot
            // reach incidentally. `u64::MAX / 2` is well above any
            // wall-clock seed but still leaves room for the +1 of
            // observe_atomic without saturating.
            const SENT_LAMPORT: u64 = u64::MAX / 2;
            let tombstone = Tombstone::new("doc-rx-1", "tracks", "deleter-id", SENT_LAMPORT);
            let tombstone_msg =
                TombstoneSyncMessage::new(tombstone, PropagationDirection::Bidirectional);
            let payload = tombstone_msg.encode();

            let doc_key = b"tombstones:tracks:doc-rx-1";
            let frame = encode_v2_frame(doc_key, SyncMessageType::Tombstone, &payload);

            send_b
                .write_all(&frame)
                .await
                .expect("write tombstone frame");
            send_b.finish().expect("finish send_b");

            let (send_a, recv_a) = conn_a_from_b
                .accept_bi()
                .await
                .expect("accept_bi on A side");
            let peer_id = conn_a_from_b.remote_id();

            let coordinator = make_coordinator_with_transport(Arc::new(PinnedConnTransport {
                peer_id,
                conn: conn_a_from_b.clone(),
            }));

            // Attach a fresh Lamport clock starting below SENT_LAMPORT
            // so the observe is unambiguously the source of the
            // advance.
            let clock = Arc::new(AtomicU64::new(1));
            coordinator.set_lamport_clock(Arc::clone(&clock));

            coordinator
                .handle_incoming_sync_stream(peer_id, send_a, recv_a)
                .await
                .expect("handle_incoming_sync_stream must complete cleanly");

            let advanced = clock.load(std::sync::atomic::Ordering::Relaxed);
            assert!(
                advanced > SENT_LAMPORT,
                "wired Lamport clock must advance past inbound Tombstone.lamport: \
                 sent={SENT_LAMPORT}, clock_after={advanced}"
            );

            endpoint_a.close().await;
            endpoint_b.close().await;
        }

        /// peat-mesh#196: same pin for `TombstoneBatch`. The batch
        /// handler observes the **max** Lamport across all tombstones
        /// in the batch (a single CAS rather than N), and the resulting
        /// clock value must exceed every per-entry Lamport — exercised
        /// here by sending three tombstones with mixed Lamports and
        /// asserting `clock > max(L1, L2, L3)`.
        #[tokio::test]
        async fn tombstone_batch_receive_advances_wired_lamport_clock_past_max() {
            use crate::qos::{Tombstone, TombstoneBatch, TombstoneSyncMessage};

            let _lock = TEST_LOCK.lock().await;
            let (endpoint_a, endpoint_b, conn_b_to_a, conn_a_from_b) = connected_pair().await;

            let (mut send_b, _recv_b) = conn_b_to_a.open_bi().await.expect("open_bi on B side");

            // Three Lamports with the largest in the middle position
            // — proves the handler computes max-over-batch, not e.g.
            // first-only or last-only.
            const L1: u64 = 1_000_000;
            const L_MAX: u64 = u64::MAX / 4;
            const L3: u64 = 2_000_000;

            let tombstones = vec![
                Tombstone::new("doc-batch-1", "nodes", "deleter-id", L1),
                Tombstone::new("doc-batch-2", "nodes", "deleter-id", L_MAX),
                Tombstone::new("doc-batch-3", "beacons", "deleter-id", L3),
            ];
            let tombstone_msgs: Vec<TombstoneSyncMessage> = tombstones
                .into_iter()
                .map(TombstoneSyncMessage::from_tombstone)
                .collect();
            let batch = TombstoneBatch::with_messages(tombstone_msgs);
            let payload = batch.encode();

            let doc_key = b"tombstones:batch";
            let frame = encode_v2_frame(doc_key, SyncMessageType::TombstoneBatch, &payload);

            send_b
                .write_all(&frame)
                .await
                .expect("write tombstone batch frame");
            send_b.finish().expect("finish send_b");

            let (send_a, recv_a) = conn_a_from_b
                .accept_bi()
                .await
                .expect("accept_bi on A side");
            let peer_id = conn_a_from_b.remote_id();

            let coordinator = make_coordinator_with_transport(Arc::new(PinnedConnTransport {
                peer_id,
                conn: conn_a_from_b.clone(),
            }));

            let clock = Arc::new(AtomicU64::new(1));
            coordinator.set_lamport_clock(Arc::clone(&clock));

            coordinator
                .handle_incoming_sync_stream(peer_id, send_a, recv_a)
                .await
                .expect("handle_incoming_sync_stream must complete cleanly");

            let advanced = clock.load(std::sync::atomic::Ordering::Relaxed);
            assert!(
                advanced > L_MAX,
                "wired Lamport clock must advance past max(batch.lamports): \
                 max={L_MAX}, clock_after={advanced}"
            );

            endpoint_a.close().await;
            endpoint_b.close().await;
        }

        /// peat-mesh#196 no-op guarantee: when the coordinator has no
        /// Lamport clock wired (standalone construction — peat-mesh-node
        /// binary, unit tests), inbound Tombstone messages still
        /// dispatch cleanly without crashing or otherwise misbehaving.
        /// The same `tombstone_receive_advances_wired_lamport_clock`
        /// frame, but with `set_lamport_clock` omitted, must complete
        /// successfully — proves the wire-up is opt-in, not required.
        #[tokio::test]
        async fn tombstone_receive_without_wired_clock_is_a_clean_noop() {
            use crate::qos::{PropagationDirection, Tombstone, TombstoneSyncMessage};

            let _lock = TEST_LOCK.lock().await;
            let (endpoint_a, endpoint_b, conn_b_to_a, conn_a_from_b) = connected_pair().await;

            let (mut send_b, _recv_b) = conn_b_to_a.open_bi().await.expect("open_bi on B side");

            let tombstone = Tombstone::new("doc-noop", "tracks", "deleter-id", 12345);
            let tombstone_msg =
                TombstoneSyncMessage::new(tombstone, PropagationDirection::Bidirectional);
            let payload = tombstone_msg.encode();

            let doc_key = b"tombstones:tracks:doc-noop";
            let frame = encode_v2_frame(doc_key, SyncMessageType::Tombstone, &payload);

            send_b.write_all(&frame).await.expect("write tombstone");
            send_b.finish().expect("finish send_b");

            let (send_a, recv_a) = conn_a_from_b
                .accept_bi()
                .await
                .expect("accept_bi on A side");
            let peer_id = conn_a_from_b.remote_id();

            // Note: no `set_lamport_clock` call. The coordinator's
            // `lamport_clock` is `Arc<RwLock<Option<_>>>` initialised
            // to `None`; `observe_inbound_lamport` early-returns.
            let coordinator = make_coordinator_with_transport(Arc::new(PinnedConnTransport {
                peer_id,
                conn: conn_a_from_b.clone(),
            }));

            coordinator
                .handle_incoming_sync_stream(peer_id, send_a, recv_a)
                .await
                .expect("handle_incoming_sync_stream must complete cleanly without a clock");

            endpoint_a.close().await;
            endpoint_b.close().await;
        }

        /// peat-mesh#196 QA defensive cap: `observe_atomic` must refuse
        /// to advance the clock past `u64::MAX - 1`. Without this cap,
        /// a hostile or buggy peer sending `Tombstone.lamport == u64::MAX`
        /// (or any local consumer making the same call) would saturate
        /// the atomic at `u64::MAX`; the next `next_lamport()` would
        /// then `fetch_add(1).wrapping_add(1)` → `0`, breaking the
        /// "never return 0" invariant the seed-floor protects. The cap
        /// keeps the atomic at most `u64::MAX - 1` so `next_lamport`
        /// can still produce `u64::MAX` once before any wrap risk.
        #[test]
        fn observe_atomic_caps_advance_at_u64_max_minus_one() {
            use crate::sync::automerge_backend::observe_atomic;

            // Pre-cap baseline: a small starting clock.
            let atomic = AtomicU64::new(1);

            // Try to advance to `u64::MAX` — must be capped.
            observe_atomic(&atomic, u64::MAX);
            assert_eq!(
                atomic.load(std::sync::atomic::Ordering::Relaxed),
                u64::MAX - 1,
                "observe_atomic must cap the advance at u64::MAX - 1"
            );

            // A subsequent `fetch_add(1).wrapping_add(1)` (the
            // shape of `next_lamport`) returns u64::MAX, NOT 0 —
            // proving the cap forestalls the wrap-to-zero.
            let cur = atomic.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
            let next = cur.wrapping_add(1);
            assert_eq!(
                next,
                u64::MAX,
                "next_lamport after capped observe must return u64::MAX, not 0"
            );
        }

        /// peat-mesh#196 QA defensive cap, end-to-end via the
        /// receive path: send a Tombstone with `lamport == u64::MAX`
        /// through `handle_incoming_sync_stream`, assert the wired
        /// clock stops at `u64::MAX - 1` rather than saturating to
        /// `u64::MAX`. Pairs with the unit pin above.
        #[tokio::test]
        async fn tombstone_receive_caps_wired_clock_against_u64_max() {
            use crate::qos::{PropagationDirection, Tombstone, TombstoneSyncMessage};

            let _lock = TEST_LOCK.lock().await;
            let (endpoint_a, endpoint_b, conn_b_to_a, conn_a_from_b) = connected_pair().await;

            let (mut send_b, _recv_b) = conn_b_to_a.open_bi().await.expect("open_bi on B side");

            let tombstone = Tombstone::new("doc-evil", "tracks", "evil-peer", u64::MAX);
            let tombstone_msg =
                TombstoneSyncMessage::new(tombstone, PropagationDirection::Bidirectional);
            let payload = tombstone_msg.encode();

            let doc_key = b"tombstones:tracks:doc-evil";
            let frame = encode_v2_frame(doc_key, SyncMessageType::Tombstone, &payload);

            send_b.write_all(&frame).await.expect("write tombstone");
            send_b.finish().expect("finish send_b");

            let (send_a, recv_a) = conn_a_from_b
                .accept_bi()
                .await
                .expect("accept_bi on A side");
            let peer_id = conn_a_from_b.remote_id();

            let coordinator = make_coordinator_with_transport(Arc::new(PinnedConnTransport {
                peer_id,
                conn: conn_a_from_b.clone(),
            }));

            let clock = Arc::new(AtomicU64::new(1));
            coordinator.set_lamport_clock(Arc::clone(&clock));

            coordinator
                .handle_incoming_sync_stream(peer_id, send_a, recv_a)
                .await
                .expect("handle_incoming_sync_stream must complete cleanly");

            let advanced = clock.load(std::sync::atomic::Ordering::Relaxed);
            assert_eq!(
                advanced,
                u64::MAX - 1,
                "inbound u64::MAX Lamport must cap the wired clock at u64::MAX - 1, not saturate to u64::MAX"
            );

            endpoint_a.close().await;
            endpoint_b.close().await;
        }

        /// peat-mesh#202: inbound tombstone must fire `observer_tx`
        /// on the receiving node — the documented "fires for ALL
        /// document changes" contract on
        /// `AutomergeStore::subscribe_to_observer_changes` includes
        /// deletes from remote peers. Pre-fix, `apply_tombstone`
        /// called `store.delete(&doc_key)` which silently removed
        /// from disk + cache without firing the broadcast.
        /// Post-fix, it routes through
        /// `delete_with_origin(.., Remote(peer))` which fires
        /// `observer_tx` + `gossip_tx(Remote(peer))` while
        /// suppressing `change_tx` per peat-mesh#115.
        ///
        /// Pin shape: seed a document on the receiver's store, drive
        /// a Tombstone frame with the matching `(collection,
        /// document_id)` through `handle_incoming_sync_stream`, and
        /// assert (a) the document is removed AND (b) the observer
        /// channel woke with the doc key.
        #[tokio::test]
        async fn tombstone_receive_fires_observer_tx() {
            use crate::qos::{PropagationDirection, Tombstone, TombstoneSyncMessage};

            let _lock = TEST_LOCK.lock().await;
            let (endpoint_a, endpoint_b, conn_b_to_a, conn_a_from_b) = connected_pair().await;

            let (mut send_b, _recv_b) = conn_b_to_a.open_bi().await.expect("open_bi on B side");

            // The doc key the receiver will seed and the tombstone
            // will target. peat-mesh's tombstone-to-doc-key shape is
            // `"{collection}:{document_id}"`.
            let collection = "contacts";
            let document_id = "c-rx-1";
            let doc_key = format!("{collection}:{document_id}");

            let tombstone = Tombstone::new(document_id, collection, "deleter-id", 42);
            let tombstone_msg =
                TombstoneSyncMessage::new(tombstone, PropagationDirection::Bidirectional);
            let payload = tombstone_msg.encode();

            // Wire `doc_key` matches the convention `handle_incoming_tombstone`
            // re-derives internally from `tombstone.collection` +
            // `tombstone.document_id` — the doc_key here is for the
            // v2 framing layer, not the tombstone payload.
            let frame_doc_key = doc_key.as_bytes();
            let frame = encode_v2_frame(frame_doc_key, SyncMessageType::Tombstone, &payload);

            send_b.write_all(&frame).await.expect("write tombstone");
            send_b.finish().expect("finish send_b");

            let (send_a, recv_a) = conn_a_from_b
                .accept_bi()
                .await
                .expect("accept_bi on A side");
            let peer_id = conn_a_from_b.remote_id();

            let coordinator = make_coordinator_with_transport(Arc::new(PinnedConnTransport {
                peer_id,
                conn: conn_a_from_b.clone(),
            }));

            // Seed the document on the receiver's store. Drives the
            // `store.get(&doc_key)?.is_some()` branch inside
            // `apply_tombstone` that conditionally invokes the
            // delete path. Without a seeded doc, the delete is
            // skipped and observer_tx never fires for the right
            // reason (vacuously absent).
            let mut seed = automerge::Automerge::new();
            seed.transact::<_, _, automerge::AutomergeError>(|tx| {
                use automerge::transaction::Transactable;
                tx.put(automerge::ROOT, "field", "alice")?;
                Ok(())
            })
            .expect("seed transaction");
            coordinator
                .store
                .put(&doc_key, &seed)
                .expect("seed document on receiver");

            // Subscribe AFTER the seed put so the seed doesn't
            // pollute the assertion stream.
            let mut observer_rx = coordinator.store.subscribe_to_observer_changes();

            coordinator
                .handle_incoming_sync_stream(peer_id, send_a, recv_a)
                .await
                .expect("handle_incoming_sync_stream must complete cleanly");

            // observer_tx must have fired with `doc_key` — the
            // peat-mesh#202 contract fix. Pre-fix this recv would
            // time out.
            let observed_key =
                tokio::time::timeout(std::time::Duration::from_millis(500), observer_rx.recv())
                    .await
                    .expect("observer_tx must fire on remote-driven delete (peat-mesh#202)")
                    .expect("observer_tx must yield Ok");
            assert_eq!(observed_key, doc_key);

            // And the document is actually gone.
            let after = coordinator
                .store
                .get(&doc_key)
                .expect("get post-tombstone-apply");
            assert!(
                after.is_none(),
                "apply_tombstone must have removed the local document"
            );

            endpoint_a.close().await;
            endpoint_b.close().await;
        }

        /// Combined parse-side pin for the two Negentropy variants
        /// (msg_type=0x08 NegentropyInit, msg_type=0x09 NegentropyResponse).
        ///
        /// peat-mesh#177: closes the remaining-variant gap for Negentropy.
        /// Unlike the other 5 variants, the Negentropy handlers
        /// (`handle_negentropy_init` / `handle_negentropy_response`) do
        /// **not** increment `total_bytes_received` (Negentropy framing
        /// is its own subprotocol with its own accounting), and they
        /// short-circuit on malformed payload BEFORE any observable side
        /// effect. The closest dispatch-side pin is therefore a parse-side
        /// roundtrip: `receive_sync_payload_from_stream` correctly decodes
        /// both variants into the expected `ReceivedSyncPayload` shape.
        /// Combined with the exhaustive-match in `dispatch_received_payload`
        /// (compile-checked), this pins the variant + the dispatch
        /// routing without depending on handler semantics.
        ///
        /// The full Negentropy protocol roundtrip (init → response →
        /// reconciliation) is a separate concern with its own tests
        /// in this file.
        #[tokio::test]
        async fn receive_sync_payload_from_stream_decodes_negentropy_variants() {
            let _lock = TEST_LOCK.lock().await;
            let (endpoint_a, endpoint_b, conn_b_to_a, conn_a_from_b) = connected_pair().await;

            let (mut send_b, _recv_b) = conn_b_to_a.open_bi().await.expect("open_bi on B side");

            // NegentropyInit and NegentropyResponse payloads are opaque
            // bytes at the framing layer — the dispatch layer hands them
            // to `handle_negentropy_init` / `handle_negentropy_response`
            // verbatim. Any non-empty payload exercises the parse path.
            let init_payload = vec![0x01, 0x02, 0x03, 0x04];
            let resp_payload = vec![0xAA, 0xBB, 0xCC, 0xDD];

            send_b
                .write_all(&encode_v2_frame(
                    b"negentropy:init",
                    SyncMessageType::NegentropyInit,
                    &init_payload,
                ))
                .await
                .expect("write negentropy init frame");
            send_b
                .write_all(&encode_v2_frame(
                    b"negentropy:resp",
                    SyncMessageType::NegentropyResponse,
                    &resp_payload,
                ))
                .await
                .expect("write negentropy response frame");
            send_b.finish().expect("finish send_b");

            let (_send_a, mut recv_a) = conn_a_from_b
                .accept_bi()
                .await
                .expect("accept_bi on A side");

            let coordinator = make_minimal_coordinator();

            // Frame 1: NegentropyInit.
            let (doc_key_1, payload_1, _size_1) = coordinator
                .receive_sync_payload_from_stream(&mut recv_a)
                .await
                .expect("frame 1 must parse");
            assert_eq!(doc_key_1, "negentropy:init");
            match payload_1 {
                ReceivedSyncPayload::NegentropyInit(bytes) => {
                    assert_eq!(bytes, init_payload, "NegentropyInit payload must roundtrip");
                }
                other => panic!("frame 1 must decode as NegentropyInit, got {:?}", other),
            }

            // Frame 2: NegentropyResponse.
            let (doc_key_2, payload_2, _size_2) = coordinator
                .receive_sync_payload_from_stream(&mut recv_a)
                .await
                .expect("frame 2 must parse");
            assert_eq!(doc_key_2, "negentropy:resp");
            match payload_2 {
                ReceivedSyncPayload::NegentropyResponse(bytes) => {
                    assert_eq!(
                        bytes, resp_payload,
                        "NegentropyResponse payload must roundtrip"
                    );
                }
                other => panic!("frame 2 must decode as NegentropyResponse, got {:?}", other),
            }

            endpoint_a.close().await;
            endpoint_b.close().await;
        }

        /// Pin: `SyncChannel::receive_loop` parses v2 frames written by a
        /// peer into the connect-side recv stream and dispatches them to
        /// the coordinator.
        ///
        /// peat-mesh#176 QA finding 1 (WARNING): the receive_loop rewrite
        /// itself was not directly tested by the earlier pins, only its
        /// `receive_sync_payload_from_stream` delegate. This test drives
        /// frames into the connect-side stream via the SyncChannel public
        /// API (constructed via `SyncChannel::connect`) and observes the
        /// coordinator's `total_bytes_received` advance — catching any
        /// future refactor that breaks the connect-side framing parse
        /// without breaking the delegate.
        ///
        /// Setup: B opens `SyncChannel::connect`, which calls
        /// `conn.open_bi()` and spawns `receive_loop` on the recv half.
        /// The test then accepts the corresponding bi-stream on A's side
        /// (`conn_a_from_b.accept_bi()`) and manually writes v2-encoded
        /// frames into the send half (the symmetric `SyncChannel::accept`
        /// constructor is ADR-063 §D3 deferred work). B's receive_loop
        /// must parse those frames and dispatch them to B's coordinator.
        #[tokio::test]
        async fn sync_channel_receive_loop_dispatches_frames_to_coordinator() {
            use crate::storage::sync_channel::SyncChannel;

            let _lock = TEST_LOCK.lock().await;
            let (endpoint_a, endpoint_b, conn_b_to_a, conn_a_from_b) = connected_pair().await;

            // B's coordinator wired to PinnedConnTransport — when the
            // dispatch path generates a response for the non-empty entry
            // below (`receive_sync_message → send_sync_message_for_doc`),
            // it has a real connection to write to. Addresses
            // peat-mesh#176 round-2 QA suggestion 1.
            let peer_id_a = conn_b_to_a.remote_id();
            let coord_b = Arc::new(make_coordinator_with_transport(Arc::new(
                PinnedConnTransport {
                    peer_id: peer_id_a,
                    conn: conn_b_to_a.clone(),
                },
            )));

            // SyncChannel's own transport-pinning instance (same shape).
            let transport_b: Arc<dyn SyncTransport> = Arc::new(PinnedConnTransport {
                peer_id: peer_id_a,
                conn: conn_b_to_a.clone(),
            });

            // B opens the SyncChannel — this calls conn_b_to_a.open_bi()
            // and spawns receive_loop on the recv half.
            let channel = SyncChannel::connect(transport_b, peer_id_a, Arc::clone(&coord_b))
                .await
                .expect("SyncChannel::connect on B");

            // Materialise the bi-stream on the wire by sending an initial
            // empty SyncBatch from B. Per iroh's noq layer, "Calling
            // open_bi() then waiting on RecvStream without writing
            // anything to SendStream will never succeed" — A's accept_bi
            // blocks until B writes at least one byte.
            channel
                .send(&SyncBatch::new())
                .await
                .expect("initial send from B materialises the bi-stream");

            // A accepts the bi-stream B's connect+send just established.
            let (mut send_a, _recv_a) = conn_a_from_b
                .accept_bi()
                .await
                .expect("accept_bi on A side");

            // Write two v2 SyncBatch frames into the A → B direction.
            // receive_loop on B must parse both and dispatch via
            // receive_batch_message → total_bytes_received advances.
            //
            // Frame 1: empty batch — exercises receive_batch_message's
            // accounting on the no-op-entries path.
            // Frame 2: a SyncBatch carrying a real Automerge DeltaSync
            // entry. This drives the per-entry processing path inside
            // receive_batch_message, which calls receive_sync_message →
            // generate_sync_message → send_sync_message_for_doc (the
            // PinnedConnTransport above lets that response-send succeed).
            // Addresses peat-mesh#176 round-2 QA suggestion 2.
            let batch_one = SyncBatch::new();

            let mut remote_doc = automerge::Automerge::new();
            let mut tx = remote_doc.transaction();
            tx.put(automerge::ROOT, "field", "receive-loop-pin-non-empty")
                .expect("put on root");
            tx.commit();
            let mut remote_state = automerge::sync::State::new();
            let sync_msg =
                automerge::sync::SyncDoc::generate_sync_message(&remote_doc, &mut remote_state)
                    .expect("fresh remote doc must produce an initial sync message");

            let mut batch_two = SyncBatch::new();
            batch_two.entries.push(SyncEntry::new(
                "nodes:receive-loop-pin-non-empty".to_string(),
                SyncMessageType::DeltaSync,
                sync_msg.encode(),
            ));

            let frame_one = encode_batch_frame(&batch_one);
            let frame_two = encode_batch_frame(&batch_two);
            let total_bytes_expected = (frame_one.len() + frame_two.len()) as u64;

            send_a.write_all(&frame_one).await.expect("write frame 1");
            send_a.write_all(&frame_two).await.expect("write frame 2");
            send_a.finish().expect("finish send_a");

            // Bounded wait with `await_counter_at_least` — replaces the
            // prior 10s sleep-loop. On a regression the budget fires in
            // ~3s and the assertion below surfaces the observed value
            // crisply rather than hanging until a test-runner timeout.
            // Addresses peat-mesh#176 round-2 QA suggestion 3.
            let observed = await_counter_at_least(
                &coord_b.total_bytes_received,
                total_bytes_expected,
                Duration::from_secs(3),
            )
            .await;
            // `>=` rather than `==`: receive_batch_message increments the
            // counter by the outer frame size, and the inner DeltaSync
            // entry's `receive_sync_message` increments by its own
            // encoded size on top (see `automerge_sync.rs:1242`). The
            // lower bound proves both frames reached the dispatcher;
            // the surplus is the per-entry accounting from the non-empty
            // batch. Pre-#175 the counter would have been 0 because the
            // loop's framing parse desynchronized on byte 0 of frame 1.
            assert!(
                observed >= total_bytes_expected,
                "receive_loop must dispatch both frames to coord_b — observed \
                 {observed} bytes, expected ≥ {total_bytes_expected}. Pre-#175 \
                 the loop misparsed the framing and silently desynchronized."
            );

            // The non-empty batch entry must materialise in B's store —
            // exercises the per-entry receive_batch_message → receive_sync_message
            // path inside the dispatch.
            let stored = coord_b
                .store
                .get("nodes:receive-loop-pin-non-empty")
                .expect("get doc_key");
            assert!(
                stored.is_some(),
                "Non-empty SyncBatch entry must materialise in coord_b's store"
            );

            channel.close().await;
            endpoint_a.close().await;
            endpoint_b.close().await;
        }

        /// peat-mesh#239 regression guard for the receive-idle-timeout removal
        /// (`SyncChannel::receive_loop`): an idle-but-live channel must still
        /// deliver a frame sent after a quiet period that exceeds the removed
        /// 30s `RECV_TIMEOUT`. Pre-#239 the receiver task returned
        /// `Err("Receive idle timeout")` at 30s and was never re-armed, so the
        /// late frame was lost; with the timeout gone the loop blocks for the
        /// next frame and QUIC keepalive holds the connection.
        ///
        /// Uses `tokio::time::advance` to cross the >30s window without a real
        /// wall-clock wait (quinn's timers run on the tokio time driver).
        #[tokio::test(start_paused = true)]
        async fn receive_loop_delivers_frame_after_idle_beyond_old_recv_timeout() {
            use crate::storage::sync_channel::SyncChannel;

            let _lock = TEST_LOCK.lock().await;
            let (endpoint_a, endpoint_b, conn_b_to_a, conn_a_from_b) = connected_pair().await;

            let peer_id_a = conn_b_to_a.remote_id();
            let coord_b = Arc::new(make_coordinator_with_transport(Arc::new(
                PinnedConnTransport {
                    peer_id: peer_id_a,
                    conn: conn_b_to_a.clone(),
                },
            )));
            let transport_b: Arc<dyn SyncTransport> = Arc::new(PinnedConnTransport {
                peer_id: peer_id_a,
                conn: conn_b_to_a.clone(),
            });

            let channel = SyncChannel::connect(transport_b, peer_id_a, Arc::clone(&coord_b))
                .await
                .expect("SyncChannel::connect on B");
            channel
                .send(&SyncBatch::new())
                .await
                .expect("initial send materialises the bi-stream");
            let (mut send_a, _recv_a) = conn_a_from_b
                .accept_bi()
                .await
                .expect("accept_bi on A side");

            // Idle the channel past the OLD 30s RECV_TIMEOUT before sending the
            // first measured frame. With the timeout removed, the receiver is
            // still alive to process it. (Real-QUIC I/O means `advance` does not
            // fully virtualize away the wall-clock here, so keep the margin just
            // over 30s rather than larger.)
            tokio::time::advance(Duration::from_secs(31)).await;

            let frame = encode_batch_frame(&SyncBatch::new());
            let expected = frame.len() as u64;
            send_a
                .write_all(&frame)
                .await
                .expect("write frame after idle");
            send_a.finish().expect("finish send_a");

            let observed = await_counter_at_least(
                &coord_b.total_bytes_received,
                expected,
                Duration::from_secs(10),
            )
            .await;
            assert!(
                observed >= expected,
                "receive_loop must still deliver a frame after a >30s idle gap \
                 (peat-mesh#239: pre-fix the 30s RECV_TIMEOUT killed the idle receiver). \
                 observed {observed}, expected ≥ {expected}"
            );

            channel.close().await;
            endpoint_a.close().await;
            endpoint_b.close().await;
        }

        /// E2E pin for `SyncChannel::send`'s writer-task pipeline
        /// (peat-mesh#178 QA [WARNING]: encode side was not directly
        /// covered).
        ///
        /// Drives `channel.send(batch)` from N concurrent tasks on the
        /// initiator side, observes the bytes arriving at the responder's
        /// coordinator via `handle_incoming_sync_stream`'s frame loop,
        /// and asserts every batch was dispatched. Catches future
        /// refactors that break the writer-task pipeline (e.g.
        /// re-introducing the per-peer mutex, queue capacity regressions,
        /// reconnect-race state clobbering) without breaking the encode
        /// helper in isolation.
        #[tokio::test]
        async fn sync_channel_send_writer_task_delivers_concurrent_batches() {
            use crate::storage::sync_channel::SyncChannel;

            let _lock = TEST_LOCK.lock().await;
            let (endpoint_a, endpoint_b, conn_b_to_a, conn_a_from_b) = connected_pair().await;

            // B-side coordinator wired to a real transport so the
            // receive-side dispatch (which generates response messages
            // for any DeltaSync entries within the batch) doesn't fail
            // at the response-send tail.
            let peer_id_a = conn_b_to_a.remote_id();
            let coord_b = Arc::new(make_coordinator_with_transport(Arc::new(
                PinnedConnTransport {
                    peer_id: peer_id_a,
                    conn: conn_b_to_a.clone(),
                },
            )));
            let transport_b: Arc<dyn SyncTransport> = Arc::new(PinnedConnTransport {
                peer_id: peer_id_a,
                conn: conn_b_to_a.clone(),
            });

            // B opens the SyncChannel — the writer task is now the
            // source of truth for the send side; user `send()` calls
            // enqueue into the mpsc.
            let channel = Arc::new(
                SyncChannel::connect(transport_b, peer_id_a, Arc::clone(&coord_b))
                    .await
                    .expect("SyncChannel::connect on B"),
            );

            // Materialise the bi-stream on the wire so A's accept_bi
            // returns, then accept on A and spawn the dispatch loop.
            channel
                .send(&SyncBatch::new())
                .await
                .expect("initial materialiser send");

            let (send_a, recv_a) = conn_a_from_b
                .accept_bi()
                .await
                .expect("accept_bi on A side");
            let peer_id_b = conn_a_from_b.remote_id();
            let coord_a = Arc::new(make_coordinator_with_transport(Arc::new(
                PinnedConnTransport {
                    peer_id: peer_id_b,
                    conn: conn_a_from_b.clone(),
                },
            )));
            let coord_a_for_task = Arc::clone(&coord_a);
            let dispatch_task = tokio::spawn(async move {
                let _ = coord_a_for_task
                    .handle_incoming_sync_stream(peer_id_b, send_a, recv_a)
                    .await;
            });

            // Concurrently `send(batch)` from N tasks. Each batch carries
            // one DeltaSync entry with a unique doc_key so the writer
            // task's interleaving is observable in coord_a's store after
            // all sends complete.
            const N_SENDERS: usize = 8;
            let mut sender_handles = Vec::with_capacity(N_SENDERS);
            for i in 0..N_SENDERS {
                let channel = Arc::clone(&channel);
                sender_handles.push(tokio::spawn(async move {
                    let mut remote_doc = automerge::Automerge::new();
                    let mut tx = remote_doc.transaction();
                    tx.put(automerge::ROOT, "field", format!("concurrent-send-pin-{i}"))
                        .expect("put on root");
                    tx.commit();
                    let mut remote_state = automerge::sync::State::new();
                    let sync_msg = automerge::sync::SyncDoc::generate_sync_message(
                        &remote_doc,
                        &mut remote_state,
                    )
                    .expect("fresh remote doc must produce an initial sync message");

                    let mut batch = SyncBatch::new();
                    batch.entries.push(SyncEntry::new(
                        format!("nodes:concurrent-send-pin-{i}"),
                        SyncMessageType::DeltaSync,
                        sync_msg.encode(),
                    ));
                    channel
                        .send(&batch)
                        .await
                        .expect("concurrent channel.send must succeed");
                }));
            }
            for h in sender_handles {
                h.await.expect("sender task must not panic");
            }

            // Close the channel so the writer drains and finishes the
            // SendStream; the dispatch loop on A's side then observes
            // the graceful end-of-stream and returns.
            channel.close().await;
            let _ = tokio::time::timeout(Duration::from_secs(10), dispatch_task).await;

            // Every concurrent send's DeltaSync entry must have
            // materialised in coord_a's store. Plus the initial empty
            // batch from `materialiser send` (no doc, so not asserted).
            for i in 0..N_SENDERS {
                let key = format!("nodes:concurrent-send-pin-{i}");
                let stored = coord_a.store.get(&key).expect("get doc_key");
                assert!(
                    stored.is_some(),
                    "concurrent SyncChannel::send #{i} must materialise on the receiver — \
                     missing doc_key '{key}'"
                );
            }

            // Counter must advance at least by the per-entry bytes plus
            // the outer batch frame for each of the N concurrent sends
            // (plus the materialiser empty batch). A bound rather than
            // exact equality because per-DeltaSync inner accounting
            // varies with sync-message size.
            let observed = coord_a
                .total_bytes_received
                .load(std::sync::atomic::Ordering::Relaxed);
            assert!(
                observed > 0,
                "coord_a.total_bytes_received must advance from concurrent sends; observed={observed}"
            );

            endpoint_a.close().await;
            endpoint_b.close().await;
        }
    }

    // ---- peat-mesh#236 FullHistory bandwidth-floor regression ---------------

    /// Regression for the FullHistory branch of `prepare_doc_for_sync` and
    /// the FULL_HISTORY_BANDWIDTH_FLOOR constant (peat-mesh#236).
    ///
    /// (a) A document in a FullHistory collection (`commands`) must come back
    ///     with `doc_bytes == None` — the O(N) `doc.save()` is skipped.
    /// (b) A document in a LatestOnly collection (`beacons`) must come back
    ///     with `doc_bytes == Some(_)` — the snapshot is present.
    ///
    /// The permit_size computation in `send_prepared_to_peer` is covered
    /// by the assertion on `doc_bytes` — if the save is skipped (`None`),
    /// the bandwidth-floor constant (`FULL_HISTORY_BANDWIDTH_FLOOR`) is
    /// used; if the save ran (`Some`), `b.len()` is used.  Both paths are
    /// exercised here.
    #[cfg(feature = "automerge-backend")]
    #[tokio::test]
    async fn full_history_prepare_skips_save_and_uses_bandwidth_floor() {
        use crate::storage::automerge_store::AutomergeStore;
        use crate::storage::flow_control::{FlowControlConfig, FlowController};
        use crate::storage::negentropy_sync::NegentropySync;
        use crate::storage::partition_detection::PartitionDetector;
        use crate::storage::sync_errors::SyncErrorHandler;
        use automerge::transaction::Transactable;
        use std::sync::atomic::AtomicU64;
        use tokio::sync::Notify;

        // Minimal no-op transport: never returns a connection.
        struct NoOpTransport;
        #[async_trait::async_trait]
        impl SyncTransport for NoOpTransport {
            fn get_connection(&self, _peer_id: &EndpointId) -> Option<Connection> {
                None
            }
            fn connected_peers(&self) -> Vec<EndpointId> {
                Vec::new()
            }
        }

        let store = Arc::new(AutomergeStore::in_memory());

        let coordinator = AutomergeSyncCoordinator {
            store: store.clone(),
            transport: Arc::new(NoOpTransport),
            peer_states: Arc::new(RwLock::new(HashMap::new())),
            peer_stats: Arc::new(RwLock::new(HashMap::new())),
            total_bytes_sent: Arc::new(AtomicU64::new(0)),
            total_bytes_received: Arc::new(AtomicU64::new(0)),
            error_handler: Arc::new(SyncErrorHandler::default()),
            partition_detector: Arc::new(PartitionDetector::new()),
            flow_controller: Arc::new(FlowController::with_config(FlowControlConfig::default())),
            sync_mode_registry: Arc::new(crate::qos::SyncModeRegistry::with_defaults()),
            next_batch_id: Arc::new(AtomicU64::new(1)),
            sync_router: None,
            channel_manager: Arc::new(RwLock::new(None)),
            negentropy_sync: Arc::new(NegentropySync::new()),
            ttl_manager: Arc::new(RwLock::new(None)),
            bandwidth_allocation: Arc::new(RwLock::new(None)),
            lamport_clock: Arc::new(RwLock::new(None)),
            fanout: Arc::new(FanoutState {
                inner: Mutex::new(FanoutQueue::default()),
                notify: Notify::new(),
            }),
        };

        // Write a doc into a FullHistory collection ("commands") and into a
        // LatestOnly collection ("beacons").
        let mut doc_fh = automerge::Automerge::new();
        doc_fh
            .transact::<_, _, automerge::AutomergeError>(|tx| {
                tx.put(automerge::ROOT, "order", "advance")?;
                Ok(())
            })
            .unwrap();
        store.put("commands:cmd-1", &doc_fh).unwrap();

        let mut doc_lo = automerge::Automerge::new();
        doc_lo
            .transact::<_, _, automerge::AutomergeError>(|tx| {
                tx.put(automerge::ROOT, "lat", 37.0f64)?;
                Ok(())
            })
            .unwrap();
        store.put("beacons:node-1", &doc_lo).unwrap();

        // (a) FullHistory: doc.save() must be skipped → doc_bytes is None.
        let prepared_fh = coordinator
            .prepare_doc_for_sync("commands:cmd-1")
            .expect("prepare must succeed")
            .expect("doc is present — must return Some(PreparedDoc)");
        assert_eq!(
            prepared_fh.sync_mode,
            crate::qos::SyncMode::FullHistory,
            "commands collection must be FullHistory"
        );
        assert!(
            prepared_fh.doc_bytes.is_none(),
            "FullHistory prepare must skip doc.save() — doc_bytes must be None (peat-mesh#236)"
        );

        // (b) LatestOnly: doc.save() must run → doc_bytes is Some.
        let prepared_lo = coordinator
            .prepare_doc_for_sync("beacons:node-1")
            .expect("prepare must succeed")
            .expect("doc is present — must return Some(PreparedDoc)");
        assert_eq!(
            prepared_lo.sync_mode,
            crate::qos::SyncMode::LatestOnly,
            "beacons collection must be LatestOnly"
        );
        assert!(
            prepared_lo.doc_bytes.is_some(),
            "LatestOnly prepare must run doc.save() — doc_bytes must be Some(_)"
        );
    }
}
