Skip to main content

meerkat_store/
migrate.rs

1//! Offline migration primitives behind `rkat storage migrate` / `rkat
2//! storage prune` (Phase 6 of the storage unification arc).
3//!
4//! This module is a **reusable library layer** over an arbitrary state
5//! directory — standalone mobkit gateways with no `rkat` CLI on the box run
6//! their migrations through these same primitives. It owns:
7//!
8//! - [`RealmMaintenanceFence`]: the realm-wide exclusive maintenance fence.
9//!   It takes the realm-level write-admission fence
10//!   (`<realm_dir>/realm.mfence`, honored by the JSONL session store and the
11//!   filesystem blob/artifact stores on every write) first, then the
12//!   per-file [`meerkat_sqlite::ExclusiveFence`] on the **full fixed
13//!   inventory** ([`REALM_SQLITE_FILES`], whether or not each database
14//!   exists yet — the fence file is a sibling lock), then the dynamic
15//!   `mobs/*.db` set with a re-enumeration check so a database created
16//!   mid-acquisition cannot escape. Acquisition order is deterministic
17//!   (admission fence first serializes concurrent migrators; per-file
18//!   fences are sorted), waiting up to a deadline for in-flight
19//!   per-operation guards to drain. Acquisition is all-or-nothing: any
20//!   failure releases everything already acquired (RAII).
21//! - The **backup naming discipline**: structural changes rename, never
22//!   delete. [`backup_artifact_name`] produces
23//!   `<original>.pre-<workspace-version>-<unix-ts>[.<purpose>]`; doctor
24//!   lists `*.pre-*` as `backup-artifact` findings and `rkat storage prune`
25//!   owns their lifecycle. External retention tooling can recognize backup
26//!   artifacts by the `.pre-` name segment.
27//! - The shape-stable **report vocabulary** ([`MigrateReport`],
28//!   [`PruneReport`], ...) serialized by `rkat storage migrate --json` /
29//!   `rkat storage prune --json` and reused by downstream orchestrators.
30//! - Read-only helpers the orchestration layer composes: realm-directory
31//!   listing, ledger-version reads, and split-brain divergence computation
32//!   (sessions row-level by id + content digest; other domains at
33//!   file-digest level in v1).
34//!
35//! Orchestration (which store constructors to run, when to fence, what to
36//! archive) lives with the caller — the CLI's `storage migrate` verb for
37//! disk realms — because only the top of the dependency graph can see every
38//! store crate.
39
40use std::collections::BTreeMap;
41use std::fs;
42use std::path::{Path, PathBuf};
43use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
44
45use meerkat_core::REALM_MANIFEST_FILE_NAME;
46use rusqlite::OptionalExtension;
47use serde::{Deserialize, Serialize};
48use sha2::{Digest, Sha256};
49
50use crate::error::StoreError;
51
52/// SQLite database files a realm directory can materialize, relative to the
53/// realm root. This is the same inventory `doctor` sweeps and the file set
54/// the [`RealmMaintenanceFence`] covers — every path in this list is fenced
55/// whether or not the database exists yet (plus the per-mob `mobs/*.db`
56/// databases, which are enumerated dynamically).
57pub const REALM_SQLITE_FILES: &[&str] = &[
58    "sessions.sqlite3",
59    "runtime.sqlite3",
60    "workgraph.sqlite3",
61    "jobs.sqlite3",
62    "memory/memory.sqlite3",
63    "tasks.db",
64    "sessions_jsonl/session_index.sqlite3",
65];
66
67/// Stable finding code: a legacy `<home>/.rkat/sessions` directory exists
68/// (report-only; migrate never moves it).
69pub const FINDING_LEGACY_HOME_SESSIONS_DIR: &str = "legacy-home-sessions-dir";
70
71/// Enumerate every SQLite database file currently materialized under a realm
72/// directory: the fixed [`REALM_SQLITE_FILES`] list plus per-mob `mobs/*.db`
73/// databases. Sorted (deterministic order). Symlinked entries are excluded
74/// (see [`enumerate_realm_sqlite_inventory`] for the reporting variant).
75pub fn enumerate_realm_sqlite_files(realm_dir: &Path) -> Vec<PathBuf> {
76    enumerate_realm_sqlite_inventory(realm_dir).files
77}
78
79/// Realm SQLite inventory with symlinked entries reported instead of
80/// silently followed.
81#[non_exhaustive]
82#[derive(Debug, Default)]
83pub struct RealmSqliteInventory {
84    /// Regular database files, sorted.
85    pub files: Vec<PathBuf>,
86    /// Inventory paths occupied by symlinks, sorted. Never followed:
87    /// digesting or archiving through a link would reach a target outside
88    /// the realm. Callers surface these as per-realm errors/notes.
89    pub symlinks: Vec<PathBuf>,
90}
91
92/// [`enumerate_realm_sqlite_files`] variant that also reports symlinked
93/// inventory entries (checked via `symlink_metadata`, never followed).
94pub fn enumerate_realm_sqlite_inventory(realm_dir: &Path) -> RealmSqliteInventory {
95    let mut inventory = RealmSqliteInventory::default();
96    for path in REALM_SQLITE_FILES
97        .iter()
98        .map(|relative| realm_dir.join(relative))
99    {
100        classify_inventory_path(path, &mut inventory);
101    }
102    if let Ok(entries) = fs::read_dir(realm_dir.join("mobs")) {
103        for entry in entries.filter_map(Result::ok) {
104            let path = entry.path();
105            if path.extension().and_then(|ext| ext.to_str()) == Some("db") {
106                classify_inventory_path(path, &mut inventory);
107            }
108        }
109    }
110    inventory.files.sort();
111    inventory.symlinks.sort();
112    inventory
113}
114
115fn classify_inventory_path(path: PathBuf, inventory: &mut RealmSqliteInventory) {
116    match fs::symlink_metadata(&path) {
117        Ok(metadata) if metadata.file_type().is_symlink() => inventory.symlinks.push(path),
118        Ok(metadata) if metadata.is_file() => inventory.files.push(path),
119        // Absent, wrong-typed, or unprobeable: not a digestible database.
120        _ => {}
121    }
122}
123
124/// The dynamic per-mob `mobs/*.db` databases currently materialized under a
125/// realm directory, sorted.
126fn mob_database_files(realm_dir: &Path) -> Vec<PathBuf> {
127    let mut files = Vec::new();
128    if let Ok(entries) = fs::read_dir(realm_dir.join("mobs")) {
129        for entry in entries.filter_map(Result::ok) {
130            let path = entry.path();
131            if path.extension().and_then(|ext| ext.to_str()) == Some("db") && path.is_file() {
132                files.push(path);
133            }
134        }
135    }
136    files.sort();
137    files
138}
139
140/// File-name stem of the realm-level write-admission fence: guards and the
141/// maintenance fence both target `<realm_dir>/realm`, so the shared lock
142/// file is `<realm_dir>/realm.mfence` (see
143/// [`meerkat_sqlite::fence_lock_path`]).
144///
145/// SQLite stores are quiesced by their per-file fences; the JSONL session
146/// store and the filesystem blob/artifact stores have no database file, so
147/// their write paths take the shared [`meerkat_sqlite::OperationGuard`] on
148/// this target instead. Holding the exclusive side (as
149/// [`RealmMaintenanceFence`] does) therefore excludes every live durable
150/// writer in the realm, not just the SQLite ones.
151pub const REALM_WRITE_ADMISSION_STEM: &str = "realm";
152
153/// The path whose `.mfence` sibling is the realm-level write-admission lock
154/// for `realm_dir`.
155pub fn realm_write_admission_target(realm_dir: &Path) -> PathBuf {
156    realm_dir.join(REALM_WRITE_ADMISSION_STEM)
157}
158
159/// Detect at store construction whether `store_dir` sits inside a realm
160/// directory (its parent holds a realm manifest); if so, return the
161/// write-admission target the store's per-operation guards must use.
162///
163/// Deriving this once at construction keeps the write hot path cheap:
164/// standalone stores (arbitrary directories, tests) get `None` and skip the
165/// guard entirely.
166pub fn store_realm_admission_target(store_dir: &Path) -> Option<PathBuf> {
167    let parent = store_dir.parent()?;
168    parent
169        .join(REALM_MANIFEST_FILE_NAME)
170        .is_file()
171        .then(|| realm_write_admission_target(parent))
172}
173
174/// The realm-wide exclusive maintenance fence.
175///
176/// Holds the realm-level write-admission fence
177/// ([`realm_write_admission_target`], honored per operation by the JSONL
178/// session store and the filesystem blob/artifact stores) plus one
179/// [`meerkat_sqlite::ExclusiveFence`] per SQLite database in the **full
180/// fixed inventory** ([`REALM_SQLITE_FILES`], whether or not the file exists
181/// yet) and per currently-materialized `mobs/*.db` database (re-enumerated
182/// until stable). While held, every foreign process's per-operation guard
183/// fails typed (`MaintenanceFenceHeld`); the holder's own in-process store
184/// operations self-admit (see `meerkat_sqlite::fence`), which is what lets
185/// bulk maintenance reuse production store code paths.
186///
187/// Not covered: a `mobs/*.db` database created by a foreign process *after*
188/// acquisition completes (mob store openers take only their own per-file
189/// guard), and writers that bypass the guard seams entirely.
190///
191/// Acquisition blocks the calling thread (bounded by the deadline); async
192/// callers should wrap it in `spawn_blocking`.
193#[derive(Debug)]
194pub struct RealmMaintenanceFence {
195    admission: meerkat_sqlite::ExclusiveFence,
196    fences: Vec<meerkat_sqlite::ExclusiveFence>,
197    databases: Vec<PathBuf>,
198}
199
200impl RealmMaintenanceFence {
201    /// Fence `realm_dir`, waiting up to `deadline` (total, across all locks)
202    /// for in-flight operations to drain.
203    ///
204    /// The write-admission fence is taken first (serializing concurrent
205    /// migrators outright), then the fixed inventory in sorted order, then
206    /// the dynamic mob databases — so two concurrent migrators acquire in
207    /// the same sequence instead of deadlocking ABBA. Fixed-inventory fence
208    /// files are sibling locks; creating a missing enclosing directory (for
209    /// example `memory/`) is the only mutation acquisition performs. On any
210    /// failure the already-acquired fences are released (RAII) and the typed
211    /// error surfaces ([`StoreError::MaintenanceFenceHeld`] when a foreign
212    /// holder owns a fence past the deadline).
213    pub fn acquire(realm_dir: &Path, deadline: Duration) -> Result<Self, StoreError> {
214        let started = Instant::now();
215        let remaining = |started: &Instant| deadline.saturating_sub(started.elapsed());
216
217        let admission = meerkat_sqlite::ExclusiveFence::acquire(
218            &realm_write_admission_target(realm_dir),
219            remaining(&started),
220        )
221        .map_err(StoreError::from)?;
222
223        // Full fixed inventory, existing or not: a database created during
224        // the maintenance window is already excluded by its sibling fence.
225        let mut databases: Vec<PathBuf> = REALM_SQLITE_FILES
226            .iter()
227            .map(|relative| realm_dir.join(relative))
228            .collect();
229        databases.sort();
230        let mut fences = Vec::with_capacity(databases.len());
231        for database in &databases {
232            if let Some(parent) = database.parent() {
233                fs::create_dir_all(parent)?;
234            }
235            let fence = meerkat_sqlite::ExclusiveFence::acquire(database, remaining(&started))
236                .map_err(|error| {
237                    // Dropping `admission`/`fences` here releases everything.
238                    StoreError::from(error)
239                })?;
240            fences.push(fence);
241        }
242
243        // Dynamic per-mob databases, enumerated under the already-held fixed
244        // fences. Re-enumerate until the set is stable: a database created
245        // between enumeration and fencing must not escape the fence. The
246        // first pass always runs; later growth is bounded by the deadline.
247        let mut first_pass = true;
248        loop {
249            let new: Vec<PathBuf> = mob_database_files(realm_dir)
250                .into_iter()
251                .filter(|database| !databases.contains(database))
252                .collect();
253            if new.is_empty() {
254                break;
255            }
256            if !first_pass && started.elapsed() >= deadline {
257                return Err(StoreError::Internal(format!(
258                    "mob databases kept appearing under '{}' during maintenance-fence \
259                     acquisition; realm is not quiescent",
260                    realm_dir.join("mobs").display()
261                )));
262            }
263            first_pass = false;
264            for database in new {
265                let fence = meerkat_sqlite::ExclusiveFence::acquire(&database, remaining(&started))
266                    .map_err(StoreError::from)?;
267                fences.push(fence);
268                databases.push(database);
269            }
270        }
271        databases.sort();
272
273        Ok(Self {
274            admission,
275            fences,
276            databases,
277        })
278    }
279
280    /// The database files this fence covers (fixed inventory plus mob
281    /// databases), sorted. Fixed-inventory entries are fenced even when the
282    /// database does not exist yet.
283    pub fn fenced_databases(&self) -> &[PathBuf] {
284        &self.databases
285    }
286
287    /// The lock file of the realm-level write-admission fence.
288    pub fn admission_lock_path(&self) -> &Path {
289        self.admission.lock_path()
290    }
291
292    /// Number of held per-file fences (excludes the admission fence).
293    pub fn len(&self) -> usize {
294        self.fences.len()
295    }
296
297    /// True when no per-file fence is held. The full fixed inventory is
298    /// always fenced, so this is never true after a successful `acquire`.
299    pub fn is_empty(&self) -> bool {
300        self.fences.is_empty()
301    }
302}
303
304/// Compose the registered backup-artifact name for `original`:
305/// `<original>.pre-<workspace-version>-<unix-ts>[.<purpose>]`.
306///
307/// The `.pre-` segment is the **registered recognition token**: doctor lists
308/// matching paths as `backup-artifact` findings, `rkat storage prune` owns
309/// their lifecycle, and external retention tooling may rely on it. The
310/// version is the workspace version this binary was built from; the
311/// timestamp is Unix seconds; the optional purpose suffix names why the
312/// artifact exists (for example `split-brain`).
313pub fn backup_artifact_name(original: &str, purpose: &str) -> String {
314    let timestamp = SystemTime::now()
315        .duration_since(UNIX_EPOCH)
316        .map(|elapsed| elapsed.as_secs())
317        .unwrap_or(0);
318    let version = env!("CARGO_PKG_VERSION");
319    if purpose.is_empty() {
320        format!("{original}.pre-{version}-{timestamp}")
321    } else {
322        format!("{original}.pre-{version}-{timestamp}.{purpose}")
323    }
324}
325
326/// True when a file/directory name matches the registered backup-artifact
327/// naming ([`backup_artifact_name`]): `<original>.pre-<version>-<unix-ts>`
328/// with an optional trailing `.<purpose>` segment.
329///
330/// The full suffix shape is validated — a mere `.pre-` substring (for
331/// example `notes.pre-release`) is NOT a registered artifact, so prune can
332/// never sweep unrelated files.
333pub fn is_backup_artifact_name(name: &str) -> bool {
334    let Some(idx) = name.rfind(".pre-") else {
335        return false;
336    };
337    if idx == 0 {
338        // No original name before the marker.
339        return false;
340    }
341    // suffix = <version>-<unix-ts>[.<purpose>]; the version contains no '-'
342    // (plain x.y.z workspace versions), so the first '-' ends it.
343    let suffix = &name[idx + ".pre-".len()..];
344    let Some((version, rest)) = suffix.split_once('-') else {
345        return false;
346    };
347    let version_ok = !version.is_empty()
348        && version.split('.').count() >= 2
349        && version
350            .split('.')
351            .all(|part| !part.is_empty() && part.chars().all(|c| c.is_ascii_digit()));
352    if !version_ok {
353        return false;
354    }
355    let (timestamp, purpose) = match rest.split_once('.') {
356        Some((timestamp, purpose)) => (timestamp, Some(purpose)),
357        None => (rest, None),
358    };
359    let timestamp_ok = !timestamp.is_empty() && timestamp.chars().all(|c| c.is_ascii_digit());
360    let purpose_ok = purpose.is_none_or(|p| {
361        !p.is_empty()
362            && p.chars()
363                .all(|c| c.is_ascii_alphanumeric() || c == '-' || c == '_')
364    });
365    timestamp_ok && purpose_ok
366}
367
368/// True when a file name matches the registered index-quarantine naming
369/// (`*.corrupt-<timestamp>`): the suffix after `.corrupt-` must be all
370/// digits, and something must precede the marker.
371pub fn is_quarantine_artifact_name(name: &str) -> bool {
372    let Some(idx) = name.rfind(".corrupt-") else {
373        return false;
374    };
375    if idx == 0 {
376        return false;
377    }
378    let timestamp = &name[idx + ".corrupt-".len()..];
379    !timestamp.is_empty() && timestamp.chars().all(|c| c.is_ascii_digit())
380}
381
382/// One completed archive rename plus its post-rename warnings.
383///
384/// The rename is the preservation; read-only hardening and parent-directory
385/// durability are best-effort and surface here as warnings instead of
386/// failing an archive that already succeeded.
387#[non_exhaustive]
388#[derive(Debug)]
389pub struct ArchivedPath {
390    /// The archive path the original was renamed to.
391    pub archive: PathBuf,
392    /// Post-rename hardening/durability failures (write-permission strip,
393    /// parent-directory fsync). Report-only.
394    pub warnings: Vec<String>,
395}
396
397/// Rename `path` (file or directory) to its registered backup-artifact name
398/// next to the original, strip write permission so the archive is
399/// read-only, and fsync the parent directory so the rename is
400/// crash-durable. Hardening/durability failures never fail the archive —
401/// they surface in [`ArchivedPath::warnings`].
402pub fn archive_path_read_only_reported(
403    path: &Path,
404    purpose: &str,
405) -> Result<ArchivedPath, StoreError> {
406    let name = path
407        .file_name()
408        .and_then(|name| name.to_str())
409        .ok_or_else(|| {
410            StoreError::Internal(format!(
411                "cannot archive '{}': path has no UTF-8 file name",
412                path.display()
413            ))
414        })?;
415    let archive = path.with_file_name(backup_artifact_name(name, purpose));
416    fs::rename(path, &archive)?;
417    let mut warnings = Vec::new();
418    strip_write_permissions(&archive, &mut warnings);
419    sync_parent_dir_reported(&archive, &mut warnings);
420    Ok(ArchivedPath { archive, warnings })
421}
422
423/// [`archive_path_read_only_reported`] with the hardening/durability
424/// warnings dropped. Prefer the reported variant so partial hardening is
425/// visible in reports.
426pub fn archive_path_read_only(path: &Path, purpose: &str) -> Result<PathBuf, StoreError> {
427    archive_path_read_only_reported(path, purpose).map(|archived| archived.archive)
428}
429
430/// Fsync the parent directory of `path` so a completed rename survives a
431/// crash (unix; directory handles are not fsyncable elsewhere). Failures
432/// are warnings: the rename itself already happened.
433fn sync_parent_dir_reported(path: &Path, warnings: &mut Vec<String>) {
434    #[cfg(unix)]
435    {
436        let Some(parent) = path
437            .parent()
438            .filter(|parent| !parent.as_os_str().is_empty())
439        else {
440            return;
441        };
442        let synced = fs::File::open(parent).and_then(|dir| dir.sync_all());
443        if let Err(error) = synced {
444            warnings.push(format!(
445                "archive rename to '{}' is not crash-durable: fsync of parent directory '{}' \
446                 failed: {error}",
447                path.display(),
448                parent.display()
449            ));
450        }
451    }
452    #[cfg(not(unix))]
453    {
454        let _ = (path, &mut *warnings);
455    }
456}
457
458/// Recursively strip write permission, recording failures as warnings.
459/// Symlinks are skipped entirely: `fs::set_permissions` follows them, and a
460/// link inside an archive must never chmod its external target.
461fn strip_write_permissions(path: &Path, warnings: &mut Vec<String>) {
462    let metadata = match fs::symlink_metadata(path) {
463        Ok(metadata) => metadata,
464        Err(error) => {
465            warnings.push(format!(
466                "archive hardening skipped for '{}': {error}",
467                path.display()
468            ));
469            return;
470        }
471    };
472    if metadata.file_type().is_symlink() {
473        return;
474    }
475    if metadata.is_dir() {
476        match fs::read_dir(path) {
477            Ok(entries) => {
478                for entry in entries.filter_map(Result::ok) {
479                    strip_write_permissions(&entry.path(), warnings);
480                }
481            }
482            Err(error) => warnings.push(format!(
483                "archive hardening could not list '{}': {error}",
484                path.display()
485            )),
486        }
487    }
488    let mut permissions = metadata.permissions();
489    permissions.set_readonly(true);
490    if let Err(error) = fs::set_permissions(path, permissions) {
491        warnings.push(format!(
492            "archive '{}' may remain writable: {error}",
493            path.display()
494        ));
495    }
496}
497
498fn restore_write_permissions_best_effort(path: &Path) {
499    let Ok(metadata) = fs::symlink_metadata(path) else {
500        return;
501    };
502    if metadata.file_type().is_symlink() {
503        // `fs::set_permissions` follows symlinks; never chmod through one.
504        return;
505    }
506    #[allow(clippy::permissions_set_readonly_false)] // deliberate: prune restores
507    // write permission on registered archives it is about to delete.
508    {
509        let mut permissions = metadata.permissions();
510        permissions.set_readonly(false);
511        let _ = fs::set_permissions(path, permissions);
512    }
513    if metadata.is_dir()
514        && let Ok(entries) = fs::read_dir(path)
515    {
516        for entry in entries.filter_map(Result::ok) {
517            restore_write_permissions_best_effort(&entry.path());
518        }
519    }
520}
521
522// ─────────────────────────────────────────────────────────────────────────
523// Shape-stable report vocabulary (serde; `#[non_exhaustive]` + defaults,
524// like the `meerkat_core::storage_diagnostics` types).
525// ─────────────────────────────────────────────────────────────────────────
526
527/// Dry-run vs apply.
528#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
529#[serde(rename_all = "snake_case")]
530pub enum MigrateMode {
531    /// Read-only: report what would change; touch nothing.
532    #[default]
533    DryRun,
534    /// Fenced structural migration.
535    Apply,
536}
537
538/// The full `storage migrate` report.
539#[non_exhaustive]
540#[derive(Debug, Clone, Default, Serialize, Deserialize)]
541pub struct MigrateReport {
542    /// Dry-run or apply.
543    #[serde(default)]
544    pub mode: MigrateMode,
545    /// State roots swept (explicit roots, or the resolver's candidates).
546    #[serde(default)]
547    pub swept_roots: Vec<PathBuf>,
548    /// Per-realm migration outcomes (cases 1 and 2).
549    #[serde(default)]
550    pub realms: Vec<RealmMigrateReport>,
551    /// Split-brain twins and their resolution (case 3).
552    #[serde(default)]
553    pub split_brain: Vec<SplitBrainReport>,
554    /// Deprecated-leftover findings (case 5; report-only) plus sweep
555    /// context, reusing the doctor finding vocabulary.
556    #[serde(default)]
557    pub findings: Vec<meerkat_core::StorageFinding>,
558    /// Fail-closed refusals and hard failures. Non-empty ⇒ nonzero exit.
559    #[serde(default)]
560    pub errors: Vec<String>,
561}
562
563impl MigrateReport {
564    /// New empty report for one run.
565    pub fn new(mode: MigrateMode, swept_roots: Vec<PathBuf>) -> Self {
566        Self {
567            mode,
568            swept_roots,
569            ..Self::default()
570        }
571    }
572
573    /// True when the run must exit nonzero (refusals, fence failures,
574    /// per-realm errors).
575    pub fn has_errors(&self) -> bool {
576        !self.errors.is_empty() || self.realms.iter().any(|realm| !realm.errors.is_empty())
577    }
578}
579
580/// Migration outcome for one realm materialization.
581#[non_exhaustive]
582#[derive(Debug, Clone, Serialize, Deserialize)]
583pub struct RealmMigrateReport {
584    /// Realm id (manifest identity; directory name when unreadable).
585    pub realm: String,
586    /// The realm directory (case 2 — state-root adoption is report-only:
587    /// the realm is used where it lies).
588    pub root: PathBuf,
589    /// Backend pinned in the manifest, when readable.
590    #[serde(default, skip_serializing_if = "Option::is_none")]
591    pub backend: Option<String>,
592    /// Ledger baseline entries per database × domain (case 1).
593    #[serde(default)]
594    pub ledger: Vec<LedgerBaselineEntry>,
595    /// Human-readable per-realm notes (skips, report-only carve-outs).
596    #[serde(default)]
597    pub notes: Vec<String>,
598    /// Per-realm failures (fence not acquirable, store open failures).
599    #[serde(default)]
600    pub errors: Vec<String>,
601}
602
603impl RealmMigrateReport {
604    /// New empty per-realm report.
605    pub fn new(realm: impl Into<String>, root: PathBuf) -> Self {
606        Self {
607            realm: realm.into(),
608            root,
609            backend: None,
610            ledger: Vec::new(),
611            notes: Vec::new(),
612            errors: Vec::new(),
613        }
614    }
615}
616
617/// One database × ledger-domain baseline row.
618#[non_exhaustive]
619#[derive(Debug, Clone, Serialize, Deserialize)]
620pub struct LedgerBaselineEntry {
621    /// Database file.
622    pub database: PathBuf,
623    /// Ledger domain (`session-store`, `runtime-store`, ...).
624    pub domain: String,
625    /// Version before the run (`None` = no ledger row).
626    #[serde(default, skip_serializing_if = "Option::is_none")]
627    pub before: Option<i64>,
628    /// Version after the run (`None` in dry-run).
629    #[serde(default, skip_serializing_if = "Option::is_none")]
630    pub after: Option<i64>,
631    /// What happened (or would happen).
632    pub action: LedgerBaselineAction,
633}
634
635impl LedgerBaselineEntry {
636    /// New entry with no before/after versions (set the public fields).
637    pub fn new(database: PathBuf, domain: impl Into<String>, action: LedgerBaselineAction) -> Self {
638        Self {
639            database,
640            domain: domain.into(),
641            before: None,
642            after: None,
643            action,
644        }
645    }
646}
647
648/// Disposition of one ledger baseline row.
649#[non_exhaustive]
650#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
651#[serde(rename_all = "kebab-case")]
652pub enum LedgerBaselineAction {
653    /// Dry-run: no ledger row. Apply will initialize only if the domain owns
654    /// zero catalog objects; an unledgered owned object is refused rather
655    /// than inferred or stamped.
656    MissingRow,
657    /// Dry-run: a ledger row exists. Apply still verifies that it is current
658    /// or an exact supported released predecessor before migrating.
659    Recorded,
660    /// Apply: the ledger row was created or advanced.
661    Stamped,
662    /// Apply: already at the current version; no-op.
663    AlreadyCurrent,
664    /// Not migrated by this verb in v1 (per-mob databases); the owning
665    /// store verifies and, when eligible, migrates the file on its next open.
666    ReportOnly,
667}
668
669/// Case 3: one split-brain realm (same realm id under 2+ swept roots).
670#[non_exhaustive]
671#[derive(Debug, Clone, Serialize, Deserialize)]
672pub struct SplitBrainReport {
673    /// Realm id.
674    pub realm: String,
675    /// Every realm directory materializing this id.
676    pub locations: Vec<PathBuf>,
677    /// Sessions identical across every copy (count only; not enumerated).
678    #[serde(default)]
679    pub sessions_equal: usize,
680    /// Divergent / single-copy sessions (row-level compare by id + content
681    /// digest over the session tables).
682    #[serde(default)]
683    pub sessions: Vec<SessionDivergenceEntry>,
684    /// Per-file content-digest comparison over every authoritative store the
685    /// realm materializes: the SQLite databases (runtime / workgraph /
686    /// schedule / memory / tasks / index / mob), the canonical
687    /// `sessions_jsonl/*.jsonl` session files, and the `blobs/` and
688    /// `artifacts/` trees.
689    #[serde(default)]
690    pub files: Vec<FileDivergenceEntry>,
691    /// How the twin was (or was not) resolved.
692    pub resolution: SplitBrainResolution,
693    /// Divergence-computation failures (report-only; archiving is
694    /// non-destructive either way, but see
695    /// [`SplitBrainReport::comparison_is_conclusive`]).
696    #[serde(default)]
697    pub errors: Vec<String>,
698}
699
700impl SplitBrainReport {
701    /// New unresolved (fail-closed) split-brain report for one realm.
702    pub fn new(realm: impl Into<String>, locations: Vec<PathBuf>) -> Self {
703        Self {
704            realm: realm.into(),
705            locations,
706            sessions_equal: 0,
707            sessions: Vec::new(),
708            files: Vec::new(),
709            resolution: SplitBrainResolution::Refused {
710                reason: "split-brain unresolved: rerun with `--apply --adopt-root <path>` to \
711                         adopt one root and archive the other copies read-only"
712                    .to_string(),
713            },
714            errors: Vec::new(),
715        }
716    }
717
718    /// True when every authoritative entry on every copy was readable and
719    /// conclusively classified: no comparison errors and no
720    /// [`DivergenceStatus::Unknown`] entries.
721    ///
722    /// An adopt-and-archive decision must not rest on an inconclusive
723    /// report — an unreadable side may hold the only copy of content the
724    /// report could not account for. (Divergent entries do not make a
725    /// report inconclusive; archiving preserves divergent content.)
726    pub fn comparison_is_conclusive(&self) -> bool {
727        self.errors.is_empty()
728            && self
729                .sessions
730                .iter()
731                .all(|entry| entry.status != DivergenceStatus::Unknown)
732            && self
733                .files
734                .iter()
735                .all(|entry| entry.status != DivergenceStatus::Unknown)
736    }
737}
738
739/// One non-equal session across split-brain copies.
740#[non_exhaustive]
741#[derive(Debug, Clone, Serialize, Deserialize)]
742pub struct SessionDivergenceEntry {
743    /// Session id.
744    pub session_id: String,
745    /// Divergence classification.
746    pub status: DivergenceStatus,
747}
748
749/// One database file compared across split-brain copies.
750#[non_exhaustive]
751#[derive(Debug, Clone, Serialize, Deserialize)]
752pub struct FileDivergenceEntry {
753    /// Path relative to the realm directory.
754    pub file: String,
755    /// Divergence classification.
756    pub status: DivergenceStatus,
757}
758
759/// Divergence classification across split-brain copies.
760#[non_exhaustive]
761#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
762#[serde(rename_all = "snake_case", tag = "kind")]
763pub enum DivergenceStatus {
764    /// Present in every copy with identical content.
765    Equal,
766    /// Present in more than one copy with differing content (or missing
767    /// from some copies).
768    Divergent,
769    /// Present in exactly one copy.
770    OnlyIn {
771        /// The realm directory holding the single copy.
772        location: PathBuf,
773    },
774    /// A read failure on some copy poisoned this comparison: the entry may
775    /// be equal, divergent, or unique, and no archive decision may rest on
776    /// it. The failure is recorded in [`SplitBrainReport::errors`].
777    Unknown,
778}
779
780/// Case 3 resolution.
781#[non_exhaustive]
782#[derive(Debug, Clone, Serialize, Deserialize)]
783#[serde(rename_all = "snake_case", tag = "kind")]
784pub enum SplitBrainResolution {
785    /// Fail-closed refusal: nothing moved. Pass `--apply --adopt-root
786    /// <path>` to adopt one root and archive the others read-only.
787    Refused {
788        /// Why the run refused.
789        reason: String,
790    },
791    /// One copy adopted where it lies; every other copy archived read-only
792    /// under the registered backup naming. No synthesis, no merging —
793    /// divergent content is preserved in the archives.
794    Archived {
795        /// The adopted realm directory (left untouched).
796        adopted: PathBuf,
797        /// Archive paths of the non-adopted copies.
798        archived: Vec<PathBuf>,
799    },
800    /// Archiving stopped partway: the archives listed succeeded before the
801    /// failure and stay where they are (renames are the preservation;
802    /// nothing is rolled back). Rerun after fixing the fault. A partial
803    /// archive must never be reported as "nothing moved".
804    ArchiveFailed {
805        /// The adopted realm directory (left untouched).
806        adopted: PathBuf,
807        /// Archive paths that succeeded before the failure.
808        archived: Vec<PathBuf>,
809        /// The failure that stopped the run.
810        reason: String,
811    },
812}
813
814/// The full `storage prune` report.
815#[non_exhaustive]
816#[derive(Debug, Clone, Default, Serialize, Deserialize)]
817pub struct PruneReport {
818    /// Dry-run or apply.
819    #[serde(default)]
820    pub mode: MigrateMode,
821    /// State roots swept.
822    #[serde(default)]
823    pub swept_roots: Vec<PathBuf>,
824    /// Age threshold in days (artifacts at least this old are deleted on
825    /// apply; `0` = all).
826    #[serde(default)]
827    pub older_than_days: u64,
828    /// Registered artifacts found, with dispositions.
829    #[serde(default)]
830    pub artifacts: Vec<PruneArtifact>,
831    /// Failures (delete errors). Non-empty ⇒ nonzero exit.
832    #[serde(default)]
833    pub errors: Vec<String>,
834}
835
836impl PruneReport {
837    /// New empty report for one run.
838    pub fn new(mode: MigrateMode, swept_roots: Vec<PathBuf>, older_than_days: u64) -> Self {
839        Self {
840            mode,
841            swept_roots,
842            older_than_days,
843            ..Self::default()
844        }
845    }
846}
847
848/// One registered maintenance artifact.
849#[non_exhaustive]
850#[derive(Debug, Clone, Serialize, Deserialize)]
851pub struct PruneArtifact {
852    /// Artifact path (file or directory).
853    pub path: PathBuf,
854    /// Which registered naming pattern matched.
855    pub kind: PruneArtifactKind,
856    /// Total size in bytes (recursive for directories).
857    #[serde(default)]
858    pub bytes: u64,
859    /// Age in whole days (from mtime).
860    #[serde(default)]
861    pub age_days: u64,
862    /// Disposition.
863    pub action: PruneAction,
864}
865
866/// Registered artifact classes prune may touch. Anything else is never
867/// touched.
868#[non_exhaustive]
869#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
870#[serde(rename_all = "kebab-case")]
871pub enum PruneArtifactKind {
872    /// `*.pre-<version>-<timestamp>[.<purpose>]` migration backup.
873    BackupArtifact,
874    /// `*.corrupt-<timestamp>` quarantined index.
875    QuarantinedIndex,
876}
877
878/// Prune disposition for one artifact.
879#[non_exhaustive]
880#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
881#[serde(rename_all = "kebab-case")]
882pub enum PruneAction {
883    /// Dry-run: old enough; `--apply` would delete it.
884    WouldDelete,
885    /// Apply: deleted.
886    Deleted,
887    /// Younger than the threshold; kept.
888    Kept,
889    /// Apply: deletion failed (reason in `PruneReport::errors`).
890    DeleteFailed,
891}
892
893// ─────────────────────────────────────────────────────────────────────────
894// Read-only helpers for the orchestration layer.
895// ─────────────────────────────────────────────────────────────────────────
896
897/// One materialized realm directory under a state root.
898#[non_exhaustive]
899#[derive(Debug, Clone)]
900pub struct RealmDirEntry {
901    /// Realm id (manifest identity; sanitized directory name when the
902    /// manifest is unreadable).
903    pub realm_id: String,
904    /// Backend string from the manifest, when readable.
905    pub backend: Option<String>,
906    /// The state root swept.
907    pub state_root: PathBuf,
908    /// The realm directory.
909    pub dir: PathBuf,
910    /// False when the manifest failed to read/parse.
911    pub manifest_readable: bool,
912}
913
914/// Lenient realm-directory listing under one state root: directories with a
915/// `realm_manifest.json`, read as raw JSON (no backend validation), skipping
916/// registered backup artifacts (`*.pre-*`). An absent root lists empty.
917pub fn list_realm_dirs(state_root: &Path) -> Vec<RealmDirEntry> {
918    let Ok(entries) = fs::read_dir(state_root) else {
919        return Vec::new();
920    };
921    let mut dirs: Vec<PathBuf> = entries
922        .filter_map(Result::ok)
923        .map(|entry| entry.path())
924        .filter(|path| path.is_dir())
925        .collect();
926    dirs.sort();
927
928    let mut realms = Vec::new();
929    for dir in dirs {
930        let name = dir
931            .file_name()
932            .map(|name| name.to_string_lossy().into_owned())
933            .unwrap_or_default();
934        if is_backup_artifact_name(&name) {
935            continue; // archived realm copy, not a live realm
936        }
937        let manifest_path = dir.join(REALM_MANIFEST_FILE_NAME);
938        if !manifest_path.is_file() {
939            continue;
940        }
941        let parsed = fs::read(&manifest_path)
942            .ok()
943            .and_then(|bytes| serde_json::from_slice::<serde_json::Value>(&bytes).ok())
944            .and_then(|value| {
945                let realm_id = value.get("realm_id")?.as_str()?.to_string();
946                let backend = value
947                    .get("backend")
948                    .and_then(|backend| backend.as_str())
949                    .map(ToString::to_string);
950                Some((realm_id, backend))
951            });
952        let (realm_id, backend, manifest_readable) = match parsed {
953            Some((realm_id, backend)) => (realm_id, backend, true),
954            None => (name, None, false),
955        };
956        realms.push(RealmDirEntry {
957            realm_id,
958            backend,
959            state_root: state_root.to_path_buf(),
960            dir,
961            manifest_readable,
962        });
963    }
964    realms
965}
966
967/// Read the schema-ledger rows of one database, read-only. `Ok(None)` means
968/// no `meerkat_schema` table; it does not certify a fresh domain.
969pub fn read_domain_versions(db_path: &Path) -> Result<Option<Vec<(String, i64)>>, StoreError> {
970    // No schema preflight: this read-only seam is how future versions get
971    // reported, so it must open files it would otherwise refuse.
972    let conn = meerkat_sqlite::open(db_path, meerkat_sqlite::ConnectionProfile::ReadOnly)
973        .map_err(StoreError::from)?;
974    let table_exists = conn
975        .query_row(
976            "SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = 'meerkat_schema'",
977            [],
978            |_| Ok(()),
979        )
980        .optional()?;
981    if table_exists.is_none() {
982        return Ok(None);
983    }
984    let mut statement =
985        conn.prepare("SELECT domain, version FROM meerkat_schema ORDER BY domain")?;
986    let rows = statement
987        .query_map([], |row| {
988            Ok((row.get::<_, String>(0)?, row.get::<_, i64>(1)?))
989        })?
990        .collect::<Result<Vec<_>, _>>()?;
991    Ok(Some(rows))
992}
993
994/// Highest ledger version this binary supports for domains whose owning
995/// store crates are visible from `meerkat-store`. Domains owned by crates
996/// above this one in the dependency order (runtime-store, workgraph,
997/// memory, mob, tools-tasks) return `None` and are reported without
998/// judgment.
999pub fn supported_domain_version(domain: &str) -> Option<i64> {
1000    match domain {
1001        #[cfg(feature = "sqlite")]
1002        "session-store" => Some(crate::sqlite_store::SESSION_STORE_DOMAIN.supported_version()),
1003        #[cfg(feature = "sqlite")]
1004        "schedule-store" => {
1005            Some(crate::schedule_sqlite_store::SCHEDULE_STORE_DOMAIN.supported_version())
1006        }
1007        "jsonl-index" => Some(crate::index::JSONL_INDEX_DOMAIN.supported_version()),
1008        _ => None,
1009    }
1010}
1011
1012/// One database × domain ledger reading (`None` = expected domain with no
1013/// ledger row).
1014#[non_exhaustive]
1015#[derive(Debug, Clone)]
1016pub struct LedgerDomainReading {
1017    /// Database file.
1018    pub database: PathBuf,
1019    /// Ledger domain.
1020    pub domain: String,
1021    /// Recorded version, when a row exists.
1022    pub version: Option<i64>,
1023}
1024
1025/// One domain recorded at a version newer than this binary supports.
1026#[non_exhaustive]
1027#[derive(Debug, Clone)]
1028pub struct FutureDomainVersion {
1029    /// Database file.
1030    pub database: PathBuf,
1031    /// Ledger domain.
1032    pub domain: String,
1033    /// Version recorded in the file.
1034    pub found: i64,
1035    /// Highest version this binary supports for the domain.
1036    pub supported: i64,
1037}
1038
1039/// Read-only ledger baseline of one realm's inventoried databases, with
1040/// failures surfaced typed.
1041#[non_exhaustive]
1042#[derive(Debug, Default)]
1043pub struct RealmLedgerBaseline {
1044    /// Readable ledger rows plus expected-domain gaps.
1045    pub rows: Vec<LedgerDomainReading>,
1046    /// Domains recorded at a future version — dry-run must report these as
1047    /// refusals exactly as `--apply`'s guarded constructors would refuse
1048    /// them ([`StoreError::SchemaFromTheFuture`]).
1049    pub future: Vec<FutureDomainVersion>,
1050    /// Ledger read failures. A corrupt/unreadable database is a per-realm
1051    /// error, never a missing ledger.
1052    pub errors: Vec<String>,
1053}
1054
1055/// Read the ledger baseline of every inventoried database currently on disk
1056/// under `realm_dir` (the doctor's file × domain matrix), read-only.
1057///
1058/// Unlike a bare [`read_domain_versions`] sweep, read failures and future
1059/// versions are first-class outcomes: dry-run reporting built on this can
1060/// never launder a corrupt database into "no ledger" or a future version
1061/// into "safely recorded".
1062pub fn read_realm_ledger_baseline(realm_dir: &Path) -> RealmLedgerBaseline {
1063    let mut baseline = RealmLedgerBaseline::default();
1064    for (relative, expected_domains) in crate::doctor::REALM_DATABASE_FILES {
1065        let db_path = realm_dir.join(relative);
1066        if !db_path.is_file() {
1067            continue;
1068        }
1069        let rows = match read_domain_versions(&db_path) {
1070            Ok(rows) => rows.unwrap_or_default(),
1071            Err(error) => {
1072                baseline.errors.push(format!(
1073                    "ledger unreadable for {}: {error}",
1074                    db_path.display()
1075                ));
1076                continue;
1077            }
1078        };
1079        for (domain, version) in &rows {
1080            if let Some(supported) = supported_domain_version(domain)
1081                && *version > supported
1082            {
1083                baseline.future.push(FutureDomainVersion {
1084                    database: db_path.clone(),
1085                    domain: domain.clone(),
1086                    found: *version,
1087                    supported,
1088                });
1089            }
1090            baseline.rows.push(LedgerDomainReading {
1091                database: db_path.clone(),
1092                domain: domain.clone(),
1093                version: Some(*version),
1094            });
1095        }
1096        for expected in *expected_domains {
1097            if !rows.iter().any(|(domain, _)| domain == expected) {
1098                baseline.rows.push(LedgerDomainReading {
1099                    database: db_path.clone(),
1100                    domain: (*expected).to_string(),
1101                    version: None,
1102                });
1103            }
1104        }
1105    }
1106    baseline
1107}
1108
1109fn table_exists(conn: &rusqlite::Connection, table: &str) -> Result<bool, rusqlite::Error> {
1110    Ok(conn
1111        .query_row(
1112            "SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = ?1",
1113            [table],
1114            |_| Ok(()),
1115        )
1116        .optional()?
1117        .is_some())
1118}
1119
1120// ─────────────────────────────────────────────────────────────────────────
1121// Split-brain divergence (case 3).
1122// ─────────────────────────────────────────────────────────────────────────
1123
1124type SessionDigests = BTreeMap<String, [u8; 32]>;
1125
1126/// Per-session content digest over every authoritative session-domain table,
1127/// fed in a fixed table order with ordered rows so digests are comparable
1128/// across copies. This includes strand links, component events, and both
1129/// the immutable and mutable halves of the authenticated HeadCanonical
1130/// metadata state graph; a copy missing a cell/state/delta/lineage row or
1131/// pointing an owner at a different head token and state is therefore
1132/// split-brain even when its compact `session_heads` row matches.
1133fn session_digests(db_path: &Path) -> Result<SessionDigests, StoreError> {
1134    // No schema preflight: divergence comparison is read-only and must not
1135    // refuse a copy merely because a newer binary migrated it.
1136    let conn = meerkat_sqlite::open(db_path, meerkat_sqlite::ConnectionProfile::ReadOnly)
1137        .map_err(StoreError::from)?;
1138    let mut hashers: BTreeMap<String, Sha256> = BTreeMap::new();
1139    let mut feed = |session_id: String, tag: &str, chunks: &[&[u8]]| {
1140        let hasher = hashers.entry(session_id).or_default();
1141        hasher.update(tag.as_bytes());
1142        hasher.update([0u8]);
1143        for chunk in chunks {
1144            hasher.update(chunk);
1145            hasher.update([0u8]);
1146        }
1147    };
1148
1149    // `JsonColumnBytes` tolerates TEXT-or-BLOB storage for the JSON columns.
1150    if table_exists(&conn, "sessions")? {
1151        let mut statement = conn.prepare(
1152            "SELECT session_id, metadata_json, session_json FROM sessions ORDER BY session_id",
1153        )?;
1154        let mut rows = statement.query([])?;
1155        while let Some(row) = rows.next()? {
1156            let session_id: String = row.get(0)?;
1157            let metadata = row
1158                .get::<_, meerkat_sqlite::JsonColumnBytes>(1)?
1159                .into_bytes();
1160            let document = row
1161                .get::<_, meerkat_sqlite::JsonColumnBytes>(2)?
1162                .into_bytes();
1163            feed(session_id, "sessions", &[&metadata, &document]);
1164        }
1165    }
1166    if table_exists(&conn, "session_heads")? {
1167        let mut statement = conn.prepare(
1168            "SELECT session_id, metadata_json, head_json FROM session_heads ORDER BY session_id",
1169        )?;
1170        let mut rows = statement.query([])?;
1171        while let Some(row) = rows.next()? {
1172            let session_id: String = row.get(0)?;
1173            let metadata = row
1174                .get::<_, meerkat_sqlite::JsonColumnBytes>(1)?
1175                .into_bytes();
1176            let head = row
1177                .get::<_, meerkat_sqlite::JsonColumnBytes>(2)?
1178                .into_bytes();
1179            feed(session_id, "head", &[&metadata, &head]);
1180        }
1181    }
1182    if table_exists(&conn, "session_strand_messages")? {
1183        let mut statement = conn.prepare(
1184            "SELECT session_id, strand, seq, message_json FROM session_strand_messages \
1185             ORDER BY session_id, strand, seq",
1186        )?;
1187        let mut rows = statement.query([])?;
1188        while let Some(row) = rows.next()? {
1189            let session_id: String = row.get(0)?;
1190            let strand: String = row.get(1)?;
1191            let seq: i64 = row.get(2)?;
1192            let message = row
1193                .get::<_, meerkat_sqlite::JsonColumnBytes>(3)?
1194                .into_bytes();
1195            feed(
1196                session_id,
1197                "message",
1198                &[strand.as_bytes(), seq.to_be_bytes().as_slice(), &message],
1199            );
1200        }
1201    }
1202    if table_exists(&conn, "session_strand_links")? {
1203        let mut statement = conn.prepare(
1204            "SELECT session_id, strand, successor, strand_len, splice_start, splice_end, \
1205             successor_end FROM session_strand_links ORDER BY session_id, strand",
1206        )?;
1207        let mut rows = statement.query([])?;
1208        while let Some(row) = rows.next()? {
1209            let session_id: String = row.get(0)?;
1210            let strand: String = row.get(1)?;
1211            let successor: String = row.get(2)?;
1212            let strand_len: i64 = row.get(3)?;
1213            let splice_start: i64 = row.get(4)?;
1214            let splice_end: i64 = row.get(5)?;
1215            let successor_end: i64 = row.get(6)?;
1216            feed(
1217                session_id,
1218                "strand-link",
1219                &[
1220                    strand.as_bytes(),
1221                    successor.as_bytes(),
1222                    strand_len.to_be_bytes().as_slice(),
1223                    splice_start.to_be_bytes().as_slice(),
1224                    splice_end.to_be_bytes().as_slice(),
1225                    successor_end.to_be_bytes().as_slice(),
1226                ],
1227            );
1228        }
1229    }
1230    if table_exists(&conn, "session_rewrites")? {
1231        let mut statement = conn.prepare(
1232            "SELECT session_id, rewrite_idx, commit_json FROM session_rewrites \
1233             ORDER BY session_id, rewrite_idx",
1234        )?;
1235        let mut rows = statement.query([])?;
1236        while let Some(row) = rows.next()? {
1237            let session_id: String = row.get(0)?;
1238            let index: i64 = row.get(1)?;
1239            let commit = row
1240                .get::<_, meerkat_sqlite::JsonColumnBytes>(2)?
1241                .into_bytes();
1242            feed(
1243                session_id,
1244                "rewrite",
1245                &[index.to_be_bytes().as_slice(), &commit],
1246            );
1247        }
1248    }
1249    if table_exists(&conn, "session_component_events")? {
1250        let mut statement = conn.prepare(
1251            "SELECT session_id, component, seq, event_json, event_digest \
1252             FROM session_component_events ORDER BY session_id, component, seq",
1253        )?;
1254        let mut rows = statement.query([])?;
1255        while let Some(row) = rows.next()? {
1256            let session_id: String = row.get(0)?;
1257            let component: String = row.get(1)?;
1258            let seq: i64 = row.get(2)?;
1259            let event = row
1260                .get::<_, meerkat_sqlite::JsonColumnBytes>(3)?
1261                .into_bytes();
1262            let event_digest: String = row.get(4)?;
1263            feed(
1264                session_id,
1265                "component-event",
1266                &[
1267                    component.as_bytes(),
1268                    seq.to_be_bytes().as_slice(),
1269                    &event,
1270                    event_digest.as_bytes(),
1271                ],
1272            );
1273        }
1274    }
1275    if table_exists(&conn, "session_head_metadata_cells")? {
1276        let mut statement = conn.prepare(
1277            "SELECT session_id, metadata_key, key_route, exact_value_digest, \
1278             metadata_json, created_at_ms \
1279             FROM session_head_metadata_cells \
1280             ORDER BY session_id, metadata_key, exact_value_digest",
1281        )?;
1282        let mut rows = statement.query([])?;
1283        while let Some(row) = rows.next()? {
1284            let session_id: String = row.get(0)?;
1285            let metadata_key: String = row.get(1)?;
1286            let key_route: Vec<u8> = row.get(2)?;
1287            let exact_value_digest: String = row.get(3)?;
1288            let metadata = row
1289                .get::<_, meerkat_sqlite::JsonColumnBytes>(4)?
1290                .into_bytes();
1291            let created_at_ms: i64 = row.get(5)?;
1292            feed(
1293                session_id,
1294                "head-metadata-cell",
1295                &[
1296                    metadata_key.as_bytes(),
1297                    &key_route,
1298                    exact_value_digest.as_bytes(),
1299                    &metadata,
1300                    created_at_ms.to_be_bytes().as_slice(),
1301                ],
1302            );
1303        }
1304    }
1305    if table_exists(&conn, "session_head_metadata_current")? {
1306        let mut statement = conn.prepare(
1307            "SELECT session_id, metadata_key, key_route, exact_value_digest \
1308             FROM session_head_metadata_current ORDER BY session_id, metadata_key",
1309        )?;
1310        let mut rows = statement.query([])?;
1311        while let Some(row) = rows.next()? {
1312            let session_id: String = row.get(0)?;
1313            let metadata_key: String = row.get(1)?;
1314            let key_route: Vec<u8> = row.get(2)?;
1315            let exact_value_digest: String = row.get(3)?;
1316            feed(
1317                session_id,
1318                "head-metadata-current",
1319                &[
1320                    metadata_key.as_bytes(),
1321                    &key_route,
1322                    exact_value_digest.as_bytes(),
1323                ],
1324            );
1325        }
1326    }
1327    if table_exists(&conn, "session_head_metadata_states")? {
1328        let mut statement = conn.prepare(
1329            "SELECT session_id, state_id, predecessor_state_id, identity_json, \
1330             transition_id, created_at_ms FROM session_head_metadata_states \
1331             ORDER BY session_id, state_id",
1332        )?;
1333        let mut rows = statement.query([])?;
1334        while let Some(row) = rows.next()? {
1335            let session_id: String = row.get(0)?;
1336            let state_id: String = row.get(1)?;
1337            let predecessor_state_id: Option<String> = row.get(2)?;
1338            let identity = row
1339                .get::<_, meerkat_sqlite::JsonColumnBytes>(3)?
1340                .into_bytes();
1341            let transition_id: String = row.get(4)?;
1342            let created_at_ms: i64 = row.get(5)?;
1343            let predecessor_present = [u8::from(predecessor_state_id.is_some())];
1344            feed(
1345                session_id,
1346                "head-metadata-state",
1347                &[
1348                    state_id.as_bytes(),
1349                    &predecessor_present,
1350                    predecessor_state_id
1351                        .as_deref()
1352                        .unwrap_or_default()
1353                        .as_bytes(),
1354                    &identity,
1355                    transition_id.as_bytes(),
1356                    created_at_ms.to_be_bytes().as_slice(),
1357                ],
1358            );
1359        }
1360    }
1361    if table_exists(&conn, "session_head_metadata_state_deltas")? {
1362        let mut statement = conn.prepare(
1363            "SELECT session_id, state_id, ordinal, metadata_key, key_route, \
1364             predecessor_exact_value_digest, successor_exact_value_digest \
1365             FROM session_head_metadata_state_deltas \
1366             ORDER BY session_id, state_id, ordinal",
1367        )?;
1368        let mut rows = statement.query([])?;
1369        while let Some(row) = rows.next()? {
1370            let session_id: String = row.get(0)?;
1371            let state_id: String = row.get(1)?;
1372            let ordinal: i64 = row.get(2)?;
1373            let metadata_key: String = row.get(3)?;
1374            let key_route: Vec<u8> = row.get(4)?;
1375            let predecessor_exact_value_digest: Option<String> = row.get(5)?;
1376            let successor_exact_value_digest: Option<String> = row.get(6)?;
1377            let predecessor_present = [u8::from(predecessor_exact_value_digest.is_some())];
1378            let successor_present = [u8::from(successor_exact_value_digest.is_some())];
1379            feed(
1380                session_id,
1381                "head-metadata-state-delta",
1382                &[
1383                    state_id.as_bytes(),
1384                    ordinal.to_be_bytes().as_slice(),
1385                    metadata_key.as_bytes(),
1386                    &key_route,
1387                    &predecessor_present,
1388                    predecessor_exact_value_digest
1389                        .as_deref()
1390                        .unwrap_or_default()
1391                        .as_bytes(),
1392                    &successor_present,
1393                    successor_exact_value_digest
1394                        .as_deref()
1395                        .unwrap_or_default()
1396                        .as_bytes(),
1397                ],
1398            );
1399        }
1400    }
1401    if table_exists(&conn, "session_head_metadata_refs")? {
1402        let mut statement = conn.prepare(
1403            "SELECT session_id, owner, head_cas_token, state_id \
1404             FROM session_head_metadata_refs ORDER BY session_id, owner",
1405        )?;
1406        let mut rows = statement.query([])?;
1407        while let Some(row) = rows.next()? {
1408            let session_id: String = row.get(0)?;
1409            let owner: String = row.get(1)?;
1410            let head_cas_token: String = row.get(2)?;
1411            let state_id: String = row.get(3)?;
1412            feed(
1413                session_id,
1414                "head-metadata-ref",
1415                &[
1416                    owner.as_bytes(),
1417                    head_cas_token.as_bytes(),
1418                    state_id.as_bytes(),
1419                ],
1420            );
1421        }
1422    }
1423    if table_exists(&conn, "session_head_metadata_head_lineage")? {
1424        let mut statement = conn.prepare(
1425            "SELECT session_id, transition_id, predecessor_head_cas_token, \
1426             successor_head_cas_token, predecessor_state_id, successor_state_id, \
1427             created_at_ms FROM session_head_metadata_head_lineage \
1428             ORDER BY session_id, transition_id",
1429        )?;
1430        let mut rows = statement.query([])?;
1431        while let Some(row) = rows.next()? {
1432            let session_id: String = row.get(0)?;
1433            let transition_id: String = row.get(1)?;
1434            let predecessor_head_cas_token: Option<String> = row.get(2)?;
1435            let successor_head_cas_token: String = row.get(3)?;
1436            let predecessor_state_id: Option<String> = row.get(4)?;
1437            let successor_state_id: String = row.get(5)?;
1438            let created_at_ms: i64 = row.get(6)?;
1439            let predecessor_head_present = [u8::from(predecessor_head_cas_token.is_some())];
1440            let predecessor_state_present = [u8::from(predecessor_state_id.is_some())];
1441            feed(
1442                session_id,
1443                "head-metadata-head-lineage",
1444                &[
1445                    transition_id.as_bytes(),
1446                    &predecessor_head_present,
1447                    predecessor_head_cas_token
1448                        .as_deref()
1449                        .unwrap_or_default()
1450                        .as_bytes(),
1451                    successor_head_cas_token.as_bytes(),
1452                    &predecessor_state_present,
1453                    predecessor_state_id
1454                        .as_deref()
1455                        .unwrap_or_default()
1456                        .as_bytes(),
1457                    successor_state_id.as_bytes(),
1458                    created_at_ms.to_be_bytes().as_slice(),
1459                ],
1460            );
1461        }
1462    }
1463
1464    Ok(hashers
1465        .into_iter()
1466        .map(|(session_id, hasher)| (session_id, hasher.finalize().into()))
1467        .collect())
1468}
1469
1470/// Stream one file's bytes into `hasher` through the fixed-size buffer
1471/// `std::io::copy` maintains — divergence hashing must never materialize an
1472/// entire database (or blob) in memory.
1473fn stream_file_into(hasher: &mut Sha256, path: &Path) -> Result<(), StoreError> {
1474    let mut file = fs::File::open(path)?;
1475    std::io::copy(&mut file, hasher)?;
1476    Ok(())
1477}
1478
1479/// Streamed digest of one file's bytes; for SQLite databases the `-wal`
1480/// sidecar (if present) is folded in, so uncheckpointed frames register as
1481/// divergence (conservative: never reports "equal" for possibly-different
1482/// content).
1483fn file_digest(path: &Path) -> Result<[u8; 32], StoreError> {
1484    let mut hasher = Sha256::new();
1485    stream_file_into(&mut hasher, path)?;
1486    let mut wal = path.as_os_str().to_os_string();
1487    wal.push("-wal");
1488    let wal = PathBuf::from(wal);
1489    match fs::symlink_metadata(&wal) {
1490        // A symlinked sidecar would smuggle external bytes into the digest;
1491        // refuse it so the entry poisons as Unknown instead of comparing.
1492        Ok(metadata) if metadata.file_type().is_symlink() => {
1493            return Err(StoreError::Internal(format!(
1494                "refusing to follow symlink {} while digesting {}",
1495                wal.display(),
1496                path.display()
1497            )));
1498        }
1499        Ok(metadata) if metadata.is_file() => stream_file_into(&mut hasher, &wal)?,
1500        _ => {}
1501    }
1502    Ok(hasher.finalize().into())
1503}
1504
1505/// Everything authoritative one split-brain copy materializes, read in a
1506/// single sequential pass for that side.
1507#[derive(Debug)]
1508struct RealmContentSnapshot {
1509    /// `None`: a sessions database exists but its rows could not be read —
1510    /// the whole session comparison is poisoned (a failed read is not an
1511    /// empty side). An absent database is legitimately empty (`Some`).
1512    sessions: Option<SessionDigests>,
1513    /// Relative path (`/`-separated) → streamed content digest.
1514    files: BTreeMap<String, [u8; 32]>,
1515    /// Entries that exist on this side but could not be read/enumerated.
1516    /// Directory entries carry a trailing `/` and poison every path under
1517    /// the prefix.
1518    unreadable: Vec<String>,
1519    /// Human-readable read failures (folded into the report's errors).
1520    errors: Vec<String>,
1521}
1522
1523impl RealmContentSnapshot {
1524    fn poisons(&self, relative: &str) -> bool {
1525        self.unreadable.iter().any(|entry| {
1526            entry == relative || (entry.ends_with('/') && relative.starts_with(entry.as_str()))
1527        })
1528    }
1529
1530    fn record_unreadable(&mut self, relative: String, error: String) {
1531        self.errors.push(error);
1532        self.unreadable.push(relative);
1533    }
1534}
1535
1536fn snapshot_relative(location: &Path, path: &Path) -> Option<String> {
1537    path.strip_prefix(location)
1538        .ok()
1539        .map(|relative| relative.to_string_lossy().replace('\\', "/"))
1540}
1541
1542/// Digest one directory entry into the snapshot. Symlinks are never
1543/// followed (a link could smuggle external bytes into — or hide realm bytes
1544/// from — the comparison) and non-regular files cannot be accounted for:
1545/// both poison their entry instead of silently passing.
1546fn digest_entry_into(path: &Path, relative: String, snapshot: &mut RealmContentSnapshot) {
1547    match fs::symlink_metadata(path) {
1548        Ok(metadata) if metadata.file_type().is_symlink() => snapshot.record_unreadable(
1549            relative,
1550            format!(
1551                "refusing to follow symlink {} during divergence comparison",
1552                path.display()
1553            ),
1554        ),
1555        Ok(metadata) if metadata.is_file() => match file_digest(path) {
1556            Ok(digest) => {
1557                snapshot.files.insert(relative, digest);
1558            }
1559            Err(error) => snapshot.record_unreadable(
1560                relative,
1561                format!("file digest unavailable for {}: {error}", path.display()),
1562            ),
1563        },
1564        Ok(_) => snapshot.record_unreadable(
1565            relative,
1566            format!(
1567                "{} is not a regular file; divergence cannot account for it",
1568                path.display()
1569            ),
1570        ),
1571        Err(error) => {
1572            snapshot
1573                .record_unreadable(relative, format!("cannot stat {}: {error}", path.display()));
1574        }
1575    }
1576}
1577
1578/// The canonical JSONL session files (`sessions_jsonl/*.jsonl`) — the
1579/// durable truth of a jsonl-backend realm; the SQLite index next to them is
1580/// a derived projection covered by the database sweep.
1581fn snapshot_jsonl_sessions(location: &Path, snapshot: &mut RealmContentSnapshot) {
1582    let dir = location.join("sessions_jsonl");
1583    let entries = match fs::read_dir(&dir) {
1584        Ok(entries) => entries,
1585        Err(error) if error.kind() == std::io::ErrorKind::NotFound => return,
1586        Err(error) => {
1587            snapshot.record_unreadable(
1588                "sessions_jsonl/".to_string(),
1589                format!("cannot list {}: {error}", dir.display()),
1590            );
1591            return;
1592        }
1593    };
1594    let mut paths: Vec<PathBuf> = entries
1595        .filter_map(Result::ok)
1596        .map(|entry| entry.path())
1597        .collect();
1598    paths.sort();
1599    for path in paths {
1600        if path.extension().and_then(|ext| ext.to_str()) != Some("jsonl") {
1601            continue;
1602        }
1603        let Some(relative) = snapshot_relative(location, &path) else {
1604            continue;
1605        };
1606        digest_entry_into(&path, relative, snapshot);
1607    }
1608}
1609
1610/// Recursive content walk of one authoritative tree (`blobs/`,
1611/// `artifacts/`).
1612fn snapshot_tree(location: &Path, tree: &str, snapshot: &mut RealmContentSnapshot) {
1613    let root = location.join(tree);
1614    match fs::symlink_metadata(&root) {
1615        Ok(metadata) if metadata.file_type().is_symlink() => {
1616            snapshot.record_unreadable(
1617                format!("{tree}/"),
1618                format!(
1619                    "refusing to follow symlink {} during divergence comparison",
1620                    root.display()
1621                ),
1622            );
1623            return;
1624        }
1625        Ok(metadata) if metadata.is_dir() => {}
1626        Ok(_) | Err(_) => return, // absent (or a stray file the db sweep ignores)
1627    }
1628    walk_tree_into(location, &root, snapshot);
1629}
1630
1631fn walk_tree_into(location: &Path, dir: &Path, snapshot: &mut RealmContentSnapshot) {
1632    let entries = match fs::read_dir(dir) {
1633        Ok(entries) => entries,
1634        Err(error) => {
1635            let relative = snapshot_relative(location, dir).unwrap_or_default();
1636            snapshot.record_unreadable(
1637                format!("{relative}/"),
1638                format!("cannot list {}: {error}", dir.display()),
1639            );
1640            return;
1641        }
1642    };
1643    let mut paths: Vec<PathBuf> = entries
1644        .filter_map(Result::ok)
1645        .map(|entry| entry.path())
1646        .collect();
1647    paths.sort();
1648    for path in paths {
1649        let Some(relative) = snapshot_relative(location, &path) else {
1650            continue;
1651        };
1652        match fs::symlink_metadata(&path) {
1653            Ok(metadata) if metadata.is_dir() && !metadata.file_type().is_symlink() => {
1654                walk_tree_into(location, &path, snapshot);
1655            }
1656            _ => digest_entry_into(&path, relative, snapshot),
1657        }
1658    }
1659}
1660
1661/// Snapshot one copy: session rows first, then every authoritative file's
1662/// bytes, in one sequential pass. The caller's fence is what quiesces the
1663/// side; reading rows and bytes together per side is what makes the
1664/// snapshot represent a single moment under that fence.
1665fn snapshot_realm_content(location: &Path) -> RealmContentSnapshot {
1666    let mut snapshot = RealmContentSnapshot {
1667        sessions: Some(SessionDigests::new()),
1668        files: BTreeMap::new(),
1669        unreadable: Vec::new(),
1670        errors: Vec::new(),
1671    };
1672
1673    let sessions_db = location.join("sessions.sqlite3");
1674    match fs::symlink_metadata(&sessions_db) {
1675        // A symlinked sessions database is never opened (the row-level SQL
1676        // would read the external target); the inventory sweep below records
1677        // the refusal, and `None` poisons the session comparison.
1678        Ok(metadata) if metadata.file_type().is_symlink() => snapshot.sessions = None,
1679        Ok(metadata) if metadata.is_file() => match session_digests(&sessions_db) {
1680            Ok(digests) => snapshot.sessions = Some(digests),
1681            Err(error) => {
1682                snapshot.sessions = None;
1683                snapshot.errors.push(format!(
1684                    "session divergence unavailable for {}: {error}",
1685                    sessions_db.display()
1686                ));
1687            }
1688        },
1689        _ => {}
1690    }
1691
1692    let inventory = enumerate_realm_sqlite_inventory(location);
1693    for link in &inventory.symlinks {
1694        let Some(relative) = snapshot_relative(location, link) else {
1695            continue;
1696        };
1697        snapshot.record_unreadable(
1698            relative,
1699            format!(
1700                "refusing to follow symlink {} during divergence comparison",
1701                link.display()
1702            ),
1703        );
1704    }
1705    for db in inventory.files {
1706        let Some(relative) = snapshot_relative(location, &db) else {
1707            continue;
1708        };
1709        match file_digest(&db) {
1710            Ok(digest) => {
1711                snapshot.files.insert(relative, digest);
1712            }
1713            Err(error) => snapshot.record_unreadable(
1714                relative,
1715                format!("file digest unavailable for {}: {error}", db.display()),
1716            ),
1717        }
1718    }
1719
1720    snapshot_jsonl_sessions(location, &mut snapshot);
1721    snapshot_tree(location, "blobs", &mut snapshot);
1722    snapshot_tree(location, "artifacts", &mut snapshot);
1723    snapshot
1724}
1725
1726/// Compute the per-domain divergence report for one split-brain realm.
1727///
1728/// Sessions are compared row-level (id + content digest over the session
1729/// tables, read-only SQL); every other authoritative store — the SQLite
1730/// databases, the canonical `sessions_jsonl/*.jsonl` files, and the
1731/// `blobs/` / `artifacts/` trees — is compared at streamed file-digest
1732/// level. Read failures are folded into [`SplitBrainReport::errors`] AND
1733/// poison the affected entries as [`DivergenceStatus::Unknown`]: a failed
1734/// read must never manufacture an equal / only-in claim
1735/// ([`SplitBrainReport::comparison_is_conclusive`] is the archive-decision
1736/// gate).
1737///
1738/// Callers that act on the report (`--apply --adopt-root`) must hold the
1739/// [`RealmMaintenanceFence`] on every location for the whole
1740/// compare-to-archive interval: the fence excludes foreign guarded writers,
1741/// and each side is then read in a single sequential pass (rows, then
1742/// bytes) so the snapshot reflects one quiesced moment. Unfenced (dry-run)
1743/// reports are advisory.
1744pub fn compute_split_brain_report(realm: &str, locations: &[PathBuf]) -> SplitBrainReport {
1745    let mut report = SplitBrainReport::new(realm, locations.to_vec());
1746
1747    let snapshots: Vec<(PathBuf, RealmContentSnapshot)> = locations
1748        .iter()
1749        .map(|location| (location.clone(), snapshot_realm_content(location)))
1750        .collect();
1751    for (_, snapshot) in &snapshots {
1752        report.errors.extend(snapshot.errors.iter().cloned());
1753    }
1754
1755    // Row-level session comparison. One unreadable sessions database
1756    // poisons every session entry (Unknown, never equal / only-in).
1757    let sessions_poisoned = snapshots
1758        .iter()
1759        .any(|(_, snapshot)| snapshot.sessions.is_none());
1760    let mut all_ids: Vec<String> = snapshots
1761        .iter()
1762        .filter_map(|(_, snapshot)| snapshot.sessions.as_ref())
1763        .flat_map(|digests| digests.keys().cloned())
1764        .collect();
1765    all_ids.sort();
1766    all_ids.dedup();
1767    for session_id in all_ids {
1768        if sessions_poisoned {
1769            report.sessions.push(SessionDivergenceEntry {
1770                session_id,
1771                status: DivergenceStatus::Unknown,
1772            });
1773            continue;
1774        }
1775        let holders: Vec<(&PathBuf, &[u8; 32])> = snapshots
1776            .iter()
1777            .filter_map(|(location, snapshot)| {
1778                snapshot
1779                    .sessions
1780                    .as_ref()?
1781                    .get(&session_id)
1782                    .map(|digest| (location, digest))
1783            })
1784            .collect();
1785        if holders.len() == 1 {
1786            report.sessions.push(SessionDivergenceEntry {
1787                session_id,
1788                status: DivergenceStatus::OnlyIn {
1789                    location: holders[0].0.clone(),
1790                },
1791            });
1792        } else if holders.len() == snapshots.len()
1793            && holders.iter().all(|(_, digest)| *digest == holders[0].1)
1794        {
1795            report.sessions_equal += 1;
1796        } else {
1797            report.sessions.push(SessionDivergenceEntry {
1798                session_id,
1799                status: DivergenceStatus::Divergent,
1800            });
1801        }
1802    }
1803
1804    // Per-file comparison across every authoritative store.
1805    let mut relative_files: Vec<String> = snapshots
1806        .iter()
1807        .flat_map(|(_, snapshot)| {
1808            snapshot.files.keys().cloned().chain(
1809                snapshot
1810                    .unreadable
1811                    .iter()
1812                    .filter(|entry| !entry.ends_with('/'))
1813                    .cloned(),
1814            )
1815        })
1816        .collect();
1817    relative_files.sort();
1818    relative_files.dedup();
1819    for relative in relative_files {
1820        if snapshots
1821            .iter()
1822            .any(|(_, snapshot)| snapshot.poisons(&relative))
1823        {
1824            report.files.push(FileDivergenceEntry {
1825                file: relative,
1826                status: DivergenceStatus::Unknown,
1827            });
1828            continue;
1829        }
1830        let holders: Vec<(&PathBuf, &[u8; 32])> = snapshots
1831            .iter()
1832            .filter_map(|(location, snapshot)| {
1833                snapshot
1834                    .files
1835                    .get(&relative)
1836                    .map(|digest| (location, digest))
1837            })
1838            .collect();
1839        let status = if holders.len() == 1 {
1840            DivergenceStatus::OnlyIn {
1841                location: holders[0].0.clone(),
1842            }
1843        } else if holders.len() == snapshots.len()
1844            && holders.iter().all(|(_, digest)| *digest == holders[0].1)
1845        {
1846            DivergenceStatus::Equal
1847        } else {
1848            DivergenceStatus::Divergent
1849        };
1850        report.files.push(FileDivergenceEntry {
1851            file: relative,
1852            status,
1853        });
1854    }
1855
1856    report
1857}
1858
1859// ─────────────────────────────────────────────────────────────────────────
1860// Prune (registered backup-artifact lifecycle).
1861// ─────────────────────────────────────────────────────────────────────────
1862
1863fn recursive_size(path: &Path) -> u64 {
1864    let Ok(metadata) = fs::symlink_metadata(path) else {
1865        return 0;
1866    };
1867    if metadata.is_dir() {
1868        let Ok(entries) = fs::read_dir(path) else {
1869            return 0;
1870        };
1871        entries
1872            .filter_map(Result::ok)
1873            .map(|entry| recursive_size(&entry.path()))
1874            .sum()
1875    } else {
1876        metadata.len()
1877    }
1878}
1879
1880/// The Unix timestamp a registered artifact name embeds (backup names carry
1881/// `.pre-<version>-<ts>[.purpose]`, quarantines `.corrupt-<ts>`). `None`
1882/// for names outside the registered patterns.
1883pub fn registered_artifact_timestamp(name: &str) -> Option<u64> {
1884    if is_backup_artifact_name(name) {
1885        let idx = name.rfind(".pre-")?;
1886        let suffix = &name[idx + ".pre-".len()..];
1887        let (_, rest) = suffix.split_once('-')?;
1888        let timestamp = rest
1889            .split_once('.')
1890            .map_or(rest, |(timestamp, _)| timestamp);
1891        return timestamp.parse().ok();
1892    }
1893    if is_quarantine_artifact_name(name) {
1894        let idx = name.rfind(".corrupt-")?;
1895        return name[idx + ".corrupt-".len()..].parse().ok();
1896    }
1897    None
1898}
1899
1900/// Age of one registered artifact, from the timestamp its name registered
1901/// at archive time. `fs::rename` preserves the source's mtime, so a
1902/// long-idle file archived today would look 30+ days old to an
1903/// mtime-based clock and be prunable immediately; the registered name is
1904/// the archival record. Filesystem mtime is only the fallback for
1905/// registered names whose timestamp overflows.
1906fn age_days(path: &Path, name: &str) -> u64 {
1907    let now = SystemTime::now()
1908        .duration_since(UNIX_EPOCH)
1909        .map(|elapsed| elapsed.as_secs())
1910        .unwrap_or(0);
1911    if let Some(registered) = registered_artifact_timestamp(name) {
1912        return now.saturating_sub(registered) / 86_400;
1913    }
1914    fs::symlink_metadata(path)
1915        .and_then(|metadata| metadata.modified())
1916        .ok()
1917        .and_then(|modified| SystemTime::now().duration_since(modified).ok())
1918        .map(|age| age.as_secs() / 86_400)
1919        .unwrap_or(0)
1920}
1921
1922fn artifact_kind(name: &str) -> Option<PruneArtifactKind> {
1923    if is_backup_artifact_name(name) {
1924        Some(PruneArtifactKind::BackupArtifact)
1925    } else if is_quarantine_artifact_name(name) {
1926        Some(PruneArtifactKind::QuarantinedIndex)
1927    } else {
1928        None
1929    }
1930}
1931
1932/// The original name a registered artifact was archived from (the part
1933/// before the registered `.pre-` / `.corrupt-` suffix). `None` for names
1934/// outside the registered patterns.
1935pub fn registered_artifact_original(name: &str) -> Option<&str> {
1936    if is_backup_artifact_name(name) {
1937        return Some(&name[..name.rfind(".pre-")?]);
1938    }
1939    if is_quarantine_artifact_name(name) {
1940        return Some(&name[..name.rfind(".corrupt-")?]);
1941    }
1942    None
1943}
1944
1945fn push_artifacts_in(
1946    dir: &Path,
1947    dirs_too: bool,
1948    original_filter: Option<&str>,
1949    artifacts: &mut Vec<PruneArtifact>,
1950) {
1951    let Ok(entries) = fs::read_dir(dir) else {
1952        return;
1953    };
1954    let mut paths: Vec<PathBuf> = entries
1955        .filter_map(Result::ok)
1956        .map(|entry| entry.path())
1957        .collect();
1958    paths.sort();
1959    for path in paths {
1960        if path.is_dir() && !dirs_too {
1961            continue;
1962        }
1963        let Some(name) = path.file_name().and_then(|name| name.to_str()) else {
1964            continue;
1965        };
1966        let Some(kind) = artifact_kind(name) else {
1967            continue;
1968        };
1969        // Quarantine artifacts are owned index FILES; a directory (or
1970        // symlink) wearing the `.corrupt-*` name is not one and must never
1971        // enter prune's deletion authority.
1972        if kind == PruneArtifactKind::QuarantinedIndex
1973            && !fs::symlink_metadata(&path)
1974                .map(|metadata| metadata.is_file())
1975                .unwrap_or(false)
1976        {
1977            continue;
1978        }
1979        if let Some(filter) = original_filter
1980            && registered_artifact_original(name) != Some(filter)
1981        {
1982            continue;
1983        }
1984        artifacts.push(PruneArtifact {
1985            bytes: recursive_size(&path),
1986            age_days: age_days(&path, name),
1987            path,
1988            kind,
1989            action: PruneAction::Kept,
1990        });
1991    }
1992}
1993
1994/// Enumerate every registered maintenance artifact under the swept state
1995/// roots: root-level `*.pre-*` archives (files or whole archived realm
1996/// directories) and, inside each realm directory, `*.pre-*` backups and
1997/// `*.corrupt-*` quarantines next to the databases (realm root, `memory/`,
1998/// `sessions_jsonl/`, `mobs/`). Nothing outside these naming patterns is
1999/// ever returned — prune's deletion authority is exactly this listing.
2000pub fn enumerate_maintenance_artifacts(state_roots: &[PathBuf]) -> Vec<PruneArtifact> {
2001    enumerate_maintenance_artifacts_filtered(state_roots, None)
2002}
2003
2004/// [`enumerate_maintenance_artifacts`] scoped to one realm. Realm-directory
2005/// artifacts are returned only for realms whose manifest identity matches;
2006/// root-level archives are returned only when they were archived from the
2007/// realm's directory name (whole-realm split-brain archives). Root-level
2008/// file archives that carry no realm identity are excluded under a filter —
2009/// a scoped prune must never delete another realm's preserved copy.
2010pub fn enumerate_maintenance_artifacts_filtered(
2011    state_roots: &[PathBuf],
2012    realm_filter: Option<&str>,
2013) -> Vec<PruneArtifact> {
2014    let realm_dir_name = realm_filter.map(meerkat_core::sanitize_realm_id);
2015    let mut artifacts = Vec::new();
2016    let mut seen_roots: Vec<PathBuf> = Vec::new();
2017    for root in state_roots {
2018        let canonical = fs::canonicalize(root).unwrap_or_else(|_| root.clone());
2019        if seen_roots.contains(&canonical) {
2020            continue;
2021        }
2022        seen_roots.push(canonical);
2023
2024        // Root level: archived realm directories and archived files.
2025        push_artifacts_in(root, true, realm_dir_name.as_deref(), &mut artifacts);
2026
2027        // Inside each live realm directory (backup artifacts inside archived
2028        // copies are part of the archive, owned by the archive's own entry).
2029        for realm in list_realm_dirs(root) {
2030            if let Some(filter) = realm_filter
2031                && realm.realm_id != filter
2032            {
2033                continue;
2034            }
2035            for scan_dir in [
2036                realm.dir.clone(),
2037                realm.dir.join("memory"),
2038                realm.dir.join("sessions_jsonl"),
2039                realm.dir.join("mobs"),
2040            ] {
2041                push_artifacts_in(&scan_dir, false, None, &mut artifacts);
2042            }
2043        }
2044    }
2045    artifacts
2046}
2047
2048/// Delete one registered artifact (file or directory), restoring write
2049/// permissions first (archives are stored read-only). Refuses paths whose
2050/// name does not match the registered patterns — prune never touches
2051/// anything else, even if asked.
2052pub fn remove_maintenance_artifact(path: &Path) -> Result<(), StoreError> {
2053    let name = path
2054        .file_name()
2055        .and_then(|name| name.to_str())
2056        .unwrap_or_default();
2057    if artifact_kind(name).is_none() {
2058        return Err(StoreError::Internal(format!(
2059            "refusing to remove '{}': not a registered maintenance artifact \
2060             (*.pre-* / *.corrupt-*)",
2061            path.display()
2062        )));
2063    }
2064    restore_write_permissions_best_effort(path);
2065    let metadata = fs::symlink_metadata(path)?;
2066    if metadata.is_dir() {
2067        fs::remove_dir_all(path)?;
2068    } else {
2069        fs::remove_file(path)?;
2070    }
2071    Ok(())
2072}
2073
2074#[cfg(test)]
2075#[allow(clippy::expect_used, clippy::unwrap_used, clippy::panic)]
2076mod tests {
2077
2078    #[test]
2079    fn artifact_name_validation_rejects_lookalikes() {
2080        // Full-shape validation: a mere ".pre-" substring is not registered.
2081        for lookalike in [
2082            "notes.pre-release",
2083            "config.pre-prod",
2084            ".pre-0.8.3-1700000000",
2085            "db.pre-0.8.3-notadigit",
2086            "db.pre-083-1700000000",
2087            "db.pre-0.8.3-1700000000.",
2088            "db.pre-0.8.3-1700000000.bad purpose",
2089        ] {
2090            assert!(!is_backup_artifact_name(lookalike), "{lookalike}");
2091        }
2092        for valid in [
2093            "sessions.sqlite3.pre-0.8.3-1700000000",
2094            "sessions.sqlite3.pre-0.8.3-1700000000.split-brain",
2095            "realm-dir.pre-10.20.30-1.adopt_other",
2096        ] {
2097            assert!(is_backup_artifact_name(valid), "{valid}");
2098        }
2099        for lookalike in ["notes.corrupt-ish", ".corrupt-123", "x.corrupt-12a"] {
2100            assert!(!is_quarantine_artifact_name(lookalike), "{lookalike}");
2101        }
2102        assert!(is_quarantine_artifact_name(
2103            "session_index.sqlite3.corrupt-1700000000"
2104        ));
2105    }
2106    use super::*;
2107    use rusqlite::Connection;
2108
2109    fn create_db(path: &Path) {
2110        if let Some(parent) = path.parent() {
2111            fs::create_dir_all(parent).unwrap();
2112        }
2113        let conn = Connection::open(path).unwrap();
2114        conn.execute_batch("CREATE TABLE t (x INTEGER)").unwrap();
2115    }
2116
2117    fn write_manifest(state_root: &Path, realm_id: &str, backend: &str) -> PathBuf {
2118        let dir = state_root.join(meerkat_core::sanitize_realm_id(realm_id));
2119        fs::create_dir_all(&dir).unwrap();
2120        fs::write(
2121            dir.join(REALM_MANIFEST_FILE_NAME),
2122            serde_json::to_vec(&serde_json::json!({
2123                "realm_id": realm_id,
2124                "backend": backend,
2125                "origin": "explicit",
2126                "created_at": "0",
2127            }))
2128            .unwrap(),
2129        )
2130        .unwrap();
2131        dir
2132    }
2133
2134    #[test]
2135    fn fence_acquires_full_inventory_and_mobs_in_sorted_order() {
2136        let temp = tempfile::tempdir().unwrap();
2137        let realm = temp.path().join("realm");
2138        create_db(&realm.join("sessions.sqlite3"));
2139        create_db(&realm.join("workgraph.sqlite3"));
2140        create_db(&realm.join("memory/memory.sqlite3"));
2141        create_db(&realm.join("mobs/alpha.db"));
2142        create_db(&realm.join("mobs/realm_profiles.db"));
2143        // Non-database files are not fenced.
2144        fs::write(realm.join("notes.txt"), b"x").unwrap();
2145
2146        let fence = RealmMaintenanceFence::acquire(&realm, Duration::from_secs(1)).unwrap();
2147        // Full fixed inventory (whether or not the file exists) + mob dbs.
2148        assert_eq!(fence.len(), REALM_SQLITE_FILES.len() + 2);
2149        assert!(!fence.is_empty());
2150        let mut expected: Vec<PathBuf> = REALM_SQLITE_FILES
2151            .iter()
2152            .map(|relative| realm.join(relative))
2153            .collect();
2154        expected.push(realm.join("mobs/alpha.db"));
2155        expected.push(realm.join("mobs/realm_profiles.db"));
2156        expected.sort();
2157        assert_eq!(fence.fenced_databases(), expected.as_slice());
2158        // Every fence lock file exists — including for databases that do
2159        // not exist yet (the fence file is a sibling lock).
2160        for database in fence.fenced_databases() {
2161            assert!(meerkat_sqlite::fence_lock_path(database).is_file());
2162        }
2163        assert!(!realm.join("tasks.db").exists());
2164        // The realm-level write-admission fence is held too.
2165        assert!(fence.admission_lock_path().ends_with("realm.mfence"));
2166        assert!(fence.admission_lock_path().is_file());
2167    }
2168
2169    #[test]
2170    fn foreign_admission_holder_blocks_fence_acquisition() {
2171        let temp = tempfile::tempdir().unwrap();
2172        let realm = temp.path().join("realm");
2173        create_db(&realm.join("sessions.sqlite3"));
2174
2175        // A FOREIGN process holding the write-admission fence (raw exclusive
2176        // lock, no in-process holder registry entry) blocks the realm fence.
2177        let admission_lock = meerkat_sqlite::fence_lock_path(&realm_write_admission_target(&realm));
2178        let foreign = fs::OpenOptions::new()
2179            .read(true)
2180            .write(true)
2181            .create(true)
2182            .truncate(false)
2183            .open(&admission_lock)
2184            .unwrap();
2185        foreign.try_lock().unwrap();
2186
2187        let error = RealmMaintenanceFence::acquire(&realm, Duration::from_millis(100))
2188            .expect_err("foreign admission holder must fail acquisition");
2189        assert!(
2190            matches!(error, StoreError::MaintenanceFenceHeld { ref path }
2191                if path.ends_with(REALM_WRITE_ADMISSION_STEM)),
2192            "{error:?}"
2193        );
2194        drop(foreign);
2195        RealmMaintenanceFence::acquire(&realm, Duration::from_secs(1)).unwrap();
2196    }
2197
2198    #[test]
2199    fn foreign_holder_fails_typed_and_releases_partial_acquisition() {
2200        let temp = tempfile::tempdir().unwrap();
2201        let realm = temp.path().join("realm");
2202        create_db(&realm.join("sessions.sqlite3"));
2203        create_db(&realm.join("workgraph.sqlite3"));
2204
2205        // Simulate a FOREIGN process holding the second fence: a raw
2206        // exclusive lock without the in-process holder registry entry.
2207        let foreign_lock = meerkat_sqlite::fence_lock_path(&realm.join("workgraph.sqlite3"));
2208        let foreign = fs::OpenOptions::new()
2209            .read(true)
2210            .write(true)
2211            .create(true)
2212            .truncate(false)
2213            .open(&foreign_lock)
2214            .unwrap();
2215        foreign.try_lock().unwrap();
2216
2217        let error = RealmMaintenanceFence::acquire(&realm, Duration::from_millis(100))
2218            .expect_err("foreign fence holder must fail acquisition");
2219        assert!(
2220            matches!(error, StoreError::MaintenanceFenceHeld { ref path }
2221                if path.ends_with("workgraph.sqlite3")),
2222            "{error:?}"
2223        );
2224
2225        // RAII: the earlier fences (admission, sessions) were released on
2226        // failure — fresh exclusive acquisitions succeed immediately.
2227        let reacquired =
2228            meerkat_sqlite::ExclusiveFence::try_acquire(&realm.join("sessions.sqlite3")).unwrap();
2229        assert!(reacquired.is_some(), "partial acquisition must be released");
2230        drop(reacquired);
2231        let admission =
2232            meerkat_sqlite::ExclusiveFence::try_acquire(&realm_write_admission_target(&realm))
2233                .unwrap();
2234        assert!(admission.is_some(), "admission fence must be released");
2235        drop(admission);
2236        drop(foreign);
2237
2238        // With the foreign lock gone, full acquisition succeeds.
2239        let fence = RealmMaintenanceFence::acquire(&realm, Duration::from_secs(1)).unwrap();
2240        assert_eq!(fence.len(), REALM_SQLITE_FILES.len());
2241    }
2242
2243    #[test]
2244    fn empty_realm_still_fences_the_full_fixed_inventory() {
2245        let temp = tempfile::tempdir().unwrap();
2246        let fence = RealmMaintenanceFence::acquire(temp.path(), Duration::ZERO).unwrap();
2247        // A database created during maintenance is excluded by its sibling
2248        // fence, so the fixed inventory is fenced even when no file exists.
2249        assert_eq!(fence.len(), REALM_SQLITE_FILES.len());
2250        assert!(!fence.is_empty());
2251        for relative in REALM_SQLITE_FILES {
2252            let lock = meerkat_sqlite::fence_lock_path(&temp.path().join(relative));
2253            assert!(lock.is_file(), "{}", lock.display());
2254        }
2255    }
2256
2257    #[test]
2258    fn backup_names_follow_the_registered_discipline() {
2259        let name = backup_artifact_name("sessions.sqlite3", "split-brain");
2260        assert!(
2261            name.starts_with(&format!(
2262                "sessions.sqlite3.pre-{}-",
2263                env!("CARGO_PKG_VERSION")
2264            )),
2265            "{name}"
2266        );
2267        assert!(name.ends_with(".split-brain"), "{name}");
2268        assert!(is_backup_artifact_name(&name));
2269        let bare = backup_artifact_name("team", "");
2270        assert!(is_backup_artifact_name(&bare));
2271        assert!(!bare.ends_with('.'), "{bare}");
2272        assert!(is_quarantine_artifact_name(
2273            "session_index.sqlite3.corrupt-1"
2274        ));
2275        assert!(!is_backup_artifact_name("sessions.sqlite3"));
2276    }
2277
2278    #[test]
2279    fn archive_renames_and_strips_write_permission() {
2280        let temp = tempfile::tempdir().unwrap();
2281        let dir = temp.path().join("team");
2282        fs::create_dir_all(&dir).unwrap();
2283        fs::write(dir.join("sessions.sqlite3"), b"data").unwrap();
2284
2285        let archive = archive_path_read_only(&dir, "split-brain").unwrap();
2286        assert!(!dir.exists(), "original must be gone");
2287        assert!(archive.is_dir());
2288        assert!(archive.join("sessions.sqlite3").is_file());
2289        let name = archive.file_name().unwrap().to_str().unwrap();
2290        assert!(is_backup_artifact_name(name), "{name}");
2291        let inner = fs::metadata(archive.join("sessions.sqlite3")).unwrap();
2292        assert!(inner.permissions().readonly(), "archive must be read-only");
2293
2294        // Prune can remove it (write permission restored first).
2295        remove_maintenance_artifact(&archive).unwrap();
2296        assert!(!archive.exists());
2297    }
2298
2299    #[test]
2300    fn archive_reported_returns_archive_path_and_no_warnings_on_success() {
2301        let temp = tempfile::tempdir().unwrap();
2302        let file = temp.path().join("sessions.sqlite3");
2303        fs::write(&file, b"data").unwrap();
2304
2305        let archived = archive_path_read_only_reported(&file, "split-brain").unwrap();
2306        assert!(archived.archive.is_file());
2307        assert!(
2308            archived.warnings.is_empty(),
2309            "successful hardening + parent fsync must not warn: {:?}",
2310            archived.warnings
2311        );
2312        let name = archived.archive.file_name().unwrap().to_str().unwrap();
2313        assert!(is_backup_artifact_name(name), "{name}");
2314        remove_maintenance_artifact(&archived.archive).unwrap();
2315    }
2316
2317    #[cfg(unix)]
2318    #[test]
2319    fn archive_hardening_never_chmods_through_symlinks() {
2320        use std::os::unix::fs::PermissionsExt;
2321
2322        let temp = tempfile::tempdir().unwrap();
2323        let external = temp.path().join("external.txt");
2324        fs::write(&external, b"do not chmod").unwrap();
2325        let dir = temp.path().join("team");
2326        fs::create_dir_all(&dir).unwrap();
2327        std::os::unix::fs::symlink(&external, dir.join("link")).unwrap();
2328
2329        let archived = archive_path_read_only_reported(&dir, "split-brain").unwrap();
2330        let mode = fs::metadata(&external).unwrap().permissions().mode();
2331        assert!(
2332            mode & 0o200 != 0,
2333            "external symlink target must stay writable (mode {mode:o})"
2334        );
2335        // Restore-then-delete must not chmod through the link either.
2336        remove_maintenance_artifact(&archived.archive).unwrap();
2337        let mode = fs::metadata(&external).unwrap().permissions().mode();
2338        assert!(mode & 0o200 != 0, "mode {mode:o}");
2339        assert!(external.is_file());
2340    }
2341
2342    #[test]
2343    fn remove_refuses_unregistered_paths() {
2344        let temp = tempfile::tempdir().unwrap();
2345        let victim = temp.path().join("precious.txt");
2346        fs::write(&victim, b"do not touch").unwrap();
2347        let error = remove_maintenance_artifact(&victim).expect_err("must refuse");
2348        assert!(matches!(error, StoreError::Internal(_)));
2349        assert!(victim.is_file(), "unregistered path must survive");
2350    }
2351
2352    #[test]
2353    fn split_brain_divergence_classifies_sessions_and_files() {
2354        let temp = tempfile::tempdir().unwrap();
2355        let dir_a = temp.path().join("a/team");
2356        let dir_b = temp.path().join("b/team");
2357        for dir in [&dir_a, &dir_b] {
2358            fs::create_dir_all(dir).unwrap();
2359        }
2360        let ddl = "CREATE TABLE sessions (
2361            session_id TEXT PRIMARY KEY,
2362            created_at_ms INTEGER NOT NULL,
2363            updated_at_ms INTEGER NOT NULL,
2364            message_count INTEGER NOT NULL,
2365            total_tokens INTEGER NOT NULL,
2366            metadata_json TEXT NOT NULL,
2367            session_json BLOB NOT NULL
2368        )";
2369        let insert = |conn: &Connection, id: &str, body: &str| {
2370            conn.execute(
2371                "INSERT INTO sessions VALUES (?1, 0, 0, 0, 0, '{}', ?2)",
2372                rusqlite::params![id, body.as_bytes()],
2373            )
2374            .unwrap();
2375        };
2376        let shared = "00000000-0000-4000-8000-000000000001";
2377        let divergent = "00000000-0000-4000-8000-000000000002";
2378        let only_a = "00000000-0000-4000-8000-000000000003";
2379        {
2380            let conn = Connection::open(dir_a.join("sessions.sqlite3")).unwrap();
2381            conn.execute_batch(ddl).unwrap();
2382            insert(&conn, shared, "same");
2383            insert(&conn, divergent, "version-a");
2384            insert(&conn, only_a, "solo");
2385        }
2386        {
2387            let conn = Connection::open(dir_b.join("sessions.sqlite3")).unwrap();
2388            conn.execute_batch(ddl).unwrap();
2389            insert(&conn, shared, "same");
2390            insert(&conn, divergent, "version-b");
2391        }
2392        // A file present only under B.
2393        create_db(&dir_b.join("workgraph.sqlite3"));
2394
2395        let report = compute_split_brain_report("team", &[dir_a.clone(), dir_b.clone()]);
2396        assert!(report.errors.is_empty(), "{:?}", report.errors);
2397        assert_eq!(report.sessions_equal, 1);
2398        let status_of = |id: &str| {
2399            report
2400                .sessions
2401                .iter()
2402                .find(|entry| entry.session_id == id)
2403                .map(|entry| entry.status.clone())
2404        };
2405        assert_eq!(status_of(divergent), Some(DivergenceStatus::Divergent));
2406        assert_eq!(
2407            status_of(only_a),
2408            Some(DivergenceStatus::OnlyIn { location: dir_a })
2409        );
2410        assert_eq!(
2411            status_of(shared),
2412            None,
2413            "equal sessions are counted, not listed"
2414        );
2415        let workgraph = report
2416            .files
2417            .iter()
2418            .find(|entry| entry.file == "workgraph.sqlite3")
2419            .expect("workgraph file entry");
2420        assert_eq!(
2421            workgraph.status,
2422            DivergenceStatus::OnlyIn { location: dir_b }
2423        );
2424        let sessions_file = report
2425            .files
2426            .iter()
2427            .find(|entry| entry.file == "sessions.sqlite3")
2428            .expect("sessions file entry");
2429        assert_eq!(sessions_file.status, DivergenceStatus::Divergent);
2430        assert!(report.comparison_is_conclusive());
2431    }
2432
2433    #[test]
2434    fn split_brain_covers_jsonl_blobs_and_artifacts() {
2435        let temp = tempfile::tempdir().unwrap();
2436        let dir_a = temp.path().join("a/team");
2437        let dir_b = temp.path().join("b/team");
2438        for dir in [&dir_a, &dir_b] {
2439            fs::create_dir_all(dir.join("sessions_jsonl")).unwrap();
2440            fs::create_dir_all(dir.join("artifacts")).unwrap();
2441        }
2442        // Canonical JSONL session files: one equal, one divergent.
2443        for dir in [&dir_a, &dir_b] {
2444            fs::write(dir.join("sessions_jsonl/equal.jsonl"), b"{\"same\":1}").unwrap();
2445        }
2446        fs::write(dir_a.join("sessions_jsonl/split.jsonl"), b"version-a").unwrap();
2447        fs::write(dir_b.join("sessions_jsonl/split.jsonl"), b"version-b").unwrap();
2448        // A blob present only under A (nested shard directory).
2449        fs::create_dir_all(dir_a.join("blobs/ab")).unwrap();
2450        fs::write(dir_a.join("blobs/ab/abcd.json"), b"blob-bytes").unwrap();
2451        // An artifact record that diverges.
2452        fs::write(dir_a.join("artifacts/r1.json"), b"{\"v\":1}").unwrap();
2453        fs::write(dir_b.join("artifacts/r1.json"), b"{\"v\":2}").unwrap();
2454
2455        let report = compute_split_brain_report("team", &[dir_a.clone(), dir_b]);
2456        assert!(report.errors.is_empty(), "{:?}", report.errors);
2457        let status_of = |file: &str| {
2458            report
2459                .files
2460                .iter()
2461                .find(|entry| entry.file == file)
2462                .map(|entry| entry.status.clone())
2463        };
2464        assert_eq!(
2465            status_of("sessions_jsonl/equal.jsonl"),
2466            Some(DivergenceStatus::Equal)
2467        );
2468        assert_eq!(
2469            status_of("sessions_jsonl/split.jsonl"),
2470            Some(DivergenceStatus::Divergent)
2471        );
2472        assert_eq!(
2473            status_of("blobs/ab/abcd.json"),
2474            Some(DivergenceStatus::OnlyIn { location: dir_a })
2475        );
2476        assert_eq!(
2477            status_of("artifacts/r1.json"),
2478            Some(DivergenceStatus::Divergent)
2479        );
2480        assert!(report.comparison_is_conclusive());
2481    }
2482
2483    #[test]
2484    fn split_brain_read_failure_poisons_the_session_comparison() {
2485        let temp = tempfile::tempdir().unwrap();
2486        let dir_a = temp.path().join("a/team");
2487        let dir_b = temp.path().join("b/team");
2488        for dir in [&dir_a, &dir_b] {
2489            fs::create_dir_all(dir).unwrap();
2490        }
2491        // Copy A's sessions database is unreadable garbage; copy B holds a
2492        // real session row. B's row must classify Unknown, never OnlyIn —
2493        // a failed read is not an empty side.
2494        fs::write(dir_a.join("sessions.sqlite3"), b"this is not sqlite").unwrap();
2495        {
2496            let conn = Connection::open(dir_b.join("sessions.sqlite3")).unwrap();
2497            conn.execute_batch(
2498                "CREATE TABLE sessions (
2499                    session_id TEXT PRIMARY KEY,
2500                    created_at_ms INTEGER NOT NULL,
2501                    updated_at_ms INTEGER NOT NULL,
2502                    message_count INTEGER NOT NULL,
2503                    total_tokens INTEGER NOT NULL,
2504                    metadata_json TEXT NOT NULL,
2505                    session_json BLOB NOT NULL
2506                );
2507                INSERT INTO sessions VALUES
2508                    ('00000000-0000-4000-8000-000000000001', 0, 0, 0, 0, '{}', X'AA');",
2509            )
2510            .unwrap();
2511        }
2512
2513        let report = compute_split_brain_report("team", &[dir_a, dir_b]);
2514        assert!(!report.errors.is_empty());
2515        assert_eq!(report.sessions_equal, 0);
2516        assert_eq!(report.sessions.len(), 1);
2517        assert_eq!(report.sessions[0].status, DivergenceStatus::Unknown);
2518        assert!(!report.comparison_is_conclusive());
2519    }
2520
2521    #[cfg(unix)]
2522    #[test]
2523    fn split_brain_never_follows_symlinks_and_marks_them_unknown() {
2524        let temp = tempfile::tempdir().unwrap();
2525        let outside = temp.path().join("outside.json");
2526        fs::write(&outside, b"external bytes").unwrap();
2527        let dir_a = temp.path().join("a/team");
2528        let dir_b = temp.path().join("b/team");
2529        for dir in [&dir_a, &dir_b] {
2530            fs::create_dir_all(dir.join("blobs")).unwrap();
2531        }
2532        std::os::unix::fs::symlink(&outside, dir_a.join("blobs/link.json")).unwrap();
2533        fs::write(dir_b.join("blobs/link.json"), b"external bytes").unwrap();
2534
2535        let report = compute_split_brain_report("team", &[dir_a, dir_b]);
2536        assert!(!report.errors.is_empty());
2537        let entry = report
2538            .files
2539            .iter()
2540            .find(|entry| entry.file == "blobs/link.json")
2541            .expect("symlinked entry");
2542        assert_eq!(entry.status, DivergenceStatus::Unknown);
2543        assert!(!report.comparison_is_conclusive());
2544    }
2545
2546    /// A symlinked database in the fixed inventory must never be digested
2547    /// through to its external target: the entry is refused (Unknown), the
2548    /// refusal is a per-realm error, and for `sessions.sqlite3` the
2549    /// row-level session comparison is poisoned too.
2550    #[cfg(unix)]
2551    #[test]
2552    fn split_brain_never_digests_through_symlinked_databases() {
2553        let temp = tempfile::tempdir().unwrap();
2554        let outside = temp.path().join("outside.sqlite3");
2555        {
2556            let conn = Connection::open(&outside).unwrap();
2557            conn.execute_batch(
2558                "CREATE TABLE sessions (
2559                    session_id TEXT PRIMARY KEY,
2560                    created_at_ms INTEGER NOT NULL,
2561                    updated_at_ms INTEGER NOT NULL,
2562                    message_count INTEGER NOT NULL,
2563                    total_tokens INTEGER NOT NULL,
2564                    metadata_json TEXT NOT NULL,
2565                    session_json BLOB NOT NULL
2566                );
2567                INSERT INTO sessions VALUES
2568                    ('00000000-0000-4000-8000-000000000001', 0, 0, 0, 0, '{}', X'AA');",
2569            )
2570            .unwrap();
2571        }
2572        let dir_a = temp.path().join("a/team");
2573        let dir_b = temp.path().join("b/team");
2574        fs::create_dir_all(&dir_a).unwrap();
2575        fs::create_dir_all(&dir_b).unwrap();
2576        std::os::unix::fs::symlink(&outside, dir_a.join("sessions.sqlite3")).unwrap();
2577        fs::copy(&outside, dir_b.join("sessions.sqlite3")).unwrap();
2578
2579        let inventory = enumerate_realm_sqlite_inventory(&dir_a);
2580        assert!(inventory.files.is_empty());
2581        assert_eq!(inventory.symlinks, vec![dir_a.join("sessions.sqlite3")]);
2582        assert!(enumerate_realm_sqlite_files(&dir_a).is_empty());
2583
2584        let report = compute_split_brain_report("team", &[dir_a, dir_b]);
2585        assert!(
2586            report
2587                .errors
2588                .iter()
2589                .any(|error| error.contains("refusing to follow symlink")),
2590            "{:?}",
2591            report.errors
2592        );
2593        let entry = report
2594            .files
2595            .iter()
2596            .find(|entry| entry.file == "sessions.sqlite3")
2597            .expect("sessions database entry");
2598        assert_eq!(entry.status, DivergenceStatus::Unknown);
2599        // The poisoned side must also poison the row-level comparison.
2600        assert_eq!(report.sessions.len(), 1, "{:?}", report.sessions);
2601        assert_eq!(report.sessions[0].status, DivergenceStatus::Unknown);
2602        assert!(!report.comparison_is_conclusive());
2603    }
2604
2605    #[test]
2606    fn ledger_baseline_surfaces_read_failures_and_future_versions() {
2607        let temp = tempfile::tempdir().unwrap();
2608        let realm = temp.path().join("team");
2609        fs::create_dir_all(realm.join("sessions_jsonl")).unwrap();
2610        // An unreadable database must surface as an error, never as a
2611        // missing ledger row.
2612        fs::write(realm.join("tasks.db"), b"this is not sqlite").unwrap();
2613        // A future jsonl-index version must be reported as a refusal in
2614        // dry-run exactly as apply's guarded constructor would refuse it.
2615        {
2616            let conn =
2617                Connection::open(realm.join("sessions_jsonl/session_index.sqlite3")).unwrap();
2618            conn.execute_batch(
2619                "CREATE TABLE meerkat_schema (domain TEXT PRIMARY KEY, version INTEGER NOT NULL);",
2620            )
2621            .unwrap();
2622            conn.execute(
2623                "INSERT INTO meerkat_schema VALUES ('jsonl-index', ?1)",
2624                [i64::MAX],
2625            )
2626            .unwrap();
2627        }
2628        // A ledger-less database is a missing-row reading, not an error.
2629        create_db(&realm.join("runtime.sqlite3"));
2630
2631        let baseline = read_realm_ledger_baseline(&realm);
2632        assert_eq!(baseline.errors.len(), 1, "{:?}", baseline.errors);
2633        assert!(
2634            baseline.errors[0].contains("tasks.db"),
2635            "{:?}",
2636            baseline.errors
2637        );
2638        assert!(
2639            !baseline
2640                .rows
2641                .iter()
2642                .any(|row| row.database.ends_with("tasks.db")),
2643            "an unreadable database must contribute no ledger rows"
2644        );
2645        assert_eq!(baseline.future.len(), 1);
2646        assert_eq!(baseline.future[0].domain, "jsonl-index");
2647        assert_eq!(baseline.future[0].found, i64::MAX);
2648        assert_eq!(
2649            baseline.future[0].supported,
2650            crate::index::JSONL_INDEX_DOMAIN.supported_version()
2651        );
2652        let runtime_row = baseline
2653            .rows
2654            .iter()
2655            .find(|row| row.database.ends_with("runtime.sqlite3"))
2656            .expect("runtime row");
2657        assert_eq!(runtime_row.domain, "runtime-store");
2658        assert_eq!(runtime_row.version, None);
2659    }
2660
2661    #[test]
2662    fn prune_enumeration_sees_only_registered_patterns() {
2663        let temp = tempfile::tempdir().unwrap();
2664        let root = temp.path().join("realms");
2665        let realm_dir = write_manifest(&root, "artifacts", "sqlite");
2666        fs::write(
2667            realm_dir.join("sessions.sqlite3.pre-0.0.1-1700000000"),
2668            b"backup",
2669        )
2670        .unwrap();
2671        let jsonl = realm_dir.join("sessions_jsonl");
2672        fs::create_dir_all(&jsonl).unwrap();
2673        fs::write(jsonl.join("session_index.sqlite3.corrupt-42"), b"q").unwrap();
2674        // Root-level archived realm directory.
2675        let archived = root.join("team.pre-0.0.1-1700000000.split-brain");
2676        fs::create_dir_all(&archived).unwrap();
2677        fs::write(archived.join("sessions.sqlite3"), b"old").unwrap();
2678        // Distractors that must never appear.
2679        fs::write(realm_dir.join("sessions.sqlite3"), b"live").unwrap();
2680        fs::write(realm_dir.join("notes.txt"), b"keep me").unwrap();
2681
2682        let artifacts = enumerate_maintenance_artifacts(&[root.clone(), root]);
2683        let mut names: Vec<String> = artifacts
2684            .iter()
2685            .map(|artifact| {
2686                artifact
2687                    .path
2688                    .file_name()
2689                    .unwrap()
2690                    .to_string_lossy()
2691                    .into_owned()
2692            })
2693            .collect();
2694        names.sort();
2695        assert_eq!(
2696            names,
2697            vec![
2698                "session_index.sqlite3.corrupt-42".to_string(),
2699                "sessions.sqlite3.pre-0.0.1-1700000000".to_string(),
2700                "team.pre-0.0.1-1700000000.split-brain".to_string(),
2701            ],
2702            "duplicate roots must not double-count"
2703        );
2704        let archived_entry = artifacts
2705            .iter()
2706            .find(|artifact| artifact.path == archived)
2707            .expect("archived dir entry");
2708        assert_eq!(archived_entry.kind, PruneArtifactKind::BackupArtifact);
2709        assert_eq!(archived_entry.bytes, 3);
2710    }
2711
2712    #[test]
2713    fn quarantine_artifacts_must_be_files_even_at_root_level() {
2714        let temp = tempfile::tempdir().unwrap();
2715        let root = temp.path().join("realms");
2716        fs::create_dir_all(&root).unwrap();
2717        // Quarantine names belong to owned index FILES; a directory wearing
2718        // the name must not enter prune's deletion authority even at the
2719        // root level, where archived realm DIRECTORIES (backup naming) are
2720        // legitimately enumerated.
2721        fs::create_dir_all(root.join("data.corrupt-1700000000")).unwrap();
2722        fs::write(root.join("index.sqlite3.corrupt-1700000000"), b"q").unwrap();
2723
2724        let artifacts = enumerate_maintenance_artifacts(&[root]);
2725        let names: Vec<&str> = artifacts
2726            .iter()
2727            .filter_map(|artifact| artifact.path.file_name().and_then(|name| name.to_str()))
2728            .collect();
2729        assert_eq!(names, vec!["index.sqlite3.corrupt-1700000000"]);
2730    }
2731
2732    #[test]
2733    fn report_shapes_round_trip_through_json() {
2734        let mut report = MigrateReport {
2735            mode: MigrateMode::Apply,
2736            ..MigrateReport::default()
2737        };
2738        let mut realm = RealmMigrateReport::new("team", PathBuf::from("/roots/a/team"));
2739        realm.backend = Some("sqlite".to_string());
2740        realm.ledger.push(LedgerBaselineEntry {
2741            database: PathBuf::from("/roots/a/team/sessions.sqlite3"),
2742            domain: "session-store".to_string(),
2743            before: None,
2744            after: Some(1),
2745            action: LedgerBaselineAction::Stamped,
2746        });
2747        report.realms.push(realm);
2748        report.split_brain.push(SplitBrainReport {
2749            realm: "team".to_string(),
2750            locations: vec![
2751                PathBuf::from("/roots/a/team"),
2752                PathBuf::from("/roots/b/team"),
2753            ],
2754            sessions_equal: 4,
2755            sessions: vec![SessionDivergenceEntry {
2756                session_id: "s".to_string(),
2757                status: DivergenceStatus::Divergent,
2758            }],
2759            files: vec![],
2760            resolution: SplitBrainResolution::Archived {
2761                adopted: PathBuf::from("/roots/a/team"),
2762                archived: vec![PathBuf::from("/roots/b/team.pre-0.8.3-1-split-brain")],
2763            },
2764            errors: vec![],
2765        });
2766        report.split_brain.push(SplitBrainReport {
2767            realm: "solo".to_string(),
2768            locations: vec![PathBuf::from("/roots/a/solo")],
2769            sessions_equal: 0,
2770            sessions: vec![SessionDivergenceEntry {
2771                session_id: "s2".to_string(),
2772                status: DivergenceStatus::Unknown,
2773            }],
2774            files: vec![],
2775            resolution: SplitBrainResolution::ArchiveFailed {
2776                adopted: PathBuf::from("/roots/a/solo"),
2777                archived: vec![PathBuf::from("/roots/b/solo.pre-0.8.3-1.split-brain")],
2778                reason: "rename failed".to_string(),
2779            },
2780            errors: vec!["sessions unreadable".to_string()],
2781        });
2782        let json = serde_json::to_string(&report).expect("serialize");
2783        let parsed: MigrateReport = serde_json::from_str(&json).expect("deserialize");
2784        assert!(matches!(parsed.mode, MigrateMode::Apply));
2785        assert_eq!(parsed.realms.len(), 1);
2786        assert_eq!(
2787            parsed.realms[0].ledger[0].action,
2788            LedgerBaselineAction::Stamped
2789        );
2790        assert!(matches!(
2791            parsed.split_brain[0].resolution,
2792            SplitBrainResolution::Archived { .. }
2793        ));
2794        // Partial-archive vocabulary round-trips: successes stay visible
2795        // next to the failure, and Unknown entries mark the report
2796        // inconclusive.
2797        assert!(matches!(
2798            &parsed.split_brain[1].resolution,
2799            SplitBrainResolution::ArchiveFailed { archived, .. } if archived.len() == 1
2800        ));
2801        assert!(!parsed.split_brain[1].comparison_is_conclusive());
2802        assert!(parsed.split_brain[0].comparison_is_conclusive());
2803        assert!(!parsed.has_errors());
2804
2805        // Forward compatibility: unknown fields tolerated, defaults fill in.
2806        let sparse: PruneReport = serde_json::from_str(r#"{"future_field":1}"#).expect("sparse");
2807        assert!(matches!(sparse.mode, MigrateMode::DryRun));
2808        assert!(sparse.artifacts.is_empty());
2809    }
2810
2811    #[test]
2812    fn read_domain_versions_reads_ledgers_read_only() {
2813        let temp = tempfile::tempdir().unwrap();
2814        let db = temp.path().join("db.sqlite3");
2815        create_db(&db);
2816        assert!(read_domain_versions(&db).unwrap().is_none());
2817        {
2818            let conn = Connection::open(&db).unwrap();
2819            conn.execute_batch(
2820                "CREATE TABLE meerkat_schema (domain TEXT PRIMARY KEY, version INTEGER NOT NULL);
2821                 INSERT INTO meerkat_schema VALUES ('session-store', 1);",
2822            )
2823            .unwrap();
2824        }
2825        assert_eq!(
2826            read_domain_versions(&db).unwrap(),
2827            Some(vec![("session-store".to_string(), 1)])
2828        );
2829    }
2830
2831    #[test]
2832    fn list_realm_dirs_skips_archives_and_reads_manifests_leniently() {
2833        let temp = tempfile::tempdir().unwrap();
2834        let root = temp.path().join("realms");
2835        write_manifest(&root, "alpha", "sqlite");
2836        let corrupt = root.join("corrupt");
2837        fs::create_dir_all(&corrupt).unwrap();
2838        fs::write(corrupt.join(REALM_MANIFEST_FILE_NAME), b"not-json").unwrap();
2839        let archived = root.join("beta.pre-0.0.1-1700000000");
2840        fs::create_dir_all(&archived).unwrap();
2841        fs::write(
2842            archived.join(REALM_MANIFEST_FILE_NAME),
2843            serde_json::to_vec(&serde_json::json!({
2844                "realm_id": "beta", "backend": "sqlite",
2845                "origin": "explicit", "created_at": "0",
2846            }))
2847            .unwrap(),
2848        )
2849        .unwrap();
2850        // A directory without a manifest is not a realm.
2851        fs::create_dir_all(root.join("not-a-realm")).unwrap();
2852
2853        let realms = list_realm_dirs(&root);
2854        let ids: Vec<&str> = realms.iter().map(|realm| realm.realm_id.as_str()).collect();
2855        assert_eq!(ids, vec!["alpha", "corrupt"]);
2856        assert!(realms[0].manifest_readable);
2857        assert_eq!(realms[0].backend.as_deref(), Some("sqlite"));
2858        assert!(!realms[1].manifest_readable);
2859    }
2860}