car-state 0.55.0

//! Conflict-free replicated state merge (strong eventual consistency).
//!
//! Applies *CodeCRDT: Observation-Driven Coordination for Multi-Agent LLM Code
//! Generation* (arXiv 2510.18893) to CAR — see
//! `docs/proposals/convergent-shared-state.md`. The paper coordinates stochastic
//! agents through a **shared CRDT state** with deterministic convergence ("100%
//! convergence, zero merge failures") instead of message passing, plus an
//! optimistic *claim* protocol so agents don't duplicate work.
//!
//! CAR already has the substrate: `car_state::StateStore` is a versioned
//! key→value store, and `car_verify::transaction` *detects* write-write /
//! read-write conflicts across concurrent actions/agents — but only reports
//! them. This module supplies the missing *resolution*: merge divergent replicas
//! (concurrent agents, or the offline devices of `docs/proposals/multi-device-sync.md`)
//! into a single state both converge to, deterministically.
//!
//! Two CRDTs, both pure:
//! - [`LwwMap`] — a last-writer-wins key→[`LwwRegister`] map. The shared state.
//! - [`ClaimRegistry`] — a first-claim-wins map for the observation-driven
//!   task-claiming coordination (an agent claims a unit of work; the earliest
//!   claim deterministically wins, so two agents never both own it after merge).
//!
//! All merges are **commutative, associative, and idempotent**, so replicas
//! reach the same state regardless of message order, duplication, or batching —
//! the CRDT strong-eventual-consistency guarantee.

use serde::{Deserialize, Serialize};
use serde_json::Value;
use std::collections::HashMap;

/// A value tagged for last-writer-wins convergence: a logical `version` (e.g.
/// `StateStore::version(key)`) and the `replica` (device/agent id) that wrote
/// it. The pair `(version, replica)` is a total order, so two replicas always
/// agree on the winner — `replica` is the deterministic tiebreaker when versions
/// collide (concurrent writes at the same logical clock).
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct LwwRegister {
    pub value: Value,
    pub version: u64,
    pub replica: String,
}

impl LwwRegister {
    pub fn new(value: Value, version: u64, replica: impl Into<String>) -> Self {
        Self {
            value,
            version,
            replica: replica.into(),
        }
    }

    /// Does `self` win over `other` under the `(version, replica)` total order?
    fn dominates(&self, other: &LwwRegister) -> bool {
        (self.version, self.replica.as_str()) > (other.version, other.replica.as_str())
    }

    /// The deterministic merge of two registers — the dominating one. Returns a
    /// clone so the operation is pure (commutative, associative, idempotent).
    pub fn merge(a: &LwwRegister, b: &LwwRegister) -> LwwRegister {
        if a.dominates(b) {
            a.clone()
        } else {
            b.clone()
        }
    }
}

/// A last-writer-wins key→register map: a conflict-free shared state.
pub type LwwMap = HashMap<String, LwwRegister>;

/// Merge two replicas of an [`LwwMap`] into the state both converge to. The
/// union of keys; per shared key, the dominating register wins. Zero merge
/// failures by construction (the paper's SEC).
pub fn merge_maps(a: &LwwMap, b: &LwwMap) -> LwwMap {
    let mut out = a.clone();
    for (k, rb) in b {
        out.entry(k.clone())
            .and_modify(|ra| *ra = LwwRegister::merge(ra, rb))
            .or_insert_with(|| rb.clone());
    }
    out
}

/// Merge any number of replicas (left fold over [`merge_maps`]). Order-independent
/// by SEC, so the result is the same for any permutation of `replicas`.
pub fn merge_many(replicas: &[LwwMap]) -> LwwMap {
    let mut iter = replicas.iter();
    match iter.next() {
        Some(first) => iter.fold(first.clone(), |acc, r| merge_maps(&acc, r)),
        None => LwwMap::new(),
    }
}

/// Project a merged [`LwwMap`] to a plain key→value state for consumers that
/// don't care about the CRDT tags (e.g. feeding `verify`/`simulate`).
pub fn materialize(m: &LwwMap) -> HashMap<String, Value> {
    m.iter()
        .map(|(k, r)| (k.clone(), r.value.clone()))
        .collect()
}

