car_sync/

lib.rs

//! Multi-device sync core for CAR — the oplog + deterministic fold (slice B1
//! of `docs/proposals/multi-device-sync.md`).
//!
//! The proposal's frame: **sync events, not files.** Every state-changing
//! operation is appended to a content-addressed, append-only, replica-tagged
//! [`oplog::OpRecord`] log; sync is "send me the ops I don't have"; and each
//! device [`fold::fold`]s the full op-set into materialized state
//! **deterministically** — commutative, associative, and idempotent over the
//! op-set (CRDT properties), so two laptops writing simultaneously converge
//! the moment they exchange ops.
//!
//! What this slice ships (pure, library-only — no network, no daemon wiring):
//!
//! - [`oplog`] — [`oplog::OpRecord`] exactly as the proposal specs it
//!   (`op_id` content-derived, `hlc {wall_ms, counter, device_id}`,
//!   `scope: Personal | Shared{org}`, the eight-variant `surface` enum,
//!   surface-specific `payload`), plus the per-device `seq`/`prev` hash-chain
//!   linkage that makes a device's log **order-verifiable**
//!   ([`oplog::verify_log`]) and the [`oplog::DeviceLog`] writer that stamps
//!   [`oplog::Hlc`] values from the hybrid clock (see the B3 note below).
//! - [`fold`](mod@fold) — [`fold::fold`]`(ops) -> `[`fold::SyncState`] under the
//!   proposal's per-surface fold rules: **grow-only union by stable ID** for
//!   the log tier, **LWW-register per record ordered by HLC** for the
//!   registry tier, and ordered-observation [`fold::SyncState::replay`] for
//!   the path-dependent routing tier ("sync the observations, not the
//!   result" — the EMA apply is injected, execution stays out of the crate).
//!   [`fold::state_hash`] is the divergence-detection invariant ("same
//!   frontier ⇒ same snapshot hash"), and
//!   [`fold::registry_as_lww`] projects a folded registry onto
//!   `car_state::crdt::LwwMap` so the fold provably agrees with the shipped
//!   `crdt_merge` primitives where the domains overlap.
//! - [`journal`] — durable JSONL persistence for the log in the
//!   `car-eventlog` journal idiom: append-only, torn-line tolerant on load,
//!   plus B4's [`journal::OplogJournal::truncate_to`] (atomic
//!   temp+rename rewrite under the existing advisory lock, stamping a
//!   [`journal::TruncationMarker`] that fences the naive
//!   `load`+`resume` path into a runtime error — a truncated tail resumes
//!   only through [`checkpoint::resume_anchored`]).
//! - [`checkpoint`] (B4) — [`checkpoint::Checkpoint`]: a serialized fold at
//!   a frontier — per-device `{seq, hlc, head}` frontier entries, covered
//!   scopes, the [`fold::SyncState`] snapshot, [`fold::state_hash`] as the
//!   divergence invariant, and a whole-record `checkpoint_hash` (frontier +
//!   scopes + state) as the content address / file name — so "same file ⇒
//!   same checkpoint" holds even when two frontiers fold to one deduped
//!   state, and a tampered frontier is rejected on load.
//!   [`checkpoint::verify_anchored`] proves a truncated tail continues the
//!   checkpoint's recorded chain heads (the checkpoint IS the anchored
//!   head); [`checkpoint::resume_anchored`] resumes a device chain past a
//!   truncation without forking. [`fold::fold_onto`] is the consumption
//!   primitive: `fold_onto(checkpoint.state, tail) == fold(full log)`.
//! - [`compact`] (B4) — per-surface retention
//!   ([`compact::RetentionPolicy::proposal_default`]: conversations last-N,
//!   runs 50/agent + 30 days, trajectories last-D, knowledge/skills/routing
//!   keep-all — every dropped id-bearing entry leaves a minimal tombstone
//!   stub so `supersedes` references resolve even when they arrive after
//!   compaction, and event-stream trims are rejected), the monotone-only
//!   [`compact::AckTable`] fold-frontier bookkeeping (an ack asserts
//!   durably-folded state — MUST, binding on B3), and
//!   [`compact::compact_and_truncate`] enforcing the crash-ordering
//!   invariant **checkpoint durable FIRST, then truncate** — acknowledged
//!   data is never lost, and compaction refuses to drop anything above ANY
//!   device's acked frontier.
//!
//! Determinism discipline (the proposal's "free property" depends on it):
//! all folded state lives in `BTreeMap`s — no `HashMap` iteration order, no
//! wall-clock reads anywhere in the fold/retention path (the age rules'
//! reference instant defaults to [`compact::as_of_from_ops`], pure over the
//! below-frontier ops).
//!
//! B3 adds the missing middle — how ops actually travel:
//!
//! - [`oplog::HlcClock`] — the **real hybrid logical clock**: `{wall_ms,
//!   counter}` state with the standard send/receive rules (max of local
//!   wall and everything witnessed; counter ticks on ties), monotone under
//!   clock skew, regression, and same-millisecond bursts. It replaces B1's
//!   pure-Lamport stamp source behind the SAME wire shape, exactly as
//!   promised — [`DeviceLog::new`] still defaults to the degenerate
//!   logical (always-0 wall) mode, and wall readings are **injectable**
//!   ([`oplog::WallClock`]; [`oplog::system_clock`] is the one opt-in
//!   place system time exists in this crate).
//! - [`relay`] — the [`relay::Relay`] trait (`push` / `pull(since seq
//!   frontier) → {ops, latest_checkpoint_ptr}` / `ack` /
//!   `checkpoint_put/get` / `roster`) with two reference implementations:
//!   [`relay::InMemoryRelay`] and the shared-directory
//!   [`relay::FsRelay`] loopback (the single-user two-Mac case). The relay
//!   admits only ops that *continue* a device's relay-held chain (fork =
//!   runtime error), computes the **stable frontier** = `min(acked)` over
//!   non-evicted roster devices, marks a device silent past the horizon
//!   `H` [`relay::DeviceStatus::Evicted`] (its ack no longer pins GC;
//!   reinstated on a caught-up ack), and GC-drops an op **only** when it
//!   is both at/below the stable frontier AND covered by a stored
//!   checkpoint — checkpoints dedup on `checkpoint_hash`, the whole-record
//!   content address, never `state_hash` (the B4 contract).
//! - [`session`] — [`session::SyncSession`], the device-side pump holding
//!   the B1/B4 contracts **by construction**: append journals (flushed)
//!   before an op is pushable (journal-durable before transmit); pulls are
//!   verified before folding; folds are journaled before the ack, whose
//!   value is *derived from journal-held ops only* (acking merely-received
//!   state is impossible). Retry-safe at every crash point (`op_id` dedup
//!   both ways). Cold bootstrap / straggler re-entry is
//!   [`session::SyncSession::bootstrap`]/[`session::SyncSession::rebase`]:
//!   `checkpoint_get` + `pull(since = checkpoint frontier)` +
//!   [`checkpoint::resume_anchored`] — never `DeviceLog::resume` — with
//!   locally-held uncovered ops (a returning straggler's unpushed writes)
//!   carried across the rebase and pushed after.
//!
//! B5 adds **execution lease + fencing** — single-leader *execution* layered
//! on top of the leaderless *replication* above:
//!
//! - [`lease`] — the [`lease::LeaseCoordinator`] trait: a **linearizable**
//!   compare-and-swap register per agent (exactly one holder at a time; a new
//!   acquire after TTL-expiry or release bumps the monotone `epoch` = the
//!   fencing token). It is deliberately **separate** from
//!   [`relay::Relay`] — an eventually-consistent relay structurally cannot
//!   host a lease (no consensus). [`lease::InMemoryLeaseCoordinator`] is the
//!   honest in-process reference (`Arc<Mutex>` CAS is genuinely linearizable
//!   in one process); a distributed backend is B6. The lease register holds
//!   only non-sensitive metadata, so it never breaches the E2E guarantee on
//!   the actual agent data (the proposal's data/control-plane split).
//! - **Fencing as a fold property, over two views** — the leased
//!   [`oplog::Surface::Intent`] surface ([`fold::FoldTier::Leased`]) carries
//!   the `epoch`, and the fold yields (a) [`fold::SyncState::committed_run`],
//!   the **fence-independent, keep-all idempotency ORACLE** (survives epoch
//!   bumps AND compaction — the correct "did this run already execute?"
//!   lookup), and (b) [`fold::SyncState::intent`], the "who holds now" view
//!   where *pending* intents are per-agent fenced (a stale zombie's pending
//!   loses **deterministically, order-independently, without a wall-clock
//!   race** — fencing beats HLC) while committed/failed are terminal-immune.
//!   Idempotency keys on the B7 `car_proto::deterministic_run_id`. **This
//!   converges the ledger and provides the durable oracle; it is NOT
//!   exactly-once execution** — that is B6's dispatch fence (a linearizable
//!   "still epoch N?" plus the oracle read, before the external effect). See
//!   [`lease`] and [`session::SyncSession::record_intent`] (terminal-guarded).
//!
//! B2 adds **transcript resume** — the conversation surface as an ordered,
//! role-threaded projection of the oplog:
//!
//! - [`conversation`] — [`fold::SyncState::transcript`] folds the
//!   [`oplog::Surface::Conversation`] entries for one `conversation_id` into a
//!   causally-ordered `Vec<`[`conversation::Turn`]`>` (the crate's canonical
//!   `(hlc, op_id)` order — two devices talking to the same agent concurrently
//!   interleave deterministically), and [`fold::SyncState::resume_messages`]
//!   returns the **repaired, provider-valid** [`car_inference_types::Message`]
//!   sequence car-inference's multi-turn path replays to continue the
//!   conversation — the verbatim conversation-resume API
//!   `docs/solutions/conversation-persistence-removed-in-0.25.md` says does not
//!   exist today. A conversation turn is an **event stream keyed by `op_id`**
//!   (op identity IS turn identity — the kernel-review correction: content
//!   keying silently dropped two genuine same-timestamp turns), so a resent op
//!   dedups but two distinct authorings never collapse; it differs from routing
//!   only in being an *independent* multiset (no path-dependent replay), so it
//!   tolerates `LastN` retention. Because HLC order is deterministic but says
//!   nothing about *concurrent* turns, `resume_messages` runs a repair (coalesce
//!   adjacent same-role turns, drop orphan/dangling tool exchanges) so the
//!   `Message` sequence is never provider-invalid — the "runtime validates"
//!   thesis applied to the projection. The 0.25 *compaction-vs-store
//!   incoherence* cannot recur: the oplog is the one source of truth and the
//!   transcript is a projection of the same folded state B4's checkpoint
//!   serializes. Built on the shared `car-inference-types` crate, so a
//!   `Message` shape change is a compile error here, not a runtime break in B6.
//!
//! Later slices: rerouting today's file write paths through the oplog and the
//! daemon/memgine adoption of transcript resume (B6), the `sync.*` WS/FFI
//! surface + E2E encryption + checkpoint/op signing + per-scope streams + the
//! **distributed lease coordinator** (B6 — the network backend speaks the
//! [`relay::Relay`] and [`lease::LeaseCoordinator`] contracts).

