//! Automerge document storage with redb persistence
//!
//! This module provides persistent storage for Automerge CRDT documents using redb,
//! a pure Rust embedded database. This replaces the previous RocksDB implementation
//! to eliminate C/C++ build dependencies and align with Iroh's storage layer.

use crate::storage::cipher::Cipher;
use crate::storage::traits::{Collection, DocumentPredicate};
use crate::storage::ttl_manager::TtlManager;
use automerge::{transaction::Transactable, Automerge, ReadDoc};
use lru::LruCache;
use redb::{Builder, Database, ReadableTable, ReadableTableMetadata, TableDefinition};
use std::hash::{Hash, Hasher};
use std::num::NonZeroUsize;
use std::path::Path;
use std::sync::{Arc, RwLock};
use tokio::sync::broadcast;

use anyhow::{Context, Result};

/// Table definition for document storage
/// Key: document key as string bytes
/// Value: serialized Automerge document bytes
const DOCUMENTS_TABLE: TableDefinition<&[u8], &[u8]> = TableDefinition::new("documents");

/// Default redb cache size in bytes (Issue #446)
///
/// The redb default is 1 GiB which is excessive for our use case.
/// We set a much smaller default (16 MiB) that's sufficient for typical
/// document storage while preventing unbounded memory growth.
///
/// Can be overridden via `CAP_REDB_CACHE_SIZE` environment variable (in bytes).
const DEFAULT_REDB_CACHE_SIZE: usize = 16 * 1024 * 1024; // 16 MiB

/// Table definition for tombstone storage (ADR-034 Phase 2)
/// Key: "collection:document_id" as string bytes
/// Value: serialized Tombstone bytes (JSON)
const TOMBSTONES_TABLE: TableDefinition<&[u8], &[u8]> = TableDefinition::new("tombstones");

/// Table definition for backend-level metadata (peat-mesh#195).
///
/// Reserved for small, infrequently-mutated values that survive process
/// restart but don't belong in a document or tombstone payload. First
/// occupant is the Lamport-clock high-water mark for
/// `AutomergeBackend::next_lamport`; future entries (sync resume cursor,
/// last-compaction timestamp, etc.) plug in by adding a new key under
/// the same table.
///
/// Key: ASCII identifier (e.g. `"lamport_highwater"`).
/// Value: caller-defined bytes — for `lamport_highwater`, 8 bytes of
/// little-endian `u64`. The table is intentionally untyped so the
/// schema for new keys stays local to their accessor methods.
const METADATA_TABLE: TableDefinition<&str, &[u8]> = TableDefinition::new("metadata");

/// Metadata key for the Lamport-clock high-water value persisted by
/// [`AutomergeStore::write_lamport_highwater`] (peat-mesh#195).
const METADATA_KEY_LAMPORT_HIGHWATER: &str = "lamport_highwater";

/// Storage layer for Automerge documents with redb persistence
///
/// # Change Notifications (Phase 6.3)
///
/// The store emits change notifications when documents are modified via `put()`.
/// Subscribers can listen for these notifications to trigger automatic sync.
///
/// # In-Memory Mode
///
/// For high-throughput testing, the store can operate in pure in-memory mode
/// where all documents are stored only in the LRU cache (no disk persistence).
/// Enable via `AutomergeStore::in_memory()` constructor.
/// Number of striped lock buckets for per-document concurrency control.
/// Documents are hashed into buckets, so operations on different documents
/// rarely contend while operations on the same document serialize correctly.
const DOC_LOCK_STRIPES: usize = 64;

/// Origin of a document mutation observed by the store.
///
/// Used by [`DocChange`] / [`AutomergeStore::subscribe_to_changes_with_origin`]
/// so consumers driving onward sync (gossip) can avoid echoing a change back
/// to the peer it just arrived from.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum ChangeOrigin {
    /// Change originated from a local write (caller is on this node).
    Local,
    /// Change was received from a remote peer via the sync coordinator.
    ///
    /// The string is the peer's stable identifier — for the Iroh transport,
    /// `EndpointId::to_string()` (hex-encoded NodeId). The store keeps the
    /// representation transport-agnostic so non-Iroh transports (BLE, Lite)
    /// can populate the same field.
    ///
    /// **Equality contract:** consumers driving transitive gossip compare
    /// this string against the identifier they hold for each connected peer
    /// to decide whether to skip pushing the doc back to its source. Both
    /// sides MUST use the same stringification of the peer identifier;
    /// otherwise echo suppression silently breaks and a doc received from
    /// peer X is re-sent back to X. On the Iroh transport the canonical form
    /// is `EndpointId::to_string()` (hex-encoded NodeId) — the consumer's
    /// `connected_peers()` iteration should call `peer_id.to_string()` and
    /// compare to this field. A future BLE/Lite/LoRa transport populating
    /// this field with a different encoding MUST coordinate the same change
    /// in its consumer pushers, or echo suppression will not work on
    /// mixed-transport meshes.
    Remote(String),
}

/// Document-change event published on the origin-tagged broadcast (peat#891).
///
/// Local puts emit `DocChange { key, origin: ChangeOrigin::Local }`.
/// Remote-sync receives emit `DocChange { key, origin: ChangeOrigin::Remote(peer) }`.
/// Consumers driving onward propagation can use the `origin` field to skip
/// the source peer and prevent echoes without suppressing the broadcast.
#[derive(Debug, Clone)]
pub struct DocChange {
    pub key: String,
    pub origin: ChangeOrigin,
}

/// An opaque delta between two Automerge documents (peat-mesh#187).
///
/// Built by [`AutomergeStore::diff`] from a `(current, proposed)` pair and
/// consumed by [`AutomergeStore::apply_delta`] to advance a stored document
/// without round-tripping the full byte image. The CRDT-level effect of
/// applying a delta is identical to what the receive-side sync path does
/// when it merges a `SyncMessage` from a peer — the document's operation
/// history is preserved (the ADR-021 "create once, evolve through deltas"
/// invariant), and the resulting state has both histories causally joined.
///
/// Internally the delta is the set of `automerge::Change` records that
/// exist in `proposed` but not in `current` (the same shape automerge's
/// own `Automerge::get_changes_added` returns), encoded as raw change
/// bytes. Each change is self-describing in automerge's binary format,
/// but `AutomergeDelta` carries an explicit `[u32-le length][bytes]`
/// framing in [`Self::to_bytes`] / [`Self::from_bytes`] so the
/// serialised form is parseable without out-of-band schema.
#[derive(Debug, Clone, Default)]
pub struct AutomergeDelta {
    /// Encoded `automerge::Change` entries, one per change in the delta.
    /// `pub(crate)` so `apply_delta_inner` can iterate without going
    /// through a public accessor.
    pub(crate) changes: Vec<Vec<u8>>,
}

impl AutomergeDelta {
    /// Number of underlying Automerge changes carried by this delta.
    pub fn change_count(&self) -> usize {
        self.changes.len()
    }

    /// `true` when applying this delta would be a no-op (proposed has no
    /// causal history beyond current's heads).
    pub fn is_empty(&self) -> bool {
        self.changes.is_empty()
    }

    /// Serialise the delta to a self-contained byte buffer. Each change is
    /// length-prefixed (`u32` little-endian) so the buffer round-trips via
    /// [`Self::from_bytes`] without out-of-band schema. Suitable for
    /// transport between a client and a peat-mesh process where the
    /// `Automerge` value itself cannot cross the boundary.
    ///
    /// Returns `Err` only if a single change exceeds `u32::MAX` bytes
    /// (~4 GiB). Automerge changes are individually-bounded in practice
    /// (kilobytes), so the realistic risk is zero; the checked
    /// conversion exists to make the impossible case loud rather than
    /// silently corrupt the framed buffer.
    pub fn to_bytes(&self) -> Result<Vec<u8>> {
        let mut out = Vec::with_capacity(self.changes.iter().map(|c| 4 + c.len()).sum());
        for change in &self.changes {
            let len = u32::try_from(change.len()).map_err(|_| {
                anyhow::anyhow!(
                    "AutomergeDelta::to_bytes: single change body of {} bytes exceeds u32::MAX",
                    change.len()
                )
            })?;
            out.extend_from_slice(&len.to_le_bytes());
            out.extend_from_slice(change);
        }
        Ok(out)
    }

    /// Inverse of [`Self::to_bytes`]. Rejects truncated buffers with a
    /// descriptive error rather than silently dropping the trailing
    /// partial entry.
    pub fn from_bytes(bytes: &[u8]) -> Result<Self> {
        let mut changes = Vec::new();
        let mut cursor = 0usize;
        while cursor < bytes.len() {
            if cursor + 4 > bytes.len() {
                anyhow::bail!(
                    "AutomergeDelta::from_bytes: truncated length prefix at offset {cursor}"
                );
            }
            let len = u32::from_le_bytes(bytes[cursor..cursor + 4].try_into().unwrap()) as usize;
            cursor += 4;
            if cursor + len > bytes.len() {
                anyhow::bail!(
                    "AutomergeDelta::from_bytes: truncated change body at offset {cursor} \
                     (need {len} bytes, have {})",
                    bytes.len() - cursor
                );
            }
            changes.push(bytes[cursor..cursor + len].to_vec());
            cursor += len;
        }
        Ok(Self { changes })
    }
}

pub struct AutomergeStore {
    /// Database handle - None when running in memory-only mode
    db: Option<Arc<Database>>,
    cache: Arc<RwLock<LruCache<String, Automerge>>>,
    /// Broadcast channel for sync coordinator - used to trigger P2P sync
    /// Only notified for local puts (not synced documents)
    change_tx: broadcast::Sender<String>,
    /// Broadcast channel for observers - used for hierarchical aggregation (Issue #377)
    /// Notified for ALL document changes (local and synced) so observers can react
    observer_tx: broadcast::Sender<String>,
    /// Origin-tagged broadcast for transitive gossip drivers (peat#891 / peat#907).
    ///
    /// Fires on ALL document mutations (local and sync-received), carrying the
    /// origin attribution so consumers can re-propagate without echoing back to
    /// the source peer. Unlike `change_tx` (local-only) and `observer_tx`
    /// (untagged), this channel is the one a gossip pusher should subscribe to.
    gossip_tx: broadcast::Sender<DocChange>,
    /// Striped locks for per-document concurrency control (Issue #74).
    /// Serializes read-modify-write operations (compact, merge) on the
    /// same document key to prevent silent data loss.
    doc_locks: Box<[std::sync::Mutex<()>]>,
    /// Optional at-rest cipher (peat-mesh#124). When `Some`, document
    /// bytes are encrypted before redb writes and decrypted on read.
    /// Tombstones and the LRU cache are not affected by this layer.
    cipher: Option<Arc<dyn Cipher>>,
    /// Test-only counter of `scan_prefix` invocations. Used by the
    /// peat-mesh#131 regression test to assert that a circuit-breaker-
    /// gated call to `sync_all_documents_with_peer` does not scan the
    /// store. Behind cfg(test) so production builds don't carry the
    /// per-call atomic.
    #[cfg(test)]
    scan_prefix_calls: std::sync::atomic::AtomicUsize,
}

impl AutomergeStore {
    /// Test-only: read the cumulative `scan_prefix` call count. Used by
    /// the peat-mesh#131 regression test on `sync_all_documents_with_peer`.
    #[cfg(test)]
    pub(crate) fn scan_prefix_call_count(&self) -> usize {
        self.scan_prefix_calls
            .load(std::sync::atomic::Ordering::Relaxed)
    }
}

impl AutomergeStore {
    /// Open or create storage at the given path, without at-rest encryption.
    ///
    /// Convenience wrapper around [`open_with_cipher`](Self::open_with_cipher)
    /// for callers that don't need encryption. Backends and tests that
    /// haven't been ported to thread a cipher through can keep using this.
    pub fn open(path: impl AsRef<Path>) -> Result<Self> {
        Self::open_with_cipher(path, None)
    }

