car-sync 0.34.0

//! Compaction + oplog GC (slice B4 of `docs/proposals/multi-device-sync.md`,
//! §"Deep dive: the checkpoint / compaction / GC protocol").
//!
//! Compaction = **checkpoint at the stable frontier, then drop the ops below
//! it**. Three guards make that safe, each straight from the proposal:
//!
//! 1. **The frontier is the min acked HLC over every known device**
//!    ([`AckTable::stable_frontier`]) — "GC of a CRDT log is only safe once
//!    every replica has folded past the truncation point". Compaction
//!    refuses to run at all when a device present in the log has no ack
//!    entry ([`CompactError::UnackedDevice`]) or when nothing is acked
//!    ([`CompactError::NothingAcked`]): ops above ANY device's acked
//!    frontier are data another device hasn't seen, and are never dropped.
//!    (Device eviction — the horizon `H` that unpins a dead laptop — is
//!    relay policy, B3.)
//! 2. **Checkpoint durable FIRST, then truncate**
//!    ([`compact_and_truncate`]). The crash-ordering invariant: a crash
//!    after the checkpoint fsync but before the journal rename leaves the
//!    full journal plus a redundant checkpoint (harmless — rerunning
//!    compaction recomputes the identical content-addressed file); the
//!    rename itself is atomic, so mid-truncation crashes leave either the
//!    old complete journal or the new complete tail. **At no point does
//!    acknowledged data exist only in a file that isn't durably written.**
//! 3. **Retention is applied to the checkpoint state, never to the tail.**
//!    The proposal's per-surface retention table
//!    ([`RetentionPolicy::proposal_default`]) trims what the snapshot
//!    *keeps*; ops above the frontier are untouched by policy. **Replay**
//!    surfaces (routing observations — path-dependent, replayed from genesis;
//!    [`crate::oplog::Surface::is_replay_stream`]) additionally reject any rule
//!    but keep-all ([`CompactError::EventStreamRetention`]), so an unacked (or
//!    acked!) observation tail can never be compacted away. (Conversation is
//!    an event-stream multiset too — B2 — but its turns are independent, so it
//!    tolerates `LastN`; only replay streams are forbidden.) **Every
//!    retention-dropped id-bearing entry leaves a minimal tombstone stub**
//!    (`{"id": …, "tombstone": true}`, original `op_id`/`hlc` kept), so a
//!    `"supersedes"` reference always resolves against a tombstone, not a
//!    hole — **including a reference that arrives AFTER compaction**.
//!    (Preserving only the references visible at compaction time was a
//!    reproduced divergence: a later tail op superseding an
//!    already-dropped entry made the global fold retain what the compacted
//!    device could not. Universal stubs are time-hole-free: both sides
//!    reduce the same record to the same stub, deterministically. Entries
//!    with no entity id — e.g. content-hash-keyed conversation turns —
//!    cannot be referenced by id and drop entirely.) Stubs are carried
//!    unchanged by later retention passes and never count against a
//!    surface's retention quota.
//!
//! Determinism note (the "same frontier ⇒ same snapshot hash" invariant):
//! retention is deterministic over a fixed state, and the age-based rules'
//! reference instant defaults to [`as_of_from_ops`] — the max payload
//! `"timestamp"` among the ops at/below the frontier, i.e. **pure over the
//! same inputs the fold already consumes**, so every device compacting the
//! same frontier derives the same instant with no out-of-band agreement
//! (this crate never reads a clock; a caller may still pass an explicit
//! `Some(as_of_ms)`). The derived value is conservative: a stale max
//! under-drops, and undated entries never age-drop anyway.
//! Ordering/recency come from the payload's numeric `"timestamp"` field
//! (ms) and grouping from `"agent_id"`; an entry with no timestamp is
//! treated as newest / never age-dropped — undated data is never silently
//! discarded.

use crate::checkpoint::Checkpoint;
use crate::fold::{FoldedRecord, SyncState};
use crate::journal::OplogJournal;
use crate::oplog::{verify_log, ChainError, Hlc, OpRecord};
use serde::{Deserialize, Serialize};
use serde_json::{json, Value};
use std::collections::{BTreeMap, BTreeSet};
use std::fmt;
use std::fs::{self, File};
use std::io::Write;
use std::path::{Path, PathBuf};

/// Parity constants with today's run GC (`car-server-core::run_store`):
/// keep the 50 most recent runs per agent…
pub const RUNS_MAX_PER_AGENT: usize = 50;
/// …and drop runs older than 30 days (in ms), whichever is more
/// restrictive — exactly `RunStore::gc`'s rule, made globally coherent.
pub const RUNS_MAX_AGE_MS: u64 = 30 * 24 * 60 * 60 * 1000;

/// How one surface's checkpoint retention trims — the proposal's table rows.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum RetentionRule {
    /// Keep everything (knowledge/skills — "the valuable distilled state" —
    /// and forced for event-stream surfaces).
    KeepAll,
    /// Keep the last `n` entries by payload `"timestamp"` (conversations —
    /// "last N turns by timestamp (= today's `max_turns`)").
    LastN { n: usize },
    /// Keep entries no older than `max_age_ms` relative to the compaction's
    /// `as_of_ms` (trajectories — "last D days").
    MaxAgeMs { max_age_ms: u64 },
    /// Keep the most recent `max_per_agent` per payload `"agent_id"` AND
    /// drop anything older than `max_age_ms` — both restrictive, the
    /// `RunStore::gc` rule (runs — "50 / agent + 30 days").
    PerAgentWithAge {
        max_per_agent: usize,
        max_age_ms: u64,
    },
}

/// Per-surface retention policy: surface tag → rule, with a conservative
/// keep-all default for unknown surfaces. Applies to the grow-only log tier
/// only — the LWW registry tier is already one compact record per id, and
/// the proposal's table assigns it no retention.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct RetentionPolicy {
    pub rules: BTreeMap<String, RetentionRule>,
    pub default_rule: RetentionRule,
}

impl Default for RetentionPolicy {
    fn default() -> Self {
        Self::keep_all()
    }
}

impl RetentionPolicy {
    /// Retain everything — the exact-checkpoint policy (compaction still
    /// drops ops; the snapshot just keeps their full folded state).
    pub fn keep_all() -> Self {
        Self {
            rules: BTreeMap::new(),
            default_rule: RetentionRule::KeepAll,
        }
    }

