//! Client-side conflict detection and resolution for SyncKit.
//!
//! SyncKit uses E2E encryption, the server never sees row contents and cannot
//! merge or compare data. All conflict resolution must happen client-side after
//! decryption.
//!
//! This module provides:
//! - [`detect_conflicts`]: pure function that splits pulled changes into clean
//!   (non-conflicting) and conflicting sets
//! - [`resolve_lww`]: last-write-wins resolution by `client_timestamp`
//! - [`resolve_field_merge`]: 3-way JSON object merge using a base version
//! - [`ConflictResolver`]: trait for custom resolution strategies

use std::cmp::Ordering;
use std::collections::HashMap;

use chrono::{DateTime, Utc};
#[cfg(test)]
use uuid::Uuid;

use crate::ids::DeviceId;
use crate::types::{ChangeEntry, Hlc, PulledChange};

/// A remote change that conflicts with a local pending change.
#[derive(Debug, Clone)]
pub struct ConflictPair {
    /// The remote change from pull.
    pub remote: PulledChange,
    /// The local pending change that conflicts.
    pub local: ChangeEntry,
}

/// The non-conflicting changes from a pull, withheld behind the HLC gate.
///
/// "Clean" means only that no local *pending* edit contests the row, **not**
/// that the remote change is newer than what is already committed locally. A
/// device that applied and pushed a newer edit, then pulls an older remote edit
/// for the same row, sees that older edit here. Applying it blindly clobbers the
/// newer local value.
///
/// So the applicable entries cannot be read out directly: [`gated`](Self::gated)
/// is the only door, and it requires the row's committed HLC. This makes
/// "apply a clean change without checking the committed clock" unrepresentable
/// rather than a contract the caller has to remember.
#[derive(Debug, Clone)]
pub struct CleanChanges(Vec<PulledChange>);

impl CleanChanges {
    /// HLC-gate the clean changes against committed local clocks.
    ///
    /// `committed_hlc(table, row_id)` returns the HLC currently committed for a
    /// row, or `None` if it has never been applied locally. A change is dropped
    /// when its HLC is older than or equal to the committed clock; everything
    /// else is returned in pull order, ready to apply.
    ///
    /// Samples the wall clock once for the poisoning check; use
    /// [`gated_at`](Self::gated_at) to pin `now` across a whole sync batch (or in
    /// tests) so every change in the batch is classified against the same instant.
    pub fn gated(self, committed_hlc: impl Fn(&str, &str) -> Option<Hlc>) -> Vec<ChangeEntry> {
        self.gated_at(Utc::now(), committed_hlc)
    }

    /// [`gated`](Self::gated) with an explicit `now`, so the clock-poisoning
    /// classification is deterministic (testable) and consistent across a batch.
    ///
    /// The poisoning threshold is still fundamentally wall-clock-relative: two
    /// devices whose clocks straddle the drift boundary by less than their skew
    /// can classify a near-boundary entry differently. Pinning `now` per batch
    /// removes the *within-device* drift between the batch's first and last change;
    /// the residual cross-device boundary window is the bounded, documented
    /// tradeoff of [`MAX_HLC_DRIFT_MS`] (see [`resolve_lww_at`]).
    pub fn gated_at(
        self,
        now: DateTime<Utc>,
        committed_hlc: impl Fn(&str, &str) -> Option<Hlc>,
    ) -> Vec<ChangeEntry> {
        self.0
            .into_iter()
            .filter(|p| {
                // A clean change is applied without passing through `resolve_lww`,
                // so the clock-poisoning guard must also live here: never apply an
                // entry whose wall clock is implausibly far in the future.
                if is_clock_poisoned(&p.entry.hlc, now) {
                    tracing::warn!(
                        table = %p.entry.table,
                        wall_ms = p.entry.hlc.wall_ms,
                        "dropping clean change with implausibly-future HLC"
                    );
                    return false;
                }
                let apply = committed_hlc(&p.entry.table, &p.entry.row_id)
                    .is_none_or(|committed| p.entry.hlc > committed);
                if !apply {
                    tracing::debug!(
                        table = %p.entry.table,
                        row_id = %p.entry.row_id,
                        "clean change gated out: older than the committed HLC"
                    );
                }
                apply
            })
            .map(|p| p.entry)
            .collect()
    }

    /// The `(table, row_id)` addresses of the clean changes, so a caller can
    /// pre-fetch their committed clocks in one query before [`gated`](Self::gated).
    /// Yields addresses only, the change payloads stay sealed until gating.
    pub fn row_keys(&self) -> impl Iterator<Item = (&str, &str)> {
        self.0
            .iter()
            .map(|p| (p.entry.table.as_str(), p.entry.row_id.as_str()))
    }

    /// Number of clean changes before gating (for metrics/logging).
    pub fn len(&self) -> usize {
        self.0.len()
    }

    /// Whether there are no clean changes at all.
    pub fn is_empty(&self) -> bool {
        self.0.is_empty()
    }
}

/// How a conflict should be resolved.
#[derive(Debug, Clone)]
pub enum Resolution {
    /// Keep the local version; skip the remote change.
    /// The local version will push on the next sync cycle.
    KeepLocal,
    /// Apply the remote change, discarding the local version.
    KeepRemote,
    /// Apply a merged result that combines both changes.
    Merged(serde_json::Value),
    /// Skip both changes (neither apply nor push).
    Skip,
}

/// Trait for custom conflict resolution strategies.
///
/// Implement this to plug in app-specific merge logic. The `base` parameter
/// is `Some` only if the app provides a base-store adapter.
pub trait ConflictResolver: Send + Sync {
    /// Resolve a conflict between a `local` change and a `remote` change for the
    /// same row, optionally given the last-synced `base` value. Returns which
    /// side wins (or a merged result).
    fn resolve(
        &self,
        local: &ChangeEntry,
        remote: &PulledChange,
        base: Option<&serde_json::Value>,
    ) -> Resolution;
}

