car_sync/

checkpoint.rs

//! Checkpoints — a serialized fold at a frontier (slice B4 of
//! `docs/proposals/multi-device-sync.md`, §"Snapshots: bounding the oplog").
//!
//! A [`Checkpoint`] is exactly what the proposal specs: "a serialized
//! **folded state at a frontier F**, content-addressed" — the
//! [`crate::fold::SyncState`] a device folded from every op at or below the
//! frontier, plus the bookkeeping a *truncated* log needs to stay
//! verifiable:
//!
//! - **`frontier`** — per device, the `{seq, hlc, head}` of the last op the
//!   checkpoint covers. Because [`crate::oplog::verify_log`] enforces
//!   HLC-monotone chains, "hlc ≤ F" is always a per-device chain *prefix*,
//!   so the frontier is a clean cut.
//! - **`head`** — the covered chain tail's `op_id`. After truncation the
//!   first retained op's `prev` must link to it: the checkpoint IS the
//!   anchored chain head, which upgrades the A9-style truncation-honesty
//!   story (a truncated log + its checkpoint prove that nothing was dropped
//!   *silently* — the cut is signed into the anchor). [`verify_anchored`]
//!   checks the whole composition; [`crate::oplog::verify_log`] alone
//!   already accepts chains that don't start at `seq 0` (designed for this).
//! - **`state_hash`** — [`crate::fold::state_hash`] of the stored state, the
//!   proposal's divergence invariant ("same frontier ⇒ same snapshot hash").
//! - **`checkpoint_hash`** — the **content address**, covering the WHOLE
//!   record (frontier + scopes + state), not just the state. This is
//!   load-bearing: the fold DEDUPS (e.g. the same conversation turn emitted
//!   by two devices), so two checkpoints with *different frontiers* can fold
//!   to the *same* `SyncState` — a state-only address would give them one
//!   file name, the relay (B3) would dedup-keep the wrong one, and
//!   [`resume_anchored`] would seed `next_seq` from the wrong frontier — a
//!   permanent duplicate-seq chain fork. Addressing the whole record makes
//!   "same file ⇒ same checkpoint" true (and gives B6 a correct thing to
//!   sign); "same frontier ⇒ same file" still holds because the fold is
//!   deterministic.
//!
//! **Scopes, honestly:** the proposal tracks frontiers per scope, but B1's
//! `DeviceLog` stamps ONE `seq`/`prev` chain across all scopes (personal and
//! shared ops interleave in a single device chain), so a per-scope
//! truncation would punch unverifiable holes in the chain. B4 therefore
//! checkpoints a device log **whole-chain** and records the [`Checkpoint::scopes`]
//! it covers; true per-scope frontiers arrive when B3/B6 split the relay
//! streams (and with them the chains) by scope.
//!
//! Durability discipline: [`Checkpoint::save`] writes temp + atomic rename
//! (fsync'd file, best-effort fsync'd dir); [`Checkpoint::load`] re-derives
//! BOTH hashes from the stored content — `state_hash` from the state,
//! `checkpoint_hash` from the whole record — and cross-checks the file name
//! against the content address, **rejecting any mismatch loudly**
//! ([`CheckpointError::HashMismatch`] / [`CheckpointError::ContentMismatch`]
//! / [`CheckpointError::AddressMismatch`]) — a tampered or bit-rotted
//! snapshot (including a tampered *frontier*) never folds and never anchors.

use crate::fold::{fold, state_hash, SyncState};
use crate::oplog::{canonical_json, verify_log, ChainError, DeviceLog, Hlc, OpRecord};
use serde::{Deserialize, Serialize};
use sha2::{Digest, Sha256};
use std::collections::{BTreeMap, BTreeSet};
use std::fmt;
use std::fs::{self, File};
use std::io::Write;
use std::path::{Path, PathBuf};

