memstead_base/filesystem/

changelog.rs

//! JSONL changelog at `.memstead/changes.jsonl` for filesystem mems.
//!
//! filesystem mems have no commit history, so the agent's `note` parameter and
//! the per-mutation provenance trail (timestamp, mutation type, entity
//! id, actor) lands here instead. Append-only; one line per mutation;
//! the file is created lazily on first append.
//!
//! ## Line shape
//!
//! Each line is a JSON object terminated by `\n`:
//!
//! ```json
//! {"ts":"2026-05-08T15:42:13.001Z","kind":"create","entity":"mem:slug","actor":"agent","note":"first draft"}
//! ```
//!
//! - `ts` — RFC 3339 timestamp with millisecond precision, always UTC
//!   (`Z` suffix). Sortable lexicographically.
//! - `kind` — mutation kind: `create`, `update`, `delete`, `relate`,
//!   `rename`, or `batch`. Future kinds extend the set; readers should
//!   tolerate unknown values.
//! - `entity` — mem-relative entity id (`mem:slug` form), or
//!   `null` for batch mutations that span multiple entities.
//! - `actor` — caller category from
//!   [`memstead_base::vcs::Actor::as_trailer`]: `agent`, `cli`, `app`,
//!   `external`, `unknown`.
//! - `note` — agent-authored provenance note when present. Omitted
//!   from the JSON object when absent or whitespace-only.
//! - `client` — optional caller identity (`name@version`). Omitted
//!   when absent.
//!
//! ## Atomicity
//!
//! Writes use `OpenOptions::append(true)` with a single `write_all`
//! call against the buffered line. POSIX guarantees `O_APPEND` writes
//! ≤ `PIPE_BUF` (4 KiB on Linux/macOS) appear atomically; the line
//! shape stays well under that. Concurrent writers from a single
//! process are serialised by the file's append-mode kernel lock.
//! Cross-process concurrency is out of scope (filesystem mems are single-writer
//! per the plan).

use std::io::Write as _;
use std::path::{Path, PathBuf};

use serde::Serialize;

use crate::provenance::ProvenanceKind;
use crate::vcs::{Actor, ClientId};

/// Conventional path of the changelog inside a workspace root.
pub fn changelog_path(workspace_root: &Path) -> PathBuf {
    workspace_root
        .join(crate::mem::MEM_META_DIR)
        .join("changes.jsonl")
}

/// Mutation kind written to the `kind` field of each line. The set
/// covers today's MCP mutating tools; new mutations extend the enum.
/// Readers branch on the string form (stable wire shape).
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum MutationKind {
    Create,
    Update,
    Delete,
    Relate,
    Rename,
    Batch,
}

impl MutationKind {
    pub fn as_str(&self) -> &'static str {
        match self {
            MutationKind::Create => "create",
            MutationKind::Update => "update",
            MutationKind::Delete => "delete",
            MutationKind::Relate => "relate",
            MutationKind::Rename => "rename",
            MutationKind::Batch => "batch",
        }
    }
}

/// Two enums name the same six mutation classes — one is the legacy
/// folder-backend on-disk encoder, the other is the backend-neutral
/// shape consumed by [`crate::backend::MemBackend`]. Bridge here so
/// callers crossing between the two surfaces don't drift.
impl From<ProvenanceKind> for MutationKind {
    fn from(k: ProvenanceKind) -> Self {
        match k {
            ProvenanceKind::Create => MutationKind::Create,
            ProvenanceKind::Update => MutationKind::Update,
            ProvenanceKind::Delete => MutationKind::Delete,
            ProvenanceKind::Relate => MutationKind::Relate,
            ProvenanceKind::Rename => MutationKind::Rename,
            ProvenanceKind::Batch => MutationKind::Batch,
        }
    }
}

impl From<MutationKind> for ProvenanceKind {
    fn from(k: MutationKind) -> Self {
        match k {
            MutationKind::Create => ProvenanceKind::Create,
            MutationKind::Update => ProvenanceKind::Update,
            MutationKind::Delete => ProvenanceKind::Delete,
            MutationKind::Relate => ProvenanceKind::Relate,
            MutationKind::Rename => ProvenanceKind::Rename,
            MutationKind::Batch => ProvenanceKind::Batch,
        }
    }
}

