car-sync 0.32.1

//! The deterministic fold: `fold(ops) → materialized state`.
//!
//! The proposal's convergence contract, verbatim: "each daemon **folds** the
//! full op-set into local state deterministically. Because the fold is
//! commutative, associative, and idempotent over the op-set (CRDT
//! properties), two laptops writing simultaneously converge the moment they
//! exchange ops."
//!
//! Fold rules per surface tier (the proposal's table):
//! - **Grow-only** (conversations, knowledge, skills, trajectories, runs,
//!   routing observations): union by [`crate::oplog::OpRecord::fold_key`] —
//!   the stable entity key for logical-entity surfaces, the `op_id` for
//!   event-stream surfaces (routing), which fold as a MULTISET: the proposal
//!   replays "the merged **multiset** of observations", so two
//!   byte-identical observations are two events and both survive.
//!   Entities are immutable in this tier (a change is a new op — e.g. a
//!   `Supersedes` fact), so on a key collision with *different* content the
//!   earliest `(hlc, op_id)` writer wins, deterministically.
//! - **Registry** (declagents, the file registries): LWW-register per
//!   record keyed by id, ordered by HLC — *not per file*. Latest
//!   `(hlc, op_id)` wins; concurrent edits to different records both
//!   survive.
//! - **Routing**: the fold materializes the hlc-ordered observation stream;
//!   the EMA replay is the caller-injected [`SyncState::replay`] ("sync the
//!   observations, not the result" — same observations + same canonical
//!   order ⇒ bit-identical result on every device).
//! - **Leased** (`Intent`, B5): LWW-per-run_id (monotone status) with
//!   **per-agent epoch fencing** — a stale-epoch intent from a failed-over
//!   lease holder loses at the fold, order-independently. See [`crate::lease`]
//!   and the [`fold_onto`] `FoldTier::Leased` arm.
//!
//! Determinism discipline: all state is `BTreeMap`-backed and nothing here
//! reads a clock — the proposal calls out "a non-determinism leak
//! (wall-clock or HashMap iteration order sneaking into a fold)" as the bug
//! class [`state_hash`] exists to catch.

pub use crate::oplog::FoldTier;
use crate::oplog::{canonical_json, Hlc, OpRecord};
use car_state::crdt::{LwwMap, LwwRegister};
use serde::{Deserialize, Serialize};
use serde_json::Value;
use sha2::{Digest, Sha256};
use std::collections::BTreeMap;

/// One folded entity: the winning op's payload plus the stamp/id it won with.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct FoldedRecord {
    pub op_id: String,
    pub hlc: Hlc,
    pub payload: Value,
}

/// One agent's leased execution-intent ledger — the [`FoldTier::Leased`]
/// tier's per-agent folded state (B5).
///
/// This slice delivers **deterministic ledger convergence plus a durable
/// idempotency oracle** — it does NOT by itself make execution exactly-once.
/// The exactly-once *execution* gate is B6's dispatch fence (a linearizable
/// "am I still epoch N?" check plus the durable non-fenced idempotency read
/// **before** the external side effect); the fold decides who wins the
/// *ledger*, not whether the effect happens.
///
/// Two distinct views live here, and confusing them causes double-execution:
///
/// - **`committed_runs` is the idempotency oracle** — a **fence-INDEPENDENT,
///   keep-all** map `run_id → committed record`. Once a run commits, it stays
///   here forever (a commit is a fact; no epoch bump erases it), so
///   [`SyncState::committed_run`] is the correct "did this run already
///   execute?" lookup. It is carried in the checkpoint and never trimmed by
///   retention/compaction (see [`crate::compact`]).
/// - **`runs` is the "who holds now" view** — per-agent epoch **fencing**
///   applies to *pending* intents (a stale zombie holder's pending is fenced),
///   while committed/failed records are **terminal-immune** (never reverted to
///   pending, never cleared by a fence raise). Read via [`SyncState::intent`].
///   Do **NOT** use `runs`/`intent()` as the idempotency oracle — a pending
///   fenced by a later epoch is absent here yet the run may have committed; ask
///   `committed_runs` / [`SyncState::committed_run`].
///
/// **Fencing is per AGENT, not per run** (the proposal's spec, deliberately):
/// a zombie's *unique* post-failover **pending** — one the new holder never
/// re-ran — is fenced too (per-run fencing would let it through). `fencing_epoch`
/// is the agent's max-seen lease epoch. `committed_runs` is what makes that
/// safe for idempotency: even after prior-epoch pendings drop from `runs`, the
/// committed fact survives keep-all.
#[derive(Debug, Clone, PartialEq, Default, Serialize, Deserialize)]
pub struct IntentAgent {
    /// The agent's fencing epoch — the max lease epoch any of its intents
    /// carried. **Pending** intents below it are fenced (terminals are immune).
    pub fencing_epoch: u64,
    /// run_id key (`id:<run_id>`) → the "who holds now" winner (terminal-immune;
    /// pendings fenced to `fencing_epoch`). NOT the idempotency oracle.
    pub runs: BTreeMap<String, FoldedRecord>,
    /// run_id key → the committed record. **Fence-independent, keep-all** — the
    /// durable idempotency oracle that survives epoch bumps AND compaction.
    /// Grow-only (highest `(epoch, hlc, op_id)` committed record wins on a
    /// collision); never cleared by fencing. `#[serde(default)]` so a state
    /// serialized before this field parses.
    #[serde(default)]
    pub committed_runs: BTreeMap<String, FoldedRecord>,
}

