//! Automerge + Iroh-backed implementation of the sync backend traits.
//!
//! `AutomergeBackend` composes the production primitives that already
//! live in `peat-mesh::storage` — `AutomergeStore`, `MeshSyncTransport`,
//! `AutomergeSyncCoordinator`, `SyncProtocolHandler`,
//! `NetworkedIrohBlobStore` — and presents them as a turnkey
//! `Arc<dyn DataSyncBackend>` for consumers of `peat_mesh::Node`.
//!
//! Background: prior to this adapter, every consumer that wanted the
//! full Automerge+Iroh stack reproduced the same 7-step bootstrap by
//! hand. The only `DataSyncBackend` impl shipped from peat-mesh itself
//! was `InMemoryBackend` (test-only); the production implementation
//! lived downstream in `peat-protocol::storage::AutomergeBackend`,
//! which sibling repos cannot reach without violating the
//! dependency-flow rule that peat-protocol depends on peat-mesh, not
//! the other way around. peat-mesh#106 closes that gap.
//!
//! Wire-compatible with on-disk data written by the prior 7-step
//! bootstrap — `AutomergeStore::open(data_dir.join("automerge"))`
//! reads existing `automerge.redb` directly; no migration.
//!
//! # Example
//!
//! ```ignore
//! use peat_mesh::sync::{AutomergeBackend, AutomergeBackendConfig};
//! use peat_mesh::Node;
//! use std::path::PathBuf;
//! use std::sync::Arc;
//!
//! let backend = AutomergeBackend::with_iroh(AutomergeBackendConfig {
//!     data_dir: PathBuf::from("/var/lib/myapp"),
//!     formation_id: "my-formation".to_string(),
//!     base64_shared_key: "base64-shared-key=".to_string(),
//!     iroh_bind_addr: None,
//!     cipher: None,
//! }).await?;
//! let node = Node::new(backend as Arc<dyn DataSyncBackend>);
//! ```
//!
//! To pin the Iroh QUIC endpoint to a specific UDP port (required for
//! direct-addressed deployments — Docker compose, k8s `hostPort`,
//! cross-cluster harnesses), set `iroh_bind_addr`:
//!
//! ```ignore
//! AutomergeBackendConfig {
//!     iroh_bind_addr: Some("0.0.0.0:51071".parse().unwrap()),
//!     ..cfg
//! }
//! ```

use crate::security::FormationKey;
use crate::storage::blob_announce;
use crate::storage::sync_transport::SyncTransport;
use crate::storage::{
    AutomergeStore, AutomergeSyncCoordinator, Cipher, InboundPeerRegistry, MeshSyncTransport,
    NetworkedIrohBlobStore, SyncChannelManager, SyncProtocolHandler, CAP_AUTOMERGE_ALPN,
};
use crate::sync::in_memory::evaluate_query;
use crate::sync::traits::{BackendInfo, DataSyncBackend, DocumentStore, PeerDiscovery, SyncEngine};
use crate::sync::types::{
    BackendConfig, ChangeEvent, ChangeStream, Document, DocumentId, PeerEvent, PeerId, PeerInfo,
    Priority, Query, SyncSubscription, TransportType,
};
use anyhow::{anyhow, bail, Context, Result};
use async_trait::async_trait;
use std::collections::{HashMap, HashSet};
use std::net::SocketAddr;
use std::path::PathBuf;
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::{Arc, Mutex as StdMutex, OnceLock};
use std::time::{Duration, SystemTime, UNIX_EPOCH};
use tokio::sync::{broadcast, mpsc, RwLock};
use tokio::task::JoinHandle;

/// Advance the given `AtomicU64` so its value ends strictly greater than
/// `observed` (peat-mesh#192 / peat-mesh#196).
///
/// Shared implementation of the Lamport receive-side rule
/// `L := max(L, observed) + 1`. Called from
/// [`AutomergeBackend::observe_lamport`] (consumer-facing API) and from
/// the sync coordinator's tombstone-ingest paths
/// (`AutomergeSyncCoordinator::observe_inbound_lamport`) so both call
/// sites share one CAS retry loop and one ordering choice.
///
/// Idempotent: a no-op when the atomic is already past `observed`.
/// Never regresses the clock — a strictly-greater local value wins.
///
/// Concurrency: `compare_exchange` with `AcqRel` on success / `Acquire`
/// on failure is sufficient to publish the new value to subsequent
/// loads on other threads. The loop is bounded by the number of
/// concurrent advancers — in practice it spins at most a handful of
/// times.
pub(crate) fn observe_atomic(atomic: &AtomicU64, observed: u64) {
    // Defensive cap: clamp the atomic's *final* value at `u64::MAX - 1`.
    // Without this, an `observed == u64::MAX` (reaching this through
    // `AutomergeSyncCoordinator::observe_inbound_lamport` from a
    // hostile or buggy peer) would set the atomic to `u64::MAX`;
    // the next `next_lamport()` would then run
    // `fetch_add(1).wrapping_add(1)` → return `0`, violating the
    // "never return 0" invariant the construction-time seed-floor
    // specifically protects (the value 0 is ambiguous with "unset"
    // for consumers stamping Tombstones with the result).
    //
    // Capping at `u64::MAX - 1` preserves all realistic Lamport
    // semantics (2^64 - 1 distinct values is more than any node
    // would ever issue in any plausible operating lifetime) and
    // forecloses the wrap-to-zero pathology. The local
    // `next_lamport` can still produce `u64::MAX` once when the
    // atomic sits at `u64::MAX - 1` (fetch_add yields the next
    // value `u64::MAX`); any subsequent call would wrap, which is
    // a 2^64-issuance regime not worth defending further.
    //
    // The cap goes on `candidate` (the *new* atomic value) rather
    // than on `observed` itself: capping `observed` would compute
    // `candidate = capped + 1 = u64::MAX` for `observed == u64::MAX`,
    // defeating the protection. Capping `candidate` correctly clamps
    // the post-observe state at `u64::MAX - 1`.
    let candidate = observed.saturating_add(1).min(u64::MAX - 1);
    loop {
        let cur = atomic.load(Ordering::Relaxed);
        // `cur >= candidate` rather than `cur > observed` so the
        // cap acts symmetrically with the candidate computation:
        // when `observed == u64::MAX` and the local atomic is
        // already at the cap (`u64::MAX - 1`), we short-circuit
        // instead of attempting a no-op CAS.
        if cur >= candidate {
            return;
        }
        match atomic.compare_exchange(cur, candidate, Ordering::AcqRel, Ordering::Acquire) {
            Ok(_) => return,
            // Lost the race; another writer advanced the clock.
            // Re-read and re-evaluate the predicate.
            Err(_) => continue,
        }
    }
}

/// Inputs for [`AutomergeBackend::with_iroh`].
///
/// Matches the peat-mesh house convention of `Config` struct + `::new(cfg)`
/// (see [`MeshConfig`](crate::config::MeshConfig),
/// [`TransportManagerConfig`](crate::transport::TransportManagerConfig),
/// [`LiteTransportConfig`](crate::transport::LiteTransportConfig), etc.).
/// Future knobs the original issue punted as out-of-scope (TTL semantics,
/// sync-mode enforcement, encryption-at-rest hooks) land here as additional
/// fields without breaking existing callers a second time.
#[derive(Debug, Clone)]
#[non_exhaustive]
pub struct AutomergeBackendConfig {
    /// Root directory for on-disk state. The adapter creates two
    /// subdirectories beneath it: `automerge/` (redb-backed CRDT
    /// store) and `iroh/` (blob store). Wire-compatible with the
    /// layout written by `peat-node`'s prior 7-step bootstrap.
    pub data_dir: PathBuf,
    /// Formation identifier — first arg to
    /// [`FormationKey::from_base64`](crate::security::FormationKey::from_base64).
    pub formation_id: String,
    /// Base64-encoded shared formation secret — second arg to
    /// [`FormationKey::from_base64`](crate::security::FormationKey::from_base64).
    pub base64_shared_key: String,
    /// Optional pinned bind address for the Iroh QUIC endpoint.
    ///
    /// - `None` (default): Iroh picks an ephemeral UDP port and binds
    ///   both the IPv4 unspecified address (`0.0.0.0`) and the IPv6
    ///   unspecified address (`[::]`). Suitable for relay-mediated
    ///   and dynamic deployments.
    /// - `Some(addr)`: pin the UDP socket to `addr`. Required for
    ///   direct-addressed deployments where peers reach this node by
    ///   `<host>:<port>` without relay assistance — Docker compose
    ///   stacks, k8s `hostPort` / `containerPort`, cross-cluster test
    ///   harnesses, and any deployment that exposes the Iroh UDP
    ///   port through an L4 load balancer.
    ///
    /// **Single-stack when pinned.** Under the hood the builder calls
    /// `.clear_ip_transports().bind_addr(addr)`, which *replaces* the
    /// default dual-stack `0.0.0.0` / `[::]` socket pair rather than
    /// adding to it. As a result only the address family of `addr`
    /// listens: passing `0.0.0.0:51071` binds IPv4 only; passing
    /// `[::]:51071` binds IPv6 only. If you need dual-stack with a
    /// pinned port, file an issue — exposing a `Vec<SocketAddr>` (one
    /// per family) is a small future extension once a real consumer
    /// asks.
    pub iroh_bind_addr: Option<SocketAddr>,
    /// Optional at-rest cipher (peat-mesh#124). When `Some`, the
    /// underlying [`AutomergeStore`] encrypts document bytes before
    /// every redb write and decrypts on every read. Sync continues
    /// to exchange plaintext patches over the wire — encryption is
    /// at-rest only. Defaults to `None` (no encryption).
    pub cipher: Option<Arc<dyn Cipher>>,
    /// Optional override for the blob-download stall threshold
    /// (peat-mesh#137 follow-up). `None` uses the [`IrohConfig`] default
    /// (30s). Lower it for deployments that want an unreachable preferred
    /// peer abandoned faster on the *first* fetch — e.g. dual-C2 failover,
    /// where dropping this to a few seconds cuts the one-time learning
    /// cost before the peer-health index demotes the dead peer. A live
    /// transfer never trips it: the #122 watchdog resets on progress, so
    /// only genuinely stuck attempts rotate to the next peer.
    ///
    /// [`IrohConfig`]: crate::config::IrohConfig
    pub download_stall_timeout: Option<Duration>,
    /// Optional TTL configuration for short-lived document eviction
    /// (beacon soft-delete, node-state expiry, etc.).
    ///
    /// `None` (default) skips the [`TtlManager`](crate::storage::TtlManager)
    /// entirely — useful for test consumers and environments that manage
    /// TTL outside the backend. `Some(cfg)` starts a background cleanup
    /// task that reaps documents past their collection TTL.
    ///
    /// The correctness invariant from ADR-016: `tombstone_ttl_hours` must
    /// be ≥ the longest expected peer offline / partition window. The default
    /// `TtlConfig::new()` value of 168 h (7 days) is the safe floor for DDIL
    /// deployments; lower values risk tombstone-before-reconnect resurrection
    /// (the failure mode fixed in peat-mesh#217/#218). A warn-level log is
    /// emitted when `tombstone_ttl_hours` is set below 24 h.
    ///
    /// peat-node#136.
    pub ttl_config: Option<crate::storage::TtlConfig>,
    /// Optional garbage-collection configuration for tombstone reaping and
    /// implicit-TTL eviction. `None` (default) skips the
    /// [`GarbageCollector`](crate::qos::GarbageCollector) — useful for test
    /// consumers and environments running GC via their own scheduler.
    ///
    /// `Some(cfg)` starts a periodic GC task alongside the TtlManager.
    ///
    /// peat-node#136.
    pub gc_config: Option<crate::qos::GcConfig>,
    /// Deterministic Iroh identity seed (32 raw bytes).
    ///
    /// When `Some`, the Iroh endpoint is bound with a fixed `SecretKey` derived
    /// from these bytes (`iroh::SecretKey::from_bytes`), giving this node a
    /// stable, reproducible `EndpointId`. Required for Kubernetes-based peer
    /// discovery (peat-node#63): K8s discovery emits pod names, not iroh
    /// endpoint IDs, so every pod must derive its keypair deterministically
    /// from `HKDF-SHA256(formation_secret, "iroh:" + pod_name)` so peers can
    /// compute each other's `EndpointId` without an out-of-band lookup.
    ///
    /// When `None` (default), iroh generates a random keypair per process
    /// restart, which is appropriate for static-peer and mDNS deployments.
    pub iroh_secret_key: Option<[u8; 32]>,
}

impl Default for AutomergeBackendConfig {
    fn default() -> Self {
        Self {
            data_dir: PathBuf::new(),
            formation_id: String::new(),
            base64_shared_key: String::new(),
            iroh_bind_addr: None,
            cipher: None,
            download_stall_timeout: None,
            ttl_config: None,
            iroh_secret_key: None,
            gc_config: None,
        }
    }
}

