x0x 0.34.0

//! KvStore synchronization using anti-entropy gossip.
//!
//! Wraps a KvStore in `Arc<RwLock<>>` for concurrent access and
//! synchronizes it via gossip pub/sub delta propagation.

use crate::gossip::wire::{decode_delta, encode_delta};
use crate::gossip::PubSubManager;
use crate::identity::AgentId;
use crate::kv::store::AccessPolicy;
use crate::kv::{KvStore, KvStoreDelta, Result};
use saorsa_gossip_types::PeerId;
use serde::{Deserialize, Serialize};
use std::path::{Path, PathBuf};
use std::sync::Arc;
use tokio::sync::RwLock;

/// Suffix appended to a store topic to form its state-sync side channel.
///
/// State requests travel on a separate topic so the main topic keeps its
/// existing `(PeerId, KvStoreDelta)` wire format — pre-#96 nodes simply
/// never subscribe to the side channel and are unaffected.
const STATE_SYNC_TOPIC_SUFFIX: &str = "/state-sync";

/// Delays between state-request retries for a first-time joiner whose
/// store is still empty. Spread out so a slow mesh (peer discovery,
/// subscription propagation) still converges without flooding.
const STATE_REQUEST_RETRY_SECS: [u64; 4] = [1, 5, 15, 30];

/// First persistent-tail delay after the front-loaded schedule exhausts.
const STATE_REQUEST_TAIL_START_SECS: u64 = 30;

/// Ceiling for the persistent tail's exponential backoff. While a replica
/// is still empty it keeps requesting at most this often — the steady-state
/// cost is one ~50-byte side-topic message per store per 5 minutes.
const STATE_REQUEST_TAIL_CAP_SECS: u64 = 300;

/// The complete state-request delay schedule: the front-loaded burst, then
/// an infinite exponential tail (30s doubling to a 300s ceiling).
///
/// Infinite BY DESIGN (issue #238): the owner answers state requests only
/// reactively and never volunteers state to late subscribers, so a finite
/// schedule left a replica that rehydrated while the owner was offline
/// permanently un-synced (a "zombie subscription" — even the owner
/// returning did not revive it; only a full daemon restart did, by minting
/// a fresh schedule). Convergence — `StateServed` evidence matched against
/// local state, see [`bootstrap_converged`] — is the only legitimate stop
/// condition, and the requester loop owns that check.
fn state_request_delays() -> impl Iterator<Item = u64> {
    let tail = std::iter::successors(Some(STATE_REQUEST_TAIL_START_SECS), |d| {
        Some(d.saturating_mul(2).min(STATE_REQUEST_TAIL_CAP_SECS))
    });
    STATE_REQUEST_RETRY_SECS.into_iter().chain(tail)
}

/// Minimum spacing between full-state responses from ONE holder for ONE
/// store. Every empty replica's request would otherwise make every holder
/// republish its complete state — after a fleet restart N replicas × M
/// holders align on the same schedule and the amplification is N×M full
/// publications per cadence. One response per window per holder serves all
/// concurrently-bootstrapping replicas (the response is a broadcast on the
/// main topic); a request that lands inside the window is served by the
/// requester's next scheduled attempt.
const STATE_RESPONSE_COOLDOWN_SECS: u64 = 15;

/// Sleep duration for a scheduled delay with ±20% jitter, so a fleet of
/// replicas restarted together does not phase-lock its request (and thus
/// full-state response) schedule. Mirrors the reconnect-backoff jitter in
/// `lib.rs`.
fn jittered_secs(secs: u64) -> std::time::Duration {
    let factor = 0.8 + rand::random::<f64>() * 0.4;
    std::time::Duration::from_secs_f64(secs as f64 * factor)
}

/// Message exchanged on the state-sync side topic.
///
/// Wire compatibility: `StateRequest` keeps its variant index and shape, so
/// v0.30.1 peers decode it unchanged. Older peers receiving the newer
/// `OwnerAnnounce` variant fail to deserialize it and skip the message
/// (their receive loop tolerates undecodable payloads), so the addition is
/// purely additive.
#[derive(Debug, Serialize, Deserialize)]
enum KvSyncMessage {
    /// A peer with no local state for the store asks holders to republish
    /// their full state (as a regular delta) on the main topic.
    StateRequest { requester: PeerId },
    /// The store owner's self-attestation of the store's authoritative
    /// metadata, published in response to a `StateRequest`.
    ///
    /// Trust model: the pub/sub layer verifies the ML-DSA-65 signature of
    /// every delivered v2 message and exposes the verified sender `AgentId`.
    /// The verified sender must equal the claimed `owner` — an owner can only
    /// attest to its own stores, and no third party can assign ownership.
    ///
    /// **Ownership is never established from this message.** A receiver's
    /// owner is anchored only at construction (see `KvStore::new_replica`).
    /// The announce can solely refresh policy (when the owner matches AND
    /// `policy_version` is strictly newer, blocking a replayed stale announce
    /// from downgrading policy) or record a conflict.
    OwnerAnnounce {
        /// The owning agent (must equal the verified message sender).
        owner: AgentId,
        /// The store's access policy as set by the owner.
        policy: AccessPolicy,
        /// Monotonic freshness counter — a refresh applies only when this is
        /// strictly greater than the receiver's current `policy_version`.
        policy_version: u64,
    },
    /// A holder's declaration that it has answered a `StateRequest`: its
    /// full state either was republished on the main topic (possibly
    /// earlier, within the response cooldown) or there is nothing to serve.
    /// Requesters match this against their OWN state to decide whether the
    /// bootstrap tail may stop — mere non-emptiness is not convergence
    /// evidence (a single incremental delta must not silence recovery).
    ///
    /// Wire compatibility: additive variant, same precedent as
    /// `OwnerAnnounce` — v0.33.0 peers fail to deserialize it and skip the
    /// message. Against a fleet of only-older responders no markers arrive
    /// and the requester keeps its capped-cadence tail (bounded chatter,
    /// never a zombie).
    StateServed {
        /// The declaring holder (receivers skip their own echo).
        responder: PeerId,
        /// True when the holder's store is empty. Only the OWNER ever
        /// declares emptiness (an empty non-owner replica stays silent) —
        /// otherwise two empty bootstrapping replicas would convince each
        /// other the store is legitimately empty and re-create the zombie.
        empty: bool,
        /// The holder's owner-signed checkpoint high-water mark, when it
        /// holds one. A requester whose own mark has reached this value has
        /// provably absorbed at least this much owner history — the exact
        /// convergence gate for checkpoint-bearing stores.
        checkpoint_seq: Option<u64>,
    },
    /// A holder's digest-committed declaration that it has answered a
    /// `StateRequest` (issue #240): `digest` commits to the FULL served
    /// entry set (see `served_content_digest`), so a requester can verify
    /// "served us, completely" against its OWN state instead of trusting
    /// that the full delta and the marker both arrived (the v1 cross-topic
    /// loss window). A requester stops only when its local digest matches a
    /// declared digest — a lost full delta leaves the local digest
    /// different, so it keeps asking.
    ///
    /// Empty holders (owner or not) declare the digest of the empty set,
    /// which any empty requester can compute locally — authoritative
    /// emptiness without an owner, terminating the genuinely-empty chatter
    /// tail (v1 behavior for old peers is unchanged: only the OWNER declares
    /// emptiness there). Because the digest is self-verifying, an empty
    /// declaration needs no full-delta broadcast to witness.
    ///
    /// Wire compatibility: additive variant, same precedent as
    /// `OwnerAnnounce`/`StateServed` — older peers fail to deserialize it
    /// and skip the message; new peers treat v1 markers as weaker evidence
    /// when no v2 digest has been seen. The marker rides along with the
    /// full-state broadcast (never separately) when there is state to serve.
    StateServedV2 {
        /// The declaring holder (receivers skip their own echo).
        responder: PeerId,
        /// Canonical BLAKE3 digest over the served entry set.
        digest: [u8; 32],
        /// Number of entries in the served set — a cheap shape check for the
        /// verified full-replace adopt path (and useful in logs).
        entry_count: u32,
    },
}

/// One responder's latest v2 digest declaration.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
struct ServedState {
    /// The declared content digest.
    digest: [u8; 32],
    /// The declared number of entries.
    entry_count: u32,
}

/// Aggregated `StateServed`/`StateServedV2` evidence observed by one
/// replica's responder loop, consumed by its bootstrap requester to decide
/// convergence.
#[derive(Debug, Default, Clone)]
struct ServedEvidence {
    /// A holder declared non-empty state.
    saw_nonempty: bool,
    /// The OWNER declared the store legitimately empty.
    saw_owner_empty: bool,
    /// Highest checkpoint sequence any holder declared.
    max_checkpoint_seq: u64,
    /// Latest digest declaration per responder (bounded by mesh size — a
    /// responder's newer declaration REPLACES its older one, so a replayed
    /// stale serve can never roll a verified full-replace adopt backwards).
    digests: std::collections::HashMap<PeerId, ServedState>,
}

/// Disarms the bootstrap-active flag on ANY requester exit path (converged,
/// silenced, cancelled, torn down) so the listener's digest-verified
/// full-replace adopt can never fire outside the bootstrap window.
struct BootstrapGuard(std::sync::Arc<std::sync::atomic::AtomicBool>);

impl Drop for BootstrapGuard {
    fn drop(&mut self) {
        self.0.store(false, std::sync::atomic::Ordering::Relaxed);
    }
}

/// Convergence rule for the bootstrap tail (pure for unit-testing).
///
/// Strongest available evidence wins:
/// - a declared checkpoint sequence must be matched by this replica's own
///   high-water mark (exact, survives partial/lost responses);
/// - otherwise, when any v2 digest declarations exist, the local digest
///   must match one of them — with the data-bearing-claim-wins rule: if any
///   declaration is non-empty, only a non-empty match converges (an empty
///   holder's declaration must not retire a requester while a full holder
///   advertised content);
/// - otherwise (only v1 markers seen — old peers) the legacy weak rules:
///   declared non-empty requires local non-emptiness; emptiness counts only
///   when the owner declared it.
///
/// No evidence at all (`ServedEvidence::default()`) is NEVER convergence.
fn bootstrap_converged(
    ev: &ServedEvidence,
    is_empty: bool,
    highest_checkpoint_seq: u64,
    local_digest: [u8; 32],
) -> bool {
    if ev.max_checkpoint_seq > 0 {
        return highest_checkpoint_seq >= ev.max_checkpoint_seq;
    }
    if !ev.digests.is_empty() {
        let any_nonempty = ev.digests.values().any(|d| d.entry_count > 0);
        return ev
            .digests
            .values()
            .any(|d| d.digest == local_digest && (d.entry_count > 0 || !any_nonempty));
    }
    if ev.saw_nonempty {
        !is_empty
    } else {
        ev.saw_owner_empty
    }
}

/// Synchronization wrapper for a KvStore.
///
/// Manages automatic background synchronization using anti-entropy gossip.
/// Changes are propagated via deltas published to a gossip topic.
pub struct KvStoreSync {
    /// The store being synchronized.
    store: Arc<RwLock<KvStore>>,

    /// Pub/sub manager for topic-based messaging.
    pubsub: Arc<PubSubManager>,

    /// Topic name for this store.
    topic: String,

    /// This node's gossip peer id — identifies our deltas and state
    /// requests on the wire.
    local_peer_id: PeerId,

    /// This node's agent id, when known. Used to decide whether this node
    /// is the store owner (and should answer state requests with an
    /// [`KvSyncMessage::OwnerAnnounce`]) and to ignore its own announces.
    local_agent_id: Option<AgentId>,

    /// Optional persistence context. When armed (see
    /// [`set_persist_path`](Self::set_persist_path)), the full store state is
    /// snapshotted atomically after every local mutation and every merged
    /// remote delta, so a restart restores policy, keyset, entry contents,
    /// the latest adopted checkpoint, the checkpoint high-water mark, and the
    /// OR-Set sequence-counter ceiling instead of coming back as an empty
    /// replica. This is what makes `AppendOnly` immutability survive a
    /// restart: an owner (or replica) with amnesia would otherwise accept
    /// rewrites of keys it no longer remembers holding.
    persist: std::sync::Mutex<Option<Arc<PersistCtx>>>,

    /// Set by [`silence_bootstrap`](Self::silence_bootstrap). The bootstrap
    /// requester checks it every iteration: its schedule is infinite (issue
    /// #238), so a sync that should stop generating traffic — but keep
    /// serving (e.g. the deletion-test harness) — arms this without ending
    /// the listener/responder loops.
    stopped: Arc<std::sync::atomic::AtomicBool>,

    /// Cancelled by [`cancel_sync`](Self::cancel_sync) / [`stop`](Self::stop).
    /// ALL background loops (delta listener, responder, requester) select on
    /// it, so a discarded sync tears down completely without the topic-wide
    /// `unsubscribe` that would kill unrelated subscribers sharing the topic
    /// string (round-4 review: flag-only teardown left ghost listeners and a
    /// live responder until daemon shutdown).
    cancel: tokio_util::sync::CancellationToken,
}

/// Structural teardown (parallel-review finding): the background loops hold
/// clones of the token, the store, and the pubsub — never the sync itself —
/// so when the last `KvStoreSync` reference drops, every loop (including the
/// INFINITE bootstrap requester, issue #238) is cancelled without any caller
/// having to remember `cancel_sync()`. The explicit rollback calls remain as
/// belt-and-braces.
impl Drop for KvStoreSync {
    fn drop(&mut self) {
        self.cancel.cancel();
    }
}

/// Shared persistence context for one store's snapshot file.
struct PersistCtx {
    /// Snapshot file path.
    path: PathBuf,
    /// Serializes snapshot commits AND records the last durably-persisted
    /// store version. `(version, bytes)` are captured under this lock, so
    /// commit order equals capture order — a concurrent persist burst can
    /// never rename an older snapshot over a newer one — and the version
    /// gate skips writes that would not advance durable state.
    gate: tokio::sync::Mutex<Option<u64>>,
    /// True after a failed snapshot write; cleared by the next success.
    /// While set, LOCAL writes are refused (fail-closed for what this node
    /// controls); remote-delta merges continue (replication is not wedged).
    degraded: std::sync::atomic::AtomicBool,
}

impl KvStoreSync {
    /// Create a new KvStore synchronization manager.
    ///
    /// # Arguments
    ///
    /// * `store` - The KvStore to synchronize.
    /// * `pubsub` - Pub/sub manager for gossip messaging.
    /// * `topic` - Topic name for pub/sub.
    /// * `local_peer_id` - This node's gossip peer id.
    /// * `local_agent_id` - This node's agent id, if available. Required for
    ///   the owner to answer state requests with an ownership announcement;
    ///   `None` disables announcing (joined replicas can still adopt).
    pub fn new(
        store: KvStore,
        pubsub: Arc<PubSubManager>,
        topic: String,
        local_peer_id: PeerId,
        local_agent_id: Option<AgentId>,
    ) -> Result<Self> {
        let store = Arc::new(RwLock::new(store));

        Ok(Self {
            store,
            pubsub,
            topic,
            local_peer_id,
            local_agent_id,
            persist: std::sync::Mutex::new(None),
            stopped: Arc::new(std::sync::atomic::AtomicBool::new(false)),
            cancel: tokio_util::sync::CancellationToken::new(),
        })
    }