    /// The proposal's retention table. Two rows are deployment-configurable
    /// in the proposal itself and therefore parameters here: conversations'
    /// `N` ("= today's `max_turns`" — a memgine config value, not a global
    /// constant) and trajectories' `D` days (given as `max_age_ms`). Runs
    /// use the shipped `RunStore::gc` constants
    /// ([`RUNS_MAX_PER_AGENT`]/[`RUNS_MAX_AGE_MS`]); knowledge, skills, and
    /// routing observations keep all (routing's keep-all is also enforced
    /// structurally — see [`CompactError::EventStreamRetention`]).
    pub fn proposal_default(conversation_last_n: usize, trajectory_max_age_ms: u64) -> Self {
        let mut rules = BTreeMap::new();
        rules.insert(
            "conversation".to_string(),
            RetentionRule::LastN {
                n: conversation_last_n,
            },
        );
        rules.insert(
            "run".to_string(),
            RetentionRule::PerAgentWithAge {
                max_per_agent: RUNS_MAX_PER_AGENT,
                max_age_ms: RUNS_MAX_AGE_MS,
            },
        );
        rules.insert(
            "trajectory".to_string(),
            RetentionRule::MaxAgeMs {
                max_age_ms: trajectory_max_age_ms,
            },
        );
        rules.insert("knowledge".to_string(), RetentionRule::KeepAll);
        rules.insert("skill".to_string(), RetentionRule::KeepAll);
        rules.insert("routing".to_string(), RetentionRule::KeepAll);
        Self {
            rules,
            default_rule: RetentionRule::KeepAll,
        }
    }

    pub fn rule_for(&self, surface_tag: &str) -> &RetentionRule {
        self.rules.get(surface_tag).unwrap_or(&self.default_rule)
    }
}

/// What retention did: per-surface counts of entries dropped entirely
/// (id-less — nothing can reference them) and the entries reduced to
/// tombstone stubs, as `(surface_tag, key)`.
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
pub struct RetentionReport {
    pub dropped: BTreeMap<String, usize>,
    pub tombstoned: Vec<(String, String)>,
}

fn value_ts(payload: &Value) -> Option<u64> {
    payload
        .get("timestamp")
        .and_then(|v| v.as_u64().or_else(|| v.as_f64().map(|f| f as u64)))
}

fn payload_ts(record: &FoldedRecord) -> Option<u64> {
    value_ts(&record.payload)
}

/// The deterministic default reference instant for age-based retention:
/// the max payload `"timestamp"` across `ops` (0 when none carry one).
/// Pure over the same inputs the fold consumes, so every device compacting
/// the same frontier derives the identical value — no out-of-band
/// agreement, no clock read. Conservative by construction: a stale max
/// under-drops (age rules see everything as newer), and undated entries
/// never age-drop regardless.
pub fn as_of_from_ops(ops: &[OpRecord]) -> u64 {
    ops.iter()
        .filter_map(|op| value_ts(&op.payload))
        .max()
        .unwrap_or(0)
}

fn within_age(record: &FoldedRecord, as_of_ms: u64, max_age_ms: u64) -> bool {
    match payload_ts(record) {
        // Future-stamped entries have age 0 (saturating) — kept.
        Some(ts) => as_of_ms.saturating_sub(ts) <= max_age_ms,
        // Undated data is never age-dropped.
        None => true,
    }
}

/// Deterministic recency order: ascending `(timestamp, hlc, op_id)`, with a
/// missing timestamp sorting as newest (never preferentially dropped).
fn recency_sorted(entries: &BTreeMap<String, FoldedRecord>) -> Vec<(&String, &FoldedRecord)> {
    let mut sorted: Vec<(&String, &FoldedRecord)> = entries.iter().collect();
    sorted.sort_by(|(_, a), (_, b)| {
        (payload_ts(a).unwrap_or(u64::MAX), &a.hlc, &a.op_id).cmp(&(
            payload_ts(b).unwrap_or(u64::MAX),
            &b.hlc,
            &b.op_id,
        ))
    });
    sorted
}

fn select_retained(
    entries: &BTreeMap<String, FoldedRecord>,
    rule: &RetentionRule,
    as_of_ms: u64,
) -> BTreeSet<String> {
    match rule {
        RetentionRule::KeepAll => entries.keys().cloned().collect(),
        RetentionRule::LastN { n } => recency_sorted(entries)
            .into_iter()
            .rev()
            .take(*n)
            .map(|(k, _)| k.clone())
            .collect(),
        RetentionRule::MaxAgeMs { max_age_ms } => entries
            .iter()
            .filter(|(_, r)| within_age(r, as_of_ms, *max_age_ms))
            .map(|(k, _)| k.clone())
            .collect(),
        RetentionRule::PerAgentWithAge {
            max_per_agent,
            max_age_ms,
        } => {
            // Rank per agent over ALL entries (recency), then apply both
            // caps restrictively — RunStore::gc's exact semantics.
            let mut per_agent_rank: BTreeMap<&str, usize> = BTreeMap::new();
            let mut keep = BTreeSet::new();
            for (key, record) in recency_sorted(entries).into_iter().rev() {
                let agent = record
                    .payload
                    .get("agent_id")
                    .and_then(Value::as_str)
                    .unwrap_or("");
                let rank = per_agent_rank.entry(agent).or_insert(0);
                let over_count = *rank >= *max_per_agent;
                *rank += 1;
                if !over_count && within_age(record, as_of_ms, *max_age_ms) {
                    keep.insert(key.clone());
                }
            }
            keep
        }
    }
}

/// The entity id a folded log entry answers to for tombstone stubs: the
/// `id:` key form, else the payload's own `"id"`.
fn entity_id<'a>(key: &'a str, record: &'a FoldedRecord) -> Option<&'a str> {
    key.strip_prefix("id:")
        .or_else(|| record.payload.get("id").and_then(Value::as_str))
}

/// Is this record a retention tombstone stub? (`"tombstone": true` is the
/// reserved marker [`apply_retention`] stamps.)
pub fn is_tombstone(record: &FoldedRecord) -> bool {
    record
        .payload
        .get("tombstone")
        .and_then(Value::as_bool)
        .unwrap_or(false)
}

