car-sync 0.33.0

//! The device-side sync session — the pump that drives a [`DeviceLog`] +
//! [`OplogJournal`] pair against a [`Relay`] (slice B3 of
//! `docs/proposals/multi-device-sync.md`).
//!
//! [`SyncSession::pump`] is one reconciliation round, sequenced so the
//! binding contracts from B1/B4 hold **by construction**, not by caller
//! discipline:
//!
//! 1. **Push** — only ops read from (or appended through) the journal are
//!    ever handed to the relay, and the journal is **fsync'd**
//!    ([`OplogJournal::sync`]) before the push, so **an op is journal-durable
//!    before it is transmitted** (B1 MUST — `flush` is only the page cache;
//!    a power loss in the writeback window would otherwise lose a
//!    transmitted op and re-mint its seq into a permanent Fork). A crash
//!    that loses the push cursor is harmless — a re-push dedups relay-side
//!    on `op_id`.
//! 2. **Pull** — `pull(since = my per-device seq frontier)`, then
//!    **verify before fold** (B1 MUST): the union of held + pulled ops
//!    must pass [`verify_log`] (or [`verify_anchored`] against the
//!    session's base checkpoint) before anything is folded or journaled.
//! 3. **Fold durably, then ack** — verified remote ops are appended to the
//!    journal, **advancing `self.ops` in lockstep** (each op enters
//!    `self.ops` the instant its append succeeds, so a mid-loop failure
//!    leaves `self.ops == journal` and the retry re-pull filters the
//!    already-journaled ops instead of duplicating them — a duplicate line
//!    would `DuplicateSeq`-brick the next open). The journal is then fsync'd
//!    and only then is `ack` sent. The ack value is *derived from the
//!    journal-held ops* — there is no API to ack anything else, so **acking
//!    merely-received (un-journaled) state is impossible by construction**
//!    (B4 MUST). A crash between the fold and the ack leaves the relay's ack
//!    table behind — the safe direction: GC can't drop what we haven't
//!    acked, and the next pump re-acks.
//!
//! Idempotence: re-running `pump` after any crash point re-pushes
//! (relay dedups by `op_id`), re-pulls (already-held ops are filtered by
//! `op_id`; re-folding is a no-op), and re-acks (monotone) — the whole
//! round is retry-safe.
//!
//! **Cold bootstrap / straggler re-entry** ([`SyncSession::bootstrap`] /
//! [`SyncSession::rebase`]): when the relay has GC'd past the session's
//! frontier (`RelayError::FrontierTruncated`) or the device is brand new,
//! the path is exactly the proposal's — `checkpoint_get()` → `pull(since =
//! checkpoint frontier)` → verify → journal rewritten as checkpoint-anchored
//! tail (`truncate_to`, stamping the truncation marker so the naive
//! `load`/`resume` stays fenced — B4 contract 5) → **`resume_anchored`**,
//! never `DeviceLog::resume`. Local ops **not covered** by the checkpoint —
//! including a returning straggler's never-pushed writes — are carried into
//! the rebased tail and pushed on the next pump: re-entry is lossless (see
//! the relay module docs for why this is safe under seq-based frontiers).

use crate::checkpoint::{resume_anchored, AnchorError, Checkpoint, CheckpointError};
use crate::fold::{fold_onto, state_hash, FoldedRecord, SyncState};
use crate::journal::OplogJournal;
use crate::lease::{Intent, IntentStatus, LeaseCoordinator, LeaseError};
use crate::oplog::{verify_log, ChainError, DeviceLog, Hlc, OpRecord, Scope, Surface, WallClock};
use crate::relay::{checkpoint_frontier, frontier_of, Frontier, Relay, RelayError};
use serde_json::Value;
use std::fmt;
use std::path::{Path, PathBuf};

/// A sync-session failure.
#[derive(Debug)]
pub enum SessionError {
    Io(std::io::Error),
    Chain(ChainError),
    Anchor(AnchorError),
    Relay(RelayError),
    Checkpoint(CheckpointError),
    /// The journal carries a truncation marker naming a checkpoint that is
    /// not present in the session's checkpoint directory — resume is
    /// impossible without it (fetch it from the relay and retry).
    MissingCheckpoint { checkpoint_hash: String },
    /// The relay served a "latest" checkpoint that does not cover the
    /// session's current base — rebasing onto it would silently lose state.
    CheckpointRegression { held: String, offered: String },
    /// A lease-coordinator call failed (B5) — used by the best-effort local
    /// gate [`SyncSession::record_intent_if_current`].
    Lease(LeaseError),
}

impl fmt::Display for SessionError {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            SessionError::Io(e) => write!(f, "sync session io error: {e}"),
            SessionError::Chain(e) => write!(f, "sync session chain error: {e}"),
            SessionError::Anchor(e) => write!(f, "sync session anchor error: {e}"),
            SessionError::Relay(e) => write!(f, "sync session relay error: {e}"),
            SessionError::Checkpoint(e) => write!(f, "sync session checkpoint error: {e}"),
            SessionError::MissingCheckpoint { checkpoint_hash } => write!(
                f,
                "journal is truncated below checkpoint {checkpoint_hash}, which is not in the \
                 session checkpoint directory — fetch it (relay checkpoint_get) and retry"
            ),
            SessionError::CheckpointRegression { held, offered } => write!(
                f,
                "relay's latest checkpoint {offered} does not cover the session's base {held} — \
                 refusing to rebase onto it (state would be lost)"
            ),
            SessionError::Lease(e) => write!(f, "sync session lease error: {e}"),
        }
    }
}

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