/// Automerge + Iroh-backed sync backend.
///
/// Implements all four traits in the `peat_mesh::sync::traits` stack
/// (`DocumentStore`, `PeerDiscovery`, `SyncEngine`, `DataSyncBackend`)
/// by composing existing peat-mesh storage primitives. See module-level
/// docs for the rationale.
///
/// Clone is shallow — every internal field is an `Arc` — so handing
/// `Arc::new(self.clone())` to the `DataSyncBackend` accessors gives
/// each sub-trait view of the same backing state.
#[derive(Clone)]
pub struct AutomergeBackend {
    /// CRDT document store (redb-backed unless `with_in_memory_store` is used).
    store: Arc<AutomergeStore>,
    /// QUIC sync transport (formation-key authenticated).
    transport: Arc<MeshSyncTransport>,
    /// Sync coordinator (per-document sync state, retry, circuit breaker, flow control).
    coordinator: Arc<AutomergeSyncCoordinator>,
    /// Persistent-stream channel manager (peat-mesh#175 / ADR-063).
    ///
    /// Owned here so the strong reference outlives the construction
    /// scope. `AutomergeSyncCoordinator::set_channel_manager` downgrades
    /// to a `Weak` internally (the coordinator → manager → coordinator
    /// would otherwise cycle); without this field, the manager would be
    /// dropped at the end of `new()` and the coordinator's `Weak` would
    /// dangle, silently routing every consumer through the legacy
    /// per-message-stream fallback. peat-mesh#178 QA BLOCKER.
    ///
    /// `#[allow(dead_code)]`: the field is "never read" by definition —
    /// its purpose is to be alive, not to be accessed. The coordinator
    /// reaches the manager through the `Weak` it owns.
    #[allow(dead_code)]
    channel_manager: Arc<SyncChannelManager>,
    /// Networked blob store (owns the iroh::Router + endpoint shutdown).
    /// Kept alive for the lifetime of the backend so the router stays accepting
    /// incoming sync ALPN connections.
    blob_store: Arc<NetworkedIrohBlobStore>,
    /// `true` between successful `start_sync` and `stop_sync`.
    sync_active: Arc<AtomicBool>,
    /// `true` between successful `initialize` and `shutdown`.
    initialized: Arc<AtomicBool>,
    /// Collections explicitly subscribed via `SyncEngine::subscribe`.
    /// The on-change auto-sync task only forwards keys whose collection
    /// prefix is in this set (or, if empty, all collections — preserving
    /// the all-collections-sync behavior of the prior 7-step bootstrap).
    subscriptions: Arc<RwLock<HashSet<String>>>,
    /// `on_peer_event` registrations.
    #[allow(clippy::type_complexity)]
    peer_callbacks: Arc<StdMutex<Vec<Box<dyn Fn(PeerEvent) + Send + Sync>>>>,
    /// Backend event channel — `DocumentStore::observe` and the
    /// observer-forwarder task both publish through this to all
    /// live observers. `remove` also publishes here (Removed events
    /// don't flow through the forwarder because
    /// `AutomergeStore::delete` does not broadcast).
    events: broadcast::Sender<EventEnvelope>,
    /// Per-key pending-emission map for the observer-forwarder.
    ///
    /// `upsert_with_origin` writes the pre-converted `Document` and
    /// the caller's origin here, keyed by the storage key, *before*
    /// calling `AutomergeStore::put`. The forwarder, when it sees
    /// that key on the broadcast channel, **consumes** the entry
    /// and emits the pre-baked `Updated` event verbatim.
    ///
    /// If no entry exists for a key, the forwarder treats the event
    /// as sync-ingested and fetches the latest state from the store
    /// to build an `Updated{origin: None}` per ADR-059.
    ///
    /// Two properties this guarantees:
    /// 1. **Exactly one `Updated` event per local write** — pre-
    ///    round-2 the adapter direct-emitted from `upsert_with_origin`
    ///    AND the forwarder replayed the same put, producing
    ///    duplicate events once sync was active.
    /// 2. **Local upserts never lose their Updated event under a
    ///    follow-on remove race** — if a `remove` deletes the key
    ///    before the forwarder fetches, the pre-baked Document in
    ///    this map keeps the event payload alive. Pre-round-2 the
    ///    forwarder did `store.get` after the broadcast and would
    ///    silently drop the Updated event if the key was already
    ///    gone.
    pending_emissions: Arc<StdMutex<HashMap<String, PendingEmission>>>,
    /// Handle to the on-change auto-sync task (spawned by `start_sync`).
    sync_task: Arc<StdMutex<Option<JoinHandle<()>>>>,
    /// Handle to the periodic FlowController cleanup task (spawned by
    /// `start_sync`). Bounds the rate-limiter map under doc churn —
    /// post-#115 the map is `O(peers × distinct_doc_keys_ever_seen)`
    /// and without periodic eviction grows for the process lifetime.
    flow_cleanup_task: Arc<StdMutex<Option<JoinHandle<()>>>>,
    /// Handle to the QoS-priority fanout worker (spawned by `start_sync`,
    /// peat-mesh#247 / ADR-0013). Drains the coordinator's fanout queue
    /// highest-priority-first; the change loop enqueues onto it.
    fanout_task: Arc<StdMutex<Option<JoinHandle<()>>>>,
    /// Handle to the always-on observer-forwarder task (spawned in
    /// `with_iroh`). Runs for the lifetime of the backend so
    /// `observe()` sees events regardless of sync state — this is
    /// the trait contract `observe` carries.
    #[allow(dead_code)]
    observer_task: Arc<JoinHandle<()>>,
    /// Node-local Lamport clock for consumer-authored operations
    /// (peat-mesh#192). Monotonically advances on every call to
    /// [`Self::next_lamport`]; can be jumped forward to track an
    /// observed remote clock via [`Self::observe_lamport`].
    ///
    /// **Scope of guarantee on landing.** This field provides node-
    /// local strict monotonicity — every `next_lamport` return on this
    /// node is strictly greater than every previous one, including
    /// under concurrent callers. Full Lamport partial-order semantics
    /// across nodes require the receive-side rule
    /// `L_local := max(L_local, L_remote) + 1` to fire on every
    /// inbound Lamport-bearing message; that wire-up is **not** in
    /// this PR. `observe_lamport` exists on the public surface but
    /// is not yet called from the sync-receive / tombstone-ingest
    /// path. Until that wires up, cross-node ordering of values
    /// returned by `next_lamport` reflects the `SystemTime`-nanos
    /// seed plus local advances only — i.e. roughly the same wall-
    /// clock ordering the prior `peat-cli` proxy gave, just at the
    /// right surface and with strict per-node uniqueness.
    ///
    /// **Cross-restart persistence.** Disk-backed stores persist a
    /// high-water mark in the `metadata` redb table under the
    /// `lamport_highwater` key (peat-mesh#195). On construction the
    /// seed is `max(persisted_highwater, SystemTime::now() nanos, 1)`
    /// — so even if the system clock has regressed (NTP slew, manual
    /// time-set, battery-less RTC boot-at-epoch), the resumed clock
    /// stays above any value the prior run could have issued. A
    /// background flush task ([`Self::lamport_flush_task`]) writes
    /// the current atomic to redb at a fixed cadence; explicit
    /// callers (test rigs, clean-shutdown hooks) can force a flush
    /// via [`Self::flush_lamport_highwater`]. Worst-case window of
    /// loss on an unclean shutdown is one flush interval.
    ///
    /// In-memory stores (`AutomergeBackend::with_iroh` is currently
    /// always disk-backed, but `AutomergeStore::in_memory()` is
    /// exposed at the storage layer) cannot persist — the seed is
    /// wall-clock only, and the flush task is still spawned but
    /// no-ops at the store layer.
    ///
    /// Motivating consumer: `peat-cli delete` builds
    /// `peat_mesh::qos::Tombstone::new(doc_id, collection, author,
    /// lamport: u64)`. The current CLI workaround uses
    /// `SystemTime::now()` nanos directly; this clock replaces that
    /// at the right surface — strict node-local monotonicity plus
    /// cross-restart persistence is strictly stronger than the
    /// wall-clock proxy. Cross-node Lamport semantics require the
    /// receive-side wire-up tracked in peat-mesh#196.
    lamport: Arc<AtomicU64>,
    /// Handle to the background Lamport-highwater flush task
    /// (peat-mesh#195). Reads `self.lamport` at a fixed cadence and
    /// writes the value through `AutomergeStore::write_lamport_highwater`
    /// when it has changed since the previous flush. Exits when the
    /// store is dropped (via a `Weak<AutomergeStore>` upgrade check —
    /// same pattern as `observer_task`).
    ///
    /// `#[allow(dead_code)]`: the field's purpose is to keep the task
    /// alive for the lifetime of the backend, not to be read.
    #[allow(dead_code)]
    lamport_flush_task: Arc<JoinHandle<()>>,
    /// Optional TtlManager started from [`AutomergeBackendConfig::ttl_config`].
    /// Held here so the manager (and its background cleanup task) outlives
    /// the construction scope. `None` when no `ttl_config` was supplied.
    ///
    /// `#[allow(dead_code)]`: the field's purpose is to keep the manager
    /// alive (and its background task running) for the lifetime of the
    /// backend — accessing it is not the goal.
    #[allow(dead_code)]
    ttl_manager: Option<Arc<crate::storage::TtlManager>>,
    /// Optional GC task handle started from [`AutomergeBackendConfig::gc_config`].
    /// Held in an `Arc<JoinHandle>` (matching the `lamport_flush_task` pattern)
    /// so the handle can be shared across `Clone` while keeping the task alive.
    /// `None` when no `gc_config` was supplied.
    ///
    /// `#[allow(dead_code)]`: kept alive, not read.
    #[allow(dead_code)]
    gc_task: Option<Arc<JoinHandle<()>>>,
    /// Idempotency guard for [`Self::shutdown_and_release`] (peat-mesh#197).
    ///
    /// Flipped atomically from `false` → `true` by the first
    /// `shutdown_and_release` call. Subsequent calls observe `true`
    /// and short-circuit with `Ok(())`, avoiding a second
    /// `Router::shutdown` against an already-closed Router (which
    /// would surface a misleading iroh error). Defensive against
    /// signal-handling races and double-shutdown bugs in
    /// shutdown-hook code.
    shutdown_completed: Arc<AtomicBool>,
}

/// Internal event envelope used by the broadcast bus.
///
/// `ChangeEvent` is `Clone` but `broadcast::Sender` requires `Send +
/// Sync`. Wrapping in a thin envelope keeps the public `ChangeEvent`
/// type out of the broadcast plumbing and gives future room to attach
/// per-event metadata that doesn't belong on `ChangeEvent` itself.
#[derive(Debug, Clone)]
struct EventEnvelope {
    collection: String,
    event: ChangeEvent,
}

/// Pre-baked emission staged by `upsert_with_origin` for the
/// observer-forwarder to deliver. See `pending_emissions` docs.
#[derive(Debug, Clone)]
struct PendingEmission {
    collection: String,
    document: Document,
    origin: Option<String>,
}