    /// Open or create storage at the given path, optionally with at-rest
    /// encryption (peat-mesh#124).
    ///
    /// When `cipher` is `Some`, every byte stored in `DOCUMENTS_TABLE`
    /// passes through [`Cipher::encrypt`] on write and
    /// [`Cipher::decrypt`] on read. Tombstones in `TOMBSTONES_TABLE`
    /// are left plaintext in this version — encrypting tombstone
    /// metadata is a follow-up if needed.
    pub fn open_with_cipher(
        path: impl AsRef<Path>,
        cipher: Option<Arc<dyn Cipher>>,
    ) -> Result<Self> {
        // Ensure the directory exists
        let path = path.as_ref();
        if let Some(parent) = path.parent() {
            std::fs::create_dir_all(parent).ok();
        }

        // redb stores in a single file, append .redb extension if it's a directory path
        let db_path = if path.is_dir() || !path.exists() {
            std::fs::create_dir_all(path).ok();
            path.join("automerge.redb")
        } else {
            path.to_path_buf()
        };

        // Check for corrupted (0-byte) database file and remove it
        // This can happen on Android if the previous initialization was interrupted
        if db_path.exists() {
            if let Ok(metadata) = std::fs::metadata(&db_path) {
                if metadata.len() == 0 {
                    tracing::warn!("Removing corrupted 0-byte redb database at {:?}", db_path);
                    std::fs::remove_file(&db_path).ok();
                }
            }
        }

        // Configure redb cache size (Issue #446)
        // Default redb cache is 1 GiB which causes excessive memory growth.
        // Use a smaller cache (default 16 MiB) or allow override via environment variable.
        let cache_size = std::env::var("CAP_REDB_CACHE_SIZE")
            .ok()
            .and_then(|s| s.parse::<usize>().ok())
            .unwrap_or(DEFAULT_REDB_CACHE_SIZE);

        tracing::debug!("Opening redb database with cache_size={} bytes", cache_size);

        let db = Builder::new()
            .set_cache_size(cache_size)
            .create(&db_path)
            .context("Failed to open redb database")?;

        // Initialize the tables (redb requires this on first use)
        {
            let write_txn = db
                .begin_write()
                .context("Failed to begin write transaction")?;
            // Creating the tables if they don't exist
            let _ = write_txn.open_table(DOCUMENTS_TABLE);
            let _ = write_txn.open_table(TOMBSTONES_TABLE); // ADR-034 Phase 2
            let _ = write_txn.open_table(METADATA_TABLE); // peat-mesh#195
            write_txn
                .commit()
                .context("Failed to commit table creation")?;
        }

        let cache = LruCache::new(NonZeroUsize::new(1000).unwrap());

        // Create broadcast channel for sync coordinator notifications
        // Issue #346: Increased from 1024 to 8192 to reduce lagging under high load.
        // When this channel lags, the auto_sync_task must do a full resync which is expensive.
        // A larger buffer trades memory (8KB per doc_key) for reduced resync frequency.
        let (change_tx, _) = broadcast::channel(8192);

        // Create broadcast channel for observer notifications (Issue #377)
        // This channel notifies ALL document changes including synced documents
        // so hierarchical aggregation can react to remotely synced cohort summaries
        let (observer_tx, _) = broadcast::channel(8192);

        // Origin-tagged channel for the transitive-gossip pusher (peat#891).
        // Sized to match the other two for symmetric backpressure.
        let (gossip_tx, _) = broadcast::channel(8192);

        Ok(Self {
            db: Some(Arc::new(db)),
            cache: Arc::new(RwLock::new(cache)),
            change_tx,
            observer_tx,
            gossip_tx,
            doc_locks: (0..DOC_LOCK_STRIPES)
                .map(|_| std::sync::Mutex::new(()))
                .collect::<Vec<_>>()
                .into_boxed_slice(),
            cipher,
            #[cfg(test)]
            scan_prefix_calls: std::sync::atomic::AtomicUsize::new(0),
        })
    }

    /// Create an in-memory store (no disk persistence)
    ///
    /// Documents are stored only in the LRU cache. This mode is useful for
    /// high-throughput testing where persistence is not required.
    ///
    /// Note: Cache size is 10,000 documents in memory mode (vs 1,000 for disk mode)
    /// to accommodate larger working sets.
    pub fn in_memory() -> Self {
        // Larger cache for in-memory mode since we have no disk backing
        let cache = LruCache::new(NonZeroUsize::new(10000).unwrap());
        let (change_tx, _) = broadcast::channel(8192);
        let (observer_tx, _) = broadcast::channel(8192);
        let (gossip_tx, _) = broadcast::channel(8192);

        tracing::info!("AutomergeStore: Running in MEMORY-ONLY mode (no disk persistence)");

        Self {
            db: None,
            cache: Arc::new(RwLock::new(cache)),
            change_tx,
            observer_tx,
            gossip_tx,
            doc_locks: (0..DOC_LOCK_STRIPES)
                .map(|_| std::sync::Mutex::new(()))
                .collect::<Vec<_>>()
                .into_boxed_slice(),
            // In-memory mode never hits redb, so the cipher field is
            // unused even if a caller wanted one. Skip the constructor
            // permutation — `Cipher` is at-rest only by definition.
            cipher: None,
            #[cfg(test)]
            scan_prefix_calls: std::sync::atomic::AtomicUsize::new(0),
        }
    }