/// A single mutation event. Built at the call site in the filesystem
/// engine (criterion 1's wiring) and passed to [`append_change`].
pub struct ChangeEntry<'a> {
    pub kind: MutationKind,
    /// Mem-relative entity id, or `None` for batch mutations.
    pub entity: Option<&'a str>,
    pub actor: Actor,
    /// Optional caller identity (e.g. `claude-code@2.1.0`).
    pub client: Option<&'a ClientId>,
    /// Optional agent-authored provenance note. Whitespace-only values
    /// are treated as absent.
    pub note: Option<&'a str>,
    /// Correlation id linking every commit produced by a single
    /// logical operation (notably multi-mem `memstead_rename`). `None`
    /// for single-call mutations that don't participate in
    /// correlation. Round-trips through the JSONL wire shape; reader
    /// reconstructs the same value on `read_provenance`.
    pub logical_operation_id: Option<&'a str>,
    /// Caller-declared role (agent-trust plan 13); `Unspecified`
    /// omits the field — absence recorded as absence.
    pub role: crate::vcs::Role,
    /// Caller-declared identity (agent-trust plan 15); `None` omits
    /// the field — absence recorded as absence, same as the role.
    pub identity: Option<&'a str>,
}

/// Errors surfaced by [`append_change`].
#[derive(Debug, thiserror::Error)]
pub enum ChangelogError {
    #[error("changelog io error at {path}: {source}")]
    Io {
        path: PathBuf,
        #[source]
        source: std::io::Error,
    },
    #[error("changelog serialisation: {0}")]
    Serialise(#[from] serde_json::Error),
}

/// Append a single change to `<workspace_root>/.memstead/changes.jsonl`.
/// Creates the `.memstead/` parent directory and the file if absent.
pub fn append_change(workspace_root: &Path, entry: &ChangeEntry<'_>) -> Result<(), ChangelogError> {
    let now = std::time::SystemTime::now();
    append_change_monotonic(workspace_root, entry, now)
}

/// Variant of [`append_change_at`] that guarantees the written `ts`
/// is strictly greater than the changelog's current last-line `ts`.
///
/// The last-line `ts` doubles as the folder backend's drift cursor
/// (`current_head()`), and the engine's self-write bookkeeping treats
/// "cursor unchanged" as "no commit landed" — so two mutations landing
/// inside the same millisecond must not share a `ts`, or the second
/// one becomes invisible to drift detection and the events channel.
/// When `now` is not strictly after the last line (same-millisecond
/// commits, or a clock that stepped backwards), the written timestamp
/// is bumped to `last + 1ms`. The format and the lexicographic-cursor
/// dialect are unchanged — strictly increasing fixed-width RFC 3339
/// stays strictly increasing lexicographically.
///
/// Production callers route here; [`append_change_at`] stays the exact
/// write-this-timestamp primitive for deterministic tests.
pub fn append_change_monotonic(
    workspace_root: &Path,
    entry: &ChangeEntry<'_>,
    now: std::time::SystemTime,
) -> Result<(), ChangelogError> {
    // Compare at the cursor's own granularity — the formatted
    // millisecond string — not raw `SystemTime`: `now` carries
    // sub-millisecond nanos that make it "later" than the parsed
    // last-line ts even when both format to the same millisecond,
    // which is exactly the collision the clamp exists to prevent.
    let effective = match last_line_ts(&changelog_path(workspace_root)) {
        Some(last) if format_rfc3339_utc(now) <= format_rfc3339_utc(last) => {
            last + std::time::Duration::from_millis(1)
        }
        _ => now,
    };
    append_change_at(workspace_root, entry, effective)
}

/// Read the `ts` of the changelog's last non-empty line, parsed back
/// to a `SystemTime`. `None` when the file is absent, unreadable, or
/// its last line doesn't carry a parseable `ts` — the monotonic clamp
/// is best-effort and must never block an append. Reads only the tail
/// of the file (last 4 KiB) so appends stay O(1) in changelog length;
/// a line is well under 4 KiB per the atomicity contract above.
fn last_line_ts(path: &Path) -> Option<std::time::SystemTime> {
    use std::io::{Read as _, Seek as _, SeekFrom};
    let mut file = std::fs::File::open(path).ok()?;
    let len = file.metadata().ok()?.len();
    let start = len.saturating_sub(4096);
    file.seek(SeekFrom::Start(start)).ok()?;
    let mut buf = Vec::new();
    file.read_to_end(&mut buf).ok()?;
    // Lossy: a seek that lands mid-UTF-8-char only garbles the first
    // (incomplete) line of the tail, which the rev() scan never needs.
    let tail = String::from_utf8_lossy(&buf);
    tail.lines()
        .rev()
        .map(str::trim)
        .filter(|l| !l.is_empty())
        .filter_map(|l| {
            let ts = serde_json::from_str::<serde_json::Value>(l)
                .ok()?
                .get("ts")?
                .as_str()?
                .to_string();
            parse_rfc3339_utc(&ts)
        })
        .next()
}

/// Variant of [`append_change`] that takes the timestamp explicitly.
/// Used by tests that need deterministic line ordering; production
/// callers go through [`append_change`].
pub fn append_change_at(
    workspace_root: &Path,
    entry: &ChangeEntry<'_>,
    now: std::time::SystemTime,
) -> Result<(), ChangelogError> {
    let target = changelog_path(workspace_root);
    if let Some(parent) = target.parent() {
        std::fs::create_dir_all(parent).map_err(|e| ChangelogError::Io {
            path: parent.to_path_buf(),
            source: e,
        })?;
    }

    let ts = format_rfc3339_utc(now);
    let note = entry
        .note
        .map(str::trim)
        .filter(|n| !n.is_empty())
        .map(|s| s.to_string());
    let client = entry.client.map(|c| format!("{}@{}", c.name, c.version));

    #[derive(Serialize)]
    struct Wire<'a> {
        ts: &'a str,
        kind: &'a str,
        entity: Option<&'a str>,
        actor: &'a str,
        #[serde(skip_serializing_if = "Option::is_none")]
        note: Option<String>,
        #[serde(skip_serializing_if = "Option::is_none")]
        client: Option<String>,
        #[serde(skip_serializing_if = "Option::is_none", rename = "logical_op")]
        logical_operation_id: Option<&'a str>,
        #[serde(skip_serializing_if = "Option::is_none")]
        role: Option<&'static str>,
        #[serde(skip_serializing_if = "Option::is_none")]
        identity: Option<&'a str>,
    }