/// The materialized read model a full op-set folds to. On-disk files
/// (`conversations/*.jsonl`, `declagents.json`, …) are projections of this
/// (the proposal's "files are projections" reframe); B4's checkpoint is a
/// serialized `SyncState` at a frontier.
#[derive(Debug, Clone, PartialEq, Default, Serialize, Deserialize)]
pub struct SyncState {
    /// Grow-only tier: surface tag → stable key → record (union;
    /// first-writer-wins on a key collision).
    pub logs: BTreeMap<String, BTreeMap<String, FoldedRecord>>,
    /// Registry tier: surface tag → record id → LWW winner.
    pub registries: BTreeMap<String, BTreeMap<String, FoldedRecord>>,
    /// Leased execution-intent tier (B5): agent_id → its fenced intent
    /// ledger. Folded from [`crate::oplog::Surface::Intent`] ops with **epoch
    /// fencing** applied deterministically. `#[serde(default)]` so a pre-B5
    /// serialized state still parses.
    #[serde(default)]
    pub intents: BTreeMap<String, IntentAgent>,
}

impl SyncState {
    /// A grow-only surface's entries in canonical `(hlc, op_id)` order — the
    /// deterministic total order every device agrees on (used by the routing
    /// replay, and the order B2's transcript materialization will consume).
    pub fn log_entries(&self, surface_tag: &str) -> Vec<&FoldedRecord> {
        let mut entries: Vec<&FoldedRecord> = self
            .logs
            .get(surface_tag)
            .map(|m| m.values().collect())
            .unwrap_or_default();
        entries.sort_by(|a, b| (&a.hlc, &a.op_id).cmp(&(&b.hlc, &b.op_id)));
        entries
    }

    /// Replay an order-sensitive fold (e.g. the routing EMA) over a surface's
    /// canonically-ordered entries: `fold(routing) =
    /// observations.sorted_by(hlc).fold(empty_store, apply_ema)`. The apply
    /// function is injected — execution (and the EMA itself) stays out of
    /// this crate, like the other pure cores.
    pub fn replay<T, F>(&self, surface_tag: &str, init: T, apply: F) -> T
    where
        F: FnMut(T, &FoldedRecord) -> T,
    {
        self.log_entries(surface_tag).into_iter().fold(init, apply)
    }

    /// The **"who holds now"** leased intent for a run (terminal-immune,
    /// pending-fenced) — NOT the idempotency oracle. `None` when the agent has
    /// no such run, or the run is a *pending* fenced by a later, higher-epoch
    /// holder. A committed run is terminal-immune and stays visible here.
    ///
    /// **For "did this run already execute?" use [`SyncState::committed_run`]**
    /// — `intent()` can return `None`/pending for a run that actually committed
    /// under a prior epoch, which would cause a double-execution if trusted as
    /// the idempotency check.
    pub fn intent(&self, agent_id: &str, run_id: &str) -> Option<&FoldedRecord> {
        self.intents.get(agent_id)?.runs.get(&format!("id:{run_id}"))
    }

    /// The idempotency oracle (B5): the committed record for a run, if it has
    /// **ever** committed for this agent. **Fence-independent and keep-all** —
    /// unaffected by epoch bumps and by compaction — so this is the correct
    /// "did `run_id` already run?" lookup before dispatching a side effect.
    /// `None` iff no committed intent for `(agent_id, run_id)` exists.
    pub fn committed_run(&self, agent_id: &str, run_id: &str) -> Option<&FoldedRecord> {
        self.intents
            .get(agent_id)?
            .committed_runs
            .get(&format!("id:{run_id}"))
    }

    /// Every run_id this agent has committed (the keep-all oracle's keys, with
    /// the `id:` prefix stripped) — for a failover executor scanning "what has
    /// already run".
    pub fn committed_run_ids(&self, agent_id: &str) -> Vec<&str> {
        self.intents
            .get(agent_id)
            .map(|a| {
                a.committed_runs
                    .keys()
                    .filter_map(|k| k.strip_prefix("id:"))
                    .collect()
            })
            .unwrap_or_default()
    }

    /// The agent's current fencing epoch — the max lease epoch its intents
    /// carry — or `None` if it has none. Pending intents below this are fenced.
    pub fn fencing_epoch(&self, agent_id: &str) -> Option<u64> {
        self.intents.get(agent_id).map(|a| a.fencing_epoch)
    }
}

/// Is this folded intent record terminal (committed/failed)? Terminals are
/// immune to fencing; only pending intents are fenced.
fn intent_is_terminal(rec: &FoldedRecord) -> bool {
    crate::lease::intent_status_rank(&rec.payload) == 1
}

/// Total priority for the leased tier's winner selection: terminal-flag (a
/// terminal always outranks a pending — terminal-immunity), then `epoch`
/// (fencing / latest-holder), then `(hlc, op_id)`. `max` under this key is
/// order-independent.
fn intent_priority(rec: &FoldedRecord) -> (u8, u64, &Hlc, &str) {
    (
        crate::lease::intent_status_rank(&rec.payload),
        crate::lease::intent_epoch(&rec.payload),
        &rec.hlc,
        rec.op_id.as_str(),
    )
}

/// Fold an op-set into its materialized state. Order-independent (per-key
/// winner selection under a total order), idempotent (ops dedup on `op_id`
/// first), and pure.
///
/// # Input contract: verify before folding untrusted input
///
/// `fold` does NOT verify ids or chains — that is the explicit, separate
/// [`crate::oplog::verify_log`] pass, and any caller feeding ops from a
/// remote/untrusted source (the B3 relay pull path) MUST run it first.
/// `fold` stays order-independent even on invalid input (two records forging
/// the *same* claimed `op_id` with *different* content dedup by a
/// content-deterministic tiebreak, not arrival order), but which forged
/// record wins is meaningless — verification is what makes the answer mean
/// something.
pub fn fold(ops: &[OpRecord]) -> SyncState {
    fold_onto(&SyncState::default(), ops)
}