/// Tag a device/agent's plain state for replication: build an [`LwwMap`] from a
/// state `snapshot` (key→value, e.g. `StateStore::snapshot`) and its per-key
/// `versions` (e.g. `StateStore::versions`), stamped with this `replica` id.
/// This is the *export* half of multi-device sync — each replica exports its
/// state, the maps are exchanged and [`merge_many`]'d, and the result
/// [`materialize`]d back. Keys absent from `versions` default to version `0`.
pub fn export_lww(
    snapshot: &HashMap<String, Value>,
    versions: &HashMap<String, u64>,
    replica: impl Into<String>,
) -> LwwMap {
    let replica = replica.into();
    snapshot
        .iter()
        .map(|(k, v)| {
            let version = versions.get(k).copied().unwrap_or(0);
            (
                k.clone(),
                LwwRegister::new(v.clone(), version, replica.clone()),
            )
        })
        .collect()
}

/// One agent's claim on a unit of work, ordered for *first*-claim-wins: the
/// lowest `(version, replica)` wins, so the earliest claimant keeps the task
/// after any merge and late duplicate claims are dropped.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct Claim {
    /// The agent/replica that claimed the task.
    pub claimant: String,
    /// Logical time of the claim (lower = earlier).
    pub version: u64,
    /// The replica that recorded it (deterministic tiebreaker on equal version).
    pub replica: String,
}

impl Claim {
    fn earlier_than(&self, other: &Claim) -> bool {
        (self.version, self.replica.as_str()) < (other.version, other.replica.as_str())
    }

    fn merge(a: &Claim, b: &Claim) -> Claim {
        if a.earlier_than(b) {
            a.clone()
        } else {
            b.clone()
        }
    }
}

/// A first-claim-wins map task_id→[`Claim`] — the observation-driven coordination
/// primitive. After merging replicas, every task has exactly one owner, chosen
/// deterministically, so concurrent agents never both execute the same unit.
pub type ClaimRegistry = HashMap<String, Claim>;

/// Merge two claim registries: union of tasks; per shared task the *earlier*
/// claim wins (first-claim-wins, vs. last-writer-wins for data).
pub fn merge_claims(a: &ClaimRegistry, b: &ClaimRegistry) -> ClaimRegistry {
    let mut out = a.clone();
    for (task, cb) in b {
        out.entry(task.clone())
            .and_modify(|ca| *ca = Claim::merge(ca, cb))
            .or_insert_with(|| cb.clone());
    }
    out
}

/// Merge any number of claim registries (left fold over [`merge_claims`]).
/// Order-independent: same result for any permutation.
pub fn merge_claims_many(registries: &[ClaimRegistry]) -> ClaimRegistry {
    let mut iter = registries.iter();
    match iter.next() {
        Some(first) => iter.fold(first.clone(), |acc, r| merge_claims(&acc, r)),
        None => ClaimRegistry::new(),
    }
}

/// Record a local claim on `task` for `claimant`. Returns `true` if `claimant`
/// now owns the task (the claim won or was already held), `false` if an existing
/// earlier claim wins — the optimistic "observe then claim" step. Idempotent:
/// re-claiming a task you already own returns `true` without changing the owner.
pub fn claim(
    registry: &mut ClaimRegistry,
    task: impl Into<String>,
    claimant: impl Into<String>,
    version: u64,
    replica: impl Into<String>,
) -> bool {
    let task = task.into();
    let candidate = Claim {
        claimant: claimant.into(),
        version,
        replica: replica.into(),
    };
    match registry.get(&task) {
        Some(existing) if !candidate.earlier_than(existing) => {
            // An existing claim is earlier (or equal) — it keeps the task.
            existing.claimant == candidate.claimant
        }
        _ => {
            let owns = candidate.claimant.clone();
            registry.insert(task, candidate);
            // We just installed our claim; we own it.
            let _ = owns;
            true
        }
    }
}

