car_sync/

relay.rs

//! Relay transport — the account-scoped op-stream devices push/pull against
//! (slice B3 of `docs/proposals/multi-device-sync.md`, §"Transport" +
//! §"Stragglers and the single coherence knob" + §"Sync protocol surface").
//!
//! The [`Relay`] trait is the proposal's surface — `push(ops)`,
//! `pull(since_frontier) → {ops, latest_checkpoint_ptr}`, `ack(frontier)`,
//! `checkpoint_put/get`, `roster()` — as a **library trait** with two
//! reference implementations: [`InMemoryRelay`] (tests, in-process
//! coordination) and [`FsRelay`] (a shared directory — the realistic
//! single-user two-Mac loopback). The network daemon surface (`sync.*`
//! JSON-RPC + the Parslee-hosted backend) is B6; anything that speaks this
//! trait is that surface's contract.
//!
//! ## What the relay validates (and what it can't)
//!
//! The relay holds one **chain per device** and admits a pushed op only if
//! it *continues* that chain exactly: contiguous `seq`, `prev` linking the
//! relay-held head, HLC advancing (a re-push of an already-held op dedups on
//! `op_id`; a *different* op claiming a held `seq` is a
//! [`RelayError::Fork`] — the B1 permanent-fork hazard surfaced as a
//! runtime error at the transport). A device pushes only its **own** chain
//! ([`RelayError::ForeignOps`]). In B3 payloads are cleartext so the relay
//! can run this full verification; under B6's E2E encryption the same
//! checks still work — `op_id`/`hlc`/`seq`/`prev` stay cleartext metadata
//! by design ("the relay can route and dedup on `op_id` and `hlc`").
//! Device *identity* remains asserted, not authenticated, until B6 signing.
//!
//! ## Stable frontier, eviction horizon `H`, and GC
//!
//! Straight from the proposal's stragglers section:
//!
//! - **Stable frontier** = `min(acked)` over **active (non-evicted) roster
//!   devices** — `None` (nothing droppable) while any active device has
//!   never acked, mirroring `compact::AckTable`'s refusal semantics.
//! - **Eviction**: a device silent (no push/pull/ack) longer than
//!   [`RelayConfig::eviction_horizon_ms`] is marked
//!   [`DeviceStatus::Evicted`] on the roster; its ack no longer holds the
//!   frontier. An evicted device may still push/pull/ack (its re-entry path
//!   is cold bootstrap — `checkpoint_get` + `pull(since checkpoint
//!   frontier)` + `resume_anchored`, see `crate::session`), and it is
//!   **reinstated** when it acks at/above the current stable frontier
//!   (i.e. it has provably caught up) — or whenever nothing is
//!   GC-eligible anyway (`stable frontier == None`).
//! - **GC** ([`Relay::gc`]): an op is droppable only when **both** (a) its
//!   HLC is at/below the stable frontier AND (b) a stored checkpoint
//!   **covers** it (`checkpoint.frontier[device].seq >= op.seq`) — "ops
//!   below the stable frontier are GC-eligible relay-side only after a
//!   covering checkpoint exists". Both conditions are per-device chain
//!   *prefixes* (HLC-monotone chains; frontier seqs), so GC always drops a
//!   prefix and the retained chain stays gap-free; the relay keeps the last
//!   dropped op's `(seq, op_id, hlc)` as the chain anchor for continuity
//!   checks and remembers the drop floor so a `pull` whose `since` frontier
//!   reaches into truncated space fails loudly
//!   ([`RelayError::FrontierTruncated`]) instead of silently serving a
//!   gapped log — the signal that sends the puller to cold bootstrap.
//!
//! **Deliberate deviation from the proposal, binding on B6:** the proposal's
//! third `H` rule ("max op age the relay accepts = `H`; older ops are
//! rejected") exists to keep accepted ops out of *time-truncated* space
//! (`hlc.wall < now − H`). This relay's truncation is not time-based — it
//! is per-device-seq + covering-checkpoint, under which a returning
//! straggler's late ops are structurally *outside* truncated space (its own
//! chain frontier is behind them) and converge losslessly once pushed,
//! which is strictly stronger than the proposal's "loses only writes it
//! made while >H-offline" degenerate case. So B3 accepts old ops rather
//! than rejecting them. If B6's hosted backend adopts wall-clock-bounded
//! storage, it must reintroduce the age bound *and* the proposal's
//! surface-it-to-the-user story together.
//!
//! ## Checkpoints
//!
//! `checkpoint_put` runs TWO checks, not one. `Checkpoint::verify` proves
//! internal self-consistency (both hashes recompute) — but that alone lets
//! a checkpoint built from a *different* chain merely CLAIMING a device's
//! name walk in and, via the coverage claim, make GC drop that device's
//! real ops (the kernel-review data-loss defect). So `checkpoint_put` also
//! **cross-checks the frontier against the relay-held chains**
//! ([`RelayState::validate_frontier`]): every frontier entry must name an op
//! the relay actually holds/held for that device at that seq (or, for a
//! seq already GC'd, be consistent with the remembered dropped head). A
//! device the relay has never seen a chain for cannot be validated. Only
//! validated checkpoints are stored, so `gc` — which counts stored
//! checkpoints as coverage — counts only validated ones, and the
//! bare-seq coverage test is then sound (a validated frontier head at seq S
//! means the contiguous hash-linked prefix 0..S is the device's real
//! chain). `op_id`/`seq`/`prev` stay cleartext metadata under B6 E2E, so
//! this cross-check survives encryption. Storage **dedups on
//! `checkpoint_hash` — the whole-record content address, never
//! `state_hash`** (binding contract from B4: two frontiers can fold to one
//! state; keying on the state would keep the wrong checkpoint and fork
//! chains on resume). The "latest" pointer served by `pull`/`checkpoint_get`
//! advances only to a checkpoint whose frontier **dominates** the current
//! latest (covers at least every device/seq it covers), so a stale or
//! concurrent upload can never regress the bootstrap pointer.

use crate::checkpoint::{Checkpoint, CheckpointError, FrontierEntry};
use crate::oplog::{Hlc, OpRecord, WallClock};
use serde::{Deserialize, Serialize};
use std::collections::BTreeMap;
use std::fmt;
use std::fs::{self, File, OpenOptions};
use std::io::Write;
use std::path::{Path, PathBuf};

/// A pull cursor: device_id → highest `seq` the puller already holds for
/// that device. Seq-based (not HLC-based) on purpose: a straggler's late
/// ops carry *old* HLCs but *new* seqs, so a seq frontier still delivers
/// them to every peer — an HLC cursor would silently skip them.
pub type Frontier = BTreeMap<String, u64>;

/// The frontier of an op-set: max seq per device.
pub fn frontier_of(ops: &[OpRecord]) -> Frontier {
    let mut frontier = Frontier::new();
    for op in ops {
        let entry = frontier.entry(op.device_id.clone()).or_insert(op.seq);
        if op.seq > *entry {
            *entry = op.seq;
        }
    }
    frontier
}

/// The pull cursor a checkpoint's coverage corresponds to — the cold
/// bootstrap's `pull(since = F)`.
pub fn checkpoint_frontier(checkpoint: &Checkpoint) -> Frontier {
    checkpoint
        .frontier
        .iter()
        .map(|(device, entry)| (device.clone(), entry.seq))
        .collect()
}

