car_sync/

oplog.rs

//! The append-only, replica-tagged operation log.
//!
//! [`OpRecord`] follows `docs/proposals/multi-device-sync.md` §"The frame:
//! sync events, not files" field-for-field, with two B1 specifics:
//!
//! - **`op_id` is content-derived via the shipped B7 discipline**
//!   (`car_proto::deterministic_run_id`'s SHA-256 + `0x1f` field separators;
//!   the proposal's `blake3(payload)` is the same content-addressing idea —
//!   we reuse the hash the codebase already standardized on rather than add a
//!   dependency). The digest covers `device_id ‖ seq ‖ prev ‖ hlc ‖ scope ‖
//!   surface ‖ canonical(payload)`, so the id is simultaneously the natural
//!   dedup key for op *retransmission* AND a tamper-evident cover of the
//!   record, including its position in the device chain. Logical-entity
//!   dedup across devices (the proposal's "conversations dedup on
//!   (speaker,text,timestamp); knowledge on fact_id") happens at the fold's
//!   stable-key level, not on `op_id` — see [`OpRecord::stable_key`].
//!   **Event-stream surfaces are the exception**: routing observations fold
//!   as a MULTISET (the proposal replays "the merged multiset of
//!   observations"), so they key by `op_id` — two byte-identical
//!   observations are two events, and only retransmission dedups. See
//!   [`Surface::is_event_stream`] / [`OpRecord::fold_key`].
//! - **The HLC is shape-only in B1.** [`Hlc`] carries the proposal's
//!   `{wall_ms, counter, device_id}` total order; [`DeviceLog`] stamps pure
//!   Lamport values into `wall_ms` (`counter` stays 0) with the standard
//!   send/receive rules, so nothing in this crate reads a wall clock. B3
//!   replaces the stamp *source* with the true hybrid clock — the wire shape
//!   and the fold are unchanged.
//!
//! Order-verifiability: each op carries a per-device `seq` and the `prev`
//! op_id of the same device's preceding op — a per-device hash chain.
//! [`verify_log`] recomputes every id and walks every chain, so a loaded or
//! received log proves its own order and integrity.
//!
//! **Honesty note — device identity is asserted, not authenticated.** The
//! hash chain proves internal consistency (nothing was reordered or mutated
//! after the fact), but a forger who recomputes the hashes can emit a chain
//! claiming any `device_id` and it will pass [`verify_log`]. Cryptographic
//! device identity (signing ops/checkpoints with a device key) lands with
//! the checkpoint/relay slices (B4/B6); until then, trust in a log's origin
//! comes from the transport that delivered it.

use serde::{Deserialize, Serialize};
use serde_json::Value;
use sha2::{Digest, Sha256};
use std::collections::BTreeMap;
use std::fmt;
use std::sync::Arc;

/// Hybrid-logical-clock stamp — the proposal's `{wall_ms, counter, device_id}`.
/// The derived `Ord` (field order) IS the total order every device agrees on.
/// B1 stamped pure Lamport values into this shape; B3's [`HlcClock`] supplies
/// the real hybrid clock — the wire shape is unchanged, exactly as promised.
#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)]
pub struct Hlc {
    pub wall_ms: u64,
    pub counter: u32,
    pub device_id: String,
}

/// An injectable wall-clock reading (milliseconds since the Unix epoch).
///
/// Library logic never reads the system time directly — the clock is a
/// value the caller hands in (the `run_cascade`/`EffectModel` injection
/// idiom), so tests stay fully deterministic. Production callers pass
/// [`system_clock`]; [`DeviceLog::new`] defaults to [`logical_clock`]
/// (always 0), under which the HLC degenerates to exactly B1's pure
/// Lamport order (the wall component never advances, so every event is a
/// counter tick).
pub type WallClock = Arc<dyn Fn() -> u64 + Send + Sync>;

/// The real wall clock — the ONE place system time enters this crate, and
/// only ever by explicit caller opt-in.
pub fn system_clock() -> WallClock {
    Arc::new(|| {
        std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .map(|d| d.as_millis() as u64)
            .unwrap_or(0)
    })
}

/// A wall clock that never advances (always 0): the HLC's degenerate
/// pure-Lamport mode — B1's stamp semantics, now produced by the same
/// hybrid-clock code path.
///
/// **Counter cap (binding on B6 daemon wiring).** In this mode the wall
/// component is pinned at 0, so *every* event is a counter tick and the
/// `u32` counter never resets — [`HlcClock::tick`] panics after `2^32`
/// events on one device without a wall advance (~4.3 billion). Fine for
/// tests and short-lived tools, but a long-running daemon MUST NOT ship the
/// default: pass [`system_clock`] (or a real monotonic source) via
/// [`DeviceLog::with_wall_clock`], under which the counter resets every
/// millisecond the wall advances and the cap is unreachable in practice.
/// (Daemon wiring is B6; this is the note that keeps the default out of
/// production.)
pub fn logical_clock() -> WallClock {
    Arc::new(|| 0)
}

/// The real hybrid logical clock (B3) — the proposal's `{wall_ms, counter}`
/// state with the standard HLC send/receive rules (Kulkarni et al.):
///
/// - **tick** (local/send event): `l' = max(l, wall_now)`; if the wall
///   didn't advance past everything witnessed, bump the counter, else reset
///   it — the issued stamp is strictly greater than every stamp this clock
///   has issued or observed.
/// - **observe** (receive rule): fold a remote stamp into `(l, c)` as a
///   component-wise max, so the *next* tick lands strictly above it.
///
/// Monotone by construction under clock **skew** (a peer's future stamp is
/// absorbed via `observe`; local ticks ride the counter until the local
/// wall catches up), clock **regression** (a wall reading below `l` is
/// ignored — the counter carries the order), and same-millisecond
/// **bursts** (counter ties, broken across devices by `Hlc::device_id`).
/// The wall component never runs *behind* the physical clock reading it
/// was given, so `wall_ms` stays a meaningful timestamp bounded by the
/// max skew among devices — the "hybrid" in HLC.
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
pub struct HlcClock {
    /// Max wall-clock ms witnessed (own readings and observed stamps).
    l: u64,
    /// Logical tie counter within `l`.
    c: u32,
}

