//! Tests for [`super`].

use super::*;
use crate::types::{ChangeOp, Hlc};
use serde_json::json;

/// Properties of the resolver.
///
/// The contract here is convergence, which is a statement about every pair
/// of changes rather than about the pairs someone wrote down. The 43 tests
/// below are examples; these state the rule. See wiki `testing-posture`,
/// Phase 2.
mod properties {
    use super::*;
    use proptest::prelude::*;

    /// Small device pool: node is the final tiebreak, so collisions are the
    /// interesting case and random UUIDs would never produce them.
    fn device_id() -> impl Strategy<Value = DeviceId> {
        (0u8..3).prop_map(|n| {
            let mut bytes = [0u8; 16];
            bytes[15] = n;
            DeviceId::new(Uuid::from_bytes(bytes))
        })
    }

    /// Walls clustered tightly so ties and near-ties are common, plus a
    /// far-future band that trips the clock-poisoning guard.
    fn any_hlc() -> impl Strategy<Value = Hlc> {
        let wall = prop_oneof![
            6 => 1_700_000_000_000i64..1_700_000_000_010,
            2 => 0i64..2_000_000_000_000,
            2 => 4_000_000_000_000i64..8_000_000_000_000,
        ];
        (wall, 0u32..4, device_id()).prop_map(|(wall_ms, counter, node)| Hlc {
            wall_ms,
            counter,
            node,
        })
    }

    /// Pairs of clocks, weighted so exact ties are common.
    ///
    /// Two independent draws almost never collide, and the tie is exactly
    /// where convergence is hardest: it is the case the payload tiebreak in
    /// `resolve_tie` exists for. Generating the pair rather than two
    /// independent clocks is what gives this property teeth, verified by
    /// removing that tiebreak and watching the convergence test fail.
    fn hlc_pair() -> impl Strategy<Value = (Hlc, Hlc)> {
        prop_oneof![
            3 => (any_hlc(), any_hlc()),
            3 => any_hlc().prop_map(|h| (h, h)),
            2 => (any_hlc(), 0u32..4).prop_map(|(h, counter)| (h, Hlc { counter, ..h })),
        ]
    }

    fn entry_with(hlc: Hlc, payload: u8) -> ChangeEntry {
        let mut e = make_entry("tasks", "row-1", ChangeOp::Update, Utc::now());
        e.hlc = hlc;
        e.data = Some(json!({ "v": payload }));
        e
    }

    fn pulled_with(hlc: Hlc, payload: u8) -> PulledChange {
        let mut p = make_pulled(
            "tasks",
            "row-1",
            ChangeOp::Update,
            Utc::now(),
            hlc.node.as_uuid(),
            1,
        );
        p.entry.hlc = hlc;
        p.entry.data = Some(json!({ "v": payload }));
        p
    }