impl From<std::io::Error> for SessionError {
    fn from(e: std::io::Error) -> Self {
        SessionError::Io(e)
    }
}
impl From<ChainError> for SessionError {
    fn from(e: ChainError) -> Self {
        SessionError::Chain(e)
    }
}
impl From<AnchorError> for SessionError {
    fn from(e: AnchorError) -> Self {
        SessionError::Anchor(e)
    }
}
impl From<RelayError> for SessionError {
    fn from(e: RelayError) -> Self {
        SessionError::Relay(e)
    }
}
impl From<CheckpointError> for SessionError {
    fn from(e: CheckpointError) -> Self {
        SessionError::Checkpoint(e)
    }
}
impl From<LeaseError> for SessionError {
    fn from(e: LeaseError) -> Self {
        SessionError::Lease(e)
    }
}

/// What one [`SyncSession::pump`] round did.
#[derive(Debug, Clone, Default, PartialEq)]
pub struct PumpReport {
    /// Own ops newly admitted by the relay.
    pub pushed: usize,
    /// Own ops the relay already held.
    pub push_deduped: usize,
    /// Remote ops pulled, verified, journaled, and folded this round.
    pub folded: usize,
    /// The fold frontier acked (derived from journal-held ops), if any.
    pub acked: Option<Hlc>,
}

/// A device's live sync endpoint: its append chain, its durable journal,
/// and the pump. See the module docs for the contract sequencing.
pub struct SyncSession {
    device_id: String,
    device: DeviceLog,
    journal: OplogJournal,
    checkpoint_dir: PathBuf,
    wall: WallClock,
    /// Every op the journal holds (the tail, when `base` is set).
    ops: Vec<OpRecord>,
    /// The checkpoint the journal is anchored on, when truncated.
    base: Option<Checkpoint>,
    /// In-memory push cursor (own max seq pushed). Deliberately NOT
    /// persisted: losing it in a crash only causes a deduped re-push.
    pushed_through: Option<u64>,
}

impl fmt::Debug for SyncSession {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("SyncSession")
            .field("device_id", &self.device_id)
            .field("ops", &self.ops.len())
            .field("base", &self.base.as_ref().map(|c| &c.checkpoint_hash))
            .field("pushed_through", &self.pushed_through)
            .finish_non_exhaustive()
    }
}

impl SyncSession {
    /// Open a session over an existing (possibly empty, possibly truncated)
    /// journal. A truncated journal resumes **only** through its covering
    /// checkpoint (`resume_anchored` — B4 contract): the checkpoint file
    /// must be present in `checkpoint_dir` under its content address.
    pub fn open(
        device_id: impl Into<String>,
        journal_path: &Path,
        checkpoint_dir: &Path,
        wall: WallClock,
    ) -> Result<Self, SessionError> {
        let device_id = device_id.into();
        let (marker, ops) = OplogJournal::load_with_marker(journal_path)?;
        let journal = OplogJournal::open(journal_path)?;
        let (device, base) = match marker {
            Some(marker) => {
                let path = checkpoint_dir
                    .join(format!("{}.checkpoint.json", marker.checkpoint_hash));
                if !path.exists() {
                    return Err(SessionError::MissingCheckpoint {
                        checkpoint_hash: marker.checkpoint_hash,
                    });
                }
                let checkpoint = Checkpoint::load(&path)?;
                let device = resume_anchored(device_id.clone(), &checkpoint, &ops)?;
                (device, Some(checkpoint))
            }
            None => (DeviceLog::resume(device_id.clone(), &ops)?, None),
        };
        let mut session = Self {
            device_id,
            device,
            journal,
            checkpoint_dir: checkpoint_dir.to_path_buf(),
            wall,
            ops,
            base,
            pushed_through: None,
        };
        session.device.set_wall_clock(session.wall.clone());
        Ok(session)
    }

    /// Open + immediately [`SyncSession::rebase`] onto the relay's latest
    /// checkpoint — the cold-device / returning-straggler entry point
    /// (proposal §"Cold / new device bootstrap"). With no relay checkpoint
    /// (young account) this degrades to a plain open; the first `pump`
    /// replays from genesis.
    pub fn bootstrap(
        device_id: impl Into<String>,
        journal_path: &Path,
        checkpoint_dir: &Path,
        relay: &mut dyn Relay,
        wall: WallClock,
    ) -> Result<Self, SessionError> {
        let mut session = Self::open(device_id, journal_path, checkpoint_dir, wall)?;
        session.rebase(relay)?;
        Ok(session)
    }

    /// Record a local mutation: stamp (hybrid clock), **journal (flushed)**,
    /// then hold for push — the op is journal-durable before it can ever be
    /// transmitted (B1 MUST). If the journal write fails the device chain
    /// is rolled back (rebuilt from the durable ops), so the failed op can
    /// never leave a hole for the next append to chain onto.
    pub fn append(
        &mut self,
        scope: Scope,
        surface: Surface,
        payload: Value,
    ) -> Result<OpRecord, SessionError> {
        let op = self.device.append(scope, surface, payload);
        if let Err(e) = self.journal.append(&op) {
            self.rebuild_device()?;
            return Err(SessionError::Io(e));
        }
        self.ops.push(op.clone());
        Ok(op)
    }

    /// The fence-independent **committed-run idempotency oracle** (B5) over
    /// this session's folded state (checkpoint base + journal tail): "has
    /// `run_id` already committed for `agent_id`?" Keep-all and immune to epoch
    /// bumps AND compaction, so it is the CORRECT idempotency lookup — unlike
    /// [`crate::fold::SyncState::intent`], which is the fenced "who holds now"
    /// view and can read `None`/pending for a run that actually committed. A B6
    /// dispatch fence performs exactly this read before an external side effect.
    pub fn committed_run(&self, agent_id: &str, run_id: &str) -> Option<FoldedRecord> {
        self.state().committed_run(agent_id, run_id).cloned()
    }