/// Fold additional ops **onto an already-folded base state** — the B4
/// checkpoint-consumption primitive: a truncated device reconstructs
/// `fold(full log)` as `fold_onto(checkpoint.state, retained tail)`.
///
/// Uses the same per-key winner selection as [`fold`] (grow-only earliest
/// `(hlc, op_id)` wins; registry latest wins; event streams keyed by
/// `op_id`), so `fold_onto(fold(prefix), suffix) == fold(prefix ∪ suffix)`
/// exactly — the equivalence that makes compaction safe, and the invariant
/// the lib-level B4 tests pin per surface. Re-delivering an op already in
/// the base is idempotent (equal `(hlc, op_id)` never displaces the slot).
///
/// Same input contract as [`fold`]: verify (via
/// [`crate::oplog::verify_log`] / [`crate::checkpoint::verify_anchored`])
/// before folding untrusted input. One caveat unique to invalid input: the
/// base keeps only `FoldedRecord`s, so a forged op colliding with a
/// *base* record's `op_id` cannot use the full-record content tiebreak
/// [`fold`] applies within one op-set — verification is what makes the
/// answer mean something.
pub fn fold_onto(base: &SyncState, ops: &[OpRecord]) -> SyncState {
    // Idempotence: a retransmitted op (same op_id) folds once. On an id
    // collision with DIFFERENT content (invalid input — verify_log rejects
    // it) the tiebreak must not depend on arrival order, so the
    // lexicographically-smaller canonical serialization wins.
    let canonical_record = |op: &OpRecord| -> String {
        canonical_json(&serde_json::to_value(op).expect("OpRecord serializes"))
    };
    let mut unique: BTreeMap<&str, &OpRecord> = BTreeMap::new();
    for op in ops {
        unique
            .entry(&op.op_id)
            .and_modify(|existing| {
                if *existing != op && canonical_record(op) < canonical_record(existing) {
                    *existing = op;
                }
            })
            .or_insert(op);
    }

    let mut state = base.clone();

    // Leased-tier pre-pass (B5): establish the FINAL per-agent `fencing_epoch`
    // (max over the base and every new intent op) BEFORE the main loop, and
    // evict base *pending* records that the raised fence makes stale — terminals
    // (committed/failed) are immune and kept. Deciding pending-fencing against
    // the final fence (not an intermediate one built up mid-loop) is what keeps
    // the leased fold order-independent and base-composable
    // (`fold_onto(checkpoint, tail) == fold(full)`).
    {
        let mut agent_max: BTreeMap<String, u64> = BTreeMap::new();
        for op in unique.values() {
            if op.surface.fold_tier() == FoldTier::Leased {
                let slot = agent_max
                    .entry(crate::lease::intent_agent(&op.payload).to_string())
                    .or_insert(0);
                *slot = (*slot).max(crate::lease::intent_epoch(&op.payload));
            }
        }
        for (agent, max_epoch) in agent_max {
            let entry = state.intents.entry(agent).or_default();
            if max_epoch > entry.fencing_epoch {
                entry.fencing_epoch = max_epoch;
                entry.runs.retain(|_, r| intent_is_terminal(r)); // keep terminals, fence pendings
            }
        }
    }

    for op in unique.values() {
        let record = FoldedRecord {
            op_id: op.op_id.clone(),
            hlc: op.hlc.clone(),
            payload: op.payload.clone(),
        };
        match op.surface.fold_tier() {
            FoldTier::GrowOnly => {
                let slot = state
                    .logs
                    .entry(op.surface.tag())
                    .or_default()
                    .entry(op.fold_key());
                slot.and_modify(|existing| {
                    // Immutable-entity union: earliest (hlc, op_id) wins.
                    // (Unreachable for event-stream surfaces — their fold_key
                    // IS the op_id, so a collision is the same op.)
                    if (&record.hlc, &record.op_id) < (&existing.hlc, &existing.op_id) {
                        *existing = record.clone();
                    }
                })
                .or_insert(record);
            }
            FoldTier::Registry => {
                let slot = state
                    .registries
                    .entry(op.surface.tag())
                    .or_default()
                    .entry(op.fold_key());
                slot.and_modify(|existing| {
                    // LWW: latest (hlc, op_id) wins.
                    if (&record.hlc, &record.op_id) > (&existing.hlc, &existing.op_id) {
                        *existing = record.clone();
                    }
                })
                .or_insert(record);
            }
            FoldTier::Leased => {
                // `fencing_epoch` is already final for this agent (pre-pass).
                let agent = crate::lease::intent_agent(&op.payload).to_string();
                let epoch = crate::lease::intent_epoch(&op.payload);
                let key = op.fold_key();
                let is_terminal = crate::lease::intent_status_rank(&op.payload) == 1;
                let is_committed = crate::lease::intent_is_committed(&op.payload);
                let entry = state.intents.entry(agent).or_default();

                // (A) The idempotency ORACLE: grow-only, fence-INDEPENDENT,
                //     keep-all. A commit is a permanent fact — recorded whatever
                //     its epoch, never cleared by the fence. Highest
                //     `(epoch, hlc, op_id)` committed record wins a collision.
                if is_committed {
                    let better = entry
                        .committed_runs
                        .get(&key)
                        .is_none_or(|existing| intent_priority(&record) > intent_priority(existing));
                    if better {
                        entry.committed_runs.insert(key.clone(), record.clone());
                    }
                }

                // (B) The "who holds now" view: terminals are immune, pendings
                //     are fenced to `fencing_epoch`. Eligible = terminal (always)
                //     OR a live pending at the fence; below-fence pendings drop.
                //     The winner is `max` under `intent_priority`, so a terminal
                //     never reverts to a pending (terminal-immunity) and a
                //     stale-epoch pending never wins.
                let eligible = is_terminal || epoch == entry.fencing_epoch;
                if eligible {
                    let better = entry
                        .runs
                        .get(&key)
                        .is_none_or(|existing| intent_priority(&record) > intent_priority(existing));
                    if better {
                        entry.runs.insert(key, record);
                    }
                }
            }
        }
    }
    state
}