/// Per-device frontier bookkeeping: the last op the checkpoint covers for
/// one device — its chain position (`seq`), stamp (`hlc`), and `op_id`
/// (`head`, the anchor a truncated log's first retained `prev` links to).
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct FrontierEntry {
    pub seq: u64,
    pub hlc: Hlc,
    pub head: String,
}

/// A serialized fold at a frontier — see the module docs for the design.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct Checkpoint {
    /// device_id → last covered op (`{seq, hlc, head}`). A device absent
    /// here has NO ops below the frontier (its full chain is retained).
    pub frontier: BTreeMap<String, FrontierEntry>,
    /// Sorted, deduped scope tags covered ([`crate::oplog::Scope::tag`]) —
    /// whole-chain checkpointing, see module docs.
    pub scopes: Vec<String>,
    /// [`state_hash`] of `state` — the divergence invariant ("same frontier
    /// ⇒ same snapshot hash"). Recomputed and enforced on
    /// [`Checkpoint::load`].
    pub state_hash: String,
    /// Whole-record content address covering frontier + scopes + state (see
    /// module docs for why state-only addressing forks chains). The file
    /// name; recomputed and enforced on [`Checkpoint::load`].
    pub checkpoint_hash: String,
    /// The folded state at the frontier (possibly retention-compacted by
    /// [`crate::compact::plan_compaction`]).
    pub state: SyncState,
}

impl Checkpoint {
    /// Assemble a checkpoint from its parts, stamping both hashes — the one
    /// construction path, so a `Checkpoint` value is coherent by build.
    pub fn assemble(
        frontier: BTreeMap<String, FrontierEntry>,
        scopes: Vec<String>,
        state: SyncState,
    ) -> Self {
        let hash = state_hash(&state);
        let mut checkpoint = Self {
            frontier,
            scopes,
            state_hash: hash,
            checkpoint_hash: String::new(),
            state,
        };
        checkpoint.checkpoint_hash = checkpoint.content_hash();
        checkpoint
    }

    /// Checkpoint an op-set: verify it (the B1 verify-before-fold
    /// contract), fold it, and record the per-device frontier + scopes.
    /// The state is the **exact** fold — retention is applied separately by
    /// [`crate::compact::plan_compaction`], so the equivalence
    /// `fold_onto(checkpoint.state, tail) == fold(full log)` holds exactly.
    pub fn from_ops(ops: &[OpRecord]) -> Result<Self, ChainError> {
        verify_log(ops)?;
        let state = fold(ops);
        let mut frontier: BTreeMap<String, FrontierEntry> = BTreeMap::new();
        let mut scopes: BTreeSet<String> = BTreeSet::new();
        for op in ops {
            scopes.insert(op.scope.tag());
            let replace = match frontier.get(&op.device_id) {
                Some(existing) => op.seq > existing.seq,
                None => true,
            };
            if replace {
                frontier.insert(
                    op.device_id.clone(),
                    FrontierEntry {
                        seq: op.seq,
                        hlc: op.hlc.clone(),
                        head: op.op_id.clone(),
                    },
                );
            }
        }
        Ok(Self::assemble(frontier, scopes.into_iter().collect(), state))
    }

    /// Recompute the whole-record content address from the checkpoint's
    /// fields (everything except `checkpoint_hash` itself, over the
    /// canonical serialization). Covers the frontier and scopes, not just
    /// the state — see the module docs for the chain-fork this prevents.
    pub fn content_hash(&self) -> String {
        let mut value = serde_json::to_value(self).expect("Checkpoint serializes");
        if let Some(obj) = value.as_object_mut() {
            obj.remove("checkpoint_hash");
        }
        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!("ckpt-{hex}")
    }