pub mod checkpoint;
pub mod compact;
pub mod conversation;
pub mod crypto;
pub mod fence;
pub mod fold;
pub mod journal;
pub mod lease;
pub mod net_relay;
pub mod oplog;
pub mod org_key_directory;
pub mod org_key_provider;
pub mod org_key_resolver;
pub mod org_rotation;
pub mod partition;
pub mod relay;
pub mod session;

pub use checkpoint::{
    resume_anchored, verify_anchored, AnchorError, Checkpoint, CheckpointError, FrontierEntry,
};
pub use compact::{
    apply_retention, as_of_from_ops, compact_and_truncate, is_tombstone, plan_compaction, AckTable,
    CompactError, CompactionOutcome, CompactionPlan, RetentionPolicy, RetentionReport,
    RetentionRule, RUNS_MAX_AGE_MS, RUNS_MAX_PER_AGENT,
};
pub use conversation::{Role, Turn, DEFAULT_CONVERSATION};
pub use crypto::{
    derive_ed25519_identity, derive_key, derive_x25519_identity, ed25519_verifying,
    encryption_audience, generate_org_key, parse_ed25519_verifying, parse_x25519_pub,
    require_canonical_org, unwrap_org_key, wrap_org_key, x25519_public, CryptoError,
    DerivedKeyProvider, Envelope, KdfProfile, LocalKeyCipher, PayloadCipher, StretchedMaster,
    SyncKeyProvider, WrappedOrgKey, ALG_CHACHA20POLY1305, ALG_ORG_KEY_WRAP,
};
// Re-exported because the org-key wrap/unwrap public API takes/returns these
// dalek key types (so downstream doesn't version-match an ed25519-dalek dep).
pub use ed25519_dalek::{SigningKey as OrgSigningKey, VerifyingKey as OrgVerifyingKey};
pub use fence::{check_dispatch, FenceDecision};
pub use fold::{
    fold, fold_at, fold_onto, hlc_version, registry_as_lww, state_hash, FoldTier, FoldedRecord,
    IntentAgent, SyncState,
};
pub use journal::{OplogJournal, TruncationMarker};
pub use lease::{
    InMemoryLeaseCoordinator, Intent, IntentStatus, Lease, LeaseCoordinator, LeaseError,
};
pub use net_relay::{
    LeaseWire, LoopbackTransport, NetworkLeaseCoordinator, NetworkOrgKeyDirectory, NetworkRelay,
    OrgKeyTransport, SyncTransport, TransportError,
};
pub use oplog::{
    canonical_json, logical_clock, system_clock, verify_log, ChainError, DeviceLog, Hlc, HlcClock,
    OpRecord, Scope, Surface, WallClock,
};
pub use org_key_directory::{
    FsOrgKeyDirectory, InMemoryOrgKeyDirectory, MemberPublicKey, OrgKeyDirectory,
    OrgKeyDirectoryError, OrgKeyDirectoryState,
};
pub use org_key_provider::OrgAwareKeyProvider;
pub use org_key_resolver::{resolve_org_root, ResolvedOrgRoot};
pub use org_rotation::{
    is_authoritative, provision_org_members, resolve_all_org_roots, rotate_org_key,
    sign_rotation_floor, verify_rotation_floor, ProvisionReport, RotationError, RotationFloor,
};
pub use partition::{
    is_portable, policy_for, portable_domains, SurfacePolicy, SyncClass, SURFACE_POLICIES,
};
pub use relay::{
    checkpoint_frontier, frontier_of, AckOutcome, DeviceStatus, Frontier, FsRelay, GcReport,
    InMemoryRelay, PullResult, PushOutcome, Relay, RelayConfig, RelayError, RosterEntry,
};
pub use session::{PumpReport, SessionError, SyncSession};

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

    /// Build a two-device op-set exercising both fold tiers:
    /// grow-only Knowledge facts + LWW Declagent registry records.
    fn two_device_ops() -> Vec<OpRecord> {
        let mut a = DeviceLog::new("device-a");
        let mut b = DeviceLog::new("device-b");

        let mut ops = vec![
            a.append(
                Scope::Personal,
                Surface::Knowledge,
                json!({"id": "fact-1", "body": "the sky is blue"}),
            ),
            a.append(
                Scope::Personal,
                Surface::Declagent,
                json!({"id": "agent-1", "name": "milo", "rev": "a1"}),
            ),
            b.append(
                Scope::Personal,
                Surface::Knowledge,
                json!({"id": "fact-2", "body": "water is wet"}),
            ),
        ];
        // b observes a's ops (lamport receive rule) then overwrites agent-1.
        for op in &ops {
            b.observe(&op.hlc);
        }
        ops.push(b.append(
            Scope::Personal,
            Surface::Declagent,
            json!({"id": "agent-1", "name": "milo", "rev": "b2"}),
        ));
        ops.push(a.append(
            Scope::Personal,
            Surface::Conversation,
            json!({"speaker": "user", "text": "hi", "timestamp": 1}),
        ));
        ops
    }

    /// Heap's algorithm — every permutation of `items`, no rand dependency.
    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
    }

    #[test]
    fn fold_is_permutation_invariant() {
        // The core CRDT law: same op-SET in any order → the same state and
        // the same state hash. All 120 permutations of a 5-op set.
        let ops = two_device_ops();
        let baseline = fold(&ops);
        let baseline_hash = state_hash(&baseline);
        for perm in permutations(&ops) {
            let folded = fold(&perm);
            assert_eq!(folded, baseline, "fold must be order-independent");
            assert_eq!(state_hash(&folded), baseline_hash);
        }
    }

    #[test]
    fn fold_is_idempotent_over_duplicated_ops() {
        // Re-delivered ops (relay retransmission) dedup on op_id: folding the
        // set twice-concatenated equals folding it once.
        let ops = two_device_ops();
        let mut doubled = ops.clone();
        doubled.extend(ops.iter().cloned());
        assert_eq!(fold(&doubled), fold(&ops));
        // Re-folding the identical set is stable (idempotent re-fold).
        assert_eq!(fold(&ops), fold(&ops));
    }

    #[test]
    fn divergent_replica_union_matches_crdt_merge() {
        // The overlap contract with the shipped car-state CRDT primitives:
        // folding the UNION of two devices' ops must resolve a registry to
        // exactly the state crdt_merge produces from the per-device exports.
        let ops = two_device_ops();
        let a_ops: Vec<OpRecord> = ops
            .iter()
            .filter(|o| o.device_id == "device-a")
            .cloned()
            .collect();
        let b_ops: Vec<OpRecord> = ops
            .iter()
            .filter(|o| o.device_id == "device-b")
            .cloned()
            .collect();

        let union_lww = registry_as_lww(&fold(&ops), &Surface::Declagent.tag());
        let a_lww = registry_as_lww(&fold(&a_ops), &Surface::Declagent.tag());
        let b_lww = registry_as_lww(&fold(&b_ops), &Surface::Declagent.tag());

        let merged_ab = car_state::crdt::merge_maps(&a_lww, &b_lww);
        let merged_ba = car_state::crdt::merge_maps(&b_lww, &a_lww);
        assert_eq!(merged_ab, union_lww, "fold(union) == crdt_merge(exports)");
        assert_eq!(merged_ba, union_lww, "in either merge order");

        // And the winner is b's later write (b observed a first — higher HLC).
        assert_eq!(union_lww["id:agent-1"].value["rev"], json!("b2"));
        assert_eq!(union_lww["id:agent-1"].replica, "device-b");
    }

    #[test]
    fn journal_round_trip_load_fold_verify() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("oplog.jsonl");
        let ops = two_device_ops();
        {
            let mut journal = OplogJournal::open(&path).unwrap();
            for op in &ops {
                journal.append(op).unwrap();
            }
        }
        let loaded = OplogJournal::load(&path).unwrap();
        assert_eq!(loaded, ops);
        verify_log(&loaded).expect("loaded log must chain-verify");
        assert_eq!(state_hash(&fold(&loaded)), state_hash(&fold(&ops)));
    }

    /// Build a two-device op-set touching EVERY surface tier, with a valid
    /// frontier cut at `split` (every op before it is HLC-≤ every device's
    /// ack): grow-only entities, an LWW registry record overwritten across
    /// the cut, and a routing observation multiset spanning the cut
    /// (including a byte-identical repeat — the multiset trap).
    fn all_surface_ops() -> (Vec<OpRecord>, usize) {
        let mut a = DeviceLog::new("dev-a");
        let mut b = DeviceLog::new("dev-b");
        let mut ops = vec![
            a.append(
                Scope::Personal,
                Surface::Knowledge,
                json!({"id": "f1", "timestamp": 10}),
            ),
            a.append(Scope::Personal, Surface::Skill, json!({"id": "s1"})),
            a.append(
                Scope::Personal,
                Surface::Conversation,
                json!({"speaker": "u", "text": "hi", "timestamp": 11}),
            ),
            a.append(
                Scope::Personal,
                Surface::Run,
                json!({"id": "r1", "agent_id": "milo", "timestamp": 12}),
            ),
            a.append(
                Scope::Personal,
                Surface::Trajectory,
                json!({"id": "t1", "timestamp": 13}),
            ),
            a.append(
                Scope::Personal,
                Surface::Declagent,
                json!({"id": "agent-1", "rev": "a"}),
            ),
            a.append(
                Scope::Shared { org: "acme".into() },
                Surface::Registry {
                    kind: "agents".into(),
                },
                json!({"id": "reg-1", "v": 1}),
            ),
            a.append(Scope::Personal, Surface::Routing, json!({"sample": 1.0})),
            a.append(Scope::Personal, Surface::Routing, json!({"sample": 1.0})), // byte-identical repeat
        ];
        for op in &ops {
            b.observe(&op.hlc);
        }
        ops.push(b.append(
            Scope::Personal,
            Surface::Knowledge,
            json!({"id": "f2", "timestamp": 20}),
        ));
        let split = ops.len();
        // Tail: every tier mutates again, above the frontier.
        ops.push(b.append(
            Scope::Personal,
            Surface::Conversation,
            json!({"speaker": "a", "text": "yo", "timestamp": 21}),
        ));
        ops.push(b.append(
            Scope::Personal,
            Surface::Declagent,
            json!({"id": "agent-1", "rev": "b"}),
        )); // LWW across the cut
        ops.push(b.append(Scope::Personal, Surface::Routing, json!({"sample": 0.0})));
        ops.push(b.append(
            Scope::Personal,
            Surface::Run,
            json!({"id": "r2", "agent_id": "milo", "timestamp": 22}),
        ));
        for op in &ops[split..] {
            a.observe(&op.hlc);
        }
        ops.push(a.append(
            Scope::Personal,
            Surface::Knowledge,
            json!({"id": "f1", "timestamp": 99}),
        )); // grow-only collision across the cut
        (ops, split)
    }

    fn acks_at(ops: &[OpRecord], split: usize) -> AckTable {
        // Every device acks the max HLC of the prefix ("I have folded
        // everything at or below this stamp"), so the stable frontier is
        // exactly the cut.
        let frontier = ops[..split].iter().map(|o| o.hlc.clone()).max().unwrap();
        let mut acks = AckTable::new();
        for op in ops {
            acks.ack(op.device_id.clone(), frontier.clone());
        }
        acks
    }

    #[test]
    fn compaction_equivalence_fold_full_equals_checkpoint_plus_tail() {
        // THE invariant that makes compaction safe, per surface:
        // fold(full log) == fold_onto(checkpoint.state, retained tail),
        // byte-identical state AND state_hash — including the routing
        // observation MULTISET and its order-sensitive replay.
        let (ops, split) = all_surface_ops();
        let acks = acks_at(&ops, split);
        let plan = plan_compaction(&ops, &acks, &RetentionPolicy::keep_all(), None).unwrap();
        assert_eq!(plan.dropped_ops, split);
        assert_eq!(plan.retained_ops.len(), ops.len() - split);

        let full = fold(&ops);
        let reconstructed = fold_onto(&plan.checkpoint.state, &plan.retained_ops);
        assert_eq!(reconstructed, full);
        assert_eq!(state_hash(&reconstructed), state_hash(&full));

        // Per-surface spot checks across the cut:
        assert_eq!(
            reconstructed.registries[&Surface::Declagent.tag()]["id:agent-1"].payload["rev"],
            json!("b"),
            "LWW: the tail's later write wins over the checkpointed one"
        );
        assert_eq!(
            reconstructed.logs[&Surface::Knowledge.tag()]["id:f1"].payload["timestamp"],
            json!(10),
            "grow-only: the checkpointed earliest writer keeps the slot"
        );
        assert_eq!(
            reconstructed.log_entries(&Surface::Routing.tag()).len(),
            3,
            "multiset: 2 checkpointed observations (incl. the repeat) + 1 tail"
        );
        let ema =
            |s: f64, rec: &FoldedRecord| 0.7 * s + 0.3 * rec.payload["sample"].as_f64().unwrap();
        assert_eq!(
            reconstructed.replay(&Surface::Routing.tag(), 0.5_f64, ema),
            full.replay(&Surface::Routing.tag(), 0.5_f64, ema),
            "order-sensitive replay agrees across the compaction"
        );

        // And the composition is verifiable: the checkpoint anchors the tail.
        verify_anchored(&plan.checkpoint, &plan.retained_ops).unwrap();
    }

    #[test]
    fn retention_coherence_local_compaction_equals_global() {
        // The proposal's "local compaction is just an eager application of
        // the same retention the checkpoint applies globally — the two can
        // never disagree": retention(fold_onto(retained ckpt, tail)) ==
        // retention(fold(full)).
        let (ops, split) = all_surface_ops();
        let acks = acks_at(&ops, split);
        let policy = RetentionPolicy::proposal_default(1, u64::MAX);
        let as_of = 1_000u64;
        let plan = plan_compaction(&ops, &acks, &policy, Some(as_of)).unwrap();
        assert_eq!(
            plan.as_of_ms, as_of,
            "explicit as_of wins over the derived default"
        );

        let (global, _) = apply_retention(&fold(&ops), &policy, as_of).unwrap();
        let (local, _) = apply_retention(
            &fold_onto(&plan.checkpoint.state, &plan.retained_ops),
            &policy,
            as_of,
        )
        .unwrap();
        assert_eq!(local, global);
        assert_eq!(state_hash(&local), state_hash(&global));
        // The retained checkpoint really did trim: only the newest turn
        // survives conversations' last-1 rule (turns are content-hash-keyed
        // — no id — so the trimmed one drops without a stub).
        assert_eq!(
            plan.checkpoint.state.logs[&Surface::Conversation.tag()].len(),
            1
        );
    }

    #[test]
    fn retention_coherence_survives_cross_frontier_supersedes() {
        // Kernel-review repro: LastN{1} knowledge drops f1 at compaction
        // time; a LATER tail op f3 supersedes f1. Under
        // preserve-only-what's-referenced-now, the global fold retained f1
        // (it sees f3's reference) while the compacted device could not —
        // divergence under identical policy + as_of. Universal tombstone
        // stubs close the time hole: both sides hold the same f1 stub.
        let mut a = DeviceLog::new("a");
        let mut b = DeviceLog::new("b");
        let mut ops = vec![
            a.append(
                Scope::Personal,
                Surface::Knowledge,
                json!({"id": "f1", "timestamp": 1}),
            ),
            a.append(
                Scope::Personal,
                Surface::Knowledge,
                json!({"id": "f2", "timestamp": 2}),
            ),
        ];
        let split = ops.len();
        for op in &ops {
            b.observe(&op.hlc);
        }
        ops.push(b.append(
            Scope::Personal,
            Surface::Knowledge,
            json!({"id": "f3", "timestamp": 3, "supersedes": "f1"}),
        ));

        let acks = acks_at(&ops, split);
        let mut policy = RetentionPolicy::keep_all();
        policy
            .rules
            .insert("knowledge".to_string(), RetentionRule::LastN { n: 1 });
        let plan = plan_compaction(&ops, &acks, &policy, Some(10)).unwrap();
        assert_eq!(plan.dropped_ops, split);
        let tag = Surface::Knowledge.tag();
        // The checkpoint stubbed f1 BEFORE anything referenced it…
        assert!(is_tombstone(&plan.checkpoint.state.logs[&tag]["id:f1"]));

        // …and the coherence equivalence holds ACROSS the late reference.
        let (global, _) = apply_retention(&fold(&ops), &policy, 10).unwrap();
        let (local, _) = apply_retention(
            &fold_onto(&plan.checkpoint.state, &plan.retained_ops),
            &policy,
            10,
        )
        .unwrap();
        assert_eq!(
            local, global,
            "no divergence despite the cross-frontier supersedes"
        );
        assert_eq!(state_hash(&local), state_hash(&global));
        // f3 is live, its supersedes target resolves against the f1 stub on
        // BOTH sides — a tombstone, not a hole.
        assert_eq!(
            global.logs[&tag]["id:f3"].payload["supersedes"],
            json!("f1")
        );
        assert!(is_tombstone(&global.logs[&tag]["id:f1"]));
        assert!(is_tombstone(&local.logs[&tag]["id:f1"]));
    }

    #[test]
    fn crash_ordering_checkpoint_durable_first_then_truncate() {
        // Simulate a crash between the two durable steps and prove no
        // acknowledged data can be lost at any point.
        let dir = tempfile::tempdir().unwrap();
        let journal_path = dir.path().join("oplog.jsonl");
        let ckpt_dir = dir.path().join("checkpoints");
        let (ops, split) = all_surface_ops();
        let acks = acks_at(&ops, split);
        let policy = RetentionPolicy::keep_all();

        {
            let mut journal = OplogJournal::open(&journal_path).unwrap();
            for op in &ops {
                journal.append(op).unwrap();
            }

            // Step 1+2: plan and persist the checkpoint… then "crash"
            // before truncation (we simply don't truncate).
            let plan = plan_compaction(
                &OplogJournal::load(&journal_path).unwrap(),
                &acks,
                &policy,
                None,
            )
            .unwrap();
            let ckpt_path = plan.checkpoint.save(&ckpt_dir).unwrap();

            // Post-"crash" state: the journal is UNTOUCHED (full data,
            // no truncation marker — the normal load path still works),
            // and the checkpoint is valid but redundant. Nothing lost.
            let survived = OplogJournal::load(&journal_path).unwrap();
            assert_eq!(survived, ops, "journal intact after crash-before-truncate");
            let ckpt = Checkpoint::load(&ckpt_path).unwrap();
            assert_eq!(fold_onto(&ckpt.state, &plan.retained_ops), fold(&ops));
        } // journal lock released — "process died"

        // "Restart": rerun the whole compaction. Idempotent — the same
        // frontier recomputes the same content-addressed checkpoint file —
        // and now the truncation completes.
        let mut journal = OplogJournal::open(&journal_path).unwrap();
        let outcome = compact_and_truncate(&mut journal, &ckpt_dir, &acks, &policy, None).unwrap();
        assert_eq!(outcome.plan.dropped_ops, split);

        // The truncated journal is marked: the naive load path is a runtime
        // error, and the marker names the covering checkpoint.
        assert!(
            OplogJournal::load(&journal_path).is_err(),
            "naive load is fenced"
        );
        let (marker, tail) = OplogJournal::load_with_marker(&journal_path).unwrap();
        assert_eq!(
            marker.unwrap().checkpoint_hash,
            outcome.plan.checkpoint.checkpoint_hash,
            "marker names the covering checkpoint"
        );
        assert_eq!(tail, ops[split..].to_vec());
        verify_log(&tail).expect("truncated journal verifies on its own (anchored non-zero start)");
        let ckpt = Checkpoint::load(&outcome.checkpoint_path.unwrap()).unwrap();
        verify_anchored(&ckpt, &tail).unwrap();
        assert_eq!(
            fold_onto(&ckpt.state, &tail),
            fold(&ops),
            "nothing acknowledged was lost"
        );

        // Exactly one checkpoint file exists (the rerun deduped on content).
        let count = std::fs::read_dir(&ckpt_dir).unwrap().count();
        assert_eq!(count, 1);
    }

    #[test]
    fn truncated_journal_resumes_and_keeps_verifying_end_to_end() {
        let dir = tempfile::tempdir().unwrap();
        let journal_path = dir.path().join("oplog.jsonl");
        let ckpt_dir = dir.path().join("checkpoints");
        let (ops, split) = all_surface_ops();
        let acks = acks_at(&ops, split);

        let mut journal = OplogJournal::open(&journal_path).unwrap();
        for op in &ops {
            journal.append(op).unwrap();
        }
        let outcome = compact_and_truncate(
            &mut journal,
            &ckpt_dir,
            &acks,
            &RetentionPolicy::keep_all(),
            None,
        )
        .unwrap();
        let ckpt = Checkpoint::load(&outcome.checkpoint_path.unwrap()).unwrap();

        // Life goes on after truncation: resume the device chain from
        // checkpoint + tail (never from seq 0 — DeviceLog::resume is fenced
        // and refuses the truncated tail at runtime), append, journal,
        // reload.
        let (_, tail) = OplogJournal::load_with_marker(&journal_path).unwrap();
        assert!(matches!(
            DeviceLog::resume("dev-a", &tail),
            Err(ChainError::TruncatedChain { .. })
        ));
        let mut dev_a = resume_anchored("dev-a", &ckpt, &tail).unwrap();
        let next = dev_a.append(Scope::Personal, Surface::Knowledge, json!({"id": "f-new"}));
        journal.append(&next).unwrap();

        let (marker, reloaded) = OplogJournal::load_with_marker(&journal_path).unwrap();
        assert!(marker.is_some(), "marker survives post-truncation appends");
        verify_anchored(&ckpt, &reloaded).expect("checkpoint anchors the growing truncated log");
        let full_plus = {
            let mut v = ops.clone();
            v.push(next);
            v
        };
        assert_eq!(fold_onto(&ckpt.state, &reloaded), fold(&full_plus));
    }

    // ------------------------------------------------------------------
    // B3 integration: relay transport + roster + eviction + re-entry.
    // ------------------------------------------------------------------

    use crate::session::SyncSession;
    use std::sync::atomic::{AtomicU64, Ordering};
    use std::sync::Arc;

    fn manual_clock() -> (Arc<AtomicU64>, WallClock) {
        let t = Arc::new(AtomicU64::new(0));
        let reader = t.clone();
        (t, Arc::new(move || reader.load(Ordering::SeqCst)))
    }

    fn session(device: &str, root: &std::path::Path, wall: WallClock) -> SyncSession {
        SyncSession::open(
            device,
            &root.join(device).join("oplog.jsonl"),
            &root.join(device).join("checkpoints"),
            wall,
        )
        .unwrap()
    }

    #[test]
    fn two_macs_converge_through_the_filesystem_loopback_relay() {
        // The realistic single-user case: two DeviceLogs syncing through a
        // shared directory, real HLC wall clocks with skew between them.
        let tmp = tempfile::tempdir().unwrap();
        let relay_dir = tmp.path().join("shared-relay");
        let (ta, wall_a) = manual_clock();
        let (tb, wall_b) = manual_clock();
        let (tr, wall_r) = manual_clock();
        ta.store(1_000, Ordering::SeqCst);
        tb.store(940, Ordering::SeqCst); // 60ms of skew
        tr.store(970, Ordering::SeqCst);

        // Each Mac holds its own FsRelay handle on the shared dir — state
        // travels through the files, never through shared memory.
        let mut relay_a =
            FsRelay::open(&relay_dir, RelayConfig::default(), wall_r.clone()).unwrap();
        let mut relay_b = FsRelay::open(&relay_dir, RelayConfig::default(), wall_r).unwrap();
        let mut a = session("mac-a", tmp.path(), wall_a);
        let mut b = session("mac-b", tmp.path(), wall_b);

        a.append(
            Scope::Personal,
            Surface::Knowledge,
            json!({"id": "f1", "v": 1}),
        )
        .unwrap();
        a.append(
            Scope::Personal,
            Surface::Declagent,
            json!({"id": "milo", "owner": "a"}),
        )
        .unwrap();
        b.append(Scope::Personal, Surface::Routing, json!({"sample": 1.0}))
            .unwrap();
        b.append(
            Scope::Personal,
            Surface::Conversation,
            json!({"speaker": "u", "text": "hi", "timestamp": 5}),
        )
        .unwrap();

        a.pump(&mut relay_a).unwrap();
        b.pump(&mut relay_b).unwrap();
        // b writes after folding a's record; despite b's slower wall clock
        // the HLC orders it causally after (receive rule).
        b.append(
            Scope::Personal,
            Surface::Declagent,
            json!({"id": "milo", "owner": "b"}),
        )
        .unwrap();
        b.pump(&mut relay_b).unwrap();
        a.pump(&mut relay_a).unwrap();

        assert_eq!(a.state_hash(), b.state_hash());
        assert_eq!(
            a.state().registries[&Surface::Declagent.tag()]["id:milo"].payload["owner"],
            json!("b"),
            "causality beats wall-clock skew"
        );
        // The journals themselves verify end-to-end.
        verify_log(a.ops()).unwrap();
        verify_log(b.ops()).unwrap();
    }

    #[test]
    fn straggler_eviction_and_lossless_cold_reentry() {
        // The full stragglers arc from the proposal: three devices, c goes
        // dark holding UNPUSHED local writes → evicted at the horizon →
        // frontier unpinned → checkpoint + relay GC → c returns, its pull
        // is FrontierTruncated → cold re-entry (checkpoint_get + pull since
        // the checkpoint frontier + resume_anchored) carrying its unpushed
        // ops → it pushes them (chain-valid) → everyone converges.
        let tmp = tempfile::tempdir().unwrap();
        let (t, wall) = manual_clock();
        let mut relay = InMemoryRelay::new(
            RelayConfig {
                eviction_horizon_ms: Some(1_000),
            },
            wall.clone(),
        );
        let mut a = session("dev-a", tmp.path(), wall.clone());
        let mut b = session("dev-b", tmp.path(), wall.clone());
        let mut c = session("dev-c", tmp.path(), wall.clone());

        // t=100: everyone writes, pumps, acks.
        t.store(100, Ordering::SeqCst);
        a.append(
            Scope::Personal,
            Surface::Knowledge,
            json!({"id": "a1", "timestamp": 100}),
        )
        .unwrap();
        c.append(
            Scope::Personal,
            Surface::Knowledge,
            json!({"id": "c1", "timestamp": 100}),
        )
        .unwrap();
        a.pump(&mut relay).unwrap();
        c.pump(&mut relay).unwrap();
        b.pump(&mut relay).unwrap();
        a.pump(&mut relay).unwrap();
        c.pump(&mut relay).unwrap();
        assert_eq!(a.state_hash(), c.state_hash());

        // t=500: c writes LOCALLY (journal-durable, never pushed) and goes
        // dark.
        t.store(500, Ordering::SeqCst);
        let unpushed = c
            .append(
                Scope::Personal,
                Surface::Knowledge,
                json!({"id": "c-dark", "timestamp": 500}),
            )
            .unwrap();

        // t=800..2000: a and b keep working; c stays silent.
        t.store(800, Ordering::SeqCst);
        a.append(
            Scope::Personal,
            Surface::Knowledge,
            json!({"id": "a2", "timestamp": 800}),
        )
        .unwrap();
        a.pump(&mut relay).unwrap();
        b.pump(&mut relay).unwrap();
        a.pump(&mut relay).unwrap();

        // c's stale ack pins the frontier while it is still active.
        let pinned = relay.stable_frontier().unwrap().unwrap();

        // t=2000: past the horizon (last seen 500) — the sweep on any
        // contact evicts c and the frontier advances past its stale ack.
        t.store(2_000, Ordering::SeqCst);
        a.pump(&mut relay).unwrap();
        b.pump(&mut relay).unwrap();
        let roster: std::collections::BTreeMap<String, RosterEntry> = relay
            .roster()
            .unwrap()
            .into_iter()
            .map(|e| (e.device_id.clone(), e))
            .collect();
        assert_eq!(roster["dev-c"].status, DeviceStatus::Evicted);
        let unpinned = relay.stable_frontier().unwrap().unwrap();
        assert!(
            unpinned > pinned,
            "the evicted device's ack no longer holds the frontier"
        );

        // a checkpoints at the stable frontier and the relay GCs. Before a
        // covering checkpoint exists, NOTHING drops even below the frontier.
        assert_eq!(
            relay.gc().unwrap().total(),
            0,
            "no covering checkpoint → no GC"
        );
        let ckpt = a.publish_checkpoint(&mut relay).unwrap().unwrap();
        let report = relay.gc().unwrap();
        assert!(report.total() > 0, "covered + below-frontier ops now drop");
        // Ops above the stable frontier never drop, covered or not.
        let mut since = Frontier::new();
        for (device, entry) in &ckpt.frontier {
            since.insert(device.clone(), entry.seq);
        }
        for op in relay.pull("dev-a", &since).unwrap().ops {
            assert!(
                op.hlc > unpinned
                    || ckpt
                        .frontier
                        .get(&op.device_id)
                        .is_none_or(|e| op.seq > e.seq)
            );
        }

        // t=3000: c returns. Its normal pump hits truncated space → the
        // cold bootstrap signal. NOTE the pump pushes BEFORE it pulls, so
        // even this failed round already delivered c's journal-durable
        // unpushed op to the relay (chain-valid against c's GC'd chain
        // anchor) — contract 1 makes that safe at any time.
        t.store(3_000, Ordering::SeqCst);
        let err = c.pump(&mut relay).unwrap_err();
        assert!(
            matches!(
                err,
                SessionError::Relay(RelayError::FrontierTruncated { .. })
            ),
            "got {err:?}"
        );
        assert!(
            relay
                .pull("dev-a", &{
                    let mut f = since.clone();
                    f.insert("dev-c".to_string(), 0);
                    f
                })
                .unwrap()
                .ops
                .iter()
                .any(|op| op.op_id == unpushed.op_id),
            "the failed pump's push half already landed the unpushed op"
        );

        // Cold re-entry: rebase onto the checkpoint. The unpushed local op
        // SURVIVES the rebase (uncovered by the checkpoint frontier)…
        assert!(c.rebase(&mut relay).unwrap());
        assert_eq!(c.base().unwrap().checkpoint_hash, ckpt.checkpoint_hash);
        assert!(
            c.ops().iter().any(|op| op.op_id == unpushed.op_id),
            "the straggler's unpushed write survives cold re-entry"
        );
        // …and the naive resume path is fenced on c's rebased journal.
        let c_journal = tmp.path().join("dev-c").join("oplog.jsonl");
        assert!(
            OplogJournal::load(&c_journal).is_err(),
            "truncation marker fences load()"
        );

        // c pumps: its unpushed op is re-offered (the push cursor reset on
        // rebase) and dedups against the failed round's delivery — pushed
        // exactly once overall — then c acks at the new frontier →
        // reinstated.
        let report = c.pump(&mut relay).unwrap();
        assert_eq!(
            (report.pushed, report.push_deduped),
            (0, 1),
            "the unpushed op reached the relay exactly once"
        );
        let roster: std::collections::BTreeMap<String, RosterEntry> = relay
            .roster()
            .unwrap()
            .into_iter()
            .map(|e| (e.device_id.clone(), e))
            .collect();
        assert_eq!(
            roster["dev-c"].status,
            DeviceStatus::Active,
            "caught-up ack reinstates"
        );

        // c's late op has an OLD hlc (below the stable frontier) but is not
        // GC-eligible: no checkpoint covers its seq yet.
        assert!(unpushed.hlc < relay.stable_frontier().unwrap().unwrap());
        assert_eq!(
            relay.gc().unwrap().total(),
            0,
            "late op is safe until a checkpoint covers it"
        );

        // Everyone pulls c's late write and converges — lossless re-entry.
        a.pump(&mut relay).unwrap();
        b.pump(&mut relay).unwrap();
        assert_eq!(a.state_hash(), b.state_hash());
        assert_eq!(a.state_hash(), c.state_hash());
        assert!(a.state().logs[&Surface::Knowledge.tag()].contains_key("id:c-dark"));

        // Chain validity end to end: every journal still proves itself.
        verify_log(a.ops()).unwrap();
        verify_log(b.ops()).unwrap();
        verify_anchored(c.base().unwrap(), c.ops()).unwrap();
    }

    #[test]
    fn replay_over_permuted_opsets_is_deterministic() {
        // The routing rule ("sync the observations, not the result"): an
        // order-sensitive injected fold (EMA-like) over the hlc-ordered
        // observation stream yields the same value from any delivery order.
        let mut a = DeviceLog::new("dev-a");
        let mut b = DeviceLog::new("dev-b");
        let mut ops = vec![
            a.append(Scope::Personal, Surface::Routing, json!({"sample": 1.0})),
            a.append(Scope::Personal, Surface::Routing, json!({"sample": 0.0})),
        ];
        for op in &ops {
            b.observe(&op.hlc);
        }
        ops.push(b.append(Scope::Personal, Surface::Routing, json!({"sample": 1.0})));

        let ema = |state: f64, rec: &FoldedRecord| {
            0.7 * state + 0.3 * rec.payload["sample"].as_f64().unwrap()
        };
        let folded = fold(&ops);
        // Multiset guard: the third observation is byte-identical to the
        // first and must still be a distinct event (this test previously
        // passed while silently losing it).
        assert_eq!(folded.log_entries(&Surface::Routing.tag()).len(), 3);
        let baseline = folded.replay(&Surface::Routing.tag(), 0.5_f64, ema);
        // 0.5 →(1.0) 0.65 →(0.0) 0.455 →(1.0) 0.6185
        assert!((baseline - 0.6185).abs() < 1e-12, "got {baseline}");
        for perm in permutations(&ops) {
            assert_eq!(
                fold(&perm).replay(&Surface::Routing.tag(), 0.5_f64, ema),
                baseline
            );
        }
    }
}