/// Deterministic content hash of a folded state — the proposal's built-in
/// divergence invariant: "Same frontier ⇒ same snapshot hash,
/// deterministically. A mismatch is a fold bug or a non-determinism leak."
/// B4's checkpoint hash is this value at a frontier.
pub fn state_hash(state: &SyncState) -> String {
    let value = serde_json::to_value(state).expect("SyncState serializes");
    let mut hasher = Sha256::new();
    hasher.update(canonical_json(&value).as_bytes());
    let digest = hasher.finalize();
    let hex: String = digest.iter().take(16).map(|b| format!("{b:02x}")).collect();
    format!("state-{hex}")
}

/// Encode an [`Hlc`] as a single `u64` version that preserves the
/// `(wall_ms, counter)` order — the bridge onto `car_state::crdt`'s
/// `(version, replica)` total order. 44 bits of wall-clock milliseconds
/// (good past year 2500) and 20 bits of counter; a counter ≥ 2^20 within one
/// millisecond is outside the HLC's operating range (B3's clock guarantees
/// far less) and would break the order-preservation, so it is debug-asserted.
pub fn hlc_version(hlc: &Hlc) -> u64 {
    debug_assert!(hlc.counter < (1 << 20), "HLC counter exceeds encoding range");
    debug_assert!(
        hlc.wall_ms < (1 << 44),
        "HLC wall_ms exceeds encoding range (the << 20 would drop high bits in release)"
    );
    (hlc.wall_ms << 20) | (u64::from(hlc.counter) & 0xF_FFFF)
}