    /// Acquire the striped lock for a document key.
    ///
    /// Returns a `MutexGuard` that serializes read-modify-write operations
    /// on documents hashing to the same stripe. Hold this guard across
    /// the entire get→modify→put sequence.
    ///
    /// Used by `compact()` and `AutomergeSyncCoordinator::receive_sync_message()`
    /// to prevent concurrent modifications from silently dropping changes.
    pub fn lock_doc(&self, key: &str) -> std::sync::MutexGuard<'_, ()> {
        let mut hasher = std::collections::hash_map::DefaultHasher::new();
        key.hash(&mut hasher);
        let idx = (hasher.finish() as usize) % DOC_LOCK_STRIPES;
        self.doc_locks[idx]
            .lock()
            .unwrap_or_else(|e| e.into_inner())
    }

    /// Check if the store is running in memory-only mode
    pub fn is_in_memory(&self) -> bool {
        self.db.is_none()
    }

    /// Serialize an Automerge doc into the byte form written to redb,
    /// applying the configured at-rest cipher if any (peat-mesh#124).
    fn save_doc_for_storage(&self, doc: &Automerge) -> Result<Vec<u8>> {
        let bytes = doc.save();
        match &self.cipher {
            Some(c) => c.encrypt(&bytes).context("at-rest encrypt failed"),
            None => Ok(bytes),
        }
    }

    /// Inverse of [`save_doc_for_storage`](Self::save_doc_for_storage):
    /// decrypt the redb-stored bytes (if a cipher is configured) and
    /// load them as an `Automerge` document.
    fn load_doc_from_storage(&self, raw: &[u8]) -> Result<Automerge> {
        match &self.cipher {
            Some(c) => {
                let plaintext = c.decrypt(raw).context("at-rest decrypt failed")?;
                Automerge::load(&plaintext).context("Failed to load Automerge document")
            }
            None => Automerge::load(raw).context("Failed to load Automerge document"),
        }
    }

    /// Save an Automerge document
    ///
    /// # Change Notifications (Phase 6.3)
    ///
    /// This method emits a change notification after successfully persisting the document.
    /// Subscribers will receive the document key to trigger automatic sync.
    pub fn put(&self, key: &str, doc: &Automerge) -> Result<()> {
        self.put_inner(key, doc, true, ChangeOrigin::Local)
    }

    /// Save an Automerge document without emitting the sync-coordinator
    /// change notification (Issue #346 / Issue #115).
    ///
    /// Use this method when storing documents received via sync to avoid
    /// triggering a sync-back that would be blocked by cooldown and waste resources.
    /// The sending peer already has this document, so syncing back is unnecessary.
    ///
    /// Note: this still fires `observer_tx` (hierarchical aggregation) and
    /// `gossip_tx` with `ChangeOrigin::Local` — callers that have a peer
    /// identifier should prefer [`put_with_origin`](Self::put_with_origin)
    /// so the gossip pipeline can skip echoing back to the source.
    pub fn put_without_notify(&self, key: &str, doc: &Automerge) -> Result<()> {
        self.put_inner(key, doc, false, ChangeOrigin::Local)
    }

    /// Save an Automerge document tagged with an explicit [`ChangeOrigin`]
    /// (peat#891 / peat#907 — transitive gossip support).
    ///
    /// The sync coordinator's receive-from-peer path uses this with
    /// `origin = ChangeOrigin::Remote(peer_id)` so the origin-tagged
    /// broadcast (`subscribe_to_changes_with_origin`) receives a fully
    /// attributed event. The legacy `change_tx` (driven by
    /// `subscribe_to_changes`) remains local-only and is **not** fired
    /// when origin is `Remote(_)` — that's the same #115 ping-pong fix,
    /// just lifted from "suppress the broadcast" to "attribute the
    /// broadcast and let the consumer filter".
    pub fn put_with_origin(&self, key: &str, doc: &Automerge, origin: ChangeOrigin) -> Result<()> {
        let notify_change_tx = matches!(origin, ChangeOrigin::Local);
        self.put_inner(key, doc, notify_change_tx, origin)
    }

    /// Save an Automerge document with origin attribution AND TTL-based
    /// expiration in one call (peat-mesh#151 QA finding: avoid duplicating
    /// the key-prefix → collection extraction in consumers).
    ///
    /// Mirrors the existing [`put_with_ttl`](Self::put_with_ttl) /
    /// [`put_without_notify_with_ttl`](Self::put_without_notify_with_ttl)
    /// pair: the put step routes through [`put_with_origin`](Self::put_with_origin)
    /// (so `change_tx` is silenced for `Remote(_)` origins per the
    /// #115 invariant), and the TTL step shares the single canonical
    /// [`apply_ttl`](Self::apply_ttl) helper so the key-prefix
    /// convention stays defined in one place.
    pub fn put_with_origin_and_ttl(
        &self,
        key: &str,
        doc: &Automerge,
        origin: ChangeOrigin,
        ttl_manager: &TtlManager,
    ) -> Result<()> {
        self.put_with_origin(key, doc, origin)?;
        Self::apply_ttl(key, ttl_manager)
    }

    /// Save an Automerge document with TTL-based expiration.
    ///
    /// Delegates to `put()` and then schedules TTL cleanup via the TtlManager
    /// if the collection has a configured TTL. The collection name is extracted
    /// from the key prefix (before the first '/' or ':').
    pub fn put_with_ttl(&self, key: &str, doc: &Automerge, ttl_manager: &TtlManager) -> Result<()> {
        self.put(key, doc)?;
        Self::apply_ttl(key, ttl_manager)
    }

    /// Save an Automerge document with TTL-based expiration, without emitting a
    /// sync change notification (Issue #115).
    ///
    /// Why: receive-side writes (delta-sync and state-snapshot apply) must not
    /// fire `change_tx`, or the sync coordinator re-broadcasts the doc to the
    /// peer it just received from — creating a self-sustaining ping-pong that
    /// drains the per-peer rate-limit bucket and starves first-sync of new
    /// documents.
    ///
    /// **Deprecated as of rc.16 / peat#891:** the receive-side path was the
    /// only intended caller of this method, and it now routes through
    /// [`put_with_origin_and_ttl`](Self::put_with_origin_and_ttl) so the
    /// origin-tagged broadcast carries the source peer attribution and
    /// downstream transitive gossip can suppress echoes correctly. Callers
    /// using this method get `ChangeOrigin::Local` on the gossip channel,
    /// which is wrong for sync-received docs — a transitive-gossip pusher
    /// will treat the doc as a local write and push it to every peer
    /// (including the source), wasting bandwidth per peat-mesh#115. Migrate
    /// to `put_with_origin_and_ttl(key, doc, ChangeOrigin::Remote(peer), mgr)`
    /// on the receive path; if you genuinely have no peer attribution (a
    /// non-sync code path that needs TTL without change_tx), call
    /// `put_without_notify` + `apply_ttl` explicitly to signal that.
    #[deprecated(
        since = "0.9.0-rc.16",
        note = "Use `put_with_origin_and_ttl(.., ChangeOrigin::Remote(peer), ..)` on the receive \
                path so transitive gossip can suppress echoes (peat#891 / peat#907). Callers \
                with no peer attribution should use `put_without_notify` + apply TTL externally."
    )]
    pub fn put_without_notify_with_ttl(
        &self,
        key: &str,
        doc: &Automerge,
        ttl_manager: &TtlManager,
    ) -> Result<()> {
        self.put_without_notify(key, doc)?;
        Self::apply_ttl(key, ttl_manager)
    }

    fn apply_ttl(key: &str, ttl_manager: &TtlManager) -> Result<()> {
        let collection = key.find(['/', ':']).map(|pos| &key[..pos]).unwrap_or(key);
        if let Some(ttl) = ttl_manager.config().get_collection_ttl(collection) {
            ttl_manager.set_ttl(key, ttl)?;
        }
        Ok(())
    }

    /// Internal put implementation
    fn put_inner(
        &self,
        key: &str,
        doc: &Automerge,
        notify: bool,
        origin: ChangeOrigin,
    ) -> Result<()> {
        // Only persist to disk if we have a database (not in-memory mode)
        if let Some(ref db) = self.db {
            let bytes = self.save_doc_for_storage(doc)?;

            let write_txn = db
                .begin_write()
                .context("Failed to begin write transaction")?;
            {
                let mut table = write_txn
                    .open_table(DOCUMENTS_TABLE)
                    .context("Failed to open documents table")?;
                table
                    .insert(key.as_bytes(), bytes.as_slice())
                    .context("Failed to insert document")?;
            }
            write_txn.commit().context("Failed to commit write")?;
        }

        self.cache
            .write()
            .unwrap()
            .put(key.to_string(), doc.clone());

        // Always notify observers for ALL document changes (Issue #377)
        // This enables hierarchical aggregation to react to remotely synced docs
        let _ = self.observer_tx.send(key.to_string());

        // Notify sync coordinator only for local changes (Phase 6.3, Issue #346)
        // Skip notification for documents received via sync to avoid sync-back loops
        if notify {
            // Ignore send errors - if no one is listening, that's fine
            let _ = self.change_tx.send(key.to_string());
        }

        // Always fire the origin-tagged gossip channel (peat#891). Consumers
        // that drive onward propagation subscribe here and filter on `origin`
        // to skip echoes. The legacy `change_tx` gate above keeps the old
        // local-only-pusher contract untouched.
        let _ = self.gossip_tx.send(DocChange {
            key: key.to_string(),
            origin,
        });

        Ok(())
    }

    /// Load an Automerge document
    pub fn get(&self, key: &str) -> Result<Option<Automerge>> {
        // Always check cache first
        {
            let mut cache = self.cache.write().unwrap_or_else(|e| e.into_inner());
            if let Some(doc) = cache.get(key) {
                return Ok(Some(doc.clone()));
            }
        }

        // In memory-only mode, cache miss means document doesn't exist
        let Some(ref db) = self.db else {
            return Ok(None);
        };

        let read_txn = db
            .begin_read()
            .context("Failed to begin read transaction")?;
        let table = read_txn
            .open_table(DOCUMENTS_TABLE)
            .context("Failed to open documents table")?;

        match table.get(key.as_bytes())? {
            Some(value) => {
                let doc = self.load_doc_from_storage(value.value())?;

                self.cache
                    .write()
                    .unwrap()
                    .put(key.to_string(), doc.clone());

                Ok(Some(doc))
            }
            None => Ok(None),
        }
    }

    /// Delete a document.
    ///
    /// Treats the call as a **local** mutation and fires the same
    /// observer / sync-coordinator / origin-tagged-gossip broadcast
    /// pipeline as a local [`Self::put`] — peat-mesh#202 closure
    /// of the documented "fires for ALL document changes" contract
    /// on [`Self::subscribe_to_observer_changes`].
    ///
    /// Mirrors `put` / `put_with_origin`:
    /// - `observer_tx` always fires (CDC contract).
    /// - `change_tx` fires (local sync-out trigger).
    /// - `gossip_tx` fires with [`ChangeOrigin::Local`].
    ///
    /// Receive-side callers (the sync coordinator's tombstone-ingest
    /// path) must use [`Self::delete_with_origin`] with a
    /// `ChangeOrigin::Remote(peer_id)` instead, to suppress
    /// `change_tx` per the peat-mesh#115 ping-pong invariant.
    pub fn delete(&self, key: &str) -> Result<()> {
        self.delete_inner(key, true, ChangeOrigin::Local)
    }

    /// Delete a document with explicit origin attribution
    /// (peat-mesh#202).
    ///
    /// Mirrors the [`Self::put_with_origin`] split: `ChangeOrigin::Local`
    /// fires the local-only `change_tx` (triggering sync-out), while
    /// `ChangeOrigin::Remote(peer)` suppresses it (preventing the
    /// peat-mesh#115 ping-pong) and rides the peer attribution on the
    /// origin-tagged gossip channel so transitive-gossip drivers can
    /// skip the source peer.
    ///
    /// The sync coordinator's `apply_tombstone` path is the canonical
    /// `Remote(peer)` caller: an inbound tombstone removes the local
    /// document via this method so CDC observers see the delete event
    /// (the observer_tx fires) without the auto-sync pusher attempting
    /// to push the delete back to the originating peer.
    pub fn delete_with_origin(&self, key: &str, origin: ChangeOrigin) -> Result<()> {
        let notify_change_tx = matches!(origin, ChangeOrigin::Local);
        self.delete_inner(key, notify_change_tx, origin)
    }

    fn delete_inner(&self, key: &str, notify: bool, origin: ChangeOrigin) -> Result<()> {
        // Only delete from disk if we have a database
        if let Some(ref db) = self.db {
            let write_txn = db
                .begin_write()
                .context("Failed to begin write transaction")?;
            {
                let mut table = write_txn
                    .open_table(DOCUMENTS_TABLE)
                    .context("Failed to open documents table")?;
                table.remove(key.as_bytes())?;
            }
            write_txn.commit().context("Failed to commit delete")?;
        }

        self.cache
            .write()
            .unwrap_or_else(|e| e.into_inner())
            .pop(key);

        // peat-mesh#202: fire the same broadcast pipeline as `put_inner`
        // so a CDC consumer subscribed to `subscribe_to_observer_changes`
        // sees deletes — closing the documented "fires for ALL document
        // changes" contract that previously held for inserts/updates
        // only. Subscribers detect the delete by reading `store.get(key)`
        // on the event and observing `Ok(None)` (the pattern peat-cli's
        // `cli/observe.rs` already implements for locally-observed
        // put-then-tombstone races).
        //
        // Always notify observers for ALL document changes (Issue #377,
        // peat-mesh#202). Hierarchical aggregation reacts to remotely
        // synced docs through this channel; the channel is now
        // delete-complete.
        let _ = self.observer_tx.send(key.to_string());

        // Notify sync coordinator only for local changes (Phase 6.3,
        // Issue #346, peat-mesh#115). Skip notification for documents
        // received via sync — the originating peer already has the
        // tombstone, and re-pushing would create a ping-pong loop.
        if notify {
            let _ = self.change_tx.send(key.to_string());
        }

        // Always fire the origin-tagged gossip channel (peat#891).
        // Consumers driving onward propagation subscribe here and
        // filter on `origin` to skip the source peer. Symmetric with
        // `put_inner` — the same channel carries inserts, updates,
        // and deletes alike, attributed identically.
        let _ = self.gossip_tx.send(DocChange {
            key: key.to_string(),
            origin,
        });

        Ok(())
    }

    /// Compute a delta from `current` to `proposed` — the set of
    /// `automerge::Change` records that exist in `proposed` but not in
    /// `current` (peat-mesh#187).
    ///
    /// This is an associated function rather than a method because the
    /// computation has no dependency on store state: callers hand two
    /// `Automerge` values, get back an opaque [`AutomergeDelta`]. Pair
    /// with [`Self::apply_delta`] to advance a stored document by the
    /// computed delta.
    ///
    /// Empty result (`delta.is_empty()`) when `proposed`'s heads are a
    /// subset of `current`'s heads — i.e. nothing causally new to apply.
    ///
    /// The `peat-cli` round-trip-edit pattern (peat-node ADR-001 Phase
    /// 4b — `peat update --from <PATH>`) is the motivating consumer:
    /// query → external edit → diff against the queried state → apply
    /// the delta. Re-creating the document via `put` would break ADR-021's
    /// "create once, evolve through deltas" invariant; this path preserves
    /// it.
    pub fn diff(current: &Automerge, proposed: &Automerge) -> AutomergeDelta {
        // `Automerge::get_changes_added(other)` returns "changes in `other`
        // that are not in `self`" — exactly the delta-from-current-to-proposed
        // semantic this API contracts. See automerge 0.7.4 src/automerge.rs:1244.
        let added = current.get_changes_added(proposed);
        let changes = added.iter().map(|c| c.raw_bytes().to_vec()).collect();
        AutomergeDelta { changes }
    }

    /// Apply a delta to a stored document, advancing it through the same
    /// observer / sync-coordinator / origin-tagged-gossip notification
    /// pipeline as [`Self::put`] (peat-mesh#187).
    ///
    /// `key` must already exist in the store. Use [`Self::put`] to
    /// initialise a new document — `apply_delta` is the "evolve" half
    /// of ADR-021's create-once-evolve-through-deltas split.
    ///
    /// Applying an empty delta is a no-op (returns `Ok(())` without
    /// touching the document, the cache, or any broadcast channel).
    ///
    /// Origin attribution: the apply is treated as a local mutation —
    /// the sync coordinator's local-only `change_tx` fires, observers
    /// see [`ChangeOrigin::Local`] on the origin-tagged gossip channel,
    /// and the resulting state syncs to peers on the next push. Callers
    /// that need to apply a delta as if it were sync-received from a
    /// specific peer should use [`Self::apply_delta_with_origin`] and
    /// pass [`ChangeOrigin::Remote`] — the same suppression semantics
    /// `put_with_origin` already enforces apply here.
    ///
    /// Concurrency: the call acquires the per-key striped lock (same
    /// one [`Self::lock_doc`] returns) for the duration of the read →
    /// `apply_changes` → put cycle, so two concurrent `apply_delta`
    /// calls on the same key serialise correctly — neither writer's
    /// changes are silently dropped (the peat#868 substrate-fix
    /// invariant: any get→modify→put cycle on the same key must hold
    /// the doc lock across the whole sequence).
    ///
    /// **Concurrent advance between `diff` and `apply_delta`.** Automerge
    /// is a CRDT, so if the stored document advanced between the
    /// `diff(current, proposed)` call and this `apply_delta` call —
    /// for example, a peer's sync receive landed on the key in between
    /// — the apply still succeeds and merges correctly: the resulting
    /// state contains *both* histories, joined causally, and no data
    /// is lost. Callers expecting a last-writer-wins overwrite (the
    /// non-CRDT mental model) should re-`get`, recompute `diff`, and
    /// re-apply; the current API gives merge semantics, not overwrite
    /// semantics, by construction.
    ///
    /// **Re-applying the same delta.** The argument is `&AutomergeDelta`
    /// (not `AutomergeDelta` by-value) so callers can re-apply a delta
    /// to multiple stores or retry on transient errors without
    /// re-computing it. The per-change `Vec<u8>` clone inside this
    /// method is the cost of keeping the source `delta` reusable; if a
    /// future workload reveals the clone is in a hot path, a
    /// `into_changes` consuming variant is the obvious follow-up
    /// (deferred; the round-trip-edit consumer that motivated this
    /// API does a single apply per CLI invocation, so the clone
    /// cost is irrelevant on the unblocking path).
    pub fn apply_delta(&self, key: &str, delta: &AutomergeDelta) -> Result<()> {
        self.apply_delta_inner(key, delta, true, ChangeOrigin::Local)
    }

    /// Apply a delta with explicit origin attribution (peat-mesh#187 +
    /// peat#891 / peat#907).
    ///
    /// Mirrors the `put` / `put_with_origin` split: `ChangeOrigin::Local`
    /// fires the local-only `change_tx` (triggering sync-out), while
    /// `ChangeOrigin::Remote(peer)` suppresses it (preventing the
    /// peat-mesh#115 ping-pong) and rides the peer attribution on the
    /// origin-tagged gossip channel so transitive-gossip drivers can
    /// skip the source peer.
    pub fn apply_delta_with_origin(
        &self,
        key: &str,
        delta: &AutomergeDelta,
        origin: ChangeOrigin,
    ) -> Result<()> {
        let notify_change_tx = matches!(origin, ChangeOrigin::Local);
        self.apply_delta_inner(key, delta, notify_change_tx, origin)
    }

    fn apply_delta_inner(
        &self,
        key: &str,
        delta: &AutomergeDelta,
        notify: bool,
        origin: ChangeOrigin,
    ) -> Result<()> {
        if delta.is_empty() {
            return Ok(());
        }

        // Hold the per-key lock across the entire read-modify-write
        // cycle — same invariant the SKILL.md gotcha pins for `put`-
        // based mutations.
        let _guard = self.lock_doc(key);

        let mut doc = self.get(key)?.ok_or_else(|| {
            anyhow::anyhow!(
                "AutomergeStore::apply_delta: document {key:?} not found — \
                 use `put` to initialise a new document"
            )
        })?;

        let changes: Vec<automerge::Change> = delta
            .changes
            .iter()
            .map(|bytes| {
                automerge::Change::from_bytes(bytes.clone())
                    .context("AutomergeStore::apply_delta: decode change bytes")
            })
            .collect::<Result<_>>()?;

        doc.apply_changes(changes)
            .context("AutomergeStore::apply_delta: apply_changes")?;

        // Routes the resulting state through the same put_inner pipeline
        // as `put` / `put_with_origin` — observers fire, the sync
        // coordinator's local-only `change_tx` fires when `notify` is
        // true, and the origin-tagged gossip channel fires with the
        // caller's `origin`. Issue #187 §"Decide whether apply_delta
        // participates in the same observer-change / sync-on-change
        // pipeline as put" — yes, it does.
        self.put_inner(key, &doc, notify, origin)
    }

    /// Scan documents with prefix
    pub fn scan_prefix(&self, prefix: &str) -> Result<Vec<(String, Automerge)>> {
        // Test-only call counter (peat-mesh#131 regression test for the
        // peer-level circuit-breaker gate on `sync_all_documents_with_peer`
        // verifies this stays at zero when the breaker is open). Behind
        // cfg(test) so production builds carry no per-call atomic
        // increment.
        #[cfg(test)]
        self.scan_prefix_calls
            .fetch_add(1, std::sync::atomic::Ordering::Relaxed);

        // In memory-only mode, scan the cache
        if self.db.is_none() {
            let cache = self.cache.read().unwrap_or_else(|e| e.into_inner());
            let results: Vec<(String, Automerge)> = cache
                .iter()
                .filter(|(k, _)| k.starts_with(prefix))
                .map(|(k, v)| (k.clone(), v.clone()))
                .collect();
            return Ok(results);
        }

        let mut results = Vec::new();

        let read_txn = self
            .db
            .as_ref()
            .unwrap()
            .begin_read()
            .context("Failed to begin read transaction")?;
        let table = read_txn
            .open_table(DOCUMENTS_TABLE)
            .context("Failed to open documents table")?;

        // Use range to scan from prefix onwards
        let prefix_bytes = prefix.as_bytes();
        for entry in table.range(prefix_bytes..)? {
            let (key, value) = entry?;
            let key_bytes = key.value();

            // Stop if we've passed the prefix
            if !key_bytes.starts_with(prefix_bytes) {
                break;
            }

            let key_str = String::from_utf8_lossy(key_bytes).to_string();
            let doc = self.load_doc_from_storage(value.value())?;
            results.push((key_str, doc));
        }

        Ok(results)
    }

    /// Scan document keys with prefix, without loading any document bodies.
    ///
    /// Returns just the matching keys — the on-disk Automerge payloads are
    /// not decrypted, decoded, or held in memory at any point. Use this
    /// when the caller only needs the key set (e.g. to drive a per-key
    /// sync loop that re-loads each document individually). The
    /// `scan_prefix` companion eager-loads every Automerge document into
    /// memory simultaneously; for stores with even a handful of long-
    /// history CRDT documents that peak load is multi-GB, which under
    /// sustained mDNS rediscovery on Android tripped the OOM chain
    /// tracked in [defenseunicorns/peat#873][peat873] (and observed
    /// upstream after the [peat-mesh#131][m131] breaker gate landed —
    /// every breaker-Open→HalfOpen→Open cycle slipped one
    /// `sync_all_documents_with_peer` through, and that one call
    /// reconstructed the entire store).
    ///
    /// Mirrors `scan_prefix`'s lookup semantics exactly: same redb table
    /// range scan, same memory-only-mode cache fallback, same prefix
    /// boundary stop. The only behavioral difference is the absent
    /// `load_doc_from_storage` call per entry.
    ///
    /// [peat873]: https://github.com/defenseunicorns/peat/issues/873
    /// [m131]: https://github.com/defenseunicorns/peat-mesh/issues/131
    pub fn keys_with_prefix(&self, prefix: &str) -> Result<Vec<String>> {
        // In memory-only mode, scan the cache for matching keys.
        if self.db.is_none() {
            let cache = self.cache.read().unwrap_or_else(|e| e.into_inner());
            let results: Vec<String> = cache
                .iter()
                .filter(|(k, _)| k.starts_with(prefix))
                .map(|(k, _)| k.clone())
                .collect();
            return Ok(results);
        }

        let mut results = Vec::new();

        let read_txn = self
            .db
            .as_ref()
            .unwrap()
            .begin_read()
            .context("Failed to begin read transaction")?;
        let table = read_txn
            .open_table(DOCUMENTS_TABLE)
            .context("Failed to open documents table")?;

        let prefix_bytes = prefix.as_bytes();
        for entry in table.range(prefix_bytes..)? {
            let (key, _value) = entry?;
            let key_bytes = key.value();

            // Stop once past the prefix range
            if !key_bytes.starts_with(prefix_bytes) {
                break;
            }

            results.push(String::from_utf8_lossy(key_bytes).to_string());
        }

        Ok(results)
    }

    /// Count total documents
    pub fn count(&self) -> usize {
        // In memory-only mode, count cache entries
        let Some(ref db) = self.db else {
            return self.cache.read().unwrap_or_else(|e| e.into_inner()).len();
        };

        let read_txn = match db.begin_read() {
            Ok(txn) => txn,
            Err(_) => return 0,
        };
        let table = match read_txn.open_table(DOCUMENTS_TABLE) {
            Ok(t) => t,
            Err(_) => return 0,
        };

        table.len().unwrap_or(0) as usize
    }

    /// Subscribe to document change notifications (Phase 6.3)
    ///
    /// Returns a receiver that will receive document keys whenever documents are modified.
    /// Multiple subscribers are supported - each gets their own receiver.
    ///
    /// # Example
    ///
    /// ```ignore
    /// let store = AutomergeStore::open("./data")?;
    /// let mut rx = store.subscribe_to_changes();
    /// while let Ok(doc_key) = rx.recv().await {
    ///     println!("Document changed: {}", doc_key);
    /// }
    /// ```
    pub fn subscribe_to_changes(&self) -> broadcast::Receiver<String> {
        self.change_tx.subscribe()
    }

    /// Subscribe to observer notifications (Issue #377)
    ///
    /// Returns a receiver that receives document keys for ALL changes,
    /// including documents received via sync — inserts, updates, AND
    /// deletes (peat-mesh#202). Use this for hierarchical aggregation
    /// where you need to react to remotely synced documents (e.g.,
    /// federation coordinator reacting to cohort summaries synced from
    /// cohort coordinators), or for CDC consumers that need to see every
    /// mutation regardless of origin.
    ///
    /// Unlike [`Self::subscribe_to_changes`] which fires only for
    /// **local** writes (puts and deletes from this node), this
    /// channel fires for ALL document changes regardless of origin —
    /// local writes plus sync-received writes from peers, including
    /// tombstone-driven deletes that propagated in over the wire.
    ///
    /// Subscribers detect insert/update vs delete by reading
    /// [`Self::get`] on the event: `Ok(Some(_))` → the doc is
    /// present (insert or update); `Ok(None)` → the doc has been
    /// removed (delete, whether via local `delete` or a
    /// remote-driven tombstone apply).
    pub fn subscribe_to_observer_changes(&self) -> broadcast::Receiver<String> {
        self.observer_tx.subscribe()
    }

    /// Subscribe to origin-tagged change notifications (peat#891 / peat#907).
    ///
    /// Returns a receiver of [`DocChange`] events covering both local writes
    /// and sync-received remote applies. Each event carries a [`ChangeOrigin`]
    /// so the consumer can drive transitive gossip without echoing a change
    /// back to the peer it came from:
    ///
    /// - `ChangeOrigin::Local`: push the document to all connected peers.
    /// - `ChangeOrigin::Remote(peer_id)`: push the document to all connected
    ///   peers **except** `peer_id`.
    ///
    /// Use this subscriber when wiring a gossip pusher; `subscribe_to_changes`
    /// (local-only, untagged) is the right channel for a sync-coordinator
    /// outbound pusher that already presumes local-write semantics.
    pub fn subscribe_to_changes_with_origin(&self) -> broadcast::Receiver<DocChange> {
        self.gossip_tx.subscribe()
    }

    /// Get a collection handle for a specific namespace
    pub fn collection(self: &Arc<Self>, name: &str) -> Arc<dyn Collection> {
        Arc::new(AutomergeCollection {
            store: Arc::clone(self),
            prefix: format!("{}:", name),
        })
    }

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

    /// Store a tombstone
    ///
    /// Tombstones are stored with key format "collection:document_id"
    /// In memory-only mode, this is a no-op (tombstones aren't needed without persistence)
    pub fn put_tombstone(&self, tombstone: &crate::qos::Tombstone) -> Result<()> {
        let Some(ref db) = self.db else {
            return Ok(()); // No-op in memory mode
        };

        let key = format!("{}:{}", tombstone.collection, tombstone.document_id);
        let bytes = serde_json::to_vec(tombstone).context("Failed to serialize tombstone")?;

        let write_txn = db
            .begin_write()
            .context("Failed to begin write transaction")?;
        {
            let mut table = write_txn
                .open_table(TOMBSTONES_TABLE)
                .context("Failed to open tombstones table")?;
            table
                .insert(key.as_bytes(), bytes.as_slice())
                .context("Failed to insert tombstone")?;
        }
        write_txn
            .commit()
            .context("Failed to commit tombstone write")?;

        tracing::debug!(
            "Stored tombstone for document {} in collection {}",
            tombstone.document_id,
            tombstone.collection
        );

        Ok(())
    }

    /// Get a tombstone by collection and document ID
    pub fn get_tombstone(
        &self,
        collection: &str,
        document_id: &str,
    ) -> Result<Option<crate::qos::Tombstone>> {
        let Some(ref db) = self.db else {
            return Ok(None); // No tombstones in memory mode
        };

        let key = format!("{}:{}", collection, document_id);

        let read_txn = db
            .begin_read()
            .context("Failed to begin read transaction")?;
        let table = read_txn
            .open_table(TOMBSTONES_TABLE)
            .context("Failed to open tombstones table")?;

        match table.get(key.as_bytes())? {
            Some(value) => {
                let bytes = value.value();
                let tombstone: crate::qos::Tombstone =
                    serde_json::from_slice(bytes).context("Failed to deserialize tombstone")?;
                Ok(Some(tombstone))
            }
            None => Ok(None),
        }
    }

    /// Get all tombstones for a collection
    pub fn get_tombstones_for_collection(
        &self,
        collection: &str,
    ) -> Result<Vec<crate::qos::Tombstone>> {
        let Some(ref db) = self.db else {
            return Ok(Vec::new()); // No tombstones in memory mode
        };

        let prefix = format!("{}:", collection);
        let mut tombstones = Vec::new();

        let read_txn = db
            .begin_read()
            .context("Failed to begin read transaction")?;
        let table = read_txn
            .open_table(TOMBSTONES_TABLE)
            .context("Failed to open tombstones table")?;

        // Iterate all entries and filter by prefix
        for entry in table.iter()? {
            let (key, value) = entry?;
            let key_str = String::from_utf8_lossy(key.value());
            if key_str.starts_with(&prefix) {
                let tombstone: crate::qos::Tombstone = serde_json::from_slice(value.value())
                    .context("Failed to deserialize tombstone")?;
                tombstones.push(tombstone);
            }
        }

        Ok(tombstones)
    }

    /// Get all tombstones across all collections
    pub fn get_all_tombstones(&self) -> Result<Vec<crate::qos::Tombstone>> {
        let Some(ref db) = self.db else {
            return Ok(Vec::new()); // No tombstones in memory mode
        };

        let mut tombstones = Vec::new();

        let read_txn = db
            .begin_read()
            .context("Failed to begin read transaction")?;
        let table = read_txn
            .open_table(TOMBSTONES_TABLE)
            .context("Failed to open tombstones table")?;

        for entry in table.iter()? {
            let (_key, value) = entry?;
            let tombstone: crate::qos::Tombstone =
                serde_json::from_slice(value.value()).context("Failed to deserialize tombstone")?;
            tombstones.push(tombstone);
        }

        Ok(tombstones)
    }

    /// Remove a tombstone
    pub fn remove_tombstone(&self, collection: &str, document_id: &str) -> Result<bool> {
        let Some(ref db) = self.db else {
            return Ok(false); // No tombstones in memory mode
        };

        let key = format!("{}:{}", collection, document_id);

        let write_txn = db
            .begin_write()
            .context("Failed to begin write transaction")?;
        let existed = {
            let mut table = write_txn
                .open_table(TOMBSTONES_TABLE)
                .context("Failed to open tombstones table")?;
            let result = table.remove(key.as_bytes())?;
            result.is_some()
        };
        write_txn
            .commit()
            .context("Failed to commit tombstone removal")?;

        if existed {
            tracing::debug!(
                "Removed tombstone for document {} in collection {}",
                document_id,
                collection
            );
        }

        Ok(existed)
    }

    /// Check if a tombstone exists
    pub fn has_tombstone(&self, collection: &str, document_id: &str) -> Result<bool> {
        Ok(self.get_tombstone(collection, document_id)?.is_some())
    }

    // === Garbage Collection support methods (ADR-034 Phase 3) ===

    /// Get list of all collections (by scanning document key prefixes)
    pub fn list_collections(&self) -> Result<Vec<String>> {
        let mut collections = std::collections::HashSet::new();

        // In memory-only mode, scan the cache
        if self.db.is_none() {
            let cache = self.cache.read().unwrap_or_else(|e| e.into_inner());
            for key in cache.iter().map(|(k, _)| k) {
                if let Some(colon_pos) = key.find(':') {
                    let collection = &key[..colon_pos];
                    collections.insert(collection.to_string());
                }
            }
            return Ok(collections.into_iter().collect());
        }

        let read_txn = self
            .db
            .as_ref()
            .unwrap()
            .begin_read()
            .context("Failed to begin read transaction")?;
        let table = read_txn
            .open_table(DOCUMENTS_TABLE)
            .context("Failed to open documents table")?;

        for result in table.iter().context("Failed to iterate documents")? {
            let (key, _) = result.context("Failed to read document entry")?;
            let key_str =
                std::str::from_utf8(key.value()).context("Invalid UTF-8 in document key")?;

            // Keys are formatted as "collection:document_id"
            if let Some(colon_pos) = key_str.find(':') {
                let collection = &key_str[..colon_pos];
                collections.insert(collection.to_string());
            }
        }

        Ok(collections.into_iter().collect())
    }

    /// Get documents in a collection that were created before the cutoff time
    ///
    /// This checks the _created_at field stored in the Automerge document.
    /// Used for ImplicitTTL garbage collection.
    pub fn get_expired_documents(
        &self,
        collection: &str,
        cutoff: std::time::SystemTime,
    ) -> Result<Vec<String>> {
        let prefix = format!("{}:", collection);
        let docs = self.scan_prefix(&prefix)?;
        let mut expired = Vec::new();

        let cutoff_ms = cutoff
            .duration_since(std::time::UNIX_EPOCH)
            .unwrap_or_default()
            .as_millis() as u64;

        for (key, doc) in docs {
            // Extract _created_at timestamp if present
            if let Ok(Some((automerge::Value::Scalar(scalar), _))) =
                doc.get(automerge::ROOT, "_created_at")
            {
                if let automerge::ScalarValue::Uint(created_at) = scalar.as_ref() {
                    if *created_at < cutoff_ms {
                        // Document is older than cutoff
                        if let Some(doc_id) = key.strip_prefix(&prefix) {
                            expired.push(doc_id.to_string());
                        }
                    }
                }
            }
        }

        Ok(expired)
    }

    /// Hard delete a document (permanent removal, no tombstone created)
    ///
    /// Used by garbage collection for ImplicitTTL collections where
    /// tombstones are not needed.
    pub fn hard_delete(&self, collection: &str, document_id: &str) -> Result<()> {
        let key = format!("{}:{}", collection, document_id);
        self.delete(&key)?;
        tracing::debug!(
            "Hard deleted document {} from collection {}",
            document_id,
            collection
        );
        Ok(())
    }

    // === Document Compaction (Issue #401 - Memory Blowout Fix) ===

    /// Compact a document by discarding CRDT history
    ///
    /// Automerge documents accumulate operation history with every change.
    /// This method replaces the document with a forked copy that contains
    /// only the current state, freeing memory used by historical operations.
    ///
    /// # When to use
    ///
    /// - After many updates to high-frequency documents (beacons, node-states)
    /// - When memory pressure is detected
    /// - Periodically for long-running simulations
    ///
    /// # Returns
    ///
    /// - `Ok(Some(old_size, new_size))` - Document was compacted, returns sizes before/after
    /// - `Ok(None)` - Document not found
    /// - `Err(_)` - Compaction failed
    ///
    /// # Example
    ///
    /// ```ignore
    /// let store = AutomergeStore::open("./data")?;
    /// if let Some((old, new)) = store.compact("node-states:soldier-1")? {
    ///     tracing::info!("Compacted {} -> {} bytes ({}% reduction)", old, new, 100 - (new * 100 / old));
    /// }
    /// ```
    pub fn compact(&self, key: &str) -> Result<Option<(usize, usize)>> {
        // Hold per-document striped lock across the entire get→fork→put
        // sequence to prevent concurrent sync writes from being lost (Issue #74).
        let _guard = self.lock_doc(key);
        let doc = match self.get(key)? {
            Some(d) => d,
            None => return Ok(None),
        };

        let old_size = doc.save().len();

        // Fork creates a new document with current state but no history
        let compacted = doc.fork();
        let new_size = compacted.save().len();

        // Save the compacted document (without triggering sync notification)
        self.put_without_notify(key, &compacted)?;

        tracing::debug!(
            "Compacted document {}: {} -> {} bytes ({:.1}% reduction)",
            key,
            old_size,
            new_size,
            if old_size > 0 {
                100.0 - (new_size as f64 * 100.0 / old_size as f64)
            } else {
                0.0
            }
        );

        Ok(Some((old_size, new_size)))
    }

    /// Compact all documents with a given prefix
    ///
    /// Useful for batch-compacting all documents in a collection.
    ///
    /// # Returns
    ///
    /// `(documents_compacted, total_bytes_before, total_bytes_after)`
    ///
    /// # Example
    ///
    /// ```ignore
    /// let store = AutomergeStore::open("./data")?;
    /// let (count, before, after) = store.compact_prefix("node-states:")?;
    /// tracing::info!("Compacted {} documents: {} -> {} bytes", count, before, after);
    /// ```
    pub fn compact_prefix(&self, prefix: &str) -> Result<(usize, usize, usize)> {
        let docs = self.scan_prefix(prefix)?;
        let mut count = 0;
        let mut total_before = 0;
        let mut total_after = 0;

        for (key, _) in docs {
            if let Some((before, after)) = self.compact(&key)? {
                count += 1;
                total_before += before;
                total_after += after;
            }
        }

        if count > 0 {
            tracing::info!(
                "Compacted {} documents with prefix '{}': {} -> {} bytes ({:.1}% reduction)",
                count,
                prefix,
                total_before,
                total_after,
                if total_before > 0 {
                    100.0 - (total_after as f64 * 100.0 / total_before as f64)
                } else {
                    0.0
                }
            );
        }

        Ok((count, total_before, total_after))
    }

    /// Compact all documents in the store
    ///
    /// # Returns
    ///
    /// `(documents_compacted, total_bytes_before, total_bytes_after)`
    pub fn compact_all(&self) -> Result<(usize, usize, usize)> {
        // In memory-only mode, iterate cache
        if self.db.is_none() {
            let keys: Vec<String> = {
                let cache = self.cache.read().unwrap_or_else(|e| e.into_inner());
                cache.iter().map(|(k, _)| k.clone()).collect()
            };

            let mut count = 0;
            let mut total_before = 0;
            let mut total_after = 0;

            for key in keys {
                if let Some((before, after)) = self.compact(&key)? {
                    count += 1;
                    total_before += before;
                    total_after += after;
                }
            }

            return Ok((count, total_before, total_after));
        }

        // With disk persistence, scan all documents
        let read_txn = self
            .db
            .as_ref()
            .unwrap()
            .begin_read()
            .context("Failed to begin read transaction")?;
        let table = read_txn
            .open_table(DOCUMENTS_TABLE)
            .context("Failed to open documents table")?;

        let keys: Vec<String> = table
            .iter()?
            .filter_map(|entry| {
                entry
                    .ok()
                    .map(|(k, _)| String::from_utf8_lossy(k.value()).to_string())
            })
            .collect();

        drop(table);
        drop(read_txn);

        let mut count = 0;
        let mut total_before = 0;
        let mut total_after = 0;

        for key in keys {
            if let Some((before, after)) = self.compact(&key)? {
                count += 1;
                total_before += before;
                total_after += after;
            }
        }

        if count > 0 {
            tracing::info!(
                "Compacted {} documents: {} -> {} bytes ({:.1}% reduction)",
                count,
                total_before,
                total_after,
                if total_before > 0 {
                    100.0 - (total_after as f64 * 100.0 / total_before as f64)
                } else {
                    0.0
                }
            );
        }

        Ok((count, total_before, total_after))
    }

    /// Start a background task that periodically compacts documents in
    /// specific collections that exceed a size threshold.
    ///
    /// Only `LatestOnly` sync-mode collections should be compacted — compacting
    /// `FullHistory` collections destroys change history needed for delta sync.
    ///
    /// - `interval`: How often to run compaction (default: 5 minutes)
    /// - `size_threshold_bytes`: Only compact documents larger than this (default: 64 KiB)
    /// - `collections`: Collection prefixes to compact (e.g., `["beacons", "platforms"]`)
    pub fn start_background_compaction(
        self: &Arc<Self>,
        interval: std::time::Duration,
        size_threshold_bytes: usize,
        collections: Vec<String>,
        token: tokio_util::sync::CancellationToken,
    ) {
        let store = Arc::clone(self);
        tokio::spawn(async move {
            let mut timer = tokio::time::interval(interval);
            // Don't run immediately on startup
            timer.tick().await;

            loop {
                tokio::select! {
                    _ = token.cancelled() => {
                        tracing::info!("background compaction cancelled");
                        break;
                    }
                    _ = timer.tick() => {
                        match store.compact_collections_above_threshold(&collections, size_threshold_bytes) {
                            Ok((count, before, after)) => {
                                if count > 0 {
                                    tracing::info!(count, before, after, "background compaction complete");
                                }
                            }
                            Err(e) => {
                                tracing::warn!("background compaction failed: {e}");
                            }
                        }
                    }
                }
            }
        });
    }

    /// Compact documents in specific collections that exceed a size threshold.
    ///
    /// Only scans documents matching the given collection prefixes (e.g.,
    /// `"beacons"` matches `"beacons:beacon-1"`, `"beacons:beacon-2"`, etc.).
    ///
    /// Returns `(documents_compacted, total_bytes_before, total_bytes_after)`.
    pub fn compact_collections_above_threshold(
        &self,
        collections: &[String],
        threshold_bytes: usize,
    ) -> Result<(usize, usize, usize)> {
        let mut count = 0;
        let mut total_before = 0;
        let mut total_after = 0;

        for collection in collections {
            let prefix = format!("{}:", collection);
            let docs = self.scan_prefix(&prefix)?;
            for (key, _) in docs {
                let size = self.document_size(&key)?.unwrap_or(0);
                if size >= threshold_bytes {
                    if let Some((before, after)) = self.compact(&key)? {
                        count += 1;
                        total_before += before;
                        total_after += after;
                    }
                }
            }
        }

        Ok((count, total_before, total_after))
    }

    /// Compact only documents exceeding a size threshold.
    ///
    /// Returns `(documents_compacted, total_bytes_before, total_bytes_after)`.
    pub fn compact_above_threshold(&self, threshold_bytes: usize) -> Result<(usize, usize, usize)> {
        let keys = self.all_keys()?;
        let mut count = 0;
        let mut total_before = 0;
        let mut total_after = 0;

        for key in keys {
            let size = self.document_size(&key)?.unwrap_or(0);
            if size >= threshold_bytes {
                if let Some((before, after)) = self.compact(&key)? {
                    count += 1;
                    total_before += before;
                    total_after += after;
                }
            }
        }

        Ok((count, total_before, total_after))
    }

    /// List all document keys in the store.
    fn all_keys(&self) -> Result<Vec<String>> {
        if self.db.is_none() {
            let cache = self.cache.read().unwrap_or_else(|e| e.into_inner());
            return Ok(cache.iter().map(|(k, _)| k.clone()).collect());
        }

        let read_txn = self
            .db
            .as_ref()
            .unwrap()
            .begin_read()
            .context("Failed to begin read transaction")?;
        let table = read_txn
            .open_table(DOCUMENTS_TABLE)
            .context("Failed to open documents table")?;

        let keys: Vec<String> = table
            .iter()?
            .filter_map(|entry| {
                entry
                    .ok()
                    .map(|(k, _)| String::from_utf8_lossy(k.value()).to_string())
            })
            .collect();

        Ok(keys)
    }

    /// Get a typed collection handle for serde-based access.
    ///
    /// Returns a `TypedCollection<T>` that provides automatic serde
    /// serialization/deserialization, integrated queries, and
    /// prefix-filtered change subscriptions.
    ///
    /// # Example
    ///
    /// ```ignore
    /// let sensors = store.typed_collection::<SensorReading>("sensors");
    /// sensors.upsert("r001", &reading)?;
    /// let result = sensors.get("r001")?;
    /// ```
    pub fn typed_collection<T: serde::Serialize + serde::de::DeserializeOwned>(
        self: &Arc<Self>,
        name: &str,
    ) -> super::typed_collection::TypedCollection<T> {
        super::typed_collection::TypedCollection::new(Arc::clone(self), name)
    }

    /// Get the serialized size of a document (for monitoring)
    pub fn document_size(&self, key: &str) -> Result<Option<usize>> {
        match self.get(key)? {
            Some(doc) => Ok(Some(doc.save().len())),
            None => Ok(None),
        }
    }

    /// Read the persisted Lamport-clock high-water mark, if any
    /// (peat-mesh#195).
    ///
    /// Returns `Ok(None)` when running in-memory (no `db`) or when the
    /// metadata key has never been written — both shapes route the
    /// `AutomergeBackend` seed path to its wall-clock fallback. Returns
    /// `Ok(Some(value))` with the persisted high-water on disk-backed
    /// stores that have previously called
    /// [`Self::write_lamport_highwater`].
    ///
    /// The value is stored as 8 bytes of little-endian `u64`; a stored
    /// blob of any other length is treated as a corruption indicator
    /// and surfaces as `Err` rather than silently truncating to a
    /// possibly-wrong value.
    pub fn read_lamport_highwater(&self) -> Result<Option<u64>> {
        let Some(ref db) = self.db else {
            return Ok(None);
        };
        let read_txn = db
            .begin_read()
            .context("read_lamport_highwater: begin_read")?;
        let table = match read_txn.open_table(METADATA_TABLE) {
            Ok(t) => t,
            // Table-not-found is the "never-written" case on stores
            // that pre-date the metadata table.
            Err(redb::TableError::TableDoesNotExist(_)) => return Ok(None),
            Err(e) => {
                return Err(anyhow::Error::new(e).context("read_lamport_highwater: open_table"));
            }
        };
        let Some(value) = table
            .get(METADATA_KEY_LAMPORT_HIGHWATER)
            .context("read_lamport_highwater: get")?
        else {
            return Ok(None);
        };
        let bytes = value.value();
        if bytes.len() != 8 {
            // Corrupt entry (wrong length) is treated as `None` to
            // align with `write_lamport_highwater`'s recovery policy
            // (which overwrites corrupt entries with the in-memory
            // value). The previous "fail-closed on read, fail-open on
            // write" split was asymmetric — silently recoverable in
            // some call orders, hard-fatal in others. Logging + None
            // surfaces the corruption to operators without aborting
            // backend construction, and the subsequent first periodic
            // flush will overwrite the corrupt blob with a valid one.
            tracing::warn!(
                expected_bytes = 8,
                actual_bytes = bytes.len(),
                "read_lamport_highwater: corrupt entry — treating as None and \
                 letting the next write_lamport_highwater overwrite"
            );
            return Ok(None);
        }
        Ok(Some(u64::from_le_bytes(bytes.try_into().unwrap())))
    }

    /// Persist the Lamport-clock high-water mark to redb
    /// (peat-mesh#195).
    ///
    /// **Monotonic write semantics.** The write is contained inside a
    /// single redb transaction that reads the existing value first and
    /// commits only when `value > existing`. This closes the
    /// concurrent-flush race the rc.28 QA review flagged on the
    /// initial #197 surface: if a periodic flush task reads the atomic
    /// at time T1 and an explicit `flush_lamport_highwater` writes a
    /// strictly-larger value at T4, the periodic flush arriving at T5
    /// observes the post-T4 value as `existing > value` and skips the
    /// regressing write. Outcome: the persisted high-water is
    /// monotonic regardless of how the periodic and explicit flush
    /// paths interleave.
    ///
    /// No-op (returns `Ok(())`) on in-memory stores — there's no disk
    /// to write to and the wall-clock seed path is the only baseline
    /// available there.
    pub fn write_lamport_highwater(&self, value: u64) -> Result<()> {
        let Some(ref db) = self.db else {
            return Ok(());
        };
        let write_txn = db
            .begin_write()
            .context("write_lamport_highwater: begin_write")?;
        {
            let mut table = write_txn
                .open_table(METADATA_TABLE)
                .context("write_lamport_highwater: open_table")?;

            // Read existing first — under the redb write-txn lock, so
            // no concurrent writer can interleave between the read and
            // the insert below. Extract the existing `u64` (if any)
            // into a scope-bounded local so the `AccessGuard`'s
            // borrow of `table` is released before the subsequent
            // `table.insert`.
            let existing: Option<u64> = {
                let guard = table
                    .get(METADATA_KEY_LAMPORT_HIGHWATER)
                    .context("write_lamport_highwater: read existing")?;
                guard.and_then(|g| {
                    let raw = g.value();
                    if raw.len() == 8 {
                        Some(u64::from_le_bytes(raw.try_into().unwrap()))
                    } else {
                        // Corrupt entry (wrong length): treat as
                        // "no existing high-water" so the upcoming
                        // insert overwrites it. The in-memory atomic
                        // is the authoritative source for the current
                        // run; a corrupt historical entry is best
                        // replaced rather than preserved.
                        None
                    }
                })
            };

            // Monotonic-write check: skip the insert when the existing
            // value is ≥ `value`. The transaction still commits with
            // empty changes — cheap, and keeps the lock-release shape
            // uniform across both branches.
            if existing.is_some_and(|e| e >= value) {
                drop(table);
                write_txn
                    .commit()
                    .context("write_lamport_highwater: commit (no-op)")?;
                return Ok(());
            }

            table
                .insert(METADATA_KEY_LAMPORT_HIGHWATER, &value.to_le_bytes()[..])
                .context("write_lamport_highwater: insert")?;
        }
        write_txn
            .commit()
            .context("write_lamport_highwater: commit")?;
        Ok(())
    }
}