/// The minimal stub a retention-dropped entry leaves behind: the entity id
/// plus the tombstone marker, under the record's original `op_id`/`hlc`.
/// Deterministic from the dropped record, so every device reduces it to the
/// identical stub. Stubbing is idempotent (a stub of a stub is itself).
fn tombstone_of(record: &FoldedRecord, id: &str) -> FoldedRecord {
    FoldedRecord {
        op_id: record.op_id.clone(),
        hlc: record.hlc.clone(),
        payload: json!({"id": id, "tombstone": true}),
    }
}

/// Apply per-surface retention to a folded (checkpoint) state — the
/// proposal's "the snapshot applies each surface's retention", with the
/// §"Two free properties fall out" referential-integrity rule realized as
/// **universal tombstone stubs**: every dropped id-bearing entry is reduced
/// to `{"id", "tombstone": true}` (original `op_id`/`hlc` kept) rather than
/// erased, so a `"supersedes"` reference — even one that arrives after
/// compaction — always resolves. Stubs never count against a rule's quota
/// and are carried unchanged by later passes. Pure over its inputs;
/// `as_of_ms` is the reference instant for the age rules (no clock reads
/// here — [`plan_compaction`] defaults it via [`as_of_from_ops`]).
pub fn apply_retention(
    state: &SyncState,
    policy: &RetentionPolicy,
    as_of_ms: u64,
) -> Result<(SyncState, RetentionReport), CompactError> {
    // Path-dependent REPLAY surfaces (routing's EMA) are retention-forbidden:
    // the folded result is recomputed from the ordered multiset, so any trim
    // silently corrupts every device's replay. Reject loudly.
    //
    // This is NARROWER than "event stream": conversation turns are an
    // op_id-keyed multiset too (B2 — so two genuine same-content turns never
    // collapse), but they are INDEPENDENT entries, so `LastN` over them is
    // well-defined and MUST be allowed (conversations need last-N retention).
    // Only replay streams are rejected here — see `Surface::is_replay_stream`.
    // Routing is the only one today; its tag is the guard key.
    let routing_tag = crate::oplog::Surface::Routing.tag();
    debug_assert!(crate::oplog::Surface::Routing.is_replay_stream());
    if state.logs.contains_key(&routing_tag)
        && policy.rule_for(&routing_tag) != &RetentionRule::KeepAll
    {
        return Err(CompactError::EventStreamRetention {
            surface: routing_tag,
        });
    }

    // Leased execution intents (B5) are KEEP-ALL, structurally protected.
    // `state.intents` (incl. the fence-independent `committed_runs` idempotency
    // oracle) is carried untouched below — the fold routes `Surface::Intent`
    // into `state.intents`, never `state.logs`, so retention here cannot reach
    // it. A retention rule keyed on "intent" would therefore be a silent no-op
    // that misleads a reader into thinking committed records get trimmed; reject
    // it loudly so the keep-all invariant is explicit and reviewable, mirroring
    // the event-stream guard. (The idempotency oracle surviving compaction
    // depends on this: a committed run's record must NEVER be trimmable.)
    if policy.rule_for(&crate::oplog::Surface::Intent.tag()) != &RetentionRule::KeepAll {
        return Err(CompactError::IntentRetention);
    }

    let mut retained = state.clone();
    let mut report = RetentionReport::default();
    for (tag, entries) in &state.logs {
        let rule = policy.rule_for(tag);
        if rule == &RetentionRule::KeepAll {
            continue;
        }
        // Selection ranks LIVE entries only: existing stubs are carried
        // unchanged and never displace a live entry from the quota.
        let live: BTreeMap<String, FoldedRecord> = entries
            .iter()
            .filter(|(_, record)| !is_tombstone(record))
            .map(|(key, record)| (key.clone(), record.clone()))
            .collect();
        let keep = select_retained(&live, rule, as_of_ms);
        let surface = retained.logs.get_mut(tag).expect("cloned from state");
        for (key, record) in &live {
            if keep.contains(key) {
                continue;
            }
            match entity_id(key, record) {
                Some(id) => {
                    surface.insert(key.clone(), tombstone_of(record, id));
                    report.tombstoned.push((tag.clone(), key.clone()));
                }
                None => {
                    surface.remove(key);
                    *report.dropped.entry(tag.clone()).or_insert(0) += 1;
                }
            }
        }
    }
    report.tombstoned.sort();
    Ok((retained, report))
}

/// The fold-frontier / ack bookkeeping B3's relay `ack(frontier)` reports
/// against: per-device max folded HLC, **monotone-only** advance, persisted
/// alongside the checkpoint (temp + atomic rename, like everything durable
/// here). This is the proposal's `acked[device]` table, device-local.
///
/// **MUST, binding on B3: an ack asserts durably-folded state.** A device
/// may report `ack(frontier)` only after the ops at/below that frontier are
/// durably persisted on it (journal-durable, fold applied) — an ack sent
/// from memory ahead of the fsync lets compaction drop ops the acking
/// device then loses in a crash, which is exactly the data loss the stable
/// frontier exists to prevent. The mirror of B1's
/// journal-durable-before-transmit rule.
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
pub struct AckTable {
    acked: BTreeMap<String, Hlc>,
}

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

    /// Advance a device's acked frontier. Monotone-only: an ack at or below
    /// the current frontier is ignored (returns `false`) — a delayed or
    /// replayed ack can never move GC eligibility backwards.
    pub fn ack(&mut self, device_id: impl Into<String>, frontier: Hlc) -> bool {
        let device_id = device_id.into();
        match self.acked.get(&device_id) {
            Some(current) if frontier <= *current => false,
            _ => {
                self.acked.insert(device_id, frontier);
                true
            }
        }
    }

    pub fn get(&self, device_id: &str) -> Option<&Hlc> {
        self.acked.get(device_id)
    }

    pub fn devices(&self) -> impl Iterator<Item = &str> {
        self.acked.keys().map(String::as_str)
    }

    /// The stable frontier — `min(acked[d])` over every known device
    /// (proposal §"Snapshots"). Ops at or below it are folded by everyone
    /// and are the only truncation candidates. `None` when no device has
    /// acked (nothing is ever droppable then).
    pub fn stable_frontier(&self) -> Option<&Hlc> {
        self.acked.values().min()
    }

    /// Durably persist (temp + atomic rename); pairs with
    /// [`AckTable::load`]. Kept alongside the checkpoint directory by
    /// convention.
    pub fn save(&self, path: &Path) -> std::io::Result<()> {
        if let Some(parent) = path.parent() {
            if !parent.as_os_str().is_empty() {
                fs::create_dir_all(parent)?;
            }
        }
        let tmp_path = {
            let mut s = path.as_os_str().to_owned();
            s.push(".tmp");
            PathBuf::from(s)
        };
        {
            let mut tmp = File::create(&tmp_path)?;
            tmp.write_all(
                serde_json::to_string(self)
                    .map_err(std::io::Error::other)?
                    .as_bytes(),
            )?;
            tmp.sync_all()?;
        }
        fs::rename(&tmp_path, path)
    }

    /// Load a persisted table; a missing file is an empty table (a fresh
    /// device knows of no acks — and an empty table makes compaction
    /// refuse, the safe default).
    pub fn load(path: &Path) -> std::io::Result<Self> {
        match fs::read_to_string(path) {
            Ok(raw) => serde_json::from_str(&raw).map_err(std::io::Error::other),
            Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(Self::default()),
            Err(e) => Err(e),
        }
    }
}