    let mut line = serde_json::to_string(&Wire {
        ts: &ts,
        kind: entry.kind.as_str(),
        entity: entry.entity,
        actor: entry.actor.as_trailer(),
        note,
        client,
        role: entry.role.as_trailer(),
        identity: entry.identity,
        logical_operation_id: entry.logical_operation_id,
    })?;
    line.push('\n');

    let mut file = std::fs::OpenOptions::new()
        .create(true)
        .append(true)
        .open(&target)
        .map_err(|e| ChangelogError::Io {
            path: target.clone(),
            source: e,
        })?;
    file.write_all(line.as_bytes())
        .map_err(|e| ChangelogError::Io {
            path: target,
            source: e,
        })?;
    Ok(())
}

/// Format a `SystemTime` as RFC 3339 with millisecond precision and
/// the `Z` suffix. Always UTC. Hand-rolled because the project does
/// not pull in `chrono` and the underlying `std::time::SystemTime` is
/// epoch-based.
///
/// Public so the folder-backend [`crate::backend::MemBackend`] impl
/// can reuse the same encoder when it constructs cursor strings.
pub fn format_rfc3339_utc(now: std::time::SystemTime) -> String {
    let dur = now
        .duration_since(std::time::UNIX_EPOCH)
        .unwrap_or_default();
    let total_secs = dur.as_secs();
    let millis = dur.subsec_millis();
    let (y, m, d, hh, mm, ss) = decompose_unix_seconds(total_secs);
    format!("{y:04}-{m:02}-{d:02}T{hh:02}:{mm:02}:{ss:02}.{millis:03}Z")
}

/// Inverse of [`format_rfc3339_utc`]. Returns `None` when `s` is not
/// the exact 24-character `YYYY-MM-DDTHH:MM:SS.mmmZ` shape this
/// crate emits — round-trip precision is the contract; lenient
/// parsing is not. Used by the folder-backend `MemBackend::read_provenance`
/// impl when reconstructing [`crate::provenance::Provenance`] records
/// from on-disk JSONL lines.
pub fn parse_rfc3339_utc(s: &str) -> Option<std::time::SystemTime> {
    let bytes = s.as_bytes();
    if bytes.len() != 24
        || bytes[23] != b'Z'
        || bytes[4] != b'-'
        || bytes[7] != b'-'
        || bytes[10] != b'T'
        || bytes[13] != b':'
        || bytes[16] != b':'
        || bytes[19] != b'.'
    {
        return None;
    }
    let year: i32 = s.get(0..4)?.parse().ok()?;
    let month: u32 = s.get(5..7)?.parse().ok()?;
    let day: u32 = s.get(8..10)?.parse().ok()?;
    let hour: u32 = s.get(11..13)?.parse().ok()?;
    let minute: u32 = s.get(14..16)?.parse().ok()?;
    let second: u32 = s.get(17..19)?.parse().ok()?;
    let millis: u32 = s.get(20..23)?.parse().ok()?;
    if month == 0
        || month > 12
        || day == 0
        || day > 31
        || hour > 23
        || minute > 59
        || second > 60
        || millis > 999
    {
        return None;
    }
    let days = ymd_to_days(year, month, day)?;
    let total_secs = days
        .checked_mul(86_400)?
        .checked_add(hour as i64 * 3_600)?
        .checked_add(minute as i64 * 60)?
        .checked_add(second as i64)?;
    if total_secs < 0 {
        return None;
    }
    Some(std::time::UNIX_EPOCH + std::time::Duration::new(total_secs as u64, millis * 1_000_000))
}