/// Collection implementation for AutomergeStore
///
/// Wraps AutomergeStore and provides Collection trait implementation.
/// Uses key prefixing to namespace collections (e.g., "cells:cell-1", "nodes:node-1").
pub struct AutomergeCollection {
    store: Arc<AutomergeStore>,
    prefix: String,
}

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

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

impl Collection for AutomergeCollection {
    fn upsert(&self, doc_id: &str, data: Vec<u8>) -> Result<()> {
        // Get existing document or create a new one
        // This is critical for CRDT sync: we must UPDATE the existing document
        // rather than replacing it with a new one. If we create a new document,
        // Automerge will see it as a conflicting branch and may pick the wrong
        // value during merge.
        let key = self.prefixed_key(doc_id);
        let mut doc = match self.store.get(&key)? {
            Some(existing) => {
                // Fork the existing document to update it
                existing.fork()
            }
            None => {
                // No existing doc, create a new one
                Automerge::new()
            }
        };

        match doc.transact(|tx| {
            tx.put(
                automerge::ROOT,
                "data",
                automerge::ScalarValue::Bytes(data.clone()),
            )?;
            Ok::<(), automerge::AutomergeError>(())
        }) {
            Ok(_) => self.store.put(&key, &doc),
            Err(e) => Err(anyhow::anyhow!(
                "Failed to update Automerge document: {:?}",
                e
            )),
        }
    }