    /// Enable on-disk snapshot persistence at `path`.
    ///
    /// Call before [`start`](Self::start) so no merged delta can land
    /// unpersisted. The caller is responsible for loading any existing
    /// snapshot BEFORE constructing this sync (see
    /// [`load_snapshot`]); this method only arms writes.
    pub fn set_persist_path(&self, path: PathBuf) {
        if let Ok(mut guard) = self.persist.lock() {
            *guard = Some(Arc::new(PersistCtx {
                path,
                gate: tokio::sync::Mutex::new(None),
                degraded: std::sync::atomic::AtomicBool::new(false),
            }));
        }
    }

    /// Clone the armed persistence context, if any.
    fn persist_ctx(&self) -> Option<Arc<PersistCtx>> {
        self.persist.lock().ok().and_then(|g| g.clone())
    }

    /// Snapshot the store to the configured persist path (`Ok` no-op when
    /// persistence is not armed).
    ///
    /// Durability contract:
    /// - Commits are serialized per store and version-gated, so concurrent
    ///   persists can never regress durable state.
    /// - On failure the store is flagged **durability-degraded**
    ///   ([`durability_degraded`](Self::durability_degraded)): callers on the
    ///   LOCAL write path must propagate the error to the writer and MUST NOT
    ///   publish the mutation (durability before announcement); callers on
    ///   the REMOTE merge path log and continue (replication is not wedged —
    ///   peers hold the data; only this node's disk is behind).
    /// - The next successful persist (including via
    ///   [`ensure_durable`](Self::ensure_durable)) clears the flag.
    ///
    /// # Errors
    ///
    /// I/O or serialization failure writing the snapshot.
    pub async fn persist(&self) -> Result<()> {
        match self.persist_ctx() {
            Some(ctx) => persist_snapshot(&self.store, &ctx).await,
            None => Ok(()),
        }
    }

    /// True while the last snapshot attempt failed and no retry has
    /// succeeded. Local writes are refused in this state (fail-closed).
    pub fn durability_degraded(&self) -> bool {
        self.persist_ctx()
            .is_some_and(|c| c.degraded.load(std::sync::atomic::Ordering::Relaxed))
    }

    /// If the store is durability-degraded, retry persisting the CURRENT
    /// state before any new mutation is accepted. `Ok` when not degraded,
    /// not persistent, or the retry succeeded.
    ///
    /// # Errors
    ///
    /// The retry failed — the caller must refuse the local write.
    pub async fn ensure_durable(&self) -> Result<()> {
        match self.persist_ctx() {
            Some(ctx) if ctx.degraded.load(std::sync::atomic::Ordering::Relaxed) => {
                persist_snapshot(&self.store, &ctx).await
            }
            _ => Ok(()),
        }
    }

    /// The state-sync side topic for this store.
    fn state_sync_topic(&self) -> String {
        format!("{}{}", self.topic, STATE_SYNC_TOPIC_SUFFIX)
    }

    /// Start background synchronization.
    ///
    /// Subscribes to the gossip topic and begins receiving remote deltas.
    /// Also joins the state-sync side channel: holders answer state
    /// requests by republishing their full state, and — issue #96 — a
    /// first-time joiner (empty local store) requests that state so it
    /// bootstraps keys written before it joined. Without this, only
    /// deltas published *after* subscribing ever arrive.
    pub async fn start(&self) -> Result<()> {
        self.start_with_spawner(|fut| {
            tokio::spawn(fut);
        })
        .await
    }