impl AutomergeBackend {
    /// Build the full Automerge+Iroh sync backend.
    ///
    /// Mirrors the 7-step bootstrap in `peat-node::SidecarNode::new`:
    ///   1. open `AutomergeStore` at `data_dir/automerge`
    ///   2. bind an iroh `Endpoint`
    ///   3. derive `FormationKey` from `(formation_id, base64_shared_key)`
    ///   4. wrap the endpoint in `MeshSyncTransport`
    ///   5. create `AutomergeSyncCoordinator`
    ///   6. create `SyncProtocolHandler` (HMAC-auth incoming sync streams)
    ///   7. start `NetworkedIrohBlobStore` with the sync handler registered
    ///      on `CAP_AUTOMERGE_ALPN`
    ///
    /// The returned backend is **not yet running sync** — call
    /// `initialize(...)` then `sync_engine().start_sync()`. The
    /// `DataSyncBackend::initialize` step is preserved for parity
    /// with `InMemoryBackend`'s lifecycle even though most setup
    /// already happened in this constructor.
    ///
    /// Wire-compat: any directory layout the prior bootstrap wrote
    /// to is read directly. `AutomergeStore::open` creates the redb
    /// file if absent or opens it if present.
    pub async fn with_iroh(cfg: AutomergeBackendConfig) -> Result<Arc<Self>> {
        let AutomergeBackendConfig {
            data_dir,
            formation_id,
            base64_shared_key,
            iroh_bind_addr,
            cipher,
            download_stall_timeout,
            ttl_config,
            gc_config,
            iroh_secret_key,
        } = cfg;
        let automerge_dir = data_dir.join("automerge");
        let iroh_dir = data_dir.join("iroh");
        tokio::fs::create_dir_all(&automerge_dir)
            .await
            .context("create automerge data dir")?;
        tokio::fs::create_dir_all(&iroh_dir)
            .await
            .context("create iroh data dir")?;

        // 1. Open the CRDT store (peat-mesh#124: pass at-rest cipher through).
        let store = Arc::new(
            AutomergeStore::open_with_cipher(&automerge_dir, cipher)
                .context("open AutomergeStore at data_dir")?,
        );

        // 2. Bind iroh endpoint. `N0DisableRelay` keeps the endpoint
        //    independent of n0's public relay pool; consumers that
        //    need NAT traversal layer their own relay config via
        //    `add_peer` / `connect_to_peer` static-peer flow.
        //
        // If the caller pinned `iroh_bind_addr`, clear the builder's
        // default IPv4-`0.0.0.0`/IPv6-`[::]` socket pair and replace
        // it with the requested address. Per iroh's
        // `Builder::bind_addr` docs, `bind_addr` is additive on top
        // of the default sockets — `clear_ip_transports` first is
        // the documented way to make it a true replacement.
        let memory_lookup = iroh::address_lookup::memory::MemoryLookup::new();
        // peat#923 / ADR-060 §5: routes through the shared
        // `crate::security::tls_provider::iroh_quic_provider()` helper so
        // FIPS-approved aws-lc-rs is wired by construction across every
        // iroh endpoint site in peat-mesh. See that module's doc for the
        // full rationale.
        // peat-mesh#226: apply the same tactical QUIC keepalive + idle-timeout
        // (1 s keepalive / 5 s idle, Issue #315) that every `IrohTransport`
        // endpoint already uses. This endpoint backs the persistent CRDT sync
        // channels (and the blob store); without transport-level liveness a
        // half-open connection was detected only by the 30 s app-level
        // `SyncChannel::RECV_TIMEOUT`, so a local write sat undelivered — and
        // convergence to *reachable* peers stalled — for multiples of 30 s
        // whenever a channel went half-open (dead peer or a transient path
        // disruption at a topology failover). The 1 s keepalive also keeps a
        // healthy-but-idle channel verified, so it is never mistaken for stale.
        let mut builder = iroh::Endpoint::builder(iroh::endpoint::presets::N0DisableRelay)
            .crypto_provider(crate::security::tls_provider::iroh_quic_provider())
            .transport_config(crate::network::iroh_transport::create_tactical_transport_config())
            .address_lookup(memory_lookup.clone());
        if let Some(addr) = iroh_bind_addr {
            builder = builder
                .clear_ip_transports()
                .bind_addr(addr)
                .map_err(|e| anyhow!("invalid iroh bind address {addr}: {e}"))?;
        }
        // peat-node#63: deterministic keypair for Kubernetes peer discovery.
        // HKDF-SHA256(formation_secret, "iroh:" + pod_name) → 32 bytes → SecretKey.
        if let Some(key_bytes) = iroh_secret_key {
            builder = builder.secret_key(iroh::SecretKey::from_bytes(&key_bytes));
        }
        let endpoint = builder.bind().await.context("bind iroh endpoint")?;

        // 3. Derive the formation key for peer authentication.
        let formation_key = FormationKey::from_base64(&formation_id, &base64_shared_key)
            .map_err(|e| anyhow!("invalid formation key: {e}"))?;

        // 4. Sync transport over the endpoint.
        let transport = Arc::new(MeshSyncTransport::new(
            endpoint.clone(),
            formation_key.clone(),
        ));

        // 5. Sync coordinator atop the transport + store.
        let coordinator = Arc::new(AutomergeSyncCoordinator::new(
            Arc::clone(&store),
            transport.clone(),
        ));

        // 5a. Persistent sync channels (peat-mesh#175 / ADR-063).
        //
        // Pre-#175 every consumer of `AutomergeBackend::new` (peat-protocol,
        // peat-node, peat-sim) ran the legacy per-message-stream fallback
        // in `send_sync_message_for_doc` because `set_channel_manager` was
        // only called from the standalone `peat-mesh-node` binary. That
        // capped sync throughput at ~5 rounds/sec/peer on a 200 ms RTT
        // link (the symptom in peat-mesh#175). With the channel manager
        // wired here, every consumer gets the persistent multiplexed
        // bi-stream per peer, with the legacy path retained as the
        // reconnect-window fallback (cf. ADR-063 §D4).
        //
        // `set_channel_manager` downgrades to `Weak` internally (to avoid
        // the coordinator-channel_manager cycle since SyncChannelManager
        // holds an Arc<coordinator>). The strong reference is preserved
        // as a field on AutomergeBackend so the manager outlives this
        // construction scope. peat-mesh#178 QA BLOCKER.
        let channel_manager = Arc::new(SyncChannelManager::new(
            transport.clone() as Arc<dyn SyncTransport>,
            coordinator.clone(),
        ));
        coordinator.set_channel_manager(Arc::clone(&channel_manager));

        // 5b. Lifecycle services: TtlManager + GarbageCollector (peat-node#136).
        //
        // Both are optional and default None. Consumers (e.g. peat-node) that
        // set `ttl_config` / `gc_config` in `AutomergeBackendConfig` get the
        // background tasks started automatically here; consumers that don't
        // (tests, peat-mesh-node which starts them independently) see no change.
        let ttl_manager: Option<Arc<crate::storage::TtlManager>> = ttl_config.map(|cfg| {
            if cfg.tombstone_ttl_hours < 24 {
                tracing::warn!(
                    tombstone_ttl_hours = cfg.tombstone_ttl_hours,
                    "tombstone_ttl_hours is below 24 h — peers offline longer than this \
                         may resurrect deleted documents on reconnect (ADR-016 invariant)"
                );
            }
            let mgr = Arc::new(crate::storage::TtlManager::new(Arc::clone(&store), cfg));
            mgr.start_background_cleanup();
            coordinator.set_ttl_manager(Arc::clone(&mgr));
            mgr
        });
        let gc_task: Option<Arc<JoinHandle<()>>> = gc_config.map(|cfg| {
            let gc = Arc::new(crate::qos::GarbageCollector::new(Arc::clone(&store), cfg));
            Arc::new(crate::qos::start_periodic_gc(gc))
        });

        // 6. Protocol handlers for incoming ALPN connections.
        //
        // The sync handler must register inbound-authenticated peers into the
        // blob store's `known_peers` (peat-node#170) — but it's about to be
        // moved into the store's iroh Router (step 7), before the store exists.
        // Hand it a shared `OnceLock` slot now and fill it with a `Weak` ref once
        // the store is built, breaking the construction cycle.
        let peer_registry: InboundPeerRegistry = Arc::new(OnceLock::new());
        let handler = SyncProtocolHandler::new(
            transport.clone(),
            coordinator.clone(),
            formation_key.clone(),
        )
        .with_inbound_peer_registry(peer_registry.clone());

        // Blob provider gossip handler (peat-node#170). It needs a reference to
        // the blob store to apply announces, but the store doesn't exist yet and
        // is about to own this handler via its Router — so hand it a slot now
        // and fill it with a Weak ref once the store is built (cycle-break).
        let blob_announce_slot: blob_announce::BlobStoreSlot = Arc::new(OnceLock::new());
        let announce_handler = blob_announce::BlobAnnounceHandler::new(blob_announce_slot.clone());

        // 7. Blob store owns the iroh::Router; register the sync + blob-announce
        //    ALPN handlers alongside iroh-blobs' default protocol.
        // Default timeouts except the stall threshold, which the caller may
        // override (peat-mesh#137 follow-up — lets a consumer like peat-node
        // shrink the first-fetch dead-peer wait in dual-C2 failover).
        let iroh_defaults = crate::config::IrohConfig::default();
        let blob_store = NetworkedIrohBlobStore::from_endpoint_with_protocols_with_timeouts(
            iroh_dir,
            endpoint,
            memory_lookup,
            vec![
                (CAP_AUTOMERGE_ALPN, Box::new(handler) as Box<_>),
                (
                    blob_announce::CAP_BLOB_ANNOUNCE_ALPN,
                    Box::new(announce_handler) as Box<_>,
                ),
            ],
            iroh_defaults.shutdown_timeout,
            iroh_defaults.connect_timeout,
            download_stall_timeout.unwrap_or(iroh_defaults.download_stall_timeout),
            iroh_defaults.peer_health_cooldown,
        )
        .await
        .context("start NetworkedIrohBlobStore")?;

        // Now that the store exists, hand both handlers a Weak ref to it so the
        // sync handler can register inbound peers into `known_peers` and the
        // announce handler can apply provider gossip (peat-node#170). `set` only
        // fails if already populated — impossible here.
        let _ = peer_registry.set(Arc::downgrade(&blob_store));
        let _ = blob_announce_slot.set(Arc::downgrade(&blob_store));

        // Bus capacity 1024: matches the order-of-magnitude of
        // `AutomergeStore::observer_tx` (8192 from issue #346) but
        // smaller because observers re-fetch the doc from the store
        // anyway, so a lag here just means re-reading the latest
        // snapshot — no data loss.
        let (events, _) = broadcast::channel(1024);

        let pending_emissions: Arc<StdMutex<HashMap<String, PendingEmission>>> =
            Arc::new(StdMutex::new(HashMap::new()));

        // Spawn the always-on observer-forwarder. This bridges
        // `AutomergeStore::subscribe_to_observer_changes` (which
        // fires for ALL puts — both local upserts and remote-sync
        // merges) onto our `events` broadcast bus.
        //
        // Two paths converge here:
        // - **Local upserts** stage a pre-baked `PendingEmission`
        //   (Document + origin) via `pending_emissions` BEFORE
        //   `store.put` triggers the broadcast. The forwarder
        //   consumes the entry and emits the pre-baked event
        //   verbatim — robust against a follow-on `remove` racing
        //   the forwarder's read of the store.
        // - **Sync-ingested puts** have no `pending_emissions` entry.
        //   The forwarder fetches the latest state from the store,
        //   converts, and emits `Updated{origin: None}` per ADR-059.
        //
        // This single-path design is what makes `observe()` emit
        // exactly one event per write, regardless of whether sync
        // is currently active.
        // CRITICAL: hold ONLY a `Weak<AutomergeStore>` across the
        // `rx.recv().await` parking point. The strong upgrade
        // happens AFTER we have a key in hand, not before.
        //
        // Otherwise we form an ownership cycle (task → strong
        // Arc<store> → observer_tx Sender → task parked on
        // rx.recv() waiting for the Sender to close → Sender can
        // never close because the task holds the Arc) and every
        // backend drop leaks the task + store for the process
        // lifetime.
        //
        // Round-3 of QA #106 caught this: an earlier shape upgraded
        // `Weak → strong Arc` at the top of each loop iteration and
        // held that strong Arc across the await. Functionally
        // identical to capturing `Arc::clone(&store)` directly —
        // the strong handle was preserved in the future's state
        // machine across the park, so the Sender never dropped.
        // peat-mesh#106 QA round-4 moves the upgrade *after* the
        // recv resolves so the strong Arc is only held during the
        // synchronous event-processing window.
        let observer_handle = {
            let store_weak = Arc::downgrade(&store);
            let events = events.clone();
            let pending = Arc::clone(&pending_emissions);
            let mut rx = store.subscribe_to_observer_changes();
            tokio::spawn(async move {
                loop {
                    // Park holding ONLY the Weak (and the Receiver,
                    // which doesn't keep the Sender alive). When
                    // the last strong Arc<store> outside this task
                    // drops, the store drops, the Sender drops,
                    // `rx.recv()` returns Closed, the task exits.
                    let key = match rx.recv().await {
                        Ok(k) => k,
                        Err(broadcast::error::RecvError::Lagged(_)) => continue,
                        Err(broadcast::error::RecvError::Closed) => return,
                    };
                    // Now we have a key to process — upgrade the
                    // Weak. If the backend has been dropped in the
                    // narrow window between the recv returning Ok
                    // and us getting CPU here, exit cleanly (the
                    // event would have nowhere to go anyway).
                    let Some(store) = store_weak.upgrade() else {
                        return;
                    };

                    // Prefer the pre-baked emission from a local
                    // upsert. If it's there, deliver it verbatim
                    // and skip the store fetch (which could race
                    // a concurrent remove and drop the event).
                    if let Some(emission) = pending
                        .lock()
                        .unwrap_or_else(|e| e.into_inner())
                        .remove(&key)
                    {
                        let _ = events.send(EventEnvelope {
                            collection: emission.collection.clone(),
                            event: ChangeEvent::Updated {
                                collection: emission.collection,
                                document: emission.document,
                                origin: emission.origin,
                            },
                        });
                        continue;
                    }
                    // No pending emission — sync-ingested path.
                    // Reconstruct the Document from the store.
                    let Some((collection, doc_id)) = key.split_once(':') else {
                        continue;
                    };
                    let am = match store.get(&key) {
                        Ok(Some(am)) => am,
                        Ok(None) => continue,
                        Err(_) => continue,
                    };
                    let json = crate::storage::json_convert::automerge_to_json(&am);
                    let doc = AutomergeBackend::json_to_document(json, doc_id);
                    let _ = events.send(EventEnvelope {
                        collection: collection.to_string(),
                        event: ChangeEvent::Updated {
                            collection: collection.to_string(),
                            document: doc,
                            origin: None,
                        },
                    });
                }
            })
        };

        // peat-mesh#192 + peat-mesh#195: seed the node-local Lamport
        // clock from `max(persisted_highwater, SystemTime::now() nanos, 1)`.
        // The wall-clock floor protects the seed when the persisted
        // value is `None` (fresh data dir, in-memory store) and when
        // it's older than the current wall clock (normal case after a
        // graceful shutdown). The persisted-value path strictly wins
        // when the system clock has regressed below the prior run's
        // high-water (the tactical-hardware fragility case the rc.28
        // QA round-1 [WARNING] flagged).
        //
        // The `1u64` floor protects against the `UNIX_EPOCH` skew
        // case — `next_lamport` must never return 0, since consumers
        // using its return as a Lamport-typed `u64` may treat 0 as
        // "unset".
        let persisted_highwater = store
            .read_lamport_highwater()
            .context("read persisted lamport highwater")?
            .unwrap_or(0);
        let wallclock_nanos = SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .map(|d| d.as_nanos() as u64)
            .unwrap_or(0);
        let lamport_seed = persisted_highwater.max(wallclock_nanos).max(1);
        let lamport = Arc::new(AtomicU64::new(lamport_seed));

        // peat-mesh#196: wire the node-local Lamport clock into the
        // sync coordinator's receive path. Every inbound
        // `Tombstone.lamport` will now flow through `observe_atomic`
        // before the tombstone is applied — locally-authored
        // operations following a receive sort strictly after the
        // observed Lamport. Standalone coordinator consumers (the
        // peat-mesh-node binary, unit tests) don't make this call and
        // the receive-side wire-up no-ops, preserving existing
        // behaviour for non-AutomergeBackend assemblies.
        coordinator.set_lamport_clock(Arc::clone(&lamport));

        // peat-mesh#195: background Lamport-highwater flush. Runs at a
        // fixed cadence; writes the current atomic to redb when it has
        // changed since the previous flush. Worst-case loss on an
        // unclean shutdown is one flush interval's worth of issued
        // values.
        //
        // 1-second cadence chosen as a reasonable bound for the
        // motivating consumer (CLI `delete`, low-frequency): on the
        // realistic workload — handfuls of tombstones per minute,
        // bursty under recovery — the flush task wakes once per
        // second, observes that the atomic hasn't changed, and skips
        // the write. Under sustained activity it flushes every
        // second; the prior-flushed-value short-circuit makes the
        // idle case zero-cost beyond the 1-second sleep.
        //
        // CRITICAL: hold ONLY a `Weak<AutomergeStore>` across the
        // sleep. The strong upgrade happens AFTER the tick fires.
        // Same ownership-cycle prevention as the `observer_task` —
        // a strong `Arc` held across the await would prevent backend
        // drop from closing the task.
        let lamport_flush_handle = {
            let store_weak = Arc::downgrade(&store);
            let lamport = Arc::clone(&lamport);
            // Seed `last_flushed` to the value already on disk so the
            // first non-skipped tick doesn't redundantly re-write the
            // same value it just loaded at startup. The 0 fallback
            // covers in-memory stores and pre-#195 disk files.
            let last_flushed_init = persisted_highwater;
            tokio::spawn(async move {
                let mut interval = tokio::time::interval(Duration::from_secs(1));
                // First tick fires immediately; skip it so we don't
                // flush before any next_lamport has been called.
                interval.tick().await;
                let mut last_flushed: u64 = last_flushed_init;
                loop {
                    interval.tick().await;
                    let Some(store) = store_weak.upgrade() else {
                        // Backend has been dropped.
                        return;
                    };
                    let current = lamport.load(Ordering::Relaxed);
                    if current == last_flushed {
                        continue;
                    }
                    if let Err(e) = store.write_lamport_highwater(current) {
                        // Per peat-mesh#195 constraint: "flush errors
                        // must not block the issued value — log + carry
                        // on." A persistent redb fault will surface in
                        // other paths (put/get); we don't elevate the
                        // Lamport flush to terminate the task.
                        tracing::warn!(
                            error = %e,
                            current,
                            "lamport_highwater flush failed; will retry next tick"
                        );
                        continue;
                    }
                    last_flushed = current;
                }
            })
        };

        Ok(Arc::new(Self {
            store,
            transport,
            coordinator,
            channel_manager,
            blob_store,
            sync_active: Arc::new(AtomicBool::new(false)),
            initialized: Arc::new(AtomicBool::new(false)),
            subscriptions: Arc::new(RwLock::new(HashSet::new())),
            peer_callbacks: Arc::new(StdMutex::new(Vec::new())),
            events,
            pending_emissions,
            sync_task: Arc::new(StdMutex::new(None)),
            flow_cleanup_task: Arc::new(StdMutex::new(None)),
            fanout_task: Arc::new(StdMutex::new(None)),
            observer_task: Arc::new(observer_handle),
            lamport,
            lamport_flush_task: Arc::new(lamport_flush_handle),
            ttl_manager,
            gc_task,
            shutdown_completed: Arc::new(AtomicBool::new(false)),
        }))
    }