    /// Record a leased execution [`Intent`] (B5) — journal-durable exactly
    /// like any [`SyncSession::append`], so it is durable before it can be
    /// transmitted. The **partitioned / ungated** path: a zombie that cannot
    /// reach the coordinator still records here, and the fold converges the
    /// ledger deterministically.
    ///
    /// **Terminal guard (C3):** a committed run is terminal. A `pending`/`failed`
    /// write for a run already committed (per the fence-independent
    /// [`SyncSession::committed_run`] oracle) is a **no-op** (`Ok(None)`), so a
    /// failed-over holder writing `pending` before checking cannot revert the
    /// committed ledger.
    ///
    /// This converges the **ledger**; it is NOT the exactly-once execution gate.
    /// The B6 dispatch fence — a linearizable "am I still epoch N?" plus this
    /// committed-oracle read **before** the external effect — is what makes
    /// execution single-shot.
    pub fn record_intent(
        &mut self,
        scope: Scope,
        intent: &Intent,
    ) -> Result<Option<OpRecord>, SessionError> {
        if intent.status != IntentStatus::Committed
            && self.committed_run(&intent.agent_id, &intent.run_id).is_some()
        {
            return Ok(None); // terminal: the run already committed — do not revert
        }
        Ok(Some(self.append(scope, Surface::Intent, intent.payload())?))
    }

    /// Record a leased [`Intent`] **only if the local epoch is still current**
    /// — the best-effort local gate: a linearizable coordinator read confirms
    /// this device still holds the lease at `intent.epoch` before the op is
    /// journaled. Returns `Ok(None)` when the coordinator says this device is
    /// no longer the holder at that epoch (skipping a write the fold would fence)
    /// or when the run is already committed (the [`SyncSession::record_intent`]
    /// terminal guard).
    ///
    /// This is a *liveness optimization*, NOT the safety gate: a partitioned
    /// zombie that cannot reach the coordinator falls back to
    /// [`SyncSession::record_intent`]. Exactly-once execution is the B6 dispatch
    /// fence (this check races a pause-after-check — the Kleppmann residual).
    pub fn record_intent_if_current(
        &mut self,
        scope: Scope,
        intent: &Intent,
        coordinator: &mut dyn LeaseCoordinator,
    ) -> Result<Option<OpRecord>, SessionError> {
        let current = coordinator.current(&intent.agent_id)?;
        let still_ours = current
            .as_ref()
            .is_some_and(|l| l.holder == self.device_id && l.epoch == intent.epoch);
        if !still_ours {
            return Ok(None); // not the current holder — don't even write it
        }
        self.record_intent(scope, intent) // terminal-guarded
    }

    /// Rebuild the append chain from what is actually durable — the
    /// journal-held ops (+ base anchor).
    fn rebuild_device(&mut self) -> Result<(), SessionError> {
        let mut device = match &self.base {
            Some(checkpoint) => resume_anchored(self.device_id.clone(), checkpoint, &self.ops)?,
            None => DeviceLog::resume(self.device_id.clone(), &self.ops)?,
        };
        device.set_wall_clock(self.wall.clone());
        self.device = device;
        Ok(())
    }

    /// The per-device seq cursor of everything this session holds
    /// (checkpoint coverage + journal tail).
    fn held_frontier(&self) -> Frontier {
        let mut frontier = self
            .base
            .as_ref()
            .map(checkpoint_frontier)
            .unwrap_or_default();
        for (device, seq) in frontier_of(&self.ops) {
            let entry = frontier.entry(device).or_insert(seq);
            if seq > *entry {
                *entry = seq;
            }
        }
        frontier
    }

    /// The fold frontier this session may ack: the max HLC across
    /// journal-held ops and the base checkpoint's coverage — derived from
    /// durable state ONLY, which is what makes ack-before-fold impossible.
    fn ack_frontier(&self) -> Option<Hlc> {
        let from_ops = self.ops.iter().map(|op| &op.hlc).max();
        let from_base = self
            .base
            .as_ref()
            .and_then(|c| c.frontier.values().map(|e| &e.hlc).max());
        [from_ops, from_base].into_iter().flatten().max().cloned()
    }

    /// One reconciliation round: push journal-durable own ops → pull →
    /// verify → journal the folds → ack. See the module docs for the
    /// contract sequencing; retry-safe at every crash point.
    ///
    /// Returns [`RelayError::FrontierTruncated`] (wrapped) when the relay
    /// has GC'd past this session's frontier — call
    /// [`SyncSession::rebase`] and pump again.
    pub fn pump(&mut self, relay: &mut dyn Relay) -> Result<PumpReport, SessionError> {
        let mut report = PumpReport::default();

        // 1. Push own journal-durable ops the relay may not have.
        let mut own: Vec<OpRecord> = self
            .ops
            .iter()
            .filter(|op| {
                op.device_id == self.device_id
                    && self.pushed_through.is_none_or(|through| op.seq > through)
            })
            .cloned()
            .collect();
        own.sort_by_key(|op| op.seq);
        if !own.is_empty() {
            // Journal-durable BEFORE transmit (B1 MUST): the per-op appends
            // only reached the OS page cache (flush, not fsync). One batch
            // barrier here makes the whole own-op tail stable before any of
            // it leaves the device — a power loss in the writeback window
            // otherwise loses a transmitted op and re-mints its seq (a
            // permanent Fork). fsync once, not per append.
            self.journal.sync()?;
            let outcome = relay.push(&self.device_id, &own)?;
            report.pushed = outcome.accepted;
            report.push_deduped = outcome.deduped;
            self.pushed_through = own.last().map(|op| op.seq);
        }

        // 2. Pull everything above what we hold.
        let pulled = relay.pull(&self.device_id, &self.held_frontier())?;

        // 3. Filter already-held (idempotent re-pull) and VERIFY BEFORE FOLD.
        let held: std::collections::BTreeSet<&str> =
            self.ops.iter().map(|op| op.op_id.as_str()).collect();
        let new_ops: Vec<OpRecord> = pulled
            .ops
            .into_iter()
            .filter(|op| !held.contains(op.op_id.as_str()))
            .collect();
        if !new_ops.is_empty() {
            let mut candidate = self.ops.clone();
            candidate.extend(new_ops.iter().cloned());
            match &self.base {
                Some(checkpoint) => crate::checkpoint::verify_anchored(checkpoint, &candidate)?,
                None => verify_log(&candidate)?,
            }

            // 4. Journal the folds — advancing self.ops IN LOCKSTEP with the
            // journal. Recording each op into self.ops the instant its
            // append succeeds is load-bearing for retry-safety: a mid-loop
            // append failure then leaves self.ops == the journal, so the
            // retry pump filters the already-journaled ops out of the
            // re-pull instead of appending them a second time (a duplicate
            // journal line = DuplicateSeq on the next open = a bricked
            // device). `candidate` is discarded on failure.
            drop(candidate);
            for op in &new_ops {
                self.journal.append(op)?;
                self.ops.push(op.clone());
                self.device.observe(&op.hlc);
                report.folded += 1;
            }
            // Durable fold BEFORE ack (B4 MUST: ack asserts durably-folded
            // state). One batch fsync over the folds just journaled.
            self.journal.sync()?;
        }

        // 5. Ack — derived from journal-held state only, after it is
        // durable. (B4 MUST: ack asserts durably-folded state.)
        if let Some(frontier) = self.ack_frontier() {
            relay.ack(&self.device_id, frontier.clone())?;
            report.acked = Some(frontier);
        }
        Ok(report)
    }