/// Split pulled changes into non-conflicting and conflicting sets.
///
/// A conflict exists when a remote change and a local pending change both
/// modify the same `(table, row_id)`. Classification is by *contest*, not by the
/// server-asserted `device_id`: any pulled change that a local pending edit
/// contests is resolved, even one labeled as our own echo, trusting that label
/// would let a server relabel a hostile row as our echo to route it around
/// conflict detection. `our_device_id` is used only to log that spoof signal. A
/// pulled change with no contesting local edit is clean (a genuine echo, or a
/// remote change we hold no competing edit for); it is still HLC-gated at apply
/// time via [`CleanChanges::gated`].
///
/// **Precondition:** `local_pending` should contain at most one entry per
/// `(table, row_id)`. If duplicates exist, only the last one participates
/// in conflict detection (earlier entries are silently ignored). Callers
/// should compact local pending changes before calling this function.
///
/// If `remote` contains multiple changes for the same `(table, row_id)`,
/// each produces a separate `ConflictPair` with a clone of the same local
/// entry. The caller must resolve them in order (by `seq`).
///
/// Returns `(clean, conflicts)` where `clean` changes have no *un-pushed* local
/// edit to conflict with.
///
/// **Contract, clean changes still need HLC gating.** "Clean" means only that
/// no local *pending* edit contests the row. It does **not** mean the remote
/// change is newer than what is already committed locally: a device that has
/// applied and pushed a newer edit, then pulls an older remote edit for the same
/// row, will see that older edit in `clean`. The caller must compare each clean
/// change's [`Hlc`](crate::types::Hlc) against the row's committed HLC and skip
/// it if older. Applying clean changes blindly can clobber a newer local value.
pub fn detect_conflicts(
    remote: Vec<PulledChange>,
    local_pending: &[ChangeEntry],
    our_device_id: DeviceId,
) -> (CleanChanges, Vec<ConflictPair>) {
    // Build lookup: (table, row_id) -> last local pending entry.
    // If local_pending has duplicates for the same key, the last entry wins.
    let mut local_map: HashMap<(&str, &str), &ChangeEntry> = HashMap::new();
    for entry in local_pending {
        local_map.insert((&entry.table, &entry.row_id), entry);
    }

    let mut clean = Vec::new();
    let mut conflicts = Vec::new();

    for pulled in remote {
        let key = (pulled.entry.table.as_str(), pulled.entry.row_id.as_str());
        if let Some(&local_entry) = local_map.get(&key) {
            // A pending local edit contests this row. Resolve it regardless of the
            // asserted device_id: a change claiming to be our echo while contesting
            // an un-pushed local edit is either a benign re-pull of an in-flight
            // change or a server relabeling a hostile row to skip detection. Either
            // way, resolve rather than trust the label, a genuine echo resolves to
            // an identical value.
            if pulled.device_id == our_device_id {
                tracing::debug!(
                    table = %pulled.entry.table,
                    row_id = %pulled.entry.row_id,
                    "pulled change claims our device_id but contests a pending local edit; resolving as a conflict",
                );
            }
            conflicts.push(ConflictPair {
                remote: pulled,
                local: local_entry.clone(),
            });
        } else {
            clean.push(pulled);
        }
    }

    (CleanChanges(clean), conflicts)
}

/// Last-write-wins resolution by hybrid logical clock.
///
/// The higher [`Hlc`](crate::types::Hlc) wins, **regardless of operation**. Because
/// the HLC carries the originating device in its `node` component, every value is
/// globally unique and the comparison is convergent: device A resolving (local=a,
/// remote=b) and device B resolving (local=b, remote=a) compare the same two HLCs
/// and keep the same physical change.
///
/// This replaces the prior wall-clock + "delete always wins" rule. Delete-wins
/// caused permanent loss when a delete on one device beat a strictly-newer edit on
/// another (ultra-fuzz Run #28); under HLC ordering a newer edit beats an older
/// delete and vice-versa, by clock value alone.
///
/// An *exactly* equal HLC normally means a same-device echo, since a distinct
/// `node` makes the value unique. It can only collide across devices if two
/// installs share a `node` UUID (a cloned config dir or a restored backup). To
/// stay convergent even then, an exact tie is broken on the canonical payload
/// bytes, a deterministic order both devices compute identically, rather than
/// unconditionally keeping local (which would let A keep `a` while B keeps `b`).
pub fn resolve_lww(local: &ChangeEntry, remote: &PulledChange) -> Resolution {
    resolve_lww_at(local, remote, Utc::now())
}

/// Maximum forward wall-clock drift tolerated before an HLC is judged
/// clock-poisoned. An entry whose wall component is beyond `now + MAX_HLC_DRIFT_MS`
/// carries an implausible future timestamp, from a misconfigured clock or a
/// hostile peer holding the shared key, and must not be allowed to win LWW
/// against an honest entry. Set generously (5 min) so honest inter-device skew is
/// never flagged.
pub const MAX_HLC_DRIFT_MS: i64 = 5 * 60 * 1000;

/// Whether `hlc`'s wall clock is implausibly far in the future relative to `now`,
/// the clock-poisoning signal. A bare wall clock, and even an unclamped HLC,
/// lets such an entry win *every* conflict until real time catches up (years,
/// for a hostile timestamp); this predicate is how [`resolve_lww_at`] refuses to
/// let it. Exposed so an app driving its own apply loop can quarantine the same
/// entries at ingest.
pub fn is_clock_poisoned(hlc: &Hlc, now: DateTime<Utc>) -> bool {
    hlc.wall_ms > now.timestamp_millis().saturating_add(MAX_HLC_DRIFT_MS)
}

/// [`resolve_lww`] with an explicit `now`, so the poisoning guard is
/// deterministic and testable.
///
/// A poisoned entry (see [`is_clock_poisoned`]) *loses* to a non-poisoned one
/// regardless of raw HLC order, this is what stops a far-future clock from
/// winning forever. Merely capping the future value would not: capped at
/// `now + drift` it would perpetually track "now" and keep winning by the drift
/// margin. When both sides are poisoned, or neither is, the normal HLC order
/// applies. Two honest devices agree except in the narrow window where an entry
/// straddles the drift threshold by less than their clock skew, vastly better
/// than the unbounded, permanent domination an unguarded compare allows.
pub fn resolve_lww_at(
    local: &ChangeEntry,
    remote: &PulledChange,
    now: DateTime<Utc>,
) -> Resolution {
    match (
        is_clock_poisoned(&local.hlc, now),
        is_clock_poisoned(&remote.entry.hlc, now),
    ) {
        (false, true) => {
            tracing::warn!(
                remote_wall_ms = remote.entry.hlc.wall_ms,
                "remote HLC wall-clock is implausibly far in the future; rejecting it in LWW"
            );
            return Resolution::KeepLocal;
        }
        (true, false) => {
            tracing::warn!(
                local_wall_ms = local.hlc.wall_ms,
                "local HLC wall-clock is implausibly far in the future; letting remote win LWW"
            );
            return Resolution::KeepRemote;
        }
        _ => {}
    }
    let resolution = match resolve_tie(
        &local.hlc,
        &canonical_payload(local.data.as_ref()),
        &remote.entry.hlc,
        &canonical_payload(remote.entry.data.as_ref()),
    ) {
        Ordering::Less => Resolution::KeepRemote,
        // Greater or an exact tie both keep local: at a true tie the two changes
        // are byte-identical, so keeping either side converges.
        Ordering::Greater | Ordering::Equal => Resolution::KeepLocal,
    };
    // Every LWW resolution discards one side; log at debug so a lost edit is
    // diagnosable after the fact rather than vanishing silently.
    tracing::debug!(
        table = %local.table,
        row_id = %local.row_id,
        kept = ?resolution,
        "LWW resolved",
    );
    resolution
}

/// The single device-independent tiebreak both resolvers route through.
///
/// Orders two conflicting changes by full [`Hlc`] first, globally unique via its
/// `node`, so both devices compute the identical result, then breaks an *exact*
/// HLC tie on the canonical payload bytes, a value every device derives the same
/// way. [`Ordering::Greater`] means the `a` side wins.
///
/// One primitive is the point. The old field-merge rule broke ties on "local
/// wins" (device-relative → device A keeps `a`, device B keeps `b`, permanent
/// silent divergence) and on bare wall-`ms` (a tie window orders of magnitude
/// wider than the exact-HLC tie here). With both [`resolve_lww_at`] and
/// [`resolve_field_merge`] deferring to this, a device-relative winner rule has
/// no code path left.
fn resolve_tie(a_hlc: &Hlc, a_payload: &[u8], b_hlc: &Hlc, b_payload: &[u8]) -> Ordering {
    a_hlc.cmp(b_hlc).then_with(|| a_payload.cmp(b_payload))
}