    fn get(&self, doc_id: &str) -> Result<Option<Vec<u8>>> {
        match self.store.get(&self.prefixed_key(doc_id))? {
            Some(doc) => {
                // Extract bytes from Automerge document
                if let Ok(Some((automerge::Value::Scalar(scalar), _))) =
                    doc.get(automerge::ROOT, "data")
                {
                    if let automerge::ScalarValue::Bytes(bytes) = scalar.as_ref() {
                        return Ok(Some(bytes.to_vec()));
                    }
                }
                Ok(None)
            }
            None => Ok(None),
        }
    }

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

    fn scan(&self) -> Result<Vec<(String, Vec<u8>)>> {
        let docs = self.store.scan_prefix(&self.prefix)?;
        tracing::debug!(
            "AutomergeCollection.scan: prefix={}, found {} docs",
            self.prefix,
            docs.len()
        );
        let mut results = Vec::new();

        for (key, doc) in docs {
            tracing::debug!(
                "AutomergeCollection.scan: processing key={}, doc_len={}",
                key,
                doc.save().len()
            );
            if let Some(doc_id) = self.strip_prefix(&key) {
                match doc.get(automerge::ROOT, "data") {
                    Ok(Some((automerge::Value::Scalar(scalar), _))) => {
                        if let automerge::ScalarValue::Bytes(bytes) = scalar.as_ref() {
                            tracing::debug!(
                                "AutomergeCollection.scan: found data bytes, doc_id={}, len={}",
                                doc_id,
                                bytes.len()
                            );
                            results.push((doc_id.to_string(), bytes.to_vec()));
                        } else {
                            tracing::debug!(
                                "AutomergeCollection.scan: data is not Bytes, doc_id={}",
                                doc_id
                            );
                        }
                    }
                    Ok(Some((value, _))) => {
                        tracing::debug!(
                            "AutomergeCollection.scan: data is not Scalar, doc_id={}, value_type={:?}",
                            doc_id,
                            value
                        );
                    }
                    Ok(None) => {
                        tracing::debug!(
                            "AutomergeCollection.scan: no 'data' field, doc_id={}",
                            doc_id
                        );
                    }
                    Err(e) => {
                        tracing::debug!(
                            "AutomergeCollection.scan: error getting 'data', doc_id={}, err={}",
                            doc_id,
                            e
                        );
                    }
                }
            }
        }

        Ok(results)
    }