    /// Recompute BOTH hashes and reject any mismatch — the load-verify half
    /// of the durability discipline (also callable on an in-memory
    /// checkpoint received from a peer). `state_hash` proves the state;
    /// `checkpoint_hash` proves the whole record, so a tampered *frontier*
    /// is caught too.
    pub fn verify(&self) -> Result<(), CheckpointError> {
        let actual_state = state_hash(&self.state);
        if actual_state != self.state_hash {
            return Err(CheckpointError::HashMismatch {
                expected: self.state_hash.clone(),
                actual: actual_state,
            });
        }
        let actual_content = self.content_hash();
        if actual_content != self.checkpoint_hash {
            return Err(CheckpointError::ContentMismatch {
                expected: self.checkpoint_hash.clone(),
                actual: actual_content,
            });
        }
        Ok(())
    }

    /// Content-addressed file name: `<checkpoint_hash>.checkpoint.json`.
    /// Same frontier + same retention ⇒ same record ⇒ same file (the relay
    /// dedup the proposal leans on) — and, because the address covers the
    /// whole record, same file ⇒ same checkpoint.
    pub fn file_name(&self) -> String {
        format!("{}.checkpoint.json", self.checkpoint_hash)
    }

    /// Durably write the checkpoint into `dir` (created if needed):
    /// temp file → fsync → atomic rename → best-effort dir fsync. Returns
    /// the final path. Writing the same checkpoint twice is idempotent
    /// (content-addressed name, rename-over-identical).
    pub fn save(&self, dir: &Path) -> std::io::Result<PathBuf> {
        fs::create_dir_all(dir)?;
        let final_path = dir.join(self.file_name());
        let tmp_path = dir.join(format!("{}.tmp", self.file_name()));
        {
            let mut tmp = File::create(&tmp_path)?;
            tmp.write_all(
                serde_json::to_string(self)
                    .map_err(std::io::Error::other)?
                    .as_bytes(),
            )?;
            tmp.sync_all()?;
        }
        fs::rename(&tmp_path, &final_path)?;
        // Make the rename itself durable where the platform allows it.
        #[cfg(unix)]
        {
            let _ = File::open(dir).and_then(|d| d.sync_all());
        }
        Ok(final_path)
    }

    /// Load a checkpoint and **verify it**: both stored hashes must
    /// recompute from the stored content ([`Checkpoint::verify`]), and the
    /// file's name must BE the content address — a renamed or
    /// wrongly-addressed file is rejected ([`CheckpointError::AddressMismatch`])
    /// so a store can never serve checkpoint X under checkpoint Y's name.
    /// Never fold a snapshot that doesn't prove itself.
    pub fn load(path: &Path) -> Result<Self, CheckpointError> {
        let raw = fs::read_to_string(path).map_err(CheckpointError::Io)?;
        let checkpoint: Checkpoint =
            serde_json::from_str(&raw).map_err(CheckpointError::Parse)?;
        checkpoint.verify()?;
        let expected = checkpoint.file_name();
        let found = path
            .file_name()
            .map(|n| n.to_string_lossy().into_owned())
            .unwrap_or_default();
        if found != expected {
            return Err(CheckpointError::AddressMismatch { expected, found });
        }
        Ok(checkpoint)
    }
}

/// A checkpoint load/verify failure.
#[derive(Debug)]
pub enum CheckpointError {
    Io(std::io::Error),
    Parse(serde_json::Error),
    /// The stored `state_hash` does not recompute from the stored state —
    /// tampering, corruption, or a fold-determinism bug. Never fold it.
    HashMismatch { expected: String, actual: String },
    /// The stored `checkpoint_hash` does not recompute from the whole
    /// record — a tampered frontier/scopes (or a forged address). Never
    /// fold or anchor on it.
    ContentMismatch { expected: String, actual: String },
    /// The file's name is not the record's content address — a renamed,
    /// swapped, or wrongly-stored checkpoint file.
    AddressMismatch { expected: String, found: String },
}