/// Deterministic byte encoding of a concrete JSON value for the exact-HLC
/// tiebreak. `serde_json`'s default `Map` is sorted, so equal values always
/// serialize to equal bytes on every device, the property the tiebreak relies on.
///
/// INVARIANT: this convergence holds only while `serde_json` keeps insertion
/// order OFF (i.e. the `preserve_order` feature is NOT enabled anywhere in the
/// dependency tree). If it is ever turned on, map-key order becomes
/// insertion-dependent and two devices can compute different tiebreak bytes for
/// equal values, divergence. The `canonical_payload_sorts_map_keys` test pins
/// this (it fails the moment insertion order leaks in).
fn canonical_value(v: &serde_json::Value) -> Vec<u8> {
    serde_json::to_vec(v).unwrap_or_default()
}

/// [`canonical_value`] for an optional payload; `None` (a delete) canonicalizes
/// to empty bytes.
fn canonical_payload(data: Option<&serde_json::Value>) -> Vec<u8> {
    data.map(canonical_value).unwrap_or_default()
}

/// 3-way field-level merge for JSON objects (top-level keys only).
///
/// Compares `local` and `remote` against `base` to determine which fields
/// each side changed, then merges non-overlapping changes. For overlapping
/// fields, the newer timestamp wins.
///
/// The overlapping-field winner (and the no-base fallback) is decided by
/// [`resolve_tie`] over the two sides' HLCs, not by a device-relative "ties go to
/// local" rule, so two devices resolving the mirror image of the same conflict
/// converge on the same merged object. Pass each side's [`Hlc`] (`local_hlc` from
/// the local [`ChangeEntry`], `remote_hlc` from the pulled entry).
///
/// If any input is not a JSON object (including `Value::Null` for a missing base
/// snapshot), a field-level merge is impossible, so this falls back to
/// last-writer-wins on the HLCs rather than unconditionally keeping remote,
/// which would silently discard a strictly-newer local edit whenever no base is
/// available (e.g. a first-ever edit).
pub fn resolve_field_merge(
    local: &serde_json::Value,
    remote: &serde_json::Value,
    base: &serde_json::Value,
    local_hlc: &Hlc,
    remote_hlc: &Hlc,
) -> Resolution {
    // Device-independent winner of a contested field (and of the whole entry in
    // the no-base fallback): identical on every device, so the merge converges.
    // `>= Equal` keeps local; only a strict `Less` hands the field to remote.
    let local_wins = resolve_tie(
        local_hlc,
        &canonical_value(local),
        remote_hlc,
        &canonical_value(remote),
    ) != Ordering::Less;

    let (Some(local_obj), Some(remote_obj), Some(base_obj)) =
        (local.as_object(), remote.as_object(), base.as_object())
    else {
        // No usable base: last-writer-wins on the HLCs, keeping local when it wins
        // the tie instead of dropping it.
        tracing::debug!(
            local_wins,
            ?local_hlc,
            ?remote_hlc,
            "field-merge base unavailable; fell back to HLC LWW"
        );
        return if local_wins {
            Resolution::KeepLocal
        } else {
            Resolution::KeepRemote
        };
    };

    // Compute diffs: keys where local/remote differ from base
    let mut local_changed: HashMap<&str, Option<&serde_json::Value>> = HashMap::new();
    let mut remote_changed: HashMap<&str, Option<&serde_json::Value>> = HashMap::new();

    // Check keys in base for changes or deletions
    for key in base_obj.keys() {
        let base_val = &base_obj[key];

        match local_obj.get(key) {
            Some(local_val) if local_val != base_val => {
                local_changed.insert(key, Some(local_val));
            }
            None => {
                // Key deleted on local side
                local_changed.insert(key, None);
            }
            _ => {}
        }

        match remote_obj.get(key) {
            Some(remote_val) if remote_val != base_val => {
                remote_changed.insert(key, Some(remote_val));
            }
            None => {
                // Key deleted on remote side
                remote_changed.insert(key, None);
            }
            _ => {}
        }
    }

    // Check for new keys added by local (not in base)
    for (key, val) in local_obj {
        if !base_obj.contains_key(key) {
            local_changed.insert(key, Some(val));
        }
    }

    // Check for new keys added by remote (not in base)
    for (key, val) in remote_obj {
        if !base_obj.contains_key(key) {
            remote_changed.insert(key, Some(val));
        }
    }

    // Build merged result starting from base
    let mut result = base_obj.clone();

    // Apply local changes
    for (key, val) in &local_changed {
        match val {
            Some(v) => {
                result.insert((*key).to_string(), (*v).clone());
            }
            None => {
                result.remove(*key);
            }
        }
    }

    // Apply remote changes (non-overlapping only, or remote won the tie)
    for (key, val) in &remote_changed {
        if local_changed.contains_key(key) {
            // Overlapping: the device-independent winner decides every contested
            // field the same way. When remote wins, apply its value; when local
            // wins, local was already applied above.
            if !local_wins {
                match val {
                    Some(v) => {
                        result.insert((*key).to_string(), (*v).clone());
                    }
                    None => {
                        result.remove(*key);
                    }
                }
            }
        } else {
            // Non-overlapping: apply remote
            match val {
                Some(v) => {
                    result.insert((*key).to_string(), (*v).clone());
                }
                None => {
                    result.remove(*key);
                }
            }
        }
    }

    Resolution::Merged(serde_json::Value::Object(result))
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::types::{ChangeOp, Hlc};
    use serde_json::json;

    /// Fixed node for locally-minted test entries, distinct from any random
    /// `other_device`, so HLC tiebreaks are deterministic.
    fn local_node() -> DeviceId {
        DeviceId::new(Uuid::from_u128(0x1111_1111_1111_1111_1111_1111_1111_1111))
    }

    /// A second fixed device node, distinct from [`local_node`], for the
    /// field-merge tests that need to name the remote side's clock.
    fn remote_node() -> DeviceId {
        DeviceId::new(Uuid::from_u128(0x2222_2222_2222_2222_2222_2222_2222_2222))
    }

    /// Map a wall-clock timestamp onto an HLC at `node`, so the timestamp-ordered
    /// field-merge tests express the same intent against the HLC-based API. A
    /// strictly later `ts` yields a strictly greater HLC (higher `wall_ms`); equal
    /// `ts` on distinct nodes ties on the node, which is exactly the convergent
    /// behavior the F1 fix guarantees.
    fn ts_hlc(ts: DateTime<Utc>, node: DeviceId) -> Hlc {
        Hlc::from_legacy(ts.timestamp_millis(), node)
    }

    fn make_entry(table: &str, row_id: &str, op: ChangeOp, ts: DateTime<Utc>) -> ChangeEntry {
        // Derive the HLC wall component from the timestamp so the time-ordered
        // tests below still express the intended ordering.
        ChangeEntry {
            table: table.to_string(),
            op,
            row_id: row_id.to_string(),
            timestamp: ts,
            hlc: Hlc::from_legacy(ts.timestamp_millis(), local_node()),
            data: Some(json!({"value": "test"})),
            extra: serde_json::Map::default(),
        }
    }

    fn make_pulled(
        table: &str,
        row_id: &str,
        op: ChangeOp,
        ts: DateTime<Utc>,
        device_id: Uuid,
        seq: i64,
    ) -> PulledChange {
        let mut entry = make_entry(table, row_id, op, ts);
        entry.hlc = Hlc::from_legacy(ts.timestamp_millis(), DeviceId::new(device_id));
        PulledChange {
            entry,
            device_id: DeviceId::new(device_id),
            seq,
        }
    }

    /// Build a pulled change with an explicit HLC, for resolution tests that need
    /// to control the clock independently of the wall timestamp.
    fn pulled_with_hlc(row_id: &str, op: ChangeOp, hlc: Hlc, device_id: Uuid) -> PulledChange {
        let mut p = make_pulled("tasks", row_id, op, Utc::now(), device_id, 1);
        p.entry.hlc = hlc;
        p
    }

    // ── detect_conflicts ──

    #[test]
    fn no_conflicts_when_different_rows() {
        let our_device = Uuid::new_v4();
        let other_device = Uuid::new_v4();
        let now = Utc::now();

        let remote = vec![make_pulled(
            "tasks",
            "r1",
            ChangeOp::Update,
            now,
            other_device,
            1,
        )];
        let local = vec![make_entry("tasks", "r2", ChangeOp::Update, now)];

        let (clean, conflicts) = detect_conflicts(remote, &local, DeviceId::new(our_device));
        assert_eq!(clean.len(), 1);
        assert!(conflicts.is_empty());
    }

    #[test]
    fn conflict_detected_same_row_different_device() {
        let our_device = Uuid::new_v4();
        let other_device = Uuid::new_v4();
        let now = Utc::now();

        let remote = vec![make_pulled(
            "tasks",
            "r1",
            ChangeOp::Update,
            now,
            other_device,
            1,
        )];
        let local = vec![make_entry("tasks", "r1", ChangeOp::Update, now)];

        let (clean, conflicts) = detect_conflicts(remote, &local, DeviceId::new(our_device));
        assert!(clean.is_empty());
        assert_eq!(conflicts.len(), 1);
        assert_eq!(conflicts[0].remote.entry.row_id, "r1");
        assert_eq!(conflicts[0].local.row_id, "r1");
    }

    #[test]
    fn own_echo_without_pending_edit_is_clean() {
        // An echo of our own device with no contesting local pending edit is
        // clean (it still passes the HLC gate at apply time).
        let our_device = Uuid::new_v4();
        let now = Utc::now();

        let remote = vec![make_pulled(
            "tasks",
            "r1",
            ChangeOp::Update,
            now,
            our_device,
            1,
        )];
        let (clean, conflicts) = detect_conflicts(remote, &[], DeviceId::new(our_device));
        assert_eq!(clean.len(), 1);
        assert!(conflicts.is_empty());
    }

    #[test]
    fn echo_contesting_a_pending_edit_is_resolved_not_trusted() {
        // Hardening: a pulled change labeled as our own echo that contests an
        // un-pushed local edit is resolved as a conflict, not waved through as
        // clean. Trusting the device_id label would let a server relabel a hostile
        // row as our echo to skip conflict detection entirely.
        let our_device = Uuid::new_v4();
        let now = Utc::now();

        let remote = vec![make_pulled(
            "tasks",
            "r1",
            ChangeOp::Update,
            now,
            our_device,
            1,
        )];
        let local = vec![make_entry("tasks", "r1", ChangeOp::Update, now)];

        let (clean, conflicts) = detect_conflicts(remote, &local, DeviceId::new(our_device));
        assert!(clean.is_empty());
        assert_eq!(
            conflicts.len(),
            1,
            "echo contesting a pending edit is resolved"
        );
    }

    #[test]
    fn clean_changes_gate_drops_stale_keeps_newer() {
        let our_device = Uuid::new_v4();
        let other_device = Uuid::new_v4();
        let now = Utc::now();
        // A clean remote change for tasks/r1; its HLC wall == now_ms.
        let remote = vec![make_pulled(
            "tasks",
            "r1",
            ChangeOp::Update,
            now,
            other_device,
            1,
        )];
        let (clean, _conflicts) = detect_conflicts(remote, &[], DeviceId::new(our_device));
        assert_eq!(clean.len(), 1);

        // No committed clock for the row → kept (first time we've seen it).
        assert_eq!(clean.clone().gated(|_, _| None).len(), 1);
        // Committed clock older than the remote → kept.
        assert_eq!(
            clean
                .clone()
                .gated(|_, _| Some(Hlc::zero(DeviceId::new(other_device))))
                .len(),
            1
        );
        // Committed clock newer than the remote → dropped (would clobber newer local).
        let newer = Hlc {
            wall_ms: now.timestamp_millis() + 1,
            counter: 0,
            node: DeviceId::new(other_device),
        };
        assert!(clean.gated(move |_, _| Some(newer)).is_empty());
    }

    #[test]
    fn different_tables_same_row_id_no_conflict() {
        let our_device = Uuid::new_v4();
        let other_device = Uuid::new_v4();
        let now = Utc::now();

        let remote = vec![make_pulled(
            "tasks",
            "r1",
            ChangeOp::Update,
            now,
            other_device,
            1,
        )];
        let local = vec![make_entry("events", "r1", ChangeOp::Update, now)];

        let (clean, conflicts) = detect_conflicts(remote, &local, DeviceId::new(our_device));
        assert_eq!(clean.len(), 1);
        assert!(conflicts.is_empty());
    }

    #[test]
    fn detect_conflicts_correct_split() {
        let our_device = Uuid::new_v4();
        let other_device = Uuid::new_v4();
        let now = Utc::now();

        let remote = vec![
            make_pulled("tasks", "r1", ChangeOp::Update, now, other_device, 1),
            make_pulled("tasks", "r2", ChangeOp::Insert, now, other_device, 2),
            make_pulled("events", "r3", ChangeOp::Delete, now, other_device, 3),
        ];
        let local = vec![
            make_entry("tasks", "r1", ChangeOp::Update, now),
            // r2 not in local → clean
            // r3 not in local → clean
        ];

        let (clean, conflicts) = detect_conflicts(remote, &local, DeviceId::new(our_device));
        assert_eq!(clean.len(), 2);
        assert_eq!(conflicts.len(), 1);
        assert_eq!(conflicts[0].remote.entry.row_id, "r1");
    }

    #[test]
    fn empty_remote_produces_no_conflicts() {
        let our_device = Uuid::new_v4();
        let now = Utc::now();

        let remote = vec![];
        let local = vec![make_entry("tasks", "r1", ChangeOp::Update, now)];

        let (clean, conflicts) = detect_conflicts(remote, &local, DeviceId::new(our_device));
        assert!(clean.is_empty());
        assert!(conflicts.is_empty());
    }

    #[test]
    fn empty_local_produces_no_conflicts() {
        let our_device = Uuid::new_v4();
        let other_device = Uuid::new_v4();
        let now = Utc::now();

        let remote = vec![make_pulled(
            "tasks",
            "r1",
            ChangeOp::Update,
            now,
            other_device,
            1,
        )];
        let local: Vec<ChangeEntry> = vec![];

        let (clean, conflicts) = detect_conflicts(remote, &local, DeviceId::new(our_device));
        assert_eq!(clean.len(), 1);
        assert!(conflicts.is_empty());
    }

    // ── resolve_lww (HLC ordering) ──

    #[test]
    fn lww_picks_newer_timestamp() {
        let other_device = Uuid::new_v4();
        let old = Utc::now() - chrono::Duration::seconds(60);
        let new = Utc::now();

        let local = make_entry("tasks", "r1", ChangeOp::Update, old);
        let remote = make_pulled("tasks", "r1", ChangeOp::Update, new, other_device, 1);

        assert!(matches!(
            resolve_lww(&local, &remote),
            Resolution::KeepRemote
        ));
    }

    #[test]
    fn lww_local_wins_when_newer() {
        let other_device = Uuid::new_v4();
        let old = Utc::now() - chrono::Duration::seconds(60);
        let new = Utc::now();

        let local = make_entry("tasks", "r1", ChangeOp::Update, new);
        let remote = make_pulled("tasks", "r1", ChangeOp::Update, old, other_device, 1);

        assert!(matches!(
            resolve_lww(&local, &remote),
            Resolution::KeepLocal
        ));
    }

    #[test]
    fn lww_counter_breaks_same_wall_tie() {
        // Same wall_ms, higher counter wins regardless of node.
        let other = Uuid::new_v4();
        let mut local = make_entry("tasks", "r1", ChangeOp::Update, Utc::now());
        local.hlc = Hlc {
            wall_ms: 1000,
            counter: 2,
            node: local_node(),
        };
        let remote = pulled_with_hlc(
            "r1",
            ChangeOp::Update,
            Hlc {
                wall_ms: 1000,
                counter: 5,
                node: DeviceId::new(other),
            },
            other,
        );
        assert!(matches!(
            resolve_lww(&local, &remote),
            Resolution::KeepRemote
        ));
    }

    #[test]
    fn lww_node_breaks_exact_tie_convergently() {
        // Identical (wall, counter): the node decides, and BOTH devices must pick
        // the same physical change. Use nodes with a known order (a < b).
        let a = Uuid::from_u128(1);
        let b = Uuid::from_u128(2);
        let hlc_a = Hlc {
            wall_ms: 1000,
            counter: 0,
            node: DeviceId::new(a),
        };
        let hlc_b = Hlc {
            wall_ms: 1000,
            counter: 0,
            node: DeviceId::new(b),
        };

        // Device A: local is a's change, remote is b's change.
        let mut local_a = make_entry("tasks", "r1", ChangeOp::Update, Utc::now());
        local_a.hlc = hlc_a;
        let remote_b = pulled_with_hlc("r1", ChangeOp::Update, hlc_b, b);
        // b > a, so A drops its local and keeps remote (b's change).
        assert!(matches!(
            resolve_lww(&local_a, &remote_b),
            Resolution::KeepRemote
        ));

        // Device B: local is b's change, remote is a's change.
        let mut local_b = make_entry("tasks", "r1", ChangeOp::Update, Utc::now());
        local_b.hlc = hlc_b;
        let remote_a = pulled_with_hlc("r1", ChangeOp::Update, hlc_a, a);
        // b > a, so B keeps its local (b's change). Both devices converge on b.
        assert!(matches!(
            resolve_lww(&local_b, &remote_a),
            Resolution::KeepLocal
        ));
    }

    #[test]
    fn lww_identical_hlc_node_collision_converges_on_payload() {
        // Pathological case: two installs share a node UUID (cloned config), so
        // two genuinely different edits produce a byte-identical HLC. The payload
        // tiebreak must still make both devices land on the same value.
        let shared = Uuid::from_u128(7);
        let hlc = Hlc {
            wall_ms: 1000,
            counter: 3,
            node: DeviceId::new(shared),
        };
        let val_a = json!({"v": "aaa"});
        let val_b = json!({"v": "bbb"}); // canonically greater than val_a

        // Device A: local = a, remote = b.
        let mut local_a = make_entry("tasks", "r1", ChangeOp::Update, Utc::now());
        local_a.hlc = hlc;
        local_a.data = Some(val_a.clone());
        let mut remote_b = pulled_with_hlc("r1", ChangeOp::Update, hlc, shared);
        remote_b.entry.data = Some(val_b.clone());
        // b's payload sorts higher, so A drops local and keeps remote (b).
        assert!(matches!(
            resolve_lww(&local_a, &remote_b),
            Resolution::KeepRemote
        ));

        // Device B: local = b, remote = a, must keep local (b). Both converge on b.
        let mut local_b = make_entry("tasks", "r1", ChangeOp::Update, Utc::now());
        local_b.hlc = hlc;
        local_b.data = Some(val_b);
        let mut remote_a = pulled_with_hlc("r1", ChangeOp::Update, hlc, shared);
        remote_a.entry.data = Some(val_a);
        assert!(matches!(
            resolve_lww(&local_b, &remote_a),
            Resolution::KeepLocal
        ));
    }

    #[test]
    fn lww_newer_update_beats_older_delete() {
        // The data-loss fix: a strictly-newer UPDATE must beat an older DELETE.
        // Under the old "delete always wins" rule this edit was silently lost.
        let other = Uuid::new_v4();
        let mut local = make_entry("tasks", "r1", ChangeOp::Update, Utc::now());
        local.hlc = Hlc {
            wall_ms: 2000,
            counter: 0,
            node: local_node(),
        };
        let remote = pulled_with_hlc(
            "r1",
            ChangeOp::Delete,
            Hlc {
                wall_ms: 1000,
                counter: 0,
                node: DeviceId::new(other),
            },
            other,
        );
        assert!(matches!(
            resolve_lww(&local, &remote),
            Resolution::KeepLocal
        ));
    }

    #[test]
    fn lww_newer_delete_beats_older_update() {
        // Symmetric: a strictly-newer DELETE beats an older UPDATE.
        let other = Uuid::new_v4();
        let mut local = make_entry("tasks", "r1", ChangeOp::Update, Utc::now());
        local.hlc = Hlc {
            wall_ms: 1000,
            counter: 0,
            node: local_node(),
        };
        let remote = pulled_with_hlc(
            "r1",
            ChangeOp::Delete,
            Hlc {
                wall_ms: 2000,
                counter: 0,
                node: DeviceId::new(other),
            },
            other,
        );
        assert!(matches!(
            resolve_lww(&local, &remote),
            Resolution::KeepRemote
        ));
    }

    // ── resolve_field_merge ──

    #[test]
    fn field_merge_non_overlapping_changes() {
        let base = json!({"title": "old", "status": "pending", "priority": 1});
        let local = json!({"title": "new title", "status": "pending", "priority": 1});
        let remote = json!({"title": "old", "status": "done", "priority": 1});
        let now = Utc::now();

        let result = resolve_field_merge(
            &local,
            &remote,
            &base,
            &ts_hlc(now, local_node()),
            &ts_hlc(now, remote_node()),
        );
        match result {
            Resolution::Merged(v) => {
                assert_eq!(v["title"], "new title");
                assert_eq!(v["status"], "done");
                assert_eq!(v["priority"], 1);
            }
            _ => panic!("Expected Merged"),
        }
    }

    #[test]
    fn field_merge_overlapping_newer_wins() {
        let base = json!({"title": "old"});
        let local = json!({"title": "local title"});
        let remote = json!({"title": "remote title"});
        let old = Utc::now() - chrono::Duration::seconds(60);
        let new = Utc::now();

        // Remote is newer → remote wins the overlapping field
        let result = resolve_field_merge(
            &local,
            &remote,
            &base,
            &ts_hlc(old, local_node()),
            &ts_hlc(new, remote_node()),
        );
        match result {
            Resolution::Merged(v) => {
                assert_eq!(v["title"], "remote title");
            }
            _ => panic!("Expected Merged"),
        }

        // Local is newer → local wins the overlapping field
        let result = resolve_field_merge(
            &local,
            &remote,
            &base,
            &ts_hlc(new, local_node()),
            &ts_hlc(old, remote_node()),
        );
        match result {
            Resolution::Merged(v) => {
                assert_eq!(v["title"], "local title");
            }
            _ => panic!("Expected Merged"),
        }
    }

    #[test]
    fn field_merge_key_deleted_on_one_side() {
        let base = json!({"title": "old", "notes": "some notes"});
        let local = json!({"title": "old"}); // notes deleted
        let remote = json!({"title": "old", "notes": "some notes"});
        let now = Utc::now();

        let result = resolve_field_merge(
            &local,
            &remote,
            &base,
            &ts_hlc(now, local_node()),
            &ts_hlc(now, remote_node()),
        );
        match result {
            Resolution::Merged(v) => {
                assert_eq!(v["title"], "old");
                assert!(v.get("notes").is_none(), "notes should be deleted");
            }
            _ => panic!("Expected Merged"),
        }
    }

    #[test]
    fn field_merge_non_object_falls_back() {
        let base = json!("string value");
        let local = json!("local string");
        let remote = json!("remote string");
        let now = Utc::now();

        assert!(matches!(
            resolve_field_merge(
                &local,
                &remote,
                &base,
                &ts_hlc(now, local_node()),
                &ts_hlc(now, remote_node())
            ),
            Resolution::KeepRemote
        ));
    }

    #[test]
    fn field_merge_empty_base_treats_all_as_changed() {
        let base = json!({});
        let local = json!({"title": "from local"});
        let remote = json!({"status": "from remote"});
        let now = Utc::now();

        let result = resolve_field_merge(
            &local,
            &remote,
            &base,
            &ts_hlc(now, local_node()),
            &ts_hlc(now, remote_node()),
        );
        match result {
            Resolution::Merged(v) => {
                assert_eq!(v["title"], "from local");
                assert_eq!(v["status"], "from remote");
            }
            _ => panic!("Expected Merged"),
        }
    }

    #[test]
    fn field_merge_new_keys_from_both_sides() {
        let base = json!({"existing": 1});
        let local = json!({"existing": 1, "local_new": "a"});
        let remote = json!({"existing": 1, "remote_new": "b"});
        let now = Utc::now();

        let result = resolve_field_merge(
            &local,
            &remote,
            &base,
            &ts_hlc(now, local_node()),
            &ts_hlc(now, remote_node()),
        );
        match result {
            Resolution::Merged(v) => {
                assert_eq!(v["existing"], 1);
                assert_eq!(v["local_new"], "a");
                assert_eq!(v["remote_new"], "b");
            }
            _ => panic!("Expected Merged"),
        }
    }

    // ── PulledChange preserves metadata ──

    #[test]
    fn pulled_change_preserves_device_id_and_seq() {
        let device_id = Uuid::new_v4();
        let now = Utc::now();
        let pulled = make_pulled("tasks", "r1", ChangeOp::Insert, now, device_id, 42);

        assert_eq!(pulled.device_id.as_uuid(), device_id);
        assert_eq!(pulled.seq, 42);
        assert_eq!(pulled.entry.table, "tasks");
        assert_eq!(pulled.entry.row_id, "r1");
    }

    #[test]
    fn pulled_change_clone_works() {
        let device_id = Uuid::new_v4();
        let now = Utc::now();
        let pulled = make_pulled("tasks", "r1", ChangeOp::Insert, now, device_id, 1);
        let cloned = pulled.clone();

        assert_eq!(cloned.device_id, pulled.device_id);
        assert_eq!(cloned.seq, pulled.seq);
        assert_eq!(cloned.entry.table, pulled.entry.table);
    }

    // ── Resolution variants ──

    #[test]
    fn resolution_debug_format() {
        let keep_local = Resolution::KeepLocal;
        let keep_remote = Resolution::KeepRemote;
        let merged = Resolution::Merged(json!({"a": 1}));
        let skip = Resolution::Skip;

        assert!(format!("{keep_local:?}").contains("KeepLocal"));
        assert!(format!("{keep_remote:?}").contains("KeepRemote"));
        assert!(format!("{merged:?}").contains("Merged"));
        assert!(format!("{skip:?}").contains("Skip"));
    }

    // ── ConflictResolver trait ──

    #[test]
    fn custom_resolver_works() {
        struct AlwaysRemote;
        impl ConflictResolver for AlwaysRemote {
            fn resolve(
                &self,
                _local: &ChangeEntry,
                _remote: &PulledChange,
                _base: Option<&serde_json::Value>,
            ) -> Resolution {
                Resolution::KeepRemote
            }
        }

        let resolver = AlwaysRemote;
        let now = Utc::now();
        let local = make_entry("tasks", "r1", ChangeOp::Update, now);
        let remote = make_pulled("tasks", "r1", ChangeOp::Update, now, Uuid::new_v4(), 1);

        assert!(matches!(
            resolver.resolve(&local, &remote, None),
            Resolution::KeepRemote
        ));
    }

    // ── Fuzz: edge cases and attack vectors ──

    // Attack vector 1: duplicate local entries for same (table, row_id).
    // HashMap insert means last entry wins. Verify the conflict pair uses
    // the LATER local entry, not the earlier one.
    #[test]
    fn detect_conflicts_duplicate_local_uses_last_entry() {
        let our_device = Uuid::new_v4();
        let other_device = Uuid::new_v4();
        let t1 = Utc::now() - chrono::Duration::seconds(60);
        let t2 = Utc::now();

        let remote = vec![make_pulled(
            "tasks",
            "r1",
            ChangeOp::Update,
            t2,
            other_device,
            1,
        )];
        // Two local entries for same (table, row_id): Insert then Update.
        // The Update (last) should participate in conflict detection.
        let local = vec![
            make_entry("tasks", "r1", ChangeOp::Insert, t1),
            make_entry("tasks", "r1", ChangeOp::Update, t2),
        ];

        let (_clean, conflicts) = detect_conflicts(remote, &local, DeviceId::new(our_device));
        assert_eq!(conflicts.len(), 1);
        // The conflict should use the Update (last entry), not the Insert
        assert_eq!(conflicts[0].local.op, ChangeOp::Update);
        assert_eq!(conflicts[0].local.timestamp, t2);
    }

    // DELETE vs DELETE under HLC: the higher clock wins like any other pair.
    // (Previously local always won and the timestamp was ignored, now a strictly
    // newer remote delete wins.)
    #[test]
    fn lww_both_delete_newer_wins() {
        let other_device = Uuid::new_v4();
        let old = Utc::now() - chrono::Duration::seconds(60);
        let new = Utc::now();

        let local = make_entry("tasks", "r1", ChangeOp::Delete, old);
        let remote = make_pulled("tasks", "r1", ChangeOp::Delete, new, other_device, 1);

        assert!(matches!(
            resolve_lww(&local, &remote),
            Resolution::KeepRemote
        ));
    }

    // F1 (regression): overlapping fields whose two sides carry the *same*
    // wall-ms must resolve convergently, not "ties go to local". The old rule
    // broke ties on local, so device A (local=A) kept A while device B (local=B)
    // kept B, permanent silent divergence. Now the exact tie breaks on the full
    // HLC (distinct node), so both devices land on the same physical value.
    #[test]
    fn field_merge_overlapping_equal_wall_converges() {
        let base = json!({"title": "base"});
        let val_a = json!({"title": "device A version"});
        let val_b = json!({"title": "device B version"});
        let now = Utc::now();
        let hlc_a = ts_hlc(now, local_node()); // node 0x111...
        let hlc_b = ts_hlc(now, remote_node()); // node 0x222... (> a), same wall

        // Device A resolves (local = A's edit, remote = B's edit); device B
        // resolves the mirror image (local = B's edit, remote = A's edit).
        let on_a = resolve_field_merge(&val_a, &val_b, &base, &hlc_a, &hlc_b);
        let on_b = resolve_field_merge(&val_b, &val_a, &base, &hlc_b, &hlc_a);

        let (title_a, title_b) = match (on_a, on_b) {
            (Resolution::Merged(a), Resolution::Merged(b)) => {
                (a["title"].clone(), b["title"].clone())
            }
            _ => panic!("Expected Merged on both devices"),
        };
        assert_eq!(
            title_a, title_b,
            "both devices must converge on the same value at an equal-wall tie"
        );
        // Specifically on B's edit, since hlc_b > hlc_a on the node tiebreak.
        assert_eq!(title_a, "device B version");
    }

    // Attack vector 4: HashMap iteration order determinism.
    // Non-overlapping keys from both sides should merge deterministically
    // regardless of HashMap iteration order.
    #[test]
    fn field_merge_deterministic_with_many_keys() {
        let base = json!({});
        let local = json!({"a": 1, "c": 3, "e": 5, "g": 7, "i": 9});
        let remote = json!({"b": 2, "d": 4, "f": 6, "h": 8, "j": 10});
        let now = Utc::now();

        // Run multiple times to catch iteration-order bugs
        for _ in 0..10 {
            let result = resolve_field_merge(
                &local,
                &remote,
                &base,
                &ts_hlc(now, local_node()),
                &ts_hlc(now, remote_node()),
            );
            match result {
                Resolution::Merged(v) => {
                    assert_eq!(v["a"], 1);
                    assert_eq!(v["b"], 2);
                    assert_eq!(v["c"], 3);
                    assert_eq!(v["d"], 4);
                    assert_eq!(v["e"], 5);
                    assert_eq!(v["f"], 6);
                    assert_eq!(v["g"], 7);
                    assert_eq!(v["h"], 8);
                    assert_eq!(v["i"], 9);
                    assert_eq!(v["j"], 10);
                }
                _ => panic!("Expected Merged"),
            }
        }
    }

    // Attack vector 5: null base with both local and remote as objects.
    // Falls back to KeepRemote, silently discarding local changes.
    #[test]
    fn field_merge_null_base_discards_local() {
        let base = json!(null);
        let local = json!({"title": "important local edit"});
        let remote = json!({"status": "remote only"});
        let now = Utc::now();

        // BUG: Both sides are valid objects but null base causes KeepRemote,
        // which silently drops "title": "important local edit".
        // A better fallback might be to merge both against an empty base,
        // or fall back to LWW.
        let result = resolve_field_merge(
            &local,
            &remote,
            &base,
            &ts_hlc(now, local_node()),
            &ts_hlc(now, remote_node()),
        );
        assert!(
            matches!(result, Resolution::KeepRemote),
            "Null base should fall back to KeepRemote (current behavior)"
        );
    }

    // Attack vector 6: empty object vs changed value in field_merge.
    // Both `{}` and a new value differ from the base value, so both are
    // "changed". This is an overlapping-field conflict.
    #[test]
    fn field_merge_empty_object_counts_as_change() {
        let base = json!({"meta": {"nested": "data"}});
        let local = json!({"meta": {}}); // Changed to empty object
        let remote = json!({"meta": "flat string"}); // Changed to string
        let old = Utc::now() - chrono::Duration::seconds(60);
        let new = Utc::now();

        // Remote is newer, so remote wins the overlapping field
        let result = resolve_field_merge(
            &local,
            &remote,
            &base,
            &ts_hlc(old, local_node()),
            &ts_hlc(new, remote_node()),
        );
        match result {
            Resolution::Merged(v) => {
                assert_eq!(v["meta"], "flat string");
            }
            _ => panic!("Expected Merged"),
        }

        // Local is newer, so local wins, meta becomes empty object
        let result = resolve_field_merge(
            &local,
            &remote,
            &base,
            &ts_hlc(new, local_node()),
            &ts_hlc(old, remote_node()),
        );
        match result {
            Resolution::Merged(v) => {
                assert_eq!(v["meta"], json!({}));
            }
            _ => panic!("Expected Merged"),
        }
    }

    // Attack vector 7: multiple remote changes for same (table, row_id).
    // Each produces a separate ConflictPair with a CLONE of the same local entry.
    #[test]
    fn detect_conflicts_multiple_remote_same_row() {
        let our_device = Uuid::new_v4();
        let other_device = Uuid::new_v4();
        let now = Utc::now();

        let remote = vec![
            make_pulled("tasks", "r1", ChangeOp::Update, now, other_device, 1),
            make_pulled("tasks", "r1", ChangeOp::Update, now, other_device, 2),
            make_pulled("tasks", "r1", ChangeOp::Delete, now, other_device, 3),
        ];
        let local = vec![make_entry("tasks", "r1", ChangeOp::Update, now)];

        let (_clean, conflicts) = detect_conflicts(remote, &local, DeviceId::new(our_device));
        // All 3 remote changes conflict with the same local entry
        assert_eq!(conflicts.len(), 3);
        // Each gets a clone of the same local entry
        assert_eq!(conflicts[0].local.row_id, "r1");
        assert_eq!(conflicts[1].local.row_id, "r1");
        assert_eq!(conflicts[2].local.row_id, "r1");
        // Verify seq ordering is preserved
        assert_eq!(conflicts[0].remote.seq, 1);
        assert_eq!(conflicts[1].remote.seq, 2);
        assert_eq!(conflicts[2].remote.seq, 3);
    }

    // Attack vector 8: numeric type coercion in serde_json PartialEq.
    // JSON `1` (u64) and `1.0` (f64) are different Value variants.
    // serde_json::Value PartialEq does NOT treat them as equal.
    #[test]
    fn field_merge_integer_vs_float_treated_as_different() {
        let base = json!({"count": 1}); // serde_json: Number(PosInt(1))
        let local = json!({"count": 1.0}); // serde_json: Number(Float(1.0))
        let remote = json!({"count": 1}); // unchanged from base
        let now = Utc::now();

        let result = resolve_field_merge(
            &local,
            &remote,
            &base,
            &ts_hlc(now, local_node()),
            &ts_hlc(now, remote_node()),
        );
        match result {
            Resolution::Merged(v) => {
                // BUG CANDIDATE: `1` != `1.0` in serde_json, so local sees
                // "count" as changed (1 -> 1.0) even though semantically
                // identical. Remote sees no change. Merge applies local's 1.0.
                assert_eq!(
                    v["count"], 1.0,
                    "serde_json treats 1 and 1.0 as different values"
                );
            }
            _ => panic!("Expected Merged"),
        }
    }

    // INSERT vs DELETE under HLC: operation is irrelevant, the higher clock wins.
    // A strictly-newer remote delete beats an older local insert.
    #[test]
    fn lww_newer_remote_delete_beats_older_local_insert() {
        let other_device = Uuid::new_v4();
        let old = Utc::now() - chrono::Duration::seconds(60);
        let new = Utc::now();

        let local = make_entry("tasks", "r1", ChangeOp::Insert, old);
        let remote = make_pulled("tasks", "r1", ChangeOp::Delete, new, other_device, 1);

        assert!(matches!(
            resolve_lww(&local, &remote),
            Resolution::KeepRemote
        ));
    }

    // Reverse: a strictly-newer local delete beats an older remote insert.
    #[test]
    fn lww_newer_local_delete_beats_older_remote_insert() {
        let other_device = Uuid::new_v4();
        let old = Utc::now() - chrono::Duration::seconds(60);
        let new = Utc::now();

        let local = make_entry("tasks", "r1", ChangeOp::Delete, new);
        let remote = make_pulled("tasks", "r1", ChangeOp::Insert, old, other_device, 1);

        assert!(matches!(
            resolve_lww(&local, &remote),
            Resolution::KeepLocal
        ));
    }

    // Bonus: field_merge where both sides delete the same key.
    // Both detect the key as deleted. Local applies deletion first.
    // Remote sees it as overlapping but equal-ts, so local's deletion stands.
    #[test]
    fn field_merge_both_delete_same_key() {
        let base = json!({"title": "old", "notes": "old notes"});
        let local = json!({"title": "old"}); // deleted "notes"
        let remote = json!({"title": "old"}); // also deleted "notes"
        let now = Utc::now();

        let result = resolve_field_merge(
            &local,
            &remote,
            &base,
            &ts_hlc(now, local_node()),
            &ts_hlc(now, remote_node()),
        );
        match result {
            Resolution::Merged(v) => {
                assert_eq!(v["title"], "old");
                assert!(
                    v.get("notes").is_none(),
                    "Both sides deleted notes, should stay deleted"
                );
            }
            _ => panic!("Expected Merged"),
        }
    }

    // field_merge where local deletes a key and remote modifies it. The
    // higher-HLC side wins the contested field: a strictly-newer local delete
    // beats an older remote modify, so the key stays deleted.
    #[test]
    fn field_merge_newer_local_delete_beats_remote_modify() {
        let base = json!({"title": "old", "notes": "original"});
        let local = json!({"title": "old"}); // deleted "notes"
        let remote = json!({"title": "old", "notes": "updated notes"});
        let old = Utc::now() - chrono::Duration::seconds(60);
        let new = Utc::now();

        // Local strictly newer → its delete wins the overlapping "notes" field.
        let result = resolve_field_merge(
            &local,
            &remote,
            &base,
            &ts_hlc(new, local_node()),
            &ts_hlc(old, remote_node()),
        );
        match result {
            Resolution::Merged(v) => {
                assert!(
                    v.get("notes").is_none(),
                    "newer local delete should win over older remote modify"
                );
            }
            _ => panic!("Expected Merged"),
        }
    }

    // Bonus: field_merge where both sides add the SAME new key with
    // different values. Both are in local_changed and remote_changed.
    #[test]
    fn field_merge_both_add_same_new_key_different_values() {
        let base = json!({"existing": 1});
        let local = json!({"existing": 1, "new_key": "local value"});
        let remote = json!({"existing": 1, "new_key": "remote value"});
        let old = Utc::now() - chrono::Duration::seconds(60);
        let new = Utc::now();

        // Remote newer: remote wins the overlapping new key
        let result = resolve_field_merge(
            &local,
            &remote,
            &base,
            &ts_hlc(old, local_node()),
            &ts_hlc(new, remote_node()),
        );
        match result {
            Resolution::Merged(v) => {
                assert_eq!(v["new_key"], "remote value");
            }
            _ => panic!("Expected Merged"),
        }

        // Local newer: local wins the overlapping new key
        let result = resolve_field_merge(
            &local,
            &remote,
            &base,
            &ts_hlc(new, local_node()),
            &ts_hlc(old, remote_node()),
        );
        match result {
            Resolution::Merged(v) => {
                assert_eq!(v["new_key"], "local value");
            }
            _ => panic!("Expected Merged"),
        }

        // Equal wall: the winner is decided by the HLC node tiebreak, not "local
        // wins", and both devices resolving the mirror image agree on it.
        let ha = ts_hlc(new, local_node());
        let hb = ts_hlc(new, remote_node());
        let on_a = resolve_field_merge(&local, &remote, &base, &ha, &hb);
        let on_b = resolve_field_merge(&remote, &local, &base, &hb, &ha);
        match (on_a, on_b) {
            (Resolution::Merged(a), Resolution::Merged(b)) => {
                assert_eq!(
                    a["new_key"], b["new_key"],
                    "both devices converge on the same new_key"
                );
            }
            _ => panic!("Expected Merged on both devices"),
        }
    }

    // 50 years in milliseconds, far beyond MAX_HLC_DRIFT_MS.
    const FIFTY_YEARS_MS: i64 = 50i64 * 365 * 24 * 3600 * 1000;

    #[test]
    fn lww_rejects_poisoned_remote_for_honest_local() {
        let now = Utc::now();
        let now_ms = now.timestamp_millis();
        let mut local = make_entry("tasks", "r1", ChangeOp::Update, now);
        local.hlc = Hlc {
            wall_ms: now_ms,
            counter: 0,
            node: local_node(),
        };
        let mut pulled = make_pulled("tasks", "r1", ChangeOp::Update, now, Uuid::new_v4(), 1);
        // Unguarded, a wall clock 50 years ahead would win every conflict for decades.
        pulled.entry.hlc = Hlc {
            wall_ms: now_ms + FIFTY_YEARS_MS,
            counter: 0,
            node: DeviceId::new(Uuid::new_v4()),
        };
        assert!(is_clock_poisoned(&pulled.entry.hlc, now));
        assert!(matches!(
            resolve_lww_at(&local, &pulled, now),
            Resolution::KeepLocal
        ));
    }

    #[test]
    fn lww_rejects_poisoned_local_for_honest_remote() {
        let now = Utc::now();
        let now_ms = now.timestamp_millis();
        let mut local = make_entry("tasks", "r1", ChangeOp::Update, now);
        local.hlc = Hlc {
            wall_ms: now_ms + FIFTY_YEARS_MS,
            counter: 0,
            node: local_node(),
        };
        let mut pulled = make_pulled("tasks", "r1", ChangeOp::Update, now, Uuid::new_v4(), 1);
        pulled.entry.hlc = Hlc {
            wall_ms: now_ms,
            counter: 0,
            node: DeviceId::new(Uuid::new_v4()),
        };
        assert!(matches!(
            resolve_lww_at(&local, &pulled, now),
            Resolution::KeepRemote
        ));
    }

    #[test]
    fn field_merge_null_base_falls_back_to_lww_not_keep_remote() {
        let local = json!({"a": 1});
        let remote = json!({"a": 2});
        let base = serde_json::Value::Null;
        let local_ts = Utc::now();
        // Local strictly newer: must be kept, not silently discarded (the old bug).
        let remote_older = local_ts - chrono::Duration::seconds(10);
        assert!(matches!(
            resolve_field_merge(
                &local,
                &remote,
                &base,
                &ts_hlc(local_ts, local_node()),
                &ts_hlc(remote_older, remote_node())
            ),
            Resolution::KeepLocal
        ));
        // Remote newer: remote wins.
        let remote_newer = local_ts + chrono::Duration::seconds(10);
        assert!(matches!(
            resolve_field_merge(
                &local,
                &remote,
                &base,
                &ts_hlc(local_ts, local_node()),
                &ts_hlc(remote_newer, remote_node())
            ),
            Resolution::KeepRemote
        ));
    }

    #[test]
    fn canonical_payload_sorts_map_keys() {
        // Pins the exact-HLC tiebreak's convergence invariant: serde_json must
        // serialize object keys in sorted order. Fails loudly if `preserve_order`
        // is ever enabled anywhere in the dependency tree.
        let bytes = canonical_payload(Some(&json!({"b": 1, "a": 2})));
        assert_eq!(bytes, br#"{"a":2,"b":1}"#.to_vec());
    }
}