    /// Re-anchor this session on the relay's latest checkpoint — the cold
    /// bootstrap / post-eviction re-entry move. Returns `true` when a
    /// rebase happened.
    ///
    /// Sequencing (each step durable before the next depends on it):
    /// `checkpoint_get` → `pull(since = checkpoint frontier)` → merge in
    /// every locally-held op the checkpoint does NOT cover (a returning
    /// straggler's unpushed writes survive) → `verify_anchored` → save the
    /// checkpoint into the session checkpoint dir → `truncate_to` (journal
    /// rewritten as the anchored tail, truncation marker stamped) →
    /// `resume_anchored`. The push cursor resets so the next `pump`
    /// re-offers every own op in the tail (relay dedups the already-pushed).
    pub fn rebase(&mut self, relay: &mut dyn Relay) -> Result<bool, SessionError> {
        let Some(checkpoint) = relay.checkpoint_get()? else {
            return Ok(false); // young account: genesis replay via pump
        };
        if let Some(base) = &self.base {
            if base.checkpoint_hash == checkpoint.checkpoint_hash {
                return Ok(false); // already anchored here
            }
            if !crate::relay::frontier_dominates(&checkpoint, base) {
                return Err(SessionError::CheckpointRegression {
                    held: base.checkpoint_hash.clone(),
                    offered: checkpoint.checkpoint_hash.clone(),
                });
            }
        }

        let pulled = relay.pull(&self.device_id, &checkpoint_frontier(&checkpoint))?;

        // Keep every held op the checkpoint does not cover — own unpushed
        // writes AND foreign tails we already folded — deduped against the
        // pull by op_id.
        let mut tail: Vec<OpRecord> = pulled.ops;
        let mut seen: std::collections::BTreeSet<String> =
            tail.iter().map(|op| op.op_id.clone()).collect();
        for op in &self.ops {
            let covered = checkpoint
                .frontier
                .get(&op.device_id)
                .is_some_and(|entry| op.seq <= entry.seq);
            if !covered && seen.insert(op.op_id.clone()) {
                tail.push(op.clone());
            }
        }
        tail.sort_by(|a, b| (&a.hlc, &a.op_id).cmp(&(&b.hlc, &b.op_id)));

        // Verify BEFORE any durable rewrite; then checkpoint durable FIRST,
        // then the journal truncation that names it (B4 crash ordering).
        crate::checkpoint::verify_anchored(&checkpoint, &tail)?;
        checkpoint.save(&self.checkpoint_dir)?;
        self.journal.truncate_to(&tail, &checkpoint.checkpoint_hash)?;

        let mut device = resume_anchored(self.device_id.clone(), &checkpoint, &tail)?;
        device.set_wall_clock(self.wall.clone());
        self.device = device;
        self.ops = tail;
        self.base = Some(checkpoint);
        self.pushed_through = None;
        Ok(true)
    }

    /// Compute a checkpoint at the relay's stable frontier from this
    /// session's held ops and upload it — the device-computed snapshot the
    /// proposal requires under E2E ("the relay holds ciphertext and cannot
    /// fold"). Call **after** a `pump` (so held == relay-known and own ops
    /// are pushed). Returns the uploaded checkpoint, or `None` when there
    /// is no stable frontier, nothing below it, or this session is itself
    /// anchored on a checkpoint (recompaction over a base is the same
    /// later slice B4 deferred).
    pub fn publish_checkpoint(
        &mut self,
        relay: &mut dyn Relay,
    ) -> Result<Option<Checkpoint>, SessionError> {
        if self.base.is_some() {
            return Ok(None);
        }
        let Some(frontier) = relay.stable_frontier()? else {
            return Ok(None);
        };
        let below: Vec<OpRecord> = self
            .ops
            .iter()
            .filter(|op| op.hlc <= frontier)
            .cloned()
            .collect();
        if below.is_empty() {
            return Ok(None);
        }
        let checkpoint = Checkpoint::from_ops(&below)?;
        relay.checkpoint_put(&self.device_id, &checkpoint)?;
        Ok(Some(checkpoint))
    }

    /// The materialized state: `fold_onto(base checkpoint, journal tail)`.
    pub fn state(&self) -> SyncState {
        let base = self
            .base
            .as_ref()
            .map(|c| c.state.clone())
            .unwrap_or_default();
        fold_onto(&base, &self.ops)
    }

    /// [`state_hash`] of [`SyncSession::state`] — the divergence invariant
    /// two synced devices must agree on.
    pub fn state_hash(&self) -> String {
        state_hash(&self.state())
    }

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

    /// The journal-held ops (the anchored tail, when a base is set).
    pub fn ops(&self) -> &[OpRecord] {
        &self.ops
    }