impl HlcClock {
    pub fn new() -> Self {
        Self::default()
    }

    /// Stamp a local event: strictly greater than every stamp previously
    /// issued by or observed on this clock, regardless of what `wall_now`
    /// reads (regression-safe).
    pub fn tick(&mut self, wall_now: u64, device_id: &str) -> Hlc {
        if wall_now > self.l {
            self.l = wall_now;
            self.c = 0;
        } else {
            self.c = self
                .c
                .checked_add(1)
                .expect("HLC counter overflow: > u32::MAX events without wall-clock progress");
        }
        Hlc {
            wall_ms: self.l,
            counter: self.c,
            device_id: device_id.to_string(),
        }
    }

    /// Receive rule: fold an observed stamp so the next [`HlcClock::tick`]
    /// lands strictly above it (and above everything observed before it).
    pub fn observe(&mut self, remote: &Hlc) {
        if remote.wall_ms > self.l {
            self.l = remote.wall_ms;
            self.c = remote.counter;
        } else if remote.wall_ms == self.l && remote.counter > self.c {
            self.c = remote.counter;
        }
    }

    /// The max `(wall_ms, counter)` witnessed so far — the state a stamp
    /// must exceed.
    pub fn witnessed(&self) -> (u64, u32) {
        (self.l, self.c)
    }
}

/// Visibility regime — the proposal's "the `scope` field is the whole answer".
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum Scope {
    /// Replicates only across one user's devices.
    Personal,
    /// Replicates to an org op-stream other members fold in.
    Shared { org: String },
}

impl Scope {
    /// Stable string form used in the `op_id` digest.
    pub fn tag(&self) -> String {
        match self {
            Scope::Personal => "personal".to_string(),
            Scope::Shared { org } => format!("shared:{org}"),
        }
    }
}

/// Which persisted surface an op mutates — the proposal's surfaces.
///
/// **`Intent` (B5) is a leased execution-intent surface.** Its [`Surface::tag`]
/// string `"intent"` enters the `op_id` content digest and is therefore
/// **frozen forever** — changing it would re-address every historical intent
/// op. The enum is deliberately NOT `#[non_exhaustive]`: the repo bans the
/// `_ =>` wildcards that would force, so a new surface variant is a compile
/// error at every match — the intended review gate. The serde wire form stays
/// a tagged union (`#[serde(rename_all = "snake_case")]`).
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum Surface {
    Routing,
    Declagent,
    Conversation,
    Knowledge,
    Skill,
    Registry { kind: String },
    Trajectory,
    Run,
    /// Leased execution-intent ledger (B5): payload
    /// `{id: run_id, agent_id, epoch, status}`. Folds under
    /// [`FoldTier::Leased`] — LWW-per-run_id with monotone status **plus
    /// per-agent epoch fencing**, so a stale-epoch write from a failed-over
    /// lease holder loses deterministically at the fold. See [`crate::lease`].
    Intent,
}

/// How a surface folds — the proposal's per-surface fold-rule tiers.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum FoldTier {
    /// Union by stable ID (conversations, knowledge, skills, trajectories,
    /// runs, routing *observations* — the grow-only tier).
    GrowOnly,
    /// LWW-register per record keyed by id, ordered by HLC (declagents and
    /// the file registries) — NOT per file.
    Registry,
    /// Leased execution-intent tier (`Intent`, B5): LWW-per-run_id with
    /// monotone status, **plus epoch fencing** — an intent whose `epoch` is
    /// below its agent's max-seen epoch is dropped at the fold (a fenced
    /// zombie writer after a lease failover loses deterministically, with no
    /// wall-clock race). NOT grow-only. See [`crate::lease`]/[`crate::fold`].
    Leased,
}

impl Surface {
    /// Stable string form: the fold's grouping key and part of the `op_id`
    /// digest.
    pub fn tag(&self) -> String {
        match self {
            Surface::Routing => "routing".to_string(),
            Surface::Declagent => "declagent".to_string(),
            Surface::Conversation => "conversation".to_string(),
            Surface::Knowledge => "knowledge".to_string(),
            Surface::Skill => "skill".to_string(),
            Surface::Registry { kind } => format!("registry:{kind}"),
            Surface::Trajectory => "trajectory".to_string(),
            Surface::Run => "run".to_string(),
            // FROZEN (B5): part of the op_id digest — never change this string.
            Surface::Intent => "intent".to_string(),
        }
    }

    /// The proposal's fold-rule table. Routing observations are grow-only
    /// log entries ("sync the observations, not the result"); the EMA replay
    /// over them is the caller-injected [`crate::fold::SyncState::replay`].
    pub fn fold_tier(&self) -> FoldTier {
        match self {
            Surface::Conversation
            | Surface::Knowledge
            | Surface::Skill
            | Surface::Trajectory
            | Surface::Run
            | Surface::Routing => FoldTier::GrowOnly,
            Surface::Declagent | Surface::Registry { .. } => FoldTier::Registry,
            Surface::Intent => FoldTier::Leased,
        }
    }