/// Roster liveness state — the proposal's "devices+last_seen" with the
/// eviction verdict made explicit.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum DeviceStatus {
    Active,
    /// Silent past the horizon `H`: its ack no longer holds the stable
    /// frontier; re-entry is cold bootstrap, reinstated on a caught-up ack.
    Evicted,
}

/// One device registry entry — `roster()`'s row.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct RosterEntry {
    pub device_id: String,
    /// The stamp the device joined at (relay wall reading, counter 0).
    pub added_at: Hlc,
    /// Relay wall-clock ms of the device's last push/pull/ack/register.
    pub last_seen_ms: u64,
    /// The device's acked fold frontier (monotone-only), `None` until it
    /// first acks — which holds the stable frontier at `None` (safe).
    pub acked: Option<Hlc>,
    pub status: DeviceStatus,
}

/// Relay policy knobs.
#[derive(Debug, Clone, Default)]
pub struct RelayConfig {
    /// The stragglers horizon `H`: a device silent longer than this is
    /// marked [`DeviceStatus::Evicted`]. `None` (the default) never evicts
    /// — the safe, GC-pinning posture a caller must opt out of.
    pub eviction_horizon_ms: Option<u64>,
}

/// What a push did. Not transactional: on an error mid-batch the already
/// accepted prefix stays (retry-safe — a re-push of it dedups).
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct PushOutcome {
    /// Ops newly admitted to the device's relay-held chain.
    pub accepted: usize,
    /// Ops the relay already held (retransmission — `op_id` identical) or
    /// that fall below its GC floor for the device (already covered by a
    /// checkpoint).
    pub deduped: usize,
}

/// What a pull returned — the proposal's `{ops, latest_checkpoint_ptr}`.
#[derive(Debug, Clone, PartialEq)]
pub struct PullResult {
    /// Every retained op above the `since` frontier, in canonical
    /// `(hlc, op_id)` order.
    pub ops: Vec<OpRecord>,
    /// `checkpoint_hash` of the current latest checkpoint, if any — the
    /// cold-bootstrap pointer.
    pub latest_checkpoint: Option<String>,
}

/// What an ack did.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub struct AckOutcome {
    /// The acked frontier advanced (monotone-only, like `AckTable::ack`).
    pub advanced: bool,
    /// An evicted device proved it caught up and is active again.
    pub reinstated: bool,
}

/// What GC dropped, per device.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct GcReport {
    pub dropped: BTreeMap<String, usize>,
}

impl GcReport {
    pub fn total(&self) -> usize {
        self.dropped.values().sum()
    }
}

/// A relay operation failure.
#[derive(Debug)]
pub enum RelayError {
    /// A pushed op fails self-verification (id/device mismatch) or does not
    /// link the relay-held chain head.
    Chain { device_id: String, detail: String },
    /// A different op claims a `seq` the relay already holds — the
    /// permanent chain fork, refused at the transport.
    Fork { device_id: String, seq: u64 },
    /// A pushed op skips ahead of the relay-held chain (ops must arrive
    /// contiguously).
    Gap { device_id: String, expected: u64, found: u64 },
    /// A device may only push its own chain.
    ForeignOps { device_id: String, op_device: String },
    /// The pull's `since` frontier reaches into GC'd space — the puller
    /// cannot be served a gapless log and must cold-bootstrap from the
    /// latest checkpoint (`checkpoint_get` + `pull(since = checkpoint
    /// frontier)` + `resume_anchored`).
    FrontierTruncated { device_id: String, dropped_below: u64 },
    /// The uploaded checkpoint does not verify.
    Checkpoint(CheckpointError),
    /// The uploaded checkpoint is internally self-consistent
    /// ([`Checkpoint::verify`] passed) but its frontier claims a chain
    /// position that does **not** match the relay-held chain for a device —
    /// a checkpoint built from a *different* chain merely claiming the
    /// device's name (a restored-from-backup re-mint, or two accounts
    /// colliding on a device id in a shared `FsRelay` folder). Rejected so
    /// it can never become GC coverage evidence and drop the device's real
    /// ops. (`op_id`/`seq`/`prev` stay cleartext metadata under B6 E2E, so
    /// this cross-check survives encryption.)
    CheckpointFrontierUnverified { device_id: String, seq: u64, detail: String },
    Io(std::io::Error),
}

impl fmt::Display for RelayError {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            RelayError::Chain { device_id, detail } => {
                write!(f, "relay push rejected for {device_id}: {detail}")
            }
            RelayError::Fork { device_id, seq } => write!(
                f,
                "relay push rejected: a different op already holds {device_id} seq {seq} — \
                 device chain fork"
            ),
            RelayError::Gap { device_id, expected, found } => write!(
                f,
                "relay push rejected for {device_id}: seq gap (relay expects {expected}, \
                 got {found}) — push contiguously"
            ),
            RelayError::ForeignOps { device_id, op_device } => write!(
                f,
                "relay push rejected: device {device_id} pushed an op emitted by {op_device} — \
                 a device pushes only its own chain"
            ),
            RelayError::FrontierTruncated { device_id, dropped_below } => write!(
                f,
                "pull frontier reaches into GC'd space (device {device_id}: ops below seq \
                 {dropped_below} were truncated) — cold-bootstrap from the latest checkpoint"
            ),
            RelayError::Checkpoint(e) => write!(f, "relay checkpoint rejected: {e}"),
            RelayError::CheckpointFrontierUnverified { device_id, seq, detail } => write!(
                f,
                "relay checkpoint rejected: frontier for device {device_id} at seq {seq} does \
                 not match the relay-held chain ({detail}) — forged or foreign checkpoint, \
                 refusing to store it (it would become GC coverage for ops it does not cover)"
            ),
            RelayError::Io(e) => write!(f, "relay io error: {e}"),
        }
    }
}

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

/// The relay contract — see the module docs. All methods take `&mut self`
/// because every contact updates roster liveness (and runs the eviction
/// sweep) even on logically-read-only calls.
pub trait Relay {
    /// Enroll (or touch) a device on the roster. Idempotent; push/pull/ack
    /// auto-register on first contact.
    fn register(&mut self, device_id: &str) -> Result<RosterEntry, RelayError>;
    /// Admit `device_id`'s own journal-durable ops (in seq order) onto its
    /// relay-held chain. Contract (B1, binding): the caller transmits only
    /// ops that are already journal-durable on the device.
    fn push(&mut self, device_id: &str, ops: &[OpRecord]) -> Result<PushOutcome, RelayError>;
    /// Every retained op above `since` (per-device seq cursor) + the latest
    /// checkpoint pointer.
    fn pull(&mut self, device_id: &str, since: &Frontier) -> Result<PullResult, RelayError>;
    /// Record a device's fold frontier. Contract (B4, binding): the device
    /// acks only what it has DURABLY folded (journaled), never merely
    /// received.
    fn ack(&mut self, device_id: &str, frontier: Hlc) -> Result<AckOutcome, RelayError>;
    /// Store a device-computed checkpoint. Verified, deduped on
    /// `checkpoint_hash` (whole-record content address — contract from B4).
    /// Returns `false` when the identical checkpoint was already stored.
    fn checkpoint_put(
        &mut self,
        device_id: &str,
        checkpoint: &Checkpoint,
    ) -> Result<bool, RelayError>;
    /// The latest stored checkpoint (dominance-monotone pointer), if any.
    fn checkpoint_get(&mut self) -> Result<Option<Checkpoint>, RelayError>;
    /// The device registry.
    fn roster(&mut self) -> Result<Vec<RosterEntry>, RelayError>;
    /// `min(acked)` over active roster devices; `None` while any active
    /// device has never acked (nothing is droppable then).
    fn stable_frontier(&mut self) -> Result<Option<Hlc>, RelayError>;
    /// Drop every op that is BOTH at/below the stable frontier AND covered
    /// by a stored checkpoint. Never drops anything else.
    fn gc(&mut self) -> Result<GcReport, RelayError>;
}