/// The current owner (claimant) of `task`, if any.
pub fn owner<'a>(registry: &'a ClaimRegistry, task: &str) -> Option<&'a str> {
    registry.get(task).map(|c| c.claimant.as_str())
}

/// The tasks currently owned by `claimant`.
pub fn tasks_claimed_by<'a>(registry: &'a ClaimRegistry, claimant: &str) -> Vec<&'a str> {
    let mut tasks: Vec<&str> = registry
        .iter()
        .filter(|(_, c)| c.claimant == claimant)
        .map(|(t, _)| t.as_str())
        .collect();
    tasks.sort_unstable(); // deterministic
    tasks
}

/// The resolved ownership map task → claimant, after any merges. The actionable
/// view for coordination: agents read it to decide who executes each unit.
pub fn claim_owners(registry: &ClaimRegistry) -> HashMap<String, String> {
    registry
        .iter()
        .map(|(t, c)| (t.clone(), c.claimant.clone()))
        .collect()
}

#[cfg(test)]
mod tests {
    use super::*;
    use serde_json::json;

    fn reg(v: Value, ver: u64, rep: &str) -> LwwRegister {
        LwwRegister::new(v, ver, rep)
    }

    #[test]
    fn higher_version_wins() {
        let a = reg(json!("a"), 1, "r1");
        let b = reg(json!("b"), 2, "r1");
        assert_eq!(LwwRegister::merge(&a, &b).value, json!("b"));
        assert_eq!(LwwRegister::merge(&b, &a).value, json!("b")); // commutative
    }

    #[test]
    fn replica_breaks_version_ties_deterministically() {
        let a = reg(json!("a"), 5, "r1");
        let b = reg(json!("b"), 5, "r2"); // same version, higher replica id wins
        assert_eq!(LwwRegister::merge(&a, &b).value, json!("b"));
        assert_eq!(LwwRegister::merge(&b, &a).value, json!("b"));
    }

    #[test]
    fn merge_is_idempotent() {
        let a = reg(json!(1), 3, "r1");
        assert_eq!(LwwRegister::merge(&a, &a), a);
    }

    fn map(entries: &[(&str, Value, u64, &str)]) -> LwwMap {
        entries
            .iter()
            .map(|(k, v, ver, rep)| (k.to_string(), reg(v.clone(), *ver, rep)))
            .collect()
    }

    #[test]
    fn divergent_replicas_converge() {
        // Two replicas edited disjoint + one shared key concurrently.
        let a = map(&[("x", json!(1), 2, "r1"), ("shared", json!("a"), 1, "r1")]);
        let b = map(&[("y", json!(2), 1, "r2"), ("shared", json!("b"), 3, "r2")]);

        let ab = merge_maps(&a, &b);
        let ba = merge_maps(&b, &a);
        assert_eq!(ab, ba, "merge must be commutative");

        // shared resolves to r2's write (version 3 > 1); disjoint keys preserved.
        assert_eq!(ab.get("shared").unwrap().value, json!("b"));
        assert_eq!(ab.get("x").unwrap().value, json!(1));
        assert_eq!(ab.get("y").unwrap().value, json!(2));
    }

    #[test]
    fn merge_many_is_order_independent() {
        let r1 = map(&[("k", json!("one"), 1, "r1")]);
        let r2 = map(&[("k", json!("two"), 2, "r2")]);
        let r3 = map(&[("k", json!("three"), 3, "r3")]);

        let forward = merge_many(&[r1.clone(), r2.clone(), r3.clone()]);
        let reverse = merge_many(&[r3, r2, r1]);
        assert_eq!(forward, reverse);
        assert_eq!(forward.get("k").unwrap().value, json!("three")); // highest version
    }

    #[test]
    fn associativity() {
        let a = map(&[("k", json!("a"), 1, "r1")]);
        let b = map(&[("k", json!("b"), 2, "r2")]);
        let c = map(&[("k", json!("c"), 2, "r3")]);
        let left = merge_maps(&merge_maps(&a, &b), &c);
        let right = merge_maps(&a, &merge_maps(&b, &c));
        assert_eq!(left, right);
    }