impl fmt::Display for CheckpointError {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            CheckpointError::Io(e) => write!(f, "checkpoint io error: {e}"),
            CheckpointError::Parse(e) => write!(f, "checkpoint parse error: {e}"),
            CheckpointError::HashMismatch { expected, actual } => write!(
                f,
                "checkpoint state_hash mismatch (stored {expected}, recomputed {actual}) — \
                 tampered or corrupt snapshot, refusing to fold it"
            ),
            CheckpointError::ContentMismatch { expected, actual } => write!(
                f,
                "checkpoint content-address mismatch (stored {expected}, recomputed {actual}) — \
                 frontier/scopes tampered or record corrupt, refusing to fold or anchor on it"
            ),
            CheckpointError::AddressMismatch { expected, found } => write!(
                f,
                "checkpoint file name {found} is not its content address {expected} — \
                 renamed or wrongly-stored checkpoint, refusing to load it"
            ),
        }
    }
}

impl std::error::Error for CheckpointError {}

/// A failure composing a checkpoint with a (truncated) tail.
#[derive(Debug)]
pub enum AnchorError {
    /// The checkpoint itself doesn't verify.
    Checkpoint(CheckpointError),
    /// The tail doesn't chain-verify on its own.
    Chain(ChainError),
    /// A device's first retained op doesn't continue the checkpoint's
    /// recorded frontier (`seq`, `prev` link, or HLC advance).
    BrokenAnchor { device_id: String, detail: String },
    /// A device absent from the checkpoint frontier must present its chain
    /// from `seq 0` — a nonzero start with no anchor is a silent hole.
    UnanchoredDevice { device_id: String, first_seq: u64 },
}

impl fmt::Display for AnchorError {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            AnchorError::Checkpoint(e) => write!(f, "anchor checkpoint invalid: {e}"),
            AnchorError::Chain(e) => write!(f, "anchored tail chain invalid: {e}"),
            AnchorError::BrokenAnchor { device_id, detail } => {
                write!(f, "device {device_id}: tail does not anchor on checkpoint ({detail})")
            }
            AnchorError::UnanchoredDevice { device_id, first_seq } => write!(
                f,
                "device {device_id}: first op has seq {first_seq} but the checkpoint \
                 records no frontier for it — unanchored truncation"
            ),
        }
    }
}

impl std::error::Error for AnchorError {}

/// Verify the composition `checkpoint + retained tail` — the truncated-log
/// analogue of [`verify_log`]:
///
/// 1. the checkpoint proves itself (both `state_hash` and the whole-record
///    `checkpoint_hash` recompute — a tampered frontier can't anchor);
/// 2. the tail chain-verifies on its own (`verify_log` — which already
///    accepts chains not starting at `seq 0`);
/// 3. per device, the first retained op continues the checkpoint's frontier
///    exactly: `seq == frontier.seq + 1`, `prev == frontier.head`, and
///    `hlc > frontier.hlc`;
/// 4. a device with **no** frontier entry must start at `seq 0` (nothing of
///    it was truncated).
///
/// A pass means the pair carries the same integrity guarantee the full log
/// carried — the checkpoint IS the anchored head of every truncated chain.
pub fn verify_anchored(checkpoint: &Checkpoint, tail: &[OpRecord]) -> Result<(), AnchorError> {
    checkpoint.verify().map_err(AnchorError::Checkpoint)?;
    verify_log(tail).map_err(AnchorError::Chain)?;

    // First retained op per device (min seq).
    let mut first: BTreeMap<&str, &OpRecord> = BTreeMap::new();
    for op in tail {
        let replace = match first.get(op.device_id.as_str()) {
            Some(existing) => op.seq < existing.seq,
            None => true,
        };
        if replace {
            first.insert(&op.device_id, op);
        }
    }

    for (device_id, op) in first {
        match checkpoint.frontier.get(device_id) {
            Some(anchor) => {
                if op.seq != anchor.seq + 1 {
                    return Err(AnchorError::BrokenAnchor {
                        device_id: device_id.to_string(),
                        detail: format!(
                            "first retained seq {} does not continue frontier seq {}",
                            op.seq, anchor.seq
                        ),
                    });
                }
                if op.prev.as_deref() != Some(anchor.head.as_str()) {
                    return Err(AnchorError::BrokenAnchor {
                        device_id: device_id.to_string(),
                        detail: format!(
                            "first retained op's prev does not link to frontier head {}",
                            anchor.head
                        ),
                    });
                }
                if op.hlc <= anchor.hlc {
                    return Err(AnchorError::BrokenAnchor {
                        device_id: device_id.to_string(),
                        detail: "first retained op's hlc does not advance past the frontier"
                            .to_string(),
                    });
                }
            }
            None => {
                if op.seq != 0 {
                    return Err(AnchorError::UnanchoredDevice {
                        device_id: device_id.to_string(),
                        first_seq: op.seq,
                    });
                }
            }
        }
    }
    Ok(())
}