    /// Is this surface an **event stream** — a multiset keyed by `op_id`
    /// rather than a set of content-deduped logical entities?
    ///
    /// Two surfaces are event streams, for the same structural reason but with
    /// different downstream handling:
    ///
    /// - **Routing** — the proposal's "replays the merged **multiset** of
    ///   observations": `agent x succeeded` twice is two events that must both
    ///   reach the EMA replay.
    /// - **Conversation** (B2, kernel-review correction) — a conversation turn
    ///   has exactly one author and propagates by op replication, so **op
    ///   identity IS turn identity**. Content-keying was a reproduced
    ///   data-loss bug: two genuine "yes" turns stamped at the same
    ///   payload-second (cached `now()`, rapid double-confirm) fold to one
    ///   entry. Keyed by `op_id`, a *resent* op dedups but two *distinct*
    ///   authorings never collapse. Unlike routing, conversation turns are
    ///   **independent** entries (no path-dependent replay), so they tolerate
    ///   `LastN` retention — see [`Surface::is_replay_stream`].
    ///
    /// Event-stream surfaces fold keyed by `op_id` — see [`OpRecord::fold_key`].
    pub fn is_event_stream(&self) -> bool {
        match self {
            Surface::Routing | Surface::Conversation => true,
            Surface::Declagent
            | Surface::Knowledge
            | Surface::Skill
            | Surface::Registry { .. }
            | Surface::Trajectory
            | Surface::Run
            | Surface::Intent => false,
        }
    }

    /// Is this surface a **path-dependent replay** stream — one whose folded
    /// result is recomputed from the ordered multiset (routing's EMA), so that
    /// dropping ANY entry corrupts every device's recomputed value? Only these
    /// are retention-forbidden (`compact` rejects any non-keep-all rule on
    /// them). This is **narrower than [`Surface::is_event_stream`]**:
    /// conversation is an event-stream multiset too, but its turns are
    /// independent, so `LastN` over them is well-defined and allowed. Routing
    /// is the only replay stream today; a new one is a compile-error here (no
    /// `_` arm), the intended review gate.
    pub fn is_replay_stream(&self) -> bool {
        match self {
            Surface::Routing => true,
            Surface::Declagent
            | Surface::Conversation
            | Surface::Knowledge
            | Surface::Skill
            | Surface::Registry { .. }
            | Surface::Trajectory
            | Surface::Run
            | Surface::Intent => false,
        }
    }
}

/// One state-changing operation in the oplog.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct OpRecord {
    /// Content-derived id (see module docs): dedup key for retransmission
    /// and tamper-evident cover of the whole record.
    pub op_id: String,
    /// Total-order stamp. Invariant: `hlc.device_id == device_id`.
    pub hlc: Hlc,
    /// The device (replica) that emitted the op. Matches the `replica`
    /// strings `car_state::crdt` already uses.
    pub device_id: String,
    /// Per-device append index (0-based, contiguous).
    pub seq: u64,
    /// `op_id` of this device's previous op (`None` iff `seq == 0`) — the
    /// per-device hash-chain link that makes the log order-verifiable.
    pub prev: Option<String>,
    pub scope: Scope,
    pub surface: Surface,
    /// Surface-specific payload (possibly E2E ciphertext in B6).
    pub payload: Value,
}

/// Canonical, key-sorted, compact JSON — the deterministic serialization the
/// `op_id` digest and [`crate::fold::state_hash`] are computed over.
/// Independent of `serde_json`'s map-ordering configuration.
pub fn canonical_json(v: &Value) -> String {
    match v {
        Value::Object(map) => {
            let mut keys: Vec<&String> = map.keys().collect();
            keys.sort();
            let inner: Vec<String> = keys
                .iter()
                .map(|k| {
                    format!(
                        "{}:{}",
                        serde_json::to_string(k).expect("string serializes"),
                        canonical_json(&map[k.as_str()])
                    )
                })
                .collect();
            format!("{{{}}}", inner.join(","))
        }
        Value::Array(items) => {
            let inner: Vec<String> = items.iter().map(canonical_json).collect();
            format!("[{}]", inner.join(","))
        }
        _ => serde_json::to_string(v).unwrap_or_default(),
    }
}

// NOTE: this reimplements car-proto's B7 content-address discipline
// (`deterministic_run_id`: SHA-256, 0x1f separators, 16-byte/32-hex prefix)
// rather than depending on car-proto, which would drag the whole protocol
// crate into this dependency-light core. Consolidating the discipline into a
// shared home (car-proto exporting just the hasher, or a tiny common crate)
// is a next-slice cleanup — keep the two in step until then.
fn sha256_hex_32(fields: &[&str]) -> String {
    let mut hasher = Sha256::new();
    for (i, f) in fields.iter().enumerate() {
        if i > 0 {
            hasher.update(b"\x1f"); // B7's field separator — no concat collisions
        }
        hasher.update(f.as_bytes());
    }
    let digest = hasher.finalize();
    digest.iter().take(16).map(|b| format!("{b:02x}")).collect()
}

impl OpRecord {
    /// Build an op and stamp its content-derived id. Callers normally go
    /// through [`DeviceLog::append`], which manages `seq`/`prev`/`hlc`.
    pub fn new(
        hlc: Hlc,
        seq: u64,
        prev: Option<String>,
        scope: Scope,
        surface: Surface,
        payload: Value,
    ) -> Self {
        let device_id = hlc.device_id.clone();
        let mut op = OpRecord {
            op_id: String::new(),
            hlc,
            device_id,
            seq,
            prev,
            scope,
            surface,
            payload,
        };
        op.op_id = op.compute_op_id();
        op
    }