    fn find(&self, predicate: DocumentPredicate) -> Result<Vec<(String, Vec<u8>)>> {
        let all_docs = self.scan()?;
        Ok(all_docs
            .into_iter()
            .filter(|(_, bytes)| predicate(bytes))
            .collect())
    }

    fn query_geohash_prefix(&self, geohash_prefix: &str) -> Result<Vec<(String, Vec<u8>)>> {
        // For AutomergeStore, geohash queries require the geohash to be in the key
        // This is a simplified implementation - in Phase 2 we'll add proper indexing
        let all_docs = self.scan()?;
        Ok(all_docs
            .into_iter()
            .filter(|(id, _)| id.starts_with(geohash_prefix))
            .collect())
    }

    fn count(&self) -> Result<usize> {
        // Use keys_with_prefix rather than scan()?.len(): scan()
        // eager-loads every Automerge document via load_doc_from_storage,
        // then .len() throws all of it away. Same over-fetch class
        // fixed in sync_all_documents_with_peer / get_local_sync_items
        // / TypedCollection::count (defenseunicorns/peat#873,
        // peat-mesh#135).
        Ok(self.store.keys_with_prefix(&self.prefix)?.len())
    }
}

// === GcStore trait implementation for AutomergeStore (ADR-034 Phase 3) ===
//
// Both GcStore trait and AutomergeStore type are in peat-mesh,
// so the impl lives here (orphan rule satisfied).

impl crate::qos::GcStore for AutomergeStore {
    fn get_all_tombstones(&self) -> anyhow::Result<Vec<crate::qos::Tombstone>> {
        self.get_all_tombstones()
    }

    fn remove_tombstone(&self, collection: &str, document_id: &str) -> anyhow::Result<bool> {
        self.remove_tombstone(collection, document_id)
    }

    fn has_tombstone(&self, collection: &str, document_id: &str) -> anyhow::Result<bool> {
        self.has_tombstone(collection, document_id)
    }

    fn get_expired_documents(
        &self,
        collection: &str,
        cutoff: std::time::SystemTime,
    ) -> anyhow::Result<Vec<String>> {
        self.get_expired_documents(collection, cutoff)
    }

    fn hard_delete(&self, collection: &str, document_id: &str) -> anyhow::Result<()> {
        self.hard_delete(collection, document_id)
    }