    /// Start background synchronization with a caller-supplied spawner.
    ///
    /// Identical to [`start`](Self::start), but routes the background loops
    /// (delta-merge listener, state-request responder, and the bounded
    /// bootstrap requester) through `spawn` instead of detaching them with
    /// `tokio::spawn`. The `Agent` passes its tracked-task spawner so these
    /// loops are registered with the `Agent::shutdown()` drain and aborted on
    /// teardown (issue #126); callers without an `Agent` use
    /// [`start`](Self::start), which detaches via `tokio::spawn` as before.
    pub async fn start_with_spawner<S>(&self, spawn: S) -> Result<()>
    where
        S: Fn(std::pin::Pin<Box<dyn std::future::Future<Output = ()> + Send + 'static>>)
            + Send
            + Sync,
    {
        let mut sub = self.pubsub.subscribe(self.topic.clone()).await;
        let store = Arc::clone(&self.store);
        // Capture the bootstrap decision BEFORE any listener can merge a
        // cached delta. Otherwise a partial cache replay landing between
        // subscribe and this check could make the store non-empty and skip
        // the bootstrap state-request schedule — aged/pruned keys would
        // never arrive.
        //
        // Non-owner replicas ALWAYS bootstrap: a snapshot-restored replica
        // may have missed deltas while it was offline, and emptiness cannot
        // distinguish "fresh join" from "restored but stale" (the gossip
        // cache only replays ~60s of history). Owners are authoritative and
        // request only when EMPTY — that is snapshot-loss recovery from
        // their replicas.
        let bootstrap_needed = {
            let s = store.read().await;
            let local_is_owner =
                self.local_agent_id.is_some() && s.owner() == self.local_agent_id.as_ref();
            !local_is_owner || s.is_empty()
        };
        // Defense in depth against cross-topic replay: the v2 signature covers
        // the embedded topic, but pub/sub delivery does not re-check it against
        // this subscription, so a raw-mesh participant could place a valid
        // owner-signed envelope from store A under topic B. Each listener binds
        // to the exact topic it subscribed to.
        let main_topic = self.topic.clone();
        // Snapshot the persist context once: it is armed before start() by
        // construction (set_persist_path docs), so the loops never observe a
        // late change.
        let persist_ctx = self.persist_ctx();

        let loop_persist_ctx = persist_ctx.clone();
        let listener_cancel = self.cancel.clone();
        // StateServed evidence: written by the responder loop (which owns
        // the side-topic subscription), read by the listener (verified
        // full-replace adopt, issue #240) and the bootstrap requester
        // (convergence). Created BEFORE the loops so all three share it.
        let served_evidence = Arc::new(std::sync::Mutex::new(ServedEvidence::default()));
        // Armed only while the bootstrap requester runs: the verified
        // full-replace adopt fires exclusively in that window — a converged
        // replica must never let a divergent holder's serve truncate state
        // it legitimately holds.
        let bootstrap_active = Arc::new(std::sync::atomic::AtomicBool::new(bootstrap_needed));
        let listener_served = Arc::clone(&served_evidence);
        let listener_bootstrap_active = Arc::clone(&bootstrap_active);
        spawn(Box::pin(async move {
            loop {
                let msg = tokio::select! {
                    // cancel_sync tears down every loop (round-4 review) —
                    // recv alone would keep this listener alive until
                    // daemon shutdown.
                    () = listener_cancel.cancelled() => return,
                    msg = sub.recv() => msg,
                };
                let Some(msg) = msg else {
                    // The main-topic subscription is gone: this sync can no
                    // longer replicate, so it is half-dead — self-cancel so
                    // the sibling loops (in particular the INFINITE
                    // bootstrap requester) never outlive it (parallel-review
                    // finding: a dead sibling must not leave the requester
                    // chattering at capped cadence forever).
                    listener_cancel.cancel();
                    return;
                };
                if msg.topic != main_topic {
                    // Cross-topic replay defense: ignore envelopes not on our
                    // subscribed topic (see start_with_spawner).
                    continue;
                }
                let decoded = decode_delta::<KvStoreDelta>(&msg.payload);
                match decoded {
                    Ok((peer_id, delta)) => {
                        let merged = {
                            let mut s = store.write().await;
                            // Pass sender identity for access control enforcement.
                            // The gossip V2 wire format includes a verified AgentId.
                            let writer = msg.sender.as_ref();
                            match s.merge_delta(&delta, peer_id, writer) {
                                Ok(()) => {
                                    // Digest-verified full-replace adopt
                                    // (issue #240, checkpoint-less deletion
                                    // cold-sync): while bootstrapping, when
                                    // the sender's latest v2 declaration
                                    // matches this delta's served content
                                    // (digest AND entry count), the delta IS
                                    // that holder's complete state — prune
                                    // local keys it does not carry. Gated on
                                    // the sender being an AUTHORIZED writer:
                                    // merge_delta silently ignores
                                    // unauthorized deltas, and the prune
                                    // must not apply what the merge would
                                    // not. Without verification any holder
                                    // could truncate local state at will.
                                    if listener_bootstrap_active
                                        .load(std::sync::atomic::Ordering::Relaxed)
                                    {
                                        let declared = listener_served
                                            .lock()
                                            .unwrap_or_else(std::sync::PoisonError::into_inner)
                                            .digests
                                            .get(&peer_id)
                                            .copied();
                                        if let Some(declared) = declared {
                                            // Prune authority (F2,
                                            // fix-loop): absence from a
                                            // verified serve proves
                                            // deletion ONLY when the
                                            // sender is the SOLE possible
                                            // content author. Under
                                            // Signed that is the anchored
                                            // owner — and every key this
                                            // replica can hold was
                                            // admitted under that same
                                            // auth (or an owner-signed
                                            // checkpoint), so the owner's
                                            // serve is complete about all
                                            // of them. Under Allowlisted
                                            // the owner's serve can be
                                            // legitimately incomplete
                                            // about co-writers' keys (an
                                            // owner that has not merged
                                            // an allowlisted write would
                                            // otherwise TRUNCATE it), so
                                            // pruning is disabled there —
                                            // deletions still propagate
                                            // via live `removed` deltas,
                                            // and the digest mismatch
                                            // resolves when the owner
                                            // absorbs the co-write and
                                            // re-serves. (The review's
                                            // alternative count guard —
                                            // prune only when local-count
                                            // <= declared-count — was
                                            // rejected: the residual-2
                                            // stale replica is exactly a
                                            // local SUPERSET of the serve,
                                            // so that rule disables
                                            // pruning precisely where
                                            // deletion cold-sync needs
                                            // it.)
                                            let sole_author =
                                                matches!(s.policy(), AccessPolicy::Signed)
                                                    && writer.is_some()
                                                    && s.owner() == writer;
                                            if sole_author
                                                && delta.added.len()
                                                    == declared.entry_count as usize
                                                && delta
                                                    .served_digest(s.id())
                                                    .is_some_and(|dg| dg == declared.digest)
                                            {
                                                let pruned = s.prune_to_served_set(&delta);
                                                if pruned > 0 {
                                                    tracing::info!(
                                                        "pruned {pruned} stale key(s) after \
                                                         digest-verified full serve for store {}",
                                                        s.id()
                                                    );
                                                }
                                            }
                                        }
                                    }
                                    true
                                }
                                Err(e) => {
                                    tracing::warn!("Failed to merge KvStore delta: {e}");
                                    false
                                }
                            }
                        };
                        // Persist OUTSIDE the write guard so disk latency
                        // never blocks other writers. A failure flags the
                        // store durability-degraded (persist_snapshot logs);
                        // remote merges continue — replication must not
                        // wedge on this node's disk.
                        if merged {
                            if let Some(ctx) = loop_persist_ctx.as_ref() {
                                let _ = persist_snapshot(&store, ctx).await;
                            }
                        }
                    }
                    Err(e) => {
                        tracing::warn!("Failed to deserialize KvStore delta: {e}");
                    }
                }
            }
        }));

        // Responder + ownership listener on the state-sync side topic.
        //
        // StateRequest: holders with non-empty state answer by republishing
        // their full state as a regular delta on the main topic. CRDT merge
        // makes duplicate responses from multiple holders harmless
        // (idempotent), so no response suppression is needed at current mesh
        // sizes. Additionally, if this node is the store OWNER it publishes
        // an OwnerAnnounce (regardless of emptiness) so joined replicas can
        // learn the authoritative owner and policy.
        //
        // OwnerAnnounce: a replica with an unknown owner adopts the owner
        // and policy — but only when the announcement's pub/sub-verified
        // sender is the claimed owner itself (see KvSyncMessage docs).
        let mut sync_sub = self.pubsub.subscribe(self.state_sync_topic()).await;
        let responder_store = Arc::clone(&self.store);
        let responder_persist_ctx = persist_ctx.clone();
        let responder_pubsub = Arc::clone(&self.pubsub);
        let responder_topic = self.topic.clone();
        let sync_topic = self.state_sync_topic();
        let local_peer_id = self.local_peer_id;
        let local_agent_id = self.local_agent_id;
        let responder_served = Arc::clone(&served_evidence);
        let responder_cancel = self.cancel.clone();
        spawn(Box::pin(async move {
            // Response-storm damping (issue #238 review): one full-state
            // response per cooldown window, regardless of how many replicas
            // are requesting — the response is a broadcast, so it serves
            // them all.
            let mut last_full_response: Option<tokio::time::Instant> = None;
            loop {
                let msg = tokio::select! {
                    // cancel_sync tears down every loop (round-4 review).
                    () = responder_cancel.cancelled() => return,
                    msg = sync_sub.recv() => msg,
                };
                let Some(msg) = msg else {
                    // The side-topic subscription is gone: this sync can no
                    // longer receive StateServed evidence, so the requester
                    // could never legitimately stop — self-cancel so it
                    // (and the sibling loops) never outlive the responder.
                    responder_cancel.cancel();
                    return;
                };
                if msg.topic != sync_topic {
                    // Cross-topic replay defense (see start_with_spawner).
                    continue;
                }
                let Ok(sync_msg) = bincode::deserialize::<KvSyncMessage>(&msg.payload) else {
                    continue;
                };
                match sync_msg {
                    KvSyncMessage::StateRequest { requester } => {
                        if requester == local_peer_id {
                            continue;
                        }
                        // Owner: announce authoritative metadata so anchored
                        // joiners can refresh policy / confirm ownership.
                        // (Ownership itself is never learned from this — a
                        // joiner anchors its owner at construction.)
                        let announce = {
                            let s = responder_store.read().await;
                            match (local_agent_id, s.owner()) {
                                (Some(me), Some(owner)) if me == *owner => {
                                    Some(KvSyncMessage::OwnerAnnounce {
                                        owner: me,
                                        policy: s.policy().clone(),
                                        policy_version: s.policy_version(),
                                    })
                                }
                                _ => None,
                            }
                        };
                        if let Some(announce) = announce {
                            match bincode::serialize(&announce) {
                                Ok(serialized) => {
                                    if let Err(e) = responder_pubsub
                                        .publish(sync_topic.clone(), bytes::Bytes::from(serialized))
                                        .await
                                    {
                                        tracing::warn!(
                                            "KvStore owner-announce publish failed: {e}"
                                        );
                                    }
                                }
                                Err(e) => {
                                    tracing::warn!("KvStore owner-announce serialize failed: {e}");
                                }
                            }
                        }
                        // Cooldown gates the full-state broadcast. The
                        // StateServed marker is published ONLY alongside an
                        // actual full-delta publish (or for an owner's
                        // checkpoint-less empty store, which has no payload
                        // at all): a marker must witness a real broadcast —
                        // a marker for a cooldown-suppressed response could
                        // convince a requester that never received the state
                        // to stop asking (round-3 review). Checked BEFORE
                        // building the full delta — no point cloning the
                        // whole store for a suppressed response.
                        let cooled_down = last_full_response.is_some_and(|t| {
                            t.elapsed()
                                < std::time::Duration::from_secs(STATE_RESPONSE_COOLDOWN_SECS)
                        });
                        // Snapshot state once for the full-delta and the
                        // StateServed markers below. A full response is
                        // served when the store is non-empty, OR when it is
                        // empty but holds an owner checkpoint: the
                        // checkpoint-adopt merge path is a full REPLACE
                        // (keys absent from the signed set are removed), so
                        // a checkpoint-bearing empty delta is exactly how a
                        // deleted-to-empty store cold-syncs to a stale
                        // replica (round-3 review — an empty owner must not
                        // be silent while stale holders keep advertising
                        // obsolete state). The v2 digest is computed over
                        // the SAME snapshot as the full delta, so the
                        // declaration always commits to exactly what was
                        // broadcast.
                        let (full, is_empty, is_owner, checkpoint_seq, has_payload, served) = {
                            let s = responder_store.read().await;
                            let is_owner =
                                local_agent_id.is_some() && s.owner() == local_agent_id.as_ref();
                            let cp =
                                (s.highest_checkpoint_seq > 0).then_some(s.highest_checkpoint_seq);
                            let has_payload = !s.is_empty() || s.latest_checkpoint.is_some();
                            let full = (has_payload && !cooled_down).then(|| s.full_delta());
                            let served = (s.served_digest(), s.checkpoint_pairs().len() as u32);
                            (full, s.is_empty(), is_owner, cp, has_payload, served)
                        };
                        let mut markers: Vec<KvSyncMessage> = Vec::new();
                        if let Some(full) = full {
                            if let Ok(serialized) = encode_delta(local_peer_id, &full) {
                                if let Err(e) = responder_pubsub
                                    .publish(
                                        responder_topic.clone(),
                                        bytes::Bytes::from(serialized),
                                    )
                                    .await
                                {
                                    tracing::warn!("KvStore state-response publish failed: {e}");
                                } else {
                                    last_full_response = Some(tokio::time::Instant::now());
                                    markers.push(KvSyncMessage::StateServed {
                                        responder: local_peer_id,
                                        empty: is_empty,
                                        checkpoint_seq,
                                    });
                                    // The v2 marker rides along with the
                                    // broadcast it commits to — never
                                    // separately (response-storm damping,
                                    // issue #240).
                                    markers.push(KvSyncMessage::StateServedV2 {
                                        responder: local_peer_id,
                                        digest: served.0,
                                        entry_count: served.1,
                                    });
                                }
                            }
                        } else if !has_payload {
                            // Checkpoint-less empty. The v2 digest of the
                            // empty set is universally computable, so ANY
                            // empty holder may declare it — an empty
                            // requester verifies locally and stops (issue
                            // #240; no broadcast to witness because there
                            // is nothing to serve).
                            markers.push(KvSyncMessage::StateServedV2 {
                                responder: local_peer_id,
                                digest: served.0,
                                entry_count: 0,
                            });
                            if is_owner {
                                // v1 behavior for older peers is unchanged:
                                // only the OWNER declares emptiness — an
                                // empty non-owner replica stays silent on
                                // v1 so bootstrapping replicas can never
                                // talk each other into a false "converged
                                // empty".
                                markers.push(KvSyncMessage::StateServed {
                                    responder: local_peer_id,
                                    empty: true,
                                    checkpoint_seq: None,
                                });
                            }
                        }
                        for marker in markers {
                            match bincode::serialize(&marker) {
                                Ok(serialized) => {
                                    if let Err(e) = responder_pubsub
                                        .publish(sync_topic.clone(), bytes::Bytes::from(serialized))
                                        .await
                                    {
                                        tracing::warn!(
                                            "KvStore state-served marker publish failed: {e}"
                                        );
                                    }
                                }
                                Err(e) => {
                                    tracing::warn!(
                                        "KvStore state-served marker serialize failed: {e}"
                                    );
                                }
                            }
                        }
                    }
                    KvSyncMessage::StateServed {
                        responder,
                        empty,
                        checkpoint_seq,
                    } => {
                        if responder == local_peer_id {
                            continue; // our own marker echoed back
                        }
                        // Trust note: markers steer only WHEN the bootstrap
                        // requester stops asking — never store content, and
                        // convergence is re-checked against local state
                        // (`bootstrap_converged`), so a forged marker cannot
                        // inject state. A forged NON-EMPTY marker at worst
                        // stops the tail no earlier than a real holder could
                        // (the any-holder-answers trust the protocol already
                        // has). The two evidence classes that could do real
                        // damage are trusted only from the pub/sub-verified
                        // anchored owner:
                        // - EMPTY would stop an empty replica's recovery
                        //   outright;
                        // - CHECKPOINT_SEQ is the exact convergence gate,
                        //   and a forged u64::MAX would pin the requester
                        //   at capped cadence forever while a forged low
                        //   value could retire a stale replica early
                        //   (round-3 review).
                        // Owner-marker checkpoints also self-correlate: the
                        // local high-water mark only rises by MERGING the
                        // checkpoint-bearing full delta, so satisfying the
                        // gate proves the state actually arrived.
                        let owner_verified = {
                            let anchored = responder_store.read().await.owner().copied();
                            anchored.is_some() && msg.sender.as_ref() == anchored.as_ref()
                        };
                        let mut ev = responder_served
                            .lock()
                            .unwrap_or_else(std::sync::PoisonError::into_inner);
                        if empty {
                            if owner_verified {
                                ev.saw_owner_empty = true;
                            }
                        } else {
                            ev.saw_nonempty = true;
                        }
                        if let Some(seq) = checkpoint_seq.filter(|_| owner_verified) {
                            ev.max_checkpoint_seq = ev.max_checkpoint_seq.max(seq);
                        }
                    }
                    KvSyncMessage::StateServedV2 {
                        responder,
                        digest,
                        entry_count,
                    } => {
                        if responder == local_peer_id {
                            continue; // our own marker echoed back
                        }
                        // Trust note: the digest is SELF-VERIFYING — a
                        // forged declaration can only match local state
                        // that actually equals the declared content, so a
                        // forgery's worst case is the requester keeps
                        // asking (the same bound as a forged v1 marker).
                        // The verified full-replace adopt additionally
                        // requires the full-delta SENDER to be an
                        // authorized writer (see the listener), so a
                        // marker alone can never truncate anything.
                        responder_served
                            .lock()
                            .unwrap_or_else(std::sync::PoisonError::into_inner)
                            .digests
                            .insert(
                                responder,
                                ServedState {
                                    digest,
                                    entry_count,
                                },
                            );
                    }
                    KvSyncMessage::OwnerAnnounce {
                        owner,
                        policy,
                        policy_version,
                    } => {
                        // Only a signature-verified sender is trusted; the
                        // pub/sub layer drops signed messages that fail
                        // verification, so `sender: Some(..)` is verified.
                        let Some(sender) = msg.sender else {
                            tracing::warn!(
                                "ignoring unsigned KvStore ownership announcement on {}",
                                msg.topic
                            );
                            continue;
                        };
                        if local_agent_id.is_some_and(|me| me == sender) {
                            continue; // our own announce echoed back
                        }
                        let learned = {
                            let mut s = responder_store.write().await;
                            // learn_ownership can only refresh policy (when the
                            // owner matches and policy_version is forward) or
                            // record a conflict; it never establishes ownership.
                            // AppendOnly is terminal: a downgrade announce is
                            // rejected inside learn_ownership regardless of
                            // policy_version.
                            match s.learn_ownership(owner, policy, policy_version, &sender) {
                                Ok(()) => {
                                    tracing::info!(
                                        "KvStore {} processed owner announce from {} (policy {}, version {})",
                                        s.id(),
                                        hex::encode(owner.as_bytes()),
                                        s.policy(),
                                        s.policy_version()
                                    );
                                    true
                                }
                                Err(e) => {
                                    tracing::warn!(
                                        "rejected KvStore ownership announcement from {}: {e}",
                                        hex::encode(sender.as_bytes())
                                    );
                                    false
                                }
                            }
                        };
                        // A policy refresh mutates durable state — persist it
                        // (outside the write guard).
                        if learned {
                            if let Some(ctx) = responder_persist_ctx.as_ref() {
                                let _ = persist_snapshot(&responder_store, ctx).await;
                            }
                        }
                    }
                }
            }
        }));

        // Bootstrap requester: a first-time joiner starts with an empty
        // store and has no other way to learn keys written before it
        // subscribed (the gossip message cache only replays ~60s, and
        // pruning on busy topics removes older deltas entirely). Ask
        // holders to republish. The full FRONT schedule always runs — a
        // partial state arriving early (for example fresh keys via cache
        // replay) must not stop the request for the complete historical
        // state. After the front schedule, an infinite backoff tail keeps
        // asking while the store is STILL EMPTY (issue #238): holders
        // answer only reactively, so a replica whose requests all fired
        // while the owner was offline would otherwise stay a zombie
        // forever. The tail self-terminates the moment any state merges
        // (the owner's full-delta response also carries its checkpoint,
        // so policy converges with the data). Requests and the full-delta
        // responses they trigger are idempotent CRDT merges, so the
        // chatter is harmless; a genuinely-new empty store costs one tiny
        // side-topic message per backoff interval until its first write.
        if bootstrap_needed {
            let requester_pubsub = Arc::clone(&self.pubsub);
            let sync_topic = self.state_sync_topic();
            // Weak: the requester must not keep the store alive on its own.
            // (Belt-and-braces — the sibling loops hold strong Arcs, so the
            // authoritative kill switch is the `stopped` flag below.)
            let requester_store = Arc::downgrade(&self.store);
            let stopped = Arc::clone(&self.stopped);
            let requester_cancel = self.cancel.clone();
            let requester_served = Arc::clone(&served_evidence);
            let requester_bootstrap_active = Arc::clone(&bootstrap_active);
            spawn(Box::pin(async move {
                // Disarms the adopt window on ANY exit (converged, silenced,
                // cancelled, torn down) — the listener's verified
                // full-replace adopt must never fire outside bootstrap.
                let _guard = BootstrapGuard(requester_bootstrap_active);
                for (attempt, delay_secs) in state_request_delays().enumerate() {
                    tokio::select! {
                        // cancel_sync tears down every loop promptly, even
                        // mid-sleep (round-4 review).
                        () = requester_cancel.cancelled() => return,
                        () = tokio::time::sleep(jittered_secs(delay_secs)) => {}
                    }
                    if stopped.load(std::sync::atomic::Ordering::Relaxed) {
                        return; // silenced — never chatter for a dead sync
                    }
                    // Tail attempts stop on convergence; front attempts
                    // always run (see above — partial early state must not
                    // cancel the request for full history). Convergence is
                    // judged against StateServed evidence matched to local
                    // state (`bootstrap_converged`) — NOT mere non-emptiness,
                    // which a single incremental delta can fake while the
                    // full historical state is still missing (round-2
                    // review). v2 digest evidence makes the check exact for
                    // checkpoint-less state (issue #240): the requester
                    // stops only when its OWN content digest matches a
                    // holder's declaration.
                    if attempt >= STATE_REQUEST_RETRY_SECS.len() {
                        let Some(store) = requester_store.upgrade() else {
                            return; // sync torn down — nothing left to bootstrap
                        };
                        let ev = requester_served
                            .lock()
                            .unwrap_or_else(std::sync::PoisonError::into_inner)
                            .clone();
                        let (is_empty, cp_hwm, local_digest) = {
                            let s = store.read().await;
                            (s.is_empty(), s.highest_checkpoint_seq, s.served_digest())
                        };
                        if bootstrap_converged(&ev, is_empty, cp_hwm, local_digest) {
                            return; // a holder served us and local state matches
                        }
                    }
                    let request = KvSyncMessage::StateRequest {
                        requester: local_peer_id,
                    };
                    let Ok(serialized) = bincode::serialize(&request) else {
                        return;
                    };
                    if let Err(e) = requester_pubsub
                        .publish(sync_topic.clone(), bytes::Bytes::from(serialized))
                        .await
                    {
                        tracing::debug!("KvStore state-request publish failed: {e}");
                    }
                }
            }));
        }

        Ok(())
    }

    /// Silence ONLY this sync's bootstrap requester (its schedule is
    /// infinite while unconverged — issue #238), leaving the listener and
    /// responder loops serving.
    ///
    /// Use when the replica should keep replicating but never generate
    /// bootstrap chatter (e.g. an authoritative holder in a single-identity
    /// test harness). Discarded handles want [`cancel_sync`](Self::cancel_sync).
    pub fn silence_bootstrap(&self) {
        self.stopped
            .store(true, std::sync::atomic::Ordering::Relaxed);
    }

    /// Tear down ALL of this sync's background loops (delta listener,
    /// state-request responder, bootstrap requester) WITHOUT touching topic
    /// subscriptions.
    ///
    /// This is the correct teardown for a discarded handle inside a daemon:
    /// `PubSubManager::unsubscribe` (what [`stop`](Self::stop) does) removes
    /// the ENTIRE topic — including subscriptions owned by other components
    /// that legally share the topic string. Ending the loops drops their
    /// `Subscription` receivers, so the pub/sub layer prunes the closed
    /// senders on its next delivery.
    pub fn cancel_sync(&self) {
        self.cancel.cancel();
    }

    /// Stop background synchronization.
    ///
    /// Topic-wide: unsubscribes the main and state-sync topics for the
    /// WHOLE process (every subscriber of those topic strings), which is
    /// only appropriate when this sync is the topics' sole consumer.
    /// In-process daemons discarding one handle should use
    /// [`cancel_sync`](Self::cancel_sync) — or simply drop every handle
    /// clone: the [`Drop`] impl cancels structurally.
    pub async fn stop(&self) -> Result<()> {
        // End the loops FIRST: the bootstrap requester's schedule is
        // infinite while the store is empty (issue #238), and unsubscribing
        // does not end that loop (it holds no subscription).
        self.cancel_sync();
        self.pubsub.unsubscribe(&self.topic).await;
        self.pubsub.unsubscribe(&self.state_sync_topic()).await;
        Ok(())
    }

    /// Publish a local delta to the gossip network.
    pub async fn publish_delta(&self, local_peer_id: PeerId, delta: KvStoreDelta) -> Result<()> {
        let serialized = encode_delta(local_peer_id, &delta)
            .map_err(|e| crate::kv::KvError::Gossip(format!("serialize delta failed: {e}")))?;

        self.pubsub
            .publish(self.topic.clone(), bytes::Bytes::from(serialized))
            .await
            .map_err(|e| crate::kv::KvError::Gossip(format!("publish delta failed: {e}")))?;

        Ok(())
    }

    /// Get a read-only reference to the store.
    pub async fn read(&self) -> tokio::sync::RwLockReadGuard<'_, KvStore> {
        self.store.read().await
    }