    /// Recompute the content-derived id from the record's fields (the B7
    /// SHA-256 + `0x1f` discipline; `op-` + 32 hex chars).
    pub fn compute_op_id(&self) -> String {
        let hex = sha256_hex_32(&[
            &self.device_id,
            &self.seq.to_string(),
            self.prev.as_deref().unwrap_or(""),
            &self.hlc.wall_ms.to_string(),
            &self.hlc.counter.to_string(),
            &self.hlc.device_id,
            &self.scope.tag(),
            &self.surface.tag(),
            &canonical_json(&self.payload),
        ]);
        format!("op-{hex}")
    }

    /// Does the stored id match the record's content?
    pub fn id_valid(&self) -> bool {
        self.op_id == self.compute_op_id()
    }

    /// The stable key the fold dedups logical entities on: the payload's
    /// `"id"` string when present (the proposal's `fact_id` / record-id
    /// keys), else the canonical content hash (which realizes e.g. the
    /// conversation `(speaker,text,timestamp)` dedup — identical content is
    /// one entity). Prefixed so the two forms can never collide.
    ///
    /// NOT used for event-stream surfaces (routing) — the fold keys those by
    /// `op_id` via [`OpRecord::fold_key`], because an event stream is a
    /// multiset: identical content is two events, not one entity.
    pub fn stable_key(&self) -> String {
        match self.payload.get("id").and_then(Value::as_str) {
            Some(id) => format!("id:{id}"),
            None => format!("h:{}", sha256_hex_32(&[&canonical_json(&self.payload)])),
        }
    }

    /// The key the fold stores this op under: [`OpRecord::stable_key`] for
    /// logical-entity surfaces, `op_id` for event-stream surfaces
    /// ([`Surface::is_event_stream`] — the proposal's routing MULTISET:
    /// every emitted observation survives the fold; only retransmission of
    /// the *same* op dedups). Prefixes keep the three key forms (`id:`,
    /// `h:`, `op:`) disjoint.
    pub fn fold_key(&self) -> String {
        if self.surface.is_event_stream() {
            format!("op:{}", self.op_id)
        } else {
            self.stable_key()
        }
    }
}

/// A chain-verification failure from [`verify_log`].
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum ChainError {
    /// A record's stored `op_id` doesn't match its content.
    IdMismatch { op_id: String },
    /// `hlc.device_id` disagrees with the record's `device_id`.
    DeviceMismatch { op_id: String },
    /// Two ops from one device claim the same `seq`.
    DuplicateSeq { device_id: String, seq: u64 },
    /// A device's seqs aren't contiguous from its first present op.
    SeqGap { device_id: String, expected: u64, found: u64 },
    /// `prev` doesn't link to the device's preceding op (or `seq 0` has one).
    PrevMismatch { op_id: String },
    /// A device's HLC stamps aren't strictly increasing along its chain.
    NonMonotonicHlc { op_id: String },
    /// [`DeviceLog::resume`] was handed a log in which the resuming
    /// device's own chain doesn't start at `seq 0` — a truncated tail.
    /// Resuming from it would re-mint truncated seqs (a permanent chain
    /// fork); go through `checkpoint::resume_anchored` instead.
    TruncatedChain { device_id: String, first_seq: u64 },
}

impl fmt::Display for ChainError {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            ChainError::IdMismatch { op_id } => {
                write!(f, "op {op_id}: stored op_id does not match content")
            }
            ChainError::DeviceMismatch { op_id } => {
                write!(f, "op {op_id}: hlc.device_id != device_id")
            }
            ChainError::DuplicateSeq { device_id, seq } => {
                write!(f, "device {device_id}: duplicate seq {seq}")
            }
            ChainError::SeqGap { device_id, expected, found } => {
                write!(f, "device {device_id}: seq gap (expected {expected}, found {found})")
            }
            ChainError::PrevMismatch { op_id } => {
                write!(f, "op {op_id}: prev does not link to the preceding op")
            }
            ChainError::NonMonotonicHlc { op_id } => {
                write!(f, "op {op_id}: hlc not strictly increasing along device chain")
            }
            ChainError::TruncatedChain { device_id, first_seq } => write!(
                f,
                "device {device_id}: own chain starts at seq {first_seq} (truncated tail) — \
                 DeviceLog::resume would fork the chain; resume via checkpoint::resume_anchored"
            ),
        }
    }
}

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

/// Verify a log's integrity and order: every id recomputes, and every
/// device's ops form a contiguous, `prev`-linked, HLC-monotone chain from
/// the first op present for that device (a checkpointed log need not start
/// at `seq 0`, but if `seq 0` is present its `prev` must be `None`).
pub fn verify_log(ops: &[OpRecord]) -> Result<(), ChainError> {
    let mut by_device: BTreeMap<&str, BTreeMap<u64, &OpRecord>> = BTreeMap::new();
    for op in ops {
        if !op.id_valid() {
            return Err(ChainError::IdMismatch { op_id: op.op_id.clone() });
        }
        if op.hlc.device_id != op.device_id {
            return Err(ChainError::DeviceMismatch { op_id: op.op_id.clone() });
        }
        if by_device
            .entry(&op.device_id)
            .or_default()
            .insert(op.seq, op)
            .is_some()
        {
            return Err(ChainError::DuplicateSeq {
                device_id: op.device_id.clone(),
                seq: op.seq,
            });
        }
    }
    for (device_id, chain) in by_device {
        let mut prev_op: Option<&OpRecord> = None;
        for (&seq, op) in &chain {
            match prev_op {
                None => {
                    if seq == 0 && op.prev.is_some() {
                        return Err(ChainError::PrevMismatch { op_id: op.op_id.clone() });
                    }
                }
                Some(previous) => {
                    if seq != previous.seq + 1 {
                        return Err(ChainError::SeqGap {
                            device_id: device_id.to_string(),
                            expected: previous.seq + 1,
                            found: seq,
                        });
                    }
                    if op.prev.as_deref() != Some(previous.op_id.as_str()) {
                        return Err(ChainError::PrevMismatch { op_id: op.op_id.clone() });
                    }
                    if op.hlc <= previous.hlc {
                        return Err(ChainError::NonMonotonicHlc { op_id: op.op_id.clone() });
                    }
                }
            }
            prev_op = Some(op);
        }
    }
    Ok(())
}