/// One device's relay-held chain.
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
struct DeviceChain {
    /// Retained ops by seq (contiguous by construction).
    ops: BTreeMap<u64, OpRecord>,
    /// Seqs below this were GC'd (0 = nothing dropped).
    dropped_below: u64,
    /// The last GC'd op's `(seq, op_id, hlc)` — the chain-continuity anchor
    /// for the first retained/pushed op after a GC.
    dropped_head: Option<(u64, String, Hlc)>,
}

impl DeviceChain {
    /// The chain head the next pushed op must link: the last retained op,
    /// else the last GC'd op.
    fn head(&self) -> Option<(u64, &str, &Hlc)> {
        self.ops
            .iter()
            .next_back()
            .map(|(seq, op)| (*seq, op.op_id.as_str(), &op.hlc))
            .or_else(|| {
                self.dropped_head
                    .as_ref()
                    .map(|(seq, id, hlc)| (*seq, id.as_str(), hlc))
            })
    }

    fn next_seq(&self) -> u64 {
        self.head().map(|(seq, _, _)| seq + 1).unwrap_or(0)
    }
}

/// The relay's pure state machine — shared verbatim by [`InMemoryRelay`]
/// (held in memory) and [`FsRelay`] (loaded/persisted around every call).
#[derive(Debug, Default, Serialize, Deserialize)]
struct RelayState {
    roster: BTreeMap<String, RosterEntry>,
    chains: BTreeMap<String, DeviceChain>,
    /// Content-addressed checkpoint store (dedup key = `checkpoint_hash`).
    /// `FsRelay` persists these as their own `<hash>.checkpoint.json`
    /// files (re-verified on load), not inside the state file.
    #[serde(skip)]
    checkpoints: BTreeMap<String, Checkpoint>,
    /// `checkpoint_hash` of the dominance-latest checkpoint.
    latest_checkpoint: Option<String>,
}

/// Does checkpoint `a`'s frontier dominate `b`'s (cover at least every
/// device/seq `b` covers)? The relay's latest-pointer advance rule and the
/// session's rebase-regression guard.
pub(crate) fn frontier_dominates(a: &Checkpoint, b: &Checkpoint) -> bool {
    b.frontier.iter().all(|(device, entry)| {
        a.frontier
            .get(device)
            .is_some_and(|ae: &FrontierEntry| ae.seq >= entry.seq)
    })
}

impl RelayState {
    /// Touch (or enroll) a device and run the eviction sweep. Order
    /// matters: the caller is touched first so it can never evict itself
    /// mid-call.
    fn touch_and_sweep(&mut self, device_id: &str, now_ms: u64, config: &RelayConfig) {
        let entry = self
            .roster
            .entry(device_id.to_string())
            .or_insert_with(|| RosterEntry {
                device_id: device_id.to_string(),
                added_at: Hlc { wall_ms: now_ms, counter: 0, device_id: device_id.to_string() },
                last_seen_ms: now_ms,
                acked: None,
                status: DeviceStatus::Active,
            });
        entry.last_seen_ms = now_ms;
        if let Some(horizon) = config.eviction_horizon_ms {
            for entry in self.roster.values_mut() {
                if entry.status == DeviceStatus::Active
                    && now_ms.saturating_sub(entry.last_seen_ms) > horizon
                {
                    entry.status = DeviceStatus::Evicted;
                }
            }
        }
    }

    fn stable_frontier(&self) -> Option<Hlc> {
        let active: Vec<&RosterEntry> = self
            .roster
            .values()
            .filter(|e| e.status == DeviceStatus::Active)
            .collect();
        if active.is_empty() || active.iter().any(|e| e.acked.is_none()) {
            return None;
        }
        active.iter().filter_map(|e| e.acked.clone()).min()
    }

    fn push(&mut self, device_id: &str, ops: &[OpRecord]) -> Result<PushOutcome, RelayError> {
        let mut outcome = PushOutcome::default();
        // Process in seq order regardless of slice order.
        let mut sorted: Vec<&OpRecord> = ops.iter().collect();
        sorted.sort_by_key(|op| op.seq);
        for op in sorted {
            if op.device_id != device_id {
                return Err(RelayError::ForeignOps {
                    device_id: device_id.to_string(),
                    op_device: op.device_id.clone(),
                });
            }
            if !op.id_valid() {
                return Err(RelayError::Chain {
                    device_id: device_id.to_string(),
                    detail: format!("op {}: stored op_id does not match content", op.op_id),
                });
            }
            if op.hlc.device_id != op.device_id {
                return Err(RelayError::Chain {
                    device_id: device_id.to_string(),
                    detail: format!("op {}: hlc.device_id != device_id", op.op_id),
                });
            }
            let chain = self.chains.entry(device_id.to_string()).or_default();
            if op.seq < chain.dropped_below {
                // The GC'd prefix: a device re-pushing its own already
                // checkpoint-covered ops (crash lost its push cursor).
                outcome.deduped += 1;
                continue;
            }
            if let Some(existing) = chain.ops.get(&op.seq) {
                if existing.op_id == op.op_id {
                    outcome.deduped += 1;
                    continue;
                }
                return Err(RelayError::Fork { device_id: device_id.to_string(), seq: op.seq });
            }
            let expected = chain.next_seq();
            if op.seq != expected {
                return Err(RelayError::Gap {
                    device_id: device_id.to_string(),
                    expected,
                    found: op.seq,
                });
            }
            match chain.head() {
                Some((_, head_id, head_hlc)) => {
                    if op.prev.as_deref() != Some(head_id) {
                        return Err(RelayError::Chain {
                            device_id: device_id.to_string(),
                            detail: format!(
                                "op {}: prev does not link the relay-held head {head_id}",
                                op.op_id
                            ),
                        });
                    }
                    if op.hlc <= *head_hlc {
                        return Err(RelayError::Chain {
                            device_id: device_id.to_string(),
                            detail: format!(
                                "op {}: hlc does not advance past the relay-held head",
                                op.op_id
                            ),
                        });
                    }
                }
                None => {
                    if op.prev.is_some() {
                        return Err(RelayError::Chain {
                            device_id: device_id.to_string(),
                            detail: format!("op {}: seq 0 must have no prev", op.op_id),
                        });
                    }
                }
            }
            chain.ops.insert(op.seq, op.clone());
            outcome.accepted += 1;
        }
        Ok(outcome)
    }