    /// Direct accessor for the underlying `AutomergeStore`.
    ///
    /// Provided for advanced consumers (TTL manager wiring, compaction
    /// scheduling, raw-Automerge ops) that need to reach past the
    /// trait surface. Most consumers should use the `DocumentStore`
    /// view via `Node`.
    pub fn store(&self) -> &Arc<AutomergeStore> {
        &self.store
    }

    /// Direct accessor for the underlying `MeshSyncTransport`.
    pub fn transport(&self) -> &Arc<MeshSyncTransport> {
        &self.transport
    }

    /// Direct accessor for the underlying `AutomergeSyncCoordinator`.
    pub fn coordinator(&self) -> &Arc<AutomergeSyncCoordinator> {
        &self.coordinator
    }

    /// Direct accessor for the underlying `NetworkedIrohBlobStore`.
    pub fn blob_store(&self) -> &Arc<NetworkedIrohBlobStore> {
        &self.blob_store
    }

    /// Read-and-advance the node-local Lamport clock, returning a value
    /// strictly greater than any previous return on this node
    /// (peat-mesh#192).
    ///
    /// Intended for consumers — `peat-cli` and any other client that
    /// authors operations on this node — to stamp an outgoing operation.
    ///
    /// **What this guarantees on landing:** strict monotonicity *on this
    /// node*. Every call returns a value strictly greater than every
    /// prior call on this node, including under concurrent callers
    /// (the monotonicity comes from `fetch_add` being a read-modify-write
    /// atomic, not from any specific memory ordering parameter).
    ///
    /// **What this does NOT guarantee yet:** full Lamport partial-order
    /// semantics across nodes. The receive-side rule
    /// `L_local := max(L_local, L_remote) + 1` is exposed via
    /// [`Self::observe_lamport`] but is *not yet wired into the sync-
    /// receive path*. Until that wire-up lands, cross-node ordering of
    /// values returned by `next_lamport` reflects only the seed
    /// (`SystemTime::now()` nanos at construction) plus local advances
    /// — i.e. roughly the same wall-clock ordering the current
    /// `peat-cli delete` workaround already gives, just at the right
    /// surface. Consumers that need true causal ordering across nodes
    /// must thread observed remote Lamports through
    /// [`Self::observe_lamport`] themselves until the receive-side
    /// wire-up lands.
    pub fn next_lamport(&self) -> u64 {
        // `fetch_add` returns the previous value; the caller's value
        // is `previous + 1`. `Relaxed` is sufficient: the only
        // observable property is that every caller sees a unique
        // value, and that comes from `fetch_add` being a single
        // atomic read-modify-write — the ordering parameter governs
        // visibility relationships with *other* memory, not the
        // counter's own monotonicity (clarifying the QA finding on
        // the prior SeqCst-justifies-monotonicity wording).
        //
        // 18-quintillion-call wrap: not a realistic operating concern;
        // if reached, the counter wraps to 0 and subsequent calls
        // would issue lower values until the wall-clock seed of the
        // next process restart restores monotonicity. `wrapping_add`
        // spells the wrap explicitly (matching the atomic's own
        // overflow semantics); a plain `+ 1` would panic in debug
        // builds and wrap silently in release — same observable
        // outcome in release, but the debug-only panic clashes with
        // the comment's "owns the wrap" stance.
        self.lamport.fetch_add(1, Ordering::Relaxed).wrapping_add(1)
    }

    /// Read the current Lamport value without advancing the clock
    /// (peat-mesh#192).
    ///
    /// Diagnostic accessor for tests and consumer-side observability.
    /// Do not use this value to stamp an outgoing operation — that's
    /// what [`Self::next_lamport`] is for; using `current_lamport`
    /// instead would let two callers race to issue the same value.
    pub fn current_lamport(&self) -> u64 {
        self.lamport.load(Ordering::Relaxed)
    }

    /// Gracefully shut down the backend, awaiting the iroh Router's
    /// shutdown future before returning (peat-mesh#197).
    ///
    /// Distinct from [`DataSyncBackend::shutdown`], which intentionally
    /// leaves the iroh Router alive so pending blob transfers can
    /// complete after the sync activity stops. This method does the
    /// full teardown sequence:
    ///
    ///   1. Stop sync activity (`stop_sync`).
    ///   2. Flush the Lamport high-water to redb
    ///      ([`Self::flush_lamport_highwater`]).
    ///   3. **Await the iroh Router's shutdown future**
    ///      ([`NetworkedIrohBlobStore::shutdown`]) — the step the
    ///      trait `shutdown` deliberately skips.
    ///
    /// After step 3 awaits to completion, no background iroh task
    /// transitively holds an `Arc` chain back to the `AutomergeStore`
    /// or its underlying redb `Database`. The caller can then `drop`
    /// the backend (or let it fall out of scope), which releases the
    /// OS-level redb file lock synchronously, and immediately reopen
    /// a new `AutomergeBackend::with_iroh` against the same
    /// `data_dir` in the same process without hitting a lock-
    /// contention race.
    ///
    /// **Use cases:** clean-shutdown hooks, hot-restart flows (config
    /// reload, daemon-style rotation, mobile pause/resume), and any
    /// integration test that drops a backend and reopens the same
    /// data dir.
    ///
    /// **Bounded by `NetworkedIrohBlobStore::shutdown_timeout`** (5s
    /// default) so a stuck QUIC stream cannot hang the caller
    /// indefinitely. Errors from the Lamport flush log-and-continue;
    /// errors from the Router shutdown propagate verbatim.
    ///
    /// **Idempotent.** A `shutdown_completed` atomic flag short-
    /// circuits the second and subsequent calls with `Ok(())`,
    /// guarding against signal-handling races and double-shutdown
    /// bugs in caller-side hook code. The first call wins the
    /// `compare_exchange` and runs the full teardown; concurrent
    /// callers see the flag flip and return immediately without
    /// invoking another `Router::shutdown` — regardless of whether
    /// the winning caller's teardown has completed yet, the loser's
    /// CAS-failure short-circuits before reaching steps 1–3.
    pub async fn shutdown_and_release(&self) -> Result<()> {
        // Idempotency guard. `compare_exchange` from false → true
        // ensures exactly one caller runs the teardown; later
        // callers see the existing `true` and short-circuit. The
        // race window for two concurrent first-callers is closed
        // by the atomic — only one can win the CAS.
        if self
            .shutdown_completed
            .compare_exchange(false, true, Ordering::AcqRel, Ordering::Acquire)
            .is_err()
        {
            return Ok(());
        }

        // 1. Stop sync activity (idempotent — `stop_sync` handles
        //    already-stopped state).
        let _ = self.stop_sync().await;
        self.initialized.store(false, Ordering::SeqCst);

        // 2. Flush the Lamport high-water (peat-mesh#195). Errors
        //    log + continue so a transient redb fault doesn't block
        //    the iroh teardown, which is the load-bearing step for
        //    lock release. The periodic flush task running the same
        //    write is also monotonic at the storage layer, so any
        //    in-flight tick can race with this without regressing.
        if let Err(e) = self.flush_lamport_highwater() {
            tracing::warn!(
                error = %e,
                "shutdown_and_release: lamport flush failed; \
                 worst-case loss is up to one periodic flush interval"
            );
        }

        // 3. Tear down the iroh Router. After this awaits, the
        //    QUIC endpoint is closed, the Router's spawned tasks
        //    are joined, and the chain of Arcs reaching back into
        //    AutomergeStore is fully released. A subsequent
        //    drop(backend) is then a synchronous release of the
        //    redb file lock.
        self.blob_store.shutdown().await
    }

    /// Synchronously persist the current Lamport high-water to redb
    /// (peat-mesh#195).
    ///
    /// Forces an immediate flush of the in-memory atomic to the
    /// `metadata` table, bypassing the periodic flush task's cadence.
    /// Two intended call sites:
    ///
    /// 1. **Clean-shutdown hooks** — a process about to exit should
    ///    call this after the last `next_lamport` to ensure the
    ///    on-disk high-water reflects every issued value, not just
    ///    those that landed before the most recent periodic flush.
    /// 2. **Test rigs** — integration tests of the cross-restart
    ///    behaviour need a deterministic flush trigger; awaiting this
    ///    call makes the persistence observable without racing the
    ///    1-second periodic cadence.
    ///
    /// On in-memory stores (`AutomergeStore::in_memory`) this is a
    /// no-op that returns `Ok(())` — there's no disk to write to.
    /// Errors from the redb layer propagate verbatim; callers in
    /// shutdown paths may want to log-and-ignore so a transient
    /// persistence fault doesn't block process exit.
    ///
    /// **Operator note: persisted value composition.** With peat-mesh#196
    /// also landed, the in-memory atomic this method snapshots is
    /// advanced by *two* paths: locally-issued [`Self::next_lamport`]
    /// calls AND inbound `Tombstone.lamport` values flowing through
    /// [`Self::observe_lamport`] from the sync-receive path. The
    /// persisted high-water therefore reflects the larger of (a) any
    /// Lamport this node has issued and (b) any remote Lamport this
    /// node has observed — which is exactly the Lamport partial-order
    /// contract (the local clock floor exceeds every observed event).
    /// Across restart, a far-future remote Lamport that landed before
    /// shutdown permanently elevates this node's seed floor on the
    /// next boot. The `u64::MAX - 1` cap inside the internal
    /// `observe_atomic` helper prevents a hostile or buggy peer
    /// from saturating the atomic to `u64::MAX` and wrapping the
    /// next `next_lamport` to `0`. (Plain text rather than an
    /// intra-doc link — the helper is `pub(crate)`, so an
    /// `[\`crate::…::observe_atomic\`]` link would render broken in
    /// downstream rustdoc.)
    pub fn flush_lamport_highwater(&self) -> Result<()> {
        let current = self.lamport.load(Ordering::Relaxed);
        self.store.write_lamport_highwater(current)
    }

    /// Advance the local Lamport clock to be strictly greater than the
    /// `observed` remote value (peat-mesh#192).
    ///
    /// Idempotent: if the local clock is already past `observed`, this
    /// is a no-op (and does not regress the clock — a strictly-greater
    /// local value is preserved).
    ///
    /// Implements the receive-side half of Lamport's rule:
    /// `L_local := max(L_local, observed) + 1`. As of peat-mesh#196,
    /// this method is **also invoked automatically** from the sync-
    /// receive path on every inbound `Tombstone.lamport` (single
    /// `Tombstone` messages and every entry in a `TombstoneBatch`),
    /// so consumers no longer need to thread observed remote Lamports
    /// through manually. Consumers that decode their own Lamport-
    /// bearing payloads outside the tombstone pipeline can still call
    /// this directly.
    ///
    /// **`u64::MAX` cap (peat-mesh#198 QA hardening).** The standard
    /// `max(L, observed) + 1` rule is enforced via [`observe_atomic`],
    /// which clamps the post-observe value at `u64::MAX - 1`. For
    /// `observed >= u64::MAX - 1`, the local atomic ends at
    /// `u64::MAX - 1` rather than `observed + 1` — preventing a
    /// peer-sent (legitimate or hostile) saturating Lamport from
    /// pushing the clock to `u64::MAX`, which would cause the next
    /// `next_lamport()` to wrap to `0` and break the never-return-zero
    /// invariant. The behaviour is still correct: a subsequent
    /// `next_lamport()` still returns a value strictly greater than
    /// `observed`, except for the single `observed == u64::MAX` corner
    /// where it returns `u64::MAX` (the highest non-wrapping value).
    ///
    /// Concurrency: a `compare_exchange` retry loop handles concurrent
    /// observers and writers; the function returns once the local value
    /// is observably greater than `observed`. The CAS uses `AcqRel` on
    /// success / `Acquire` on failure (sufficient to publish the new
    /// value to subsequent loads on other threads). The loop is bounded
    /// by the number of concurrent advancers — in practice it spins at
    /// most a handful of times.
    pub fn observe_lamport(&self, observed: u64) {
        observe_atomic(&self.lamport, observed);
    }

    /// Build the redb key for a `(collection, doc_id)` pair.
    /// Mirrors the `"collection:doc_id"` convention `AutomergeStore`
    /// uses for prefix scans (see `AutomergeStore::list_collections`).
    fn key_for(collection: &str, doc_id: &DocumentId) -> String {
        format!("{collection}:{doc_id}")
    }

    /// Inverse of `key_for`. Returns `(collection, doc_id)` if the
    /// key has the expected shape, else `None`.
    fn split_key(key: &str) -> Option<(&str, &str)> {
        key.split_once(':')
    }

    /// Convert a `Document` into the JSON shape stored under one
    /// Automerge doc. The fields map IS the document body; `id` and
    /// `updated_at` are reconstructed at read time from the key and
    /// `SystemTime::now()`, respectively.
    fn document_to_json(document: &Document) -> serde_json::Value {
        serde_json::Value::Object(
            document
                .fields
                .iter()
                .map(|(k, v)| (k.clone(), v.clone()))
                .collect(),
        )
    }

    /// Inverse of `document_to_json`. Reconstructs a `Document` from
    /// the Automerge-derived JSON value + the doc_id parsed from the key.
    fn json_to_document(value: serde_json::Value, doc_id: &str) -> Document {
        let fields = match value {
            serde_json::Value::Object(map) => map.into_iter().collect(),
            // Defensive: an Automerge doc whose root isn't an object
            // (shouldn't happen under our writer, but might appear if
            // a peer's `json_convert` round-trip produced something
            // weird). Surface as an empty field map rather than
            // panicking.
            _ => HashMap::new(),
        };
        Document {
            id: Some(doc_id.to_string()),
            fields,
            updated_at: SystemTime::now(),
        }
    }

    /// Publish a `ChangeEvent` to the in-process observer bus.
    /// Called from `upsert_with_origin`, `remove`, and the
    /// store-observer-forwarder task.
    fn publish_event(&self, collection: String, event: ChangeEvent) {
        // broadcast::send returns Err only when there are zero active
        // receivers; that's not an error, just nothing to do.
        let _ = self.events.send(EventEnvelope { collection, event });
    }