/// The per-device append discipline: maintains the `seq`/`prev` chain and
/// stamps [`Hlc`] values from an [`HlcClock`] over an injected [`WallClock`]
/// — B3's real hybrid clock, replacing B1's pure-Lamport stamp source
/// behind the same wire shape.
///
/// [`DeviceLog::new`] defaults the wall source to [`logical_clock`]
/// (always 0), under which the HLC *is* a Lamport clock (every tick is a
/// counter increment) — B1 semantics as the degenerate case of one code
/// path. Real deployments pass [`system_clock`] (or a test-controlled
/// closure) via [`DeviceLog::with_wall_clock`] / [`DeviceLog::set_wall_clock`].
#[derive(Clone)]
pub struct DeviceLog {
    pub(crate) device_id: String,
    pub(crate) next_seq: u64,
    pub(crate) prev: Option<String>,
    pub(crate) clock: HlcClock,
    wall: WallClock,
}

impl fmt::Debug for DeviceLog {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("DeviceLog")
            .field("device_id", &self.device_id)
            .field("next_seq", &self.next_seq)
            .field("prev", &self.prev)
            .field("clock", &self.clock)
            .finish_non_exhaustive() // the wall closure has no useful Debug
    }
}

impl DeviceLog {
    pub fn new(device_id: impl Into<String>) -> Self {
        Self::with_wall_clock(device_id, logical_clock())
    }

    /// A device log stamping the real HLC over the given wall source
    /// (pass [`system_clock`] in production, a controlled closure in tests).
    pub fn with_wall_clock(device_id: impl Into<String>, wall: WallClock) -> Self {
        Self {
            device_id: device_id.into(),
            next_seq: 0,
            prev: None,
            clock: HlcClock::new(),
            wall,
        }
    }

    /// Swap the wall source on a live log (e.g. after a
    /// [`DeviceLog::resume`], which has no wall parameter). Monotonicity is
    /// unaffected: the [`HlcClock`] never regresses below what it has
    /// witnessed, whatever the new source reads.
    pub fn set_wall_clock(&mut self, wall: WallClock) {
        self.wall = wall;
    }

    /// Resume a device's chain from previously persisted ops (e.g. after
    /// [`crate::journal::OplogJournal::load`]): verifies the log, adopts this
    /// device's chain tail, and advances the clock past **every** op
    /// present (local and remote), so new appends stamp above all of them.
    /// The resumed log defaults to the [`logical_clock`] wall source — call
    /// [`DeviceLog::set_wall_clock`] to attach the real one.
    ///
    /// **MUST: an op is journal-durable before it is transmitted.** Resume
    /// derives `next_seq` from the journal; if a crash lands between
    /// "op sent to a peer/relay" and "op durably journaled", the resumed
    /// device re-mints that `seq` for a *different* op, and the union of the
    /// two logs is a permanent `DuplicateSeq`/`PrevMismatch` — an
    /// unrecoverable fork of the device's chain. Always
    /// `OplogJournal::append` (which flushes) before handing an op to any
    /// transport (B3 must preserve this ordering).
    ///
    /// **Fenced against truncated tails (B4).** If the resuming device's
    /// own chain doesn't start at `seq 0`, the ops are a truncated tail and
    /// resume refuses ([`ChainError::TruncatedChain`]) — the anchored
    /// sibling `checkpoint::resume_anchored` is the correct path. (The case
    /// this check can't see — a device whose ops were ALL truncated away —
    /// is fenced one layer down: `OplogJournal::load` refuses a journal
    /// carrying a truncation marker.)
    pub fn resume(device_id: impl Into<String>, ops: &[OpRecord]) -> Result<Self, ChainError> {
        verify_log(ops)?;
        let device_id = device_id.into();
        if let Some(first_seq) = ops
            .iter()
            .filter(|op| op.device_id == device_id)
            .map(|op| op.seq)
            .min()
        {
            if first_seq > 0 {
                return Err(ChainError::TruncatedChain { device_id, first_seq });
            }
        }
        let mut log = Self::new(device_id.clone());
        for op in ops {
            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)
    }

    /// HLC receive rule: fold a received op's stamp into the local clock,
    /// so a write that causally follows received ops stamps above them.
    pub fn observe(&mut self, hlc: &Hlc) {
        self.clock.observe(hlc);
    }

    /// Append a new op: read the wall, tick the hybrid clock, stamp, link
    /// the chain.
    pub fn append(&mut self, scope: Scope, surface: Surface, payload: Value) -> OpRecord {
        let hlc = self.clock.tick((self.wall)(), &self.device_id);
        let op = OpRecord::new(hlc, self.next_seq, self.prev.take(), scope, surface, payload);
        self.next_seq += 1;
        self.prev = Some(op.op_id.clone());
        op
    }