    #[test]
    fn materialize_drops_tags() {
        let m = map(&[("k", json!(42), 1, "r1")]);
        let plain = materialize(&m);
        assert_eq!(plain.get("k"), Some(&json!(42)));
    }

    #[test]
    fn export_tags_snapshot_with_versions_and_replica() {
        let snapshot: HashMap<String, Value> =
            [("a".to_string(), json!(1)), ("b".to_string(), json!(2))].into();
        let versions: HashMap<String, u64> = [("a".to_string(), 5)].into(); // b missing -> 0
        let m = export_lww(&snapshot, &versions, "dev1");
        assert_eq!(m["a"].version, 5);
        assert_eq!(m["a"].replica, "dev1");
        assert_eq!(m["b"].version, 0);
        assert_eq!(m["b"].value, json!(2));
    }

    #[test]
    fn export_then_merge_round_trip() {
        // Two devices export their state; merge converges (higher version wins).
        let snap_a: HashMap<String, Value> = [("k".to_string(), json!("a"))].into();
        let snap_b: HashMap<String, Value> = [("k".to_string(), json!("b"))].into();
        let dev_a = export_lww(&snap_a, &[("k".to_string(), 1)].into(), "A");
        let dev_b = export_lww(&snap_b, &[("k".to_string(), 2)].into(), "B");
        let merged = merge_many(&[dev_a, dev_b]);
        assert_eq!(materialize(&merged).get("k"), Some(&json!("b")));
    }

    #[test]
    fn claim_is_first_wins_and_idempotent() {
        let mut reg = ClaimRegistry::new();
        assert!(claim(&mut reg, "t", "agent-1", 5, "r1")); // first claim wins
        assert_eq!(owner(&reg, "t"), Some("agent-1"));
        // A later claim by another agent loses.
        assert!(!claim(&mut reg, "t", "agent-2", 9, "r2"));
        assert_eq!(owner(&reg, "t"), Some("agent-1"));
        // Re-claiming what you own is idempotent-true.
        assert!(claim(&mut reg, "t", "agent-1", 5, "r1"));
        assert_eq!(owner(&reg, "t"), Some("agent-1"));
    }

    #[test]
    fn tasks_claimed_by_and_owners() {
        let mut reg = ClaimRegistry::new();
        claim(&mut reg, "t1", "a1", 1, "r1");
        claim(&mut reg, "t2", "a1", 1, "r1");
        claim(&mut reg, "t3", "a2", 1, "r2");
        assert_eq!(tasks_claimed_by(&reg, "a1"), vec!["t1", "t2"]);
        let owners = claim_owners(&reg);
        assert_eq!(owners.get("t3"), Some(&"a2".to_string()));
    }

    #[test]
    fn merge_claims_many_resolves_one_owner_per_task() {
        // Two replicas independently claimed the same task; merge picks one.
        let mut a = ClaimRegistry::new();
        claim(&mut a, "t", "a1", 2, "r1");
        let mut b = ClaimRegistry::new();
        claim(&mut b, "t", "a2", 1, "r2"); // earlier -> wins
        let merged = merge_claims_many(&[a, b]);
        assert_eq!(owner(&merged, "t"), Some("a2"));
    }

    #[test]
    fn first_claim_wins_and_converges() {
        // r1 claims "t" at version 1; r2 also claims "t" at version 2.
        let a: ClaimRegistry = [(
            "t".to_string(),
            Claim {
                claimant: "agent-1".into(),
                version: 1,
                replica: "r1".into(),
            },
        )]
        .into();
        let b: ClaimRegistry = [(
            "t".to_string(),
            Claim {
                claimant: "agent-2".into(),
                version: 2,
                replica: "r2".into(),
            },
        )]
        .into();
        let ab = merge_claims(&a, &b);
        let ba = merge_claims(&b, &a);
        assert_eq!(ab, ba, "claim merge must be commutative");
        // The earliest claim wins — agent-1 owns the task after merge.
        assert_eq!(ab.get("t").unwrap().claimant, "agent-1");
    }
}