    /// Notify all registered `on_peer_event` callbacks.
    fn dispatch_peer_event(&self, event: PeerEvent) {
        let callbacks = self
            .peer_callbacks
            .lock()
            .unwrap_or_else(|e| e.into_inner());
        for cb in callbacks.iter() {
            cb(event.clone());
        }
    }
}

// =============================================================================
// DocumentStore
// =============================================================================

#[async_trait]
impl DocumentStore for AutomergeBackend {
    async fn upsert_with_origin(
        &self,
        collection: &str,
        document: Document,
        origin: Option<String>,
    ) -> Result<DocumentId> {
        let id = document
            .id
            .clone()
            .unwrap_or_else(|| uuid::Uuid::new_v4().to_string());
        let key = Self::key_for(collection, &id);

        // Build the as-stored Document the observer-forwarder will
        // deliver. Restamping `updated_at` matches the on-disk
        // semantic.
        let mut staged = document;
        staged.id = Some(id.clone());
        staged.updated_at = SystemTime::now();

        // Read-modify-write under the per-document stripe lock so a
        // concurrent inbound sync message for the same key can't
        // race-overwrite our local upsert with a stale state.
        //
        // The `pending_emissions.insert` lives INSIDE the lock scope
        // (between lock acquisition and `store.put`) so a concurrent
        // sync-ingested put for the same key can't broadcast in the
        // window between our insert and our put — and steal our
        // pre-baked emission to attribute to its own broadcast. The
        // sync coordinator's put path also takes `lock_doc(key)`, so
        // the two RMW sequences are mutually exclusive; this
        // placement closes the attribution-race window flagged in
        // peat-mesh#106 QA round-3.
        let _guard = self.store.lock_doc(&key);

        self.pending_emissions
            .lock()
            .unwrap_or_else(|e| e.into_inner())
            .insert(
                key.clone(),
                PendingEmission {
                    collection: collection.to_string(),
                    document: staged.clone(),
                    origin,
                },
            );

        let existing = self.store.get(&key).context("read existing doc")?;
        let json = Self::document_to_json(&staged);
        let am = crate::storage::json_convert::json_to_automerge(&json, existing.as_ref())
            .context("json -> automerge")?;
        self.store.put(&key, &am).context("write doc")?;

        Ok(id)
    }

    async fn query(&self, collection: &str, query: &Query) -> Result<Vec<Document>> {
        let prefix = format!("{collection}:");
        let pairs = self
            .store
            .scan_prefix(&prefix)
            .context("scan_prefix for query")?;

        let effective_query = query.inner_query();
        let mut out = Vec::with_capacity(pairs.len());
        for (key, am) in pairs {
            let doc_id = match Self::split_key(&key) {
                Some((_, id)) => id.to_string(),
                None => continue,
            };
            let json = crate::storage::json_convert::automerge_to_json(&am);
            let doc = Self::json_to_document(json, &doc_id);
            if !query.matches_deletion_state(&doc) {
                continue;
            }
            if !evaluate_query(&doc, effective_query) {
                continue;
            }
            out.push(doc);
        }
        Ok(out)
    }

    // Override the `DocumentStore::get` trait-default. The default
    // routes through `query(Query::Eq{id})`, which in `Self::query`
    // above does a `scan_prefix(collection_prefix)` + Automerge
    // deserialize of every entry in the collection — O(N) per
    // single-doc lookup, O(N²) when looped over a set of IDs. Surfaced
    // by peat-mesh#184 sizing the in-CI UAT's `count_delivered`
    // (~17 min on the 25 Hz case before the test switched to a single
    // `Query::All` + Rust set-intersection). This override collapses
    // single-doc get to one redb point lookup + one Automerge parse,
    // using the same `key_for(collection, id)` shape the upsert path
    // already builds. Deletion semantics intentionally match the
    // trait-default path: soft-deleted docs return `None` (callers
    // that want them should `query(Query::IncludeDeleted(...))`
    // directly). `get` has no `Query` parameter — `IncludeDeleted`
    // callers continue to route through `query()`; `get` is the
    // non-deleted-only fast path.
    async fn get(&self, collection: &str, doc_id: &DocumentId) -> Result<Option<Document>> {
        let key = Self::key_for(collection, doc_id);
        let Some(am) = self.store.get(&key).context("read doc for get")? else {
            return Ok(None);
        };
        let json = crate::storage::json_convert::automerge_to_json(&am);
        let doc = Self::json_to_document(json, doc_id);
        // `Query::All`'s `matches_deletion_state` falls into the
        // non-IncludeDeleted arm (`_ => !is_deleted`), matching the
        // trait-default `get`'s `Query::Eq{id}` filtering.
        if !Query::All.matches_deletion_state(&doc) {
            return Ok(None);
        }
        Ok(Some(doc))
    }

    async fn remove(&self, collection: &str, doc_id: &DocumentId) -> Result<()> {
        let key = Self::key_for(collection, doc_id);
        // Stripe-lock to keep the existence-check + delete atomic with
        // any concurrent upsert under the same key.
        let existed = {
            let _guard = self.store.lock_doc(&key);
            let existed = self.store.get(&key).context("read for remove")?.is_some();
            if existed {
                self.store.delete(&key).context("delete doc")?;
            }
            existed
        };
        if existed {
            self.publish_event(
                collection.to_string(),
                ChangeEvent::Removed {
                    collection: collection.to_string(),
                    doc_id: doc_id.clone(),
                    origin: None,
                },
            );
        }
        Ok(())
    }

    fn observe(&self, collection: &str, query: &Query) -> Result<ChangeStream> {
        // Wire the consumer's stream directly to a filtered view of
        // our broadcast bus. We don't emit an initial snapshot here
        // because `observe` is sync; callers needing a snapshot
        // should `query()` first and then `observe()` for live deltas
        // (matches the `InMemoryBackend::observe` contract).
        //
        // **Event delivery is single-emission, regardless of sync
        // state.** The observer-forwarder task is spawned in
        // `with_iroh` and runs for the lifetime of the backend, so
        // both local upserts (with the caller's origin) and remote
        // sync merges (with `origin = None` per ADR-059) flow
        // through this stream at exactly one event per write.
        // peat-mesh#106 QA round-2 changed this from the previous
        // dual-emission shape — the prior `upsert_with_origin`
        // direct-emit + forwarder-replay produced two events per
        // local write once sync was active.
        let (tx, rx) = mpsc::unbounded_channel();
        let mut bus = self.events.subscribe();
        let want_coll = collection.to_string();
        let query = query.clone();
        tokio::spawn(async move {
            loop {
                let env = match bus.recv().await {
                    Ok(e) => e,
                    Err(broadcast::error::RecvError::Lagged(_)) => continue,
                    Err(broadcast::error::RecvError::Closed) => return,
                };
                if env.collection != want_coll {
                    continue;
                }
                // For Updated/Removed, the trait contract is that the
                // event reaches the observer if the document matches
                // the observer's query. Removed events don't carry a
                // Document so we can't evaluate the query against
                // them — pass them through unconditionally (this
                // matches InMemoryBackend's behavior).
                match &env.event {
                    ChangeEvent::Updated { document, .. } => {
                        if !query.matches_deletion_state(document) {
                            continue;
                        }
                        if !evaluate_query(document, query.inner_query()) {
                            continue;
                        }
                    }
                    ChangeEvent::Removed { .. } | ChangeEvent::Initial { .. } => {}
                }
                if tx.send(env.event).is_err() {
                    return; // consumer dropped the stream
                }
            }
        });
        Ok(ChangeStream { receiver: rx })
    }

    async fn get_tombstones(&self, collection: &str) -> Result<Vec<crate::qos::Tombstone>> {
        self.store
            .get_tombstones_for_collection(collection)
            .context("get_tombstones_for_collection")
    }

    async fn apply_tombstone(&self, tombstone: &crate::qos::Tombstone) -> Result<()> {
        self.store
            .put_tombstone(tombstone)
            .context("put_tombstone")?;
        // Tombstone-driven remove of the live document, mirroring the
        // trait default but going through `remove` so observers see
        // a `Removed` event with the correct collection routing.
        self.remove(&tombstone.collection, &tombstone.document_id)
            .await
    }
}

// =============================================================================
// PeerDiscovery
// =============================================================================

#[async_trait]
impl PeerDiscovery for AutomergeBackend {
    async fn start(&self) -> Result<()> {
        // No separate discovery phase — the iroh::Router inside the
        // blob store starts accepting incoming sync connections at
        // construction time. Outbound peers are reached via
        // `add_peer` / `SyncEngine::connect_to_peer`.
        Ok(())
    }

    async fn stop(&self) -> Result<()> {
        // Outbound connection teardown happens at `DataSyncBackend::shutdown`;
        // PeerDiscovery::stop is the trait-level signal "stop trying to
        // find new peers", which for this backend is a no-op (we never
        // run an auto-discovery scan).
        Ok(())
    }

    async fn discovered_peers(&self) -> Result<Vec<PeerInfo>> {
        // `MeshSyncTransport` only tracks currently-connected peers
        // (it doesn't run its own discovery). Surface those as
        // `discovered_peers` with `connected = true`; consumers that
        // need a richer view (e.g. last-seen-but-disconnected) should
        // maintain their own side-band record.
        let peers = self.transport.connected_peers();
        Ok(peers
            .into_iter()
            .map(|endpoint_id| PeerInfo {
                peer_id: endpoint_id.to_string(),
                address: None,
                transport: TransportType::Tcp,
                connected: true,
                last_seen: SystemTime::now(),
                metadata: HashMap::new(),
            })
            .collect())
    }

    async fn add_peer(&self, address: &str, _transport: TransportType) -> Result<()> {
        // `address` is expected to be a hex EndpointId; the
        // `TransportType` arg is ignored because this backend always
        // talks to peers over the same iroh endpoint regardless of
        // how the caller categorized the transport.
        let endpoint_id: iroh::EndpointId = address
            .parse()
            .with_context(|| format!("parse EndpointId from address: {address}"))?;
        self.transport
            .connect_and_authenticate(endpoint_id)
            .await
            .context("connect_and_authenticate")?;
        self.dispatch_peer_event(PeerEvent::Connected(PeerInfo {
            peer_id: endpoint_id.to_string(),
            address: Some(address.to_string()),
            transport: TransportType::Tcp,
            connected: true,
            last_seen: SystemTime::now(),
            metadata: HashMap::new(),
        }));
        Ok(())
    }

    async fn wait_for_peer(&self, peer_id: &PeerId, timeout: Duration) -> Result<()> {
        // Poll-based; matches `InMemoryBackend::wait_for_peer`'s
        // approach. Used by tests / bench scenarios that race
        // against connection establishment.
        let deadline = tokio::time::Instant::now() + timeout;
        let poll = Duration::from_millis(50);
        loop {
            for endpoint_id in self.transport.connected_peers() {
                if endpoint_id.to_string() == *peer_id {
                    return Ok(());
                }
            }
            if tokio::time::Instant::now() >= deadline {
                bail!("Timeout waiting for peer {peer_id}");
            }
            tokio::time::sleep(poll).await;
        }
    }

    /// Register a callback for peer lifecycle events.
    ///
    /// **Current coverage is one-way: outbound `Connected` only.**
    /// Callbacks registered here will fire when the adapter's own
    /// `add_peer` / `SyncEngine::connect_to_peer` successfully
    /// completes an outbound `connect_and_authenticate`. They will
    /// NOT fire for:
    /// - **Inbound connections** — the iroh `Router` (owned by the
    ///   blob store) accepts incoming sync streams and hands them to
    ///   `SyncProtocolHandler`, but the handler doesn't currently
    ///   surface a peer-event channel the adapter can subscribe to.
    /// - **`Disconnected`** — `MeshSyncTransport` removes the
    ///   connection from its cache on drop, but the adapter has no
    ///   hook into that removal path.
    /// - **Reconnect after a transient drop** — same root cause as
    ///   the two above; no event channel from the transport up to
    ///   the adapter yet.
    ///
    /// This is a known limitation tracked for a follow-up RC. For
    /// the moment, consumers needing full peer-lifecycle coverage
    /// (e.g. peat-node's per-peer connection metrics) should poll
    /// `discovered_peers()` periodically, or downcast to
    /// `AutomergeBackend` via `as_any()` and reach the
    /// `MeshSyncTransport` directly via `transport()`.
    fn on_peer_event(&self, callback: Box<dyn Fn(PeerEvent) + Send + Sync>) {
        self.peer_callbacks
            .lock()
            .unwrap_or_else(|e| e.into_inner())
            .push(callback);
    }

    async fn get_peer_info(&self, peer_id: &PeerId) -> Result<Option<PeerInfo>> {
        for endpoint_id in self.transport.connected_peers() {
            if endpoint_id.to_string() == *peer_id {
                return Ok(Some(PeerInfo {
                    peer_id: peer_id.clone(),
                    address: None,
                    transport: TransportType::Tcp,
                    connected: true,
                    last_seen: SystemTime::now(),
                    metadata: HashMap::new(),
                }));
            }
        }
        Ok(None)
    }
}

// =============================================================================
// SyncEngine
// =============================================================================