    /// The checkpoint this session's journal is anchored on, if truncated.
    pub fn base(&self) -> Option<&Checkpoint> {
        self.base.as_ref()
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::relay::{AckOutcome, GcReport, InMemoryRelay, PullResult, PushOutcome, RelayConfig, RosterEntry};
    use serde_json::json;
    use std::sync::atomic::{AtomicBool, Ordering};
    use std::sync::Arc;

    fn zero_wall() -> WallClock {
        Arc::new(|| 0)
    }

    fn mem_relay() -> InMemoryRelay {
        InMemoryRelay::new(RelayConfig::default(), zero_wall())
    }

    struct Dirs {
        _tmp: tempfile::TempDir,
        journal: std::path::PathBuf,
        ckpts: std::path::PathBuf,
    }

    fn dirs() -> Dirs {
        let tmp = tempfile::tempdir().unwrap();
        let journal = tmp.path().join("oplog.jsonl");
        let ckpts = tmp.path().join("checkpoints");
        Dirs { _tmp: tmp, journal, ckpts }
    }

    fn open(device: &str, d: &Dirs) -> SyncSession {
        SyncSession::open(device, &d.journal, &d.ckpts, zero_wall()).unwrap()
    }

    #[test]
    fn two_devices_converge_through_the_relay_across_all_tiers() {
        let mut relay = mem_relay();
        let (da, db) = (dirs(), dirs());
        let mut a = open("mac-a", &da);
        let mut b = open("mac-b", &db);

        // Concurrent writes on every fold tier, including an LWW conflict
        // and a routing observation multiset.
        a.append(Scope::Personal, Surface::Knowledge, json!({"id": "f1", "v": 1})).unwrap();
        a.append(Scope::Personal, Surface::Declagent, json!({"id": "milo", "owner": "a"})).unwrap();
        a.append(Scope::Personal, Surface::Routing, json!({"sample": 1.0})).unwrap();
        b.append(Scope::Personal, Surface::Knowledge, json!({"id": "f2", "v": 2})).unwrap();
        b.append(Scope::Personal, Surface::Routing, json!({"sample": 1.0})).unwrap();

        a.pump(&mut relay).unwrap();
        let rb = b.pump(&mut relay).unwrap();
        assert_eq!(rb.folded, 3, "b folded a's three ops");
        // b now writes causally AFTER folding a's registry record.
        b.append(Scope::Personal, Surface::Declagent, json!({"id": "milo", "owner": "b"})).unwrap();
        b.pump(&mut relay).unwrap();
        let ra = a.pump(&mut relay).unwrap();
        assert_eq!(ra.folded, 3);

        assert_eq!(a.state_hash(), b.state_hash(), "divergence invariant: same hash");
        let state = a.state();
        assert_eq!(
            state.registries[&Surface::Declagent.tag()]["id:milo"].payload["owner"],
            json!("b"),
            "LWW resolved by the hybrid clock's causal order"
        );
        assert_eq!(
            state.log_entries(&Surface::Routing.tag()).len(),
            2,
            "the observation multiset survived transport"
        );

        // Idempotence: an extra pump on both sides is a complete no-op.
        let ra = a.pump(&mut relay).unwrap();
        let rb = b.pump(&mut relay).unwrap();
        assert_eq!((ra.pushed, ra.folded), (0, 0));
        assert_eq!((rb.pushed, rb.folded), (0, 0));
        assert_eq!(a.state_hash(), b.state_hash());
    }

    #[test]
    fn op_is_journal_durable_before_it_is_ever_transmitted() {
        // Contract 1 (B1 MUST): append journals+flushes; the crash window
        // between append and pump loses NOTHING and re-mints NO seq.
        let mut relay = mem_relay();
        let d = dirs();
        {
            let mut a = open("mac-a", &d);
            a.append(Scope::Personal, Surface::Knowledge, json!({"id": "f1"})).unwrap();
            // "crash" before any pump: session dropped, nothing transmitted.
        }
        let mut a = open("mac-a", &d);
        // The op survived in the journal; resume did not re-mint its seq.
        let next = a.append(Scope::Personal, Surface::Knowledge, json!({"id": "f2"})).unwrap();
        assert_eq!(next.seq, 1);
        a.pump(&mut relay).unwrap();
        assert_eq!(relay.pull("x", &Frontier::new()).unwrap().ops.len(), 2);
        verify_log(a.ops()).unwrap();
    }

    #[test]
    fn ack_is_derived_from_journal_held_state_only() {
        // Contract 2 (B4 MUST), shown by construction: the acked frontier
        // equals the max HLC of what is ON DISK in the journal — never of
        // anything merely received.
        let mut relay = mem_relay();
        let (da, db) = (dirs(), dirs());
        let mut a = open("mac-a", &da);
        a.append(Scope::Personal, Surface::Knowledge, json!({"id": "f1"})).unwrap();
        a.pump(&mut relay).unwrap();

        let mut b = open("mac-b", &db);
        let report = b.pump(&mut relay).unwrap();
        let acked = report.acked.unwrap();

        // Reload b's journal from disk: the ack is exactly its max stamp.
        drop(b);
        let on_disk = OplogJournal::load(&db.journal).unwrap();
        assert_eq!(acked, on_disk.iter().map(|op| op.hlc.clone()).max().unwrap());
        let roster: std::collections::BTreeMap<String, RosterEntry> = relay
            .roster()
            .unwrap()
            .into_iter()
            .map(|e| (e.device_id.clone(), e))
            .collect();
        assert_eq!(roster["mac-b"].acked.as_ref(), Some(&acked));
    }

    /// A relay wrapper that fails `ack` while the flag is set — the
    /// crash/partition at the worst point of the pump (after the durable
    /// fold, before the ack).
    struct FlakyAckRelay<'a> {
        inner: &'a mut dyn Relay,
        fail_ack: Arc<AtomicBool>,
    }