    fn pull(&self, since: &Frontier) -> Result<PullResult, RelayError> {
        let mut ops = Vec::new();
        for (device_id, chain) in &self.chains {
            let start = since.get(device_id).map(|held| held + 1).unwrap_or(0);
            if start < chain.dropped_below {
                return Err(RelayError::FrontierTruncated {
                    device_id: device_id.clone(),
                    dropped_below: chain.dropped_below,
                });
            }
            ops.extend(chain.ops.range(start..).map(|(_, op)| op.clone()));
        }
        ops.sort_by(|a, b| (&a.hlc, &a.op_id).cmp(&(&b.hlc, &b.op_id)));
        Ok(PullResult { ops, latest_checkpoint: self.latest_checkpoint.clone() })
    }

    fn ack(&mut self, device_id: &str, frontier: Hlc) -> AckOutcome {
        // Frontier over the OTHER active devices — the bar a returning
        // straggler must clear to be reinstated.
        let others_frontier = {
            let others: Vec<&RosterEntry> = self
                .roster
                .values()
                .filter(|e| e.status == DeviceStatus::Active && e.device_id != device_id)
                .collect();
            if others.is_empty() || others.iter().any(|e| e.acked.is_none()) {
                None
            } else {
                others.iter().filter_map(|e| e.acked.clone()).min()
            }
        };
        let entry = self.roster.get_mut(device_id).expect("touched before ack");
        let advanced = match &entry.acked {
            Some(current) if frontier <= *current => false,
            _ => {
                entry.acked = Some(frontier);
                true
            }
        };
        let mut reinstated = false;
        if entry.status == DeviceStatus::Evicted {
            // Reinstate when the device has provably caught up: its acked
            // frontier reaches the active stable frontier (or nothing is
            // GC-eligible anyway). Its stale ack must never drag GC back.
            let caught_up = match (&entry.acked, &others_frontier) {
                (Some(acked), Some(frontier)) => acked >= frontier,
                (Some(_), None) => true,
                (None, _) => false,
            };
            if caught_up {
                entry.status = DeviceStatus::Active;
                reinstated = true;
            }
        }
        AckOutcome { advanced, reinstated }
    }

    /// Cross-check a checkpoint's frontier against the relay-held chains —
    /// the defense `Checkpoint::verify` (internal self-consistency) cannot
    /// give. Every frontier entry must name an op the relay actually
    /// holds/held for that device at that seq: the whole prefix a checkpoint
    /// claims to cover has to be the device's REAL chain, or GC would drop
    /// real ops on a forged coverage claim (kernel-review data-loss defect).
    /// A device the relay has never seen a chain for cannot be validated.
    fn validate_frontier(&self, checkpoint: &Checkpoint) -> Result<(), RelayError> {
        for (device_id, entry) in &checkpoint.frontier {
            let unverified = |detail: &str| RelayError::CheckpointFrontierUnverified {
                device_id: device_id.clone(),
                seq: entry.seq,
                detail: detail.to_string(),
            };
            let Some(chain) = self.chains.get(device_id) else {
                return Err(unverified("relay holds no chain for this device"));
            };
            if entry.seq >= chain.dropped_below {
                // Still-retained region: the relay must hold exactly this op.
                match chain.ops.get(&entry.seq) {
                    Some(op) if op.op_id == entry.head => {}
                    Some(_) => {
                        return Err(unverified(
                            "frontier head does not match the relay-held op at this seq",
                        ))
                    }
                    None => {
                        return Err(unverified(
                            "relay holds no op at the claimed frontier seq (claims coverage \
                             beyond its chain head)",
                        ))
                    }
                }
            } else {
                // Already-GC'd region: verify against the remembered dropped
                // head where the seq lands on it; a deeper seq was truncated
                // by a prior *validated* checkpoint, so accept it (it covers
                // only already-dropped ops — no new coverage, no new risk).
                match &chain.dropped_head {
                    Some((seq, id, _)) if *seq == entry.seq => {
                        if id != &entry.head {
                            return Err(unverified(
                                "frontier head does not match the relay's GC'd dropped head",
                            ));
                        }
                    }
                    Some((seq, _, _)) if entry.seq < *seq => {}
                    _ => {
                        return Err(unverified(
                            "claimed frontier seq is below the relay's GC floor with no \
                             matching record",
                        ))
                    }
                }
            }
        }
        Ok(())
    }

    fn checkpoint_put(&mut self, checkpoint: &Checkpoint) -> Result<bool, RelayError> {
        checkpoint.verify().map_err(RelayError::Checkpoint)?;
        // Cross-check the frontier against the relay-held chains BEFORE the
        // checkpoint can become GC coverage — self-consistency (verify) is
        // not enough (kernel-review data-loss defect). Only validated
        // checkpoints enter the store, so `gc` counting stored checkpoints
        // as coverage counts only validated ones.
        self.validate_frontier(checkpoint)?;
        // Dedup on the WHOLE-RECORD content address, never state_hash (B4
        // contract: two frontiers can fold to one state).
        let stored = if self.checkpoints.contains_key(&checkpoint.checkpoint_hash) {
            false
        } else {
            self.checkpoints
                .insert(checkpoint.checkpoint_hash.clone(), checkpoint.clone());
            true
        };
        let advance = match self
            .latest_checkpoint
            .as_ref()
            .and_then(|hash| self.checkpoints.get(hash))
        {
            Some(current) => {
                checkpoint.checkpoint_hash != current.checkpoint_hash
                    && frontier_dominates(checkpoint, current)
            }
            None => true,
        };
        if advance {
            self.latest_checkpoint = Some(checkpoint.checkpoint_hash.clone());
        }
        Ok(stored)
    }

    fn checkpoint_get(&self) -> Option<Checkpoint> {
        self.latest_checkpoint
            .as_ref()
            .and_then(|hash| self.checkpoints.get(hash))
            .cloned()
    }

    fn gc(&mut self) -> GcReport {
        let mut report = GcReport::default();
        let Some(frontier) = self.stable_frontier() else {
            return report;
        };
        // Max covered seq per device across all (validated — every stored
        // checkpoint passed `validate_frontier`) checkpoints, computed once:
        // O(checkpoints × devices), not O(ops × checkpoints) per device.
        let mut max_covered: BTreeMap<String, u64> = BTreeMap::new();
        for ckpt in self.checkpoints.values() {
            for (device, entry) in &ckpt.frontier {
                let slot = max_covered.entry(device.clone()).or_insert(entry.seq);
                if entry.seq > *slot {
                    *slot = entry.seq;
                }
            }
        }
        for (device_id, chain) in &mut self.chains {
            // Both droppability conditions are chain prefixes; walk from the
            // bottom and stop at the first op failing either.
            let covered_through = max_covered.get(device_id).copied();
            let mut droppable: Vec<u64> = Vec::new();
            for (seq, op) in &chain.ops {
                let below_frontier = op.hlc <= frontier;
                let covered = covered_through.is_some_and(|through| through >= *seq);
                if below_frontier && covered {
                    droppable.push(*seq);
                } else {
                    break;
                }
            }
            for seq in &droppable {
                let op = chain.ops.remove(seq).expect("collected from the map");
                chain.dropped_below = seq + 1;
                chain.dropped_head = Some((*seq, op.op_id, op.hlc));
            }
            if !droppable.is_empty() {
                report.dropped.insert(device_id.clone(), droppable.len());
            }
        }
        report
    }
}

/// The in-process reference relay: [`RelayState`] + an injected wall clock.
pub struct InMemoryRelay {
    state: RelayState,
    config: RelayConfig,
    wall: WallClock,
}