/// Resume a device's append chain from a checkpoint + retained tail — the
/// truncated-journal sibling of [`DeviceLog::resume`]. Without this, a
/// device whose ops were ALL below the frontier would resume at `seq 0` and
/// permanently fork its own chain (the same hazard `DeviceLog::resume`'s
/// journal-durable-before-transmit contract guards). Verifies the
/// composition first, seeds `seq`/`prev` from the frontier anchor, and
/// advances the hybrid clock past every stamp the checkpoint or tail
/// covers (the frontier holds each device's max covered HLC, so it bounds
/// everything folded into the state). The resumed log defaults to the
/// logical (always-0) wall source — attach the real one with
/// [`DeviceLog::set_wall_clock`].
pub fn resume_anchored(
    device_id: impl Into<String>,
    checkpoint: &Checkpoint,
    tail: &[OpRecord],
) -> Result<DeviceLog, AnchorError> {
    verify_anchored(checkpoint, tail)?;
    let device_id = device_id.into();
    let mut log = DeviceLog::new(device_id.clone());
    for entry in checkpoint.frontier.values() {
        log.clock.observe(&entry.hlc);
    }
    if let Some(anchor) = checkpoint.frontier.get(&device_id) {
        log.next_seq = anchor.seq + 1;
        log.prev = Some(anchor.head.clone());
    }
    for op in tail {
        log.clock.observe(&op.hlc);
        if op.device_id == device_id && op.seq >= log.next_seq {
            log.next_seq = op.seq + 1;
            log.prev = Some(op.op_id.clone());
        }
    }
    Ok(log)
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::fold::fold_onto;
    use crate::oplog::{Scope, Surface};
    use serde_json::json;

    /// Two devices, mixed scopes/surfaces; returns (all ops, split index) —
    /// ops[..split] is a valid per-device chain-prefix cut.
    fn ops_with_cut() -> (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", "v": 1})),
            a.append(
                Scope::Shared { org: "acme".into() },
                Surface::Declagent,
                json!({"id": "agent-1", "rev": "a"}),
            ),
        ];
        for op in &ops {
            b.observe(&op.hlc);
        }
        ops.push(b.append(Scope::Personal, Surface::Knowledge, json!({"id": "f2", "v": 2})));
        let split = ops.len();
        ops.push(a.append(Scope::Personal, Surface::Conversation, json!({"speaker": "u", "text": "hi", "timestamp": 10})));
        for op in &ops[split..] {
            b.observe(&op.hlc);
        }
        ops.push(b.append(
            Scope::Shared { org: "acme".into() },
            Surface::Declagent,
            json!({"id": "agent-1", "rev": "b"}),
        ));
        (ops, split)
    }

    #[test]
    fn from_ops_records_frontier_heads_and_scopes() {
        let (ops, split) = ops_with_cut();
        let ckpt = Checkpoint::from_ops(&ops[..split]).unwrap();
        assert_eq!(ckpt.scopes, vec!["personal".to_string(), "shared:acme".to_string()]);
        assert_eq!(ckpt.frontier.len(), 2);
        assert_eq!(ckpt.frontier["dev-a"].seq, 1);
        assert_eq!(ckpt.frontier["dev-a"].head, ops[1].op_id);
        assert_eq!(ckpt.frontier["dev-b"].seq, 0);
        assert_eq!(ckpt.frontier["dev-b"].head, ops[2].op_id);
        assert_eq!(ckpt.state_hash, state_hash(&fold(&ops[..split])));
    }

    #[test]
    fn from_ops_refuses_an_invalid_log() {
        let (mut ops, _) = ops_with_cut();
        ops[0].payload = json!({"forged": true}); // id no longer matches
        assert!(matches!(Checkpoint::from_ops(&ops), Err(ChainError::IdMismatch { .. })));
    }

    #[test]
    fn checkpoint_save_load_round_trips_content_addressed() {
        let dir = tempfile::tempdir().unwrap();
        let (ops, split) = ops_with_cut();
        let ckpt = Checkpoint::from_ops(&ops[..split]).unwrap();

        let path = ckpt.save(dir.path()).unwrap();
        assert_eq!(
            path.file_name().unwrap().to_str().unwrap(),
            format!("{}.checkpoint.json", ckpt.checkpoint_hash),
            "file name is the WHOLE-RECORD content address"
        );
        let loaded = Checkpoint::load(&path).unwrap();
        assert_eq!(loaded, ckpt);

        // Saving again is idempotent (same content → same file).
        let again = ckpt.save(dir.path()).unwrap();
        assert_eq!(again, path);
        // No stray temp file left behind.
        let names: Vec<String> = fs::read_dir(dir.path())
            .unwrap()
            .map(|e| e.unwrap().file_name().to_string_lossy().into_owned())
            .collect();
        assert_eq!(names.len(), 1, "only the final checkpoint file exists: {names:?}");
    }

    #[test]
    fn tampered_checkpoint_is_rejected_on_load() {
        let dir = tempfile::tempdir().unwrap();
        let (ops, split) = ops_with_cut();
        let ckpt = Checkpoint::from_ops(&ops[..split]).unwrap();
        let path = ckpt.save(dir.path()).unwrap();

        // Flip a payload value inside the stored state without recomputing
        // the hash — the load must fail loudly, not fold garbage.
        let raw = fs::read_to_string(&path).unwrap();
        let tampered = raw.replace("\"v\":1", "\"v\":999");
        assert_ne!(raw, tampered, "tamper target must exist in the serialized state");
        fs::write(&path, tampered).unwrap();
        assert!(matches!(
            Checkpoint::load(&path),
            Err(CheckpointError::HashMismatch { .. })
        ));

        // In-memory tamper too.
        let mut forged = ckpt.clone();
        forged.state.logs.clear();
        assert!(matches!(forged.verify(), Err(CheckpointError::HashMismatch { .. })));
    }

    #[test]
    fn tampered_frontier_is_rejected_on_load_and_verify() {
        // Kernel-review defect (reproduced): with a state-only hash, a
        // stored file's FRONTIER could be edited and still pass verify().
        // The whole-record content address closes that.
        let dir = tempfile::tempdir().unwrap();
        let (ops, split) = ops_with_cut();
        let ckpt = Checkpoint::from_ops(&ops[..split]).unwrap();
        let path = ckpt.save(dir.path()).unwrap();

        // On-disk frontier tamper: bump dev-a's anchor seq. state_hash still
        // recomputes (state untouched) — the content address must catch it.
        let mut value: serde_json::Value =
            serde_json::from_str(&fs::read_to_string(&path).unwrap()).unwrap();
        value["frontier"]["dev-a"]["seq"] = json!(7);
        fs::write(&path, serde_json::to_string(&value).unwrap()).unwrap();
        assert!(matches!(
            Checkpoint::load(&path),
            Err(CheckpointError::ContentMismatch { .. })
        ));

        // In-memory frontier tamper is equally rejected — and therefore
        // can never anchor or seed a resume.
        let mut forged = ckpt.clone();
        forged.frontier.get_mut("dev-a").unwrap().seq = 7;
        assert!(matches!(forged.verify(), Err(CheckpointError::ContentMismatch { .. })));
        assert!(matches!(
            verify_anchored(&forged, &ops[split..]),
            Err(AnchorError::Checkpoint(CheckpointError::ContentMismatch { .. }))
        ));

        // A renamed file is not its own address — rejected.
        let good = ckpt.save(dir.path()).unwrap();
        let renamed = dir.path().join("latest.checkpoint.json");
        fs::rename(&good, &renamed).unwrap();
        assert!(matches!(
            Checkpoint::load(&renamed),
            Err(CheckpointError::AddressMismatch { .. })
        ));
    }

    #[test]
    fn different_frontiers_never_share_a_content_address() {
        // Kernel-review defect (reproduced): the fold dedups a
        // byte-identical logical-entity fact emitted by two devices, so
        // from_ops([oa]) and from_ops([oa, ob]) fold to the SAME SyncState
        // (same state_hash) with DIFFERENT frontiers. Under state-only
        // addressing they shared one file name — the relay would dedup-keep
        // the wrong one, verify_anchored would fail, and resume_anchored would
        // seed next_seq from the wrong frontier (a permanent duplicate-seq
        // chain fork). The whole-record address keeps them distinct.
        // (Knowledge is content-keyed and dedups cross-device — unlike a
        // conversation turn, which is an op_id-keyed event stream: see the
        // conversation module CRIT-2 tests.)
        let mut a = DeviceLog::new("a");
        let mut b = DeviceLog::new("b");
        let fact = json!({"id": "f1", "body": "hi"});
        let oa = a.append(Scope::Personal, Surface::Knowledge, fact.clone());
        let ob = b.append(Scope::Personal, Surface::Knowledge, fact);

        let just_a = Checkpoint::from_ops(std::slice::from_ref(&oa)).unwrap();
        let both = Checkpoint::from_ops(&[oa, ob]).unwrap();

        assert_eq!(just_a.state, both.state, "cross-device dedup: identical folded state");
        assert_eq!(just_a.state_hash, both.state_hash, "state hash agrees (divergence invariant)");
        assert_ne!(just_a.frontier, both.frontier, "but the frontiers differ");
        assert_ne!(
            just_a.checkpoint_hash, both.checkpoint_hash,
            "so the content addresses MUST differ"
        );
        assert_ne!(just_a.file_name(), both.file_name(), "…and so must the file names");

        // Saving both into one dir stores two files — no false dedup.
        let dir = tempfile::tempdir().unwrap();
        let p1 = just_a.save(dir.path()).unwrap();
        let p2 = both.save(dir.path()).unwrap();
        assert_ne!(p1, p2);
        assert_eq!(Checkpoint::load(&p1).unwrap(), just_a);
        assert_eq!(Checkpoint::load(&p2).unwrap(), both);
    }

    #[test]
    fn verify_anchored_accepts_the_truncated_composition() {
        let (ops, split) = ops_with_cut();
        let ckpt = Checkpoint::from_ops(&ops[..split]).unwrap();
        let tail = &ops[split..];
        verify_log(tail).expect("verify_log alone accepts a non-zero-seq chain");
        verify_anchored(&ckpt, tail).expect("checkpoint anchors the truncated tail");
        // And the composition folds to the full state.
        assert_eq!(fold_onto(&ckpt.state, tail), fold(&ops));
    }

    #[test]
    fn verify_anchored_rejects_breaks() {
        let (ops, split) = ops_with_cut();
        let ckpt = Checkpoint::from_ops(&ops[..split]).unwrap();
        let tail: Vec<OpRecord> = ops[split..].to_vec();

        // A hole right after the cut: drop dev-a's first retained op. Its
        // next op (none here) — instead drop dev-b's anchor continuation:
        // dev-b's retained op is seq 1; removing it leaves only dev-a's,
        // which still anchors — so test by skipping dev-a's op while keeping
        // a later dev-a op. Simplest real break: shift the tail by one op
        // for a device that has more than one retained op.
        let mut a_extra = resume_anchored("dev-a", &ckpt, &tail).unwrap();
        let extra = a_extra.append(Scope::Personal, Surface::Knowledge, json!({"id": "f9"}));
        let mut with_extra = tail.clone();
        with_extra.push(extra.clone());
        verify_anchored(&ckpt, &with_extra).unwrap();

        // Drop dev-a's FIRST retained op but keep the later one → the chain
        // now starts past the anchor: verify_log itself can't see the hole
        // (it only checks contiguity between present ops)… but the anchor
        // check does.
        let holed: Vec<OpRecord> = with_extra
            .iter()
            .filter(|o| o.op_id != tail[0].op_id)
            .cloned()
            .collect();
        assert!(matches!(
            verify_anchored(&ckpt, &holed),
            Err(AnchorError::BrokenAnchor { .. })
        ));

        // A device with no frontier entry must start at seq 0.
        let mut stranger = DeviceLog::new("dev-c");
        stranger.append(Scope::Personal, Surface::Knowledge, json!({"id": "s0"}));
        let s1 = stranger.append(Scope::Personal, Surface::Knowledge, json!({"id": "s1"}));
        let mut with_stranger = tail.clone();
        with_stranger.push(s1); // seq 1, but s0 is missing and no anchor exists
        assert!(matches!(
            verify_anchored(&ckpt, &with_stranger),
            Err(AnchorError::UnanchoredDevice { first_seq: 1, .. })
        ));

        // A tampered checkpoint refuses to anchor anything.
        let mut forged = ckpt.clone();
        forged.state.logs.clear();
        assert!(matches!(
            verify_anchored(&forged, &tail),
            Err(AnchorError::Checkpoint(CheckpointError::HashMismatch { .. }))
        ));
    }

    #[test]
    fn resume_anchored_continues_chains_after_truncation() {
        let (ops, split) = ops_with_cut();
        let ckpt = Checkpoint::from_ops(&ops[..split]).unwrap();
        let tail: Vec<OpRecord> = ops[split..].to_vec();

        // dev-a has a retained op → resumes past it.
        let mut a = resume_anchored("dev-a", &ckpt, &tail).unwrap();
        let next_a = a.append(Scope::Personal, Surface::Knowledge, json!({"id": "na"}));
        assert_eq!(next_a.seq, 3);
        assert_eq!(next_a.prev.as_deref(), Some(tail[0].op_id.as_str()));

        // The critical case: a device whose ops were ALL truncated must
        // resume from the checkpoint anchor, not fork at seq 0.
        let ckpt_all = Checkpoint::from_ops(&ops).unwrap();
        let mut b = resume_anchored("dev-b", &ckpt_all, &[]).unwrap();
        let next_b = b.append(Scope::Personal, Surface::Knowledge, json!({"id": "nb"}));
        assert_eq!(next_b.seq, 2, "continues past the checkpointed chain");
        assert_eq!(next_b.prev.as_deref(), Some(ckpt_all.frontier["dev-b"].head.as_str()));
        assert!(
            next_b.hlc > ckpt_all.frontier["dev-a"].hlc && next_b.hlc > ckpt_all.frontier["dev-b"].hlc,
            "lamport advanced past everything the checkpoint covers"
        );

        // The composed log (anchored tail + new appends) still verifies.
        let mut composed = tail.clone();
        composed.push(next_a);
        verify_anchored(&ckpt, &composed).unwrap();
    }
}