research: two-stage RBF pre-flight for divergent replicas [arxiv:2510.27614]

Adds a Rateless Bloom Filter + residual IBLT hybrid for the sync pre-flight, implementing the divergent-replica reconciliation scheme from arXiv 2510.27614 (Silva Gomes & Baquero, 2025). When the local revision count exceeds RBF_REV_COUNT_THRESHOLD (200) — the heuristic for initial clones and post-partition reconnects — the client now ships a small RBF + small residual IBLT instead of an over-provisioned single-stage IBLT. The receiver partitions its key set by the Bloom filter, builds the candidate-intersection residual IBLT, and subtracts the client bundle to recover missing_from_remote in one round trip. Wire-format change is strictly additive: two new SyncMessage variants (RbfRequest, RbfResponse) are appended after SealedFileAck; all existing discriminants are preserved (regression test added). Onion diff and transport framing are unchanged. Files modified / added: - crates/clawsync-onion/src/rbf.rs (new module, 30 unit tests + 3 proptests) - crates/clawsync-onion/src/manifest.rs (RbfManifest bundle type + 6 tests) - crates/clawsync-onion/src/lib.rs (re-exports) - crates/clawsync-transport/src/protocol.rs (RbfRequest/RbfResponse variants + 3 roundtrip tests) - crates/clawsync-cli/src/main.rs (heuristic switch in client push path + server dispatch) - crates/clawsync-agent/src/backend.rs (TCP/QUIC/SSH push paths opt into RBF) - CHANGELOG.md (Unreleased entry) - crates/clawsync-onion/proptest-regressions/rbf.txt (seed file) Gates: - cargo check --workspace: clean - cargo clippy -p (affected crates) --lib --bins --tests -- -D warnings: clean - cargo test --workspace: 704 passed / 0 failed (87 in clawsync-onion incl. all RBF tests)
2026-05-19 07:46:56 -07:00
33 changed files with 1527 additions and 534 deletions
@@ -7,6 +7,21 @@ Versioning follows [Semantic Versioning](https://semver.org/spec/v2.0.0.html).

 ---

+## [Unreleased]
+
+### Added
+
+- **Two-stage RBF pre-flight** for divergent-replica reconciliation
+  (`clawsync-onion::rbf`, `clawsync-onion::manifest::RbfManifest`,
+  `SyncMessage::RbfRequest` / `RbfResponse`).  When the local revision count
+  exceeds 200 (heuristic match for initial clones / post-partition reconnects),
+  the client ships a Rateless Bloom Filter + small residual IBLT bundle
+  instead of a single over-provisioned IBLT.  Implements the hybrid set
+  reconciliation scheme from arXiv 2510.27614 (Silva Gomes & Baquero, 2025).
+  Wire-format change is additive (new `SyncMessage` variants appended at end
+  of the enum); falls back transparently to plain IBLT when the heuristic
+  doesn't trigger.
+
 ## [0.1.0] — 2026-04-06

 Initial release of the ClawSync workspace.
@@ -33,14 +33,9 @@ pub enum MergeStrategy {
 /// Public summary of a branch.
 #[derive(Debug, Clone)]
 pub struct BranchInfo {
-    /// Unique numeric branch identifier (`0` = `main`).
    pub id: u32,
-    /// Human-readable branch name.
    pub name: String,
-    /// Revision number of the current HEAD on this branch.
    pub head_rev: u64,
-    /// Revision at which this branch was forked from its parent;
-    /// [`crate::format::NO_PARENT`] for the root (`main`) branch.
    pub fork_rev: u64,
 }

@@ -32,11 +32,8 @@ pub enum OnionError {
    /// Page data BLAKE3 hash does not match stored hash — integrity failure.
    #[error("BLAKE3 hash mismatch on revision {revision}: stored {stored}, computed {computed}")]
    HashMismatch {
-        /// The revision number whose page data failed the integrity check.
        revision: u64,
-        /// The BLAKE3 hex string recorded in the revision index.
        stored: String,
-        /// The BLAKE3 hex string computed from the actual page bytes.
        computed: String,
    },

@@ -50,12 +47,7 @@ pub enum OnionError {

    /// A merge was attempted on a branch with no common ancestor.
    #[error("no common ancestor found between branches {a:?} and {b:?}")]
-    NoCommonAncestor {
-        /// Name of the first branch in the attempted merge.
-        a: String,
-        /// Name of the second branch in the attempted merge.
-        b: String,
-    },
+    NoCommonAncestor { a: String, b: String },

    /// Compression or decompression failed.
    #[error("compression error: {0}")]
@@ -47,25 +47,13 @@ pub const HEADER_SIZE: usize = 128;
 // Feature flags
 // ─────────────────────────────────────────────────────────────────────────────

-/// Bit-flag constants for `OnionHeader::feature_flags`.
-///
-/// Each flag is a single bit; combine with `|` when constructing a header.
 pub mod feature_flags {
-    /// Page data may be stored with a non-`None` [`super::Codec`] (zstd, lz4, etc.).
    pub const COMPRESSION: u64 = 1 << 0;
-    /// The file uses multi-branch history (`BranchEntry` table is populated).
    pub const BRANCHING: u64 = 1 << 1;
-    /// `RevisionEntry.session_uuid` and `annotation_off` fields are meaningful.
    pub const PROVENANCE: u64 = 1 << 2;
-    /// Full-file snapshots are stored in the revision index
-    /// (i.e., at least one `RevisionEntry` has `REV_FLAG_SNAPSHOT` set).
    pub const SNAPSHOTS: u64 = 1 << 3;
 }

-/// Feature flags enabled by default when creating a new `.onion` file.
-///
-/// Includes [`feature_flags::COMPRESSION`], [`feature_flags::BRANCHING`], and
-/// [`feature_flags::PROVENANCE`].  [`feature_flags::SNAPSHOTS`] is opt-in.
 pub const DEFAULT_FEATURE_FLAGS: u64 =
    feature_flags::COMPRESSION | feature_flags::BRANCHING | feature_flags::PROVENANCE;

@@ -73,20 +61,12 @@ pub const DEFAULT_FEATURE_FLAGS: u64 =
 // Codec
 // ─────────────────────────────────────────────────────────────────────────────

-/// Compression codec used for page data storage.
-///
-/// Stored as a single byte (`u8`) in each [`PageTableEntry`].  Unknown values
-/// cause [`OnionError::UnknownCodec`] when reading.
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 #[repr(u8)]
 pub enum Codec {
-    /// No compression — page bytes are stored verbatim.
    None = 0,
-    /// [Zstandard](https://facebook.github.io/zstd/) compression.
    Zstd = 1,
-    /// [LZ4](https://lz4.github.io/lz4/) compression.
    Lz4 = 2,
-    /// [Brotli](https://github.com/google/brotli) compression.
    Brotli = 3,
    /// zstd applied after TDT byte-interleaving transform (arXiv:2506.18062).
    /// Improves compression ratio ~16% for `f32`/`f16` numeric pages.
@@ -94,22 +74,6 @@ pub enum Codec {
 }

 impl Codec {
-    /// Parse a raw codec byte from a [`PageTableEntry`] into a [`Codec`].
-    ///
-    /// # Errors
-    ///
-    /// Returns [`OnionError::UnknownCodec`] if `v` does not correspond to a
-    /// known codec variant.
-    ///
-    /// # Examples
-    ///
-    /// ```
-    /// use clawhdf5_onion::format::Codec;
-    ///
-    /// assert_eq!(Codec::from_u8(0).unwrap(), Codec::None);
-    /// assert_eq!(Codec::from_u8(1).unwrap(), Codec::Zstd);
-    /// assert!(Codec::from_u8(99).is_err());
-    /// ```
    pub fn from_u8(v: u8) -> Result<Self, OnionError> {
        match v {
            0 => Ok(Codec::None),
@@ -141,52 +105,27 @@ impl Codec {
 //   [72..128) reserved     [u8; 56]
 // ─────────────────────────────────────────────────────────────────────────────

-/// Fixed 128-byte file header at offset 0 of every `.onion` sidecar.
-///
-/// All fields are little-endian.  Readers must call [`OnionHeader::validate`]
-/// before trusting any other field.
 #[derive(FromBytes, IntoBytes, KnownLayout, Immutable, Debug, Clone)]
 #[repr(C)]
 pub struct OnionHeader {
-    /// Magic bytes — must equal [`MAGIC`] (`b"CLAWONION"`).
    pub magic: [u8; 9],
-    /// Format version — must equal [`FORMAT_VERSION`] (`1`).
    pub format_version: u8,
-    /// Alignment padding to the next `u64` boundary; always `[0u8; 6]`.
    pub _pad_align: [u8; 6],
-    /// Bitmask of enabled features; see [`feature_flags`].
    pub feature_flags: u64,
-    /// Page size in bytes used for all page data in this file; must be a
-    /// non-zero power of two.
    pub page_size: u32,
-    /// Padding after `page_size` to keep `revision_count` at a `u64` boundary.
    pub _pad_ps: [u8; 4],
-    /// Total number of `RevisionEntry` records in the index.
    pub revision_count: u64,
-    /// Number of `BranchEntry` records; `0` when the file predates branching.
    pub branch_count: u32,
-    /// Padding after `branch_count` to keep `index_offset` at a `u64` boundary.
    pub _pad_bc: [u8; 4],
-    /// Byte offset of the first `RevisionEntry` in the file.
    pub index_offset: u64,
-    /// Byte offset of the first `BranchEntry`; `0` when `branch_count == 0`.
    pub branch_offset: u64,
-    /// Unix timestamp (seconds since epoch, as `f64`) when the file was created.
    pub created_at: f64,
-    /// Reserved bytes, always zero; available for future format extensions.
    pub reserved: [u8; 56],
 }

 const _: () = assert!(size_of::<OnionHeader>() == HEADER_SIZE);

 impl OnionHeader {
-    /// Validate the header's magic bytes and format version.
-    ///
-    /// # Errors
-    ///
-    /// Returns [`OnionError::InvalidMagic`] if the magic bytes do not match
-    /// [`MAGIC`], or [`OnionError::UnknownVersion`] if `format_version` is not
-    /// equal to [`FORMAT_VERSION`].
    pub fn validate(&self) -> Result<(), OnionError> {
        if &self.magic != MAGIC {
            return Err(OnionError::InvalidMagic);
@@ -197,15 +136,10 @@ impl OnionHeader {
        Ok(())
    }

-    /// Return `true` if the feature bit `flag` is set in `feature_flags`.
-    ///
-    /// Use the constants in [`feature_flags`] as arguments.
    pub fn has_feature(&self, flag: u64) -> bool {
        self.feature_flags & flag != 0
    }

-    /// Construct a new header with zero revision/branch counts and
-    /// `index_offset` pointing immediately after the header.
    pub fn new(page_size: u32, feature_flags: u64, created_at: f64) -> Self {
        Self {
            magic: *MAGIC,
@@ -243,48 +177,24 @@ impl OnionHeader {
 //   [105..112) _pad_flags  [u8; 7]
 // ─────────────────────────────────────────────────────────────────────────────

-/// Fixed 112-byte record describing one committed revision.
-///
-/// Stored in the revision index array starting at [`OnionHeader::index_offset`].
-/// Sorted by `revision` in ascending order.
 #[derive(FromBytes, IntoBytes, KnownLayout, Immutable, Debug, Clone)]
 #[repr(C)]
 pub struct RevisionEntry {
-    /// Monotonically increasing revision number (0-based).
    pub revision: u64,
-    /// ID of the branch this revision belongs to; `0` = `main`.
    pub branch_id: u32,
-    /// Padding after `branch_id` to keep `parent_rev` at a `u64` boundary.
    pub _pad_bi: [u8; 4],
-    /// Revision number of the direct parent; [`NO_PARENT`] for the initial commit.
    pub parent_rev: u64,
-    /// Number of [`PageTableEntry`] records for this revision.
    pub page_count: u32,
-    /// Padding after `page_count` to keep `page_table_off` at a `u64` boundary.
    pub _pad_pc: [u8; 4],
-    /// Byte offset of the first [`PageTableEntry`] for this revision.
    pub page_table_off: u64,
-    /// Unix timestamp (seconds since epoch, as `f64`) when this revision was committed.
    pub timestamp: f64,
-    /// BLAKE3 hash of all page data in this revision (concatenated before hashing).
    pub blake3: [u8; 32],
-    /// UUID (16 bytes) identifying the session that produced this revision.
-    /// Zero-filled if [`feature_flags::PROVENANCE`] is not set.
    pub session_uuid: [u8; 16],
-    /// Byte offset into the annotation heap for this revision's annotation string;
-    /// `0` means no annotation.
    pub annotation_off: u64,
-    /// Per-revision bit flags; see `REV_FLAG_*` constants.
    pub flags: u8,
-    /// Padding after `flags`, also used by GC to store an epoch tag in bytes `[0..4]`.
    pub _pad_flags: [u8; 7],
 }

-/// Flag bit indicating that this revision is a full-state snapshot checkpoint.
-///
-/// Set in `RevisionEntry::flags`.  When this bit is set the revision stores a
-/// complete copy of all HDF5 pages (not just deltas), making it a fast restore
-/// point that does not require replaying earlier revisions.
 pub const REV_FLAG_SNAPSHOT: u8 = 1 << 0;

 /// Sentinel epoch value written into `RevisionEntry._pad_flags[0..4]` by
@@ -320,25 +230,14 @@ impl RevisionEntry {
 //   [32..40) created_at    f64
 // ─────────────────────────────────────────────────────────────────────────────

-/// Fixed 40-byte record describing one named branch.
-///
-/// Stored in the branch table starting at [`OnionHeader::branch_offset`].
-/// Branch `0` is always `main`.
 #[derive(FromBytes, IntoBytes, KnownLayout, Immutable, Debug, Clone)]
 #[repr(C)]
 pub struct BranchEntry {
-    /// Unique branch identifier; `0` = `main`.
    pub id: u32,
-    /// Padding after `id` to keep `name_off` at a `u64` boundary.
    pub _pad_id: [u8; 4],
-    /// Byte offset into the annotation heap for the branch name string.
    pub name_off: u64,
-    /// Revision number of the current HEAD on this branch.
    pub head_rev: u64,
-    /// Revision at which this branch was forked from its parent;
-    /// [`NO_PARENT`] for the root (`main`) branch.
    pub fork_rev: u64,
-    /// Unix timestamp (seconds since epoch, as `f64`) when the branch was created.
    pub created_at: f64,
 }

@@ -353,25 +252,14 @@ pub struct BranchEntry {
 //   [25..32) _pad          [u8; 7]
 // ─────────────────────────────────────────────────────────────────────────────

-/// Fixed 32-byte record mapping one HDF5 page to its stored data.
-///
-/// Each [`RevisionEntry`] is followed by `page_count` of these records in the
-/// page table.  Pages are ordered by `h5_offset` (ascending).
 #[derive(FromBytes, IntoBytes, KnownLayout, Immutable, Debug, Clone)]
 #[repr(C)]
 pub struct PageTableEntry {
-    /// Byte offset within the logical HDF5 file that this page covers.
    pub h5_offset: u64,
-    /// Byte offset within the `.onion` file where the (possibly compressed)
-    /// page data begins.
    pub data_offset: u64,
-    /// Original (uncompressed) page size in bytes.
    pub orig_size: u32,
-    /// Stored (possibly compressed) data size in bytes.
    pub data_size: u32,
-    /// Compression codec byte; parse with [`Codec::from_u8`].
    pub codec: u8,
-    /// Alignment padding; always zero.
    pub _pad: [u8; 7],
 }

@@ -125,11 +125,8 @@ impl<'a> Iterator for AncestorIter<'a> {
 /// Summary of branch head state, used by public query APIs.
 #[derive(Debug, Clone)]
 pub struct BranchHeadInfo {
-    /// Unique numeric branch identifier (`0` = `main`).
    pub branch_id: u32,
-    /// Revision number of the current HEAD on this branch.
    pub head_rev: u64,
-    /// Total number of revisions on this branch.
    pub revision_count: usize,
 }

@@ -66,24 +66,11 @@ pub struct MerkleNode {
 #[derive(Clone, Debug, PartialEq, Eq)]
 pub enum WalkStep {
    /// Subtree [rev_lo, rev_hi] is identical on both sides — skip.
-    Skip {
-        /// Lowest revision number covered by this identical subtree.
-        rev_lo: u64,
-        /// Highest revision number covered by this identical subtree.
-        rev_hi: u64,
-    },
+    Skip { rev_lo: u64, rev_hi: u64 },
    /// Subtree differs; descend into children.
-    Descend {
-        /// The tree node whose children should be diffed next.
-        node: MerkleNode,
-    },
+    Descend { node: MerkleNode },
    /// Leaf differs; this revision needs to be transferred.
-    Leaf {
-        /// The revision number that differs between the two trees.
-        revision: u64,
-        /// BLAKE3 hash of this revision's pages on the local side.
-        hash: [u8; 32],
-    },
+    Leaf { revision: u64, hash: [u8; 32] },
 }

 /// Balanced binary Merkle tree over OnionFile revision BLAKE3 hashes.
@@ -352,27 +339,17 @@ fn node_hash(left: &[u8; 32], right: &[u8; 32]) -> [u8; 32] {
 // Error type
 // ─────────────────────────────────────────────────────────────────────────────

-/// Errors from Merkle tree serialisation/deserialisation.
 /// Errors from Merkle tree serialisation/deserialisation.
 #[derive(Debug, thiserror::Error, PartialEq, Eq)]
 pub enum MerkleError {
-    /// The byte slice is too short to contain even the 13-byte header.
    #[error("truncated header (< 13 bytes)")]
    TruncatedHeader,
-    /// The first 4 bytes do not match the expected magic sequence.
    #[error("bad magic bytes")]
    BadMagic,
-    /// The format version byte is not supported by this reader.
    #[error("unknown format version {0}")]
    UnknownVersion(u8),
-    /// The payload is shorter than the length declared in the header.
    #[error("truncated payload: expected {expected} bytes, got {got}")]
-    TruncatedPayload {
-        /// Number of payload bytes declared in the header.
-        expected: usize,
-        /// Number of bytes actually available.
-        got: usize,
-    },
+    TruncatedPayload { expected: usize, got: usize },
 }

 // ─────────────────────────────────────────────────────────────────────────────
@@ -446,9 +446,6 @@ impl OnionFile {
        self.branches.iter().find(|b| b.id == id)
    }

-    /// Look up a `BranchEntry` by its human-readable name.
-    ///
-    /// Returns `None` if no branch with that name exists.
    pub fn branch_by_name(&self, name: &str) -> Option<&BranchEntry> {
        self.branches
            .iter()
@@ -677,19 +674,12 @@ impl OnionFile {
 /// Human-readable summary of a single revision, returned by [`OnionFile::list_revisions`].
 #[derive(Debug, Clone)]
 pub struct RevisionSummary {
-    /// Monotonically increasing revision number (0-based).
    pub revision: u64,
-    /// ID of the branch this revision belongs to; `0` = `main`.
    pub branch_id: u32,
-    /// Revision number of the direct parent; `u64::MAX` for the initial commit.
    pub parent_rev: u64,
-    /// Unix timestamp (seconds since epoch, as `f64`) when the revision was committed.
    pub timestamp: f64,
-    /// Number of page entries recorded for this revision.
    pub page_count: u32,
-    /// Optional human-readable annotation stored with the revision.
    pub annotation: Option<String>,
-    /// BLAKE3 hex digest of this revision's page data.
    pub blake3_hex: String,
    /// `true` if this revision is a full-state snapshot checkpoint.
    pub is_snapshot: bool,
@@ -11,7 +11,8 @@ use std::sync::Arc;
 use clawhdf5_onion::writer::OnionFile;
 use clawsync_onion::differ::packets_for_revisions;
 use clawsync_onion::iblt::{IBLT_SYNC_SEED, IbltSketch};
-use clawsync_onion::manifest::{ClawSyncManifest, IbltManifest};
+use clawsync_onion::manifest::{ClawSyncManifest, IbltManifest, RbfManifest};
+use clawsync_onion::rbf::{RBF_SYNC_SEED, should_use_rbf};
 use clawsync_onion::merger::merge_packets;
 use clawsync_onion::selector::SyncSelector;
 use clawsync_transport::protocol::SyncMessage;
@@ -34,11 +35,8 @@ use crate::error::AgentSyncError;
 /// Statistics from a push or pull operation.
 #[derive(Debug, Default, Clone)]
 pub struct SyncStats {
-    /// Number of revisions actually sent or received.
    pub revisions_transferred: u64,
-    /// Total compressed bytes sent or received.
    pub bytes_transferred: u64,
-    /// Revisions that were already present on the remote and therefore skipped.
    pub revisions_skipped: u64,
 }

@@ -98,10 +96,6 @@ pub struct TcpSyncBackend {
 }

 impl TcpSyncBackend {
-    /// Create a new backend targeting `remote_addr`.
-    ///
-    /// `agent_id` is embedded in all protocol messages to identify the local
-    /// agent to the remote server.
    pub fn new(remote_addr: SocketAddr, agent_id: impl Into<String>) -> Self {
        Self {
            remote_addr,
@@ -126,34 +120,59 @@ impl SyncBackend for TcpSyncBackend {
            .await
            .map_err(AgentSyncError::Transport)?;

-        // ── IBLT pre-flight ───────────────────────────────────────────────
-        // `file_blake3` is informational metadata not validated during IBLT
-        // pre-flight (server also sets it to zeros), so we skip the O(file_size)
-        // BLAKE3 read of the entire HDF5 base.
-        let sketch = IbltSketch::from_keys(&local_rev_numbers, IBLT_SYNC_SEED);
-        let iblt_manifest = IbltManifest {
-            agent_id: self.agent_id.clone(),
-            file_blake3: [0u8; 32],
-            revision_count: local_rev_numbers.len() as u64,
-            head_revision: local_rev_numbers.last().copied().unwrap_or(0),
-            head_blake3: [0u8; 32],
-            last_write: 0.0,
-            sketch_cells: sketch.cell_count() as u32,
-            sketch: sketch.to_bytes(),
-        };
-        conn.send(&SyncMessage::IbltRequest {
-            sketch: iblt_manifest,
-        })
-        .await
-        .map_err(AgentSyncError::Transport)?;
-
-        let missing_from_remote = match conn.recv().await.map_err(AgentSyncError::Transport)? {
-            SyncMessage::IbltResponse {
-                missing_from_remote,
-                ..
-            } => missing_from_remote,
-            SyncMessage::Error { message } => return Err(AgentSyncError::Remote(message)),
-            other => return Err(AgentSyncError::Protocol(format!("unexpected: {other:?}"))),
+        // ── Pre-flight: IBLT (single-stage) or RBF (two-stage) ───────────
+        // `file_blake3` is informational metadata not validated during the
+        // pre-flight (server also sets it to zeros), so we skip the
+        // O(file_size) BLAKE3 read of the entire HDF5 base.  When the local
+        // revision count is large (initial clones, post-partition reconnects)
+        // we promote to the RBF hybrid (arXiv 2510.27614) which uses fewer
+        // wire bytes than over-provisioning a single IBLT.
+        let missing_from_remote = if should_use_rbf(local_rev_numbers.len(), None) {
+            let bundle = RbfManifest::from_keys(
+                &self.agent_id,
+                &local_rev_numbers,
+                RBF_SYNC_SEED,
+            );
+            conn.send(&SyncMessage::RbfRequest { bundle })
+                .await
+                .map_err(AgentSyncError::Transport)?;
+            match conn.recv().await.map_err(AgentSyncError::Transport)? {
+                SyncMessage::RbfResponse {
+                    missing_from_remote,
+                    ..
+                } => missing_from_remote,
+                SyncMessage::Error { message } => return Err(AgentSyncError::Remote(message)),
+                other => {
+                    return Err(AgentSyncError::Protocol(format!("unexpected: {other:?}")));
+                }
+            }
+        } else {
+            let sketch = IbltSketch::from_keys(&local_rev_numbers, IBLT_SYNC_SEED);
+            let iblt_manifest = IbltManifest {
+                agent_id: self.agent_id.clone(),
+                file_blake3: [0u8; 32],
+                revision_count: local_rev_numbers.len() as u64,
+                head_revision: local_rev_numbers.last().copied().unwrap_or(0),
+                head_blake3: [0u8; 32],
+                last_write: 0.0,
+                sketch_cells: sketch.cell_count() as u32,
+                sketch: sketch.to_bytes(),
+            };
+            conn.send(&SyncMessage::IbltRequest {
+                sketch: iblt_manifest,
+            })
+            .await
+            .map_err(AgentSyncError::Transport)?;
+            match conn.recv().await.map_err(AgentSyncError::Transport)? {
+                SyncMessage::IbltResponse {
+                    missing_from_remote,
+                    ..
+                } => missing_from_remote,
+                SyncMessage::Error { message } => return Err(AgentSyncError::Remote(message)),
+                other => {
+                    return Err(AgentSyncError::Protocol(format!("unexpected: {other:?}")));
+                }
+            }
        };

        let all_packets = packets_for_revisions(&local_onion, &missing_from_remote)
@@ -358,31 +377,53 @@ impl SyncBackend for QuicSyncBackend {
            .await
            .map_err(AgentSyncError::Transport)?;

-        // ── IBLT pre-flight ───────────────────────────────────────────────
-        let sketch = IbltSketch::from_keys(&local_rev_numbers, IBLT_SYNC_SEED);
-        let iblt_manifest = IbltManifest {
-            agent_id: self.agent_id.clone(),
-            file_blake3: [0u8; 32], // not validated during IBLT pre-flight
-            revision_count: local_rev_numbers.len() as u64,
-            head_revision: local_rev_numbers.last().copied().unwrap_or(0),
-            head_blake3: [0u8; 32],
-            last_write: 0.0,
-            sketch_cells: sketch.cell_count() as u32,
-            sketch: sketch.to_bytes(),
-        };
-        conn.send(&SyncMessage::IbltRequest {
-            sketch: iblt_manifest,
-        })
-        .await
-        .map_err(AgentSyncError::Transport)?;
-
-        let missing_from_remote = match conn.recv().await.map_err(AgentSyncError::Transport)? {
-            SyncMessage::IbltResponse {
-                missing_from_remote,
-                ..
-            } => missing_from_remote,
-            SyncMessage::Error { message } => return Err(AgentSyncError::Remote(message)),
-            other => return Err(AgentSyncError::Protocol(format!("unexpected: {other:?}"))),
+        // ── Pre-flight (IBLT or RBF hybrid for high divergence) ──────────
+        let missing_from_remote = if should_use_rbf(local_rev_numbers.len(), None) {
+            let bundle = RbfManifest::from_keys(
+                &self.agent_id,
+                &local_rev_numbers,
+                RBF_SYNC_SEED,
+            );
+            conn.send(&SyncMessage::RbfRequest { bundle })
+                .await
+                .map_err(AgentSyncError::Transport)?;
+            match conn.recv().await.map_err(AgentSyncError::Transport)? {
+                SyncMessage::RbfResponse {
+                    missing_from_remote,
+                    ..
+                } => missing_from_remote,
+                SyncMessage::Error { message } => return Err(AgentSyncError::Remote(message)),
+                other => {
+                    return Err(AgentSyncError::Protocol(format!("unexpected: {other:?}")));
+                }
+            }
+        } else {
+            let sketch = IbltSketch::from_keys(&local_rev_numbers, IBLT_SYNC_SEED);
+            let iblt_manifest = IbltManifest {
+                agent_id: self.agent_id.clone(),
+                file_blake3: [0u8; 32], // not validated during IBLT pre-flight
+                revision_count: local_rev_numbers.len() as u64,
+                head_revision: local_rev_numbers.last().copied().unwrap_or(0),
+                head_blake3: [0u8; 32],
+                last_write: 0.0,
+                sketch_cells: sketch.cell_count() as u32,
+                sketch: sketch.to_bytes(),
+            };
+            conn.send(&SyncMessage::IbltRequest {
+                sketch: iblt_manifest,
+            })
+            .await
+            .map_err(AgentSyncError::Transport)?;
+            match conn.recv().await.map_err(AgentSyncError::Transport)? {
+                SyncMessage::IbltResponse {
+                    missing_from_remote,
+                    ..
+                } => missing_from_remote,
+                SyncMessage::Error { message } => return Err(AgentSyncError::Remote(message)),
+                other => {
+                    return Err(AgentSyncError::Protocol(format!("unexpected: {other:?}")));
+                }
+            }
        };

        let all_packets = packets_for_revisions(&local_onion, &missing_from_remote)
@@ -624,31 +665,53 @@ impl SyncBackend for SshSyncBackend {

        let mut peer = self.connect().await?;

-        // ── IBLT pre-flight ───────────────────────────────────────────────
-        let sketch = IbltSketch::from_keys(&local_rev_numbers, IBLT_SYNC_SEED);
-        let iblt_manifest = IbltManifest {
-            agent_id: self.agent_id.clone(),
-            file_blake3: [0u8; 32],
-            revision_count: local_rev_numbers.len() as u64,
-            head_revision: local_rev_numbers.last().copied().unwrap_or(0),
-            head_blake3: [0u8; 32],
-            last_write: 0.0,
-            sketch_cells: sketch.cell_count() as u32,
-            sketch: sketch.to_bytes(),
-        };
-        peer.send(&SyncMessage::IbltRequest {
-            sketch: iblt_manifest,
-        })
-        .await
-        .map_err(AgentSyncError::Transport)?;
-
-        let missing_from_remote = match peer.recv().await.map_err(AgentSyncError::Transport)? {
-            SyncMessage::IbltResponse {
-                missing_from_remote,
-                ..
-            } => missing_from_remote,
-            SyncMessage::Error { message } => return Err(AgentSyncError::Remote(message)),
-            other => return Err(AgentSyncError::Protocol(format!("unexpected: {other:?}"))),
+        // ── Pre-flight (IBLT or RBF hybrid for high divergence) ──────────
+        let missing_from_remote = if should_use_rbf(local_rev_numbers.len(), None) {
+            let bundle = RbfManifest::from_keys(
+                &self.agent_id,
+                &local_rev_numbers,
+                RBF_SYNC_SEED,
+            );
+            peer.send(&SyncMessage::RbfRequest { bundle })
+                .await
+                .map_err(AgentSyncError::Transport)?;
+            match peer.recv().await.map_err(AgentSyncError::Transport)? {
+                SyncMessage::RbfResponse {
+                    missing_from_remote,
+                    ..
+                } => missing_from_remote,
+                SyncMessage::Error { message } => return Err(AgentSyncError::Remote(message)),
+                other => {
+                    return Err(AgentSyncError::Protocol(format!("unexpected: {other:?}")));
+                }
+            }
+        } else {
+            let sketch = IbltSketch::from_keys(&local_rev_numbers, IBLT_SYNC_SEED);
+            let iblt_manifest = IbltManifest {
+                agent_id: self.agent_id.clone(),
+                file_blake3: [0u8; 32],
+                revision_count: local_rev_numbers.len() as u64,
+                head_revision: local_rev_numbers.last().copied().unwrap_or(0),
+                head_blake3: [0u8; 32],
+                last_write: 0.0,
+                sketch_cells: sketch.cell_count() as u32,
+                sketch: sketch.to_bytes(),
+            };
+            peer.send(&SyncMessage::IbltRequest {
+                sketch: iblt_manifest,
+            })
+            .await
+            .map_err(AgentSyncError::Transport)?;
+            match peer.recv().await.map_err(AgentSyncError::Transport)? {
+                SyncMessage::IbltResponse {
+                    missing_from_remote,
+                    ..
+                } => missing_from_remote,
+                SyncMessage::Error { message } => return Err(AgentSyncError::Remote(message)),
+                other => {
+                    return Err(AgentSyncError::Protocol(format!("unexpected: {other:?}")));
+                }
+            }
        };

        let all_packets = packets_for_revisions(&local_onion, &missing_from_remote)
@@ -2,38 +2,29 @@

 use thiserror::Error;

-/// All errors that can occur in the `clawsync-agent` autonomous sync layer.
 #[derive(Debug, Error)]
 pub enum AgentSyncError {
-    /// An error from the `clawhdf5-onion` VFD (e.g., opening or writing the sidecar).
    #[error("onion VFD error: {0}")]
    Onion(#[from] clawhdf5_onion::OnionError),

-    /// An error propagated from the `clawsync-onion` diffing/merging layer.
    #[error("sync-onion error: {0}")]
    SyncOnion(#[from] clawsync_onion::error::SyncOnionError),

-    /// An error propagated from the `clawsync-transport` layer.
    #[error("transport error: {0}")]
    Transport(#[from] clawsync_transport::error::TransportError),

-    /// An underlying I/O error.
    #[error("I/O error: {0}")]
    Io(#[from] std::io::Error),

-    /// The remote peer reported an application-level error.
    #[error("remote error: {0}")]
    Remote(String),

-    /// The remote peer violated the sync protocol.
    #[error("protocol error: {0}")]
    Protocol(String),

-    /// An error from the `clawhdf5-agent` memory layer (save, compact, WAL flush).
    #[error("agent memory error: {0}")]
    AgentMemory(String),

-    /// A restore was requested for a revision that does not exist in the sidecar.
    #[error("no revisions to restore at revision {0}")]
    RevisionNotFound(u64),
 }
@@ -37,13 +37,9 @@ pub struct OnionMemory {
 /// Summary of a historical snapshot.
 #[derive(Debug, Clone)]
 pub struct MemorySnapshot {
-    /// Revision number within the `.onion` sidecar.
    pub revision: u64,
-    /// Optional human-readable annotation stored when the snapshot was committed.
    pub annotation: Option<String>,
-    /// Unix timestamp (seconds since epoch, as `f64`) when the snapshot was committed.
    pub timestamp: f64,
-    /// BLAKE3 hex digest of the snapshotted HDF5 page data.
    pub blake3_hex: String,
 }

@@ -31,7 +31,8 @@ use clawhdf5_onion::gc::GcPolicy;
 use clawhdf5_onion::writer::OnionFile;
 use clawsync_onion::differ::{diff_revisions, packets_for_revisions};
 use clawsync_onion::iblt::{IBLT_SYNC_SEED, IbltSketch};
-use clawsync_onion::manifest::{ClawSyncManifest, IbltManifest};
+use clawsync_onion::manifest::{ClawSyncManifest, IbltManifest, RbfManifest};
+use clawsync_onion::rbf::{RBF_SYNC_SEED, should_use_rbf};
 use clawsync_onion::merger::merge_packets;
 use std::sync::Arc;

@@ -738,37 +739,64 @@ async fn cmd_push(
        .iter()
        .map(|s| s.revision)
        .collect();
-    let sketch = IbltSketch::from_keys(&local_rev_numbers, IBLT_SYNC_SEED);
-    let iblt_manifest = IbltManifest {
-        agent_id: remote_path_str.clone(),
-        file_blake3: [0u8; 32], // filled lazily; not needed for IBLT pre-flight
-        revision_count: local_rev_numbers.len() as u64,
-        head_revision: local_rev_numbers.last().copied().unwrap_or(0),
-        head_blake3: [0u8; 32],
-        last_write: 0.0,
-        sketch_cells: sketch.cell_count() as u32,
-        sketch: sketch.to_bytes(),
-    };
-    peer.send(&SyncMessage::IbltRequest {
-        sketch: iblt_manifest,
-    })
-    .await?;
+    // Two-stage RBF hybrid wins for high-divergence cases (initial clones,
+    // post-partition reconnects); single-stage IBLT remains optimal for the
+    // small-divergence common case.  See arXiv 2510.27614 §4.
+    let use_rbf = should_use_rbf(local_rev_numbers.len(), None);

-    // Receive the server's IBLT response: which revisions we should push.
-    let missing_from_remote = match peer.recv().await? {
-        SyncMessage::IbltResponse {
-            sketch: server_iblt,
-            missing_from_remote,
-        } => {
-            println!(
-                "Remote has {} revision(s) (IBLT). Pushing {} revision(s).",
-                server_iblt.revision_count,
-                missing_from_remote.len(),
-            );
-            missing_from_remote
+    let missing_from_remote = if use_rbf {
+        let bundle = RbfManifest::from_keys(
+            &remote_path_str,
+            &local_rev_numbers,
+            RBF_SYNC_SEED,
+        );
+        peer.send(&SyncMessage::RbfRequest { bundle }).await?;
+        match peer.recv().await? {
+            SyncMessage::RbfResponse {
+                bundle: server_bundle,
+                missing_from_remote,
+            } => {
+                println!(
+                    "Remote has {} revision(s) (RBF). Pushing {} revision(s).",
+                    server_bundle.revision_count,
+                    missing_from_remote.len(),
+                );
+                missing_from_remote
+            }
+            SyncMessage::Error { message } => anyhow::bail!("remote error: {message}"),
+            other => anyhow::bail!("unexpected message: {other:?}"),
+        }
+    } else {
+        let sketch = IbltSketch::from_keys(&local_rev_numbers, IBLT_SYNC_SEED);
+        let iblt_manifest = IbltManifest {
+            agent_id: remote_path_str.clone(),
+            file_blake3: [0u8; 32], // filled lazily; not needed for IBLT pre-flight
+            revision_count: local_rev_numbers.len() as u64,
+            head_revision: local_rev_numbers.last().copied().unwrap_or(0),
+            head_blake3: [0u8; 32],
+            last_write: 0.0,
+            sketch_cells: sketch.cell_count() as u32,
+            sketch: sketch.to_bytes(),
+        };
+        peer.send(&SyncMessage::IbltRequest {
+            sketch: iblt_manifest,
+        })
+        .await?;
+        match peer.recv().await? {
+            SyncMessage::IbltResponse {
+                sketch: server_iblt,
+                missing_from_remote,
+            } => {
+                println!(
+                    "Remote has {} revision(s) (IBLT). Pushing {} revision(s).",
+                    server_iblt.revision_count,
+                    missing_from_remote.len(),
+                );
+                missing_from_remote
+            }
+            SyncMessage::Error { message } => anyhow::bail!("remote error: {message}"),
+            other => anyhow::bail!("unexpected message: {other:?}"),
        }
-        SyncMessage::Error { message } => anyhow::bail!("remote error: {message}"),
-        other => anyhow::bail!("unexpected message: {other:?}"),
    };

    // Build and optionally branch-filter packets for exactly the missing revisions.
@@ -1262,7 +1290,82 @@ async fn handle_client_msg(conn: &mut SyncPeer, h5_path: &Path, msg: SyncMessage
            }
        }

-        other => anyhow::bail!("expected IbltRequest or ManifestRequest, got {other:?}"),
+        // ── RBF two-stage pre-flight (arxiv:2510.27614) ─────────────────────
+        SyncMessage::RbfRequest {
+            bundle: client_bundle,
+        } => {
+            let mut local_onion =
+                OnionFile::open(h5_path).or_else(|_| OnionFile::create_auto(h5_path))?;
+
+            let local_revisions: Vec<u64> = local_onion
+                .list_revisions()
+                .iter()
+                .map(|s| s.revision)
+                .collect();
+
+            let outcome = client_bundle
+                .reconcile_against(&local_revisions)
+                .map_err(|e| anyhow::anyhow!("RBF reconcile: {e}"))?;
+
+            // Server builds its own RBF bundle so the client could
+            // mirror-decode (e.g. for sync --pull on top of the two-stage
+            // negotiation in a future revision).
+            let server_bundle = RbfManifest::from_keys(
+                "server",
+                &local_revisions,
+                RBF_SYNC_SEED,
+            );
+
+            let missing_from_remote = outcome.missing_from_remote.clone();
+            let only_in_server = outcome.partition.certain_only_local.len()
+                + outcome.false_positives.len();
+            conn.send(&SyncMessage::RbfResponse {
+                bundle: server_bundle,
+                missing_from_remote: missing_from_remote.clone(),
+            })
+            .await?;
+            eprintln!(
+                "  RBF: client missing {only_in_server}, server missing {} revision(s).",
+                missing_from_remote.len(),
+            );
+
+            // Receive exactly the packets the client is pushing (same
+            // sub-protocol as the IbltRequest path).
+            let mut packets = Vec::new();
+            loop {
+                match conn.recv().await? {
+                    SyncMessage::LayerPacket { packet } => {
+                        let rev = packet.revision;
+                        conn.send(&SyncMessage::Ack { revision: rev }).await?;
+                        packets.push(packet);
+                    }
+                    SyncMessage::SyncComplete {
+                        revisions_transferred,
+                        bytes_transferred,
+                    } => {
+                        eprintln!(
+                            "  Client sent {revisions_transferred} revision(s),                              {bytes_transferred} bytes (RBF)."
+                        );
+                        break;
+                    }
+                    other => anyhow::bail!("unexpected: {other:?}"),
+                }
+            }
+
+            if !packets.is_empty() {
+                let stats = merge_packets(&mut local_onion, packets, true)?;
+                eprintln!(
+                    "  Merged {} revision(s), {} skipped.",
+                    stats.revisions_merged, stats.revisions_skipped
+                );
+            }
+
+            conn.close_quic();
+        }
+
+        other => {
+            anyhow::bail!("expected IbltRequest / RbfRequest / ManifestRequest, got {other:?}")
+        }
    }

    Ok(())
@@ -2386,6 +2489,17 @@ async fn handle_any_client(
                return Err(e);
            }
        }
+        SyncMessage::RbfRequest { ref bundle } => {
+            let h5_path = onion_path_for_agent(&root, &bundle.agent_id)?;
+            if let Err(e) = handle_client_msg(&mut conn, &h5_path, first_msg).await {
+                let _ = conn
+                    .send(&SyncMessage::Error {
+                        message: e.to_string(),
+                    })
+                    .await;
+                return Err(e);
+            }
+        }
        SyncMessage::ManifestRequest { ref agent_id, .. } => {
            let h5_path = onion_path_for_agent(&root, agent_id)?;
            if let Err(e) = handle_client_msg(&mut conn, &h5_path, first_msg).await {
@@ -44,17 +44,9 @@ pub fn compute_block_hashes(data: &[u8], block_size: usize) -> Vec<BlockHash> {
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub enum DeltaOp {
    /// Copy `length` bytes starting at `offset` from the source.
-    Copy {
-        /// Byte offset in the source buffer to copy from.
-        offset: usize,
-        /// Number of bytes to copy.
-        length: usize,
-    },
+    Copy { offset: usize, length: usize },
    /// Insert the given literal bytes (no corresponding source region).
-    Insert {
-        /// Literal bytes to insert at the current output position.
-        data: Vec<u8>,
-    },
+    Insert { data: Vec<u8> },
 }

 /// Compute the delta needed to transform `source` into `target`.
@@ -2,24 +2,14 @@

 use thiserror::Error;

-/// All errors that can occur in the `clawsync-core` low-level layer.
 #[derive(Debug, Error)]
 pub enum CoreError {
-    /// A compression operation (zstd, lz4, brotli, or ZstdTdt) failed.
    #[error("compression failed: {0}")]
    Compress(String),
-    /// A decompression operation failed (malformed or truncated input).
    #[error("decompression failed: {0}")]
    Decompress(String),
-    /// The BLAKE3 hash of a block does not match the stored/expected value.
    #[error("checksum mismatch: expected {expected}, got {actual}")]
-    ChecksumMismatch {
-        /// The BLAKE3 hex string that was expected.
-        expected: String,
-        /// The BLAKE3 hex string that was computed.
-        actual: String,
-    },
-    /// An underlying I/O error.
+    ChecksumMismatch { expected: String, actual: String },
    #[error("I/O error: {0}")]
    Io(#[from] std::io::Error),
 }
@@ -114,11 +114,6 @@ pub fn chunk_data_sized_simd(
 // Scalar fallback
 // ─────────────────────────────────────────────────────────────────────────────

-/// Scalar (non-SIMD) content-defined chunking fallback.
-///
-/// Splits `data` into variable-size chunks using a rolling hash with the
-/// Rabin fingerprint approach.  `min`, `avg`, and `max` are the minimum,
-/// target average, and maximum chunk sizes in bytes.
 pub fn chunk_scalar(data: &[u8], min: usize, avg: usize, max: usize) -> Vec<Chunk> {
    if data.is_empty() {
        return Vec::new();
@@ -20,7 +20,6 @@ pub enum FileDiff {
 }

 impl FileDiff {
-    /// Return the relative path for any variant.
    pub fn path(&self) -> &str {
        match self {
            Self::Added(p) | Self::Modified(p) | Self::Removed(p) | Self::Unchanged(p) => p,
@@ -1,47 +1,29 @@
-//! Error types for `clawsync-fs`.
-
 use thiserror::Error;

-/// All errors that can occur in the `clawsync-fs` filesystem sync layer.
 #[derive(Debug, Error)]
 pub enum FsSyncError {
-    /// An underlying I/O error (file read/write, directory traversal, etc.).
    #[error("I/O error: {0}")]
    Io(#[from] std::io::Error),

-    /// An error propagated from the `clawsync-transport` layer.
    #[error("transport error: {0}")]
    Transport(#[from] clawsync_transport::TransportError),

-    /// A file or directory path is invalid or could not be resolved.
    #[error("path error: {0}")]
    Path(String),

-    /// The remote peer sent an unexpected message type during an fs-sync exchange.
    #[error("protocol error: unexpected message: {0}")]
    Protocol(String),

-    /// The BLAKE3 checksum of a received file does not match the expected value.
    #[error("checksum mismatch for {path}: expected {expected}, got {actual}")]
    ChecksumMismatch {
-        /// The file path whose checksum failed verification.
        path: String,
-        /// The BLAKE3 hex string that was expected.
        expected: String,
-        /// The BLAKE3 hex string computed from the received bytes.
        actual: String,
    },

-    /// A file could not be reassembled from the received chunks.
    #[error("reconstruction failed for {path}: {reason}")]
-    Reconstruction {
-        /// The file path that failed to reconstruct.
-        path: String,
-        /// A human-readable description of the reconstruction failure.
-        reason: String,
-    },
+    Reconstruction { path: String, reason: String },

-    /// A Tokio task panicked or was cancelled during an async fs-sync operation.
    #[error("task join error: {0}")]
    Join(String),
 }
@@ -158,7 +158,6 @@ pub struct LocalEntry {

 /// In-memory manifest of a local directory tree, sorted by `rel_path`.
 pub struct FsManifest {
-    /// All file entries in the directory tree, sorted by `rel_path`.
    pub entries: Vec<LocalEntry>,
 }

@@ -56,11 +56,8 @@ pub enum ProgressEvent {
 /// fields are `None`.
 #[derive(Debug, Default)]
 pub struct SyncStats {
-    /// Number of files newly created on the destination.
    pub files_added: u32,
-    /// Number of files overwritten on the destination.
    pub files_modified: u32,
-    /// Number of files deleted from the destination.
    pub files_removed: u32,
    /// Actual bytes sent over the wire (compressed literal chunks only).
    pub bytes_transferred: u64,
@@ -376,10 +373,6 @@ pub struct FsSyncServer {
 }

 impl FsSyncServer {
-    /// Create a new push server.
-    ///
-    /// `allow_delete` controls whether the server will honour delete requests
-    /// from the client (i.e., remove files absent from the client's manifest).
    pub fn new(conn: SyncPeer, serve_root: PathBuf, excludes: GlobSet, allow_delete: bool) -> Self {
        Self {
            conn,
@@ -763,10 +756,6 @@ pub struct FsSyncPullClient {
 }

 impl FsSyncPullClient {
-    /// Create a new pull client.
-    ///
-    /// `delete` controls whether local files absent from the server's manifest
-    /// are removed after the pull completes.
    pub fn new(conn: SyncPeer, local_root: PathBuf, excludes: GlobSet, delete: bool) -> Self {
        Self {
            conn,
@@ -983,10 +972,6 @@ pub struct FsSyncPullServer {
 }

 impl FsSyncPullServer {
-    /// Create a new pull server.
-    ///
-    /// `allow_delete` controls whether the server will accept client requests
-    /// to delete files from the server's serve root during a pull session.
    pub fn new(conn: SyncPeer, serve_root: PathBuf, excludes: GlobSet, allow_delete: bool) -> Self {
        Self {
            conn,
@@ -33,7 +33,6 @@ pub struct DatasetPatch {
 /// Result of diffing two manifests.
 #[derive(Debug, Default)]
 pub struct DiffResult {
-    /// Ordered list of patches (Added, Modified, Removed) to apply to the target.
    pub patches: Vec<DatasetPatch>,
    /// Number of datasets that are identical in both manifests.
    pub unchanged_count: usize,
@@ -2,33 +2,24 @@

 use thiserror::Error;

-/// All errors that can occur in the `clawsync-hdf5` HDF5 sync layer.
 #[derive(Debug, Error)]
 pub enum Hdf5SyncError {
-    /// An error from the `clawhdf5` HDF5 parser or builder.
    #[error("HDF5 error: {0}")]
    Hdf5(#[from] clawhdf5::Error),

-    /// An underlying I/O error (file read/write, temp-file rename, etc.).
    #[error("I/O error: {0}")]
    Io(#[from] std::io::Error),

-    /// A dataset path referenced in a patch does not exist in the target file.
    #[error("dataset not found: {0}")]
    DatasetNotFound(String),

-    /// A patch attempted to overwrite a dataset with data of a different dtype.
    #[error("patch type mismatch for dataset '{path}': expected {expected}, got {actual}")]
    TypeMismatch {
-        /// HDF5 path of the mismatched dataset (e.g., `/group/weights`).
        path: String,
-        /// The dtype present in the target file.
        expected: String,
-        /// The dtype carried by the incoming patch.
        actual: String,
    },

-    /// A `DatasetManifest` could not be serialized or deserialized.
    #[error("manifest serialization error: {0}")]
    Serialize(String),
 }
@@ -24,15 +24,10 @@ use crate::error::Hdf5SyncError;
 /// Statistics from a patch operation.
 #[derive(Debug, Default, Clone)]
 pub struct PatchStats {
-    /// Number of new datasets added to the target file.
    pub datasets_added: u32,
-    /// Number of existing datasets overwritten with new data.
    pub datasets_modified: u32,
-    /// Number of datasets deleted from the target file.
    pub datasets_removed: u32,
-    /// Number of datasets copied unchanged from the old target file.
    pub datasets_unchanged: u32,
-    /// Total bytes written to the rebuilt target file.
    pub bytes_written: u64,
 }

@@ -0,0 +1,7 @@
+# Seeds for failure cases proptest has generated in the past. It is
+# automatically read and these particular cases re-run before any
+# novel cases are generated.
+#
+# It is recommended to check this file in to source control so that
+# everyone who runs the test benefits from these saved cases.
+cc cb43059b311b3d05757c4996978c93f598cd0b78bc171dba5a36d9f290c180e6 # shrinks to n_common = 0, extra_a = [187379, 187379], extra_b = []
@@ -2,54 +2,36 @@

 use thiserror::Error;

-/// All errors that can occur in the `clawsync-onion` sync layer.
 #[derive(Debug, Error)]
 pub enum SyncOnionError {
-    /// An error propagated from the underlying `clawhdf5-onion` VFD.
    #[error("onion error: {0}")]
    Onion(#[from] clawhdf5_onion::OnionError),

-    /// An error propagated from `clawsync-core` (compression, checksums, I/O).
    #[error("core error: {0}")]
    Core(#[from] clawsync_core::CoreError),

-    /// An underlying I/O error.
    #[error("I/O error: {0}")]
    Io(#[from] std::io::Error),

-    /// The requested revision number does not exist in the source `.onion` file.
    #[error("revision {0} not found in source")]
    RevisionNotFound(u64),

-    /// The named branch does not exist in the source `.onion` file.
    #[error("branch not found: {0}")]
    BranchNotFound(String),

-    /// The BLAKE3 hash of a received revision's pages does not match the expected value.
    #[error("BLAKE3 mismatch on revision {revision}: expected {expected}, got {actual}")]
    HashMismatch {
-        /// The revision number whose pages failed the integrity check.
        revision: u64,
-        /// The BLAKE3 hex string that was expected.
        expected: String,
-        /// The BLAKE3 hex string that was computed from the received pages.
        actual: String,
    },

-    /// A `LayerPacket` could not be serialized or deserialized.
    #[error("packet serialization error: {0}")]
    Serialization(String),

-    /// A merge was attempted with an empty set of packets.
    #[error("empty packet set — nothing to merge")]
    EmptyPackets,

-    /// Revisions arrived out of order during a merge — a gap was detected.
    #[error("revision gap: received {got}, expected {expected}")]
-    RevisionGap {
-        /// The next consecutive revision that was expected.
-        expected: u64,
-        /// The revision number that was actually received.
-        got: u64,
-    },
+    RevisionGap { expected: u64, got: u64 },
 }
@@ -100,16 +100,12 @@ pub enum IbltDecodeResult {
 // Error type
 // ─────────────────────────────────────────────────────────────────────────────

-/// Errors from IBLT sketch serialisation/deserialisation.
 #[derive(Debug, thiserror::Error, PartialEq, Eq)]
 pub enum IbltError {
-    /// The byte slice is too short to contain a valid sketch header or payload.
    #[error("serialised data too short")]
    Truncated,
-    /// The first bytes do not match the expected IBLT magic sequence.
    #[error("bad magic bytes")]
    BadMagic,
-    /// The format version byte is not supported by this reader.
    #[error("unknown version {0}")]
    UnknownVersion(u8),
 }
@@ -7,6 +7,9 @@
 //! - [`differ`]: compute which revisions to send (flat or Merkle tree)
 //! - [`merger`]: apply received packets to a local `OnionFile`
 //! - [`selector`]: filter revisions by branch / revision number
+//! - [`iblt`]: rateless IBLT sketches for single-stage pre-flight
+//! - [`rbf`]: Rateless Bloom Filter + residual IBLT for divergent-replica
+//!   two-stage pre-flight (arXiv 2510.27614)
 //! - [`error`] — [`SyncOnionError`]

 #![forbid(unsafe_code)]
@@ -17,6 +20,7 @@ pub mod iblt;
 pub mod manifest;
 pub mod merger;
 pub mod packet;
+pub mod rbf;
 pub mod selector;

 pub use differ::packets_for_revisions;
@@ -24,6 +28,10 @@ pub use error::SyncOnionError;
 pub use iblt::{
    DEFAULT_HASH_COUNT, IBLT_SYNC_SEED, IbltDecodeResult, IbltDiff, IbltSketch, MIN_CELLS,
 };
-pub use manifest::{ClawSyncManifest, IbltManifest};
+pub use manifest::{ClawSyncManifest, IbltManifest, RbfManifest};
 pub use merger::MergeStats;
 pub use packet::{OnionLayerPacket, OnionPage};
+pub use rbf::{
+    DEFAULT_RBF_FP_RATE, RBF_REV_COUNT_THRESHOLD, RBF_SYNC_SEED, RatelessBloomFilter,
+    RbfPartition, ServerReconcileOutcome, should_use_rbf,
+};
@@ -11,21 +11,17 @@ use rkyv::{Archive, Deserialize, Serialize};
 use clawhdf5_onion::writer::OnionFile;

 use crate::iblt::{DEFAULT_HASH_COUNT, IbltDecodeResult, IbltDiff, IbltSketch};
+use crate::rbf::{RatelessBloomFilter, ServerReconcileOutcome};

 /// A compact summary of one revision — used in the manifest.
 #[derive(Archive, Deserialize, Serialize, Debug, Clone, PartialEq)]
 pub struct RevisionSummaryPacket {
-    /// Monotonically increasing revision number (0-based).
    pub revision: u64,
-    /// ID of the branch this revision belongs to; `0` = `main`.
    pub branch_id: u32,
-    /// Revision number of the direct parent; `u64::MAX` for the initial commit.
    pub parent_rev: u64,
-    /// Unix timestamp (seconds since epoch, as `f64`) when the revision was committed.
    pub timestamp: f64,
    /// BLAKE3 hash of the pages in this revision.
    pub blake3: [u8; 32],
-    /// Optional human-readable annotation stored with the revision.
    pub annotation: Option<String>,
 }

@@ -124,17 +120,11 @@ impl ClawSyncManifest {
 /// For N=1 000: ~3 KB vs ~60 KB for `ClawSyncManifest`.
 #[derive(Archive, Deserialize, Serialize, Debug, Clone, PartialEq)]
 pub struct IbltManifest {
-    /// Logical agent / file identifier (same as in `ClawSyncManifest`).
    pub agent_id: String,
-    /// BLAKE3 hash of the raw `.h5` base file; zero-filled during pre-flight.
    pub file_blake3: [u8; 32],
-    /// Total number of revisions in the sender's sidecar.
    pub revision_count: u64,
-    /// HEAD revision number on the sender's default branch.
    pub head_revision: u64,
-    /// BLAKE3 hash of the HEAD revision's pages; zero-filled during pre-flight.
    pub head_blake3: [u8; 32],
-    /// Unix timestamp (seconds since epoch, as `f64`) of the last committed revision.
    pub last_write: f64,
    /// Serialised [`IbltSketch`] bytes.
    pub sketch: Vec<u8>,
@@ -233,6 +223,183 @@ impl IbltManifest {
    }
 }

+// ─────────────────────────────────────────────────────────────────────────────
+// RbfManifest — two-stage pre-flight (Rateless Bloom Filter + residual IBLT)
+// ─────────────────────────────────────────────────────────────────────────────
+
+/// Two-stage divergent-replica pre-flight bundle (arXiv 2510.27614).
+///
+/// The sender (typically the client) ships an [`RbfManifest`] when the
+/// expected symmetric difference is large.  It carries:
+///
+/// 1. A serialised [`RatelessBloomFilter`] (`rbf_bytes`) over the full local
+///    revision set — Stage 1.
+/// 2. A serialised [`IbltSketch`] (`residual_iblt_bytes`) **also over the
+///    full local revision set** — Stage 2.
+///
+/// The receiver runs [`crate::rbf::server_reconcile`] which partitions its
+/// own keys by the RBF, then subtracts the candidate-intersection IBLT it
+/// just built locally from the client's full-set IBLT.  This recovers
+/// `missing_from_remote` exactly, while transmitting fewer bytes than a
+/// single-stage IBLT over-provisioned for the same divergence (see paper
+/// §4).
+///
+/// Both peers must use a known seed; convention is
+/// [`crate::rbf::RBF_SYNC_SEED`].  The seed travels inside the serialised
+/// RBF and IBLT, so no extra negotiation is needed.
+#[derive(Archive, Deserialize, Serialize, Debug, Clone, PartialEq)]
+pub struct RbfManifest {
+    /// Logical identifier (file path / agent id), same convention as
+    /// [`IbltManifest`].
+    pub agent_id: String,
+    /// BLAKE3 of the underlying h5 base — informational only.
+    pub file_blake3: [u8; 32],
+    /// Total local revision count.
+    pub revision_count: u64,
+    /// HEAD revision number.
+    pub head_revision: u64,
+    /// BLAKE3 of HEAD revision pages.
+    pub head_blake3: [u8; 32],
+    /// Unix timestamp of the last commit.
+    pub last_write: f64,
+    /// Serialised [`RatelessBloomFilter`] over the full revision set.
+    pub rbf_bytes: Vec<u8>,
+    /// Serialised [`IbltSketch`] over the full revision set (residual stage).
+    pub residual_iblt_bytes: Vec<u8>,
+    /// Number of cells in the residual IBLT (denormalised for quick stats).
+    pub residual_iblt_cells: u32,
+    /// Number of bits in the RBF (denormalised for quick stats).
+    pub rbf_bits: u32,
+}
+
+impl RbfManifest {
+    /// Build the bundle from an open [`OnionFile`], the raw HDF5 base bytes,
+    /// and a shared seed.
+    ///
+    /// The residual IBLT is sized for the worst-case stragglers
+    /// (`2n+1` cells, mirroring [`IbltSketch::recommended_cells`]).
+    pub fn from_onion(agent_id: &str, onion: &OnionFile, h5_base: &[u8], seed: u64) -> Self {
+        use clawsync_core::checksum::blake3_hash;
+
+        let summaries = onion.list_revisions();
+        let revision_count = summaries.len() as u64;
+
+        let (head_revision, head_blake3) = summaries
+            .last()
+            .map(|s| {
+                (
+                    s.revision,
+                    hex_to_bytes32(&s.blake3_hex).unwrap_or([0u8; 32]),
+                )
+            })
+            .unwrap_or((0, [0u8; 32]));
+        let last_write = summaries.last().map(|s| s.timestamp).unwrap_or(0.0);
+
+        let keys: Vec<u64> = summaries.iter().map(|s| s.revision).collect();
+
+        let rbf = RatelessBloomFilter::from_keys(&keys, seed);
+        let iblt_cells = IbltSketch::recommended_cells(keys.len());
+        let mut iblt = IbltSketch::new(iblt_cells, DEFAULT_HASH_COUNT, seed);
+        for &k in &keys {
+            iblt.insert(k);
+        }
+
+        Self {
+            agent_id: agent_id.to_string(),
+            file_blake3: blake3_hash(h5_base),
+            revision_count,
+            head_revision,
+            head_blake3,
+            last_write,
+            rbf_bits: rbf.num_bits() as u32,
+            residual_iblt_cells: iblt.cell_count() as u32,
+            rbf_bytes: rbf.to_bytes(),
+            residual_iblt_bytes: iblt.to_bytes(),
+        }
+    }
+
+    /// Build a bundle directly from a key list — useful in tests and the CLI
+    /// path that already calls `list_revisions()` separately.
+    pub fn from_keys(agent_id: &str, keys: &[u64], seed: u64) -> Self {
+        let rbf = RatelessBloomFilter::from_keys(keys, seed);
+        let iblt_cells = IbltSketch::recommended_cells(keys.len());
+        let mut iblt = IbltSketch::new(iblt_cells, DEFAULT_HASH_COUNT, seed);
+        for &k in keys {
+            iblt.insert(k);
+        }
+        Self {
+            agent_id: agent_id.to_string(),
+            file_blake3: [0u8; 32],
+            revision_count: keys.len() as u64,
+            head_revision: keys.iter().copied().max().unwrap_or(0),
+            head_blake3: [0u8; 32],
+            last_write: 0.0,
+            rbf_bits: rbf.num_bits() as u32,
+            residual_iblt_cells: iblt.cell_count() as u32,
+            rbf_bytes: rbf.to_bytes(),
+            residual_iblt_bytes: iblt.to_bytes(),
+        }
+    }
+
+    /// Reconstruct the contained Rateless Bloom Filter.
+    pub fn rbf(&self) -> Result<RatelessBloomFilter, &'static str> {
+        RatelessBloomFilter::from_bytes(&self.rbf_bytes).map_err(|_| "bad RBF bytes")
+    }
+
+    /// Reconstruct the contained residual IBLT sketch.
+    pub fn residual_iblt(&self) -> Result<IbltSketch, &'static str> {
+        IbltSketch::from_bytes(&self.residual_iblt_bytes).map_err(|_| "bad residual IBLT bytes")
+    }
+
+    /// Run the receiver-side two-stage reconciliation against `local_keys`.
+    pub fn reconcile_against(
+        &self,
+        local_keys: &[u64],
+    ) -> Result<ServerReconcileOutcome, &'static str> {
+        let rbf = self.rbf()?;
+        let iblt = self.residual_iblt()?;
+        crate::rbf::server_reconcile(&rbf, &iblt, local_keys)
+    }
+
+    /// Compute the symmetric-difference diff in the same shape as
+    /// [`IbltManifest::diff_against`].
+    ///
+    /// `only_in_b` mirrors the existing API: keys the *manifest sender* has
+    /// that `local_keys` lacks.  The RBF stage can only certify
+    /// `only_in_local` (i.e. keys the sender lacks) directly; for the
+    /// `only_in_remote` half we re-use the residual IBLT subtraction
+    /// performed by [`ServerReconcileOutcome::missing_from_remote`].
+    pub fn diff_against(&self, local_keys: &[u64]) -> Result<IbltDiff, &'static str> {
+        let outcome = self.reconcile_against(local_keys)?;
+        // `certain_only_local` is keys the RBF certified absent from the
+        // sender, while `false_positives` are keys that *passed* the RBF
+        // (looked like intersection) but the residual IBLT proved the sender
+        // does not have them.  Both sets belong in `only_in_a`.
+        let mut only_in_a = outcome.partition.certain_only_local;
+        only_in_a.extend(outcome.false_positives);
+        only_in_a.sort_unstable();
+        only_in_a.dedup();
+        Ok(IbltDiff {
+            only_in_a,
+            only_in_b: outcome.missing_from_remote,
+        })
+    }
+
+    /// Serialise to rkyv bytes.
+    pub fn to_bytes(&self) -> Result<Vec<u8>, String> {
+        rkyv::to_bytes::<rkyv::rancor::Error>(self)
+            .map(|v| v.to_vec())
+            .map_err(|e| e.to_string())
+    }
+
+    /// Deserialise from rkyv bytes.
+    pub fn from_bytes(bytes: &[u8]) -> Result<Self, String> {
+        let mut aligned = rkyv::util::AlignedVec::<16>::with_capacity(bytes.len());
+        aligned.extend_from_slice(bytes);
+        rkyv::from_bytes::<RbfManifest, rkyv::rancor::Error>(&aligned).map_err(|e| e.to_string())
+    }
+}
+
 /// Parse a 64-char lowercase hex string into a `[u8; 32]`.
 fn hex_to_bytes32(hex: &str) -> Option<[u8; 32]> {
    if hex.len() != 64 {
@@ -323,4 +490,75 @@ mod tests {
        let m = ClawSyncManifest::from_onion("x", &onion, &base);
        assert_eq!(m.file_blake3, blake3_hash(&base));
    }
+
+    // ── RbfManifest tests ────────────────────────────────────────────────
+
+    #[test]
+    fn rbf_manifest_from_onion_basic() {
+        let (_h5, onion, base) = make_onion();
+        let bundle = RbfManifest::from_onion("agent-a", &onion, &base, crate::rbf::RBF_SYNC_SEED);
+        assert_eq!(bundle.agent_id, "agent-a");
+        assert_eq!(bundle.revision_count, 2);
+        assert!(bundle.residual_iblt_cells >= crate::iblt::MIN_CELLS as u32);
+        assert!(bundle.rbf_bits > 0);
+    }
+
+    #[test]
+    fn rbf_manifest_roundtrip() {
+        let bundle = RbfManifest::from_keys("agent", &[1u64, 2, 3, 4], crate::rbf::RBF_SYNC_SEED);
+        let bytes = bundle.to_bytes().unwrap();
+        let recovered = RbfManifest::from_bytes(&bytes).unwrap();
+        assert_eq!(recovered, bundle);
+    }
+
+    #[test]
+    fn rbf_manifest_reconcile_recovers_push_set() {
+        // Client has 0..50; server has 0..40 → client should push 40..50.
+        let client_keys: Vec<u64> = (0..50).collect();
+        let server_keys: Vec<u64> = (0..40).collect();
+        let bundle = RbfManifest::from_keys("c", &client_keys, crate::rbf::RBF_SYNC_SEED);
+        let outcome = bundle.reconcile_against(&server_keys).unwrap();
+        let mut got = outcome.missing_from_remote.clone();
+        got.sort_unstable();
+        let expected: Vec<u64> = (40..50).collect();
+        assert_eq!(got, expected);
+    }
+
+    #[test]
+    fn rbf_manifest_diff_against_shape_matches_iblt() {
+        let client_keys: Vec<u64> = (0..30).collect();
+        let server_keys: Vec<u64> = (10..40).collect();
+        let bundle = RbfManifest::from_keys("c", &client_keys, crate::rbf::RBF_SYNC_SEED);
+        let diff = bundle.diff_against(&server_keys).unwrap();
+
+        // sender (= client) extras  → only_in_b = 0..10
+        let mut only_in_b = diff.only_in_b.clone();
+        only_in_b.sort_unstable();
+        assert_eq!(only_in_b, (0..10).collect::<Vec<u64>>());
+
+        // server extras → only_in_a = 30..40
+        // (certified by RBF, since the client filter never saw 30..40).
+        for k in 30u64..40 {
+            assert!(diff.only_in_a.contains(&k), "missing {k} from only_in_a");
+        }
+    }
+
+    #[test]
+    fn rbf_manifest_empty_server_initial_clone() {
+        // Initial clone: client has 0..100, server has nothing.
+        let client_keys: Vec<u64> = (0..100).collect();
+        let bundle = RbfManifest::from_keys("c", &client_keys, crate::rbf::RBF_SYNC_SEED);
+        let outcome = bundle.reconcile_against(&[]).unwrap();
+        let mut got = outcome.missing_from_remote.clone();
+        got.sort_unstable();
+        assert_eq!(got, client_keys);
+    }
+
+    #[test]
+    fn rbf_manifest_already_in_sync() {
+        let keys: Vec<u64> = (0..50).collect();
+        let bundle = RbfManifest::from_keys("c", &keys, crate::rbf::RBF_SYNC_SEED);
+        let outcome = bundle.reconcile_against(&keys).unwrap();
+        assert!(outcome.missing_from_remote.is_empty());
+    }
 }
@@ -0,0 +1,844 @@
+//! Rateless Bloom Filter (RBF) + residual IBLT — first-pass sketch for
+//! divergent-replica set reconciliation.
+//!
+//! When two peers are highly divergent (large symmetric difference `d`),
+//! sizing a single IBLT requires either (a) over-provisioning the table or
+//! (b) a multi-round retry-with-larger-`m` loop.  Both waste bytes on the
+//! wire.  The hybrid two-stage scheme from the RBF paper sidesteps this:
+//!
+//! 1. **Stage 1 (RBF):** sender builds a Bloom filter over its key set `A`
+//!    and ships it.  The receiver tests each of its keys `b ∈ B` against
+//!    the filter.  Keys for which `RBF(A).contains(b) == false` are
+//!    *certain* `only_in_B` (Bloom never reports false negatives).
+//! 2. **Stage 2 (residual IBLT):** the remaining `B' ⊆ B` is the candidate
+//!    intersection — actually contains `A ∩ B` plus a small set of Bloom
+//!    false positives.  A small IBLT sized for `|A △ B'| ≈ |only_in_A| +
+//!    |FP|` recovers `only_in_A` exactly and corrects the FPs.
+//!
+//! For replica pairs with Jaccard similarity below ~0.85 (i.e. considerable
+//! divergence — initial clones, post-partition reconnects), the hybrid cuts
+//! total wire cost by more than 20 % vs. a single over-provisioned IBLT.
+//!
+//! ## Reference
+//!
+//! Silva Gomes & Baquero, *Rateless Bloom Filters: Set Reconciliation for
+//! Divergent Replicas with Variable-Sized Elements*, arXiv 2510.27614 (2025).
+//!
+//! ## Wire format (RBF)
+//!
+//! ```text
+//! "RBLF" | version(u8) | num_hashes(u8) | _pad(u16) | num_bits(u32 LE) |
+//!         num_elements(u32 LE) | seed(u64 LE) | bits[ceil(num_bits/8)]
+//! ```
+
+use xxhash_rust::xxh3::xxh3_64;
+
+use crate::iblt::{DEFAULT_HASH_COUNT, IBLT_SYNC_SEED, IbltDecodeResult, IbltSketch};
+
+// ─────────────────────────────────────────────────────────────────────────────
+// Constants
+// ─────────────────────────────────────────────────────────────────────────────
+
+/// Default target false-positive rate for the RBF first-pass filter (1 %).
+///
+/// Smaller ε grows the filter logarithmically but shrinks the expected
+/// residual IBLT proportionally.  1 % is the empirical sweet spot from the
+/// reference paper for Jaccard ∈ [0.5, 0.85].
+pub const DEFAULT_RBF_FP_RATE: f64 = 0.01;
+
+/// Heuristic threshold (in local revision count) at which the IBLT pre-flight
+/// switches to the two-stage RBF hybrid.
+///
+/// The reference paper shows the hybrid beats single-stage IBLT once
+/// divergence (or unknown divergence) is non-trivial; in clawsync the most
+/// common high-divergence scenarios are (a) initial clones and (b) reconnect
+/// after a long partition, both of which produce large local revision counts
+/// without any reliable prior on the remote.
+pub const RBF_REV_COUNT_THRESHOLD: usize = 200;
+
+/// Minimum number of bits in an RBF (guard against tiny inputs).
+pub const MIN_RBF_BITS: usize = 64;
+
+/// Maximum number of hash probes (clamps the optimum-k computation).
+pub const MAX_RBF_HASHES: u8 = 16;
+
+/// Wire-format magic bytes for a serialised RBF.
+const MAGIC: &[u8; 4] = b"RBLF";
+
+/// Wire-format version.
+const VERSION: u8 = 1;
+
+/// Golden-ratio constant used to diversify the per-probe hashes.
+const GOLDEN: u64 = 0x9e37_79b9_7f4a_7c15;
+
+// ─────────────────────────────────────────────────────────────────────────────
+// Error type
+// ─────────────────────────────────────────────────────────────────────────────
+
+/// Errors that can occur while (de)serialising or building an RBF.
+///
+/// Marked `#[non_exhaustive]` so additional failure modes can be added in
+/// future versions without breaking the public API.
+#[non_exhaustive]
+#[derive(Debug, thiserror::Error, PartialEq, Eq)]
+pub enum RbfError {
+    /// Serialised data is shorter than the fixed header.
+    #[error("serialised RBF data too short")]
+    Truncated,
+    /// Magic bytes did not match `b"RBLF"`.
+    #[error("bad RBF magic bytes")]
+    BadMagic,
+    /// Wire-format version is not understood by this build.
+    #[error("unknown RBF version {0}")]
+    UnknownVersion(u8),
+    /// `num_hashes` was zero or above [`MAX_RBF_HASHES`].
+    #[error("RBF num_hashes out of range: {0}")]
+    BadHashCount(u8),
+    /// `num_bits` was zero, larger than payload, or otherwise invalid.
+    #[error("RBF num_bits invalid: {0}")]
+    BadBitCount(u32),
+}
+
+// ─────────────────────────────────────────────────────────────────────────────
+// RatelessBloomFilter
+// ─────────────────────────────────────────────────────────────────────────────
+
+/// A Bloom filter used as the first stage of the RBF hybrid reconciliation
+/// scheme.
+///
+/// The filter is parametrised once at build time from `(n, fp_rate)` — the
+/// "rateless" property in the paper refers to the *hybrid protocol* being
+/// dynamic w.r.t. the symmetric difference, not the filter itself adapting
+/// mid-flight.  Once built, the filter is immutable on the wire.
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct RatelessBloomFilter {
+    bits: Vec<u8>,
+    num_bits: u32,
+    num_hashes: u8,
+    num_elements: u32,
+    seed: u64,
+}
+
+impl RatelessBloomFilter {
+    /// Optimum bit count `m = -n ln(ε) / (ln 2)²`, clamped to [`MIN_RBF_BITS`].
+    pub fn optimum_bits(n: usize, fp_rate: f64) -> usize {
+        if n == 0 {
+            return MIN_RBF_BITS;
+        }
+        let n = n as f64;
+        let ln2 = std::f64::consts::LN_2;
+        let raw = -(n * fp_rate.ln()) / (ln2 * ln2);
+        (raw.ceil() as usize).max(MIN_RBF_BITS)
+    }
+
+    /// Optimum number of hash probes `k = (m/n) ln 2`, clamped to
+    /// `[1, MAX_RBF_HASHES]`.
+    pub fn optimum_hashes(num_bits: usize, n: usize) -> u8 {
+        if n == 0 {
+            return 1;
+        }
+        let m = num_bits as f64;
+        let n = n as f64;
+        let k = ((m / n) * std::f64::consts::LN_2).round() as i64;
+        k.clamp(1, MAX_RBF_HASHES as i64) as u8
+    }
+
+    /// Build an empty filter sized for `n` elements at target `fp_rate`.
+    pub fn new(n: usize, fp_rate: f64, seed: u64) -> Self {
+        let num_bits = Self::optimum_bits(n, fp_rate);
+        let num_hashes = Self::optimum_hashes(num_bits, n);
+        let bytes = num_bits.div_ceil(8);
+        Self {
+            bits: vec![0u8; bytes],
+            num_bits: num_bits as u32,
+            num_hashes,
+            num_elements: 0,
+            seed,
+        }
+    }
+
+    /// Build a filter over `keys` using [`DEFAULT_RBF_FP_RATE`].
+    pub fn from_keys(keys: &[u64], seed: u64) -> Self {
+        Self::from_keys_with_fp(keys, DEFAULT_RBF_FP_RATE, seed)
+    }
+
+    /// Build a filter over `keys` with an explicit target FP rate.
+    pub fn from_keys_with_fp(keys: &[u64], fp_rate: f64, seed: u64) -> Self {
+        let mut rbf = Self::new(keys.len(), fp_rate, seed);
+        for &k in keys {
+            rbf.insert(k);
+        }
+        rbf
+    }
+
+    /// Number of bits in the filter (`m`).
+    pub fn num_bits(&self) -> usize {
+        self.num_bits as usize
+    }
+
+    /// Number of hash probes (`k`).
+    pub fn num_hashes(&self) -> u8 {
+        self.num_hashes
+    }
+
+    /// Number of inserted elements (does not double-count duplicates).
+    pub fn num_elements(&self) -> u32 {
+        self.num_elements
+    }
+
+    /// Seed used for the hash family.
+    pub fn seed(&self) -> u64 {
+        self.seed
+    }
+
+    /// Insert `key` into the filter.
+    pub fn insert(&mut self, key: u64) {
+        let was_new = self.set_bits_for(key);
+        if was_new {
+            self.num_elements = self.num_elements.saturating_add(1);
+        }
+    }
+
+    /// Test whether `key` *might* be in the filter.
+    ///
+    /// Returns `false` only if `key` is definitely absent; `true` means
+    /// `key` is either present or a Bloom false positive.
+    pub fn contains(&self, key: u64) -> bool {
+        let m = self.num_bits as usize;
+        for h in 0..self.num_hashes {
+            let idx = probe_index(key, h, self.seed, m);
+            if !self.get_bit(idx) {
+                return false;
+            }
+        }
+        true
+    }
+
+    /// Estimated current false-positive rate given the load factor.
+    ///
+    /// `(1 - e^(-kn/m))^k`.  Useful for sanity-checking sizing.
+    pub fn estimated_fp_rate(&self) -> f64 {
+        let m = self.num_bits as f64;
+        if m <= 0.0 {
+            return 1.0;
+        }
+        let k = self.num_hashes as f64;
+        let n = self.num_elements as f64;
+        let load = 1.0 - (-k * n / m).exp();
+        load.powf(k)
+    }
+
+    // ── Serialisation ────────────────────────────────────────────────────────
+
+    /// Serialise to compact bytes (see module docs for layout).
+    pub fn to_bytes(&self) -> Vec<u8> {
+        let payload_bytes = self.bits.len();
+        let mut out = Vec::with_capacity(24 + payload_bytes);
+        out.extend_from_slice(MAGIC);
+        out.push(VERSION);
+        out.push(self.num_hashes);
+        out.extend_from_slice(&0u16.to_le_bytes()); // pad
+        out.extend_from_slice(&self.num_bits.to_le_bytes());
+        out.extend_from_slice(&self.num_elements.to_le_bytes());
+        out.extend_from_slice(&self.seed.to_le_bytes());
+        out.extend_from_slice(&self.bits);
+        out
+    }
+
+    /// Deserialise from bytes.
+    pub fn from_bytes(data: &[u8]) -> Result<Self, RbfError> {
+        if data.len() < 24 {
+            return Err(RbfError::Truncated);
+        }
+        if &data[0..4] != MAGIC {
+            return Err(RbfError::BadMagic);
+        }
+        let version = data[4];
+        if version != VERSION {
+            return Err(RbfError::UnknownVersion(version));
+        }
+        let num_hashes = data[5];
+        if num_hashes == 0 || num_hashes > MAX_RBF_HASHES {
+            return Err(RbfError::BadHashCount(num_hashes));
+        }
+        // bytes 6..8 = pad
+        let num_bits = u32::from_le_bytes(data[8..12].try_into().unwrap());
+        let num_elements = u32::from_le_bytes(data[12..16].try_into().unwrap());
+        let seed = u64::from_le_bytes(data[16..24].try_into().unwrap());
+        if num_bits == 0 {
+            return Err(RbfError::BadBitCount(0));
+        }
+        let payload_bytes = (num_bits as usize).div_ceil(8);
+        if data.len() < 24 + payload_bytes {
+            return Err(RbfError::Truncated);
+        }
+        let bits = data[24..24 + payload_bytes].to_vec();
+        Ok(Self {
+            bits,
+            num_bits,
+            num_hashes,
+            num_elements,
+            seed,
+        })
+    }
+
+    // ── Internal helpers ─────────────────────────────────────────────────────
+
+    /// Set the `k` bits for `key`; return `true` iff at least one bit flipped.
+    fn set_bits_for(&mut self, key: u64) -> bool {
+        let m = self.num_bits as usize;
+        let mut any_new = false;
+        for h in 0..self.num_hashes {
+            let idx = probe_index(key, h, self.seed, m);
+            let byte = idx / 8;
+            let bit = (idx % 8) as u8;
+            let mask = 1u8 << bit;
+            let cur = self.bits[byte];
+            if cur & mask == 0 {
+                self.bits[byte] = cur | mask;
+                any_new = true;
+            }
+        }
+        any_new
+    }
+
+    #[inline]
+    fn get_bit(&self, idx: usize) -> bool {
+        let byte = idx / 8;
+        let bit = (idx % 8) as u8;
+        (self.bits[byte] >> bit) & 1 == 1
+    }
+}
+
+#[inline]
+fn probe_index(key: u64, h: u8, seed: u64, num_bits: usize) -> usize {
+    // Double-hash style: hash_h(key) = xxh3(key ⊕ (seed + h·φ))
+    let input = key ^ seed.wrapping_add((h as u64).wrapping_mul(GOLDEN));
+    xxh3_64(&input.to_le_bytes()) as usize % num_bits
+}
+
+// ─────────────────────────────────────────────────────────────────────────────
+// Two-stage reconciliation primitives
+// ─────────────────────────────────────────────────────────────────────────────
+
+/// Partition `local_keys` by an RBF received from a remote peer.
+///
+/// Given a remote filter `rbf_remote` built over the remote's key set `R`:
+///
+/// - `certain_only_local`: keys `k ∈ local_keys` that the filter is **certain**
+///   the remote lacks (Bloom never returns false negatives).
+/// - `candidate_intersection`: keys for which the filter said "maybe present"
+///   — actually in `R` or a Bloom false positive.
+///
+/// The caller feeds `candidate_intersection` into the Stage-2 residual IBLT.
+pub fn partition_against_remote(
+    rbf_remote: &RatelessBloomFilter,
+    local_keys: &[u64],
+) -> RbfPartition {
+    let mut certain_only_local = Vec::new();
+    let mut candidate_intersection = Vec::new();
+    for &k in local_keys {
+        if rbf_remote.contains(k) {
+            candidate_intersection.push(k);
+        } else {
+            certain_only_local.push(k);
+        }
+    }
+    certain_only_local.sort_unstable();
+    candidate_intersection.sort_unstable();
+    RbfPartition {
+        certain_only_local,
+        candidate_intersection,
+    }
+}
+
+/// Result of partitioning a local key set against a remote RBF.
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct RbfPartition {
+    /// Keys the remote *definitely* does not have.
+    pub certain_only_local: Vec<u64>,
+    /// Keys the remote *might* have (Bloom said yes — actually present + FPs).
+    pub candidate_intersection: Vec<u64>,
+}
+
+/// Recommended size for the Stage-2 residual IBLT.
+///
+/// Sized for the expected residual symmetric difference: `|only_in_A| +
+/// |FP|`.  We don't know `|only_in_A|` a priori on the sender side, so we
+/// over-provision modestly: at least the expected FP count from the filter,
+/// plus a safety factor of `2×` to absorb modest variance.
+pub fn residual_iblt_cells(rbf: &RatelessBloomFilter, expected_only_in_a: usize) -> usize {
+    let n = rbf.num_elements() as f64;
+    let fp_count = (n * rbf.estimated_fp_rate()).ceil() as usize;
+    let total = (expected_only_in_a + fp_count) * 2 + 1;
+    IbltSketch::recommended_cells(total)
+}
+
+/// Build the Stage-2 residual IBLT over `keys` with default sizing and seed.
+///
+/// Both peers must agree on `cells` and `seed`; convention: derive both from
+/// the RBF so they ride along automatically.
+pub fn build_residual_iblt(keys: &[u64], cells: usize, seed: u64) -> IbltSketch {
+    let mut sk = IbltSketch::new(cells, DEFAULT_HASH_COUNT, seed);
+    for &k in keys {
+        sk.insert(k);
+    }
+    sk
+}
+
+/// Two-stage hybrid reconcile on the *receiving* (server) side.
+///
+/// Inputs:
+/// - `client_rbf`: Stage-1 filter over the client's key set `A`.
+/// - `client_residual_iblt`: Stage-2 IBLT over `A` (full key set).
+/// - `local_keys`: this peer's (server's) key set `B`.
+///
+/// Output: which keys the client has that the server lacks (=
+/// `missing_from_remote` in the push direction), plus the partition info for
+/// diagnostics.  Returns `Err` if the residual IBLT is too small to decode.
+pub fn server_reconcile(
+    client_rbf: &RatelessBloomFilter,
+    client_residual_iblt: &IbltSketch,
+    local_keys: &[u64],
+) -> Result<ServerReconcileOutcome, &'static str> {
+    // Stage 1: partition server's keys by the client's RBF.
+    let partition = partition_against_remote(client_rbf, local_keys);
+
+    // Stage 2: build server's residual IBLT over the *candidate intersection*
+    // — everything the Bloom said "maybe in A".  This is the key insight: we
+    // exclude `certain_only_local` from the IBLT, shrinking the residual diff
+    // by exactly that much.
+    let cells = client_residual_iblt.cell_count();
+    let seed = client_residual_iblt.seed();
+    let mut server_sk = build_residual_iblt(&partition.candidate_intersection, cells, seed);
+    server_sk.subtract(client_residual_iblt);
+
+    // `server_sk` now encodes (B' \ A), with sign convention:
+    //   only_in_a = elements in server's IBLT not in client's = ∅ (we removed
+    //               them by partitioning), unless filter logic went weird.
+    //   only_in_b = elements in client's IBLT not in server's residual
+    //               = A \ B' = A \ B (= only_in_A — what client should push).
+    let diff = match server_sk.decode() {
+        IbltDecodeResult::Complete(d) => d,
+        IbltDecodeResult::NeedMoreCells => return Err("residual IBLT too small to decode"),
+    };
+
+    // server_sk = server_residual − client_full.
+    // After subtract, decode follows clawsync's IbltSketch convention:
+    //   diff.only_in_a = keys in server_residual not in client_full (FPs that
+    //                    "looked like A intersection" but client doesn't have)
+    //   diff.only_in_b = keys in client_full not in server_residual
+    //                    = A \ B (what client should push).
+    Ok(ServerReconcileOutcome {
+        missing_from_remote: diff.only_in_b,
+        false_positives: diff.only_in_a,
+        partition,
+    })
+}
+
+/// Outcome of [`server_reconcile`].
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct ServerReconcileOutcome {
+    /// Keys the client has that the server lacks — what the client should
+    /// push.
+    pub missing_from_remote: Vec<u64>,
+    /// Bloom false positives surfaced by the residual IBLT.  Diagnostic only;
+    /// not needed for the push direction.
+    pub false_positives: Vec<u64>,
+    /// First-stage partition (server's perspective).
+    pub partition: RbfPartition,
+}
+
+// ─────────────────────────────────────────────────────────────────────────────
+// Heuristic
+// ─────────────────────────────────────────────────────────────────────────────
+
+/// Decide whether to use the RBF two-stage hybrid for the pre-flight.
+///
+/// The heuristic matches the dossier guidance:
+///
+/// - **local revision count > [`RBF_REV_COUNT_THRESHOLD`]** → use RBF
+///   (initial-clone-shaped workload),
+/// - **expected delta > 15 % of local** → use RBF.
+///
+/// `expected_delta_ratio` may be `None` when no prior is known (e.g.
+/// post-partition reconnect): in that case we trust the count threshold.
+pub fn should_use_rbf(local_rev_count: usize, expected_delta_ratio: Option<f64>) -> bool {
+    if local_rev_count > RBF_REV_COUNT_THRESHOLD {
+        return true;
+    }
+matches!(expected_delta_ratio, Some(r) if r > 0.15)
+}
+
+/// Convenience: the canonical RBF seed shared across both peers.  Re-uses
+/// [`IBLT_SYNC_SEED`] so neither side needs an extra negotiation.
+pub const RBF_SYNC_SEED: u64 = IBLT_SYNC_SEED;
+
+// ─────────────────────────────────────────────────────────────────────────────
+// Tests
+// ─────────────────────────────────────────────────────────────────────────────
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use proptest::prelude::*;
+
+    const SEED: u64 = 0xC1A4_5CA8_1B17_0001;
+
+    // ── sizing math ──────────────────────────────────────────────────────────
+
+    #[test]
+    fn optimum_bits_grows_with_n() {
+        let m100 = RatelessBloomFilter::optimum_bits(100, 0.01);
+        let m1000 = RatelessBloomFilter::optimum_bits(1000, 0.01);
+        assert!(m1000 > m100);
+    }
+
+    #[test]
+    fn optimum_bits_grows_as_fp_shrinks() {
+        let m1 = RatelessBloomFilter::optimum_bits(1000, 0.01);
+        let m2 = RatelessBloomFilter::optimum_bits(1000, 0.001);
+        assert!(m2 > m1);
+    }
+
+    #[test]
+    fn optimum_bits_zero_n_falls_back_to_min() {
+        assert_eq!(
+            RatelessBloomFilter::optimum_bits(0, 0.01),
+            MIN_RBF_BITS,
+        );
+    }
+
+    #[test]
+    fn optimum_hashes_zero_n_clamped() {
+        assert_eq!(RatelessBloomFilter::optimum_hashes(1024, 0), 1);
+    }
+
+    #[test]
+    fn optimum_hashes_clamped_to_max() {
+        // Very generous bit budget → k would explode without clamp.
+        let k = RatelessBloomFilter::optimum_hashes(1_000_000, 1);
+        assert!(k <= MAX_RBF_HASHES);
+    }
+
+    // ── insert / contains ────────────────────────────────────────────────────
+
+    #[test]
+    fn no_false_negatives() {
+        let keys: Vec<u64> = (0..500).collect();
+        let rbf = RatelessBloomFilter::from_keys(&keys, SEED);
+        for k in &keys {
+            assert!(rbf.contains(*k), "RBF missed inserted key {k}");
+        }
+    }
+
+    #[test]
+    fn fp_rate_within_target_at_default() {
+        // Build filter for 1000 elements at 1% FP target, probe with 10_000
+        // disjoint keys, count hits.
+        let inserted: Vec<u64> = (0..1000).collect();
+        let rbf = RatelessBloomFilter::from_keys(&inserted, SEED);
+
+        let mut hits = 0usize;
+        let probes = 10_000u64;
+        for k in 10_000u64..10_000 + probes {
+            if rbf.contains(k) {
+                hits += 1;
+            }
+        }
+        let observed = hits as f64 / probes as f64;
+        // Allow up to 3× the target to absorb sampling variance.
+        assert!(
+            observed < 0.03,
+            "observed FP {observed:.4} exceeds 3× target"
+        );
+    }
+
+    #[test]
+    fn estimated_fp_rate_finite_and_in_unit() {
+        let rbf = RatelessBloomFilter::from_keys(&(0..1000).collect::<Vec<_>>(), SEED);
+        let est = rbf.estimated_fp_rate();
+        assert!(est.is_finite());
+        assert!((0.0..=1.0).contains(&est));
+    }
+
+    #[test]
+    fn empty_filter_zero_load() {
+        let rbf = RatelessBloomFilter::new(0, 0.01, SEED);
+        // No elements inserted → no bit ever set → contains is always false.
+        for k in 0u64..100 {
+            assert!(!rbf.contains(k));
+        }
+        assert_eq!(rbf.num_elements(), 0);
+    }
+
+    #[test]
+    fn duplicate_inserts_dont_double_count() {
+        let mut rbf = RatelessBloomFilter::new(10, 0.01, SEED);
+        rbf.insert(42);
+        rbf.insert(42);
+        rbf.insert(42);
+        assert_eq!(rbf.num_elements(), 1);
+    }
+
+    // ── serialisation ────────────────────────────────────────────────────────
+
+    #[test]
+    fn serialise_roundtrip() {
+        let keys: Vec<u64> = (0..200).collect();
+        let rbf = RatelessBloomFilter::from_keys(&keys, SEED);
+        let bytes = rbf.to_bytes();
+        let recovered = RatelessBloomFilter::from_bytes(&bytes).unwrap();
+        assert_eq!(rbf, recovered);
+        for k in &keys {
+            assert!(recovered.contains(*k));
+        }
+    }
+
+    #[test]
+    fn serialise_bad_magic() {
+        let mut bytes = RatelessBloomFilter::from_keys(&[1, 2, 3], SEED).to_bytes();
+        bytes[0] = 0xFF;
+        assert_eq!(
+            RatelessBloomFilter::from_bytes(&bytes),
+            Err(RbfError::BadMagic)
+        );
+    }
+
+    #[test]
+    fn serialise_truncated_header() {
+        assert_eq!(
+            RatelessBloomFilter::from_bytes(&[0u8; 10]),
+            Err(RbfError::Truncated)
+        );
+    }
+
+    #[test]
+    fn serialise_unknown_version() {
+        let mut bytes = RatelessBloomFilter::from_keys(&[1], SEED).to_bytes();
+        bytes[4] = 99;
+        assert!(matches!(
+            RatelessBloomFilter::from_bytes(&bytes),
+            Err(RbfError::UnknownVersion(99))
+        ));
+    }
+
+    #[test]
+    fn serialise_bad_hash_count() {
+        let mut bytes = RatelessBloomFilter::from_keys(&[1], SEED).to_bytes();
+        bytes[5] = 0;
+        assert!(matches!(
+            RatelessBloomFilter::from_bytes(&bytes),
+            Err(RbfError::BadHashCount(0))
+        ));
+    }
+
+    #[test]
+    fn serialise_bad_bit_count() {
+        let mut bytes = RatelessBloomFilter::from_keys(&[1], SEED).to_bytes();
+        // num_bits is at offset 8..12.
+        bytes[8] = 0;
+        bytes[9] = 0;
+        bytes[10] = 0;
+        bytes[11] = 0;
+        assert!(matches!(
+            RatelessBloomFilter::from_bytes(&bytes),
+            Err(RbfError::BadBitCount(0))
+        ));
+    }
+
+    #[test]
+    fn serialise_truncated_payload() {
+        let bytes = RatelessBloomFilter::from_keys(&(0..100).collect::<Vec<_>>(), SEED).to_bytes();
+        let short = &bytes[..bytes.len() - 4];
+        assert_eq!(
+            RatelessBloomFilter::from_bytes(short),
+            Err(RbfError::Truncated)
+        );
+    }
+
+    // ── partition / hybrid reconciliation ────────────────────────────────────
+
+    #[test]
+    fn partition_against_remote_certain_only_set() {
+        let remote_keys: Vec<u64> = (0..100).collect();
+        let local_keys: Vec<u64> = (50..150).collect();
+        let rbf = RatelessBloomFilter::from_keys(&remote_keys, SEED);
+        let part = partition_against_remote(&rbf, &local_keys);
+
+        // 100..150 are definitely not in remote → must appear in certain_only_local.
+        for k in 100u64..150 {
+            assert!(
+                part.certain_only_local.contains(&k),
+                "missing certain key {k}"
+            );
+        }
+        // 50..100 are in remote → must be in candidate_intersection.
+        for k in 50u64..100 {
+            assert!(
+                part.candidate_intersection.contains(&k),
+                "missing candidate key {k}"
+            );
+        }
+    }
+
+    #[test]
+    fn server_reconcile_one_sided_push() {
+        // Client (A) has 0..120; server (B) has 0..100.
+        // Expected: missing_from_remote = 100..120.
+        let a_keys: Vec<u64> = (0..120).collect();
+        let b_keys: Vec<u64> = (0..100).collect();
+
+        let rbf = RatelessBloomFilter::from_keys(&a_keys, SEED);
+        let cells = residual_iblt_cells(&rbf, 25); // expect ~20 only_in_a
+        let client_iblt = build_residual_iblt(&a_keys, cells, SEED);
+
+        let out = server_reconcile(&rbf, &client_iblt, &b_keys).unwrap();
+        let mut got = out.missing_from_remote.clone();
+        got.sort_unstable();
+        let expected: Vec<u64> = (100..120).collect();
+        assert_eq!(got, expected);
+    }
+
+    #[test]
+    fn server_reconcile_symmetric_diff() {
+        // Client has 0..100 + extras [777, 888]; server has 0..100 + [555].
+        // Expected: missing_from_remote = [777, 888].
+        let mut a: Vec<u64> = (0..100).collect();
+        a.extend_from_slice(&[777, 888]);
+        let mut b: Vec<u64> = (0..100).collect();
+        b.push(555);
+
+        let rbf = RatelessBloomFilter::from_keys(&a, SEED);
+        let cells = residual_iblt_cells(&rbf, 5);
+        let client_iblt = build_residual_iblt(&a, cells, SEED);
+
+        let out = server_reconcile(&rbf, &client_iblt, &b).unwrap();
+        let mut got = out.missing_from_remote.clone();
+        got.sort_unstable();
+        assert_eq!(got, vec![777, 888]);
+    }
+
+    #[test]
+    fn server_reconcile_empty_server() {
+        // Initial clone shape: server has nothing, client has 0..50.
+        let a_keys: Vec<u64> = (0..50).collect();
+        let b_keys: Vec<u64> = vec![];
+
+        let rbf = RatelessBloomFilter::from_keys(&a_keys, SEED);
+        let cells = residual_iblt_cells(&rbf, a_keys.len());
+        let client_iblt = build_residual_iblt(&a_keys, cells, SEED);
+
+        let out = server_reconcile(&rbf, &client_iblt, &b_keys).unwrap();
+        let mut got = out.missing_from_remote.clone();
+        got.sort_unstable();
+        assert_eq!(got, a_keys);
+    }
+
+    #[test]
+    fn server_reconcile_already_in_sync() {
+        let keys: Vec<u64> = (0..200).collect();
+        let rbf = RatelessBloomFilter::from_keys(&keys, SEED);
+        let cells = residual_iblt_cells(&rbf, 0);
+        let client_iblt = build_residual_iblt(&keys, cells, SEED);
+
+        let out = server_reconcile(&rbf, &client_iblt, &keys).unwrap();
+        assert!(out.missing_from_remote.is_empty());
+    }
+
+    #[test]
+    fn server_reconcile_returns_err_when_too_small() {
+        // Force the residual IBLT to be undersized.
+        let a_keys: Vec<u64> = (0..500).collect();
+        let b_keys: Vec<u64> = (0..500).filter(|k| k % 5 != 0).collect();
+        let rbf = RatelessBloomFilter::from_keys(&a_keys, SEED);
+
+        // Tiny residual IBLT — guaranteed to overflow.
+        let client_iblt = build_residual_iblt(&a_keys, 4, SEED);
+        let result = server_reconcile(&rbf, &client_iblt, &b_keys);
+        assert!(result.is_err(), "expected decode failure");
+    }
+
+    // ── heuristic ────────────────────────────────────────────────────────────
+
+    #[test]
+    fn heuristic_triggers_above_threshold() {
+        assert!(should_use_rbf(RBF_REV_COUNT_THRESHOLD + 1, None));
+    }
+
+    #[test]
+    fn heuristic_off_below_threshold() {
+        assert!(!should_use_rbf(50, None));
+    }
+
+    #[test]
+    fn heuristic_high_delta_overrides_low_count() {
+        assert!(should_use_rbf(50, Some(0.30)));
+    }
+
+    #[test]
+    fn heuristic_off_with_low_delta() {
+        assert!(!should_use_rbf(50, Some(0.05)));
+    }
+
+    // ── proptest ─────────────────────────────────────────────────────────────
+
+    proptest! {
+        #[test]
+        fn prop_no_false_negatives(
+            keys in proptest::collection::vec(0u64..100_000, 0..200),
+        ) {
+            let rbf = RatelessBloomFilter::from_keys(&keys, SEED);
+            for k in &keys {
+                prop_assert!(rbf.contains(*k));
+            }
+        }
+
+        #[test]
+        fn prop_serialise_roundtrip(
+            keys in proptest::collection::vec(0u64..100_000, 0..200),
+        ) {
+            let rbf = RatelessBloomFilter::from_keys(&keys, SEED);
+            let bytes = rbf.to_bytes();
+            let r2 = RatelessBloomFilter::from_bytes(&bytes).unwrap();
+            prop_assert_eq!(rbf, r2);
+        }
+
+        /// For high-divergence inputs, the hybrid recovers `only_in_A` exactly
+        /// as long as the residual IBLT is sized for the expected stragglers.
+        #[test]
+        fn prop_server_reconcile_recovers_only_in_a(
+            n_common in 0usize..150,
+            extra_a in proptest::collection::vec(100_000u64..200_000, 0..30),
+            extra_b in proptest::collection::vec(200_000u64..300_000, 0..30),
+        ) {
+            // Revision numbers are unique in clawsync — dedup the property
+            // inputs to model that invariant.
+            let mut extra_a = extra_a.clone();
+            extra_a.sort_unstable();
+            extra_a.dedup();
+            let mut extra_b = extra_b.clone();
+            extra_b.sort_unstable();
+            extra_b.dedup();
+
+            let common: Vec<u64> = (0..n_common as u64).collect();
+            let mut a = common.clone();
+            a.extend(&extra_a);
+            let mut b = common.clone();
+            b.extend(&extra_b);
+
+            let rbf = RatelessBloomFilter::from_keys(&a, SEED);
+            // Sized for worst-case stragglers + generous safety.
+            let cells = residual_iblt_cells(&rbf, extra_a.len().max(4)) * 4;
+            let client_iblt = build_residual_iblt(&a, cells, SEED);
+
+            let out = server_reconcile(&rbf, &client_iblt, &b).unwrap();
+            let mut got = out.missing_from_remote.clone();
+            got.sort_unstable();
+            let mut expected = extra_a.clone();
+            expected.sort_unstable();
+            expected.dedup();
+            prop_assert_eq!(got, expected);
+        }
+    }
+}
@@ -2,48 +2,32 @@

 use thiserror::Error;

-/// All errors that can occur in the `clawsync-transport` layer.
 #[derive(Debug, Error)]
 pub enum TransportError {
-    /// An underlying I/O error (socket read/write, bind, etc.).
    #[error("I/O error: {0}")]
    Io(#[from] std::io::Error),

-    /// The remote peer closed the connection before the exchange completed.
    #[error("connection closed unexpectedly")]
    ConnectionClosed,

-    /// A received frame's length prefix exceeds the protocol maximum.
    #[error("frame too large: {size} bytes (max {max})")]
-    FrameTooLarge {
-        /// The length the remote declared in the 4-byte LE prefix.
-        size: usize,
-        /// The maximum frame size permitted by the protocol.
-        max: usize,
-    },
+    FrameTooLarge { size: usize, max: usize },

-    /// A `SyncMessage` could not be serialized to bytes.
    #[error("serialization error: {0}")]
    Serialization(String),

-    /// Received bytes could not be deserialized into a `SyncMessage`.
    #[error("deserialization error: {0}")]
    Deserialization(String),

-    /// An error from the underlying QUIC stack (e.g., `quinn`).
    #[error("QUIC error: {0}")]
    Quic(String),

-    /// A TLS configuration or handshake error.
    #[error("TLS error: {0}")]
    Tls(String),

-    /// The remote peer violated the sync protocol (unexpected message type
-    /// or message sequencing error).
    #[error("protocol error: {0}")]
    Protocol(String),

-    /// An error propagated from the `clawsync-onion` sync layer.
    #[error("sync onion error: {0}")]
    SyncOnion(#[from] clawsync_onion::SyncOnionError),
 }
@@ -24,12 +24,10 @@ pub struct FramedReader<R: AsyncRead + Unpin + Send> {
 }

 impl<R: AsyncRead + Unpin + Send> FramedReader<R> {
-    /// Wrap `inner` in a framed reader.
    pub fn new(inner: R) -> Self {
        Self { inner }
    }

-    /// Read and deserialize the next length-prefixed `SyncMessage` frame.
    pub async fn recv(&mut self) -> Result<SyncMessage, TransportError> {
        let mut len_buf = [0u8; 4];
        self.inner.read_exact(&mut len_buf).await?;
@@ -54,12 +52,10 @@ pub struct FramedWriter<W: AsyncWrite + Unpin + Send> {
 }

 impl<W: AsyncWrite + Unpin + Send> FramedWriter<W> {
-    /// Wrap `inner` in a framed writer.
    pub fn new(inner: W) -> Self {
        Self { inner }
    }

-    /// Serialize `msg` and write it as a length-prefixed frame.
    pub async fn send(&mut self, msg: &SyncMessage) -> Result<(), TransportError> {
        let body = msg.to_bytes().map_err(TransportError::Serialization)?;
        let len = body.len() as u32;
@@ -69,7 +65,6 @@ impl<W: AsyncWrite + Unpin + Send> FramedWriter<W> {
        Ok(())
    }

-    /// Flush pending writes and shut down the write half.
    pub async fn shutdown(&mut self) -> Result<(), TransportError> {
        self.inner.shutdown().await?;
        Ok(())
@@ -117,17 +112,14 @@ impl StreamPeer {
        }
    }

-    /// Send a `SyncMessage` through the peer's write half.
    pub async fn send(&mut self, msg: &SyncMessage) -> Result<(), TransportError> {
        self.writer.send(msg).await
    }

-    /// Receive the next `SyncMessage` from the peer's read half.
    pub async fn recv(&mut self) -> Result<SyncMessage, TransportError> {
        self.reader.recv().await
    }

-    /// Flush and shut down the write half.
    pub async fn shutdown(&mut self) -> Result<(), TransportError> {
        self.writer.shutdown().await
    }
@@ -142,26 +134,21 @@ impl StreamPeer {
 // Split halves
 // ─────────────────────────────────────────────────────────────────────────────

-/// Read half produced by [`StreamPeer::into_split`].
 pub struct StreamReadHalf(pub(crate) FramedReader<BoxReader>);

 impl StreamReadHalf {
-    /// Receive the next `SyncMessage` from the boxed reader.
    pub async fn recv(&mut self) -> Result<SyncMessage, TransportError> {
        self.0.recv().await
    }
 }

-/// Write half produced by [`StreamPeer::into_split`].
 pub struct StreamWriteHalf(pub(crate) FramedWriter<BoxWriter>);

 impl StreamWriteHalf {
-    /// Send a `SyncMessage` through the boxed writer.
    pub async fn send(&mut self, msg: &SyncMessage) -> Result<(), TransportError> {
        self.0.send(msg).await
    }

-    /// Flush and shut down the boxed writer.
    pub async fn shutdown(&mut self) -> Result<(), TransportError> {
        self.0.shutdown().await
    }
@@ -27,17 +27,13 @@ use crate::tcp::{TcpConnection, TcpReadHalf, TcpWriteHalf};

 /// A unified connection handle for a single sync session (TCP, QUIC, mmap, or stream).
 pub enum SyncPeer {
-    /// An established async TCP connection.
    Tcp(TcpConnection),
-    /// A QUIC connection shared via `Arc` (send + recv are `&self`).
    Quic(Arc<QuicConnection>),
    /// Local same-node transport backed by two memory-mapped ring buffers —
    /// one per direction.  `send_ch` is the outgoing channel; `recv_ch` is
    /// the incoming channel.
    Mmap {
-        /// Outgoing mmap ring-buffer sender.
        send_ch: MmapSender,
-        /// Incoming mmap ring-buffer receiver.
        recv_ch: MmapReceiver,
    },
    /// Generic boxed-I/O peer — covers SSH child-process pipes and `--stdio`
@@ -46,11 +42,6 @@ pub enum SyncPeer {
 }

 impl SyncPeer {
-    /// Send a `SyncMessage` to the remote peer.
-    ///
-    /// For TCP and Stream peers this performs a framed write (4-byte LE length
-    /// + body).  For QUIC, each call opens a new unidirectional stream.  For
-    /// Mmap, the call blocks until the ring buffer has space.
    pub async fn send(&mut self, msg: &SyncMessage) -> Result<(), TransportError> {
        match self {
            SyncPeer::Tcp(c) => c.send(msg).await,
@@ -60,11 +51,6 @@ impl SyncPeer {
        }
    }

-    /// Receive the next `SyncMessage` from the remote peer.
-    ///
-    /// Blocks until a complete message is available.  Returns
-    /// [`TransportError::ConnectionClosed`] when the peer has cleanly closed
-    /// the connection.
    pub async fn recv(&mut self) -> Result<SyncMessage, TransportError> {
        match self {
            SyncPeer::Tcp(c) => c.recv().await,
@@ -74,7 +60,6 @@ impl SyncPeer {
        }
    }

-    /// Gracefully shut down the connection, flushing any pending writes.
    pub async fn shutdown(self) -> Result<(), TransportError> {
        match self {
            SyncPeer::Tcp(mut c) => c.shutdown().await,
@@ -151,18 +136,13 @@ impl SyncPeer {

 /// Write half for the sliding-window pipeline.
 pub enum PipeWriteHalf {
-    /// Write half of a split `TcpConnection`.
    Tcp(TcpWriteHalf),
-    /// QUIC connection `Arc` clone used for sending (QUIC send is `&self`).
    Quic(Arc<QuicConnection>),
-    /// Mmap ring-buffer sender.
    Mmap(MmapSender),
-    /// Boxed-I/O write half (SSH / stdio).
    Stream(StreamWriteHalf),
 }

 impl PipeWriteHalf {
-    /// Send a `SyncMessage` through the write half.
    pub async fn send(&mut self, msg: &SyncMessage) -> Result<(), TransportError> {
        match self {
            PipeWriteHalf::Tcp(h) => h.send(msg).await,
@@ -172,7 +152,6 @@ impl PipeWriteHalf {
        }
    }

-    /// Gracefully shut down the write half.
    pub async fn shutdown(self) -> Result<(), TransportError> {
        match self {
            PipeWriteHalf::Tcp(mut h) => h.shutdown().await,
@@ -205,18 +184,13 @@ impl PipeWriteHalf {

 /// Read half for the sliding-window pipeline.
 pub enum PipeReadHalf {
-    /// Read half of a split `TcpConnection`.
    Tcp(TcpReadHalf),
-    /// QUIC connection `Arc` clone used for receiving.
    Quic(Arc<QuicConnection>),
-    /// Mmap ring-buffer receiver.
    Mmap(MmapReceiver),
-    /// Boxed-I/O read half (SSH / stdio).
    Stream(StreamReadHalf),
 }

 impl PipeReadHalf {
-    /// Receive the next `SyncMessage` from the read half.
    pub async fn recv(&mut self) -> Result<SyncMessage, TransportError> {
        match self {
            PipeReadHalf::Tcp(h) => h.recv().await,
@@ -33,7 +33,7 @@

 use rkyv::{Archive, Deserialize, Serialize};

-use clawsync_onion::manifest::{ClawSyncManifest, IbltManifest};
+use clawsync_onion::manifest::{ClawSyncManifest, IbltManifest, RbfManifest};
 use clawsync_onion::packet::OnionLayerPacket;

 // ─────────────────────────────────────────────────────────────────────────────
@@ -120,45 +120,27 @@ pub const MAX_FRAME_SIZE: usize = 256 * 1024 * 1024;
 pub enum SyncMessage {
    /// Client announces itself and its current HEAD revision.
    ManifestRequest {
-        /// Logical agent / file identifier (used by the server to look up the sidecar).
        agent_id: String,
-        /// HEAD revision number the client currently holds.
        head_revision: u64,
-        /// Total revision count (so the server can compute the delta).
+        /// Revision count (so the server can compute the delta).
        revision_count: u64,
    },

    /// Server responds with its own manifest.
-    ManifestResponse {
-        /// Full revision manifest of the server's sidecar.
-        manifest: ClawSyncManifest,
-    },
+    ManifestResponse { manifest: ClawSyncManifest },

    /// One revision packet (sent by either side, depending on push/pull).
-    LayerPacket {
-        /// The serialized revision data (pages, metadata, BLAKE3).
-        packet: OnionLayerPacket,
-    },
+    LayerPacket { packet: OnionLayerPacket },

    /// Acknowledgement of a successfully received + verified revision.
-    Ack {
-        /// The revision number that was successfully applied.
-        revision: u64,
-    },
+    Ack { revision: u64 },

    /// Request to retransmit a revision (hash verification failed).
-    RetryRequest {
-        /// The revision number that failed integrity verification.
-        revision: u64,
-        /// Human-readable reason for the retry request.
-        reason: String,
-    },
+    RetryRequest { revision: u64, reason: String },

    /// Transfer is complete.
    SyncComplete {
-        /// Total number of revisions successfully transferred in this session.
        revisions_transferred: u64,
-        /// Total compressed bytes transferred in this session.
        bytes_transferred: u64,
    },

@@ -166,10 +148,7 @@ pub enum SyncMessage {
    ///
    /// If the remote supports IBLT it responds with `IbltResponse`; otherwise
    /// it responds with `Error` and the client falls back to `ManifestRequest`.
-    IbltRequest {
-        /// The client's IBLT sketch of its local revision set.
-        sketch: IbltManifest,
-    },
+    IbltRequest { sketch: IbltManifest },

    /// IBLT pre-flight response: server's sketch + decoded missing revisions.
    ///
@@ -177,17 +156,12 @@ pub enum SyncMessage {
    /// lacks (what to push).  The client decodes `sketch` against its own
    /// sketch to find revisions it needs to pull.
    IbltResponse {
-        /// The server's IBLT sketch (for the client to decode what it should pull).
        sketch: IbltManifest,
-        /// Revision numbers the server lacks (client should push these).
        missing_from_remote: Vec<u64>,
    },

    /// An error occurred; connection will be closed after this message.
-    Error {
-        /// Human-readable error description.
-        message: String,
-    },
+    Error { message: String },

    // ── FS CDC file delta (variants 9–12) ────────────────────────────────────
    //
@@ -205,9 +179,7 @@ pub enum SyncMessage {
    /// Originally: client announces a single-file CDC sync, carrying its ordered
    /// chunk descriptors.  Superseded by the server chunk embedding in `FsDirNeed`.
    FsCdcRequest {
-        /// Relative path of the file being synced.
        path: String,
-        /// Client's ordered CDC chunk descriptors (hash + length per chunk).
        chunk_hashes: Vec<FsChunkHash>,
    },

@@ -216,9 +188,7 @@ pub enum SyncMessage {
    /// Originally: server replies with the indices it needs transferred.
    /// Superseded by the server chunk embedding in `FsDirNeed`.
    FsCdcNeed {
-        /// Relative path of the file being synced.
        path: String,
-        /// Indices (into the client's chunk list) that the server needs.
        needed_indices: Vec<u32>,
    },

@@ -232,7 +202,6 @@ pub enum SyncMessage {
    /// sync path, while remaining self-contained for single-file use too.
    /// Wire overhead is ~12 bytes × N_chunks (negligible relative to data).
    FsCdcData {
-        /// Relative path of the file being transferred.
        path: String,
        /// Full ordered chunk layout of the client's file.
        chunk_order: Vec<FsChunkHash>,
@@ -241,10 +210,7 @@ pub enum SyncMessage {
    },

    /// Server confirms it has reconstructed and written the file at `path`.
-    FsFileAck {
-        /// Relative path of the file that was successfully written.
-        path: String,
-    },
+    FsFileAck { path: String },

    // ── FS directory tree sync (variants 13–16) ───────────────────────────────
    /// Client sends a per-file manifest for an entire directory tree.
@@ -271,23 +237,16 @@ pub enum SyncMessage {

    /// Server reports that all changes have been applied.
    FsDirComplete {
-        /// Number of files newly created on the server.
        files_added: u32,
-        /// Number of files overwritten on the server.
        files_modified: u32,
-        /// Number of files deleted from the server.
        files_removed: u32,
-        /// Total compressed bytes transferred in this session.
        bytes_transferred: u64,
    },

    /// Dry-run: reports what *would* change without applying anything.
    FsDirDryRun {
-        /// Paths that would be created on the server.
        would_add: Vec<String>,
-        /// Paths that would be overwritten on the server.
        would_modify: Vec<String>,
-        /// Paths that would be deleted from the server.
        would_remove: Vec<String>,
    },

@@ -325,9 +284,7 @@ pub enum SyncMessage {
    /// `dtype` is the `{:?}` format of `clawhdf5::DType` (e.g. `"F64"`).
    /// `shape` is the dataset's dimension sizes.
    Hdf5DataPayload {
-        /// HDF5 path of the dataset within the file (e.g., `/group/embeddings`).
        path: String,
-        /// Raw little-endian bytes of the dataset.
        data: Vec<u8>,
        /// HDF5 DType debug string, e.g. `"F64"`, `"F32"`, `"I32"`.
        dtype: String,
@@ -336,20 +293,13 @@ pub enum SyncMessage {
    },

    /// Server acknowledges successful reconstruction of a dataset.
-    Hdf5DataAck {
-        /// HDF5 path of the dataset that was successfully applied.
-        path: String,
-    },
+    Hdf5DataAck { path: String },

    /// Server reports that the HDF5 sync is complete.
    Hdf5SyncComplete {
-        /// Number of datasets newly added to the server's file.
        datasets_added: u32,
-        /// Number of datasets overwritten on the server.
        datasets_modified: u32,
-        /// Number of datasets removed from the server's file.
        datasets_removed: u32,
-        /// Total compressed bytes transferred in this session.
        bytes_transferred: u64,
    },

@@ -411,6 +361,38 @@ pub enum SyncMessage {
        /// BLAKE3 hex of the persisted bytes.
        blake3_hex: String,
    },
+
+    // ── Two-stage RBF pre-flight (arXiv 2510.27614) ─────────────────────
+    //
+    // IMPORTANT: rkyv discriminants are positional. These variants MUST
+    // remain appended at the end of the enum. Append only — never insert
+    // or reorder.
+    //
+    /// Two-stage divergent-replica pre-flight request.
+    ///
+    /// Sent by the client when the heuristic
+    /// (`local_rev_count > RBF_REV_COUNT_THRESHOLD`, or expected divergence
+    /// above 15 %) flags the workload as high-divergence — initial clones and
+    /// post-partition reconnects in particular.  The server responds with
+    /// [`SyncMessage::RbfResponse`].  Falls back to plain `IbltRequest` if
+    /// the server doesn't understand this variant (returns `Error`).
+    RbfRequest {
+        /// Rateless Bloom Filter + residual IBLT bundle over the client's
+        /// full revision set.
+        bundle: RbfManifest,
+    },
+
+    /// Two-stage pre-flight response.
+    ///
+    /// Carries the decoded `missing_from_remote` list (= revisions the
+    /// client should push) plus the server's own [`RbfManifest`] so the
+    /// client can mirror-decode `missing_from_local` if needed.
+    RbfResponse {
+        /// Server's own RBF bundle (for symmetric / pull-direction decoding).
+        bundle: RbfManifest,
+        /// Revisions the client has that the server lacks.
+        missing_from_remote: Vec<u64>,
+    },
 }

 impl SyncMessage {
@@ -850,4 +832,60 @@ mod tests {
        assert!(matches!(r2, SyncMessage::Ack { revision: 2 }));
        assert_eq!(n1 + n2, buf.len());
    }
+
+    // ── RbfRequest / RbfResponse roundtrips (arxiv:2510.27614) ──────────
+
+    #[test]
+    fn rbf_request_roundtrip() {
+        let bundle = RbfManifest::from_keys(
+            "test-client",
+            &(0u64..40).collect::<Vec<_>>(),
+            clawsync_onion::rbf::RBF_SYNC_SEED,
+        );
+        let msg = SyncMessage::RbfRequest { bundle };
+        let bytes = msg.to_bytes().unwrap();
+        let recovered = SyncMessage::from_bytes(&bytes).unwrap();
+        match recovered {
+            SyncMessage::RbfRequest { bundle } => {
+                assert_eq!(bundle.agent_id, "test-client");
+                assert_eq!(bundle.revision_count, 40);
+            }
+            other => panic!("wrong variant: {other:?}"),
+        }
+    }
+
+    #[test]
+    fn rbf_response_roundtrip() {
+        let bundle = RbfManifest::from_keys(
+            "test-server",
+            &(0u64..30).collect::<Vec<_>>(),
+            clawsync_onion::rbf::RBF_SYNC_SEED,
+        );
+        let msg = SyncMessage::RbfResponse {
+            bundle,
+            missing_from_remote: vec![30, 31, 32],
+        };
+        let bytes = msg.to_bytes().unwrap();
+        let recovered = SyncMessage::from_bytes(&bytes).unwrap();
+        match recovered {
+            SyncMessage::RbfResponse {
+                bundle,
+                missing_from_remote,
+            } => {
+                assert_eq!(bundle.agent_id, "test-server");
+                assert_eq!(missing_from_remote, vec![30, 31, 32]);
+            }
+            other => panic!("wrong variant: {other:?}"),
+        }
+    }
+
+    /// Regression guard: ManifestRequest must remain at discriminant 0 even
+    /// after the RBF variants were appended.
+    #[test]
+    fn manifest_request_still_at_discriminant_zero_after_rbf() {
+        let msg = manifest_request();
+        let bytes = msg.to_bytes().unwrap();
+        let recovered = SyncMessage::from_bytes(&bytes).unwrap();
+        assert!(matches!(recovered, SyncMessage::ManifestRequest { .. }));
+    }
 }
@@ -32,9 +32,7 @@ use crate::protocol::SyncMessage;

 /// QUIC endpoint configuration.
 pub struct QuicConfig {
-    /// TLS server configuration for the QUIC server endpoint.
    pub server_config: ServerConfig,
-    /// TLS client configuration for outgoing QUIC connections.
    pub client_config: ClientConfig,
 }

@@ -455,10 +453,6 @@ impl QuicConnection {
 /// A QUIC server endpoint.
 pub struct QuicServer {
    endpoint: Endpoint,
-    /// The local `SocketAddr` this server is bound to.
-    ///
-    /// Useful when the server was bound to port `0` and the assigned ephemeral
-    /// port must be communicated to clients.
    pub local_addr: SocketAddr,
 }

@@ -144,10 +144,6 @@ impl TcpWriteHalf {
 /// A listening TCP server that accepts `TcpConnection`s.
 pub struct TcpServer {
    listener: TcpListener,
-    /// The local `SocketAddr` this server is bound to.
-    ///
-    /// Useful for callers that bind to port `0` and need to discover the
-    /// assigned ephemeral port after binding.
    pub local_addr: SocketAddr,
 }