impl fmt::Debug for InMemoryRelay {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("InMemoryRelay")
            .field("state", &self.state)
            .field("config", &self.config)
            .finish_non_exhaustive()
    }
}

impl InMemoryRelay {
    pub fn new(config: RelayConfig, wall: WallClock) -> Self {
        Self { state: RelayState::default(), config, wall }
    }
}

impl Relay for InMemoryRelay {
    fn register(&mut self, device_id: &str) -> Result<RosterEntry, RelayError> {
        let now = (self.wall)();
        self.state.touch_and_sweep(device_id, now, &self.config);
        Ok(self.state.roster[device_id].clone())
    }

    fn push(&mut self, device_id: &str, ops: &[OpRecord]) -> Result<PushOutcome, RelayError> {
        let now = (self.wall)();
        self.state.touch_and_sweep(device_id, now, &self.config);
        self.state.push(device_id, ops)
    }

    fn pull(&mut self, device_id: &str, since: &Frontier) -> Result<PullResult, RelayError> {
        let now = (self.wall)();
        self.state.touch_and_sweep(device_id, now, &self.config);
        self.state.pull(since)
    }

    fn ack(&mut self, device_id: &str, frontier: Hlc) -> Result<AckOutcome, RelayError> {
        let now = (self.wall)();
        self.state.touch_and_sweep(device_id, now, &self.config);
        Ok(self.state.ack(device_id, frontier))
    }

    fn checkpoint_put(
        &mut self,
        device_id: &str,
        checkpoint: &Checkpoint,
    ) -> Result<bool, RelayError> {
        let now = (self.wall)();
        self.state.touch_and_sweep(device_id, now, &self.config);
        self.state.checkpoint_put(checkpoint)
    }

    fn checkpoint_get(&mut self) -> Result<Option<Checkpoint>, RelayError> {
        Ok(self.state.checkpoint_get())
    }

    fn roster(&mut self) -> Result<Vec<RosterEntry>, RelayError> {
        Ok(self.state.roster.values().cloned().collect())
    }

    fn stable_frontier(&mut self) -> Result<Option<Hlc>, RelayError> {
        Ok(self.state.stable_frontier())
    }

    fn gc(&mut self) -> Result<GcReport, RelayError> {
        Ok(self.state.gc())
    }
}

/// The filesystem loopback relay: two (or more) `DeviceLog`s syncing
/// through a **shared directory** — the realistic single-user two-Mac case
/// (a shared volume, an external disk, a user-managed synced folder).
///
/// Layout under `dir`:
/// - `relay.lock` — exclusive advisory lock held around every call (the
///   `OplogJournal`/`car-registry` protocol, but *blocking*: concurrent
///   callers queue rather than fail);
/// - `relay-state.json` — roster + chains + latest-checkpoint pointer
///   (temp + atomic rename per mutation);
/// - `checkpoints/<checkpoint_hash>.checkpoint.json` — content-addressed
///   checkpoint files via `Checkpoint::save`/`load`. Checkpoint files are
///   **immutable** (the name IS the whole-record content address), so each
///   is `Checkpoint::load`-verified **once** — on first sight by this
///   handle — and then served from an in-memory cache; it is never
///   re-hashed or re-`fsync`'d on subsequent (including read-only) calls.
///   A tampering attacker in the shared folder is caught by the *next
///   process* to open the relay (a fresh handle with a cold cache
///   re-verifies), which is the actual two-Mac threat model; a handle
///   never serves content it did not verify.
///
/// Semantics are identical to [`InMemoryRelay`] by construction — both
/// drive the same [`RelayState`] core; this type only adds durability and
/// cross-process mutual exclusion.
pub struct FsRelay {
    dir: PathBuf,
    config: RelayConfig,
    wall: WallClock,
    /// Verified-once checkpoint cache (hash → checkpoint). Immutable
    /// content-addressed files never need re-verification within a handle's
    /// lifetime — the fix for O(all checkpoints × state) re-hash per call.
    checkpoint_cache: BTreeMap<String, Checkpoint>,
}

impl fmt::Debug for FsRelay {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("FsRelay")
            .field("dir", &self.dir)
            .field("config", &self.config)
            .finish_non_exhaustive()
    }
}

impl FsRelay {
    pub fn open(dir: &Path, config: RelayConfig, wall: WallClock) -> std::io::Result<Self> {
        fs::create_dir_all(dir.join("checkpoints"))?;
        Ok(Self {
            dir: dir.to_path_buf(),
            config,
            wall,
            checkpoint_cache: BTreeMap::new(),
        })
    }

    fn state_path(&self) -> PathBuf {
        self.dir.join("relay-state.json")
    }

    fn checkpoints_dir(&self) -> PathBuf {
        self.dir.join("checkpoints")
    }

    /// Run `f` over the loaded state under the exclusive lock, then persist
    /// — even when `f` returns a business error (`push` keeps its accepted
    /// prefix, matching the in-memory semantics).
    fn with_state<T>(
        &mut self,
        f: impl FnOnce(&mut RelayState, u64, &RelayConfig) -> Result<T, RelayError>,
    ) -> Result<T, RelayError> {
        let lock = OpenOptions::new()
            .read(true)
            .write(true)
            .create(true)
            .truncate(false)
            .open(self.dir.join("relay.lock"))
            .map_err(RelayError::Io)?;
        lock.lock().map_err(RelayError::Io)?; // blocking; released on drop

        let mut state: RelayState = match fs::read_to_string(self.state_path()) {
            Ok(raw) => serde_json::from_str(&raw)
                .map_err(|e| RelayError::Io(std::io::Error::other(e)))?,
            Err(e) if e.kind() == std::io::ErrorKind::NotFound => RelayState::default(),
            Err(e) => return Err(RelayError::Io(e)),
        };
        // Checkpoint files are immutable content addresses: `Checkpoint::load`
        // (re-verify both hashes + the file-name address) runs ONCE per
        // file, on first sight by this handle; a cache hit serves the
        // already-verified record without re-hashing the whole state.
        for entry in fs::read_dir(self.checkpoints_dir()).map_err(RelayError::Io)? {
            let path = entry.map_err(RelayError::Io)?.path();
            let Some(name) = path.file_name().and_then(|n| n.to_str()) else {
                continue;
            };
            let Some(hash) = name.strip_suffix(".checkpoint.json") else {
                continue;
            };
            let ckpt = match self.checkpoint_cache.get(hash) {
                Some(cached) => cached.clone(),
                None => {
                    let ckpt = Checkpoint::load(&path).map_err(RelayError::Checkpoint)?;
                    self.checkpoint_cache
                        .insert(ckpt.checkpoint_hash.clone(), ckpt.clone());
                    ckpt
                }
            };
            state.checkpoints.insert(ckpt.checkpoint_hash.clone(), ckpt);
        }

        let now = (self.wall)();
        let result = f(&mut state, now, &self.config);

        // Persist: checkpoint files first (content-addressed, immutable),
        // then the state file naming them (temp + atomic rename) — the same
        // durable-referent-first ordering as compact_and_truncate. Skip the
        // fsync-heavy `save` for any checkpoint already on disk (the name IS
        // the content, so an existing file is byte-identical) — the fix for
        // re-fsyncing every checkpoint on every call, including read-only
        // ones.
        for ckpt in state.checkpoints.values() {
            let path = self.checkpoints_dir().join(ckpt.file_name());
            if !path.exists() {
                ckpt.save(&self.checkpoints_dir()).map_err(RelayError::Io)?;
            }
            self.checkpoint_cache
                .entry(ckpt.checkpoint_hash.clone())
                .or_insert_with(|| ckpt.clone());
        }
        let tmp = self.dir.join("relay-state.json.tmp");
        {
            let mut file = File::create(&tmp).map_err(RelayError::Io)?;
            file.write_all(
                serde_json::to_string(&state)
                    .map_err(|e| RelayError::Io(std::io::Error::other(e)))?
                    .as_bytes(),
            )
            .map_err(RelayError::Io)?;
            file.sync_all().map_err(RelayError::Io)?;
        }
        fs::rename(&tmp, self.state_path()).map_err(RelayError::Io)?;
        result
    }
}