/// A compaction failure. No `PartialEq` (carries `io::Error`); match on
/// variants.
#[derive(Debug)]
pub enum CompactError {
    /// The input log doesn't verify — never compact what you can't trust
    /// (the B1 verify-before-fold contract).
    Chain(ChainError),
    /// No device has acked anything: no stable frontier exists, nothing is
    /// provably folded-by-everyone, nothing may be dropped.
    NothingAcked,
    /// A device present in the log has no ack entry — its fold frontier is
    /// unknown, so every op is data it may not have seen. Refuse.
    UnackedDevice {
        device_id: String,
    },
    /// The policy tried to trim an event-stream surface (op_id-keyed
    /// observation multiset). Those replay from genesis; only keep-all is
    /// sound.
    EventStreamRetention {
        surface: String,
    },
    /// The policy set a non-keep-all rule for the leased `intent` surface
    /// (B5). Intents are keep-all — the `committed_runs` idempotency oracle
    /// must survive compaction, so trimming it is never sound.
    IntentRetention,
    /// The journal was already truncated below a checkpoint (its truncation
    /// marker names it). Re-planning from the tail alone would fold a
    /// checkpoint that silently misses everything the prior checkpoint
    /// covers — recompaction over a checkpoint base is a later slice.
    TruncatedJournal {
        checkpoint_hash: String,
    },
    Io(std::io::Error),
}

impl fmt::Display for CompactError {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            CompactError::Chain(e) => write!(f, "compaction refused: log does not verify: {e}"),
            CompactError::NothingAcked => {
                write!(
                    f,
                    "compaction refused: no acked frontier exists (empty ack table)"
                )
            }
            CompactError::UnackedDevice { device_id } => write!(
                f,
                "compaction refused: device {device_id} appears in the log but has no acked \
                 frontier — its ops may include state no other replica has folded"
            ),
            CompactError::EventStreamRetention { surface } => write!(
                f,
                "retention policy for event-stream surface {surface} must be keep_all — \
                 observation multisets replay from genesis and cannot be trimmed"
            ),
            CompactError::IntentRetention => write!(
                f,
                "retention policy for the leased `intent` surface must be keep_all — the \
                 committed-run idempotency oracle must survive compaction and cannot be trimmed"
            ),
            CompactError::TruncatedJournal { checkpoint_hash } => write!(
                f,
                "compaction refused: journal already truncated below checkpoint \
                 {checkpoint_hash} — re-planning from the tail alone would drop that \
                 checkpoint's state (recompaction over a checkpoint base is a later slice)"
            ),
            CompactError::Io(e) => write!(f, "compaction io error: {e}"),
        }
    }
}

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

/// A computed (not yet executed) compaction: the retained checkpoint, the
/// ops that stay in the journal, and what happened.
#[derive(Debug)]
pub struct CompactionPlan {
    /// The checkpoint to persist BEFORE truncating — frontier heads +
    /// retention-applied state, content-addressed over the whole record.
    pub checkpoint: Checkpoint,
    /// Ops strictly above the stable frontier — the journal's new content.
    pub retained_ops: Vec<OpRecord>,
    /// How many ops fall at/below the frontier (dropped from the journal,
    /// covered by the checkpoint).
    pub dropped_ops: usize,
    /// The stable frontier the plan cut at.
    pub frontier: Hlc,
    /// The effective reference instant the age rules ran with (the caller's
    /// explicit value, or [`as_of_from_ops`] over the below-frontier ops).
    pub as_of_ms: u64,
    /// What retention trimmed inside the checkpoint state.
    pub retention: RetentionReport,
}

/// Plan a compaction of `ops` at the [`AckTable`]'s stable frontier. Pure —
/// no file IO; [`compact_and_truncate`] executes a plan durably. See the
/// module docs for the three safety guards; per-device HLC monotonicity
/// (enforced by `verify_log`) guarantees "hlc ≤ frontier" is a chain prefix,
/// so the cut is always anchorable.
///
/// `as_of_ms`: the age rules' reference instant. `None` (the default every
/// caller should want) derives it deterministically via [`as_of_from_ops`]
/// over the below-frontier ops, so identical frontiers yield identical
/// retained checkpoints on every device.
pub fn plan_compaction(
    ops: &[OpRecord],
    acks: &AckTable,
    policy: &RetentionPolicy,
    as_of_ms: Option<u64>,
) -> Result<CompactionPlan, CompactError> {
    verify_log(ops).map_err(CompactError::Chain)?;
    let frontier = acks
        .stable_frontier()
        .cloned()
        .ok_or(CompactError::NothingAcked)?;
    for op in ops {
        if acks.get(&op.device_id).is_none() {
            return Err(CompactError::UnackedDevice {
                device_id: op.device_id.clone(),
            });
        }
    }

    let (below, retained_ops): (Vec<OpRecord>, Vec<OpRecord>) =
        ops.iter().cloned().partition(|op| op.hlc <= frontier);
    let as_of_ms = as_of_ms.unwrap_or_else(|| as_of_from_ops(&below));

    // Exact fold at the frontier first (from_ops re-verifies the prefix),
    // then retention on the snapshot only — never on the tail.
    let exact = Checkpoint::from_ops(&below).map_err(CompactError::Chain)?;
    let (retained_state, retention) = apply_retention(&exact.state, policy, as_of_ms)?;
    let checkpoint = Checkpoint::assemble(exact.frontier, exact.scopes, retained_state);

    Ok(CompactionPlan {
        checkpoint,
        retained_ops,
        dropped_ops: below.len(),
        frontier,
        as_of_ms,
        retention,
    })
}