/// Project a folded registry surface onto the shipped
/// [`car_state::crdt::LwwMap`], so the oplog fold composes with (and is
/// testably equivalent to) `crdt_merge`/`crdt_export` where the domains
/// overlap: `fold(union of ops)` ≡ `merge_maps(per-device exports)`.
pub fn registry_as_lww(state: &SyncState, surface_tag: &str) -> LwwMap {
    state
        .registries
        .get(surface_tag)
        .map(|records| {
            records
                .iter()
                .map(|(key, rec)| {
                    (
                        key.clone(),
                        LwwRegister::new(
                            rec.payload.clone(),
                            hlc_version(&rec.hlc),
                            rec.hlc.device_id.clone(),
                        ),
                    )
                })
                .collect()
        })
        .unwrap_or_default()
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::lease::{Intent, IntentStatus};
    use crate::oplog::{DeviceLog, Scope, Surface};
    use serde_json::json;

    /// Append a leased execution-intent op through a device log.
    fn intent_op(
        dev: &mut DeviceLog,
        agent: &str,
        run: &str,
        epoch: u64,
        status: IntentStatus,
    ) -> OpRecord {
        dev.append(
            Scope::Personal,
            Surface::Intent,
            Intent::new(agent, run, epoch, status).payload(),
        )
    }

    #[test]
    fn grow_only_unions_by_stable_key() {
        let mut a = DeviceLog::new("a");
        let mut b = DeviceLog::new("b");
        let ops = vec![
            a.append(Scope::Personal, Surface::Knowledge, json!({"id": "f1", "v": 1})),
            b.append(Scope::Personal, Surface::Knowledge, json!({"id": "f2", "v": 2})),
        ];
        let state = fold(&ops);
        let knowledge = &state.logs[&Surface::Knowledge.tag()];
        assert_eq!(knowledge.len(), 2);
        assert_eq!(knowledge["id:f1"].payload["v"], json!(1));
        assert_eq!(knowledge["id:f2"].payload["v"], json!(2));
    }

    #[test]
    fn grow_only_key_collision_resolves_to_earliest_deterministically() {
        // Two devices emit different content under one stable id — an
        // anomaly for the immutable tier, resolved first-writer-wins.
        let mut a = DeviceLog::new("a");
        let mut b = DeviceLog::new("b");
        let oa = a.append(Scope::Personal, Surface::Knowledge, json!({"id": "f", "v": "a"}));
        b.observe(&oa.hlc); // b writes causally later
        let ob = b.append(Scope::Personal, Surface::Knowledge, json!({"id": "f", "v": "b"}));
        let fwd = fold(&[oa.clone(), ob.clone()]);
        let rev = fold(&[ob, oa]);
        assert_eq!(fwd, rev);
        assert_eq!(fwd.logs[&Surface::Knowledge.tag()]["id:f"].payload["v"], json!("a"));
    }

    #[test]
    fn registry_is_lww_per_record_not_per_file() {
        let mut a = DeviceLog::new("a");
        let mut b = DeviceLog::new("b");
        // Concurrent edits to DIFFERENT agents both survive.
        let oa = a.append(Scope::Personal, Surface::Declagent, json!({"id": "x", "owner": "a"}));
        let ob = b.append(Scope::Personal, Surface::Declagent, json!({"id": "y", "owner": "b"}));
        // Concurrent edits to the SAME agent resolve by HLC.
        b.observe(&oa.hlc);
        let ob2 = b.append(Scope::Personal, Surface::Declagent, json!({"id": "x", "owner": "b"}));
        let state = fold(&[oa, ob, ob2]);
        let reg = &state.registries[&Surface::Declagent.tag()];
        assert_eq!(reg.len(), 2, "both records survive");
        assert_eq!(reg["id:x"].payload["owner"], json!("b"), "later HLC wins the shared record");
        assert_eq!(reg["id:y"].payload["owner"], json!("b"));
    }

    #[test]
    fn registry_concurrent_tie_breaks_on_device_deterministically() {
        // Same lamport stamp on two devices (true concurrency): the HLC's
        // device_id component breaks the tie, both fold orders agree.
        let mut a = DeviceLog::new("a");
        let mut b = DeviceLog::new("b");
        let oa = a.append(Scope::Personal, Surface::Declagent, json!({"id": "x", "owner": "a"}));
        let ob = b.append(Scope::Personal, Surface::Declagent, json!({"id": "x", "owner": "b"}));
        assert_eq!(oa.hlc.wall_ms, ob.hlc.wall_ms);
        let fwd = fold(&[oa.clone(), ob.clone()]);
        let rev = fold(&[ob, oa]);
        assert_eq!(fwd, rev);
        // "b" > "a" in the device tiebreak — matches crdt's replica tiebreak.
        assert_eq!(
            fwd.registries[&Surface::Declagent.tag()]["id:x"].payload["owner"],
            json!("b")
        );
    }

    #[test]
    fn state_hash_detects_divergence_and_agrees_on_convergence() {
        let mut a = DeviceLog::new("a");
        let o1 = a.append(Scope::Personal, Surface::Knowledge, json!({"id": "f1"}));
        let o2 = a.append(Scope::Personal, Surface::Knowledge, json!({"id": "f2"}));
        let h_full = state_hash(&fold(&[o1.clone(), o2.clone()]));
        let h_full_again = state_hash(&fold(&[o2.clone(), o1.clone()]));
        assert_eq!(h_full, h_full_again, "same op-set → same hash");
        let h_partial = state_hash(&fold(&[o1]));
        assert_ne!(h_full, h_partial, "different op-set → different hash");
        assert!(h_full.starts_with("state-"));
    }

    #[test]
    fn hlc_version_preserves_order() {
        let stamps = [
            Hlc { wall_ms: 1, counter: 0, device_id: "a".into() },
            Hlc { wall_ms: 1, counter: 1, device_id: "a".into() },
            Hlc { wall_ms: 2, counter: 0, device_id: "a".into() },
        ];
        for w in stamps.windows(2) {
            assert!(hlc_version(&w[0]) < hlc_version(&w[1]));
        }
    }

    #[test]
    fn registry_as_lww_matches_crdt_merge_including_export_shape() {
        // The equivalence the proposal leans on: per-device exports merged
        // with the shipped crdt primitives == the fold of the op union.
        let mut a = DeviceLog::new("dev-a");
        let mut b = DeviceLog::new("dev-b");
        let oa1 = a.append(Scope::Personal, Surface::Registry { kind: "agents".into() }, json!({"id": "r1", "v": "a"}));
        let oa2 = a.append(Scope::Personal, Surface::Registry { kind: "agents".into() }, json!({"id": "r2", "v": "a"}));
        b.observe(&oa1.hlc);
        b.observe(&oa2.hlc);
        let ob1 = b.append(Scope::Personal, Surface::Registry { kind: "agents".into() }, json!({"id": "r1", "v": "b"}));

        let tag = Surface::Registry { kind: "agents".into() }.tag();
        let union = registry_as_lww(&fold(&[oa1.clone(), oa2.clone(), ob1.clone()]), &tag);
        let export_a = registry_as_lww(&fold(&[oa1, oa2]), &tag);
        let export_b = registry_as_lww(&fold(&[ob1]), &tag);

        assert_eq!(car_state::crdt::merge_maps(&export_a, &export_b), union);
        assert_eq!(car_state::crdt::merge_many(&[export_b, export_a]), union);
        let plain = car_state::crdt::materialize(&union);
        assert_eq!(plain["id:r1"]["v"], json!("b"));
        assert_eq!(plain["id:r2"]["v"], json!("a"));
    }

    #[test]
    fn log_entries_are_hlc_ordered() {
        let mut a = DeviceLog::new("a");
        let mut b = DeviceLog::new("b");
        let o1 = a.append(Scope::Personal, Surface::Conversation, json!({"t": "first"}));
        b.observe(&o1.hlc);
        let o2 = b.append(Scope::Personal, Surface::Conversation, json!({"t": "second"}));
        a.observe(&o2.hlc); // a's next write causally follows b's
        let o3 = a.append(Scope::Personal, Surface::Conversation, json!({"t": "third"}));
        // Deliver out of order; the view is canonical.
        let state = fold(&[o3, o1, o2]);
        let texts: Vec<&Value> = state
            .log_entries(&Surface::Conversation.tag())
            .iter()
            .map(|r| &r.payload["t"])
            .collect();
        assert_eq!(texts, vec![&json!("first"), &json!("second"), &json!("third")]);
    }

    #[test]
    fn routing_observations_fold_as_a_multiset() {
        // The demonstrated kernel-review defect: "agent x succeeded" twice is
        // TWO observations. Under content-keyed dedup the second collapsed
        // into the first (1 entry, EMA 0.65); the proposal requires the
        // merged MULTISET (2 entries, EMA 0.755).
        let mut dev = DeviceLog::new("dev-a");
        let ops = vec![
            dev.append(Scope::Personal, Surface::Routing, json!({"sample": 1.0})),
            dev.append(Scope::Personal, Surface::Routing, json!({"sample": 1.0})),
        ];
        let state = fold(&ops);
        assert_eq!(
            state.log_entries(&Surface::Routing.tag()).len(),
            2,
            "two byte-identical observations are two events"
        );
        let ema = |s: f64, rec: &FoldedRecord| 0.7 * s + 0.3 * rec.payload["sample"].as_f64().unwrap();
        let value = state.replay(&Surface::Routing.tag(), 0.5_f64, ema);
        assert!((value - 0.755).abs() < 1e-12, "EMA over both events: got {value}");
    }

    #[test]
    fn logical_entity_surfaces_dedup_identical_content() {
        // Content-keyed dedup is scoped to logical-ENTITY surfaces (knowledge,
        // skills, …): the same fact emitted identically by two devices is ONE
        // entity. (Conversation is NOT one of these — it's an event stream
        // keyed by op_id — see the conversation module's CRIT-2 tests.)
        let mut a = DeviceLog::new("a");
        let mut b = DeviceLog::new("b");
        let fact = json!({"kind": "note", "body": "the sky is blue"});
        let oa = a.append(Scope::Personal, Surface::Knowledge, fact.clone());
        let ob = b.append(Scope::Personal, Surface::Knowledge, fact);
        let state = fold(&[oa, ob]);
        assert_eq!(state.log_entries(&Surface::Knowledge.tag()).len(), 1);
    }

    #[test]
    fn forged_colliding_op_id_dedups_order_independently() {
        // Invalid input (verify_log rejects it), but the fold must stay
        // order-independent: two records claiming the SAME op_id with
        // DIFFERENT content tiebreak on content, not arrival order.
        let mut dev = DeviceLog::new("d1");
        let genuine = dev.append(Scope::Personal, Surface::Knowledge, json!({"id": "f", "v": 1}));
        let mut forged = genuine.clone();
        forged.payload = json!({"id": "f", "v": 2}); // op_id NOT recomputed
        assert!(crate::oplog::verify_log(&[forged.clone()]).is_err());

        let ab = fold(&[genuine.clone(), forged.clone()]);
        let ba = fold(&[forged, genuine]);
        assert_eq!(ab, ba, "colliding-id dedup must not depend on arrival order");
        assert_eq!(state_hash(&ab), state_hash(&ba));
    }

    #[test]
    fn fold_onto_prefix_fold_equals_full_fold() {
        // The B4 primitive: fold(prefix) then fold_onto(., suffix) must be
        // byte-identical to fold(prefix ∪ suffix) — for every fold rule at
        // once, including a grow-only collision and an LWW overwrite that
        // CROSS the split point.
        let mut a = DeviceLog::new("a");
        let mut b = DeviceLog::new("b");
        let prefix = vec![
            a.append(Scope::Personal, Surface::Knowledge, json!({"id": "f", "v": "old"})),
            a.append(Scope::Personal, Surface::Declagent, json!({"id": "x", "owner": "a"})),
            a.append(Scope::Personal, Surface::Routing, json!({"sample": 1.0})),
        ];
        for op in &prefix {
            b.observe(&op.hlc);
        }
        let suffix = vec![
            // Grow-only collision across the split: earliest wins → "old".
            b.append(Scope::Personal, Surface::Knowledge, json!({"id": "f", "v": "new"})),
            // LWW across the split: latest wins → owner "b".
            b.append(Scope::Personal, Surface::Declagent, json!({"id": "x", "owner": "b"})),
            // Event stream across the split: both observations survive.
            b.append(Scope::Personal, Surface::Routing, json!({"sample": 0.0})),
        ];
        let mut full = prefix.clone();
        full.extend(suffix.iter().cloned());

        let via_base = fold_onto(&fold(&prefix), &suffix);
        assert_eq!(via_base, fold(&full));
        assert_eq!(state_hash(&via_base), state_hash(&fold(&full)));
        assert_eq!(via_base.logs[&Surface::Knowledge.tag()]["id:f"].payload["v"], json!("old"));
        assert_eq!(
            via_base.registries[&Surface::Declagent.tag()]["id:x"].payload["owner"],
            json!("b")
        );
        assert_eq!(via_base.log_entries(&Surface::Routing.tag()).len(), 2);

        // Idempotent re-delivery: folding an op already in the base changes
        // nothing.
        assert_eq!(fold_onto(&via_base, &prefix), via_base);
    }

    #[test]
    fn empty_fold_is_empty_and_stable() {
        let state = fold(&[]);
        assert_eq!(state, SyncState::default());
        assert_eq!(state_hash(&state), state_hash(&fold(&[])));
        assert!(state.log_entries("conversation").is_empty());
        assert!(registry_as_lww(&state, "declagent").is_empty());
        assert!(state.intent("milo", "R").is_none());
        assert!(state.fencing_epoch("milo").is_none());
    }

    // ------------------------------------------------------------------
    // B5: leased execution-intent fencing as a deterministic fold property.
    // ------------------------------------------------------------------

    #[test]
    fn intent_fold_fences_stale_epoch_order_independently() {
        // Failover: dev-a held epoch 1, dev-b stole epoch 2. Both fire the
        // SAME run R — a=zombie, b=legit holder. The fold must pick epoch 2
        // (b) in ANY delivery order and fence a's epoch-1 writes, leaving one
        // ledger record — no double-commit.
        let mut a = DeviceLog::new("dev-a");
        let mut b = DeviceLog::new("dev-b");
        let ops = vec![
            intent_op(&mut a, "milo", "R", 1, IntentStatus::Pending),
            intent_op(&mut a, "milo", "R", 1, IntentStatus::Committed),
            intent_op(&mut b, "milo", "R", 2, IntentStatus::Pending),
            intent_op(&mut b, "milo", "R", 2, IntentStatus::Committed),
        ];
        let b_commit = ops[3].clone();

        let baseline = fold(&ops);
        assert_eq!(baseline.fencing_epoch("milo"), Some(2));
        assert_eq!(baseline.intents["milo"].runs.len(), 1, "single record — no double-commit");
        let winner = baseline.intent("milo", "R").expect("R survives");
        let decoded = Intent::from_payload(&winner.payload).unwrap();
        assert_eq!((decoded.epoch, decoded.status), (2, IntentStatus::Committed));
        assert_eq!(winner.op_id, b_commit.op_id, "the current holder's commit wins");

        // Order-independence: several explicit permutations agree exactly.
        for order in [
            vec![ops[3].clone(), ops[2].clone(), ops[1].clone(), ops[0].clone()],
            vec![ops[2].clone(), ops[0].clone(), ops[3].clone(), ops[1].clone()],
            vec![ops[1].clone(), ops[3].clone(), ops[0].clone(), ops[2].clone()],
        ] {
            assert_eq!(fold(&order), baseline);
            assert_eq!(state_hash(&fold(&order)), state_hash(&baseline));
        }
    }

    #[test]
    fn intent_per_agent_pending_fencing_with_committed_immunity() {
        // Per-AGENT fencing applies to PENDINGS: dev-a (epoch 1) has an
        // unshared PENDING run S that dev-b (epoch 2) never touched → S's
        // pending is fenced (a stale holder's intent-to-do is silenced). But a
        // COMMITTED run is terminal-immune — a commit is a fact, not fenced —
        // so the zombie's unshared committed run K survives (the C1 fix: an
        // unrelated higher-epoch run must not evict it).
        let mut a = DeviceLog::new("dev-a");
        let mut b = DeviceLog::new("dev-b");
        let ops = vec![
            intent_op(&mut a, "milo", "S", 1, IntentStatus::Pending), // unshared zombie pending
            intent_op(&mut a, "milo", "K", 1, IntentStatus::Committed), // unshared zombie commit
            intent_op(&mut b, "milo", "T", 2, IntentStatus::Committed), // new holder, unrelated run
        ];
        let state = fold(&ops);
        assert_eq!(state.fencing_epoch("milo"), Some(2));
        // The unshared PENDING is fenced; it never committed.
        assert!(state.intent("milo", "S").is_none(), "unshared zombie pending is fenced");
        assert!(state.committed_run("milo", "S").is_none(), "S never committed");
        // The unshared COMMITTED run survives the unrelated epoch bump (C1).
        assert!(
            state.committed_run("milo", "K").is_some(),
            "committed run survives an unrelated epoch bump (idempotency oracle)"
        );
        assert!(state.intent("milo", "K").is_some(), "committed is terminal-immune in runs too");
        assert!(state.committed_run("milo", "T").is_some());
        // A different agent is a different fencing group.
        let mut c = DeviceLog::new("dev-c");
        let mixed = {
            let mut v = ops.clone();
            v.push(intent_op(&mut c, "other", "U", 1, IntentStatus::Committed));
            v
        };
        assert!(
            fold(&mixed).committed_run("other", "U").is_some(),
            "fencing does not cross agents"
        );
    }

    #[test]
    fn intent_fold_onto_equals_full_fold_across_an_epoch_bump() {
        // The compaction-safety equivalence for the leased tier across an epoch
        // bump: a checkpoint captured the agent at epoch 1 (COMMITTED run R). A
        // later, higher-epoch tail op (run S @ 2) raises the fence — and R,
        // being committed, is terminal-immune and SURVIVES (the C1/C3 fix; a
        // commit is a durable fact, not evicted by an unrelated bump). The
        // fold_onto == fold equivalence still holds exactly.
        let mut a = DeviceLog::new("dev-a");
        let mut b = DeviceLog::new("dev-b");
        let prefix = vec![intent_op(&mut a, "milo", "R", 1, IntentStatus::Committed)];
        for op in &prefix {
            b.observe(&op.hlc);
        }
        let tail = vec![intent_op(&mut b, "milo", "S", 2, IntentStatus::Committed)];
        let full = {
            let mut v = prefix.clone();
            v.extend(tail.iter().cloned());
            v
        };

        let base = fold(&prefix); // the "checkpoint" state: epoch 1, R committed
        assert_eq!(base.fencing_epoch("milo"), Some(1));
        assert!(base.committed_run("milo", "R").is_some());

        let via_base = fold_onto(&base, &tail);
        assert_eq!(via_base, fold(&full), "fold_onto == fold across the epoch bump");
        assert_eq!(state_hash(&via_base), state_hash(&fold(&full)));
        assert_eq!(via_base.fencing_epoch("milo"), Some(2));
        // R committed@1 SURVIVES the bump in both views (terminal-immune / oracle).
        assert!(
            via_base.committed_run("milo", "R").is_some(),
            "committed R survives the epoch bump in the idempotency oracle"
        );
        assert!(via_base.intent("milo", "R").is_some(), "committed R is terminal-immune in runs");
        assert!(via_base.committed_run("milo", "S").is_some());
        // Idempotent re-delivery of the tail changes nothing.
        assert_eq!(fold_onto(&via_base, &tail), via_base);
    }

    #[test]
    fn intent_fencing_beats_a_later_hlc() {
        // Safety is by EPOCH, not wall clock: a zombie op with a LATER hlc but
        // a LOWER epoch still loses to the higher-epoch op — no wall-clock race.
        let mut cloud = DeviceLog::new("cloud");
        let mut zombie = DeviceLog::new("laptop");
        let c = intent_op(&mut cloud, "milo", "R", 2, IntentStatus::Committed);
        zombie.observe(&c.hlc); // the zombie's later write stamps a HIGHER hlc
        let z = intent_op(&mut zombie, "milo", "R", 1, IntentStatus::Committed);
        assert!(z.hlc > c.hlc, "the zombie op is later in HLC");

        let state = fold(&[c.clone(), z]);
        let winner = state.intent("milo", "R").unwrap();
        assert_eq!(winner.op_id, c.op_id, "higher epoch wins despite lower HLC");
        assert_eq!(state.fencing_epoch("milo"), Some(2));
    }

    #[test]
    fn idempotent_run_under_failover_uses_the_same_deterministic_run_id() {
        // B7 tie-in: two sites computing the same scheduled occurrence derive
        // the SAME run_id, so a failed-over holder and a zombie collapse to ONE
        // ledger record; epoch fencing then picks the legit (epoch-2) winner.
        let run_id = car_proto::deterministic_run_id("milo", "3am digest", "2026-07-02T03:00");
        assert_eq!(
            run_id,
            car_proto::deterministic_run_id("milo", "3am digest", "2026-07-02T03:00"),
            "same occurrence → same run_id"
        );
        let mut zombie = DeviceLog::new("laptop");
        let mut cloud = DeviceLog::new("cloud");
        let z = intent_op(&mut zombie, "milo", &run_id, 1, IntentStatus::Committed);
        let c = intent_op(&mut cloud, "milo", &run_id, 2, IntentStatus::Committed);

        let state = fold(&[z, c.clone()]);
        assert_eq!(state.intents["milo"].runs.len(), 1, "exactly one execution record");
        assert_eq!(
            state.intent("milo", &run_id).unwrap().op_id,
            c.op_id,
            "the epoch-2 holder's run wins; the zombie is a no-op"
        );
    }

    #[test]
    fn intent_fold_is_order_independent_over_every_permutation() {
        // Brute-force the redesigned leased fold (pre-pass + terminal-immunity
        // + committed oracle): a 5-op set mixing committed/pending across two
        // epochs and three runs must fold IDENTICALLY in all 120 orders.
        fn permutations<T: Clone>(items: &[T]) -> Vec<Vec<T>> {
            fn heap<T: Clone>(k: usize, arr: &mut Vec<T>, out: &mut Vec<Vec<T>>) {
                if k == 1 {
                    out.push(arr.clone());
                    return;
                }
                for i in 0..k {
                    heap(k - 1, arr, out);
                    if k.is_multiple_of(2) {
                        arr.swap(i, k - 1);
                    } else {
                        arr.swap(0, k - 1);
                    }
                }
            }
            let mut arr = items.to_vec();
            let mut out = Vec::new();
            heap(arr.len(), &mut arr, &mut out);
            out
        }

        let mut a = DeviceLog::new("a");
        let mut b = DeviceLog::new("b");
        let ops = vec![
            intent_op(&mut a, "milo", "R", 1, IntentStatus::Committed), // terminal-immune across bump
            intent_op(&mut a, "milo", "S", 1, IntentStatus::Pending),   // unshared pending → fenced
            intent_op(&mut a, "milo", "T", 1, IntentStatus::Committed), // unshared committed → survives
            intent_op(&mut b, "milo", "R", 2, IntentStatus::Pending),   // C3: must not revert R
            intent_op(&mut b, "milo", "S", 2, IntentStatus::Committed), // S commits at the higher epoch
        ];
        let baseline = fold(&ops);
        // Expected steady state.
        assert_eq!(baseline.fencing_epoch("milo"), Some(2));
        let mut committed = baseline.committed_run_ids("milo");
        committed.sort();
        assert_eq!(committed, vec!["R", "S", "T"], "the oracle keeps every committed run");
        assert!(baseline.intent("milo", "S").map(|r| r.op_id.clone()).is_some_and(|_| {
            Intent::from_payload(&baseline.intent("milo", "S").unwrap().payload).unwrap().status
                == IntentStatus::Committed
        }));
        assert_eq!(
            Intent::from_payload(&baseline.intent("milo", "R").unwrap().payload).unwrap().status,
            IntentStatus::Committed,
            "R is not reverted to pending"
        );

        for perm in permutations(&ops) {
            assert_eq!(fold(&perm), baseline, "leased fold must be order-independent");
            assert_eq!(state_hash(&fold(&perm)), state_hash(&baseline));
        }
    }

    #[test]
    fn c1_committed_run_survives_an_unrelated_higher_epoch_run() {
        // C1 REPRO: run R commits at epoch 1; later an UNRELATED run T lands at
        // epoch 2 for the same agent (no concurrency). The old fold cleared
        // runs on the bump, so intent(R) → None → the idempotency check said
        // "not run" → double-execution. FIX: the fence-INDEPENDENT
        // committed_run oracle answers correctly regardless of the bump.
        let mut a = DeviceLog::new("dev-a");
        let mut b = DeviceLog::new("dev-b");
        let r_commit = intent_op(&mut a, "milo", "R", 1, IntentStatus::Committed);
        b.observe(&r_commit.hlc);
        let t_pending = intent_op(&mut b, "milo", "T", 2, IntentStatus::Pending);

        let state = fold(&[r_commit.clone(), t_pending]);
        assert_eq!(state.fencing_epoch("milo"), Some(2), "the unrelated run bumped the fence");
        // The oracle still says R committed — the correct idempotency answer.
        assert_eq!(
            state.committed_run("milo", "R").unwrap().op_id,
            r_commit.op_id,
            "committed_run(R) survives the unrelated epoch bump (C1 fixed)"
        );
        assert_eq!(state.committed_run_ids("milo"), vec!["R"]);
    }

    #[test]
    fn c3_committed_then_pending_across_a_bump_stays_committed() {
        // C3 REPRO: R commits at epoch 1; a failed-over holder writes R PENDING
        // at epoch 2 (before checking). The old Greater arm unconditionally
        // cleared, reverting R to pending → looked un-run → double-execute.
        // FIX: terminal-immunity — the fold keeps committed for R in BOTH views,
        // in any delivery order.
        let mut orig = DeviceLog::new("orig");
        let mut failover = DeviceLog::new("failover");
        let committed = intent_op(&mut orig, "milo", "R", 1, IntentStatus::Committed);
        failover.observe(&committed.hlc);
        let late_pending = intent_op(&mut failover, "milo", "R", 2, IntentStatus::Pending);
        assert!(late_pending.hlc > committed.hlc, "the pending is even later in HLC");

        for order in [
            vec![committed.clone(), late_pending.clone()],
            vec![late_pending.clone(), committed.clone()],
        ] {
            let state = fold(&order);
            // Oracle: committed, unconditionally.
            assert_eq!(
                state.committed_run("milo", "R").unwrap().op_id,
                committed.op_id,
                "committed stays committed across the bump (oracle)"
            );
            // who-holds view: terminal-immune, still committed (not reverted).
            let decoded = Intent::from_payload(&state.intent("milo", "R").unwrap().payload).unwrap();
            assert_eq!(decoded.status, IntentStatus::Committed, "runs view is not reverted to pending");
        }
    }
}