impl Relay for FsRelay {
    fn register(&mut self, device_id: &str) -> Result<RosterEntry, RelayError> {
        self.with_state(|state, now, config| {
            state.touch_and_sweep(device_id, now, config);
            Ok(state.roster[device_id].clone())
        })
    }

    fn push(&mut self, device_id: &str, ops: &[OpRecord]) -> Result<PushOutcome, RelayError> {
        self.with_state(|state, now, config| {
            state.touch_and_sweep(device_id, now, config);
            state.push(device_id, ops)
        })
    }

    fn pull(&mut self, device_id: &str, since: &Frontier) -> Result<PullResult, RelayError> {
        self.with_state(|state, now, config| {
            state.touch_and_sweep(device_id, now, config);
            state.pull(since)
        })
    }

    fn ack(&mut self, device_id: &str, frontier: Hlc) -> Result<AckOutcome, RelayError> {
        self.with_state(|state, now, config| {
            state.touch_and_sweep(device_id, now, config);
            Ok(state.ack(device_id, frontier))
        })
    }

    fn checkpoint_put(
        &mut self,
        device_id: &str,
        checkpoint: &Checkpoint,
    ) -> Result<bool, RelayError> {
        self.with_state(|state, now, config| {
            state.touch_and_sweep(device_id, now, config);
            state.checkpoint_put(checkpoint)
        })
    }

    fn checkpoint_get(&mut self) -> Result<Option<Checkpoint>, RelayError> {
        self.with_state(|state, _, _| Ok(state.checkpoint_get()))
    }

    fn roster(&mut self) -> Result<Vec<RosterEntry>, RelayError> {
        self.with_state(|state, _, _| Ok(state.roster.values().cloned().collect()))
    }

    fn stable_frontier(&mut self) -> Result<Option<Hlc>, RelayError> {
        self.with_state(|state, _, _| Ok(state.stable_frontier()))
    }