/// Civil-to-days inverse of the algorithm in [`decompose_unix_seconds`].
/// Returns days since 1970-01-01. Hinnant's algorithm.
fn ymd_to_days(year: i32, month: u32, day: u32) -> Option<i64> {
    let y = if month <= 2 {
        year as i64 - 1
    } else {
        year as i64
    };
    let m = month as i64;
    let era = if y >= 0 { y } else { y - 399 } / 400;
    let yoe = y - era * 400;
    if !(0..=399).contains(&yoe) {
        return None;
    }
    let doy_offset = if m > 2 { m - 3 } else { m + 9 };
    let doy = (153 * doy_offset + 2) / 5 + day as i64 - 1;
    let doe = yoe * 365 + yoe / 4 - yoe / 100 + doy;
    Some(era * 146_097 + doe - 719_468)
}

/// Decompose a UNIX-epoch second count into (year, month, day, hour,
/// minute, second) in UTC. Algorithm from Howard Hinnant's
/// [chrono-Compatible Low-Level Date Algorithms]
/// (https://howardhinnant.github.io/date_algorithms.html).
///
/// Valid for any seconds-since-epoch value the caller is likely to
/// observe (`SystemTime` on POSIX is bounded by `time_t`, well within
/// the algorithm's i64-domain).
fn decompose_unix_seconds(total_secs: u64) -> (i32, u32, u32, u32, u32, u32) {
    const SECONDS_PER_DAY: u64 = 86_400;
    let days = (total_secs / SECONDS_PER_DAY) as i64;
    let secs_of_day = (total_secs % SECONDS_PER_DAY) as u32;
    let hh = secs_of_day / 3_600;
    let mm = (secs_of_day / 60) % 60;
    let ss = secs_of_day % 60;

    // Civil-from-days.
    let z = days + 719_468;
    let era = z.div_euclid(146_097);
    let doe = z - era * 146_097;
    let yoe = (doe - doe / 1_460 + doe / 36_524 - doe / 146_096) / 365;
    let y = (yoe + era * 400) as i32;
    let doy = doe - (365 * yoe + yoe / 4 - yoe / 100);
    let mp = (5 * doy + 2) / 153;
    let d = (doy - (153 * mp + 2) / 5 + 1) as u32;
    let m = if mp < 10 {
        (mp + 3) as u32
    } else {
        (mp - 9) as u32
    };
    let y = if m <= 2 { y + 1 } else { y };
    (y, m, d, hh, mm, ss)
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::vcs::{Actor, ClientId};
    use tempfile::TempDir;

    fn read_lines(path: &Path) -> Vec<String> {
        std::fs::read_to_string(path)
            .unwrap()
            .lines()
            .map(|s| s.to_string())
            .collect()
    }

    fn ts(seconds: u64, millis: u32) -> std::time::SystemTime {
        std::time::UNIX_EPOCH + std::time::Duration::new(seconds, millis * 1_000_000)
    }

    #[test]
    fn appends_a_create_line_with_all_fields() {
        let tmp = TempDir::new().unwrap();
        let client = ClientId {
            name: "claude-code".into(),
            version: "2.1.0".into(),
        };
        append_change_at(
            tmp.path(),
            &ChangeEntry {
                kind: MutationKind::Create,
                entity: Some("spec:hello"),
                actor: Actor::Agent,
                client: Some(&client),
                note: Some("first draft"),
                logical_operation_id: None,
                role: crate::vcs::Role::Unspecified,
                identity: None,
            },
            ts(1_715_000_000, 1),
        )
        .unwrap();

        let lines = read_lines(&changelog_path(tmp.path()));
        assert_eq!(lines.len(), 1);
        let value: serde_json::Value = serde_json::from_str(&lines[0]).unwrap();
        assert_eq!(value["kind"], "create");
        assert_eq!(value["entity"], "spec:hello");
        assert_eq!(value["actor"], "agent");
        assert_eq!(value["client"], "claude-code@2.1.0");
        assert_eq!(value["note"], "first draft");
        // Timestamp shape is RFC 3339 UTC, ms precision, sortable.
        let ts_str = value["ts"].as_str().unwrap();
        assert!(ts_str.ends_with("Z"));
        assert!(ts_str.contains("T"));
        // 2024-05-06T12:53:20 UTC + 1ms == seconds=1_715_000_000, ms=1.
        assert_eq!(ts_str, "2024-05-06T12:53:20.001Z");
        // An identity-less entry carries no identity key — absence
        // recorded as absence (agent-trust plan 15, criterion 3).
        assert!(value.get("identity").is_none());
    }

    /// Agent-trust plan 15: a declared identity lands as the
    /// `identity` field of the JSONL line, verbatim.
    #[test]
    fn appends_the_declared_identity() {
        let tmp = TempDir::new().unwrap();
        append_change_at(
            tmp.path(),
            &ChangeEntry {
                kind: MutationKind::Update,
                entity: Some("spec:hello"),
                actor: Actor::Agent,
                client: None,
                note: None,
                logical_operation_id: None,
                role: crate::vcs::Role::Unspecified,
                identity: Some("plenum-agent"),
            },
            ts(1_715_000_000, 1),
        )
        .unwrap();
        let lines = read_lines(&changelog_path(tmp.path()));
        let value: serde_json::Value = serde_json::from_str(&lines[0]).unwrap();
        assert_eq!(value["identity"], "plenum-agent");
    }

    #[test]
    fn omits_optional_fields_when_absent() {
        let tmp = TempDir::new().unwrap();
        append_change_at(
            tmp.path(),
            &ChangeEntry {
                kind: MutationKind::Update,
                entity: Some("spec:foo"),
                actor: Actor::Cli,
                client: None,
                note: None,
                logical_operation_id: None,
                role: crate::vcs::Role::Unspecified,
                identity: None,
            },
            ts(0, 0),
        )
        .unwrap();
        let lines = read_lines(&changelog_path(tmp.path()));
        let value: serde_json::Value = serde_json::from_str(&lines[0]).unwrap();
        assert!(value.get("note").is_none());
        assert!(value.get("client").is_none());
    }

    #[test]
    fn whitespace_only_note_is_treated_as_absent() {
        let tmp = TempDir::new().unwrap();
        append_change_at(
            tmp.path(),
            &ChangeEntry {
                kind: MutationKind::Update,
                entity: Some("spec:foo"),
                actor: Actor::Cli,
                client: None,
                note: Some("   \t   "),
                logical_operation_id: None,
                role: crate::vcs::Role::Unspecified,
                identity: None,
            },
            ts(0, 0),
        )
        .unwrap();
        let lines = read_lines(&changelog_path(tmp.path()));
        let value: serde_json::Value = serde_json::from_str(&lines[0]).unwrap();
        assert!(value.get("note").is_none());
    }

    #[test]
    fn batch_mutation_writes_null_entity() {
        let tmp = TempDir::new().unwrap();
        append_change_at(
            tmp.path(),
            &ChangeEntry {
                kind: MutationKind::Batch,
                entity: None,
                actor: Actor::Agent,
                client: None,
                note: Some("multi-entity refactor"),
                logical_operation_id: None,
                role: crate::vcs::Role::Unspecified,
                identity: None,
            },
            ts(0, 0),
        )
        .unwrap();
        let lines = read_lines(&changelog_path(tmp.path()));
        let value: serde_json::Value = serde_json::from_str(&lines[0]).unwrap();
        assert!(value["entity"].is_null());
        assert_eq!(value["kind"], "batch");
    }

    #[test]
    fn appends_create_then_update_in_order() {
        let tmp = TempDir::new().unwrap();
        for (kind, ent, t) in [
            (MutationKind::Create, "a", 0),
            (MutationKind::Update, "a", 1),
            (MutationKind::Delete, "a", 2),
        ] {
            append_change_at(
                tmp.path(),
                &ChangeEntry {
                    kind,
                    entity: Some(ent),
                    actor: Actor::Cli,
                    client: None,
                    note: None,
                    logical_operation_id: None,
                    role: crate::vcs::Role::Unspecified,
                    identity: None,
                },
                ts(t, 0),
            )
            .unwrap();
        }

        let lines = read_lines(&changelog_path(tmp.path()));
        assert_eq!(lines.len(), 3);
        let kinds: Vec<String> = lines
            .iter()
            .map(|line| {
                serde_json::from_str::<serde_json::Value>(line).unwrap()["kind"]
                    .as_str()
                    .unwrap()
                    .to_string()
            })
            .collect();
        assert_eq!(kinds, vec!["create", "update", "delete"]);
    }

    #[test]
    fn creates_memstead_parent_lazily() {
        let tmp = TempDir::new().unwrap();
        assert!(!tmp.path().join(".memstead").exists());
        append_change_at(
            tmp.path(),
            &ChangeEntry {
                kind: MutationKind::Create,
                entity: Some("a"),
                actor: Actor::Cli,
                client: None,
                note: None,
                logical_operation_id: None,
                role: crate::vcs::Role::Unspecified,
                identity: None,
            },
            ts(0, 0),
        )
        .unwrap();
        assert!(tmp.path().join(".memstead").is_dir());
        assert!(tmp.path().join(".memstead").join("changes.jsonl").is_file());
    }

    #[test]
    fn does_not_create_memstead_until_first_change() {
        let tmp = TempDir::new().unwrap();
        // The `memstead init` step is responsible for creating `.memstead/`,
        // but the changelog must not require an extra setup call —
        // the helper creates the parent on demand. Verify that just
        // reading `changelog_path` does not touch disk.
        let _ = changelog_path(tmp.path());
        assert!(!tmp.path().join(".memstead").exists());
    }

    #[test]
    fn timestamp_handles_year_2026() {
        // Sanity-check the civil-from-days helper at a recent epoch
        // that does not sit on a friendly boundary. seconds=1_777_077_296
        // is 2026-04-25 00:34:56 UTC.
        let s = format_rfc3339_utc(ts(1_777_077_296, 0));
        assert_eq!(s, "2026-04-25T00:34:56.000Z");
    }

    #[test]
    fn timestamp_round_trips_a_known_2026_date() {
        // 2026-05-08T12:34:56 UTC. Verifies the helper holds across
        // months — leap-year handling at year boundaries lives in the
        // epoch and 2024 tests above.
        // Days since 1970-01-01:
        //   56 full years (14 leap) + day-of-year 127 = 20581 days
        // 20581 * 86400 + 12*3600 + 34*60 + 56 = 1_778_243_696
        let s = format_rfc3339_utc(ts(1_778_243_696, 0));
        assert_eq!(s, "2026-05-08T12:34:56.000Z");
    }

    #[test]
    fn timestamp_handles_epoch() {
        let s = format_rfc3339_utc(ts(0, 0));
        assert_eq!(s, "1970-01-01T00:00:00.000Z");
    }

    fn bare_entry() -> ChangeEntry<'static> {
        ChangeEntry {
            kind: MutationKind::Create,
            entity: Some("spec:x"),
            actor: Actor::Agent,
            client: None,
            note: None,
            logical_operation_id: None,
            role: crate::vcs::Role::Unspecified,
            identity: None,
        }
    }

    fn line_ts(line: &str) -> String {
        serde_json::from_str::<serde_json::Value>(line).unwrap()["ts"]
            .as_str()
            .unwrap()
            .to_string()
    }

    #[test]
    fn monotonic_append_bumps_same_millisecond_timestamp() {
        // The last-line ts is the folder drift cursor; two commits in
        // the same millisecond must still advance it, or the second
        // becomes invisible to drift detection and the events channel.
        let tmp = TempDir::new().unwrap();
        let now = ts(1_778_243_696, 500);
        append_change_monotonic(tmp.path(), &bare_entry(), now).unwrap();
        append_change_monotonic(tmp.path(), &bare_entry(), now).unwrap();
        let lines = read_lines(&changelog_path(tmp.path()));
        assert_eq!(line_ts(&lines[0]), "2026-05-08T12:34:56.500Z");
        assert_eq!(line_ts(&lines[1]), "2026-05-08T12:34:56.501Z");
    }

    #[test]
    fn monotonic_append_bumps_backwards_clock() {
        let tmp = TempDir::new().unwrap();
        append_change_monotonic(tmp.path(), &bare_entry(), ts(1_778_243_696, 500)).unwrap();
        // Clock stepped backwards a full second — still clamps to
        // last + 1ms rather than writing a regressing cursor.
        append_change_monotonic(tmp.path(), &bare_entry(), ts(1_778_243_695, 0)).unwrap();
        let lines = read_lines(&changelog_path(tmp.path()));
        assert_eq!(line_ts(&lines[1]), "2026-05-08T12:34:56.501Z");
    }

    #[test]
    fn monotonic_append_keeps_strictly_later_timestamp_verbatim() {
        let tmp = TempDir::new().unwrap();
        append_change_monotonic(tmp.path(), &bare_entry(), ts(1_778_243_696, 500)).unwrap();
        append_change_monotonic(tmp.path(), &bare_entry(), ts(1_778_243_696, 501)).unwrap();
        let lines = read_lines(&changelog_path(tmp.path()));
        assert_eq!(line_ts(&lines[1]), "2026-05-08T12:34:56.501Z");
    }

    #[test]
    fn rfc3339_parser_round_trips_known_dates() {
        for &(secs, ms) in &[
            (0u64, 0u32),
            (1_715_000_000, 1),
            (1_777_077_296, 0),
            (1_778_243_696, 0),
            (1_778_243_696, 999),
        ] {
            let t = ts(secs, ms);
            let s = format_rfc3339_utc(t);
            let parsed = parse_rfc3339_utc(&s).expect("parse round-trip");
            assert_eq!(parsed, t, "round-trip failed for {s}");
        }
    }

    #[test]
    fn rfc3339_parser_rejects_malformed_input() {
        assert!(parse_rfc3339_utc("").is_none());
        assert!(parse_rfc3339_utc("2026-05-08T12:34:56Z").is_none()); // missing ms
        assert!(parse_rfc3339_utc("2026-05-08T12:34:56.000+02:00").is_none()); // wrong tz
        assert!(parse_rfc3339_utc("2026-13-08T12:34:56.000Z").is_none()); // bad month
        assert!(parse_rfc3339_utc("not a date at all aaaaa").is_none());
    }

    #[test]
    fn mutation_kind_str_is_stable() {
        // The string form is the wire shape — these values are read by
        // external tools (jq, grep). Lock them.
        assert_eq!(MutationKind::Create.as_str(), "create");
        assert_eq!(MutationKind::Update.as_str(), "update");
        assert_eq!(MutationKind::Delete.as_str(), "delete");
        assert_eq!(MutationKind::Relate.as_str(), "relate");
        assert_eq!(MutationKind::Rename.as_str(), "rename");
        assert_eq!(MutationKind::Batch.as_str(), "batch");
    }
}