/// The outcome of an executed compaction.
#[derive(Debug)]
pub struct CompactionOutcome {
    /// Where the checkpoint landed (content-addressed file in
    /// `checkpoint_dir`) — `None` when the plan dropped nothing (nothing at
    /// or below the frontier), in which case neither a checkpoint write nor
    /// a truncation happened: an empty compaction is a no-op, not an empty
    /// checkpoint file.
    pub checkpoint_path: Option<PathBuf>,
    pub plan: CompactionPlan,
}

/// Execute a compaction end-to-end on a live journal, enforcing the
/// crash-ordering invariant by construction:
///
/// 1. load + verify the journal (a journal already carrying a truncation
///    marker is refused — [`CompactError::TruncatedJournal`] — because
///    re-planning from the tail alone would silently lose the prior
///    checkpoint's state);
/// 2. plan at the ack table's stable frontier (a plan that drops nothing
///    is a **no-op**: no checkpoint written, journal untouched);
/// 3. **checkpoint durable FIRST** ([`Checkpoint::save`]: temp + fsync +
///    atomic rename);
/// 4. only then truncate the journal to the retained tail
///    ([`OplogJournal::truncate_to`]: temp + atomic rename under the
///    journal's advisory lock, stamping the truncation marker that fences
///    `DeviceLog::resume`).
///
/// A crash between 3 and 4 leaves the full journal plus a redundant
/// checkpoint; rerunning is idempotent (same frontier ⇒ same
/// content-addressed checkpoint file, rename-over-identical). A crash
/// inside 4 leaves either the old or the new journal, whole. Acknowledged
/// data is never lost. (Steps are inherently sequential — each depends on
/// the previous one's durability.)
pub fn compact_and_truncate(
    journal: &mut OplogJournal,
    checkpoint_dir: &Path,
    acks: &AckTable,
    policy: &RetentionPolicy,
    as_of_ms: Option<u64>,
) -> Result<CompactionOutcome, CompactError> {
    let (marker, ops) = OplogJournal::load_with_marker(journal.path()).map_err(CompactError::Io)?;
    if let Some(marker) = marker {
        return Err(CompactError::TruncatedJournal {
            checkpoint_hash: marker.checkpoint_hash,
        });
    }
    let plan = plan_compaction(&ops, acks, policy, as_of_ms)?;
    if plan.dropped_ops == 0 {
        // Nothing at or below the frontier: writing an empty checkpoint and
        // rewriting the journal to itself would be pure churn. No-op.
        return Ok(CompactionOutcome {
            checkpoint_path: None,
            plan,
        });
    }
    // INVARIANT: checkpoint durable BEFORE any op leaves the journal.
    let checkpoint_path = plan
        .checkpoint
        .save(checkpoint_dir)
        .map_err(CompactError::Io)?;
    journal
        .truncate_to(&plan.retained_ops, &plan.checkpoint.checkpoint_hash)
        .map_err(CompactError::Io)?;
    Ok(CompactionOutcome {
        checkpoint_path: Some(checkpoint_path),
        plan,
    })
}

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

    fn hlc(wall_ms: u64, device: &str) -> Hlc {
        Hlc {
            wall_ms,
            counter: 0,
            device_id: device.into(),
        }
    }

    #[test]
    fn ack_table_is_monotone_only_and_min_frontier() {
        let mut acks = AckTable::new();
        assert_eq!(acks.stable_frontier(), None);
        assert!(acks.ack("a", hlc(5, "a")));
        assert!(acks.ack("b", hlc(9, "b")));
        assert_eq!(
            acks.stable_frontier(),
            Some(&hlc(5, "a")),
            "min over devices"
        );

        // Regression is ignored — a replayed/late ack can't move GC back.
        assert!(!acks.ack("b", hlc(3, "b")));
        assert!(!acks.ack("b", hlc(9, "b")), "equal is not an advance");
        assert_eq!(acks.get("b"), Some(&hlc(9, "b")));
        assert!(acks.ack("b", hlc(12, "b")));
        assert_eq!(acks.get("b"), Some(&hlc(12, "b")));
    }

    #[test]
    fn ack_table_persists_atomically_and_loads_missing_as_empty() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("nested").join("acks.json");
        assert_eq!(
            AckTable::load(&path).unwrap(),
            AckTable::new(),
            "missing → empty"
        );

        let mut acks = AckTable::new();
        acks.ack("a", hlc(5, "a"));
        acks.ack("b", hlc(9, "b"));
        acks.save(&path).unwrap();
        assert_eq!(AckTable::load(&path).unwrap(), acks);
        // No temp file left behind.
        assert!(!path.parent().unwrap().join("acks.json.tmp").exists());
    }

    /// Ops: device a emits 3 knowledge facts, b (having observed) emits 1.
    fn simple_ops() -> Vec<OpRecord> {
        let mut a = DeviceLog::new("a");
        let mut b = DeviceLog::new("b");
        let mut ops = vec![
            a.append(Scope::Personal, Surface::Knowledge, json!({"id": "f1"})),
            a.append(Scope::Personal, Surface::Knowledge, json!({"id": "f2"})),
            a.append(Scope::Personal, Surface::Knowledge, json!({"id": "f3"})),
        ];
        for op in &ops {
            b.observe(&op.hlc);
        }
        ops.push(b.append(Scope::Personal, Surface::Knowledge, json!({"id": "f4"})));
        ops
    }

    #[test]
    fn compaction_refuses_without_acks() {
        let ops = simple_ops();
        let policy = RetentionPolicy::keep_all();
        assert!(matches!(
            plan_compaction(&ops, &AckTable::new(), &policy, None),
            Err(CompactError::NothingAcked)
        ));

        // A device in the log with no ack entry → refuse: its fold frontier
        // is unknown.
        let mut acks = AckTable::new();
        acks.ack("a", ops[2].hlc.clone());
        assert!(matches!(
            plan_compaction(&ops, &acks, &policy, None),
            Err(CompactError::UnackedDevice { .. })
        ));
    }

    #[test]
    fn compaction_never_drops_above_a_lagging_ack() {
        let ops = simple_ops();
        let mut acks = AckTable::new();
        // b has folded everything; a's ack lags at its own second op.
        acks.ack("b", ops[3].hlc.clone());
        acks.ack("a", ops[1].hlc.clone());

        let plan = plan_compaction(&ops, &acks, &RetentionPolicy::keep_all(), None).unwrap();
        assert_eq!(
            plan.frontier, ops[1].hlc,
            "stable frontier = the lagging device's ack"
        );
        assert_eq!(plan.dropped_ops, 2, "only ops ≤ the lagging frontier drop");
        assert_eq!(plan.retained_ops.len(), 2);
        assert!(
            plan.retained_ops.iter().all(|op| op.hlc > plan.frontier),
            "everything a device hasn't seen stays in the journal"
        );
    }

    #[test]
    fn retention_conversations_last_n_by_timestamp() {
        let mut dev = DeviceLog::new("a");
        let ops: Vec<OpRecord> = (0..5)
            .map(|i| {
                dev.append(
                    Scope::Personal,
                    Surface::Conversation,
                    json!({"speaker": "u", "text": format!("t{i}"), "timestamp": 100 + i}),
                )
            })
            .collect();
        let state = fold(&ops);
        let policy = RetentionPolicy::proposal_default(2, u64::MAX);
        let (retained, report) = apply_retention(&state, &policy, 1_000).unwrap();
        let tag = Surface::Conversation.tag();
        let texts: Vec<String> = retained
            .log_entries(&tag)
            .iter()
            .map(|r| r.payload["text"].as_str().unwrap().to_string())
            .collect();
        assert_eq!(texts, vec!["t3", "t4"], "last 2 turns by timestamp survive");
        assert_eq!(report.dropped[&tag], 3);
    }

    #[test]
    fn retention_runs_per_agent_and_age_matches_run_store_gc() {
        assert_eq!(
            RUNS_MAX_PER_AGENT, 50,
            "parity with run_store DEFAULT_MAX_RUNS_PER_AGENT"
        );
        assert_eq!(
            RUNS_MAX_AGE_MS,
            30 * 24 * 60 * 60 * 1000,
            "parity with DEFAULT_MAX_AGE_DAYS"
        );

        // Behavior with small numbers: keep 2 per agent AND drop older than
        // age 50 — both restrictive.
        let mut dev = DeviceLog::new("a");
        let mut ops = Vec::new();
        for (id, agent, ts) in [
            ("r1", "milo", 10u64), // over per-agent cap AND stale
            ("r2", "milo", 60),    // within both → kept
            ("r3", "milo", 70),    // within both → kept
            ("r4", "other", 10),   // within cap, but stale → dropped
            ("r5", "other", 80),   // kept
        ] {
            ops.push(dev.append(
                Scope::Personal,
                Surface::Run,
                json!({"id": id, "agent_id": agent, "timestamp": ts}),
            ));
        }
        let state = fold(&ops);
        let mut policy = RetentionPolicy::keep_all();
        policy.rules.insert(
            "run".to_string(),
            RetentionRule::PerAgentWithAge {
                max_per_agent: 2,
                max_age_ms: 50,
            },
        );
        let (retained, report) = apply_retention(&state, &policy, 100).unwrap();
        let entries = retained.log_entries(&Surface::Run.tag());
        let kept: Vec<&str> = entries
            .iter()
            .filter(|r| !is_tombstone(r))
            .map(|r| r.payload["id"].as_str().unwrap())
            .collect();
        assert_eq!(kept, vec!["r2", "r3", "r5"]);
        // Runs carry ids → the trimmed ones leave stubs, not holes.
        let stubs: Vec<&str> = entries
            .iter()
            .filter(|r| is_tombstone(r))
            .map(|r| r.payload["id"].as_str().unwrap())
            .collect();
        assert_eq!(stubs, vec!["r1", "r4"]);
        assert_eq!(report.tombstoned.len(), 2);
        assert!(
            report.dropped.is_empty(),
            "id-bearing entries are stubbed, never erased"
        );
    }

    #[test]
    fn retention_trajectories_by_age_and_undated_never_dropped() {
        let mut dev = DeviceLog::new("a");
        let ops = vec![
            dev.append(
                Scope::Personal,
                Surface::Trajectory,
                json!({"id": "old", "timestamp": 10}),
            ),
            dev.append(
                Scope::Personal,
                Surface::Trajectory,
                json!({"id": "new", "timestamp": 90}),
            ),
            dev.append(
                Scope::Personal,
                Surface::Trajectory,
                json!({"id": "undated"}),
            ),
        ];
        let state = fold(&ops);
        let mut policy = RetentionPolicy::keep_all();
        policy.rules.insert(
            "trajectory".to_string(),
            RetentionRule::MaxAgeMs { max_age_ms: 30 },
        );
        let (retained, _) = apply_retention(&state, &policy, 100).unwrap();
        let entries = retained.log_entries(&Surface::Trajectory.tag());
        let kept: Vec<&str> = entries
            .iter()
            .filter(|r| !is_tombstone(r))
            .map(|r| r.payload["id"].as_str().unwrap())
            .collect();
        assert!(kept.contains(&"new"));
        assert!(
            kept.contains(&"undated"),
            "undated data is never silently age-dropped"
        );
        assert!(!kept.contains(&"old"));
        // The aged-out trajectory left a stub, not a hole.
        assert!(entries
            .iter()
            .any(|r| is_tombstone(r) && r.payload["id"] == json!("old")));
    }

    #[test]
    fn knowledge_and_skills_keep_all_under_the_proposal_default() {
        let mut dev = DeviceLog::new("a");
        let ops = vec![
            dev.append(
                Scope::Personal,
                Surface::Knowledge,
                json!({"id": "f1", "timestamp": 1}),
            ),
            dev.append(
                Scope::Personal,
                Surface::Skill,
                json!({"id": "s1", "timestamp": 1}),
            ),
        ];
        let state = fold(&ops);
        // Aggressive everything-else policy; knowledge/skills still keep all.
        let (retained, report) =
            apply_retention(&state, &RetentionPolicy::proposal_default(1, 1), u64::MAX).unwrap();
        assert_eq!(retained.logs[&Surface::Knowledge.tag()].len(), 1);
        assert_eq!(retained.logs[&Surface::Skill.tag()].len(), 1);
        assert!(report.dropped.is_empty());
    }

    #[test]
    fn event_stream_retention_is_rejected() {
        let mut dev = DeviceLog::new("a");
        let ops = vec![
            dev.append(Scope::Personal, Surface::Routing, json!({"sample": 1.0})),
            dev.append(Scope::Personal, Surface::Routing, json!({"sample": 0.0})),
        ];
        let state = fold(&ops);
        let mut policy = RetentionPolicy::keep_all();
        policy
            .rules
            .insert("routing".to_string(), RetentionRule::LastN { n: 1 });
        assert!(matches!(
            apply_retention(&state, &policy, 0),
            Err(CompactError::EventStreamRetention { .. })
        ));
        // And the proposal default keeps the whole multiset.
        let (retained, _) =
            apply_retention(&state, &RetentionPolicy::proposal_default(10, 10), u64::MAX).unwrap();
        assert_eq!(retained.log_entries(&Surface::Routing.tag()).len(), 2);
    }

    #[test]
    fn every_dropped_id_bearing_entry_leaves_a_tombstone_stub() {
        // f3 supersedes f2 supersedes f1; LastN(1) keeps only f3 live —
        // and EVERY trimmed id-bearing entry (f0, f1, f2) leaves a stub,
        // whether or not anything references it *yet* (the time-hole fix:
        // a supersedes that arrives after compaction still resolves).
        let mut dev = DeviceLog::new("a");
        let ops = vec![
            dev.append(
                Scope::Personal,
                Surface::Knowledge,
                json!({"id": "f1", "timestamp": 1}),
            ),
            dev.append(
                Scope::Personal,
                Surface::Knowledge,
                json!({"id": "f2", "timestamp": 2, "supersedes": "f1"}),
            ),
            dev.append(
                Scope::Personal,
                Surface::Knowledge,
                json!({"id": "f3", "timestamp": 3, "supersedes": ["f2"]}),
            ),
            dev.append(
                Scope::Personal,
                Surface::Knowledge,
                json!({"id": "f0", "timestamp": 0}),
            ),
        ];
        let state = fold(&ops);
        let mut policy = RetentionPolicy::keep_all();
        policy
            .rules
            .insert("knowledge".to_string(), RetentionRule::LastN { n: 1 });
        let (retained, report) = apply_retention(&state, &policy, 10).unwrap();
        let tag = Surface::Knowledge.tag();
        let surface = &retained.logs[&tag];
        assert_eq!(
            surface["id:f3"].payload["timestamp"],
            json!(3),
            "newest stays live"
        );
        for id in ["f0", "f1", "f2"] {
            let stub = &surface[&format!("id:{id}")];
            assert!(is_tombstone(stub), "{id} left a stub");
            assert_eq!(
                stub.payload,
                json!({"id": id, "tombstone": true}),
                "minimal stub shape"
            );
            assert_eq!(
                stub.op_id,
                state.logs[&tag][&format!("id:{id}")].op_id,
                "stub keeps the original op identity"
            );
        }
        assert_eq!(report.tombstoned.len(), 3);
        assert!(report.dropped.is_empty());

        // Idempotent + quota-neutral: re-applying the policy changes
        // nothing — stubs are carried, and they don't consume f3's slot.
        let (again, report2) = apply_retention(&retained, &policy, 10).unwrap();
        assert_eq!(again, retained);
        assert!(report2.tombstoned.is_empty());
    }

    #[test]
    fn derived_as_of_is_the_max_below_frontier_timestamp() {
        let mut dev = DeviceLog::new("a");
        let ops = vec![
            dev.append(
                Scope::Personal,
                Surface::Trajectory,
                json!({"id": "t1", "timestamp": 40}),
            ),
            dev.append(
                Scope::Personal,
                Surface::Trajectory,
                json!({"id": "t2", "timestamp": 100}),
            ),
            dev.append(Scope::Personal, Surface::Skill, json!({"id": "s1"})), // undated
        ];
        assert_eq!(as_of_from_ops(&ops), 100, "max payload timestamp");
        assert_eq!(
            as_of_from_ops(&ops[2..]),
            0,
            "no timestamps → 0 (age rules drop nothing)"
        );

        // plan_compaction defaults to the derived value — pure over the
        // below-frontier ops, so every device agrees with no out-of-band
        // coordination.
        let mut acks = AckTable::new();
        acks.ack("a", ops[2].hlc.clone());
        let mut policy = RetentionPolicy::keep_all();
        policy.rules.insert(
            "trajectory".to_string(),
            RetentionRule::MaxAgeMs { max_age_ms: 30 },
        );
        let plan = plan_compaction(&ops, &acks, &policy, None).unwrap();
        assert_eq!(plan.as_of_ms, 100);
        // age(t1) = 100 - 40 = 60 > 30 → stubbed; t2 lives.
        let surface = &plan.checkpoint.state.logs[&Surface::Trajectory.tag()];
        assert!(is_tombstone(&surface["id:t1"]));
        assert!(!is_tombstone(&surface["id:t2"]));
    }

    #[test]
    fn empty_below_frontier_compaction_is_a_no_op() {
        let dir = tempfile::tempdir().unwrap();
        let journal_path = dir.path().join("oplog.jsonl");
        let ckpt_dir = dir.path().join("checkpoints");

        // All ops sit ABOVE the acked frontier (a device acked long ago and
        // never caught up): nothing is droppable.
        let ops = simple_ops();
        let mut journal = OplogJournal::open(&journal_path).unwrap();
        for op in &ops {
            journal.append(op).unwrap();
        }
        let mut acks = AckTable::new();
        acks.ack(
            "a",
            Hlc {
                wall_ms: 0,
                counter: 0,
                device_id: "a".into(),
            },
        );
        acks.ack(
            "b",
            Hlc {
                wall_ms: 0,
                counter: 0,
                device_id: "b".into(),
            },
        );

        let before = fs::read_to_string(&journal_path).unwrap();
        let outcome = compact_and_truncate(
            &mut journal,
            &ckpt_dir,
            &acks,
            &RetentionPolicy::keep_all(),
            None,
        )
        .unwrap();
        assert_eq!(outcome.plan.dropped_ops, 0);
        assert!(
            outcome.checkpoint_path.is_none(),
            "no empty checkpoint file written"
        );
        assert!(!ckpt_dir.exists(), "checkpoint dir not even created");
        assert_eq!(
            fs::read_to_string(&journal_path).unwrap(),
            before,
            "journal untouched (no marker, no rewrite)"
        );
        // And it is still loadable the normal way (no truncation happened).
        assert_eq!(OplogJournal::load(&journal_path).unwrap(), ops);
    }

    #[test]
    fn recompacting_an_already_truncated_journal_is_refused() {
        let dir = tempfile::tempdir().unwrap();
        let journal_path = dir.path().join("oplog.jsonl");
        let ckpt_dir = dir.path().join("checkpoints");
        let ops = simple_ops();
        let mut journal = OplogJournal::open(&journal_path).unwrap();
        for op in &ops {
            journal.append(op).unwrap();
        }
        let mut acks = AckTable::new();
        acks.ack("a", ops[1].hlc.clone());
        acks.ack("b", ops[1].hlc.clone());
        let outcome = compact_and_truncate(
            &mut journal,
            &ckpt_dir,
            &acks,
            &RetentionPolicy::keep_all(),
            None,
        )
        .unwrap();
        let expected_hash = outcome.plan.checkpoint.checkpoint_hash.clone();

        // A second compaction over the truncated journal would fold a
        // checkpoint missing the prior one's state — refused, naming the
        // checkpoint to re-anchor on.
        acks.ack("a", ops[3].hlc.clone());
        acks.ack("b", ops[3].hlc.clone());
        match compact_and_truncate(
            &mut journal,
            &ckpt_dir,
            &acks,
            &RetentionPolicy::keep_all(),
            None,
        ) {
            Err(CompactError::TruncatedJournal { checkpoint_hash }) => {
                assert_eq!(checkpoint_hash, expected_hash)
            }
            other => panic!("expected TruncatedJournal refusal, got {other:?}"),
        }
    }

    #[test]
    fn plan_refuses_an_invalid_log() {
        let mut ops = simple_ops();
        ops[1].payload = json!({"forged": true});
        let mut acks = AckTable::new();
        acks.ack("a", ops[2].hlc.clone());
        acks.ack("b", ops[3].hlc.clone());
        assert!(matches!(
            plan_compaction(&ops, &acks, &RetentionPolicy::keep_all(), None),
            Err(CompactError::Chain(ChainError::IdMismatch { .. }))
        ));
    }

    #[test]
    fn intent_retention_rule_is_rejected() {
        // B5 keep-all guard: a non-keep-all rule on the leased `intent` surface
        // is rejected loudly (mirroring the event-stream guard), so the
        // committed-run idempotency oracle can never be configured away.
        use crate::lease::{Intent, IntentStatus};
        let mut dev = DeviceLog::new("a");
        let ops = vec![dev.append(
            Scope::Personal,
            Surface::Intent,
            Intent::new("milo", "R", 1, IntentStatus::Committed).payload(),
        )];
        let state = fold(&ops);
        let mut policy = RetentionPolicy::keep_all();
        policy
            .rules
            .insert("intent".to_string(), RetentionRule::LastN { n: 1 });
        assert!(matches!(
            apply_retention(&state, &policy, 0),
            Err(CompactError::IntentRetention)
        ));
        // keep-all is fine and preserves the committed oracle untouched.
        let (retained, _) = apply_retention(&state, &RetentionPolicy::keep_all(), 0).unwrap();
        assert!(retained.committed_run("milo", "R").is_some());
    }

    #[test]
    fn c2_committed_run_survives_a_checkpoint_compaction() {
        // C2 REPRO: a committed run below the stable frontier is truncated from
        // the journal by compaction. Its record must NOT be lost — the
        // fence-independent committed-run oracle is carried keep-all into the
        // checkpoint, so idempotency survives compaction.
        use crate::fold::fold_onto;
        use crate::lease::{Intent, IntentStatus};

        let mut a = DeviceLog::new("a");
        let mut b = DeviceLog::new("b");
        let mut ops = vec![a.append(
            Scope::Personal,
            Surface::Intent,
            Intent::new("milo", "R", 1, IntentStatus::Committed).payload(),
        )];
        let split = ops.len();
        for op in &ops {
            b.observe(&op.hlc);
        }
        // A later, higher-epoch tail op (above the frontier) raises the fence.
        ops.push(b.append(
            Scope::Personal,
            Surface::Intent,
            Intent::new("milo", "S", 2, IntentStatus::Pending).payload(),
        ));

        // Frontier cut so committed R (epoch 1) is BELOW and dropped.
        let frontier = ops[..split].iter().map(|o| o.hlc.clone()).max().unwrap();
        let mut acks = AckTable::new();
        acks.ack("a", frontier.clone());
        acks.ack("b", frontier);
        let plan = plan_compaction(&ops, &acks, &RetentionPolicy::keep_all(), None).unwrap();
        assert_eq!(
            plan.dropped_ops, split,
            "committed R is below the frontier, dropped from journal"
        );

        // The raw R op is gone from the retained tail, but the oracle survives
        // in the checkpoint — the idempotency answer is durable.
        assert!(
            plan.checkpoint.state.committed_run("milo", "R").is_some(),
            "committed R survives compaction in the checkpoint oracle (C2 fixed)"
        );
        let reconstructed = fold_onto(&plan.checkpoint.state, &plan.retained_ops);
        assert_eq!(
            reconstructed,
            fold(&ops),
            "fold_onto(checkpoint, tail) == fold(full)"
        );
        assert!(
            reconstructed.committed_run("milo", "R").is_some(),
            "oracle intact post-compaction"
        );
    }
}