    fn list_collections(&self) -> anyhow::Result<Vec<String>> {
        self.list_collections()
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use tempfile::TempDir;

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

    #[test]
    fn test_collection_upsert_and_get() {
        let (store, _temp) = create_test_store();
        let collection = store.collection("test");

        let data = b"test data".to_vec();
        collection.upsert("doc1", data.clone()).unwrap();

        let retrieved = collection.get("doc1").unwrap().unwrap();
        assert_eq!(retrieved, data);
    }

    #[test]
    fn test_collection_scan() {
        let (store, _temp) = create_test_store();
        let collection = store.collection("test");

        collection.upsert("doc1", b"data1".to_vec()).unwrap();
        collection.upsert("doc2", b"data2".to_vec()).unwrap();

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

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

    #[test]
    fn test_collection_delete() {
        let (store, _temp) = create_test_store();
        let collection = store.collection("test");

        collection.upsert("doc1", b"data1".to_vec()).unwrap();
        assert!(collection.get("doc1").unwrap().is_some());

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

    #[test]
    fn test_collection_count() {
        let (store, _temp) = create_test_store();
        let collection = store.collection("test");

        assert_eq!(collection.count().unwrap(), 0);

        collection.upsert("doc1", b"data1".to_vec()).unwrap();
        collection.upsert("doc2", b"data2".to_vec()).unwrap();

        assert_eq!(collection.count().unwrap(), 2);

        // peat-mesh#135 / peat#873: count() must not call scan_prefix —
        // that pays the full Automerge::load cost per entry just to take
        // .len(). The structural pin against accidentally reverting to
        // `scan_prefix(...).len()`. If this counter ever becomes
        // non-zero through count() invocations, the lighter path has
        // regressed and the OOM-class allocation pattern is back.
        //
        // Note: `store.collection(name)` returns `Arc<dyn Collection>`
        // backed by `AutomergeCollection`, so this pins the
        // `AutomergeCollection::count` path specifically. The
        // `TypedCollection::count` path is pinned by an analogous
        // assertion in `typed_collection.rs::test_typed_count`.
        let baseline = store.scan_prefix_call_count();
        for _ in 0..3 {
            let _ = collection.count().unwrap();
        }
        assert_eq!(
            store.scan_prefix_call_count(),
            baseline,
            "AutomergeCollection::count must not call scan_prefix (peat-mesh#135 / peat#873)"
        );
    }

    #[test]
    fn test_collection_find_with_predicate() {
        let (store, _temp) = create_test_store();
        let collection = store.collection("test");

        collection.upsert("doc1", b"hello".to_vec()).unwrap();
        collection.upsert("doc2", b"world".to_vec()).unwrap();
        collection.upsert("doc3", b"hello world".to_vec()).unwrap();

        let results = collection
            .find(Box::new(|bytes| {
                String::from_utf8_lossy(bytes).contains("hello")
            }))
            .unwrap();

        assert_eq!(results.len(), 2);
    }

    #[test]
    fn test_collection_namespace_isolation() {
        let (store, _temp) = create_test_store();
        let collection1 = store.collection("coll1");
        let collection2 = store.collection("coll2");

        collection1.upsert("doc1", b"data1".to_vec()).unwrap();
        collection2.upsert("doc1", b"data2".to_vec()).unwrap();

        let data1 = collection1.get("doc1").unwrap().unwrap();
        let data2 = collection2.get("doc1").unwrap().unwrap();

        assert_eq!(data1, b"data1");
        assert_eq!(data2, b"data2");
        assert_ne!(data1, data2);
    }

    #[test]
    fn test_direct_put_and_get() {
        let (store, _temp) = create_test_store();

        let mut doc = Automerge::new();
        doc.transact::<_, _, automerge::AutomergeError>(|tx| {
            tx.put(automerge::ROOT, "key", "value")?;
            Ok(())
        })
        .unwrap();

        store.put("test-doc", &doc).unwrap();

        let loaded = store.get("test-doc").unwrap().unwrap();
        let value: String = loaded
            .get(automerge::ROOT, "key")
            .unwrap()
            .unwrap()
            .0
            .to_string();
        assert!(value.contains("value"));
    }

    #[test]
    fn test_scan_prefix() {
        let (store, _temp) = create_test_store();

        let mut doc1 = Automerge::new();
        doc1.transact::<_, _, automerge::AutomergeError>(|tx| {
            tx.put(automerge::ROOT, "n", "1")?;
            Ok(())
        })
        .unwrap();

        let mut doc2 = Automerge::new();
        doc2.transact::<_, _, automerge::AutomergeError>(|tx| {
            tx.put(automerge::ROOT, "n", "2")?;
            Ok(())
        })
        .unwrap();

        store.put("prefix:a", &doc1).unwrap();
        store.put("prefix:b", &doc2).unwrap();
        store.put("other:c", &doc1).unwrap();

        let results = store.scan_prefix("prefix:").unwrap();
        assert_eq!(results.len(), 2);
    }

    #[test]
    fn test_keys_with_prefix_matches_scan_prefix_keys_without_loading_docs() {
        // peat#873: this exists because `sync_all_documents_with_peer`
        // (and `get_local_sync_items` / Negentropy probes) iterate
        // doc-keys but were paying the full `Automerge::load` cost for
        // every persisted document via `scan_prefix`. The new
        // `keys_with_prefix` API must:
        //   (a) return exactly the same keys `scan_prefix` does for a
        //       given prefix, and
        //   (b) not increment the `scan_prefix_call_count` test counter
        //       — the differentiating signal that proves it took the
        //       lighter path that the gated #131 test also relies on.
        let (store, _temp) = create_test_store();

        let mut doc1 = Automerge::new();
        doc1.transact::<_, _, automerge::AutomergeError>(|tx| {
            tx.put(automerge::ROOT, "n", "1")?;
            Ok(())
        })
        .unwrap();
        let mut doc2 = Automerge::new();
        doc2.transact::<_, _, automerge::AutomergeError>(|tx| {
            tx.put(automerge::ROOT, "n", "2")?;
            Ok(())
        })
        .unwrap();

        store.put("prefix:a", &doc1).unwrap();
        store.put("prefix:b", &doc2).unwrap();
        store.put("other:c", &doc1).unwrap();

        let baseline_scan_calls = store.scan_prefix_call_count();

        let mut keys = store.keys_with_prefix("prefix:").unwrap();
        keys.sort();
        assert_eq!(keys, vec!["prefix:a".to_string(), "prefix:b".to_string()]);

        // The lighter path must not have invoked scan_prefix.
        assert_eq!(
            store.scan_prefix_call_count(),
            baseline_scan_calls,
            "keys_with_prefix must not call scan_prefix (that would defeat the peat#873 fix)"
        );

        // Empty prefix yields all keys, still without scan_prefix.
        let mut all = store.keys_with_prefix("").unwrap();
        all.sort();
        assert_eq!(
            all,
            vec![
                "other:c".to_string(),
                "prefix:a".to_string(),
                "prefix:b".to_string()
            ]
        );
        assert_eq!(
            store.scan_prefix_call_count(),
            baseline_scan_calls,
            "keys_with_prefix(\"\") must not call scan_prefix either"
        );
    }

    #[test]
    fn test_keys_with_prefix_memory_only_mode() {
        // The disk-backed branch is covered by the test above. This pins
        // the parallel memory-only-mode branch in `keys_with_prefix`
        // (line ~506): both branches must return the same key set as
        // their `scan_prefix` analogues without paying `Automerge::load`.
        // The two branches are structurally identical aside from the
        // backing store (LruCache vs redb table), and a regression in
        // either side would slip past the disk-only test.
        let store = Arc::new(AutomergeStore::in_memory());
        assert!(
            store.is_in_memory(),
            "precondition: this test exercises the memory-only branch"
        );

        let mut doc = Automerge::new();
        doc.transact::<_, _, automerge::AutomergeError>(|tx| {
            tx.put(automerge::ROOT, "n", "1")?;
            Ok(())
        })
        .unwrap();

        store.put("prefix:a", &doc).unwrap();
        store.put("prefix:b", &doc).unwrap();
        store.put("other:c", &doc).unwrap();

        let baseline_scan_calls = store.scan_prefix_call_count();

        let mut keys = store.keys_with_prefix("prefix:").unwrap();
        keys.sort();
        assert_eq!(keys, vec!["prefix:a".to_string(), "prefix:b".to_string()]);

        let mut all = store.keys_with_prefix("").unwrap();
        all.sort();
        assert_eq!(
            all,
            vec![
                "other:c".to_string(),
                "prefix:a".to_string(),
                "prefix:b".to_string()
            ]
        );

        // Memory-only mode must also avoid scan_prefix — the load-avoidance
        // invariant is global to keys_with_prefix, not just the redb branch.
        assert_eq!(
            store.scan_prefix_call_count(),
            baseline_scan_calls,
            "keys_with_prefix in memory-only mode must not call scan_prefix"
        );
    }

    // === Compaction Tests (Issue #401 - Memory Blowout Fix) ===

    #[test]
    fn test_compact_document() {
        let (store, _temp) = create_test_store();

        // Create a document and update it many times to build up history
        let mut doc = Automerge::new();
        for i in 0..100 {
            doc.transact::<_, _, automerge::AutomergeError>(|tx| {
                tx.put(automerge::ROOT, "counter", i as i64)?;
                Ok(())
            })
            .unwrap();
        }

        store.put("test-doc", &doc).unwrap();
        let size_before = store.document_size("test-doc").unwrap().unwrap();

        // Compact the document
        let result = store.compact("test-doc").unwrap();
        assert!(result.is_some());
        let (old_size, new_size) = result.unwrap();

        assert_eq!(old_size, size_before);
        assert!(
            new_size <= old_size,
            "Compaction should reduce or maintain size"
        );

        // Verify the document still has the correct value
        let loaded = store.get("test-doc").unwrap().unwrap();
        let value = loaded.get(automerge::ROOT, "counter").unwrap().unwrap();
        assert_eq!(value.0.to_i64(), Some(99));
    }

    #[test]
    fn test_compact_nonexistent_document() {
        let (store, _temp) = create_test_store();
        let result = store.compact("nonexistent").unwrap();
        assert!(result.is_none());
    }

    #[test]
    fn test_compact_prefix() {
        let (store, _temp) = create_test_store();

        // Create multiple documents with history
        for doc_num in 0..5 {
            let mut doc = Automerge::new();
            for i in 0..50 {
                doc.transact::<_, _, automerge::AutomergeError>(|tx| {
                    tx.put(automerge::ROOT, "counter", i as i64)?;
                    Ok(())
                })
                .unwrap();
            }
            store.put(&format!("test:{}", doc_num), &doc).unwrap();
        }

        // Add a document with different prefix
        let mut other_doc = Automerge::new();
        other_doc
            .transact::<_, _, automerge::AutomergeError>(|tx| {
                tx.put(automerge::ROOT, "other", "value")?;
                Ok(())
            })
            .unwrap();
        store.put("other:1", &other_doc).unwrap();

        // Compact only "test:" prefix
        let (count, before, after) = store.compact_prefix("test:").unwrap();
        assert_eq!(count, 5);
        assert!(before > 0);
        assert!(after <= before);

        // Verify other document was not affected
        let other = store.get("other:1").unwrap().unwrap();
        let value = other.get(automerge::ROOT, "other").unwrap().unwrap();
        assert!(value.0.to_str().unwrap().contains("value"));
    }

    #[test]
    fn test_compact_all() {
        let (store, _temp) = create_test_store();

        // Create documents with history
        for prefix in &["a", "b", "c"] {
            let mut doc = Automerge::new();
            for i in 0..30 {
                doc.transact::<_, _, automerge::AutomergeError>(|tx| {
                    tx.put(automerge::ROOT, "counter", i as i64)?;
                    Ok(())
                })
                .unwrap();
            }
            store.put(&format!("{}:doc", prefix), &doc).unwrap();
        }

        let (count, before, after) = store.compact_all().unwrap();
        assert_eq!(count, 3);
        assert!(before > 0);
        assert!(after <= before);
    }

    #[test]
    fn test_compact_in_memory_store() {
        let store = Arc::new(AutomergeStore::in_memory());

        // Create a document with multi-actor history to exercise compaction.
        // fork() discards the change log (individual ops from each actor),
        // which is where real savings come from in multi-peer CRDT sync.
        // Single-actor docs may not shrink much since the compressed binary
        // format is already efficient.
        let mut doc = Automerge::new();
        for i in 0..100 {
            doc.transact::<_, _, automerge::AutomergeError>(|tx| {
                tx.put(automerge::ROOT, "counter", i as i64)?;
                Ok(())
            })
            .unwrap();
        }

        // Simulate a second actor merging in (this is the pattern that
        // causes history bloat in production — many peers contributing changes)
        let mut doc2 = doc.fork();
        for i in 0..100 {
            doc2.transact::<_, _, automerge::AutomergeError>(|tx| {
                tx.put(automerge::ROOT, "peer2_counter", i as i64)?;
                Ok(())
            })
            .unwrap();
        }
        doc.merge(&mut doc2).unwrap();

        store.put("test-doc", &doc).unwrap();

        let result = store.compact("test-doc").unwrap();
        assert!(result.is_some());
        let (old_size, new_size) = result.unwrap();
        assert!(
            new_size <= old_size,
            "compaction should not increase size, got {} -> {}",
            old_size,
            new_size
        );

        // Verify both actor values are preserved
        let loaded = store.get("test-doc").unwrap().unwrap();
        let counter = loaded.get(automerge::ROOT, "counter").unwrap().unwrap();
        assert_eq!(counter.0.to_i64(), Some(99));
        let peer2 = loaded
            .get(automerge::ROOT, "peer2_counter")
            .unwrap()
            .unwrap();
        assert_eq!(peer2.0.to_i64(), Some(99));
    }

    #[test]
    fn test_compact_above_threshold() {
        let store = Arc::new(AutomergeStore::in_memory());

        // Create a small document (below threshold)
        let mut small_doc = Automerge::new();
        small_doc
            .transact::<_, _, automerge::AutomergeError>(|tx| {
                tx.put(automerge::ROOT, "key", "value")?;
                Ok(())
            })
            .unwrap();
        store.put("small-doc", &small_doc).unwrap();

        // Create a large document with multi-actor history
        let mut big_doc = Automerge::new();
        for i in 0..200 {
            big_doc
                .transact::<_, _, automerge::AutomergeError>(|tx| {
                    tx.put(automerge::ROOT, "counter", i as i64)?;
                    Ok(())
                })
                .unwrap();
        }
        // Add a second actor to simulate peer sync history
        let mut big_doc2 = big_doc.fork();
        for i in 0..200 {
            big_doc2
                .transact::<_, _, automerge::AutomergeError>(|tx| {
                    tx.put(automerge::ROOT, "peer2", i as i64)?;
                    Ok(())
                })
                .unwrap();
        }
        big_doc.merge(&mut big_doc2).unwrap();
        store.put("big-doc", &big_doc).unwrap();

        // Compact with threshold above small doc size but below big doc
        let small_size = small_doc.save().len();
        let threshold = small_size + 1;
        let (count, before, after) = store.compact_above_threshold(threshold).unwrap();

        // Only the big doc should have been compacted (small is below threshold)
        assert_eq!(count, 1);
        assert!(
            after <= before,
            "compaction should not increase size, got {} -> {}",
            before,
            after
        );

        // Verify the big doc value is preserved
        let loaded = store.get("big-doc").unwrap().unwrap();
        let value = loaded.get(automerge::ROOT, "counter").unwrap().unwrap();
        assert_eq!(value.0.to_i64(), Some(199));
    }

    #[tokio::test]
    async fn test_background_compaction_cancellation() {
        let store = Arc::new(AutomergeStore::in_memory());
        let token = tokio_util::sync::CancellationToken::new();

        store.start_background_compaction(
            std::time::Duration::from_millis(50),
            1024,
            vec!["test".to_string()],
            token.clone(),
        );

        // Let it run briefly
        tokio::time::sleep(std::time::Duration::from_millis(100)).await;

        // Cancel and verify it stops (no panic, no hang)
        token.cancel();
        tokio::time::sleep(std::time::Duration::from_millis(100)).await;
    }

    #[test]
    fn test_compact_collections_above_threshold() {
        let store = Arc::new(AutomergeStore::in_memory());

        // Create docs in two collections
        let mut beacons_doc = Automerge::new();
        for i in 0..100 {
            beacons_doc
                .transact::<_, _, automerge::AutomergeError>(|tx| {
                    tx.put(automerge::ROOT, "lat", i as i64)?;
                    Ok(())
                })
                .unwrap();
        }
        store.put("beacons:beacon-1", &beacons_doc).unwrap();

        let mut commands_doc = Automerge::new();
        for i in 0..100 {
            commands_doc
                .transact::<_, _, automerge::AutomergeError>(|tx| {
                    tx.put(automerge::ROOT, "cmd", i as i64)?;
                    Ok(())
                })
                .unwrap();
        }
        store.put("commands:cmd-1", &commands_doc).unwrap();

        let before_commands = store.document_size("commands:cmd-1").unwrap().unwrap();

        // Only compact beacons collection
        let (count, _, _) = store
            .compact_collections_above_threshold(&["beacons".to_string()], 1)
            .unwrap();

        assert_eq!(count, 1, "only the beacons doc should be compacted");

        // Commands doc should be unchanged
        let after_commands = store.document_size("commands:cmd-1").unwrap().unwrap();
        assert_eq!(
            before_commands, after_commands,
            "commands doc should not be touched"
        );
    }

    #[test]
    fn test_compact_collections_respects_threshold() {
        let store = Arc::new(AutomergeStore::in_memory());

        // Create a small doc in an opted-in collection
        let mut small_doc = Automerge::new();
        small_doc
            .transact::<_, _, automerge::AutomergeError>(|tx| {
                tx.put(automerge::ROOT, "key", "value")?;
                Ok(())
            })
            .unwrap();
        store.put("beacons:small", &small_doc).unwrap();

        let size = store.document_size("beacons:small").unwrap().unwrap();

        // Compact with threshold above the doc size
        let (count, _, _) = store
            .compact_collections_above_threshold(&["beacons".to_string()], size + 1)
            .unwrap();

        assert_eq!(count, 0, "small doc below threshold should be skipped");
    }

    // === peat-mesh#124: at-rest Cipher round-trip ===

    /// Deterministic non-identity cipher for testing — prepends a marker
    /// on encrypt, strips it on decrypt. Real consumers (peat-node)
    /// plug in AES-256-GCM via the same trait.
    #[derive(Debug)]
    struct MarkerCipher;

    const MARKER: &[u8] = b"STUB:";

    impl Cipher for MarkerCipher {
        fn encrypt(&self, plaintext: &[u8]) -> Result<Vec<u8>> {
            let mut out = Vec::with_capacity(MARKER.len() + plaintext.len());
            out.extend_from_slice(MARKER);
            out.extend_from_slice(plaintext);
            Ok(out)
        }

        fn decrypt(&self, ciphertext: &[u8]) -> Result<Vec<u8>> {
            ciphertext
                .strip_prefix(MARKER)
                .map(|rest| rest.to_vec())
                .ok_or_else(|| anyhow::anyhow!("ciphertext missing STUB: marker"))
        }
    }

    /// Cipher-enabled round-trip: put a structured Automerge doc, read
    /// it back through the store API, and assert (1) value equality
    /// end-to-end and (2) that the redb-level bytes are ciphertext, not
    /// the plaintext Automerge save() form.
    #[test]
    fn cipher_round_trip_and_on_disk_is_ciphertext() {
        let temp_dir = TempDir::new().unwrap();
        let cipher: Arc<dyn Cipher> = Arc::new(MarkerCipher);
        let store = AutomergeStore::open_with_cipher(temp_dir.path(), Some(cipher)).unwrap();

        let mut doc = Automerge::new();
        doc.transact::<_, _, automerge::AutomergeError>(|tx| {
            tx.put(automerge::ROOT, "platform_type", "vehicle")?;
            tx.put(automerge::ROOT, "count", 42_u64)?;
            Ok(())
        })
        .unwrap();

        // 1) End-to-end round-trip: put through the cipher-enabled path,
        //    drop the cache to force a redb read, then read back.
        store.put("vehicles:rover-1", &doc).unwrap();
        // Bypass the LRU cache to force the read path through the
        // cipher (the cache holds plaintext Automerge instances).
        store
            .cache
            .write()
            .unwrap()
            .pop(&"vehicles:rover-1".to_string());

        let loaded = store.get("vehicles:rover-1").unwrap().unwrap();
        let plaintext_loaded = loaded.save();
        let plaintext_original = doc.save();
        assert_eq!(
            plaintext_loaded, plaintext_original,
            "decrypt → load round-trip lost data"
        );

        // 2) Inspect the raw redb bytes for the same key. Drop the
        //    store first so its Database handle releases the file
        //    lock; redb refuses concurrent opens of the same file.
        let db_path = temp_dir.path().join("automerge.redb");
        drop(store);
        let db = Database::open(&db_path).unwrap();
        let txn = db.begin_read().unwrap();
        let table = txn.open_table(DOCUMENTS_TABLE).unwrap();
        let stored = table
            .get(b"vehicles:rover-1".as_slice())
            .unwrap()
            .unwrap()
            .value()
            .to_vec();

        assert_ne!(
            stored, plaintext_original,
            "on-disk bytes match plaintext — cipher.encrypt was not applied"
        );
        assert!(
            stored.starts_with(MARKER),
            "on-disk bytes don't start with the stub cipher marker"
        );
    }

    /// Sanity check: with `cipher = None` (the legacy path), on-disk
    /// bytes are still the plaintext Automerge save() form — i.e.
    /// the no-cipher path is unchanged by peat-mesh#124.
    #[test]
    fn no_cipher_on_disk_is_plaintext() {
        let temp_dir = TempDir::new().unwrap();
        let store = AutomergeStore::open(temp_dir.path()).unwrap();

        let mut doc = Automerge::new();
        doc.transact::<_, _, automerge::AutomergeError>(|tx| {
            tx.put(automerge::ROOT, "k", "v")?;
            Ok(())
        })
        .unwrap();
        store.put("c:1", &doc).unwrap();

        let db_path = temp_dir.path().join("automerge.redb");
        drop(store);
        let db = Database::open(&db_path).unwrap();
        let txn = db.begin_read().unwrap();
        let table = txn.open_table(DOCUMENTS_TABLE).unwrap();
        let stored = table
            .get(b"c:1".as_slice())
            .unwrap()
            .unwrap()
            .value()
            .to_vec();

        assert_eq!(stored, doc.save(), "no-cipher path changed the bytes");
    }

    /// peat-mesh#197 QA round-4: `read_lamport_highwater` must
    /// treat a wrong-length blob as `None` (with a tracing warning)
    /// rather than failing-closed. Pairs with `write_lamport_highwater`,
    /// which silently overwrites a corrupt entry — the two halves
    /// now share a single "log and recover" policy so a future
    /// metadata-table refactor that lands a mismatched value-length
    /// can't silently regress the contract.
    ///
    /// Lives as a unit test inside the module because the metadata
    /// table is module-private; reaching into the raw redb to plant
    /// a corrupt blob requires the same `pub(self)` access this
    /// test module has by construction.
    #[test]
    fn read_lamport_highwater_recovers_from_corrupt_entry() {
        let temp_dir = TempDir::new().unwrap();
        let store = AutomergeStore::open(temp_dir.path()).unwrap();

        // Plant a wrong-length blob (4 bytes instead of 8) at the
        // canonical metadata key via the raw redb handle. Drop the
        // store first so we don't double-open.
        let db_path = temp_dir.path().join("automerge.redb");
        drop(store);
        {
            let db = Database::open(&db_path).unwrap();
            let write_txn = db.begin_write().unwrap();
            {
                let mut table = write_txn.open_table(METADATA_TABLE).unwrap();
                table
                    .insert(
                        METADATA_KEY_LAMPORT_HIGHWATER,
                        &[0xDE, 0xAD, 0xBE, 0xEF][..],
                    )
                    .unwrap();
            }
            write_txn.commit().unwrap();
        }

        // Reopen the store. `read_lamport_highwater` must return
        // `Ok(None)` (not `Err(corrupt)`); a subsequent
        // `write_lamport_highwater` overwrites the corrupt entry
        // with a valid 8-byte value.
        let store = AutomergeStore::open(temp_dir.path()).unwrap();
        let read = store
            .read_lamport_highwater()
            .expect("read_lamport_highwater must recover, not error");
        assert_eq!(
            read, None,
            "corrupt entry must be treated as None (log-and-recover policy)"
        );

        // Self-heal: writing a valid value replaces the corruption.
        store.write_lamport_highwater(12345).unwrap();
        let after = store.read_lamport_highwater().unwrap();
        assert_eq!(after, Some(12345), "write must overwrite the corrupt entry");
    }
}