/// A folder mem's ledger set against its file set, reported and never
/// repaired (consistency-sweep 04/04, criteria 1 and 2).
///
/// The ledger is the change record AND the drift cursor, and it is written
/// only by the engine. A hand-edited, hand-added or hand-deleted markdown
/// file appends no line, so `changes_since` reports an edit that happened as
/// never having happened, and a ledger line can name a file that is gone.
/// Neither is a parse bug; both are the record and the files disagreeing, and
/// nothing said so.
#[derive(Debug, Clone, Default, serde::Serialize, serde::Deserialize, PartialEq, Eq)]
pub struct LedgerReconciliation {
    /// Entity ids the ledger records but whose file is absent.
    pub ledger_without_file: Vec<String>,
    /// Entity ids present as files that the ledger never mentions.
    pub file_without_ledger: Vec<String>,
}

impl LedgerReconciliation {
    pub fn is_clean(&self) -> bool {
        self.ledger_without_file.is_empty() && self.file_without_ledger.is_empty()
    }
}

/// Reconcile `mem_root`'s ledger against the markdown files beside it.
///
/// **Reads only.** No ledger line is written, rewritten or removed, and no
/// file is touched. Writing lines for edits the engine did not author would
/// fabricate provenance for a change it cannot attribute, which is why this
/// reports rather than tidies (criterion 2).
///
/// Called on demand, never on the staleness probe: that runs before every
/// operation, and turning it into a directory walk would change the cost
/// profile of the whole folder backend.
pub fn reconcile_ledger(mem_root: &Path) -> Result<LedgerReconciliation, ChangelogError> {
    let mut on_disk: std::collections::BTreeSet<String> = Default::default();
    let meta_dir = mem_root.join(crate::mem::MEM_META_DIR);
    let mut stack = vec![mem_root.to_path_buf()];
    while let Some(dir) = stack.pop() {
        let entries = match std::fs::read_dir(&dir) {
            Ok(e) => e,
            Err(e) if e.kind() == std::io::ErrorKind::NotFound => continue,
            Err(source) => {
                return Err(ChangelogError::Io {
                    path: dir.clone(),
                    source,
                });
            }
        };
        for entry in entries.flatten() {
            let path = entry.path();
            // The engine's own sidecar is not entity content.
            if path == meta_dir {
                continue;
            }
            if path.is_dir() {
                stack.push(path);
            } else if path.extension().is_some_and(|x| x == "md")
                && let Some(stem) = path.file_stem().and_then(|s| s.to_str())
            {
                on_disk.insert(stem.to_string());
            }
        }
    }

    let mut in_ledger: std::collections::BTreeSet<String> = Default::default();
    let log_path = changelog_path(mem_root);
    match std::fs::read_to_string(&log_path) {
        Ok(raw) => {
            for line in raw.lines() {
                let trimmed = line.trim();
                if trimmed.is_empty() {
                    continue;
                }
                if let Ok(v) = serde_json::from_str::<serde_json::Value>(trimmed)
                    && let Some(entity) = v.get("entity").and_then(|x| x.as_str())
                {
                    // Ledger ids are mem-qualified (`mem--slug`); files are
                    // named by the slug alone. Compare on the slug, which is
                    // what both surfaces actually agree on.
                    let slug = entity.rsplit("--").next().unwrap_or(entity);
                    in_ledger.insert(slug.to_string());
                }
            }
        }
        Err(e) if e.kind() == std::io::ErrorKind::NotFound => {}
        Err(source) => {
            return Err(ChangelogError::Io {
                path: log_path,
                source,
            });
        }
    }

    Ok(LedgerReconciliation {
        ledger_without_file: in_ledger.difference(&on_disk).cloned().collect(),
        file_without_ledger: on_disk.difference(&in_ledger).cloned().collect(),
    })
}