#[async_trait]
impl SyncEngine for AutomergeBackend {
    async fn start_sync(&self) -> Result<()> {
        if self.sync_active.swap(true, Ordering::SeqCst) {
            // Already syncing; idempotent no-op.
            return Ok(());
        }

        // The observer-forwarder task (which bridges local + remote
        // store events onto the `events` bus for `observe()` callers)
        // runs ALWAYS — spawned in `with_iroh`, not gated on sync
        // state. start_sync only owns the outbound auto-sync pusher:
        // on local-write notifications, push the new state to every
        // connected peer via the coordinator. Mirrors the
        // `sync_on_change` task in `peat-node::SidecarNode::new`.
        // Subscription filter: if any collection has been
        // `subscribe`d, only push keys whose collection is in the
        // set. If the set is empty, push everything (the prior
        // 7-step bootstrap had no per-collection filter, so we
        // preserve that default).
        let coordinator = Arc::clone(&self.coordinator);
        let subscriptions = Arc::clone(&self.subscriptions);
        let mut change_rx = self.store.subscribe_to_changes();
        let sync_handle = tokio::spawn(async move {
            loop {
                match change_rx.recv().await {
                    Ok(key) => {
                        let allow = {
                            let subs = subscriptions.read().await;
                            if subs.is_empty() {
                                true
                            } else if let Some((collection, _)) = key.split_once(':') {
                                subs.contains(collection)
                            } else {
                                false
                            }
                        };
                        if !allow {
                            continue;
                        }
                        // Enqueue onto the QoS-priority fanout queue instead of
                        // awaiting the fanout inline (peat-mesh#247 / ADR-0013).
                        // Non-blocking, so this loop drains `change_rx` at memory
                        // speed and the priority queue — not arrival order —
                        // decides which document is fanned out next. A
                        // high-QoS write (e.g. an alert) therefore preempts a
                        // backlog of lower-priority docs rather than waiting
                        // behind it. The fanout worker (below) performs the
                        // actual per-document fanout.
                        coordinator.enqueue_fanout(&key);
                    }
                    Err(broadcast::error::RecvError::Lagged(_)) => continue,
                    Err(broadcast::error::RecvError::Closed) => return,
                }
            }
        });
        *self.sync_task.lock().unwrap_or_else(|e| e.into_inner()) = Some(sync_handle);

        // QoS-priority fanout worker (peat-mesh#247 / ADR-0013): a single
        // task that drains the coordinator's fanout queue highest-priority-
        // first and runs each document's fanout. Single-worker preserves the
        // per-(peer, channel) ordering + backpressure the old inline-await
        // loop provided.
        let coordinator = Arc::clone(&self.coordinator);
        let fanout_handle = tokio::spawn(coordinator.run_fanout_worker());
        *self.fanout_task.lock().unwrap_or_else(|e| e.into_inner()) = Some(fanout_handle);

        // Periodic FlowController cleanup (peat-mesh#115 follow-up).
        // The rate-limiter map is keyed by (peer, doc) — without
        // eviction it grows for the process lifetime in clusters with
        // doc churn (deployments, beacons, tombstones). 60s strikes the
        // balance between memory bound and wakeup overhead; full,
        // idle buckets are dropped losslessly (a fresh lookup returns
        // the same default capacity), so eviction never affects
        // rate-limiting decisions.
        let coordinator = Arc::clone(&self.coordinator);
        let cleanup_handle = tokio::spawn(async move {
            let mut ticker = tokio::time::interval(std::time::Duration::from_secs(60));
            // Skip the immediate first tick — at t=0 there's nothing to clean.
            ticker.tick().await;
            loop {
                ticker.tick().await;
                coordinator.flow_controller().cleanup();
            }
        });
        *self
            .flow_cleanup_task
            .lock()
            .unwrap_or_else(|e| e.into_inner()) = Some(cleanup_handle);

        Ok(())
    }

    async fn stop_sync(&self) -> Result<()> {
        self.sync_active.store(false, Ordering::SeqCst);
        if let Some(handle) = self
            .sync_task
            .lock()
            .unwrap_or_else(|e| e.into_inner())
            .take()
        {
            handle.abort();
        }
        if let Some(handle) = self
            .flow_cleanup_task
            .lock()
            .unwrap_or_else(|e| e.into_inner())
            .take()
        {
            handle.abort();
        }
        // QoS-priority fanout worker (peat-mesh#247): abort alongside the
        // change loop. Aborting a task parked on the queue's `Notify` is
        // clean; any entries still pending are dropped (recovered by the
        // next change / re-sync on the next start_sync).
        if let Some(handle) = self
            .fanout_task
            .lock()
            .unwrap_or_else(|e| e.into_inner())
            .take()
        {
            handle.abort();
        }
        // Observer-forwarder is intentionally NOT stopped here —
        // it runs for the lifetime of the backend so `observe()`
        // stays functional after stop_sync. The forwarder shuts
        // down naturally when the backend is dropped and the
        // store's `observer_tx` Sender closes.
        Ok(())
    }

    async fn subscribe(&self, collection: &str, _query: &Query) -> Result<SyncSubscription> {
        // The `Query` arg shapes which documents are visible to the
        // calling subscriber locally; the sync engine itself doesn't
        // filter on query because peat-mesh syncs whole documents,
        // not query-projections.
        self.subscriptions
            .write()
            .await
            .insert(collection.to_string());
        // `SyncSubscription`'s drop semantics are owned by the
        // caller — we surface the collection name in the handle but
        // don't tie removal to drop here, because `SyncEngine` doesn't
        // give us a hook to forward Drop events. Consumers that
        // need to revoke a subscription should call back through
        // `subscribe` with the same name + a separate revocation
        // path if peat-mesh adds one later.
        Ok(SyncSubscription::new(collection, ()))
    }

    async fn set_priority(&self, _collection: &str, _priority: Priority) -> Result<()> {
        // The coordinator's per-collection priority is wired via
        // `SyncModeRegistry`, but plumbing the `Priority` enum into
        // that registry is out of scope for the initial adapter cut.
        // Default trait impl already returns Ok — we override to
        // keep the explicit no-op visible.
        Ok(())
    }

    async fn is_syncing(&self) -> Result<bool> {
        Ok(self.sync_active.load(Ordering::SeqCst))
    }

    async fn force_sync(&self) -> Result<()> {
        // Iterate every connected peer and ask the coordinator to
        // resync every local document with that peer. This is the
        // "ensure critical updates are sent" semantic the trait doc
        // describes — used by tests and by consumers reacting to a
        // network reconnect.
        let peers = self.transport.connected_peers();
        for peer in peers {
            let _ = self.coordinator.sync_all_documents_with_peer(peer).await;
        }
        Ok(())
    }

    async fn connect_to_peer(&self, endpoint_id_hex: &str, addresses: &[String]) -> Result<bool> {
        let endpoint_id: iroh::EndpointId = endpoint_id_hex
            .parse()
            .with_context(|| format!("parse EndpointId hex: {endpoint_id_hex}"))?;

        // peat-mesh#205: if the caller supplied direct addresses, build a
        // full `EndpointAddr` and dial via the airgap-safe
        // `connect_and_authenticate_with_addr` path that bypasses iroh's
        // `address_lookup` chain entirely. This is the consumer-surface
        // half of the #205 fix — the existing chain-based dial below is
        // preserved for the legacy "no addresses given, rely on
        // MemoryLookup or DNS to resolve" case.
        //
        // Pre-#205 this method had `let _ = addresses;` and ignored the
        // hints the caller passed — which is what stranded peat-cli and
        // peat-node sidecars in airgapped contexts where DNS doesn't
        // work. peat-cli/peat-node already build the full address list;
        // we now honour it.
        if !addresses.is_empty() {
            let mut peer_addr = iroh::EndpointAddr::new(endpoint_id);
            for addr_str in addresses {
                let socket_addr: std::net::SocketAddr = addr_str
                    .parse()
                    .with_context(|| format!("parse SocketAddr from address {addr_str:?}"))?;
                peer_addr.addrs.insert(iroh::TransportAddr::Ip(socket_addr));
            }
            return match self
                .transport
                .connect_and_authenticate_with_addr(peer_addr)
                .await
            {
                Ok(_) => Ok(true),
                Err(e) => Err(anyhow!("connect_to_peer (with_addr) failed: {e}")),
            };
        }

        match self.transport.connect_and_authenticate(endpoint_id).await {
            Ok(_) => Ok(true),
            Err(e) => Err(anyhow!("connect_to_peer failed: {e}")),
        }
    }
}

// =============================================================================
// DataSyncBackend
// =============================================================================

#[async_trait]
impl DataSyncBackend for AutomergeBackend {
    async fn initialize(&self, _config: BackendConfig) -> Result<()> {
        // Setup already happened in `with_iroh`; this method exists
        // to satisfy the trait contract and flip the `initialized`
        // flag that `is_ready` reads. The `BackendConfig` argument
        // is ignored because the adapter is configured at
        // construction time — re-configuring after the fact would
        // require swapping the bound endpoint, which iroh doesn't
        // support.
        //
        // Consumers that want config-driven setup should call
        // `with_iroh(...)` with the appropriate inputs and treat
        // `initialize` as a no-op signal.
        self.initialized.store(true, Ordering::SeqCst);
        Ok(())
    }

    async fn shutdown(&self) -> Result<()> {
        // Stop the sync tasks (idempotent — `stop_sync` handles
        // already-stopped state).
        let _ = self.stop_sync().await;
        self.initialized.store(false, Ordering::SeqCst);

        // peat-mesh#195: flush the Lamport high-water on clean
        // shutdown so the on-disk value reflects every issued
        // `next_lamport()` from this run, not just those that landed
        // before the most recent periodic 1-second flush. Errors are
        // logged-and-ignored per the same "flush errors must not
        // block the issued value" rule the periodic task honors —
        // a transient redb fault shouldn't fail process exit.
        if let Err(e) = self.flush_lamport_highwater() {
            tracing::warn!(
                error = %e,
                "shutdown: final lamport_highwater flush failed; \
                 worst-case loss is up to one periodic flush interval"
            );
        }

        // We do NOT shut down the iroh::Router here. Consumers
        // holding `Arc<NetworkedIrohBlobStore>` may still have work
        // to do (e.g. flushing a pending blob upload). The blob
        // store owns its own shutdown timeout and graceful close;
        // the adapter's contract is "stop sync activity", not
        // "tear down the endpoint."
        Ok(())
    }

    fn document_store(&self) -> Arc<dyn DocumentStore> {
        Arc::new(self.clone())
    }

    fn peer_discovery(&self) -> Arc<dyn PeerDiscovery> {
        Arc::new(self.clone())
    }

    fn sync_engine(&self) -> Arc<dyn SyncEngine> {
        Arc::new(self.clone())
    }

    async fn is_ready(&self) -> bool {
        self.initialized.load(Ordering::SeqCst)
    }

    fn backend_info(&self) -> BackendInfo {
        BackendInfo {
            name: "AutomergeIroh".to_string(),
            version: env!("CARGO_PKG_VERSION").to_string(),
        }
    }