    pub fn device_id(&self) -> &str {
        &self.device_id
    }
}

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

    #[test]
    fn op_id_is_deterministic_and_content_derived() {
        let mk = || {
            OpRecord::new(
                Hlc { wall_ms: 7, counter: 0, device_id: "d1".into() },
                0,
                None,
                Scope::Personal,
                Surface::Knowledge,
                json!({"id": "f1", "body": "x"}),
            )
        };
        let a = mk();
        let b = mk();
        assert_eq!(a.op_id, b.op_id, "identical content → identical id");
        assert!(a.op_id.starts_with("op-"));
        assert_eq!(a.op_id.len(), 3 + 32);
        assert!(a.id_valid());
    }

    #[test]
    fn op_id_covers_every_field() {
        let base = OpRecord::new(
            Hlc { wall_ms: 7, counter: 0, device_id: "d1".into() },
            1,
            Some("op-0".into()),
            Scope::Personal,
            Surface::Knowledge,
            json!({"id": "f1"}),
        );
        let variants = [
            OpRecord::new(
                Hlc { wall_ms: 8, counter: 0, device_id: "d1".into() },
                1,
                Some("op-0".into()),
                Scope::Personal,
                Surface::Knowledge,
                json!({"id": "f1"}),
            ),
            OpRecord::new(
                Hlc { wall_ms: 7, counter: 0, device_id: "d1".into() },
                2,
                Some("op-0".into()),
                Scope::Personal,
                Surface::Knowledge,
                json!({"id": "f1"}),
            ),
            OpRecord::new(
                Hlc { wall_ms: 7, counter: 0, device_id: "d1".into() },
                1,
                Some("op-1".into()),
                Scope::Personal,
                Surface::Knowledge,
                json!({"id": "f1"}),
            ),
            OpRecord::new(
                Hlc { wall_ms: 7, counter: 0, device_id: "d1".into() },
                1,
                Some("op-0".into()),
                Scope::Shared { org: "acme".into() },
                Surface::Knowledge,
                json!({"id": "f1"}),
            ),
            OpRecord::new(
                Hlc { wall_ms: 7, counter: 0, device_id: "d1".into() },
                1,
                Some("op-0".into()),
                Scope::Personal,
                Surface::Skill,
                json!({"id": "f1"}),
            ),
            OpRecord::new(
                Hlc { wall_ms: 7, counter: 0, device_id: "d1".into() },
                1,
                Some("op-0".into()),
                Scope::Personal,
                Surface::Knowledge,
                json!({"id": "f2"}),
            ),
        ];
        for v in &variants {
            assert_ne!(base.op_id, v.op_id, "changing any field changes the id");
        }
    }

    #[test]
    fn canonical_json_is_key_order_independent() {
        // parse two spellings of the same object
        let a: Value = serde_json::from_str(r#"{"b":1,"a":{"y":2,"x":3}}"#).unwrap();
        let b: Value = serde_json::from_str(r#"{"a":{"x":3,"y":2},"b":1}"#).unwrap();
        assert_eq!(canonical_json(&a), canonical_json(&b));
        assert_eq!(canonical_json(&a), r#"{"a":{"x":3,"y":2},"b":1}"#);
    }

    #[test]
    fn surface_tags_and_tiers_are_exhaustive() {
        let surfaces = [
            (Surface::Routing, "routing", FoldTier::GrowOnly),
            (Surface::Declagent, "declagent", FoldTier::Registry),
            (Surface::Conversation, "conversation", FoldTier::GrowOnly),
            (Surface::Knowledge, "knowledge", FoldTier::GrowOnly),
            (Surface::Skill, "skill", FoldTier::GrowOnly),
            (Surface::Registry { kind: "agents".into() }, "registry:agents", FoldTier::Registry),
            (Surface::Trajectory, "trajectory", FoldTier::GrowOnly),
            (Surface::Run, "run", FoldTier::GrowOnly),
            (Surface::Intent, "intent", FoldTier::Leased),
        ];
        for (s, tag, tier) in surfaces {
            assert_eq!(s.tag(), tag);
            assert_eq!(s.fold_tier(), tier);
            // Intent is a logical-entity ledger, not an observation multiset.
            assert!(!Surface::Intent.is_event_stream());
        }
        // Event streams (op_id-keyed multisets): routing AND conversation (B2 —
        // op identity is turn identity). Only routing is a path-dependent
        // REPLAY stream (retention-forbidden); conversation is an independent
        // multiset that tolerates LastN.
        assert!(Surface::Routing.is_event_stream() && Surface::Routing.is_replay_stream());
        assert!(Surface::Conversation.is_event_stream());
        assert!(!Surface::Conversation.is_replay_stream());
        assert!(!Surface::Knowledge.is_event_stream());
    }

    #[test]
    fn tampering_is_detected() {
        let mut log = DeviceLog::new("d1");
        let mut ops = vec![
            log.append(Scope::Personal, Surface::Knowledge, json!({"id": "f1"})),
            log.append(Scope::Personal, Surface::Knowledge, json!({"id": "f2"})),
        ];
        verify_log(&ops).unwrap();
        // Mutate a payload without recomputing the id.
        ops[1].payload = json!({"id": "f2", "body": "forged"});
        assert!(matches!(verify_log(&ops), Err(ChainError::IdMismatch { .. })));
    }

    #[test]
    fn chain_defects_are_detected() {
        let mut log = DeviceLog::new("d1");
        let o0 = log.append(Scope::Personal, Surface::Knowledge, json!({"id": "a"}));
        let o1 = log.append(Scope::Personal, Surface::Knowledge, json!({"id": "b"}));
        let o2 = log.append(Scope::Personal, Surface::Knowledge, json!({"id": "c"}));
        verify_log(&[o0.clone(), o1.clone(), o2.clone()]).unwrap();

        // Missing middle op → seq gap.
        assert!(matches!(
            verify_log(&[o0.clone(), o2.clone()]),
            Err(ChainError::SeqGap { expected: 1, found: 2, .. })
        ));

        // prev link forged (re-id'd so IdMismatch doesn't fire first).
        let forged = OpRecord::new(
            o1.hlc.clone(),
            o1.seq,
            Some(o2.op_id.clone()), // wrong parent
            o1.scope.clone(),
            o1.surface.clone(),
            o1.payload.clone(),
        );
        assert!(matches!(
            verify_log(&[o0.clone(), forged, o2.clone()]),
            Err(ChainError::PrevMismatch { .. })
        ));

        // hlc going backwards along the chain.
        let backwards = OpRecord::new(
            Hlc { wall_ms: 0, counter: 0, device_id: "d1".into() },
            o1.seq,
            Some(o0.op_id.clone()),
            o1.scope.clone(),
            o1.surface.clone(),
            o1.payload.clone(),
        );
        assert!(matches!(
            verify_log(&[o0.clone(), backwards]),
            Err(ChainError::NonMonotonicHlc { .. })
        ));

        // seq 0 with a parent.
        let rooted = OpRecord::new(
            o0.hlc.clone(),
            0,
            Some(o2.op_id.clone()),
            o0.scope.clone(),
            o0.surface.clone(),
            o0.payload.clone(),
        );
        assert!(matches!(verify_log(&[rooted]), Err(ChainError::PrevMismatch { .. })));

        // duplicate seq.
        let dup = OpRecord::new(
            Hlc { wall_ms: 99, counter: 0, device_id: "d1".into() },
            o1.seq,
            Some(o0.op_id.clone()),
            o1.scope.clone(),
            o1.surface.clone(),
            json!({"id": "dup"}),
        );
        assert!(matches!(
            verify_log(&[o0.clone(), o1.clone(), dup]),
            Err(ChainError::DuplicateSeq { seq: 1, .. })
        ));

        // hlc.device_id disagreeing with device_id.
        let mut cross = o0.clone();
        cross.hlc.device_id = "d2".into();
        cross.op_id = cross.compute_op_id();
        assert!(matches!(verify_log(&[cross]), Err(ChainError::DeviceMismatch { .. })));
    }

    #[test]
    fn observe_advances_clock_past_received_ops() {
        let mut a = DeviceLog::new("a");
        let mut b = DeviceLog::new("b");
        let oa = a.append(Scope::Personal, Surface::Knowledge, json!({"id": "x"}));
        // Degenerate (logical_clock) mode: pure Lamport in the counter,
        // wall component pinned at 0 — B1's order semantics, same wire shape.
        assert_eq!(oa.hlc, Hlc { wall_ms: 0, counter: 1, device_id: "a".into() });
        b.observe(&oa.hlc);
        let ob = b.append(Scope::Personal, Surface::Knowledge, json!({"id": "y"}));
        assert!(ob.hlc > oa.hlc, "causally-later write stamps above the observed op");
    }

    /// A test wall clock the test advances (or regresses) by hand.
    fn manual_clock() -> (std::sync::Arc<std::sync::atomic::AtomicU64>, WallClock) {
        let t = std::sync::Arc::new(std::sync::atomic::AtomicU64::new(0));
        let reader = t.clone();
        let wall: WallClock =
            Arc::new(move || reader.load(std::sync::atomic::Ordering::SeqCst));
        (t, wall)
    }

    #[test]
    fn hlc_is_monotone_under_wall_clock_regression() {
        use std::sync::atomic::Ordering;
        let (t, wall) = manual_clock();
        let mut dev = DeviceLog::with_wall_clock("d1", wall);
        t.store(100, Ordering::SeqCst);
        let o1 = dev.append(Scope::Personal, Surface::Knowledge, json!({"id": "a"}));
        assert_eq!((o1.hlc.wall_ms, o1.hlc.counter), (100, 0));

        // The wall clock jumps BACKWARDS (NTP step, VM restore): stamps keep
        // strictly increasing on the counter, wall pinned at the max seen.
        t.store(40, Ordering::SeqCst);
        let o2 = dev.append(Scope::Personal, Surface::Knowledge, json!({"id": "b"}));
        let o3 = dev.append(Scope::Personal, Surface::Knowledge, json!({"id": "c"}));
        assert_eq!((o2.hlc.wall_ms, o2.hlc.counter), (100, 1));
        assert_eq!((o3.hlc.wall_ms, o3.hlc.counter), (100, 2));
        assert!(o1.hlc < o2.hlc && o2.hlc < o3.hlc);

        // The wall recovers past the pinned max: counter resets.
        t.store(200, Ordering::SeqCst);
        let o4 = dev.append(Scope::Personal, Surface::Knowledge, json!({"id": "d"}));
        assert_eq!((o4.hlc.wall_ms, o4.hlc.counter), (200, 0));
        verify_log(&[o1, o2, o3, o4]).expect("regression-spanning chain stays HLC-monotone");
    }

    #[test]
    fn hlc_burst_within_one_millisecond_stays_strictly_ordered() {
        use std::sync::atomic::Ordering;
        let (t, wall) = manual_clock();
        let mut dev = DeviceLog::with_wall_clock("d1", wall);
        t.store(555, Ordering::SeqCst);
        let ops: Vec<OpRecord> = (0..50)
            .map(|i| dev.append(Scope::Personal, Surface::Routing, json!({"n": i})))
            .collect();
        for (i, op) in ops.iter().enumerate() {
            assert_eq!(op.hlc.wall_ms, 555);
            assert_eq!(op.hlc.counter, i as u32, "burst rides the counter");
        }
        verify_log(&ops).unwrap();
    }

    #[test]
    fn hlc_absorbs_skewed_peer_stamps_and_preserves_causality() {
        use std::sync::atomic::Ordering;
        // Device b's wall clock runs far BEHIND device a's (skew), yet a
        // write on b that causally follows a's op must stamp above it.
        let (ta, wall_a) = manual_clock();
        let (tb, wall_b) = manual_clock();
        let mut a = DeviceLog::with_wall_clock("a", wall_a);
        let mut b = DeviceLog::with_wall_clock("b", wall_b);
        ta.store(10_000, Ordering::SeqCst);
        tb.store(3, Ordering::SeqCst); // b is ~10s behind

        let oa = a.append(Scope::Personal, Surface::Knowledge, json!({"id": "x"}));
        b.observe(&oa.hlc); // receive rule: absorb the future stamp
        let ob = b.append(Scope::Personal, Surface::Knowledge, json!({"id": "y"}));
        assert!(ob.hlc > oa.hlc, "causality survives a 10s skew");
        assert_eq!(ob.hlc.wall_ms, 10_000, "wall pinned at the max witnessed, not b's slow clock");
        assert_eq!(ob.hlc.counter, 1);

        // …and once b's wall genuinely passes the witnessed max, the wall
        // component takes over again (the 'hybrid' half).
        tb.store(20_000, Ordering::SeqCst);
        let ob2 = b.append(Scope::Personal, Surface::Knowledge, json!({"id": "z"}));
        assert_eq!((ob2.hlc.wall_ms, ob2.hlc.counter), (20_000, 0));
        verify_log(&[ob, ob2]).unwrap();
    }

    #[test]
    fn hlc_wire_shape_is_unchanged_from_b1() {
        // The B1→B3 promise: swapping the stamp source changes no wire bytes.
        let hlc = Hlc { wall_ms: 7, counter: 2, device_id: "d1".into() };
        assert_eq!(
            serde_json::to_value(&hlc).unwrap(),
            json!({"wall_ms": 7, "counter": 2, "device_id": "d1"})
        );
    }

    #[test]
    fn resume_adopts_witnessed_stamps_under_a_real_clock() {
        use std::sync::atomic::Ordering;
        let (t, wall) = manual_clock();
        t.store(500, Ordering::SeqCst);
        let mut dev = DeviceLog::with_wall_clock("d1", wall.clone());
        let ops = vec![
            dev.append(Scope::Personal, Surface::Knowledge, json!({"id": "a"})),
            dev.append(Scope::Personal, Surface::Knowledge, json!({"id": "b"})),
        ];
        // Restart: resume from the journal, re-attach the (now regressed)
        // wall — the next stamp still lands above everything persisted.
        t.store(100, Ordering::SeqCst);
        let mut resumed = DeviceLog::resume("d1", &ops).unwrap();
        resumed.set_wall_clock(wall);
        let next = resumed.append(Scope::Personal, Surface::Knowledge, json!({"id": "c"}));
        assert!(next.hlc > ops[1].hlc);
        let mut all = ops;
        all.push(next);
        verify_log(&all).unwrap();
    }

    #[test]
    fn resume_continues_the_chain() {
        let mut log = DeviceLog::new("d1");
        let mut peer = DeviceLog::new("d2");
        let ops = vec![
            log.append(Scope::Personal, Surface::Knowledge, json!({"id": "a"})),
            log.append(Scope::Personal, Surface::Knowledge, json!({"id": "b"})),
            peer.append(Scope::Personal, Surface::Knowledge, json!({"id": "c"})),
        ];
        let mut resumed = DeviceLog::resume("d1", &ops).unwrap();
        let next = resumed.append(Scope::Personal, Surface::Knowledge, json!({"id": "d"}));
        assert_eq!(next.seq, 2);
        assert_eq!(next.prev.as_deref(), Some(ops[1].op_id.as_str()));
        let mut all = ops;
        all.push(next);
        verify_log(&all).unwrap();
    }

    #[test]
    fn stable_key_uses_payload_id_else_content_hash() {
        let mut log = DeviceLog::new("d1");
        let with_id = log.append(Scope::Personal, Surface::Knowledge, json!({"id": "f1", "v": 1}));
        assert_eq!(with_id.stable_key(), "id:f1");
        // `stable_key` is the logical-ENTITY key (knowledge/skills/registries):
        // an id-less payload keys on its canonical content hash, key-order
        // independent, so the same fact emitted by two devices dedups.
        // (Conversation is an event stream — B2 — so it does NOT use stable_key
        // in the fold; it keys on op_id. See fold_key / is_event_stream.)
        let anon1 = log.append(
            Scope::Personal,
            Surface::Knowledge,
            json!({"kind": "note", "body": "hi"}),
        );
        let anon2 = OpRecord::new(
            Hlc { wall_ms: 42, counter: 0, device_id: "d2".into() },
            0,
            None,
            Scope::Personal,
            Surface::Knowledge,
            json!({"body": "hi", "kind": "note"}),
        );
        assert_eq!(anon1.stable_key(), anon2.stable_key());
        assert!(anon1.stable_key().starts_with("h:"));
    }

    #[test]
    fn op_record_serde_round_trips() {
        let mut log = DeviceLog::new("d1");
        let op = log.append(
            Scope::Shared { org: "acme".into() },
            Surface::Registry { kind: "agents".into() },
            json!({"id": "agent-1"}),
        );
        let json = serde_json::to_string(&op).unwrap();
        let back: OpRecord = serde_json::from_str(&json).unwrap();
        assert_eq!(back, op);
        assert!(back.id_valid());
    }
}