    impl Relay for FlakyAckRelay<'_> {
        fn register(&mut self, d: &str) -> Result<RosterEntry, RelayError> {
            self.inner.register(d)
        }
        fn push(&mut self, d: &str, ops: &[OpRecord]) -> Result<PushOutcome, RelayError> {
            self.inner.push(d, ops)
        }
        fn pull(&mut self, d: &str, since: &Frontier) -> Result<PullResult, RelayError> {
            self.inner.pull(d, since)
        }
        fn ack(&mut self, d: &str, frontier: Hlc) -> Result<AckOutcome, RelayError> {
            if self.fail_ack.load(Ordering::SeqCst) {
                return Err(RelayError::Io(std::io::Error::other("network down")));
            }
            self.inner.ack(d, frontier)
        }
        fn checkpoint_put(&mut self, d: &str, c: &Checkpoint) -> Result<bool, RelayError> {
            self.inner.checkpoint_put(d, c)
        }
        fn checkpoint_get(&mut self) -> Result<Option<Checkpoint>, RelayError> {
            self.inner.checkpoint_get()
        }
        fn roster(&mut self) -> Result<Vec<RosterEntry>, RelayError> {
            self.inner.roster()
        }
        fn stable_frontier(&mut self) -> Result<Option<Hlc>, RelayError> {
            self.inner.stable_frontier()
        }
        fn gc(&mut self) -> Result<GcReport, RelayError> {
            self.inner.gc()
        }
    }

    #[test]
    fn crash_mid_pump_is_idempotent_at_every_step() {
        let mut relay = mem_relay();
        let (da, db) = (dirs(), dirs());
        let mut a = open("mac-a", &da);
        a.append(Scope::Personal, Surface::Knowledge, json!({"id": "f1"})).unwrap();
        a.pump(&mut relay).unwrap();

        // --- Crash point 1: pulled but nothing device-side happened yet
        // (transport delivered bytes; the process died before verify/fold).
        // Nothing durable changed, the relay ack table is untouched → a
        // fresh pump redoes everything.
        let _ = relay.pull("mac-b", &Frontier::new()).unwrap();
        let roster: std::collections::BTreeMap<String, RosterEntry> = relay
            .roster()
            .unwrap()
            .into_iter()
            .map(|e| (e.device_id.clone(), e))
            .collect();
        assert_eq!(roster["mac-b"].acked, None, "merely-received is never acked");

        // --- Crash point 2: fold journaled durably, ack lost (network died
        // between the fsync and the ack).
        let fail = Arc::new(AtomicBool::new(true));
        let mut b = open("mac-b", &db);
        b.append(Scope::Personal, Surface::Knowledge, json!({"id": "f2"})).unwrap();
        {
            let mut flaky = FlakyAckRelay { inner: &mut relay, fail_ack: fail.clone() };
            let err = b.pump(&mut flaky).unwrap_err();
            assert!(matches!(err, SessionError::Relay(RelayError::Io(_))));
        }
        // The fold IS durable (journal has a's op)…
        let on_disk = OplogJournal::load(&db.journal).unwrap();
        assert_eq!(on_disk.len(), 2);
        // …but the ack never landed — the SAFE direction: GC can't drop
        // what b hasn't acked.
        let roster: std::collections::BTreeMap<String, RosterEntry> = relay
            .roster()
            .unwrap()
            .into_iter()
            .map(|e| (e.device_id.clone(), e))
            .collect();
        assert_eq!(roster["mac-b"].acked, None);

        // --- Crash point 3: session lost entirely (push cursor gone).
        // Reopen from the journal and re-pump: re-push dedups by op_id,
        // re-pull folds nothing new, the ack finally lands.
        drop(b);
        fail.store(false, Ordering::SeqCst);
        let mut b = open("mac-b", &db);
        let report = b.pump(&mut relay).unwrap();
        assert_eq!(report.folded, 0, "re-pull re-fold is a no-op by op_id dedup");
        assert_eq!(report.pushed, 0, "re-push deduped relay-side");
        assert_eq!(report.push_deduped, 1);
        assert!(report.acked.is_some());

        a.pump(&mut relay).unwrap();
        assert_eq!(a.state_hash(), b.state_hash(), "convergence after every crash point");
    }

    #[test]
    fn cold_bootstrap_goes_through_resume_anchored_and_stays_fenced() {
        // Contract 5: a cold device bootstraps checkpoint-first; its
        // journal carries the truncation marker, so the naive
        // load()/DeviceLog::resume path stays a runtime error.
        let mut relay = mem_relay();
        let da = dirs();
        let mut a = open("mac-a", &da);
        for i in 0..4 {
            a.append(Scope::Personal, Surface::Knowledge, json!({"id": format!("f{i}"), "timestamp": i})).unwrap();
        }
        a.pump(&mut relay).unwrap();
        let ckpt = a.publish_checkpoint(&mut relay).unwrap().unwrap();
        relay.gc().unwrap();

        // Fresh device: bootstrap = checkpoint_get + pull(since ckpt
        // frontier) + resume_anchored.
        let db = dirs();
        let mut b = SyncSession::bootstrap("mac-b", &db.journal, &db.ckpts, &mut relay, zero_wall())
            .unwrap();
        assert_eq!(b.base().unwrap().checkpoint_hash, ckpt.checkpoint_hash);
        b.pump(&mut relay).unwrap();
        assert_eq!(b.state_hash(), a.state_hash());

        // The fences hold on the bootstrapped journal.
        let err = OplogJournal::load(&db.journal).unwrap_err();
        assert_eq!(err.kind(), std::io::ErrorKind::InvalidData);
        let (marker, tail) = OplogJournal::load_with_marker(&db.journal).unwrap();
        assert_eq!(marker.unwrap().checkpoint_hash, ckpt.checkpoint_hash);
        // b keeps working across a restart (open() resumes anchored)…
        drop(b);
        let mut b = open("mac-b", &db);
        let op = b.append(Scope::Personal, Surface::Knowledge, json!({"id": "from-b"})).unwrap();
        assert!(op.hlc > tail.iter().map(|o| o.hlc.clone()).max().unwrap_or(Hlc {
            wall_ms: 0,
            counter: 0,
            device_id: String::new()
        }));
        b.pump(&mut relay).unwrap();
        a.pump(&mut relay).unwrap();
        assert_eq!(a.state_hash(), b.state_hash());
    }

    #[test]
    fn partial_fold_failure_is_retry_safe_and_never_bricks_the_journal() {
        // Kernel-review BRICKED-DEVICE repro (mechanism, not just the
        // duplicated-line consequence): a mid-fold-loop journal append
        // failure must leave self.ops == the journal, so the retry filters
        // the already-journaled ops out of the re-pull instead of writing
        // them a SECOND time (a duplicate line = DuplicateSeq on the next
        // open = a permanently bricked device with no recovery API).
        let mut relay = mem_relay();
        let (da, db) = (dirs(), dirs());
        let mut a = open("mac-a", &da);
        a.append(Scope::Personal, Surface::Knowledge, json!({"id": "f1"})).unwrap();
        a.append(Scope::Personal, Surface::Knowledge, json!({"id": "f2"})).unwrap();
        a.pump(&mut relay).unwrap();

        let mut b = open("mac-b", &db);
        // Force the SECOND fold append to fail (ENOSPC-class).
        b.journal.fail_append_after = Some(1);
        let err = b.pump(&mut relay).unwrap_err();
        assert!(matches!(err, SessionError::Io(_)), "got {err:?}");

        // Lockstep invariant: self.ops holds exactly what the journal holds
        // (the one op that appended before the failure), never more.
        assert_eq!(b.ops().len(), 1, "only the successfully-journaled fold is in self.ops");
        let on_disk = OplogJournal::load(&db.journal).unwrap();
        assert_eq!(on_disk.len(), 1);
        assert_eq!(on_disk[0].op_id, b.ops()[0].op_id);

        // Retry: the seam auto-cleared, so this pump folds the remaining op.
        // Crucially it does NOT re-journal the first fold (filtered by op_id
        // from self.ops) — no duplicate line.
        let report = b.pump(&mut relay).unwrap();
        assert_eq!(report.folded, 1, "only the un-journaled op is folded on retry");
        let on_disk = OplogJournal::load(&db.journal).unwrap();
        assert_eq!(on_disk.len(), 2, "no duplicate line — the journal is not bricked");
        verify_log(&on_disk).expect("no DuplicateSeq: the journal opens cleanly");

        // The device is not bricked: it reopens and converges.
        drop(b);
        let b = open("mac-b", &db);
        assert_eq!(a.state_hash(), b.state_hash());
    }

    #[test]
    fn append_failure_rolls_the_chain_back() {
        // Force a journal append failure by dropping the journal file's
        // directory out from under it is not portable; instead exercise the
        // rebuild path directly: rebuild_device must reproduce the exact
        // chain position after arbitrary appends.
        let d = dirs();
        let mut a = open("mac-a", &d);
        a.append(Scope::Personal, Surface::Knowledge, json!({"id": "f1"})).unwrap();
        a.append(Scope::Personal, Surface::Knowledge, json!({"id": "f2"})).unwrap();
        let before = a.ops().to_vec();
        a.rebuild_device().unwrap();
        let next = a.append(Scope::Personal, Surface::Knowledge, json!({"id": "f3"})).unwrap();
        assert_eq!(next.seq, 2);
        assert_eq!(next.prev.as_deref(), Some(before[1].op_id.as_str()));
        let mut all = before;
        all.push(next);
        verify_log(&all).unwrap();
    }

    // ------------------------------------------------------------------
    // B5: execution lease + fencing, end-to-end through the relay.
    // ------------------------------------------------------------------

    #[test]
    fn partition_both_write_intents_converge_to_the_higher_epoch_winner() {
        // The full arc: mac-a holds the lease (epoch 1) and fires a scheduled
        // run; it pauses past its TTL; mac-b STEALS the lease (epoch 2) and
        // fires the SAME run (same B7 deterministic run_id); mac-a, partitioned
        // from the coordinator, wrongly still believes it holds epoch 1 and
        // ALSO commits the run (a real zombie). After convergence the fold
        // fences the zombie deterministically — one ledger record, the
        // higher-epoch (mac-b) commit wins, even though the zombie's op has a
        // LATER HLC. No double-execution; both devices agree.
        use crate::lease::{InMemoryLeaseCoordinator, IntentStatus};
        use std::sync::atomic::AtomicU64;

        let coord_t = Arc::new(AtomicU64::new(0));
        let reader = coord_t.clone();
        let coord_wall: WallClock = Arc::new(move || reader.load(Ordering::SeqCst));
        let mut coord = InMemoryLeaseCoordinator::new(coord_wall);

        let mut relay = mem_relay();
        let (da, db) = (dirs(), dirs());
        let mut a = open("mac-a", &da);
        let mut b = open("mac-b", &db);
        let run = car_proto::deterministic_run_id("milo", "3am digest", "occurrence-1");

        // 1. mac-a acquires (epoch 1) and records the run's pending intent
        //    through the GATED path (coordinator confirms it still holds).
        let lease_a = coord.acquire("milo", "mac-a", 100).unwrap();
        assert_eq!(lease_a.epoch, 1);
        let recorded = a
            .record_intent_if_current(
                Scope::Personal,
                &Intent::new("milo", &run, 1, IntentStatus::Pending),
                &mut coord,
            )
            .unwrap();
        assert!(recorded.is_some(), "mac-a holds the lease → intent recorded");
        a.pump(&mut relay).unwrap();

        // 2. mac-b folds a's state, then — a's lid closed past the TTL —
        //    STEALS the lease (epoch 2) and commits the same run.
        b.pump(&mut relay).unwrap();
        coord_t.store(200, Ordering::SeqCst); // past mac-a's 100ms TTL
        let lease_b = coord.acquire("milo", "mac-b", 100).unwrap();
        assert_eq!((lease_b.epoch, lease_b.holder.as_str()), (2, "mac-b"));
        b.record_intent_if_current(
            Scope::Personal,
            &Intent::new("milo", &run, 2, IntentStatus::Pending),
            &mut coord,
        )
        .unwrap()
        .expect("mac-b holds epoch 2");
        let b_commit = b
            .record_intent_if_current(
                Scope::Personal,
                &Intent::new("milo", &run, 2, IntentStatus::Committed),
                &mut coord,
            )
            .unwrap()
            .expect("mac-b holds epoch 2");
        b.pump(&mut relay).unwrap();

        // 3. mac-a is PARTITIONED from the coordinator (never learns it lost)
        //    but still syncs with the relay. It folds b's ops, then — a genuine
        //    double-execution artifact — also commits the run via the UNGATED
        //    path, stamping a LATER HLC than b's committed op. (A committed
        //    write is allowed even for an already-committed run — idempotent;
        //    the fold converges it. The terminal guard only no-ops a *pending*
        //    write for an already-committed run — see the C3 session test.)
        a.pump(&mut relay).unwrap();
        let a_zombie = a
            .record_intent(
                Scope::Personal,
                &Intent::new("milo", &run, 1, IntentStatus::Committed),
            )
            .unwrap()
            .expect("a committed write is recorded (not terminal-guarded)");
        assert!(a_zombie.hlc > b_commit.hlc, "the zombie's write is later in HLC");
        a.pump(&mut relay).unwrap();
        b.pump(&mut relay).unwrap();

        // 4. Converge: both agree; the epoch-2 (mac-b) commit is the ledger
        //    winner despite the zombie's later HLC (fencing beats the clock),
        //    and the fence-independent oracle agrees.
        assert_eq!(a.state_hash(), b.state_hash(), "divergence invariant holds");
        let state = a.state();
        assert_eq!(state.intents["milo"].runs.len(), 1, "one who-holds record");
        assert_eq!(state.fencing_epoch("milo"), Some(2));
        let winner = state.intent("milo", &run).expect("run present");
        assert_eq!(winner.op_id, b_commit.op_id, "higher-epoch commit wins despite lower HLC");
        let decoded = Intent::from_payload(&winner.payload).unwrap();
        assert_eq!((decoded.epoch, decoded.status), (2, IntentStatus::Committed));
        // The durable idempotency oracle also resolves to the epoch-2 commit,
        // on both devices — the read a B6 dispatch fence would perform.
        assert_eq!(a.committed_run("milo", &run).unwrap().op_id, b_commit.op_id);
        assert_eq!(b.committed_run("milo", &run).unwrap().op_id, b_commit.op_id);
    }

    #[test]
    fn record_intent_if_current_gates_out_a_lost_holder() {
        // The best-effort local gate when the coordinator IS reachable: once
        // mac-a no longer holds the lease, the gated write records nothing
        // (saving an op the fold would only fence). This is a liveness
        // optimization — the sound gate remains the fold's epoch fence.
        use crate::lease::{InMemoryLeaseCoordinator, IntentStatus};
        let mut coord = InMemoryLeaseCoordinator::new(zero_wall());
        let d = dirs();
        let mut a = open("mac-a", &d);
        let run = "run-x";

        coord.acquire("milo", "mac-a", 100).unwrap();
        let recorded = a
            .record_intent_if_current(
                Scope::Personal,
                &Intent::new("milo", run, 1, IntentStatus::Pending),
                &mut coord,
            )
            .unwrap();
        assert!(recorded.is_some());
        assert_eq!(a.ops().len(), 1);

        // mac-a releases; mac-b acquires epoch 2. The coordinator now reports
        // mac-b as the holder → the gate refuses mac-a's write.
        coord.release("milo", "mac-a", 1).unwrap();
        coord.acquire("milo", "mac-b", 100).unwrap();
        let gated = a
            .record_intent_if_current(
                Scope::Personal,
                &Intent::new("milo", run, 1, IntentStatus::Committed),
                &mut coord,
            )
            .unwrap();
        assert!(gated.is_none(), "coordinator says mac-a lost → not recorded");
        assert_eq!(a.ops().len(), 1, "nothing new journaled");
    }

    #[test]
    fn c3_record_intent_no_ops_a_pending_for_an_already_committed_run() {
        // C3 write-side guard: once a run has committed (per the
        // fence-independent oracle), record_intent NO-OPs a later
        // pending/failed write for it — a failed-over holder that writes
        // `pending` before checking cannot revert the committed ledger. This is
        // the ungated path (no coordinator); the guard is a local oracle read.
        let mut relay = mem_relay();
        let (da, db) = (dirs(), dirs());
        let mut a = open("mac-a", &da);
        let mut b = open("mac-b", &db);
        let run = "run-nightly";

        // mac-a commits the run at epoch 1 and syncs it to mac-b.
        a.record_intent(Scope::Personal, &Intent::new("milo", run, 1, IntentStatus::Committed))
            .unwrap()
            .expect("first commit recorded");
        a.pump(&mut relay).unwrap();
        b.pump(&mut relay).unwrap();
        assert!(b.committed_run("milo", run).is_some(), "mac-b folded the commit");

        // mac-b fails over to epoch 2 and — before checking — tries to write the
        // run PENDING. The terminal guard no-ops it (the run already committed).
        let before = b.ops().len();
        let attempt = b
            .record_intent(Scope::Personal, &Intent::new("milo", run, 2, IntentStatus::Pending))
            .unwrap();
        assert!(attempt.is_none(), "pending for an already-committed run is a no-op");
        assert_eq!(b.ops().len(), before, "nothing journaled");

        // The ledger stays committed on both sides after further sync.
        b.pump(&mut relay).unwrap();
        a.pump(&mut relay).unwrap();
        assert!(a.committed_run("milo", run).is_some());
        let decoded = Intent::from_payload(&b.state().intent("milo", run).unwrap().payload).unwrap();
        assert_eq!(decoded.status, IntentStatus::Committed, "not reverted to pending");
    }
}