    fn as_any(&self) -> &dyn std::any::Any {
        self
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::sync::types::{Query, Value};
    use std::collections::HashMap;
    use tempfile::TempDir;

    /// Build a backend rooted at a fresh temp dir. Generates a random
    /// formation secret per test so tests don't accidentally cross-talk.
    async fn fresh_backend() -> (Arc<AutomergeBackend>, TempDir) {
        let dir = TempDir::new().expect("tempdir");
        let secret = FormationKey::generate_secret();
        let backend = AutomergeBackend::with_iroh(AutomergeBackendConfig {
            data_dir: dir.path().to_path_buf(),
            formation_id: "test-formation".to_string(),
            base64_shared_key: secret,
            iroh_bind_addr: None,
            cipher: None,
            download_stall_timeout: None,
            ..Default::default()
        })
        .await
        .expect("with_iroh");
        (backend, dir)
    }

    /// Regression test for the `DocumentStore::get` keyed-lookup
    /// override (peat-mesh#186). The trait-default routes through
    /// `query(Query::Eq{id})` → `scan_prefix(collection_prefix)` —
    /// O(N) per lookup, O(N²) in a loop. This test pins the override's
    /// correctness across (a) hit, (b) miss, (c) collection isolation
    /// (same `doc_id` in a different collection must not be returned),
    /// and (d) soft-delete filtering matches the trait-default
    /// behaviour (excluded unless caller uses `Query::IncludeDeleted`).
    /// The perf characteristic itself is encoded in the override
    /// body (single `store.get(&key)` call); a reviewer reverting the
    /// override would re-introduce the O(N) trap, so this test guards
    /// the contract, not the timing.
    #[tokio::test]
    async fn get_uses_keyed_lookup_and_matches_default_deletion_semantics() {
        let (backend, _dir) = fresh_backend().await;
        let store: Arc<dyn DocumentStore> = backend.document_store();

        // (a) hit — direct key lookup returns the doc.
        let mut fields_a = HashMap::new();
        fields_a.insert("v".to_string(), Value::String("hit".to_string()));
        store
            .upsert("col-x", Document::with_id("d-1", fields_a))
            .await
            .unwrap();
        let got = store
            .get("col-x", &"d-1".to_string())
            .await
            .unwrap()
            .expect("d-1 in col-x must be present");
        assert_eq!(got.fields.get("v"), Some(&Value::String("hit".to_string())));

        // (b) miss — unknown id returns None.
        let miss = store
            .get("col-x", &"d-does-not-exist".to_string())
            .await
            .unwrap();
        assert!(miss.is_none(), "missing id must return None");

        // (c) collection isolation — same `doc_id` in a different
        // collection must not leak across. `key_for` namespaces by
        // `collection:doc_id`; a sloppy reimplementation that scanned
        // by suffix would return the wrong doc here.
        let mut fields_other = HashMap::new();
        fields_other.insert("v".to_string(), Value::String("other".to_string()));
        store
            .upsert("col-y", Document::with_id("d-1", fields_other))
            .await
            .unwrap();
        let from_x = store
            .get("col-x", &"d-1".to_string())
            .await
            .unwrap()
            .expect("col-x:d-1 still present");
        assert_eq!(
            from_x.fields.get("v"),
            Some(&Value::String("hit".to_string())),
            "col-x lookup must not return col-y's doc"
        );

        // (d) soft-delete — `_deleted=true` excludes from get(),
        // matching the trait-default `get` (which routes through
        // `Query::Eq{id}` and filters via `matches_deletion_state`).
        // This sub-case directly synthesizes the on-disk shape via
        // `upsert` — it does NOT go through `DocumentStore::remove`
        // / soft-delete policy. The contract being pinned here is
        // the *filter* on `_deleted=true`, not the soft-delete write
        // path; production code reaches this state via `remove()`,
        // not by direct `_deleted=true` upsert.
        let mut fields_deleted = HashMap::new();
        fields_deleted.insert("v".to_string(), Value::String("tombstoned".to_string()));
        fields_deleted.insert("_deleted".to_string(), Value::Bool(true));
        store
            .upsert("col-x", Document::with_id("d-tomb", fields_deleted))
            .await
            .unwrap();
        let tombstoned = store.get("col-x", &"d-tomb".to_string()).await.unwrap();
        assert!(
            tombstoned.is_none(),
            "soft-deleted doc must be filtered out by get(), matching trait-default semantics"
        );
    }

    #[tokio::test]
    async fn upsert_get_remove_round_trip() {
        let (backend, _dir) = fresh_backend().await;
        let store: Arc<dyn DocumentStore> = backend.document_store();

        let mut fields = HashMap::new();
        fields.insert("name".to_string(), Value::String("alpha".to_string()));
        let id = store
            .upsert("markers", Document::with_id("m-1", fields))
            .await
            .unwrap();
        assert_eq!(id, "m-1");

        let fetched = store
            .get("markers", &"m-1".to_string())
            .await
            .unwrap()
            .expect("doc present");
        assert_eq!(
            fetched.fields.get("name"),
            Some(&Value::String("alpha".to_string()))
        );

        store.remove("markers", &"m-1".to_string()).await.unwrap();
        let after = store.get("markers", &"m-1".to_string()).await.unwrap();
        assert!(after.is_none());
    }

    #[tokio::test]
    async fn query_filters_by_field() {
        let (backend, _dir) = fresh_backend().await;
        let store = backend.document_store();

        let mut active = HashMap::new();
        active.insert("status".to_string(), Value::String("active".to_string()));
        let mut inactive = HashMap::new();
        inactive.insert("status".to_string(), Value::String("inactive".to_string()));
        store
            .upsert("tracks", Document::with_id("t-1", active))
            .await
            .unwrap();
        store
            .upsert("tracks", Document::with_id("t-2", inactive))
            .await
            .unwrap();

        let results = store
            .query(
                "tracks",
                &Query::Eq {
                    field: "status".to_string(),
                    value: Value::String("active".to_string()),
                },
            )
            .await
            .unwrap();
        assert_eq!(results.len(), 1);
        assert_eq!(results[0].id, Some("t-1".to_string()));
    }

    #[tokio::test]
    async fn auto_generated_id() {
        let (backend, _dir) = fresh_backend().await;
        let store = backend.document_store();
        let id = store
            .upsert("platforms", Document::new(HashMap::new()))
            .await
            .unwrap();
        assert!(!id.is_empty(), "auto-generated id must be non-empty");
        let fetched = store.get("platforms", &id).await.unwrap();
        assert!(fetched.is_some());
    }

    #[tokio::test]
    async fn observe_receives_local_upserts() {
        let (backend, _dir) = fresh_backend().await;
        let store = backend.document_store();

        let mut stream = store.observe("cells", &Query::All).unwrap();
        // Tiny delay so the spawned filter task is parked on bus.recv() before the upsert fires.
        tokio::time::sleep(Duration::from_millis(20)).await;

        let mut fields = HashMap::new();
        fields.insert("label".to_string(), Value::String("ALPHA".to_string()));
        store
            .upsert("cells", Document::with_id("c-1", fields))
            .await
            .unwrap();

        let event = tokio::time::timeout(Duration::from_secs(1), stream.receiver.recv())
            .await
            .expect("observer should receive event")
            .expect("event channel still open");
        match event {
            ChangeEvent::Updated {
                collection,
                document,
                ..
            } => {
                assert_eq!(collection, "cells");
                assert_eq!(document.id, Some("c-1".to_string()));
                assert_eq!(
                    document.fields.get("label"),
                    Some(&Value::String("ALPHA".to_string()))
                );
            }
            other => panic!("expected Updated, got {other:?}"),
        }
    }

    #[tokio::test]
    async fn observe_filters_by_collection() {
        let (backend, _dir) = fresh_backend().await;
        let store = backend.document_store();

        let mut stream = store.observe("a", &Query::All).unwrap();
        tokio::time::sleep(Duration::from_millis(20)).await;

        // Upsert in a *different* collection — observer must NOT see it.
        store
            .upsert("b", Document::with_id("b-1", HashMap::new()))
            .await
            .unwrap();

        let got = tokio::time::timeout(Duration::from_millis(200), stream.receiver.recv()).await;
        assert!(
            got.is_err(),
            "observer on collection 'a' should not see upserts on 'b'"
        );
    }

    #[tokio::test]
    async fn observe_emits_removed_on_local_delete() {
        // Post round-2 the observer-forwarder is always-on and
        // routes `Updated` events asynchronously through a spawned
        // task, while `remove()` direct-emits `Removed` via
        // `publish_event` (because `AutomergeStore::delete` doesn't
        // broadcast). The two paths can interleave, so the order
        // observers see events for `upsert; remove;` on the same
        // key is not guaranteed. What IS guaranteed: both events
        // arrive within the timeout, and the Removed event names
        // the right doc_id. Real consumers treat events as a
        // stream of facts and dedup via their own state machine —
        // the test pins the same shape.
        let (backend, _dir) = fresh_backend().await;
        let store = backend.document_store();

        let mut stream = store.observe("markers", &Query::All).unwrap();
        tokio::time::sleep(Duration::from_millis(20)).await;

        store
            .upsert("markers", Document::with_id("m-1", HashMap::new()))
            .await
            .unwrap();
        store.remove("markers", &"m-1".to_string()).await.unwrap();

        let mut saw_updated = false;
        let mut saw_removed = false;
        for _ in 0..2 {
            let event = tokio::time::timeout(Duration::from_secs(1), stream.receiver.recv())
                .await
                .expect("observer should receive event")
                .expect("channel open");
            match event {
                ChangeEvent::Updated { document, .. } => {
                    assert_eq!(document.id, Some("m-1".to_string()));
                    saw_updated = true;
                }
                ChangeEvent::Removed { doc_id, .. } => {
                    assert_eq!(doc_id, "m-1");
                    saw_removed = true;
                }
                other => panic!("unexpected event {other:?}"),
            }
        }
        assert!(saw_updated, "must see Updated for m-1");
        assert!(saw_removed, "must see Removed for m-1");
    }

    #[tokio::test]
    async fn lifecycle_initialize_shutdown() {
        let (backend, _dir) = fresh_backend().await;
        assert!(!backend.is_ready().await);
        backend
            .initialize(BackendConfig {
                app_id: "test-app".to_string(),
                persistence_dir: std::path::PathBuf::from("."),
                shared_key: None,
                transport: crate::sync::types::TransportConfig::default(),
                extra: HashMap::new(),
            })
            .await
            .unwrap();
        assert!(backend.is_ready().await);
        backend.shutdown().await.unwrap();
        assert!(!backend.is_ready().await);
    }

    #[tokio::test]
    async fn sync_engine_start_stop_toggles_is_syncing() {
        let (backend, _dir) = fresh_backend().await;
        let engine = backend.sync_engine();
        assert!(!engine.is_syncing().await.unwrap());
        engine.start_sync().await.unwrap();
        assert!(engine.is_syncing().await.unwrap());
        engine.stop_sync().await.unwrap();
        assert!(!engine.is_syncing().await.unwrap());
    }

    #[tokio::test]
    async fn subscribe_tracks_collection() {
        let (backend, _dir) = fresh_backend().await;
        let engine = backend.sync_engine();
        let _sub = engine.subscribe("markers", &Query::All).await.unwrap();
        let subs = backend.subscriptions.read().await;
        assert!(subs.contains("markers"));
    }

    #[tokio::test]
    async fn peer_discovery_starts_with_zero_peers() {
        let (backend, _dir) = fresh_backend().await;
        let disc = backend.peer_discovery();
        let peers = disc.discovered_peers().await.unwrap();
        assert!(peers.is_empty());
    }

    #[tokio::test]
    async fn backend_info_identifies_adapter() {
        let (backend, _dir) = fresh_backend().await;
        let info = backend.backend_info();
        assert_eq!(info.name, "AutomergeIroh");
        assert!(!info.version.is_empty());
    }

    #[tokio::test]
    async fn data_sync_backend_accessors_share_state() {
        // All three accessors return the same underlying state via
        // Arc clones. Upserting through one view should be visible
        // through another.
        let (backend, _dir) = fresh_backend().await;
        let store_a = backend.document_store();
        let store_b = backend.document_store();
        store_a
            .upsert("col", Document::with_id("d-1", HashMap::new()))
            .await
            .unwrap();
        let fetched = store_b.get("col", &"d-1".to_string()).await.unwrap();
        assert!(fetched.is_some());
    }

    #[tokio::test]
    async fn as_any_downcasts_to_concrete() {
        let (backend, _dir) = fresh_backend().await;
        let inner = backend.as_any().downcast_ref::<AutomergeBackend>();
        assert!(inner.is_some());
    }

    #[tokio::test]
    async fn force_sync_is_noop_with_no_peers() {
        let (backend, _dir) = fresh_backend().await;
        let engine = backend.sync_engine();
        // No peers connected; force_sync should still succeed.
        engine.force_sync().await.unwrap();
    }

    #[tokio::test]
    async fn wait_for_peer_times_out_when_no_peer_connects() {
        let (backend, _dir) = fresh_backend().await;
        let disc = backend.peer_discovery();
        let result = disc
            .wait_for_peer(&"deadbeef".to_string(), Duration::from_millis(100))
            .await;
        assert!(result.is_err());
    }

    /// QA round-1 [WARNING] coverage: the receive-side observer-forwarder
    /// task spawned by `start_sync` is the entire mechanism by which
    /// sync-ingested changes reach a consumer's `observe()` stream.
    /// We can't drive a real peer in-process without standing up a
    /// second iroh endpoint, but we can simulate the inbound path
    /// by going straight to `AutomergeStore::put` (which is what the
    /// inbound `SyncProtocolHandler` does after applying a sync
    /// message). The observer-forwarder watches
    /// `subscribe_to_observer_changes` — which fires on ALL puts,
    /// not just locally-initiated ones — so a direct put is a
    /// faithful stand-in for "a peer's sync message just merged."
    ///
    /// Pins the forwarder against regressions in key parsing,
    /// missing-store lookups, and the local-vs-synced origin
    /// contract. peat-mesh#106 QA round-1.
    #[tokio::test]
    async fn observer_forwarder_surfaces_sync_ingested_changes() {
        let (backend, _dir) = fresh_backend().await;
        let store = backend.document_store();
        let engine = backend.sync_engine();
        engine.start_sync().await.unwrap();

        // Subscribe to the observe() stream BEFORE the simulated
        // inbound merge.
        let mut stream = store.observe("alerts", &Query::All).unwrap();
        tokio::time::sleep(Duration::from_millis(40)).await;

        // Simulate an inbound sync merge by writing directly to the
        // backing AutomergeStore. The observer-forwarder task
        // subscribes to `subscribe_to_observer_changes`, which fires
        // for every put — so this exercises the same code path a
        // real peer's sync message would.
        let key = "alerts:emerg-1";
        let body = serde_json::json!({ "severity": "high", "msg": "test" });
        let am = crate::storage::json_convert::json_to_automerge(&body, None).unwrap();
        backend.store().put(key, &am).unwrap();

        let event = tokio::time::timeout(Duration::from_secs(1), stream.receiver.recv())
            .await
            .expect("observer should receive forwarded event")
            .expect("event channel open");
        match event {
            ChangeEvent::Updated {
                collection,
                document,
                origin,
            } => {
                assert_eq!(collection, "alerts");
                assert_eq!(document.id, Some("emerg-1".to_string()));
                assert_eq!(
                    document.fields.get("severity"),
                    Some(&Value::String("high".to_string()))
                );
                // ADR-059 contract: sync-ingested events have
                // origin = None; only `upsert_with_origin` callers
                // attach an origin. The forwarder must honor this.
                assert!(
                    origin.is_none(),
                    "sync-ingested events must have origin = None per ADR-059, got: {origin:?}"
                );
            }
            other => panic!("expected Updated, got {other:?}"),
        }
    }

    /// QA round-1 [WARNING] coverage: `apply_tombstone` composes
    /// `put_tombstone` + a chained `remove` that re-enters the
    /// stripe lock. Pins the happy path: applying a tombstone for
    /// a live doc removes the live doc, retains the tombstone, and
    /// surfaces it via `get_tombstones`.
    #[tokio::test]
    async fn apply_tombstone_round_trip() {
        let (backend, _dir) = fresh_backend().await;
        let store = backend.document_store();

        // Seed a live doc the tombstone will reap.
        store
            .upsert("notes", Document::with_id("n-1", HashMap::new()))
            .await
            .unwrap();

        let tombstone = crate::qos::Tombstone {
            collection: "notes".to_string(),
            document_id: "n-1".to_string(),
            deleted_at: SystemTime::now(),
            deleted_by: "test-author".to_string(),
            lamport: 1,
            reason: Some("test".to_string()),
        };
        store.apply_tombstone(&tombstone).await.unwrap();

        // The live doc is gone.
        let after = store.get("notes", &"n-1".to_string()).await.unwrap();
        assert!(after.is_none(), "live doc must be removed by tombstone");

        // The tombstone is retained and surfaces via get_tombstones.
        let tombstones = store.get_tombstones("notes").await.unwrap();
        assert_eq!(tombstones.len(), 1);
        assert_eq!(tombstones[0].document_id, "n-1");
    }

    /// QA round-2 [WARNING] coverage: pre-round-2, local
    /// `upsert_with_origin` emitted **two** `Updated` events on
    /// `observe()` when sync was active — once directly from
    /// `upsert_with_origin.publish_event`, and once again via the
    /// observer-forwarder picking up the resulting `put` from
    /// `AutomergeStore::observer_tx`. Round-2 restructured the flow
    /// so that all `Updated` events flow through the forwarder via
    /// a `pending_origins` map, and `upsert_with_origin` no longer
    /// direct-publishes. This test pins the single-emission
    /// contract: one local upsert with an origin produces exactly
    /// one `Updated` event with that origin, even with sync
    /// active.
    #[tokio::test]
    async fn local_upsert_emits_exactly_one_updated_with_origin() {
        let (backend, _dir) = fresh_backend().await;
        let store = backend.document_store();

        // Start sync — the pre-round-2 dual-emission bug only
        // triggered once the forwarder was running. With the
        // always-on forwarder in round-2, the duplicate would
        // appear regardless of sync state, but enabling sync makes
        // the pre-fix repro most faithful.
        backend.sync_engine().start_sync().await.unwrap();

        let mut stream = store.observe("alerts", &Query::All).unwrap();
        tokio::time::sleep(Duration::from_millis(20)).await;

        store
            .upsert_with_origin(
                "alerts",
                Document::with_id("a-1", HashMap::new()),
                Some("ble".to_string()),
            )
            .await
            .unwrap();

        // Exactly one Updated, with the caller's origin.
        let event = tokio::time::timeout(Duration::from_secs(1), stream.receiver.recv())
            .await
            .expect("observer should receive event")
            .expect("channel open");
        match event {
            ChangeEvent::Updated {
                document, origin, ..
            } => {
                assert_eq!(document.id, Some("a-1".to_string()));
                assert_eq!(
                    origin,
                    Some("ble".to_string()),
                    "single emission must carry the caller's origin"
                );
            }
            other => panic!("expected Updated, got {other:?}"),
        }

        // No second event within a generous window — pins the
        // exactly-one invariant. Pre-round-2 this assertion would
        // fail with a second Updated{origin: None} from the
        // forwarder's redundant emission.
        let extra = tokio::time::timeout(Duration::from_millis(200), stream.receiver.recv()).await;
        assert!(
            extra.is_err(),
            "expected no second event from a single upsert; pre-round-2 bug would emit a duplicate Updated{{origin: None}}"
        );
    }

    /// QA round-4 [WARNING] coverage: the observer-forwarder task
    /// must exit when the backend is dropped. Pre-round-4 the task
    /// captured a strong `Arc<AutomergeStore>` across the
    /// `rx.recv().await` park, forming an ownership cycle (task →
    /// store → observer_tx Sender → task). Every backend drop
    /// leaked the task + store + redb handle for the process
    /// lifetime.
    ///
    /// This test pins the termination contract: drop all strong
    /// references to the backend, then poll the observer task's
    /// `is_finished()`. Pre-round-4 the assertion would hang
    /// indefinitely (the task is parked on a recv that will
    /// never close); post-round-4 the task observes the Sender's
    /// drop and exits within milliseconds.
    #[tokio::test]
    async fn observer_task_exits_when_backend_drops() {
        let (backend, _dir) = fresh_backend().await;
        // Clone the JoinHandle Arc so we can poll its termination
        // after the backend itself is gone.
        let task = Arc::clone(&backend.observer_task);
        assert!(!task.is_finished(), "task should be running pre-drop");

        // Drop the only strong reference to the backend. The
        // task's Weak<AutomergeStore> can no longer upgrade, and
        // the observer_tx Sender drops with the store, closing
        // the broadcast receiver inside the task.
        drop(backend);

        // Poll-with-timeout. 1s is generous — in practice the
        // task exits within a few ms once the recv resolves Closed.
        let deadline = tokio::time::Instant::now() + Duration::from_secs(1);
        while !task.is_finished() && tokio::time::Instant::now() < deadline {
            tokio::time::sleep(Duration::from_millis(20)).await;
        }
        assert!(
            task.is_finished(),
            "observer task must exit within 1s of backend drop \
             (pre-round-4 the strong Arc<store> held across rx.recv().await \
             kept the Sender alive and the task parked indefinitely)"
        );
    }

    /// QA round-3 SUGGESTION coverage: directly target the
    /// pre-baking guarantee documented on `pending_emissions`.
    /// Pre-pre-baking, an upsert immediately followed by a remove
    /// (same task, no scheduler yield between) would race the
    /// observer-forwarder's `store.get(&key)` against the
    /// `store.delete(&key)` — the delete would land first, the
    /// forwarder's get would return None, and the Updated event
    /// would be silently dropped. Observers would see only
    /// `Removed` even though a put-then-delete sequence is
    /// indistinguishable from a put alone at the storage layer.
    ///
    /// Post pre-baking, `upsert_with_origin` stages the full
    /// Document under the same stripe lock as the put, so the
    /// forwarder can deliver the event from the pre-bake without
    /// touching the (possibly already-deleted) store entry.
    /// Asserting the `Updated` event arrives — with the original
    /// payload — is the direct test for that invariant.
    #[tokio::test]
    async fn upsert_immediately_followed_by_remove_preserves_updated() {
        let (backend, _dir) = fresh_backend().await;
        let store = backend.document_store();

        let mut stream = store.observe("ephemera", &Query::All).unwrap();
        tokio::time::sleep(Duration::from_millis(20)).await;

        let mut fields = HashMap::new();
        fields.insert("kind".to_string(), Value::String("flash".to_string()));
        // Upsert then immediately remove — no `await` of anything
        // else between, no sleep, no yield. The pre-baking
        // guarantee says the Updated still survives.
        store
            .upsert_with_origin(
                "ephemera",
                Document::with_id("flash-1", fields),
                Some("ble".to_string()),
            )
            .await
            .unwrap();
        store
            .remove("ephemera", &"flash-1".to_string())
            .await
            .unwrap();

        let mut saw_updated_with_payload = false;
        let mut saw_removed = false;
        for _ in 0..2 {
            let event = tokio::time::timeout(Duration::from_secs(1), stream.receiver.recv())
                .await
                .expect("observer should receive event")
                .expect("channel open");
            match event {
                ChangeEvent::Updated {
                    document, origin, ..
                } => {
                    // The pre-bake must preserve the full payload
                    // (fields + origin), not a fallback fetch from
                    // a now-empty store that would synthesize an
                    // empty Document with origin=None.
                    assert_eq!(document.id, Some("flash-1".to_string()));
                    assert_eq!(
                        document.fields.get("kind"),
                        Some(&Value::String("flash".to_string())),
                        "pre-baked payload must include original fields"
                    );
                    assert_eq!(
                        origin,
                        Some("ble".to_string()),
                        "pre-baked event must carry the caller's origin"
                    );
                    saw_updated_with_payload = true;
                }
                ChangeEvent::Removed { doc_id, .. } => {
                    assert_eq!(doc_id, "flash-1");
                    saw_removed = true;
                }
                other => panic!("unexpected event {other:?}"),
            }
        }
        assert!(
            saw_updated_with_payload,
            "Updated event must survive the immediately-following remove (pre-baking guarantee)"
        );
        assert!(saw_removed, "Removed event must also arrive");
    }

    /// QA round-1 [WARNING] coverage: `apply_tombstone` must also
    /// handle the case where the live doc is already absent (the
    /// typical inbound-sync scenario: peer A deletes, peer B never
    /// had the doc to begin with). The chained `remove` is a no-op
    /// for missing keys (`AutomergeStore::delete` doesn't error on
    /// absence), so the tombstone should land without raising.
    #[tokio::test]
    async fn apply_tombstone_for_absent_doc_is_ok() {
        let (backend, _dir) = fresh_backend().await;
        let store = backend.document_store();

        let tombstone = crate::qos::Tombstone {
            collection: "ghosts".to_string(),
            document_id: "never-existed".to_string(),
            deleted_at: SystemTime::now(),
            deleted_by: "test-author".to_string(),
            lamport: 1,
            reason: None,
        };
        store
            .apply_tombstone(&tombstone)
            .await
            .expect("tombstone for absent doc must succeed");

        let tombstones = store.get_tombstones("ghosts").await.unwrap();
        assert_eq!(tombstones.len(), 1);
    }

    /// Issue #106 (reopen): the adapter must accept a pinned Iroh
    /// UDP bind address. Direct-addressed deployments
    /// (Docker compose, k8s `hostPort`, cross-cluster test harnesses)
    /// rely on a known port; the rc.8 ephemeral-only shape silently
    /// dropped the pinned-port path peat-node had landed in v0.1.1
    /// (peat-node#42).
    ///
    /// Use `127.0.0.1:0` rather than a fixed port so parallel test
    /// runs don't collide on a busy CI port. The "0" still goes
    /// through the `bind_addr` builder branch — the OS-chosen port
    /// applies to *that* socket, not the iroh-default `0.0.0.0`
    /// fallback. The behavior under test is twofold:
    /// 1. **Structural**: the `cfg.iroh_bind_addr` path runs without
    ///    `InvalidSocketAddr` or builder panic; the backend
    ///    bootstraps and is fully wired (verified by a round-trip
    ///    upsert through `DocumentStore`).
    /// 2. **Behavioral**: `Endpoint::bound_sockets()` reports a
    ///    socket whose IP matches the requested IP, and the
    ///    iroh-default dual-stack `0.0.0.0` / `[::]` pair is *not*
    ///    additionally present. This pins the
    ///    `clear_ip_transports().bind_addr(addr)` reshape against a
    ///    regression that silently dropped or augmented the default
    ///    sockets — without this check, a future refactor that
    ///    accidentally skipped `clear_ip_transports` would still
    ///    pass the bootstrap-success half of the test.
    #[tokio::test]
    async fn with_iroh_honors_pinned_bind_addr() {
        use std::net::{IpAddr, Ipv4Addr};

        let dir = TempDir::new().expect("tempdir");
        let secret = FormationKey::generate_secret();
        let requested_ip = IpAddr::V4(Ipv4Addr::LOCALHOST);
        let backend = AutomergeBackend::with_iroh(AutomergeBackendConfig {
            data_dir: dir.path().to_path_buf(),
            formation_id: "test-formation".to_string(),
            base64_shared_key: secret,
            iroh_bind_addr: Some(SocketAddr::new(requested_ip, 0)),
            cipher: None,
            download_stall_timeout: None,
            ..Default::default()
        })
        .await
        .expect("with_iroh with pinned bind addr");

        // Behavioral assertion: the resulting endpoint must actually
        // be bound on the requested IP, and nothing else. iroh's
        // default builder binds both `0.0.0.0` and `[::]`; the
        // `clear_ip_transports` step is what makes our pinning a
        // replacement rather than an addition. If that step were
        // ever removed by accident, this assertion would fail.
        let bound = backend.blob_store().endpoint().bound_sockets();
        assert!(
            !bound.is_empty(),
            "endpoint must report at least one bound socket"
        );
        assert!(
            bound.iter().all(|s| s.ip() == requested_ip),
            "every bound socket must match the requested IP {requested_ip}; got {bound:?} \
             (a stray 0.0.0.0 or [::] here means clear_ip_transports was skipped)"
        );

        // Structural sanity-check that the rest of the adapter is
        // wired and the pinned-bind path didn't break ingest.
        let store = backend.document_store();
        let id = store
            .upsert("col", Document::with_id("d-1", HashMap::new()))
            .await
            .unwrap();
        assert_eq!(id, "d-1");
    }

    /// Surface-tier wiring test: `AutomergeBackendConfig` with both
    /// `ttl_config: Some(TtlConfig::new())` and `gc_config: Some(GcConfig::default())`
    /// must start successfully — verifying that the optional background tasks
    /// are wired without panic or error (peat-node#136).
    #[tokio::test]
    async fn ttl_and_gc_config_wire_into_with_iroh() {
        use crate::qos::GcConfig;
        use crate::storage::TtlConfig;

        let dir = TempDir::new().expect("tempdir");
        let secret = FormationKey::generate_secret();
        let result = AutomergeBackend::with_iroh(AutomergeBackendConfig {
            data_dir: dir.path().to_path_buf(),
            formation_id: "ttl-gc-wiring-test".to_string(),
            base64_shared_key: secret,
            iroh_bind_addr: None,
            cipher: None,
            download_stall_timeout: None,
            ttl_config: Some(TtlConfig::new()),
            gc_config: Some(GcConfig::default()),
            iroh_secret_key: None,
        })
        .await;
        assert!(
            result.is_ok(),
            "with_iroh must succeed with ttl_config + gc_config wired; got: {:?}",
            result.err()
        );
        result.unwrap().shutdown_and_release().await.unwrap();
    }

    /// `tombstone_ttl_hours` below 24 h emits a warn-level log but must NOT
    /// cause `with_iroh` to return an error — callers that intentionally set
    /// a short TTL for test/staging environments should not have the call fail
    /// under them (peat-node#136).
    #[tokio::test]
    async fn ttl_config_below_24h_emits_warning_but_succeeds() {
        use crate::storage::TtlConfig;

        let dir = TempDir::new().expect("tempdir");
        let secret = FormationKey::generate_secret();
        let result = AutomergeBackend::with_iroh(AutomergeBackendConfig {
            data_dir: dir.path().to_path_buf(),
            formation_id: "ttl-below-24h-test".to_string(),
            base64_shared_key: secret,
            iroh_bind_addr: None,
            cipher: None,
            download_stall_timeout: None,
            // 1 h is below the 24 h threshold that triggers a warn log; the
            // call must still succeed — the warning is advisory, not fatal.
            ttl_config: Some(TtlConfig {
                tombstone_ttl_hours: 1,
                ..TtlConfig::new()
            }),
            gc_config: None,
            iroh_secret_key: None,
        })
        .await;
        assert!(
            result.is_ok(),
            "with_iroh must succeed even with tombstone_ttl_hours < 24; got: {:?}",
            result.err()
        );
        result.unwrap().shutdown_and_release().await.unwrap();
    }

    /// `iroh_secret_key` pins the iroh EndpointId to a deterministic value.
    /// Two backends constructed from the same raw key bytes must produce the
    /// same EndpointId; a backend with `None` produces a different (random)
    /// EndpointId from one constructed with a fixed key.
    #[tokio::test(flavor = "multi_thread")]
    async fn test_automerge_backend_config_iroh_secret_key() {
        let dir1 = TempDir::new().expect("tempdir1");
        let dir2 = TempDir::new().expect("tempdir2");
        let dir3 = TempDir::new().expect("tempdir3");
        let secret = FormationKey::generate_secret();
        let fixed_key = [0x42u8; 32];

        let b1 = AutomergeBackend::with_iroh(AutomergeBackendConfig {
            data_dir: dir1.path().to_path_buf(),
            formation_id: "test-formation".to_string(),
            base64_shared_key: secret.clone(),
            iroh_secret_key: Some(fixed_key),
            ..Default::default()
        })
        .await
        .expect("b1 with_iroh");

        let b2 = AutomergeBackend::with_iroh(AutomergeBackendConfig {
            data_dir: dir2.path().to_path_buf(),
            formation_id: "test-formation".to_string(),
            base64_shared_key: secret.clone(),
            iroh_secret_key: Some(fixed_key),
            ..Default::default()
        })
        .await
        .expect("b2 with_iroh");

        // None → random ephemeral key; must differ from the fixed-key endpoint.
        let b3 = AutomergeBackend::with_iroh(AutomergeBackendConfig {
            data_dir: dir3.path().to_path_buf(),
            formation_id: "test-formation".to_string(),
            base64_shared_key: secret.clone(),
            iroh_secret_key: None,
            ..Default::default()
        })
        .await
        .expect("b3 with_iroh");

        assert_eq!(
            b1.blob_store().endpoint_id(),
            b2.blob_store().endpoint_id(),
            "same iroh_secret_key must produce the same EndpointId"
        );
        assert_ne!(
            b1.blob_store().endpoint_id(),
            b3.blob_store().endpoint_id(),
            "None iroh_secret_key must produce a different (random) EndpointId"
        );

        b1.shutdown_and_release().await.unwrap();
        b2.shutdown_and_release().await.unwrap();
        b3.shutdown_and_release().await.unwrap();
    }
}