    fn gc(&mut self) -> Result<GcReport, RelayError> {
        self.with_state(|state, _, _| Ok(state.gc()))
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::oplog::{DeviceLog, Scope, Surface};
    use std::sync::atomic::{AtomicU64, Ordering};
    use std::sync::Arc;

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

    fn mem_relay() -> InMemoryRelay {
        InMemoryRelay::new(RelayConfig::default(), Arc::new(|| 0))
    }

    fn ops_for(device: &str, n: usize) -> (DeviceLog, Vec<OpRecord>) {
        let mut log = DeviceLog::new(device);
        let ops = (0..n)
            .map(|i| {
                log.append(
                    Scope::Personal,
                    Surface::Knowledge,
                    serde_json::json!({"id": format!("{device}-f{i}")}),
                )
            })
            .collect();
        (log, ops)
    }

    #[test]
    fn push_validates_the_chain_and_dedups_retransmission() {
        let mut relay = mem_relay();
        let (mut log, ops) = ops_for("a", 3);

        let outcome = relay.push("a", &ops).unwrap();
        assert_eq!(outcome, PushOutcome { accepted: 3, deduped: 0 });

        // Retransmission (crash lost the push cursor): pure dedup.
        let again = relay.push("a", &ops).unwrap();
        assert_eq!(again, PushOutcome { accepted: 0, deduped: 3 });

        // Continuation accepted.
        let next = log.append(Scope::Personal, Surface::Knowledge, serde_json::json!({"id": "x"}));
        assert_eq!(relay.push("a", &[next]).unwrap().accepted, 1);

        // A gap (skipping a seq) is rejected.
        log.append(Scope::Personal, Surface::Knowledge, serde_json::json!({"id": "skipped"}));
        let ahead = log.append(Scope::Personal, Surface::Knowledge, serde_json::json!({"id": "y"}));
        assert!(matches!(
            relay.push("a", &[ahead]),
            Err(RelayError::Gap { expected: 4, found: 5, .. })
        ));

        // A fork (different op at a held seq) is rejected.
        let mut forked = DeviceLog::new("a");
        let f0 = forked.append(Scope::Personal, Surface::Knowledge, serde_json::json!({"id": "evil"}));
        assert!(matches!(relay.push("a", &[f0]), Err(RelayError::Fork { seq: 0, .. })));

        // Foreign ops are rejected.
        let (_, b_ops) = ops_for("b", 1);
        assert!(matches!(
            relay.push("a", &b_ops),
            Err(RelayError::ForeignOps { .. })
        ));

        // A tampered op is rejected.
        let mut tampered = ops[0].clone();
        tampered.payload = serde_json::json!({"forged": true});
        assert!(matches!(relay.push("b", &[tampered]), Err(RelayError::ForeignOps { .. })));
        let mut own_tampered = ops[0].clone();
        own_tampered.payload = serde_json::json!({"id": "a-f0", "forged": true});
        assert!(matches!(relay.push("a", &[own_tampered]), Err(RelayError::Chain { .. })));
    }

    #[test]
    fn pull_is_a_seq_cursor_and_serves_canonical_order() {
        let mut relay = mem_relay();
        let (_, a_ops) = ops_for("a", 3);
        let (_, b_ops) = ops_for("b", 2);
        relay.push("a", &a_ops).unwrap();
        relay.push("b", &b_ops).unwrap();

        // Fresh puller: everything.
        let all = relay.pull("c", &Frontier::new()).unwrap();
        assert_eq!(all.ops.len(), 5);
        assert!(all.latest_checkpoint.is_none());

        // Cursor past a's seq 1 and all of b: only a's tail comes back.
        let mut since = Frontier::new();
        since.insert("a".into(), 1);
        since.insert("b".into(), 1);
        let tail = relay.pull("c", &since).unwrap();
        assert_eq!(tail.ops.len(), 1);
        assert_eq!(tail.ops[0].seq, 2);
        assert_eq!(tail.ops[0].device_id, "a");
    }

    #[test]
    fn stable_frontier_requires_every_active_device_acked() {
        let mut relay = mem_relay();
        let (_, a_ops) = ops_for("a", 2);
        relay.push("a", &a_ops).unwrap();
        relay.register("b").unwrap();

        // b never acked → no frontier (nothing droppable) — the AckTable
        // refusal semantics, relay-side.
        relay.ack("a", a_ops[1].hlc.clone()).unwrap();
        assert_eq!(relay.stable_frontier().unwrap(), None);

        relay.ack("b", a_ops[0].hlc.clone()).unwrap();
        assert_eq!(relay.stable_frontier().unwrap(), Some(a_ops[0].hlc.clone()), "min(acked)");

        // Monotone-only ack: a replayed lower ack cannot regress it.
        let outcome = relay.ack("b", a_ops[0].hlc.clone()).unwrap();
        assert!(!outcome.advanced);
        let outcome = relay.ack("b", a_ops[1].hlc.clone()).unwrap();
        assert!(outcome.advanced);
        assert_eq!(relay.stable_frontier().unwrap(), Some(a_ops[1].hlc.clone()));
    }

    #[test]
    fn gc_requires_both_frontier_and_covering_checkpoint() {
        let mut relay = mem_relay();
        let (_, ops) = ops_for("a", 4);
        relay.push("a", &ops).unwrap();
        relay.ack("a", ops[3].hlc.clone()).unwrap();

        // Everything is below the stable frontier, but NO covering
        // checkpoint exists → nothing drops.
        assert_eq!(relay.gc().unwrap().total(), 0);

        // Checkpoint covering the first two ops → exactly those drop.
        let ckpt = Checkpoint::from_ops(&ops[..2]).unwrap();
        assert!(relay.checkpoint_put("a", &ckpt).unwrap());
        let report = relay.gc().unwrap();
        assert_eq!(report.dropped["a"], 2);

        // Ops above the frontier never drop: push more, don't ack.
        let mut log = DeviceLog::resume("a", &ops).unwrap();
        let newer = log.append(Scope::Personal, Surface::Knowledge, serde_json::json!({"id": "n"}));
        relay.push("a", std::slice::from_ref(&newer)).unwrap();
        let full_ckpt = {
            let mut all = ops.clone();
            all.push(newer);
            Checkpoint::from_ops(&all).unwrap()
        };
        relay.checkpoint_put("a", &full_ckpt).unwrap();
        // Covered by a checkpoint, but seq 4 is above the acked frontier →
        // only seqs 2..=3 drop.
        let report = relay.gc().unwrap();
        assert_eq!(report.dropped["a"], 2);
        let survivors = relay.pull("b", &Frontier::new());
        // A fresh pull now reaches into truncated space → cold bootstrap.
        assert!(matches!(
            survivors,
            Err(RelayError::FrontierTruncated { dropped_below: 4, .. })
        ));
        // But a caught-up cursor is served the retained tail.
        let mut since = Frontier::new();
        since.insert("a".into(), 3);
        assert_eq!(relay.pull("b", &since).unwrap().ops.len(), 1);
    }

    #[test]
    fn gc_preserves_chain_continuity_for_later_pushes() {
        let mut relay = mem_relay();
        let (mut log, ops) = ops_for("a", 3);
        relay.push("a", &ops).unwrap();
        relay.ack("a", ops[2].hlc.clone()).unwrap();
        let ckpt = Checkpoint::from_ops(&ops).unwrap();
        relay.checkpoint_put("a", &ckpt).unwrap();
        assert_eq!(relay.gc().unwrap().dropped["a"], 3);

        // The whole chain is GC'd; a continuation still push-verifies
        // against the remembered dropped head…
        let next = log.append(Scope::Personal, Surface::Knowledge, serde_json::json!({"id": "n"}));
        assert_eq!(relay.push("a", &[next]).unwrap().accepted, 1);
        // …and a re-push of the GC'd prefix dedups instead of forking.
        assert_eq!(relay.push("a", &ops).unwrap(), PushOutcome { accepted: 0, deduped: 3 });
    }

    #[test]
    fn forged_checkpoint_frontier_is_rejected_and_never_becomes_gc_coverage() {
        // Kernel-review DATA-LOSS repro (through the public API): a
        // self-consistent checkpoint built from a DIFFERENT chain that
        // merely CLAIMS device "a"'s name must not walk in via checkpoint_put
        // (verify() only checks internal hashes) and let GC drop a's REAL
        // ops on the coverage claim.
        let mut relay = mem_relay();

        // a's REAL chain, pushed + acked.
        let mut a = DeviceLog::new("a");
        let real: Vec<OpRecord> = (0..3)
            .map(|i| {
                a.append(Scope::Personal, Surface::Knowledge, serde_json::json!({"id": format!("real-{i}")}))
            })
            .collect();
        relay.push("a", &real).unwrap();
        relay.ack("a", real[2].hlc.clone()).unwrap();

        // A DIFFERENT chain claiming device_id "a" (another account's shared
        // relay dir, a restored-from-backup re-mint). Internally valid.
        let mut fake = DeviceLog::new("a");
        let other: Vec<OpRecord> = (0..3)
            .map(|i| {
                fake.append(Scope::Personal, Surface::Knowledge, serde_json::json!({"id": format!("other-{i}")}))
            })
            .collect();
        let forged = Checkpoint::from_ops(&other).unwrap();

        // The relay REJECTS it — its frontier head does not match the
        // relay-held op at that seq.
        assert!(matches!(
            relay.checkpoint_put("a", &forged),
            Err(RelayError::CheckpointFrontierUnverified { device_id, .. }) if device_id == "a"
        ));

        // Nothing stored → GC has no coverage → a's real ops are NOT dropped.
        // (All probing calls below use "a" as the caller so no extra
        // never-acked device is auto-registered — which would pin the stable
        // frontier to None and mask the final GC assertion.)
        assert_eq!(relay.gc().unwrap().total(), 0, "no forged coverage, no data loss");
        assert!(relay.checkpoint_get().unwrap().is_none());
        // a's real ops are all still served.
        let served = relay.pull("a", &Frontier::new()).unwrap();
        assert_eq!(served.ops.len(), 3);
        assert!(served.ops.iter().all(|op| op.payload["id"].as_str().unwrap().starts_with("real-")));

        // A checkpoint claiming coverage BEYOND the relay's chain head is
        // rejected too (not just a head mismatch).
        let mut a2 = DeviceLog::resume("a", &real).unwrap();
        let mut ahead = real.clone();
        ahead.push(a2.append(Scope::Personal, Surface::Knowledge, serde_json::json!({"id": "real-3"})));
        let claims_beyond = Checkpoint::from_ops(&ahead).unwrap(); // relay never got real-3
        assert!(matches!(
            relay.checkpoint_put("a", &claims_beyond),
            Err(RelayError::CheckpointFrontierUnverified { seq: 3, .. })
        ));

        // A checkpoint claiming a device the relay has never seen is rejected
        // (the frontier device "c" is validated regardless of the caller).
        let mut c = DeviceLog::new("c");
        let c_ops: Vec<OpRecord> = (0..2)
            .map(|i| c.append(Scope::Personal, Surface::Knowledge, serde_json::json!({"id": format!("c{i}")})))
            .collect();
        let unknown_dev = Checkpoint::from_ops(&c_ops).unwrap();
        assert!(matches!(
            relay.checkpoint_put("a", &unknown_dev),
            Err(RelayError::CheckpointFrontierUnverified { device_id, .. }) if device_id == "c"
        ));

        // The HONEST checkpoint over a's real chain is accepted and DOES
        // become coverage — the fix rejects forgeries, not legitimacy.
        let honest = Checkpoint::from_ops(&real[..2]).unwrap();
        assert!(relay.checkpoint_put("a", &honest).unwrap());
        assert_eq!(relay.gc().unwrap().dropped["a"], 2);
    }

    #[test]
    fn checkpoint_dedup_keys_on_whole_record_address_not_state_hash() {
        // The B4 contract, enforced at the relay: two checkpoints folding
        // to the SAME state under DIFFERENT frontiers are BOTH stored.
        // (Knowledge is content-keyed so two devices' identical facts dedup to
        // one folded state — a conversation turn would not, being an event
        // stream keyed by op_id.)
        let mut a = DeviceLog::new("a");
        let mut b = DeviceLog::new("b");
        let fact = serde_json::json!({"id": "f1", "body": "hi"});
        let oa = a.append(Scope::Personal, Surface::Knowledge, fact.clone());
        let ob = b.append(Scope::Personal, Surface::Knowledge, fact);
        let just_a = Checkpoint::from_ops(std::slice::from_ref(&oa)).unwrap();
        let both = Checkpoint::from_ops(&[oa.clone(), ob.clone()]).unwrap();
        assert_eq!(just_a.state_hash, both.state_hash, "the cross-device dedup collision");

        let mut relay = mem_relay();
        // The relay must hold the real chains a checkpoint claims to cover —
        // checkpoint_put now cross-checks the frontier (data-loss fix).
        relay.push("a", std::slice::from_ref(&oa)).unwrap();
        relay.push("b", std::slice::from_ref(&ob)).unwrap();
        assert!(relay.checkpoint_put("a", &just_a).unwrap());
        assert!(relay.checkpoint_put("b", &both).unwrap(), "same state_hash is NOT a dedup");
        assert!(!relay.checkpoint_put("a", &just_a).unwrap(), "same checkpoint_hash IS");

        // The latest pointer sits on the dominating frontier and a stale
        // re-put cannot regress it.
        assert_eq!(
            relay.checkpoint_get().unwrap().unwrap().checkpoint_hash,
            both.checkpoint_hash
        );
        relay.checkpoint_put("a", &just_a).unwrap();
        assert_eq!(
            relay.checkpoint_get().unwrap().unwrap().checkpoint_hash,
            both.checkpoint_hash,
            "dominance-monotone pointer"
        );

        // A tampered checkpoint is refused.
        let mut forged = both.clone();
        forged.state.logs.clear();
        assert!(matches!(
            relay.checkpoint_put("b", &forged),
            Err(RelayError::Checkpoint(CheckpointError::HashMismatch { .. }))
        ));
    }

    #[test]
    fn eviction_unpins_the_frontier_and_ack_reinstates() {
        let (t, wall) = manual_clock();
        let mut relay = InMemoryRelay::new(
            RelayConfig { eviction_horizon_ms: Some(1_000) },
            wall,
        );
        let (_, a_ops) = ops_for("a", 2);
        relay.push("a", &a_ops).unwrap();

        t.store(100, Ordering::SeqCst);
        relay.ack("a", a_ops[1].hlc.clone()).unwrap();
        relay.ack("c", a_ops[0].hlc.clone()).unwrap(); // straggler acks early…
        assert_eq!(relay.stable_frontier().unwrap(), Some(a_ops[0].hlc.clone()));

        // …then goes silent past H. Any other contact sweeps it out.
        t.store(2_000, Ordering::SeqCst);
        relay.register("a").unwrap();
        let roster: BTreeMap<String, RosterEntry> = relay
            .roster()
            .unwrap()
            .into_iter()
            .map(|e| (e.device_id.clone(), e))
            .collect();
        assert_eq!(roster["c"].status, DeviceStatus::Evicted);
        assert_eq!(roster["a"].status, DeviceStatus::Active);
        assert_eq!(
            relay.stable_frontier().unwrap(),
            Some(a_ops[1].hlc.clone()),
            "the evicted device's ack no longer holds the frontier"
        );

        // A stale ack from the returning straggler does NOT reinstate (it
        // would drag GC eligibility back)…
        let outcome = relay.ack("c", a_ops[0].hlc.clone()).unwrap();
        assert!(!outcome.advanced && !outcome.reinstated);
        assert_eq!(relay.stable_frontier().unwrap(), Some(a_ops[1].hlc.clone()));

        // …a caught-up ack does.
        let outcome = relay.ack("c", a_ops[1].hlc.clone()).unwrap();
        assert!(outcome.advanced && outcome.reinstated);
        let roster: BTreeMap<String, RosterEntry> = relay
            .roster()
            .unwrap()
            .into_iter()
            .map(|e| (e.device_id.clone(), e))
            .collect();
        assert_eq!(roster["c"].status, DeviceStatus::Active);
    }

    #[test]
    fn fs_relay_matches_in_memory_semantics_and_persists() {
        let dir = tempfile::tempdir().unwrap();
        let (_, ops) = ops_for("a", 3);
        {
            let mut relay =
                FsRelay::open(dir.path(), RelayConfig::default(), Arc::new(|| 7)).unwrap();
            assert_eq!(relay.push("a", &ops).unwrap().accepted, 3);
            relay.ack("a", ops[2].hlc.clone()).unwrap();
            let ckpt = Checkpoint::from_ops(&ops[..2]).unwrap();
            assert!(relay.checkpoint_put("a", &ckpt).unwrap());
        }
        // A second handle (fresh process) sees the same durable state.
        let mut relay =
            FsRelay::open(dir.path(), RelayConfig::default(), Arc::new(|| 8)).unwrap();
        assert_eq!(relay.stable_frontier().unwrap(), Some(ops[2].hlc.clone()));
        // The pull auto-registers b — an active never-acked device drops
        // the stable frontier to None (nothing droppable) until b acks.
        assert_eq!(relay.pull("b", &Frontier::new()).unwrap().ops, ops);
        assert_eq!(relay.stable_frontier().unwrap(), None);
        relay.ack("b", ops[2].hlc.clone()).unwrap();
        assert_eq!(relay.push("a", &ops).unwrap(), PushOutcome { accepted: 0, deduped: 3 });
        let report = relay.gc().unwrap();
        assert_eq!(report.dropped["a"], 2);
        let roster = relay.roster().unwrap();
        assert_eq!(roster.len(), 2);
        assert_eq!(roster[0].added_at.wall_ms, 7, "roster added_at survives restart");

        // The checkpoint round-trips through its content-addressed file
        // (Checkpoint::load re-verified it).
        let ckpt = relay.checkpoint_get().unwrap().unwrap();
        assert_eq!(ckpt.frontier["a"].seq, 1);

        // A tampered on-disk checkpoint file is refused by the next process
        // to open the relay (a fresh handle with a cold cache re-verifies —
        // the two-Mac shared-folder threat model). The current handle keeps
        // serving the version it already verified into its cache (it never
        // serves unverified content), which is why the check uses a fresh
        // handle.
        let path = dir
            .path()
            .join("checkpoints")
            .join(format!("{}.checkpoint.json", ckpt.checkpoint_hash));
        let raw = fs::read_to_string(&path).unwrap();
        fs::write(&path, raw.replace("\"a-f0\"", "\"a-f0-forged\"")).unwrap();
        let mut fresh =
            FsRelay::open(dir.path(), RelayConfig::default(), Arc::new(|| 9)).unwrap();
        assert!(matches!(
            fresh.checkpoint_get(),
            Err(RelayError::Checkpoint(_))
        ));
    }
}