    /// Get a mutable reference to the store.
    pub async fn write(&self) -> tokio::sync::RwLockWriteGuard<'_, KvStore> {
        self.store.write().await
    }

    /// Get the topic name.
    #[must_use]
    pub fn topic(&self) -> &str {
        &self.topic
    }
}

/// Monotonic counter for unique snapshot temp-file names — concurrent
/// persists (receive loop vs. local write) must never clobber each other's
/// temp file mid-rename.
static SNAPSHOT_TMP_COUNTER: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);

/// Magic prefix of the v1 snapshot file format.
///
/// Format: `MAGIC(8) || bincode(SnapshotBody { store, seq_counter })`.
/// The envelope exists so the OR-Set sequence-counter ceiling (which is
/// `serde(skip)` on `KvStore` for wire/legacy-layout reasons) survives a
/// restart exactly. The format is introduced unreleased — no shipped binary
/// ever wrote a bare-`KvStore` snapshot — so there is no compat read path:
/// a file without the magic is rejected (fail closed) rather than guessed at.
const SNAPSHOT_MAGIC: &[u8; 8] = b"X0XKVS1\0";

/// Owned snapshot body (decode side).
#[derive(Deserialize)]
struct SnapshotBody {
    store: KvStore,
    seq_counter: u64,
}

/// Borrowing snapshot body (encode side — avoids cloning the store).
#[derive(Serialize)]
struct SnapshotBodyRef<'a> {
    store: &'a KvStore,
    seq_counter: u64,
}

/// Encode a store into v1 snapshot bytes (magic + body).
fn encode_snapshot(store: &KvStore) -> Result<Vec<u8>> {
    let body = SnapshotBodyRef {
        store,
        seq_counter: store.seq_counter_value(),
    };
    let mut out = Vec::with_capacity(256);
    out.extend_from_slice(SNAPSHOT_MAGIC);
    out.extend_from_slice(&bincode::serialize(&body)?);
    Ok(out)
}

/// Snapshot the store to the persistence context's path.
///
/// Serialized per store via `ctx.gate`: `(version, bytes)` are captured
/// under the gate, so commit order equals capture order and a slow persist
/// can never rename an older snapshot over a newer one; the recorded
/// last-persisted version additionally skips writes that would not advance
/// durable state. Success clears the degraded flag; failure sets it and is
/// error-logged here (callers decide whether to propagate — local writes
/// must, remote merges must not).
///
/// # Errors
///
/// Serialization or I/O failure writing the snapshot.
async fn persist_snapshot(store: &Arc<RwLock<KvStore>>, ctx: &PersistCtx) -> Result<()> {
    let result = async {
        let mut last = ctx.gate.lock().await;
        let (version, bytes) = {
            let s = store.read().await;
            (s.current_version(), encode_snapshot(&s)?)
        };
        if last.is_some_and(|l| l >= version) {
            // Durable state already at (or beyond) this version.
            return Ok(());
        }
        write_snapshot_atomic(&ctx.path, &bytes)?;
        *last = Some(version);
        Ok(())
    }
    .await;
    ctx.degraded
        .store(result.is_err(), std::sync::atomic::Ordering::Relaxed);
    if let Err(e) = &result {
        tracing::error!(
            "kv snapshot persist failed for {}: {e} — store is durability-degraded; \
             local writes are refused until a snapshot succeeds",
            ctx.path.display()
        );
    }
    result
}

/// Durable atomic file write: unique temp file in the same directory,
/// fsync, rename over the destination, then (Unix) fsync the parent
/// directory so the rename itself survives power loss.
///
/// Platform note: on non-Unix targets the parent-directory fsync is skipped
/// (std cannot fsync a directory handle there); the rename is still atomic,
/// but its durability across power loss is not guaranteed. SIGKILL/power
/// loss beyond the parent fsync (e.g. hardware write caches) is out of
/// scope.
fn write_snapshot_atomic(path: &Path, bytes: &[u8]) -> std::io::Result<()> {
    if let Some(parent) = path.parent() {
        std::fs::create_dir_all(parent)?;
    }
    let n = SNAPSHOT_TMP_COUNTER.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
    let tmp = path.with_extension(format!("tmp.{}.{n}", std::process::id()));
    {
        use std::io::Write;
        let mut f = std::fs::File::create(&tmp)?;
        f.write_all(bytes)?;
        f.sync_all()?;
    }
    if let Err(e) = std::fs::rename(&tmp, path) {
        let _ = std::fs::remove_file(&tmp);
        return Err(e);
    }
    #[cfg(unix)]
    if let Some(parent) = path.parent() {
        std::fs::File::open(parent)?.sync_all()?;
    }
    Ok(())
}