    proptest! {
        /// **Convergence.** Two devices hold the same pair with the roles
        /// reversed: what is local on A is remote on B. If the answer
        /// depended on which side the resolver was handed, the two devices
        /// would keep different rows and never reconcile. No example test
        /// notices unless it happens to pick that pair.
        ///
        /// Stated over the surviving payload rather than the `Resolution`
        /// variant: at an exact tie both sides keep local, which converges
        /// precisely because the two changes are then byte-identical.
        #[test]
        fn lww_picks_the_same_winner_from_either_side(
            (a_hlc, b_hlc) in hlc_pair(),
            a_payload in any::<u8>(),
            b_payload in any::<u8>(),
        ) {
            let now = Utc::now();
            let on_a = match resolve_lww_at(
                &entry_with(a_hlc, a_payload),
                &pulled_with(b_hlc, b_payload),
                now,
            ) {
                Resolution::KeepLocal => a_payload,
                Resolution::KeepRemote => b_payload,
                other => return Err(TestCaseError::fail(format!("unexpected {other:?}"))),
            };
            let on_b = match resolve_lww_at(
                &entry_with(b_hlc, b_payload),
                &pulled_with(a_hlc, a_payload),
                now,
            ) {
                Resolution::KeepLocal => b_payload,
                Resolution::KeepRemote => a_payload,
                other => return Err(TestCaseError::fail(format!("unexpected {other:?}"))),
            };

            prop_assert_eq!(
                on_a, on_b,
                "the two devices kept different payloads and will never converge: \
                 A kept {}, B kept {} (a={:?}, b={:?})",
                on_a, on_b, a_hlc, b_hlc
            );
        }

        /// Resolution is a function of its inputs. Cheap to state, and it is
        /// what lets the resolver be re-run from a retry without
        /// re-deriving the world.
        #[test]
        fn lww_is_deterministic(
            (a_hlc, b_hlc) in hlc_pair(),
            a_payload in any::<u8>(),
            b_payload in any::<u8>(),
        ) {
            let now = Utc::now();
            let local = entry_with(a_hlc, a_payload);
            let first = resolve_lww_at(&local, &pulled_with(b_hlc, b_payload), now);
            let second = resolve_lww_at(&local, &pulled_with(b_hlc, b_payload), now);
            prop_assert_eq!(format!("{first:?}"), format!("{second:?}"));
        }

        /// A poisoned clock must never beat an honest one. This is the
        /// guard's whole purpose: an unbounded future timestamp would
        /// otherwise win every conflict for years.
        #[test]
        fn an_honest_clock_beats_a_poisoned_one(
            honest_wall in 1_700_000_000_000i64..1_700_000_100_000,
            poison_offset in (MAX_HLC_DRIFT_MS + 1)..10_000_000_000i64,
            node_a in device_id(),
            node_b in device_id(),
        ) {
            let now = Utc::now();
            let honest = Hlc { wall_ms: honest_wall, counter: 0, node: node_a };
            let poisoned = Hlc {
                wall_ms: now.timestamp_millis().saturating_add(poison_offset),
                counter: 0,
                node: node_b,
            };
            prop_assume!(!is_clock_poisoned(&honest, now));

            prop_assert!(
                matches!(
                    resolve_lww_at(&entry_with(honest, 1), &pulled_with(poisoned, 2), now),
                    Resolution::KeepLocal
                ),
                "a poisoned remote won against an honest local"
            );
            prop_assert!(
                matches!(
                    resolve_lww_at(&entry_with(poisoned, 2), &pulled_with(honest, 1), now),
                    Resolution::KeepRemote
                ),
                "a poisoned local won against an honest remote"
            );
        }

        /// **A field merge converges, dependent groups included.**
        ///
        /// The two devices see mirror images of one conflict: what is local
        /// on A is remote on B. They must compute the same merged object, or
        /// they hold different bytes forever with nothing to detect it.
        ///
        /// This is aimed at the group rule specifically. Everything else in
        /// the merge decides a field from values both devices have, but the
        /// group rule picks a *side*, and "side" is the one concept that is
        /// device-relative. It converges because the winner comes from
        /// `resolve_tie` over the two entries rather than from which one the
        /// caller happened to label local, and this is what would fail if
        /// that ever regressed to a "ties go to local" rule.
        #[test]
        fn field_merge_converges_on_mirrored_inputs(
            (a_hlc, b_hlc) in hlc_pair(),
            a_state in 0u8..3,
            b_state in 0u8..3,
            a_at in 0u8..3,
            b_at in 0u8..3,
            a_note in 0u8..3,
            b_note in 0u8..3,
        ) {
            const GROUPS: &[&[&str]] = &[&["state", "state_at"]];
            let base = json!({"state": "s0", "state_at": "t0", "note": "n0"});
            let a = json!({
                "state": format!("s{a_state}"),
                "state_at": format!("t{a_at}"),
                "note": format!("n{a_note}"),
            });
            let b = json!({
                "state": format!("s{b_state}"),
                "state_at": format!("t{b_at}"),
                "note": format!("n{b_note}"),
            });

            // On device A the local side is `a`; on device B it is `b`.
            let on_a = resolve_field_merge_with(&a, &b, &base, &a_hlc, &b_hlc, GROUPS);
            let on_b = resolve_field_merge_with(&b, &a, &base, &b_hlc, &a_hlc, GROUPS);

            prop_assert_eq!(
                format!("{on_a:?}"),
                format!("{on_b:?}"),
                "two devices merged the same conflict differently and will \
                 never converge (a={:?}, b={:?})",
                a_hlc,
                b_hlc
            );
        }

        /// **A declared group never lands split across the two sides.**
        ///
        /// The property the declaration exists to buy. However the merge
        /// resolves, every member of a contested group has to come from one
        /// side, so the pair describes a state some device actually held. A
        /// merge that decided `state` and `state_at` independently fails this
        /// on the inputs where the two sides disagree about only one of them,
        /// which is exactly the GoingsOn start()-versus-complete() case.
        #[test]
        fn a_contested_group_never_lands_split(
            (a_hlc, b_hlc) in hlc_pair(),
            a_state in 0u8..3,
            b_state in 0u8..3,
            a_at in 0u8..3,
            b_at in 0u8..3,
        ) {
            const GROUPS: &[&[&str]] = &[&["state", "state_at"]];
            let base = json!({"state": "s0", "state_at": "t0"});
            let a = json!({"state": format!("s{a_state}"), "state_at": format!("t{a_at}")});
            let b = json!({"state": format!("s{b_state}"), "state_at": format!("t{b_at}")});

            let Resolution::Merged(merged) =
                resolve_field_merge_with(&a, &b, &base, &a_hlc, &b_hlc, GROUPS)
            else {
                return Err(TestCaseError::fail("an object base must merge"));
            };

            // The result's group is allowed to be A's, B's, or the base's
            // (untouched). What it must never be is one column from one side
            // and the other from a different one.
            let pair = (&merged["state"], &merged["state_at"]);
            let candidates = [
                (&a["state"], &a["state_at"]),
                (&b["state"], &b["state_at"]),
                (&base["state"], &base["state_at"]),
            ];
            prop_assert!(
                candidates.contains(&pair),
                "the group landed split: got {:?}, which is no device's version \
                 of it (a={a}, b={b})",
                merged
            );
        }
    }
}