#[cfg(test)]
mod reconcile_tests {
    use super::*;
    use tempfile::TempDir;

    fn seed(root: &Path, files: &[&str], ledger_entities: &[&str]) {
        for f in files {
            std::fs::write(root.join(format!("{f}.md")), b"# x\n").unwrap();
        }
        let meta = root.join(crate::mem::MEM_META_DIR);
        std::fs::create_dir_all(&meta).unwrap();
        let lines: String = ledger_entities
            .iter()
            .map(|e| format!(r#"{{"ts":"2026-08-27T00:00:00Z","entity":"docs--{e}"}}"#) + "\n")
            .collect();
        std::fs::write(meta.join("changes.jsonl"), lines).unwrap();
    }

    /// 04/04, criterion 1: both directions of disagreement are named, and
    /// named apart. A ledger line with no file and a file with no ledger line
    /// are different problems for the reader.
    #[test]
    fn both_directions_of_ledger_divergence_are_named_separately() {
        let tmp = TempDir::new().unwrap();
        seed(tmp.path(), &["alpha", "orphaned-file"], &["alpha", "ghost"]);
        let r = reconcile_ledger(tmp.path()).unwrap();
        assert_eq!(r.ledger_without_file, vec!["ghost".to_string()]);
        assert_eq!(r.file_without_ledger, vec!["orphaned-file".to_string()]);
        assert!(!r.is_clean());
    }

    /// Criterion 2: the check reads. A reconciliation that tidied the ledger
    /// would be fabricating provenance for changes the engine cannot
    /// attribute, so nothing it touches may move.
    #[test]
    fn reconciliation_writes_nothing_at_all() {
        let tmp = TempDir::new().unwrap();
        seed(tmp.path(), &["alpha", "orphaned-file"], &["alpha", "ghost"]);
        let ledger = tmp
            .path()
            .join(crate::mem::MEM_META_DIR)
            .join("changes.jsonl");
        let before_ledger = std::fs::read(&ledger).unwrap();
        let before_files: Vec<_> = std::fs::read_dir(tmp.path())
            .unwrap()
            .filter_map(|e| e.ok())
            .map(|e| (e.path(), e.metadata().map(|m| m.len()).unwrap_or(0)))
            .collect();

        let _ = reconcile_ledger(tmp.path()).unwrap();

        assert_eq!(std::fs::read(&ledger).unwrap(), before_ledger);
        let after_files: Vec<_> = std::fs::read_dir(tmp.path())
            .unwrap()
            .filter_map(|e| e.ok())
            .map(|e| (e.path(), e.metadata().map(|m| m.len()).unwrap_or(0)))
            .collect();
        assert_eq!(after_files.len(), before_files.len());
    }

    /// A mem whose ledger and files agree reports clean, so the check is a
    /// signal rather than a permanent complaint.
    #[test]
    fn an_agreeing_mem_reconciles_clean() {
        let tmp = TempDir::new().unwrap();
        seed(tmp.path(), &["alpha", "beta"], &["alpha", "beta"]);
        assert!(reconcile_ledger(tmp.path()).unwrap().is_clean());
    }

    /// The engine's own sidecar is not entity content: counting
    /// `.memstead/` would report every mem as diverging forever.
    #[test]
    fn the_engine_sidecar_is_not_mistaken_for_an_entity() {
        let tmp = TempDir::new().unwrap();
        seed(tmp.path(), &["alpha"], &["alpha"]);
        std::fs::write(
            tmp.path().join(crate::mem::MEM_META_DIR).join("notes.md"),
            b"not an entity\n",
        )
        .unwrap();
        assert!(reconcile_ledger(tmp.path()).unwrap().is_clean());
    }
}