/// Load a previously persisted store snapshot from `path`.
///
/// Returns:
/// - `Ok(Some(store))` — snapshot present and valid.
/// - `Ok(None)` — no snapshot at `path` (first run).
/// - `Err(_)` — snapshot present but unreadable, undecodable, or not in the
///   v1 format. Callers MUST fail closed on this (refuse to start an empty
///   replica over a corrupt snapshot): silently discarding it would reopen
///   the restart-amnesia window (an `AppendOnly` owner that forgets its keys
///   will re-accept rewrites of them).
///
/// The restored store's in-memory `seq_counter` is set to the persisted
/// counter (floored by `version` as defense in depth), so freshly minted
/// OR-Set `(peer, seq)` tags can never collide with tags issued before the
/// restart — including the extra per-put delta tag minted by
/// `KvStoreHandle::put_with_delta`.
///
/// # Errors
///
/// [`crate::kv::KvError::Io`]/[`crate::kv::KvError::Serialization`] as above.
pub fn load_snapshot(path: &Path) -> Result<Option<KvStore>> {
    let bytes = match std::fs::read(path) {
        Ok(b) => b,
        Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(None),
        Err(e) => return Err(e.into()),
    };
    let Some(body_bytes) = bytes.strip_prefix(SNAPSHOT_MAGIC.as_slice()) else {
        return Err(std::io::Error::other(
            "unrecognized kv snapshot format (missing v1 magic) — corrupt or foreign file; \
             refusing to start with amnesia",
        )
        .into());
    };
    let body: SnapshotBody = bincode::deserialize(body_bytes)?;
    let store = body.store;
    store.restore_seq_counter(body.seq_counter.max(store.current_version()));
    Ok(Some(store))
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::identity::AgentId;
    use crate::kv::store::AccessPolicy;
    use crate::kv::{KvEntry, KvStoreId};
    use crate::network::{NetworkConfig, NetworkNode};
    use std::time::Duration;

    fn agent(n: u8) -> AgentId {
        AgentId([n; 32])
    }

    fn peer(n: u8) -> PeerId {
        PeerId::new([n; 32])
    }

    fn store_id(n: u8) -> KvStoreId {
        KvStoreId::new([n; 32])
    }

    #[test]
    fn snapshot_roundtrip_missing_and_corrupt() {
        // WHY: snapshot restore is what makes AppendOnly immutability
        // survive a restart. Missing file = clean first run (Ok(None));
        // a valid snapshot must round-trip policy, entries, and the
        // checkpoint high-water mark; a corrupt file must be an Err so
        // callers FAIL CLOSED instead of silently starting empty (amnesia).
        let dir = tempfile::tempdir().expect("tmpdir");
        let path = dir.path().join("kv").join("snap.bin");

        assert!(
            matches!(load_snapshot(&path), Ok(None)),
            "missing snapshot is a clean first run"
        );

        let mut store = KvStore::new(
            store_id(7),
            "log".to_string(),
            agent(1),
            AccessPolicy::AppendOnly,
        );
        store
            .put(
                "k1".to_string(),
                b"v1".to_vec(),
                "text/plain".to_string(),
                peer(1),
            )
            .expect("put");
        store.highest_checkpoint_seq = 5;
        // Simulate the handle-layer double seq mint: the counter can run
        // ahead of `version`. The persisted counter — not a version-derived
        // floor — must be the restore ceiling.
        let _ = store.next_seq();
        let _ = store.next_seq();
        let counter_before = store.seq_counter_value();
        let bytes = encode_snapshot(&store).expect("encode");
        write_snapshot_atomic(&path, &bytes).expect("atomic write");

        let restored = load_snapshot(&path)
            .expect("load ok")
            .expect("snapshot present");
        assert_eq!(*restored.policy(), AccessPolicy::AppendOnly);
        assert_eq!(
            restored.get("k1").map(|e| e.value.clone()),
            Some(b"v1".to_vec())
        );
        assert_eq!(restored.highest_checkpoint_seq, 5);
        // Exact tag ceiling restored: the next minted seq is strictly above
        // every pre-restart seq (no OR-Set (peer, seq) tag reuse).
        assert!(
            restored.next_seq() > counter_before,
            "restored seq counter must exceed every pre-restart seq"
        );

        // A file without the v1 magic (e.g. a bare-bincode or foreign file)
        // fails closed.
        std::fs::write(&path, bincode::serialize(&store).expect("serialize")).expect("write bare");
        assert!(
            load_snapshot(&path).is_err(),
            "missing-magic snapshot must be an error (fail closed)"
        );

        std::fs::write(&path, b"not a snapshot").expect("corrupt");
        assert!(
            load_snapshot(&path).is_err(),
            "corrupt snapshot must be an error (fail closed), not a silent fresh start"
        );

        // Truncated/garbage body AFTER a valid magic also fails closed.
        let mut evil = SNAPSHOT_MAGIC.to_vec();
        evil.extend_from_slice(b"\x01\x02\x03");
        std::fs::write(&path, evil).expect("write garbage body");
        assert!(
            load_snapshot(&path).is_err(),
            "garbage body must be an error (fail closed)"
        );
    }

    /// Construct an isolated network node (mirrors the helper in
    /// `src/gossip/pubsub.rs` tests). `PubSubManager` is fully constructable
    /// in tests, so `KvStoreSync` is testable end-to-end without a live mesh.
    async fn make_node() -> Arc<NetworkNode> {
        Arc::new(
            NetworkNode::new(NetworkConfig::default(), None, None)
                .await
                .expect("network node"),
        )
    }

    /// Build a `KvStoreSync` around a fresh node + pubsub, with
    /// `owner = agent(1)` and `local_peer_id = peer(1)`.
    async fn make_sync(topic: &str, policy: AccessPolicy) -> KvStoreSync {
        let node = make_node().await;
        let pubsub = Arc::new(PubSubManager::new(node, None).expect("pubsub"));
        let store = KvStore::new(store_id(1), "Test".to_string(), agent(1), policy);
        KvStoreSync::new(store, pubsub, topic.to_string(), peer(1), Some(agent(1)))
            .expect("kv sync")
    }

    /// Build a `KvStoreSync` that shares its pubsub with the caller (so the
    /// caller can subscribe before the sync publishes).
    async fn make_sync_with_pubsub(
        topic: &str,
        policy: AccessPolicy,
    ) -> (KvStoreSync, Arc<PubSubManager>) {
        let node = make_node().await;
        let pubsub = Arc::new(PubSubManager::new(node, None).expect("pubsub"));
        let store = KvStore::new(store_id(1), "Test".to_string(), agent(1), policy);
        let sync = KvStoreSync::new(
            store,
            Arc::clone(&pubsub),
            topic.to_string(),
            peer(1),
            Some(agent(1)),
        )
        .expect("kv sync");
        (sync, pubsub)
    }

    #[tokio::test]
    async fn test_kv_store_sync_creation() {
        let owner = agent(1);
        let store = KvStore::new(store_id(1), "Test".to_string(), owner, AccessPolicy::Signed);
        let _store_for_sync = store;
    }

    #[tokio::test]
    async fn test_apply_delta_directly() {
        let owner = agent(1);
        let writer = agent(2);
        let p2 = peer(2);

        let mut store = KvStore::new(
            store_id(1),
            "Test".to_string(),
            owner,
            AccessPolicy::Allowlisted,
        );
        store.allow_writer(writer, &owner).expect("allow");
        let store_arc = Arc::new(RwLock::new(store));

        let entry = KvEntry::new(
            "newkey".to_string(),
            b"value".to_vec(),
            "text/plain".to_string(),
        );
        let mut delta = KvStoreDelta::new(1);
        delta.added.insert("newkey".to_string(), (entry, (p2, 1)));

        {
            let mut s = store_arc.write().await;
            s.merge_delta(&delta, p2, Some(&writer)).expect("merge");
        }

        {
            let s = store_arc.read().await;
            assert!(s.get("newkey").is_some());
        }
    }

    #[tokio::test]
    async fn test_concurrent_reads() {
        let owner = agent(1);
        let store = KvStore::new(store_id(1), "Test".to_string(), owner, AccessPolicy::Signed);
        let store_arc = Arc::new(RwLock::new(store));

        let s1 = store_arc.read().await;
        let s2 = store_arc.read().await;

        assert_eq!(s1.name(), "Test");
        assert_eq!(s2.name(), "Test");
    }

    // ------------------------------------------------------------------
    // new() / topic() / read() / write()
    // ------------------------------------------------------------------

    #[tokio::test]
    async fn new_sets_topic_and_yields_accessible_guards() {
        let sync = make_sync("store/A", AccessPolicy::Signed).await;

        // topic() reports exactly the topic handed to new().
        assert_eq!(sync.topic(), "store/A");

        // read() exposes the underlying store unchanged.
        {
            let s = sync.read().await;
            assert_eq!(s.name(), "Test");
            assert!(s.is_empty());
        }

        // write() returns a mutable guard; verify it is usable by merging
        // an owner-authored delta into the Signed store, then observe it via
        // read(). This also exercises the read/write guard pair end-to-end.
        let owner = agent(1);
        let entry = KvEntry::new(
            "owner-key".to_string(),
            b"v".to_vec(),
            "text/plain".to_string(),
        );
        let mut delta = KvStoreDelta::new(1);
        delta
            .added
            .insert("owner-key".to_string(), (entry, (peer(1), 1)));
        {
            let mut s = sync.write().await;
            s.merge_delta(&delta, peer(1), Some(&owner))
                .expect("owner merge");
        }

        let s = sync.read().await;
        assert!(s.get("owner-key").is_some(), "owner write must be visible");
    }

    // ------------------------------------------------------------------
    // state_sync_topic() (private helper exercised from the test module)
    // ------------------------------------------------------------------

    #[tokio::test]
    async fn state_sync_topic_appends_side_channel_suffix() {
        let sync = make_sync("store/B", AccessPolicy::Signed).await;
        // The private helper forms the side channel by appending the suffix.
        assert_eq!(sync.state_sync_topic(), "store/B/state-sync");

        // Suffix is appended exactly once, regardless of slashes in topic.
        let sync2 = make_sync("store/B/nested", AccessPolicy::Signed).await;
        assert_eq!(sync2.state_sync_topic(), "store/B/nested/state-sync");
    }

    // ------------------------------------------------------------------
    // publish_delta(): wire round-trip observed by a subscriber
    // ------------------------------------------------------------------

    #[tokio::test]
    async fn publish_delta_delivers_encoded_pair_to_subscriber() {
        let (sync, pubsub) = make_sync_with_pubsub("store/C", AccessPolicy::Signed).await;

        // Subscribe to the main topic BEFORE publishing so we observe the
        // exact bytes KvStoreSync places on the wire.
        let mut sub = pubsub.subscribe("store/C".to_string()).await;

        let sender = peer(7);
        let entry = KvEntry::new(
            "remote".to_string(),
            b"payload".to_vec(),
            "application/octet-stream".to_string(),
        );
        let mut delta = KvStoreDelta::new(9);
        delta
            .added
            .insert("remote".to_string(), (entry, (sender, 3)));

        sync.publish_delta(sender, delta)
            .await
            .expect("publish_delta");

        let msg = tokio::time::timeout(Duration::from_secs(2), sub.recv())
            .await
            .expect("timed out waiting for published delta")
            .expect("subscriber stream closed");

        // The published payload must decode back to the (sender, delta) pair
        // that publish_delta encoded — proving the wire format is correct.
        let (observed_sender, observed_delta) =
            decode_delta::<KvStoreDelta>(&msg.payload).expect("wire decode");
        assert_eq!(observed_sender, sender);
        assert_eq!(observed_delta.version, 9);
        assert!(observed_delta.added.contains_key("remote"));
        assert_eq!(msg.topic, "store/C");
        // Sanity: the same delta also round-trips through encode_delta alone.
        let reencoded = encode_delta(sender, &observed_delta).expect("re-encode");
        let (s2, d2) = decode_delta::<KvStoreDelta>(&reencoded).expect("re-decode");
        assert_eq!(s2, sender);
        assert_eq!(d2.version, 9);
    }

    // ------------------------------------------------------------------
    // start_with_spawner(): subscribes + returns Ok with a drop-spawner
    // ------------------------------------------------------------------

    #[tokio::test]
    async fn start_with_spawner_subscribes_and_returns_ok() {
        // Unique value vs `start_default_spawner_merges_remote_delta`: this
        // routes the background futures through a *custom* (non-`tokio::spawn`)
        // spawner closure — a drop-spawner — exercising that generic code path
        // and asserting `start_with_spawner` returns `Ok` without panicking.
        //
        // It deliberately does NOT assert that a subscription or merge
        // occurred: a drop-spawner makes subscription unobservable, so this
        // would still pass against a no-op `Ok(())` impl. The real
        // subscribe->merge behaviour is asserted end-to-end by
        // `start_default_spawner_merges_remote_delta`, which drives
        // `start_with_spawner(tokio::spawn)` and verifies the key lands.
        let sync = make_sync("store/D", AccessPolicy::Signed).await;
        sync.start_with_spawner(|_fut| {
            // intentionally drop the future
        })
        .await
        .expect("start_with_spawner");
    }

    // ------------------------------------------------------------------
    // start(): default spawner merges a remotely-published delta
    // ------------------------------------------------------------------

    #[tokio::test]
    async fn start_default_spawner_merges_remote_delta() {
        // End-to-end exercise of the delta-merge listener: a delta published
        // on the topic is received by the background loop spawned by start()
        // and merged into the local store. We use an Encrypted policy so an
        // unsigned (anonymous-sender) delta is accepted by the store's
        // access control — matching what the wire delivers for an unsigned
        // publish via a PubSubManager with no signing context.
        let sync = make_sync(
            "store/E",
            AccessPolicy::Encrypted {
                group_id: vec![1, 2, 3],
            },
        )
        .await;

        sync.start().await.expect("start");

        // Let the spawned subscribe-forwarder register before we publish.
        tokio::time::sleep(Duration::from_millis(100)).await;

        let entry = KvEntry::new(
            "merged-key".to_string(),
            b"hello".to_vec(),
            "text/plain".to_string(),
        );
        let mut delta = KvStoreDelta::new(1);
        delta
            .added
            .insert("merged-key".to_string(), (entry, (peer(2), 1)));
        sync.publish_delta(peer(2), delta).await.expect("publish");

        // The merge is asynchronous; poll the store until it lands.
        let landed = tokio::time::timeout(Duration::from_secs(2), async {
            loop {
                let present = {
                    let s = sync.read().await;
                    s.get("merged-key").is_some()
                };
                if present {
                    return;
                }
                tokio::time::sleep(Duration::from_millis(25)).await;
            }
        })
        .await;
        assert!(
            landed.is_ok(),
            "remote delta was not merged by start() loop"
        );
    }

    // ------------------------------------------------------------------
    // stop(): returns Ok and is idempotent
    // ------------------------------------------------------------------

    #[tokio::test]
    async fn stop_returns_ok_and_is_idempotent() {
        let sync = make_sync("store/F", AccessPolicy::Signed).await;
        sync.stop().await.expect("first stop");
        // stop() unsubscribes both the main and the state-sync topic;
        // unsubscribe is infallible and tolerant of already-removed topics,
        // so a second stop() must remain Ok.
        sync.stop().await.expect("second stop (idempotent)");
    }

    /// WHY (rounds 1+4 review): the requester's schedule is INFINITE while
    /// the store is empty, and the sibling loops hold strong `Arc`s to the
    /// store — so the cancellation token (all loops) and the
    /// silence_bootstrap flag (requester only) are the only things that end
    /// a discarded sync's background work before daemon shutdown.
    #[tokio::test]
    async fn stop_and_silence_arm_their_kill_switches() {
        let sync = make_sync("store/stopflag", AccessPolicy::Signed).await;
        assert!(
            !sync.cancel.is_cancelled() && !sync.stopped.load(std::sync::atomic::Ordering::Relaxed),
            "both switches must start disarmed"
        );
        sync.silence_bootstrap();
        assert!(
            sync.stopped.load(std::sync::atomic::Ordering::Relaxed),
            "silence_bootstrap() arms the requester-only flag"
        );
        assert!(
            !sync.cancel.is_cancelled(),
            "silence_bootstrap() must NOT cancel the listener/responder loops"
        );
        sync.stop().await.expect("stop");
        assert!(
            sync.cancel.is_cancelled(),
            "stop() must cancel ALL background loops via the token"
        );
    }

    /// WHY (parallel-review finding): teardown must be STRUCTURAL — a
    /// caller that discards its last reference without remembering
    /// cancel_sync() must still end all background loops (the requester is
    /// infinite while unconverged), so Drop cancels the token.
    #[tokio::test]
    async fn dropping_the_sync_cancels_all_loops() {
        let sync = make_sync("store/dropcancel", AccessPolicy::Signed).await;
        let token = sync.cancel.clone();
        assert!(!token.is_cancelled());
        drop(sync);
        assert!(
            token.is_cancelled(),
            "dropping the last sync reference must cancel every loop"
        );
    }

    // ------------------------------------------------------------------
    // Issue #238: the bootstrap requester must never give up while empty
    // ------------------------------------------------------------------

    /// WHY: the request schedule is the ONLY trigger for state recovery —
    /// holders answer reactively and never volunteer state to a late
    /// subscriber. A finite schedule therefore turned "owner offline while
    /// the schedule ran" into a permanent zombie: the replica stayed empty
    /// forever, even after the owner returned. The schedule must be
    /// front-loaded (a fast mesh converges in seconds) and then an
    /// infinite capped tail (bounded chatter, unbounded patience).
    #[test]
    fn state_request_schedule_never_terminates_while_unconverged() {
        let front: Vec<u64> = state_request_delays().take(4).collect();
        assert_eq!(front, STATE_REQUEST_RETRY_SECS, "front burst unchanged");
        let tail: Vec<u64> = state_request_delays().skip(4).take(8).collect();
        assert_eq!(
            tail,
            [30, 60, 120, 240, 300, 300, 300, 300],
            "tail doubles to the cap, then holds it"
        );
        assert_eq!(
            state_request_delays().nth(10_000),
            Some(STATE_REQUEST_TAIL_CAP_SECS),
            "the schedule is infinite — convergence, not the schedule, \
             is what ends the requester"
        );
    }

    /// WHY (round-2 review): the convergence rule must weigh evidence
    /// correctly — checkpoint match is exact; declared-non-empty requires
    /// local state; emptiness counts only when the owner declared it; and
    /// NO evidence is NEVER convergence (a replica that heard nothing keeps
    /// asking, whatever its local state looks like). Issue #240 adds the
    /// digest gate: when v2 declarations exist they OUTRANK the weak v1
    /// rules, and a data-bearing declaration outranks an empty one.
    #[test]
    fn bootstrap_convergence_rule() {
        const D1: [u8; 32] = [1u8; 32];
        const D2: [u8; 32] = [2u8; 32];
        const D_EMPTY: [u8; 32] = [9u8; 32];
        let none = ServedEvidence::default();
        // No evidence: never converged, empty or not.
        assert!(!bootstrap_converged(&none, true, 0, D1));
        assert!(!bootstrap_converged(&none, false, 9, D1));

        // Checkpoint evidence is exact: local HWM must reach it.
        let cp = ServedEvidence {
            saw_nonempty: true,
            saw_owner_empty: false,
            max_checkpoint_seq: 5,
            digests: std::collections::HashMap::new(),
        };
        assert!(
            !bootstrap_converged(&cp, false, 4, D1),
            "behind the served HWM"
        );
        assert!(bootstrap_converged(&cp, false, 5, D1));
        assert!(bootstrap_converged(&cp, false, 7, D1));

        // Non-empty holder, no checkpoint: local non-emptiness required.
        let nonempty = ServedEvidence {
            saw_nonempty: true,
            saw_owner_empty: false,
            max_checkpoint_seq: 0,
            digests: std::collections::HashMap::new(),
        };
        assert!(!bootstrap_converged(&nonempty, true, 0, D1));
        assert!(bootstrap_converged(&nonempty, false, 0, D1));

        // Owner-declared empty (and nothing stronger): converged even empty.
        let owner_empty = ServedEvidence {
            saw_nonempty: false,
            saw_owner_empty: true,
            max_checkpoint_seq: 0,
            digests: std::collections::HashMap::new(),
        };
        assert!(bootstrap_converged(&owner_empty, true, 0, D1));
        // A non-empty holder claim outranks owner-empty when both were seen.
        let mixed = ServedEvidence {
            saw_nonempty: true,
            saw_owner_empty: true,
            max_checkpoint_seq: 0,
            digests: std::collections::HashMap::new(),
        };
        assert!(
            !bootstrap_converged(&mixed, true, 0, D1),
            "divergent holders: the data-bearing claim must win, keep asking"
        );

        // ---- v2 digest evidence (issue #240) ----
        let digest_ev = |decls: &[(&[u8; 32], u32)]| ServedEvidence {
            saw_nonempty: false,
            saw_owner_empty: false,
            max_checkpoint_seq: 0,
            digests: decls
                .iter()
                .enumerate()
                .map(|(i, (d, c))| {
                    (
                        peer(i as u8 + 10),
                        ServedState {
                            digest: **d,
                            entry_count: *c,
                        },
                    )
                })
                .collect(),
        };
        // Match stops: local digest equals a declared non-empty digest.
        let ev = digest_ev(&[(&D1, 3)]);
        assert!(bootstrap_converged(&ev, false, 0, D1));
        // Mismatch keeps asking — the lost-full-delta window is closed.
        assert!(!bootstrap_converged(&ev, false, 0, D2));
        // Weak v1 evidence must NOT rescue a digest mismatch.
        let mut ev_v1_too = digest_ev(&[(&D1, 3)]);
        ev_v1_too.saw_nonempty = true;
        assert!(
            !bootstrap_converged(&ev_v1_too, false, 0, D2),
            "v2 declarations outrank weak v1 evidence"
        );
        // Empty-holder declarations converge an empty requester (and ONLY
        // when every declaration is empty).
        let ev_empty = digest_ev(&[(&D_EMPTY, 0)]);
        assert!(bootstrap_converged(&ev_empty, true, 0, D_EMPTY));
        let ev_divergent = digest_ev(&[(&D_EMPTY, 0), (&D1, 2)]);
        assert!(
            !bootstrap_converged(&ev_divergent, true, 0, D_EMPTY),
            "a data-bearing declaration outranks the empty one — keep asking"
        );
        assert!(bootstrap_converged(&ev_divergent, false, 0, D1));
    }

    /// WHY (round-2 review — P1: restored non-empty replicas still became
    /// zombies): round 1 made non-owner replicas run the front burst, but
    /// the tail still exited on mere non-emptiness — a restored replica
    /// whose owner stayed offline through the burst exited on its first
    /// tail check and never asked again. Convergence now requires
    /// StateServed evidence: the replica must keep asking until a holder
    /// actually answers, however long the owner is away.
    #[tokio::test(start_paused = true)]
    async fn restored_replica_keeps_asking_until_a_holder_serves() {
        let node = make_node().await;
        let kp = crate::identity::AgentKeypair::generate().expect("keypair");
        let owner_id = kp.agent_id();
        let ctx = Arc::new(crate::gossip::SigningContext::from_keypair(&kp));
        let pubsub = Arc::new(PubSubManager::new(node, Some(ctx)).expect("pubsub"));
        let topic = "kv-238-restored-late-owner";

        // Snapshot-restored replica: non-empty, owner OFFLINE.
        let mut replica = KvStore::new_replica(
            store_id(1),
            String::new(),
            Some(owner_id),
            crate::kv::store::AnchorChannel::Persistence,
        );
        replica
            .put(
                "k_old".to_string(),
                b"v_old".to_vec(),
                "text/plain".to_string(),
                peer(2),
            )
            .expect("seed restored key");
        let joiner = KvStoreSync::new(
            replica,
            Arc::clone(&pubsub),
            topic.to_string(),
            peer(2),
            Some(agent(2)),
        )
        .expect("joiner sync");
        joiner.start().await.expect("start joiner");

        // The entire front burst fires with the owner away. The round-1
        // code exited the tail right here (non-empty ⇒ "converged").
        tokio::time::sleep(Duration::from_secs(90)).await;

        // Owner returns much later, holding a key the replica missed.
        let mut owned = KvStore::new(
            store_id(1),
            "log".to_string(),
            owner_id,
            AccessPolicy::Signed,
        );
        owned
            .put(
                "k_old".to_string(),
                b"v_old".to_vec(),
                "text/plain".to_string(),
                peer(1),
            )
            .expect("owner k_old");
        owned
            .put(
                "k_new".to_string(),
                b"v_new".to_vec(),
                "text/plain".to_string(),
                peer(1),
            )
            .expect("owner k_new");
        let owner_sync = KvStoreSync::new(
            owned,
            Arc::clone(&pubsub),
            topic.to_string(),
            peer(1),
            Some(owner_id),
        )
        .expect("owner sync");
        owner_sync.start().await.expect("start owner");

        let mut recovered = false;
        for _ in 0..200 {
            tokio::time::sleep(Duration::from_secs(5)).await;
            if joiner.read().await.get("k_new").is_some() {
                recovered = true;
                break;
            }
        }
        assert!(
            recovered,
            "a restored non-empty replica must keep requesting until a \
             holder serves it — non-emptiness alone is not convergence"
        );
    }

    /// WHY (round-3 review — deleted-to-empty must cold-sync): an owner
    /// whose store is legitimately empty but checkpointed (everything
    /// deleted) must still serve state requests: the checkpoint-adopt merge
    /// path is a full REPLACE, so its checkpoint-bearing EMPTY full delta
    /// is exactly what removes a stale replica's obsolete keys. Before this
    /// fix the empty owner was silent while stale holders kept advertising
    /// old state; convergence was also unreachable (the marker declared a
    /// checkpoint the replica could never adopt).
    #[tokio::test(start_paused = true)]
    async fn checkpointed_empty_owner_deletes_stale_replica_state() {
        let node = make_node().await;
        let kp = crate::identity::AgentKeypair::generate().expect("keypair");
        let owner_id = kp.agent_id();
        let ctx = Arc::new(crate::gossip::SigningContext::from_keypair(&kp));
        let pubsub = Arc::new(PubSubManager::new(node, Some(ctx)).expect("pubsub"));
        let topic = "kv-238-deleted-to-empty";

        // Owner: EMPTY store carrying an owner-signed checkpoint over the
        // empty set (the state after deleting everything).
        let mut owned = KvStore::new(
            store_id(1),
            "log".to_string(),
            owner_id,
            AccessPolicy::Signed,
        );
        let (pub_bytes, sec_bytes) = kp.to_bytes();
        let public_key =
            ant_quic::MlDsaPublicKey::from_bytes(&pub_bytes).expect("public key bytes");
        let secret_key =
            ant_quic::MlDsaSecretKey::from_bytes(&sec_bytes).expect("secret key bytes");
        let pairs = owned.checkpoint_pairs();
        let root = crate::kv::store::content_root(owned.id(), owned.name(), &pairs);
        let cp = crate::kv::store::make_owner_checkpoint(crate::kv::store::OwnerCheckpointParams {
            topic,
            store_id: &store_id(1),
            secret_key: &secret_key,
            public_key: &public_key,
            policy: &AccessPolicy::Signed,
            policy_version: owned.policy_version(),
            checkpoint_seq: 3,
            content_root: root,
            timestamp: 1,
        })
        .expect("sign empty checkpoint");
        owned.latest_checkpoint = Some(cp);
        owned.highest_checkpoint_seq = 3;
        // Stale replica: still holds a key the owner deleted (hwm 0).
        // Started FIRST so its subscriptions are fully registered before
        // any response can be published (in-process paused-time harness:
        // a response racing the main-topic registration is silently
        // missed; production requesters simply retry, but the poll loop
        // below burns virtual time much faster than real deliveries).
        let mut replica = KvStore::new_replica(
            store_id(1),
            String::new(),
            Some(owner_id),
            crate::kv::store::AnchorChannel::Persistence,
        );
        replica
            .put(
                "k_stale".to_string(),
                b"obsolete".to_vec(),
                "text/plain".to_string(),
                peer(2),
            )
            .expect("seed stale key");
        let joiner = KvStoreSync::new(
            replica,
            Arc::clone(&pubsub),
            topic.to_string(),
            peer(2),
            Some(agent(2)),
        )
        .expect("joiner sync");
        joiner.start().await.expect("start joiner");

        let owner_sync = KvStoreSync::new(
            owned,
            Arc::clone(&pubsub),
            topic.to_string(),
            peer(1),
            Some(owner_id),
        )
        .expect("owner sync");
        // Harness artifact: both syncs share ONE signing identity (the
        // pubsub ctx), so the stale joiner's responses arrive signed AS THE
        // OWNER — the empty owner's own bootstrap requester (empty ⇒
        // snapshot-loss recovery) would adopt the stale key back and its
        // checkpoint root would never match again. Production daemons have
        // distinct identities (a stale replica's response is not
        // owner-authorized), so silence the owner's requester: it is not
        // the machinery under test.
        owner_sync.silence_bootstrap();
        owner_sync.start().await.expect("start owner");

        // The replica's request must be answered with the checkpoint-bearing
        // empty full delta; adopting it removes the stale key and raises the
        // high-water mark to the served checkpoint (convergence reachable).
        // Poll generously: virtual time advances the requester schedule, but
        // the signing/delivery pipeline runs in REAL time (blocking-pool
        // ML-DSA ops) — each iteration donates a real scheduling window, so
        // under CPU contention more iterations are needed, not more virtual
        // seconds.
        let mut cleaned = false;
        for _ in 0..400 {
            tokio::time::sleep(Duration::from_secs(2)).await;
            let s = joiner.read().await;
            if s.get("k_stale").is_none() && s.highest_checkpoint_seq == 3 {
                cleaned = true;
                break;
            }
        }
        assert!(
            cleaned,
            "the checkpointed empty owner's response must full-replace the \
             stale replica (key removed, checkpoint HWM adopted)"
        );
        assert!(
            joiner.read().await.is_empty(),
            "replica converges to the owner's (empty) state"
        );
    }

    /// WHY (round-1 review — missed-delta recovery): a snapshot-restored
    /// NON-OWNER replica is non-empty, but may have missed deltas written
    /// while it was offline, and the gossip cache only replays ~60s. The
    /// old `is_empty()` bootstrap gate meant such a replica NEVER requested
    /// state — the missed keys were unrecoverable without another restart
    /// race. Non-owner replicas must always run the front burst.
    #[tokio::test(start_paused = true)]
    async fn restored_non_empty_replica_still_requests_missed_state() {
        let node = make_node().await;
        let kp = crate::identity::AgentKeypair::generate().expect("keypair");
        let owner_id = kp.agent_id();
        let ctx = Arc::new(crate::gossip::SigningContext::from_keypair(&kp));
        let pubsub = Arc::new(PubSubManager::new(node, Some(ctx)).expect("pubsub"));
        let topic = "kv-238-missed-delta";

        // Owner holds k_old AND k_new (k_new written while the replica was
        // "offline" — i.e. absent from the replica's restored snapshot).
        let mut owned = KvStore::new(
            store_id(1),
            "log".to_string(),
            owner_id,
            AccessPolicy::Signed,
        );
        owned
            .put(
                "k_old".to_string(),
                b"v_old".to_vec(),
                "text/plain".to_string(),
                peer(1),
            )
            .expect("owner put k_old");
        owned
            .put(
                "k_new".to_string(),
                b"v_new".to_vec(),
                "text/plain".to_string(),
                peer(1),
            )
            .expect("owner put k_new");
        let owner_sync = KvStoreSync::new(
            owned,
            Arc::clone(&pubsub),
            topic.to_string(),
            peer(1),
            Some(owner_id),
        )
        .expect("owner sync");
        owner_sync.start().await.expect("start owner");

        // Snapshot-restored replica: NON-empty (has k_old), anchored on the
        // owner, local agent is NOT the owner. Under the old gate this
        // replica never requested anything.
        let mut replica = KvStore::new_replica(
            store_id(1),
            String::new(),
            Some(owner_id),
            crate::kv::store::AnchorChannel::Persistence,
        );
        replica
            .put(
                "k_old".to_string(),
                b"v_old".to_vec(),
                "text/plain".to_string(),
                peer(2),
            )
            .expect("seed restored key");
        let joiner = KvStoreSync::new(
            replica,
            Arc::clone(&pubsub),
            topic.to_string(),
            peer(2),
            Some(agent(2)),
        )
        .expect("joiner sync");
        joiner.start().await.expect("start joiner");

        // The front burst must fire despite the replica being non-empty,
        // and the owner's full-state response must deliver the missed key.
        let mut recovered = false;
        for _ in 0..30 {
            tokio::time::sleep(Duration::from_secs(2)).await;
            if joiner.read().await.get("k_new").is_some() {
                recovered = true;
                break;
            }
        }
        assert!(
            recovered,
            "a restored non-empty non-owner replica must still request \
             state and recover deltas it missed while offline"
        );
    }

    /// WHY (issue #238 — zombie subscription + transient policy misreport):
    /// a replica that joins while the store owner is offline must still
    /// converge when the owner returns AFTER the front-loaded request
    /// schedule has exhausted. Before the fix the requester died at ~51s
    /// and nothing ever asked again; the replica stayed permanently empty
    /// (and permanently reported the `signed` replica-default policy) until
    /// a full daemon restart minted a fresh schedule. Paused time drives
    /// the virtual clock, so the multi-minute scenario runs in moments.
    #[tokio::test(start_paused = true)]
    async fn requester_tail_recovers_when_owner_returns_after_front_schedule() {
        let node = make_node().await;
        // Sign as the owner so the OwnerAnnounce path (policy refresh) is
        // exercised: v2 delivery exposes the verified sender AgentId, which
        // learn_ownership requires to equal the claimed owner.
        let kp = crate::identity::AgentKeypair::generate().expect("keypair");
        let owner_id = kp.agent_id();
        let ctx = Arc::new(crate::gossip::SigningContext::from_keypair(&kp));
        let pubsub = Arc::new(PubSubManager::new(node, Some(ctx)).expect("pubsub"));
        let topic = "kv-238-zombie";

        // Empty replica anchored on the (offline) owner — exactly what the
        // daemon's rehydration path builds for a joined store.
        let replica = KvStore::new_replica(
            store_id(1),
            String::new(),
            Some(owner_id),
            crate::kv::store::AnchorChannel::RestParam,
        );
        let joiner = KvStoreSync::new(
            replica,
            Arc::clone(&pubsub),
            topic.to_string(),
            peer(2),
            Some(agent(2)),
        )
        .expect("joiner sync");
        joiner.start().await.expect("start joiner");

        // The entire front schedule (~51s) fires into the void.
        tokio::time::sleep(Duration::from_secs(60)).await;
        assert!(
            joiner.read().await.is_empty(),
            "nobody was online to answer the front schedule"
        );
        assert_eq!(
            *joiner.read().await.policy(),
            AccessPolicy::Signed,
            "replica still reports its construction-default policy while \
             the owner is away (the transient misreport under test)"
        );

        // The owner comes online only AFTER the front schedule exhausted —
        // the window in which the old requester was already dead.
        let mut owned = KvStore::new(
            store_id(1),
            "log".to_string(),
            owner_id,
            AccessPolicy::AppendOnly,
        );
        owned
            .put(
                "k1".to_string(),
                b"v1".to_vec(),
                "text/plain".to_string(),
                peer(1),
            )
            .expect("owner put");
        let owner_sync = KvStoreSync::new(
            owned,
            Arc::clone(&pubsub),
            topic.to_string(),
            peer(1),
            Some(owner_id),
        )
        .expect("owner sync");
        owner_sync.start().await.expect("start owner");

        // The persistent tail must ask again and converge the data.
        let mut converged = false;
        for _ in 0..200 {
            tokio::time::sleep(Duration::from_secs(5)).await;
            if !joiner.read().await.is_empty() {
                converged = true;
                break;
            }
        }
        assert!(
            converged,
            "the tail requester must recover the store once the owner \
             returns (zombie subscription, issue #238)"
        );
        assert_eq!(
            joiner.read().await.get("k1").map(|e| e.value.clone()),
            Some(b"v1".to_vec()),
            "the owner's key must arrive via the state response"
        );

        // Defect 3: the owner's announce rides the same recovery, so the
        // policy misreport heals with the data (poll — the announce and the
        // full-delta response are separate messages).
        let mut policy_ok = false;
        for _ in 0..60 {
            if *joiner.read().await.policy() == AccessPolicy::AppendOnly {
                policy_ok = true;
                break;
            }
            tokio::time::sleep(Duration::from_secs(1)).await;
        }
        assert!(
            policy_ok,
            "the owner announce must refresh the replica policy \
             (transient `signed` misreport, issue #238)"
        );
    }

    // ------------------------------------------------------------------
    // Issue #240: digest-verified convergence evidence
    // ------------------------------------------------------------------

    /// Drain every side-topic `StateRequest` from `from` already queued on
    /// `probe` without blocking (the 1ms virtual timeout yields immediately
    /// under the paused clock when the queue is empty).
    async fn drain_state_requests(probe: &mut crate::gossip::Subscription, from: PeerId) -> usize {
        let mut n = 0;
        while let Ok(Some(msg)) = tokio::time::timeout(Duration::from_millis(1), probe.recv()).await
        {
            if let Ok(KvSyncMessage::StateRequest { requester }) =
                bincode::deserialize::<KvSyncMessage>(&msg.payload)
            {
                if requester == from {
                    n += 1;
                }
            }
        }
        n
    }

    /// WHY: the v2 digest commits to the served entry set so a requester
    /// can verify completeness LOCALLY. It must be deterministic across
    /// replicas, sensitive to content, insensitive to the (placeholder)
    /// name, and bound to the store id; a full delta's carried digest must
    /// equal the serving store's local digest.
    #[test]
    fn served_digest_is_deterministic_content_bound_and_name_independent() {
        let owner = agent(1);
        let mut a = KvStore::new(
            store_id(1),
            "alpha".to_string(),
            owner,
            AccessPolicy::Signed,
        );
        let mut b = KvStore::new(store_id(1), "beta".to_string(), owner, AccessPolicy::Signed);
        // Empty stores: same id ⇒ same digest; the name is not content.
        assert_eq!(a.served_digest(), b.served_digest());
        let c = KvStore::new(
            store_id(2),
            "alpha".to_string(),
            owner,
            AccessPolicy::Signed,
        );
        assert_ne!(
            a.served_digest(),
            c.served_digest(),
            "the store id binds the digest (cross-store replay defense)"
        );

        // The same entry bytes in both stores ⇒ the same digest, however
        // they got there (writer tags are transport, not content).
        let entry = KvEntry::new("k".to_string(), b"v".to_vec(), "text/plain".to_string());
        let mut d1 = KvStoreDelta::new(1);
        d1.added.insert("k".to_string(), (entry, (peer(1), 1)));
        a.merge_delta(&d1, peer(1), Some(&owner)).expect("merge a");
        b.merge_delta(&d1, peer(1), Some(&owner)).expect("merge b");
        assert_eq!(a.served_digest(), b.served_digest());

        // A delta's served digest exists only for full-state shapes
        // (name_update present), and then equals the local digest.
        assert_eq!(
            d1.served_digest(&store_id(1)),
            None,
            "incremental shape must not impersonate a full serve"
        );
        d1.name_update = Some(a.name_register().clone());
        assert_eq!(d1.served_digest(&store_id(1)), Some(a.served_digest()));

        // Different membership ⇒ different digest.
        let entry2 = KvEntry::new("k2".to_string(), b"w".to_vec(), "text/plain".to_string());
        let mut d2 = KvStoreDelta::new(2);
        d2.added.insert("k2".to_string(), (entry2, (peer(1), 2)));
        b.merge_delta(&d2, peer(1), Some(&owner)).expect("merge b2");
        assert_ne!(a.served_digest(), b.served_digest());
    }

    /// WHY (issue #240, residual 1 — the cross-topic loss window): the full
    /// delta travels on the main topic, its marker on the side topic, with
    /// no delivery coupling. If the delta is lost while the marker
    /// survives, the v1 rule stopped a non-empty requester with incomplete
    /// history. With the v2 digest the requester detects the mismatch
    /// (local {k2} vs declared {k1,k2}) and keeps asking until the real
    /// state arrives.
    #[tokio::test(start_paused = true)]
    async fn lost_full_delta_with_surviving_marker_keeps_requester_asking() {
        let node = make_node().await;
        let kp = crate::identity::AgentKeypair::generate().expect("keypair");
        let owner_id = kp.agent_id();
        let ctx = Arc::new(crate::gossip::SigningContext::from_keypair(&kp));
        let pubsub = Arc::new(PubSubManager::new(node, Some(ctx)).expect("pubsub"));
        let topic = "kv-240-lost-broadcast";
        let side = format!("{topic}{STATE_SYNC_TOPIC_SUFFIX}");

        // The full holder state (owner offline for now): {k1, k2}.
        let mut owned = KvStore::new(
            store_id(1),
            "log".to_string(),
            owner_id,
            AccessPolicy::Signed,
        );
        owned
            .put(
                "k1".to_string(),
                b"v1".to_vec(),
                "text/plain".to_string(),
                peer(1),
            )
            .expect("k1");
        owned
            .put(
                "k2".to_string(),
                b"v2".to_vec(),
                "text/plain".to_string(),
                peer(1),
            )
            .expect("k2");

        // The joiner holds only k2 — "one live incremental delta" — with
        // entry bytes IDENTICAL to the holder's (merged out of the holder's
        // own full delta), so the post-recovery digests can match exactly.
        let mut replica = KvStore::new_replica(
            store_id(1),
            String::new(),
            Some(owner_id),
            crate::kv::store::AnchorChannel::RestParam,
        );
        let k2_only = {
            let full = owned.full_delta();
            let (key, (entry, tag)) = full
                .added
                .iter()
                .find(|(k, _)| k.as_str() == "k2")
                .expect("k2 in full delta");
            let mut d = KvStoreDelta::new(1);
            d.added.insert(key.clone(), (entry.clone(), *tag));
            d
        };
        replica
            .merge_delta(&k2_only, peer(1), Some(&owner_id))
            .expect("seed k2");
        let joiner = KvStoreSync::new(
            replica,
            Arc::clone(&pubsub),
            topic.to_string(),
            peer(2),
            Some(agent(2)),
        )
        .expect("joiner sync");
        joiner.start().await.expect("start joiner");
        let mut probe = pubsub.subscribe(side.clone()).await;

        // The loss window: the marker survives, the full delta does not.
        // Publish ONLY the v2 marker the holder would have sent.
        let marker = KvSyncMessage::StateServedV2 {
            responder: peer(1),
            digest: owned.served_digest(),
            entry_count: 2,
        };
        let bytes = bincode::serialize(&marker).expect("serialize marker");
        pubsub
            .publish(side.clone(), bytes::Bytes::from(bytes))
            .await
            .expect("publish marker");

        // Well past the front burst the requester must STILL be asking —
        // its local digest ({k2}) does not match the declaration ({k1,k2}).
        let mut requests = 0;
        for _ in 0..10 {
            tokio::time::sleep(Duration::from_secs(30)).await;
            requests += drain_state_requests(&mut probe, peer(2)).await;
        }
        assert!(
            requests > 0,
            "a requester whose full delta was lost must keep asking (digest mismatch)"
        );
        assert!(
            joiner.read().await.get("k1").is_none(),
            "the lost broadcast never arrived"
        );

        // The real holder returns: the next serve delivers {k1,k2}, the
        // local digest then matches the declaration, and the tail stops.
        let owner_sync = KvStoreSync::new(
            owned,
            Arc::clone(&pubsub),
            topic.to_string(),
            peer(1),
            Some(owner_id),
        )
        .expect("owner sync");
        owner_sync.start().await.expect("start owner");

        let mut recovered = false;
        for _ in 0..200 {
            tokio::time::sleep(Duration::from_secs(5)).await;
            if joiner.read().await.get("k1").is_some() {
                recovered = true;
                break;
            }
        }
        assert!(
            recovered,
            "the still-alive requester must recover the lost key"
        );

        // Convergence is terminal: no further requests.
        tokio::time::sleep(Duration::from_secs(160)).await;
        drain_state_requests(&mut probe, peer(2)).await;
        let mut relapse = 0;
        for _ in 0..20 {
            tokio::time::sleep(Duration::from_secs(30)).await;
            relapse += drain_state_requests(&mut probe, peer(2)).await;
        }
        assert_eq!(relapse, 0, "a digest-matched requester must fall silent");
    }

    /// WHY (issue #240, residual 2 — deletion cold-sync): a checkpoint-less
    /// full delta carries only live entries, so a plain merge could never
    /// delete a stale replica's obsolete keys. The digest-verified
    /// full-replace adopt closes that: the serve's delta content is bound
    /// to the holder's declared digest, so pruning local keys it omits is
    /// safe. A plain v1-era merge leaves the stale key in place (the
    /// pre-#240 behavior the issue describes).
    #[tokio::test(start_paused = true)]
    async fn digest_verified_full_serve_prunes_stale_keys() {
        let node = make_node().await;
        let kp = crate::identity::AgentKeypair::generate().expect("keypair");
        let owner_id = kp.agent_id();
        let ctx = Arc::new(crate::gossip::SigningContext::from_keypair(&kp));
        let pubsub = Arc::new(PubSubManager::new(node, Some(ctx)).expect("pubsub"));
        let topic = "kv-240-prune-stale";
        let side = format!("{topic}{STATE_SYNC_TOPIC_SUFFIX}");

        // Owner: checkpoint-less, holding only k_live.
        let mut owned = KvStore::new(
            store_id(1),
            "log".to_string(),
            owner_id,
            AccessPolicy::Signed,
        );
        owned
            .put(
                "k_live".to_string(),
                b"v".to_vec(),
                "text/plain".to_string(),
                peer(1),
            )
            .expect("owner put");

        // Stale replica: k_live (byte-identical, from the owner's own full
        // delta) PLUS k_stale, an obsolete key the owner deleted while the
        // replica was away. Started FIRST so its subscriptions are fully
        // registered before any response can be published (in-process
        // paused-time harness).
        let mut replica = KvStore::new_replica(
            store_id(1),
            String::new(),
            Some(owner_id),
            crate::kv::store::AnchorChannel::Persistence,
        );
        let live_only = {
            let full = owned.full_delta();
            let (key, (entry, tag)) = full
                .added
                .iter()
                .find(|(k, _)| k.as_str() == "k_live")
                .expect("k_live in full delta");
            let mut d = KvStoreDelta::new(1);
            d.added.insert(key.clone(), (entry.clone(), *tag));
            d
        };
        replica
            .merge_delta(&live_only, peer(1), Some(&owner_id))
            .expect("seed k_live");
        replica
            .put(
                "k_stale".to_string(),
                b"obsolete".to_vec(),
                "text/plain".to_string(),
                peer(2),
            )
            .expect("seed stale key");
        let joiner = KvStoreSync::new(
            replica,
            Arc::clone(&pubsub),
            topic.to_string(),
            peer(2),
            Some(agent(2)),
        )
        .expect("joiner sync");
        joiner.start().await.expect("start joiner");
        let mut probe = pubsub.subscribe(side.clone()).await;

        let owner_sync = KvStoreSync::new(
            owned,
            Arc::clone(&pubsub),
            topic.to_string(),
            peer(1),
            Some(owner_id),
        )
        .expect("owner sync");
        owner_sync.start().await.expect("start owner");

        // The verified adopt must remove the stale key while k_live stays.
        let mut pruned = false;
        for _ in 0..200 {
            tokio::time::sleep(Duration::from_secs(5)).await;
            let s = joiner.read().await;
            if s.get("k_stale").is_none() && s.get("k_live").is_some() {
                pruned = true;
                break;
            }
        }
        assert!(
            pruned,
            "the digest-verified full serve must prune the stale key \
             (checkpoint-less deletion cold-sync)"
        );

        // And convergence follows: local state now equals the declared
        // digest, so the requester falls silent.
        tokio::time::sleep(Duration::from_secs(160)).await;
        drain_state_requests(&mut probe, peer(2)).await;
        let mut relapse = 0;
        for _ in 0..20 {
            tokio::time::sleep(Duration::from_secs(30)).await;
            relapse += drain_state_requests(&mut probe, peer(2)).await;
        }
        assert_eq!(relapse, 0, "the requester must stop once the digests match");
    }

    /// WHY (issue #240, residual 3 — genuinely-empty chatter): the v1 rule
    /// kept every empty checkpoint-less replica requesting forever (~1
    /// side-topic message per 5 minutes) because an empty non-owner holder
    /// had to stay silent. The v2 digest of the empty set is universally
    /// computable, so an empty holder can now declare authoritative
    /// emptiness ANY empty requester verifies locally — two empty replicas
    /// converging on empty is verifiably CORRECT, not false convergence.
    #[tokio::test(start_paused = true)]
    async fn empty_holder_v2_marker_terminates_empty_requester() {
        let node = make_node().await;
        let kp = crate::identity::AgentKeypair::generate().expect("keypair");
        let owner_id = kp.agent_id();
        let ctx = Arc::new(crate::gossip::SigningContext::from_keypair(&kp));
        let pubsub = Arc::new(PubSubManager::new(node, Some(ctx)).expect("pubsub"));
        let topic = "kv-240-empty-silence";
        let side = format!("{topic}{STATE_SYNC_TOPIC_SUFFIX}");

        // Both EMPTY non-owner replicas anchored on the (offline) owner.
        let joiner = KvStoreSync::new(
            KvStore::new_replica(
                store_id(1),
                String::new(),
                Some(owner_id),
                crate::kv::store::AnchorChannel::RestParam,
            ),
            Arc::clone(&pubsub),
            topic.to_string(),
            peer(2),
            Some(agent(2)),
        )
        .expect("joiner sync");
        joiner.start().await.expect("start joiner");
        let holder = KvStoreSync::new(
            KvStore::new_replica(
                store_id(1),
                String::new(),
                Some(owner_id),
                crate::kv::store::AnchorChannel::RestParam,
            ),
            Arc::clone(&pubsub),
            topic.to_string(),
            peer(1),
            Some(agent(3)),
        )
        .expect("holder sync");
        holder.start().await.expect("start holder");
        let mut probe = pubsub.subscribe(side.clone()).await;

        // Warm-up: the front burst fires and the empty holder's v2
        // declarations arrive; convergence should follow within a few tail
        // checks.
        tokio::time::sleep(Duration::from_secs(160)).await;
        drain_state_requests(&mut probe, peer(2)).await;
        let mut late = 0;
        for _ in 0..20 {
            tokio::time::sleep(Duration::from_secs(30)).await;
            late += drain_state_requests(&mut probe, peer(2)).await;
        }
        assert_eq!(
            late, 0,
            "an empty holder's verifiable digest must terminate the empty \
             requester's tail (genuinely-empty stores converge silently)"
        );
    }

    /// WHY: wire compatibility is additive — a fleet with only v1 (older)
    /// responders must behave exactly as before: a v1 marker plus local
    /// state converges the requester. The v2 machinery must not require
    /// v2 markers to make progress against old peers.
    #[tokio::test(start_paused = true)]
    async fn v1_marker_from_old_peer_still_converges() {
        let node = make_node().await;
        let kp = crate::identity::AgentKeypair::generate().expect("keypair");
        let owner_id = kp.agent_id();
        let ctx = Arc::new(crate::gossip::SigningContext::from_keypair(&kp));
        let pubsub = Arc::new(PubSubManager::new(node, Some(ctx)).expect("pubsub"));
        let topic = "kv-240-v1-compat";
        let side = format!("{topic}{STATE_SYNC_TOPIC_SUFFIX}");

        let joiner = KvStoreSync::new(
            KvStore::new_replica(
                store_id(1),
                String::new(),
                Some(owner_id),
                crate::kv::store::AnchorChannel::RestParam,
            ),
            Arc::clone(&pubsub),
            topic.to_string(),
            peer(2),
            Some(agent(2)),
        )
        .expect("joiner sync");
        joiner.start().await.expect("start joiner");
        let mut probe = pubsub.subscribe(side.clone()).await;

        // An "old peer" answers a request: full delta on the main topic,
        // v1 StateServed marker on the side topic — never a v2 marker.
        tokio::time::sleep(Duration::from_secs(20)).await;
        let mut owned = KvStore::new(
            store_id(1),
            "log".to_string(),
            owner_id,
            AccessPolicy::Signed,
        );
        owned
            .put(
                "k1".to_string(),
                b"v1".to_vec(),
                "text/plain".to_string(),
                peer(1),
            )
            .expect("k1");
        let full = owned.full_delta();
        let encoded = encode_delta(peer(1), &full).expect("encode full");
        pubsub
            .publish(topic.to_string(), bytes::Bytes::from(encoded))
            .await
            .expect("publish full delta");
        let marker = KvSyncMessage::StateServed {
            responder: peer(1),
            empty: false,
            checkpoint_seq: None,
        };
        let marker_bytes = bincode::serialize(&marker).expect("serialize v1 marker");
        pubsub
            .publish(side.clone(), bytes::Bytes::from(marker_bytes))
            .await
            .expect("publish v1 marker");

        let mut recovered = false;
        for _ in 0..60 {
            tokio::time::sleep(Duration::from_secs(5)).await;
            if joiner.read().await.get("k1").is_some() {
                recovered = true;
                break;
            }
        }
        assert!(recovered, "the old peer's full delta must merge");

        // Weak-evidence convergence: v1 marker + local state ⇒ the tail stops.
        tokio::time::sleep(Duration::from_secs(160)).await;
        drain_state_requests(&mut probe, peer(2)).await;
        let mut late = 0;
        for _ in 0..20 {
            tokio::time::sleep(Duration::from_secs(30)).await;
            late += drain_state_requests(&mut probe, peer(2)).await;
        }
        assert_eq!(
            late, 0,
            "v1 evidence from an old peer must still converge the requester"
        );
    }

    /// WHY: a v2 marker whose digest does not correspond to any state the
    /// requester can hold (forged or corrupt) must NEVER converge it — the
    /// digest is self-verifying, so a bad declaration can only delay, never
    /// cause, convergence. Nor may it wedge later recovery: a genuine
    /// holder's fresh declaration replaces the bad one (per-responder
    /// latest-wins).
    #[tokio::test(start_paused = true)]
    async fn tampered_digest_is_rejected_and_does_not_wedge_recovery() {
        let node = make_node().await;
        let kp = crate::identity::AgentKeypair::generate().expect("keypair");
        let owner_id = kp.agent_id();
        let ctx = Arc::new(crate::gossip::SigningContext::from_keypair(&kp));
        let pubsub = Arc::new(PubSubManager::new(node, Some(ctx)).expect("pubsub"));
        let topic = "kv-240-tampered";
        let side = format!("{topic}{STATE_SYNC_TOPIC_SUFFIX}");

        let joiner = KvStoreSync::new(
            KvStore::new_replica(
                store_id(1),
                String::new(),
                Some(owner_id),
                crate::kv::store::AnchorChannel::RestParam,
            ),
            Arc::clone(&pubsub),
            topic.to_string(),
            peer(2),
            Some(agent(2)),
        )
        .expect("joiner sync");
        joiner.start().await.expect("start joiner");
        let mut probe = pubsub.subscribe(side.clone()).await;

        // The tampered marker: a random digest no real state can match.
        let marker = KvSyncMessage::StateServedV2 {
            responder: peer(1),
            digest: [0xAB; 32],
            entry_count: 2,
        };
        let bytes = bincode::serialize(&marker).expect("serialize marker");
        pubsub
            .publish(side.clone(), bytes::Bytes::from(bytes))
            .await
            .expect("publish tampered marker");

        // The requester keeps asking: its (empty) local digest can never
        // equal the forged declaration.
        let mut requests = 0;
        for _ in 0..8 {
            tokio::time::sleep(Duration::from_secs(30)).await;
            requests += drain_state_requests(&mut probe, peer(2)).await;
        }
        assert!(
            requests > 0,
            "a tampered digest must not converge the requester"
        );
        assert!(
            joiner.read().await.is_empty(),
            "no state can have been adopted from a forged declaration"
        );

        // The genuine holder appears; its fresh declaration replaces the
        // tampered one (per-responder latest-wins) and recovery completes.
        let mut owned = KvStore::new(
            store_id(1),
            "log".to_string(),
            owner_id,
            AccessPolicy::Signed,
        );
        owned
            .put(
                "k1".to_string(),
                b"v1".to_vec(),
                "text/plain".to_string(),
                peer(1),
            )
            .expect("k1");
        let owner_sync = KvStoreSync::new(
            owned,
            Arc::clone(&pubsub),
            topic.to_string(),
            peer(1),
            Some(owner_id),
        )
        .expect("owner sync");
        owner_sync.start().await.expect("start owner");

        let mut recovered = false;
        for _ in 0..200 {
            tokio::time::sleep(Duration::from_secs(5)).await;
            if joiner.read().await.get("k1").is_some() {
                recovered = true;
                break;
            }
        }
        assert!(
            recovered,
            "a tampered marker must not wedge recovery from a genuine holder"
        );
    }

    /// WHY (F1, fix-loop — delete+recreate wedge): `KvEntry::merge`
    /// preserves the EARLIEST `created_at`, so a replica holding the
    /// original entry and an owner that deleted+re-created the key
    /// permanently disagree on `created_at`. A served digest committing it
    /// could never match and the requester would wedge at capped cadence
    /// forever. The digest excludes `created_at` (merge-converged fields
    /// only), so the re-created key converges.
    #[tokio::test(start_paused = true)]
    async fn recreated_key_converges_after_owner_delete_and_recreate() {
        let node = make_node().await;
        let kp = crate::identity::AgentKeypair::generate().expect("keypair");
        let owner_id = kp.agent_id();
        let ctx = Arc::new(crate::gossip::SigningContext::from_keypair(&kp));
        let pubsub = Arc::new(PubSubManager::new(node, Some(ctx)).expect("pubsub"));
        let topic = "kv-240-recreate";
        let side = format!("{topic}{STATE_SYNC_TOPIC_SUFFIX}");

        // The ORIGINAL entry, as the replica synced it before going offline.
        let mut owned = KvStore::new(
            store_id(1),
            "log".to_string(),
            owner_id,
            AccessPolicy::Signed,
        );
        owned
            .put(
                "k".to_string(),
                b"v1".to_vec(),
                "text/plain".to_string(),
                peer(1),
            )
            .expect("original put");
        // Age the replica's copy deterministically: even in the same
        // wall-clock millisecond, the replica's created/updated timestamps
        // are strictly older than the re-created entry's.
        let mut aged = owned.get("k").expect("original entry").clone();
        aged.created_at -= 10_000;
        aged.updated_at -= 10_000;

        // The owner DELETES and RE-CREATES the key while the replica is
        // away: the owner's entry now carries a fresh created_at.
        owned.remove("k").expect("owner delete");
        owned
            .put(
                "k".to_string(),
                b"v2".to_vec(),
                "text/plain".to_string(),
                peer(1),
            )
            .expect("owner recreate");

        // The replica returns holding the ORIGINAL (aged) entry.
        let mut replica = KvStore::new_replica(
            store_id(1),
            String::new(),
            Some(owner_id),
            crate::kv::store::AnchorChannel::Persistence,
        );
        let mut seed = KvStoreDelta::new(1);
        seed.added.insert("k".to_string(), (aged, (peer(1), 1)));
        replica
            .merge_delta(&seed, peer(1), Some(&owner_id))
            .expect("seed original entry");
        let joiner = KvStoreSync::new(
            replica,
            Arc::clone(&pubsub),
            topic.to_string(),
            peer(2),
            Some(agent(2)),
        )
        .expect("joiner sync");
        joiner.start().await.expect("start joiner");
        let mut probe = pubsub.subscribe(side.clone()).await;
        let owner_sync = KvStoreSync::new(
            owned,
            Arc::clone(&pubsub),
            topic.to_string(),
            peer(1),
            Some(owner_id),
        )
        .expect("owner sync");
        owner_sync.start().await.expect("start owner");

        // The re-created value must arrive…
        let mut recovered = false;
        for _ in 0..200 {
            tokio::time::sleep(Duration::from_secs(5)).await;
            if joiner.read().await.get("k").map(|e| e.value.clone()) == Some(b"v2".to_vec()) {
                recovered = true;
                break;
            }
        }
        assert!(recovered, "the re-created entry must merge");

        // …and the requester must CONVERGE, not wedge on the created_at
        // mismatch (the pre-fix failure mode).
        tokio::time::sleep(Duration::from_secs(160)).await;
        drain_state_requests(&mut probe, peer(2)).await;
        let mut relapse = 0;
        for _ in 0..20 {
            tokio::time::sleep(Duration::from_secs(30)).await;
            relapse += drain_state_requests(&mut probe, peer(2)).await;
        }
        assert_eq!(
            relapse, 0,
            "a delete+recreate must not wedge the requester on created_at"
        );
    }

    /// WHY (F2, fix-loop — multi-writer truncation): absence from a
    /// verified serve proves deletion ONLY when the sender is the sole
    /// possible content author. In an Allowlisted store the owner's serve
    /// can be legitimately incomplete about co-writers' keys, so the adopt
    /// must NOT prune there — otherwise an owner-only serve truncates an
    /// allowlisted writer's legitimate key during the bootstrap window.
    #[tokio::test(start_paused = true)]
    async fn allowlisted_writer_key_survives_owner_only_serve() {
        let node = make_node().await;
        let kp = crate::identity::AgentKeypair::generate().expect("keypair");
        let owner_id = kp.agent_id();
        let ctx = Arc::new(crate::gossip::SigningContext::from_keypair(&kp));
        let pubsub = Arc::new(PubSubManager::new(node, Some(ctx)).expect("pubsub"));
        let topic = "kv-240-allowlisted";
        let writer = agent(7);

        // Owner: Allowlisted, holding only k_owner.
        let mut owned = KvStore::new(
            store_id(1),
            "log".to_string(),
            owner_id,
            AccessPolicy::Allowlisted,
        );
        owned.allow_writer(writer, &owner_id).expect("allow writer");
        owned
            .put(
                "k_owner".to_string(),
                b"v".to_vec(),
                "text/plain".to_string(),
                peer(1),
            )
            .expect("owner put");

        // Replica: anchored on the owner and already POLICY-AWARE (a
        // restored replica has the allowlist persisted; the merge below
        // requires it — under the Signed default the writer's key would be
        // rejected outright). It holds k_owner (aged copy from the owner's
        // own delta — the serve must refresh its updated_at, proving the
        // verified adopt path actually ran) and k_writer, written by the
        // allowlisted writer, which the owner has NOT merged.
        let mut replica = KvStore::new_replica(
            store_id(1),
            String::new(),
            Some(owner_id),
            crate::kv::store::AnchorChannel::Persistence,
        );
        replica
            .learn_ownership(
                owner_id,
                AccessPolicy::Allowlisted,
                owned.policy_version(),
                &owner_id,
            )
            .expect("learn policy");
        // The replica must also know the allowlist itself (learned via
        // owner-gated deltas in production) or the writer's key merge is
        // rejected by access control.
        replica
            .allow_writer(writer, &owner_id)
            .expect("learn allowlist");
        let k_owner_seed = {
            let full = owned.full_delta();
            let (key, (entry, tag)) = full
                .added
                .iter()
                .find(|(k, _)| k.as_str() == "k_owner")
                .expect("k_owner in full delta");
            let mut aged = entry.clone();
            aged.created_at -= 10_000;
            aged.updated_at -= 10_000;
            let mut d = KvStoreDelta::new(1);
            d.added.insert(key.clone(), (aged, *tag));
            d
        };
        replica
            .merge_delta(&k_owner_seed, peer(1), Some(&owner_id))
            .expect("seed k_owner");
        let writer_entry = KvEntry::new(
            "k_writer".to_string(),
            b"w".to_vec(),
            "text/plain".to_string(),
        );
        let mut writer_delta = KvStoreDelta::new(2);
        writer_delta
            .added
            .insert("k_writer".to_string(), (writer_entry, (peer(7), 1)));
        replica
            .merge_delta(&writer_delta, peer(7), Some(&writer))
            .expect("seed k_writer");
        let owner_updated_at = owned.get("k_owner").expect("owner entry").updated_at;

        let joiner = KvStoreSync::new(
            replica,
            Arc::clone(&pubsub),
            topic.to_string(),
            peer(2),
            Some(agent(2)),
        )
        .expect("joiner sync");
        joiner.start().await.expect("start joiner");
        let owner_sync = KvStoreSync::new(
            owned,
            Arc::clone(&pubsub),
            topic.to_string(),
            peer(1),
            Some(owner_id),
        )
        .expect("owner sync");
        owner_sync.start().await.expect("start owner");

        // Drive several serve windows (the requester keeps asking — its
        // digest {k_owner,k_writer} never matches the owner's {k_owner}
        // declaration until the owner absorbs the co-write).
        let mut serve_landed = false;
        for _ in 0..60 {
            tokio::time::sleep(Duration::from_secs(5)).await;
            let s = joiner.read().await;
            assert!(
                s.get("k_writer").is_some(),
                "an owner-only serve must NOT prune an allowlisted writer's key"
            );
            if s.get("k_owner").map(|e| e.updated_at) == Some(owner_updated_at) {
                serve_landed = true;
            }
        }
        assert!(
            serve_landed,
            "the owner's verified serve must have merged (aged entry refreshed)"
        );
    }

    /// WHY (F3, fix-loop — tombstone + hardcoded-tag deadlock): the adopt's
    /// prune is a local observe-remove, tombstoning the tags a previous
    /// full delta used. With a hardcoded synthetic tag, a later serve
    /// re-adding the same key would be silently rejected forever. Full
    /// deltas now mint FRESH tags, so a re-served key is accepted.
    #[test]
    fn pruned_key_is_accepted_when_re_served_with_fresh_tags() {
        let owner = agent(1);
        let mut holder = KvStore::new(store_id(1), "log".to_string(), owner, AccessPolicy::Signed);
        holder
            .put(
                "k_live".to_string(),
                b"v".to_vec(),
                "text/plain".to_string(),
                peer(1),
            )
            .expect("put live");
        holder
            .put(
                "k_doomed".to_string(),
                b"x".to_vec(),
                "text/plain".to_string(),
                peer(1),
            )
            .expect("put doomed");

        // The replica absorbs a first full serve (both keys).
        let mut replica = KvStore::new_replica(
            store_id(1),
            String::new(),
            Some(owner),
            crate::kv::store::AnchorChannel::RestParam,
        );
        let s1 = holder.full_delta();
        replica
            .merge_delta(&s1, peer(1), Some(&owner))
            .expect("serve 1");
        assert!(replica.get("k_doomed").is_some());

        // The holder deletes the key; the next VERIFIED serve prunes it
        // (tombstoning the first serve's synthetic tag locally).
        holder.remove("k_doomed").expect("delete");
        let s2 = holder.full_delta();
        assert_eq!(
            s2.served_digest(&store_id(1)),
            Some(holder.served_digest()),
            "the serve must carry the holder's declared digest"
        );
        replica
            .merge_delta(&s2, peer(1), Some(&owner))
            .expect("serve 2");
        assert_eq!(replica.prune_to_served_set(&s2), 1);
        assert!(replica.get("k_doomed").is_none());

        // The holder RE-ADDS the key: a later serve must be accepted —
        // pre-fix, its synthetic tag was tombstoned by the prune and the
        // re-add silently dropped.
        holder
            .put(
                "k_doomed".to_string(),
                b"y".to_vec(),
                "text/plain".to_string(),
                peer(1),
            )
            .expect("re-add");
        let s3 = holder.full_delta();
        replica
            .merge_delta(&s3, peer(1), Some(&owner))
            .expect("serve 3");
        let entry = replica
            .get("k_doomed")
            .expect("a re-served key must be accepted after a prune");
        assert_eq!(entry.value, b"y");
    }
}