/// 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 {
        storage_version: None,
        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));
    // The negative side of `is_empty`: every other assertion in this file is
    // `assert!(clean.is_empty())`, which a constant-`true` `is_empty` also
    // satisfies. Two clean changes must report non-empty.
    assert!(
        !clean.is_empty(),
        "two uncontested rows must not report an empty CleanChanges"
    );
    assert_eq!(clean.len(), 2);
    assert_eq!(conflicts.len(), 1);
    assert_eq!(conflicts[0].remote.entry.row_id, "r1");
}

/// `row_keys` is what a caller pre-fetches committed clocks with, so it has
/// to name every clean row and nothing else, in the order the changes were
/// pulled.
#[test]
fn row_keys_names_every_clean_row_in_order() {
    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("events", "r2", ChangeOp::Insert, now, other_device, 2),
    ];

    let (clean, conflicts) = detect_conflicts(remote, &[], DeviceId::new(our_device));
    assert!(conflicts.is_empty());
    assert_eq!(
        clean.row_keys().collect::<Vec<_>>(),
        vec![("tasks", "r1"), ("events", "r2")]
    );
}

#[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() {
    // A strictly-newer UPDATE must beat an older DELETE, or the edit is
    // 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, so
// 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: overlapping fields whose two sides carry the *same* wall-ms must
// resolve convergently, not "ties go to local". Breaking a tie on local
// would leave device A holding A and device B holding B, a permanent silent
// divergence, so the exact tie breaks on the full HLC (distinct node) and
// 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.
    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
    ));
}

/// The drift window is a stated contract (5 minutes), not an arbitrary
/// number: it is the honest inter-device skew SyncKit promises to tolerate,
/// and every other test in the tree refers to it symbolically, so an
/// arithmetic slip in `5 * 60 * 1000` would change no other outcome.
#[test]
fn max_hlc_drift_is_five_minutes() {
    assert_eq!(MAX_HLC_DRIFT_MS, 300_000);
}

/// The poisoning threshold is exclusive: an entry sitting exactly on
/// `now + MAX_HLC_DRIFT_MS` is still honest, and only the next millisecond
/// is poisoned. Both sides of the boundary, because `>` and `>=` differ
/// only at equality.
#[test]
fn poisoning_threshold_is_exclusive_at_the_drift_boundary() {
    let now = Utc::now();
    let now_ms = now.timestamp_millis();

    let at_boundary = Hlc {
        wall_ms: now_ms + MAX_HLC_DRIFT_MS,
        counter: 0,
        node: remote_node(),
    };
    assert!(
        !is_clock_poisoned(&at_boundary, now),
        "an entry exactly at the drift limit is within tolerated skew"
    );

    let one_past = Hlc {
        wall_ms: now_ms + MAX_HLC_DRIFT_MS + 1,
        counter: 0,
        node: remote_node(),
    };
    assert!(
        is_clock_poisoned(&one_past, now),
        "one millisecond past the drift limit is poisoned"
    );

    let one_before = Hlc {
        wall_ms: now_ms + MAX_HLC_DRIFT_MS - 1,
        counter: 0,
        node: remote_node(),
    };
    assert!(!is_clock_poisoned(&one_before, now));
}

/// The boundary as `resolve_lww_at` sees it: a remote sitting exactly on the
/// limit is honest, so it wins on raw HLC order; one millisecond further out
/// loses to the older local. Same two inputs, opposite resolutions, so the
/// threshold cannot be shifted by a millisecond without this failing.
#[test]
fn lww_keeps_a_remote_at_the_drift_boundary_and_rejects_the_next_ms() {
    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 at_boundary = make_pulled("tasks", "r1", ChangeOp::Update, now, Uuid::new_v4(), 1);
    at_boundary.entry.hlc = Hlc {
        wall_ms: now_ms + MAX_HLC_DRIFT_MS,
        counter: 0,
        node: remote_node(),
    };
    assert!(matches!(
        resolve_lww_at(&local, &at_boundary, now),
        Resolution::KeepRemote
    ));

    let mut one_past = make_pulled("tasks", "r1", ChangeOp::Update, now, Uuid::new_v4(), 1);
    one_past.entry.hlc = Hlc {
        wall_ms: now_ms + MAX_HLC_DRIFT_MS + 1,
        counter: 0,
        node: remote_node(),
    };
    assert!(matches!(
        resolve_lww_at(&local, &one_past, now),
        Resolution::KeepLocal
    ));
}

#[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());
}