meerkat-mobkit 0.8.6

//! Offline migration and prune for MobKit state directories (Phase M6 of
//! the storage-unification arc): the mutation half of MobKit's storage
//! maintenance, behind `mobkit_gateway storage-migrate` / `storage-prune`.
//!
//! MobKit builds no second fence: [`MobKitMaintenanceFence`] composes one
//! [`meerkat_sqlite::ExclusiveFence`] per materialized database (meerkat's
//! Phase 6 primitive), and the backup discipline is meerkat's verbatim —
//! [`meerkat_store::migrate::backup_artifact_name`] renames
//! (`<original>.pre-<version>-<timestamp>[.<purpose>]`), never deletes;
//! doctor lists `*.pre-*` artifacts and [`prune_state_dir`] owns their
//! lifecycle.
//!
//! # The five migration cases ([`migrate_state_dir`])
//!
//! 1. **Ledger baseline (auto-safe).** Under the fence, every existing
//!    MobKit-owned database is opened through its NORMAL M3 constructor —
//!    the guarded `meerkat_schema` ledger migrations ARE the structural
//!    verification (continuity via `LocalContinuityStore::open`, metadata
//!    via `SqliteMetadataStore::open`, console via
//!    `SqliteConsoleLogStore::open`, per-realm agent-memory files via the
//!    `SqliteAgentMemoryStore` realm-connection path). Dry-run is a
//!    read-only version matrix. The workgraph admission sidecar is
//!    **exempt** (M3 decision: the lock database deliberately carries no
//!    tables and no ledger); most meerkat-shared databases (sessions, runtime,
//!    schedule, workgraph) are report-only here. The inherited jobs database
//!    is the deliberate exception: this verb invokes Meerkat's canonical
//!    `SqliteDetachedJobStore` constructor while holding the same state-wide
//!    fence, so offline migration covers the Phase-4 jobs domain without
//!    introducing a MobKit-owned schema authority.
//! 2. **File-name unification (auto-safe, rename-only).** A lone legacy
//!    spelling ([`crate::storage_layout::DatabaseSlot::legacy_names`])
//!    renames to its M2 canonical name under the fence, WITH its `-wal` /
//!    `-shm` siblings: a non-empty WAL is checkpointed (`TRUNCATE`) through
//!    a maintenance connection first, and the move refuses if it cannot be.
//!    Renames preserve content, so the "registered backup" is a registered
//!    **rename marker** (`<legacy>.pre-<version>-<ts>.renamed`, a small JSON
//!    record) — doctor lists it as a backup artifact and prune owns it, so
//!    retention tooling sees every rename.
//! 3. **Twin reconciliation (manual, fail-closed).** Both spellings
//!    populated → a per-domain divergence report (row-level by primary key +
//!    content digest for continuity / metadata / console; file-digest for
//!    the rest) and a typed refusal. Byte-identical twins dedup under plain
//!    `--apply` (exact-equality dedup, the one auto resolution the plan
//!    sanctions); divergent twins resolve only with `--apply --adopt
//!    <path>`: the adopted file keeps its place (renamed to canonical if
//!    legacy-named), every other copy is archived read-only under the
//!    registered backup naming. **No synthesis** — continuity fencing
//!    tokens and console `AUTOINCREMENT` cursors are per-database sequences;
//!    merging them corrupts CAS and cursor replay.
//! 4. **Continuity checkpoint adoption.** H3's
//!    [`crate::identity_first::checkpoint_adoption`] walk over the
//!    canonical-resolved continuity database, invoked under the SAME fence
//!    pass via
//!    [`crate::identity_first::adopt_continuity_snapshots_already_fenced`]
//!    (the fence lock is not re-entrant, so case 4 composes with the held
//!    fence instead of re-acquiring); its report is merged.
//! 5. **Deprecated leftovers (report-only).** The doctor's artifact
//!    findings (legacy sharded-FS blobs, `*.pre-*` backups, `*.corrupt-*`
//!    quarantines, the admission sidecar, fence lock files) plus dead
//!    `tux-runtimes.json` registry entries (recorded pid no longer alive).
//!    Nothing in case 5 is ever moved.
//!
//! Any unresolved twin fails the whole run closed (mirroring `rkat storage
//! migrate`): the divergence report is the entire output and cases 1/2/4
//! do not run, because every one of them would have to pick a spelling.

use std::collections::BTreeMap;
use std::fs;
use std::path::{Path, PathBuf};
use std::time::Duration;

use meerkat_core::storage_diagnostics::{DiagnoseScope, FindingSeverity, StorageFinding};
pub use meerkat_store::migrate::{DivergenceStatus, MigrateMode, PruneAction, PruneArtifactKind};
use meerkat_store::migrate::{
    archive_path_read_only, backup_artifact_name, remove_maintenance_artifact,
};
use serde::{Deserialize, Serialize};
use sha2::{Digest, Sha256};

use crate::console_aggregator::SqliteConsoleLogStore;
use crate::identity_first::LocalContinuityStore;
use crate::identity_first::checkpoint_adoption::{
    AdoptionMode, ContinuityAdoptionReport, adopt_continuity_snapshots_already_fenced,
    adopt_continuity_snapshots_blocking,
};
use crate::memory::sqlite_store::SqliteAgentMemoryStore;
use crate::runtime::SqliteMetadataStore;
use crate::storage_doctor::{self, DATABASE_FAMILIES, MEMORY_LEDGER_DOMAIN, MEMORY_ROOT_SPELLINGS};
use crate::storage_layout::{
    DatabaseProvenance, DatabaseSlot, MobKitStorageLayout, RUNTIME_REGISTRY_FILE_NAME,
    StorageLayoutError,
};
use crate::storage_marker_stamp::{
    MarkerStampError, MarkerStampMode, SessionDocumentStore, SessionMarkerStampReport,
    stamp_session_document_markers_already_fenced, stamp_session_document_markers_blocking,
};
use crate::workgraph_admission::WORKGRAPH_ADMISSION_SIDECAR_FILE;

/// How long the fence waits (total, across all files) for in-flight
/// per-operation guards to drain before reporting a foreign holder.
const FENCE_DRAIN_DEADLINE: Duration = Duration::from_secs(10);

/// Stable finding code: a `tux-runtimes.json` registry entry whose recorded
/// pid is no longer alive (report-only; migrate never edits the registry).
pub const FINDING_DEAD_RUNTIME_REGISTRY_ENTRY: &str = "dead-runtime-registry-entry";

/// Strict validation of the COMPLETE generated backup-artifact shape:
/// `<original>.pre-<version>-<unix-ts>[.<purpose>]` — non-empty
/// `<original>`, `<version>` a dotted crate version (two or more all-digit
/// components, e.g. `0.8.3`), an all-digit Unix timestamp, and, when
/// present, a non-empty purpose suffix.
///
/// Deliberately stricter than matching a `.pre-` substring: prune's
/// deletion authority is exactly the names this accepts, and a loose
/// substring match would claim unrelated user files like
/// `notes.pre-release`. Every name [`backup_artifact_name`] generates
/// (including the `-wal`/`-shm` sibling archives, whose suffix folds into
/// the purpose) validates.
pub fn is_registered_backup_artifact_name(name: &str) -> bool {
    let Some(index) = name.rfind(".pre-") else {
        return false;
    };
    if index == 0 {
        return false; // empty <original>
    }
    let rest = &name[index + ".pre-".len()..];
    let Some((version, tail)) = rest.split_once('-') else {
        return false;
    };
    let version_components: Vec<&str> = version.split('.').collect();
    if version_components.len() < 2
        || version_components
            .iter()
            .any(|part| part.is_empty() || !part.bytes().all(|byte| byte.is_ascii_digit()))
    {
        return false;
    }
    let (timestamp, purpose) = match tail.split_once('.') {
        Some((timestamp, purpose)) => (timestamp, Some(purpose)),
        None => (tail, None),
    };
    if timestamp.is_empty() || !timestamp.bytes().all(|byte| byte.is_ascii_digit()) {
        return false;
    }
    purpose != Some("")
}

/// Strict validation of the registered index-quarantine shape: an exact
/// `.corrupt-<digits>` suffix on a non-empty original name (same rationale
/// as [`is_registered_backup_artifact_name`] — `report.corrupt-12a` or a
/// `.bak` copy of a quarantine is never prune's to delete).
pub fn is_registered_quarantine_artifact_name(name: &str) -> bool {
    let Some(index) = name.rfind(".corrupt-") else {
        return false;
    };
    if index == 0 {
        return false;
    }
    let digits = &name[index + ".corrupt-".len()..];
    !digits.is_empty() && digits.bytes().all(|byte| byte.is_ascii_digit())
}

/// Case-5 doctor finding codes copied into the migrate report (deprecated
/// leftovers; report-only).
const LEFTOVER_FINDING_CODES: &[&str] = &[
    storage_doctor::FINDING_LEGACY_FS_BLOBS,
    storage_doctor::FINDING_BACKUP_ARTIFACT,
    storage_doctor::FINDING_QUARANTINE_ARTIFACT,
    storage_doctor::FINDING_WORKGRAPH_ADMISSION_SIDECAR,
    storage_doctor::FINDING_MAINTENANCE_FENCE_LOCK,
];

// ─────────────────────────────────────────────────────────────────────────
// The maintenance fence (composed from meerkat's per-file primitive).
// ─────────────────────────────────────────────────────────────────────────

/// Enumerate every SQLite database file currently materialized under a
/// MobKit state directory: each [`DatabaseSlot`]'s canonical **and** legacy
/// spellings that exist as files, the inherited canonical jobs database, plus
/// the per-realm agent-memory databases under both memory-root spellings.
/// Sorted (deterministic fence order).
///
/// The workgraph admission sidecar is deliberately excluded: it carries no
/// per-operation fence guard by M3 design (it IS a cross-process lock), so
/// fencing it gates nothing, and migrate never mutates it.
pub fn enumerate_state_dir_databases(state_dir: &Path) -> Vec<PathBuf> {
    let mut files: Vec<PathBuf> = Vec::new();
    for slot in DatabaseSlot::ALL {
        if slot == DatabaseSlot::AgentMemory {
            continue; // a directory slot; realm files are enumerated below
        }
        let mut names: Vec<&str> = vec![slot.canonical_name()];
        names.extend(slot.legacy_names());
        for name in names {
            let path = state_dir.join(name);
            if path.is_file() {
                files.push(path);
            }
        }
    }
    for spelling in MEMORY_ROOT_SPELLINGS {
        let root = state_dir.join(spelling);
        let Ok(entries) = fs::read_dir(&root) else {
            continue;
        };
        for entry in entries.filter_map(Result::ok) {
            let path = entry.path();
            if path.is_file() && path.extension().and_then(|ext| ext.to_str()) == Some("sqlite3") {
                files.push(path);
            }
        }
    }
    let jobs = MobKitStorageLayout::with_injected_roots(state_dir.to_path_buf(), None).jobs_db();
    if jobs.is_file() {
        files.push(jobs);
    }
    files.sort();
    files.dedup();
    files
}

/// The state-dir-wide exclusive maintenance fence: one
/// [`meerkat_sqlite::ExclusiveFence`] per database from
/// [`enumerate_state_dir_databases`], acquired in sorted order (two
/// concurrent migrators can never ABBA-deadlock), all-or-nothing — any
/// failure releases everything already acquired (RAII). While held, foreign
/// processes' per-operation guards fail typed
/// ([`meerkat_sqlite::SqliteStoreError::MaintenanceFenceHeld`]); this
/// process's own store operations self-admit, which is what lets the
/// migrate pass reuse production store constructors.
#[derive(Debug)]
pub struct MobKitMaintenanceFence {
    fences: Vec<meerkat_sqlite::ExclusiveFence>,
    databases: Vec<PathBuf>,
}

impl MobKitMaintenanceFence {
    /// Fence every materialized database under `state_dir`, waiting up to
    /// `deadline` (total, across all files) for in-flight operations to
    /// drain. Blocking; async callers should wrap it in `spawn_blocking`.
    pub fn acquire(
        state_dir: &Path,
        deadline: Duration,
    ) -> Result<Self, meerkat_sqlite::SqliteStoreError> {
        let started = std::time::Instant::now();
        let databases = enumerate_state_dir_databases(state_dir);
        let mut fences = Vec::with_capacity(databases.len());
        for database in &databases {
            let remaining = deadline.saturating_sub(started.elapsed());
            // Dropping `fences` on error releases everything acquired.
            let fence = meerkat_sqlite::ExclusiveFence::acquire(database, remaining)?;
            fences.push(fence);
        }
        Ok(Self { fences, databases })
    }

    /// The database files this fence covers, in acquisition (sorted) order.
    pub fn fenced_databases(&self) -> &[PathBuf] {
        &self.databases
    }

    /// Number of held per-file fences.
    pub fn len(&self) -> usize {
        self.fences.len()
    }

    /// True when the state dir materializes no databases (nothing to fence).
    pub fn is_empty(&self) -> bool {
        self.fences.is_empty()
    }

    /// Re-fence after a rename: the original fence rides the OLD lock path,
    /// so the file's NEW canonical path must get a fresh fence for the rest
    /// of the pass. Failure is fail-closed for the renamed database — the
    /// caller records the typed error and skips further mutation of it,
    /// never continuing unfenced.
    fn cover_renamed(&mut self, path: &Path) -> Result<(), meerkat_sqlite::SqliteStoreError> {
        // Canonical-name-first fencing may have covered the canonical path
        // up front (twin adoption renames INTO a path this fence already
        // holds); custody is already ours, and a second exclusive fence on
        // it would self-refuse.
        let canonical = std::fs::canonicalize(path).unwrap_or_else(|_| path.to_path_buf());
        if self
            .databases
            .iter()
            .any(|held| std::fs::canonicalize(held).unwrap_or_else(|_| held.clone()) == canonical)
        {
            return Ok(());
        }
        match meerkat_sqlite::ExclusiveFence::try_acquire(path)? {
            Some(fence) => {
                self.fences.push(fence);
                self.databases.push(path.to_path_buf());
                Ok(())
            }
            None => Err(meerkat_sqlite::SqliteStoreError::MaintenanceFenceHeld {
                path: path.to_path_buf(),
            }),
        }
    }
}

// ─────────────────────────────────────────────────────────────────────────
// Report vocabulary (serde; `#[non_exhaustive]` + defaults, mirroring
// `meerkat_store::migrate`'s shapes).
// ─────────────────────────────────────────────────────────────────────────

/// The full `storage-migrate` report for one state directory.
#[non_exhaustive]
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
pub struct MobKitMigrateReport {
    /// Dry-run or apply.
    #[serde(default)]
    pub mode: MigrateMode,
    /// The state directory swept.
    #[serde(default)]
    pub state_dir: PathBuf,
    /// Databases the fence covers (in apply mode, the fence actually held;
    /// in dry-run, what an apply run would fence).
    #[serde(default)]
    pub fenced_databases: Vec<PathBuf>,
    /// Ledger baseline entries per database × domain (case 1).
    #[serde(default)]
    pub ledger: Vec<LedgerBaselineEntry>,
    /// File-name unification renames (case 2, plus case-3 canonicalization
    /// renames — every physical rename this run performed or would perform).
    #[serde(default)]
    pub renames: Vec<FileRenameEntry>,
    /// File-name twins and their resolution (case 3).
    #[serde(default)]
    pub twins: Vec<TwinReport>,
    /// Continuity checkpoint-evidence adoption outcome (case 4).
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub adoption: Option<CheckpointAdoptionOutcome>,
    /// Digest-format marker stamping per session-document store (case 6):
    /// verified-but-marker-less documents (written by pre-marker builds)
    /// are respelled in place so decode-time heal probes stop re-running
    /// on every load. One outcome per store (continuity, runtime,
    /// sessions).
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub marker_stamping: Vec<MarkerStampStoreOutcome>,
    /// Deprecated-leftover findings (case 5; report-only), reusing the
    /// doctor finding vocabulary.
    #[serde(default)]
    pub findings: Vec<StorageFinding>,
    /// Human-readable notes (exemptions, report-only carve-outs).
    #[serde(default)]
    pub notes: Vec<String>,
    /// Fail-closed refusals and hard failures. Non-empty ⇒ nonzero exit.
    #[serde(default)]
    pub errors: Vec<String>,
}

impl MobKitMigrateReport {
    /// New empty report for one run.
    pub fn new(mode: MigrateMode, state_dir: &Path) -> Self {
        Self {
            mode,
            state_dir: state_dir.to_path_buf(),
            ..Self::default()
        }
    }

    /// True when the run must exit nonzero (refusals, fence failures,
    /// constructor failures, adoption refusals).
    pub fn has_errors(&self) -> bool {
        !self.errors.is_empty()
    }
}

/// One database × ledger-domain baseline row (case 1).
#[non_exhaustive]
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct LedgerBaselineEntry {
    /// Database file.
    pub database: PathBuf,
    /// Ledger domain (`mobkit-continuity`, `session-store`, ...).
    pub domain: String,
    /// Version before the run (`None` = no ledger row).
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub before: Option<i64>,
    /// Version after the run (`None` in dry-run and for report-only rows).
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub after: Option<i64>,
    /// What happened (or would happen).
    pub action: LedgerBaselineAction,
}

/// Disposition of one ledger baseline row.
#[non_exhaustive]
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "kebab-case")]
pub enum LedgerBaselineAction {
    /// Dry-run: no ledger row; the owning store baseline-stamps on
    /// `--apply` (guarded migrations converge files of any vintage).
    WouldStamp,
    /// Dry-run: a ledger row exists; pending migrations converge on
    /// `--apply` through the owning store's normal constructor.
    Recorded,
    /// Apply: the ledger row was created or advanced.
    Stamped,
    /// Apply: already at the current version; no-op.
    AlreadyCurrent,
    /// A meerkat-owned database (sessions / runtime / schedule /
    /// workgraph): versions are reported read-only; the owning meerkat
    /// store converges the file on its next open.
    ReportOnly,
    /// The workgraph admission sidecar: ledger-exempt by design (M3) — the
    /// lock database deliberately carries no tables and no ledger.
    Exempt,
}

/// One physical rename (case 2, or a case-3 canonicalization).
#[non_exhaustive]
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct FileRenameEntry {
    /// The layout slot this rename converges.
    pub slot: String,
    /// Legacy path.
    pub from: PathBuf,
    /// Canonical path.
    pub to: PathBuf,
    /// `-wal` / `-shm` sibling files moved along with the database.
    #[serde(default)]
    pub siblings: Vec<SiblingRename>,
    /// True when a non-empty WAL was checkpointed before the move.
    #[serde(default)]
    pub wal_checkpointed: bool,
    /// The registered rename marker (`<legacy>.pre-<version>-<ts>.renamed`),
    /// written on apply so doctor and prune see the rename.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub marker: Option<PathBuf>,
    /// What happened (or would happen).
    pub action: RenameAction,
}

/// One `-wal` / `-shm` sibling moved with its database.
#[non_exhaustive]
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct SiblingRename {
    pub from: PathBuf,
    pub to: PathBuf,
}

/// Disposition of one rename.
#[non_exhaustive]
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "kebab-case")]
pub enum RenameAction {
    /// Dry-run: a lone legacy spelling; `--apply` renames it.
    WouldRename,
    /// Apply: renamed (siblings and marker recorded alongside).
    Renamed,
    /// Apply: refused (reason in [`MobKitMigrateReport::errors`]) — for
    /// example a WAL that cannot be checkpointed.
    Refused,
}

/// Case 3: one file-name twin (two or more spellings of one slot populated
/// in the same state directory).
#[non_exhaustive]
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct TwinReport {
    /// The layout slot.
    pub slot: String,
    /// Every populated spelling (canonical first when present).
    pub paths: Vec<PathBuf>,
    /// True when every copy is byte-identical (database bytes plus any WAL
    /// sidecar) — the exact-equality dedup precondition.
    #[serde(default)]
    pub byte_identical: bool,
    /// Rows identical across every copy (count only; not enumerated).
    #[serde(default)]
    pub rows_equal: usize,
    /// Divergent / single-copy rows. For continuity / metadata / console
    /// the key is `table/<primary key>`; for the agent-memory directory
    /// twin the key is the relative realm-database name.
    #[serde(default)]
    pub rows: Vec<RowDivergenceEntry>,
    /// How the twin was (or was not) resolved.
    pub resolution: TwinResolution,
    /// Per-slot notes (the no-synthesis rationale, digest granularity).
    #[serde(default)]
    pub notes: Vec<String>,
    /// Divergence-computation failures (report-only; resolution is
    /// non-destructive either way).
    #[serde(default)]
    pub errors: Vec<String>,
}

/// One non-equal row (or file) across twin copies.
#[non_exhaustive]
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct RowDivergenceEntry {
    /// `table/<primary key>` (row-level domains) or a relative file name.
    pub key: String,
    /// Divergence classification.
    pub status: DivergenceStatus,
}

/// Case 3 resolution.
#[non_exhaustive]
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "snake_case", tag = "kind")]
pub enum TwinResolution {
    /// Fail-closed refusal: nothing moved.
    Refused {
        /// Why the run refused.
        reason: String,
    },
    /// Exact-equality dedup: the copies were byte-identical; the canonical
    /// spelling was kept (renaming it into place when needed) and every
    /// redundant copy archived read-only.
    Deduped {
        /// The surviving copy (at its canonical path).
        kept: PathBuf,
        /// Archive paths of the redundant copies.
        archived: Vec<PathBuf>,
    },
    /// One copy adopted as authority (at its canonical path); every other
    /// copy archived read-only under the registered backup naming. No
    /// synthesis — divergent content is preserved in the archives.
    Adopted {
        /// The adopted copy (at its canonical path).
        adopted: PathBuf,
        /// Archive paths of the non-adopted copies.
        archived: Vec<PathBuf>,
    },
}

/// Case 4 outcome: the merged H3 adoption walk (or why it did not run).
#[non_exhaustive]
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct CheckpointAdoptionOutcome {
    /// The continuity database the walk covered (canonical-resolved).
    pub database: PathBuf,
    /// The H3 walk report (`None` when skipped).
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub report: Option<ContinuityAdoptionReport>,
    /// Set when adoption did not run (no continuity database).
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub skipped: Option<String>,
}

/// Case 6 outcome: the digest-format marker-stamping walk over one
/// session-document store (or why it did not run).
#[non_exhaustive]
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct MarkerStampStoreOutcome {
    /// Store label (`continuity`, `runtime`, `sessions`).
    pub store: String,
    /// The database the walk covered (canonical-resolved).
    pub database: PathBuf,
    /// The walk report (`None` when skipped).
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub report: Option<SessionMarkerStampReport>,
    /// Set when the walk did not run (no database, wrong schema, unfenced
    /// after a rename).
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub skipped: Option<String>,
}

/// The full `storage-prune` report for one state directory. Shape mirrors
/// `meerkat_store::migrate::PruneReport` (mobkit re-declares the structs
/// because meerkat's are `#[non_exhaustive]` and cannot be constructed
/// here; the enums are shared).
#[non_exhaustive]
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
pub struct MobKitPruneReport {
    /// Dry-run or apply.
    #[serde(default)]
    pub mode: MigrateMode,
    /// The state directory swept.
    #[serde(default)]
    pub state_dir: PathBuf,
    /// Age threshold in days (artifacts at least this old are deleted on
    /// apply; `0` = all).
    #[serde(default)]
    pub older_than_days: u64,
    /// Registered artifacts found, with dispositions.
    #[serde(default)]
    pub artifacts: Vec<MobKitPruneArtifact>,
    /// Failures (delete errors). Non-empty ⇒ nonzero exit.
    #[serde(default)]
    pub errors: Vec<String>,
}

impl MobKitPruneReport {
    /// New empty report for one run.
    pub fn new(mode: MigrateMode, state_dir: &Path, older_than_days: u64) -> Self {
        Self {
            mode,
            state_dir: state_dir.to_path_buf(),
            older_than_days,
            ..Self::default()
        }
    }

    /// True when the run must exit nonzero.
    pub fn has_errors(&self) -> bool {
        !self.errors.is_empty()
    }
}

/// One registered maintenance artifact.
#[non_exhaustive]
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct MobKitPruneArtifact {
    /// Artifact path (file or directory).
    pub path: PathBuf,
    /// Which registered naming pattern matched.
    pub kind: PruneArtifactKind,
    /// Total size in bytes (recursive for directories).
    #[serde(default)]
    pub bytes: u64,
    /// Age in whole days (from mtime).
    #[serde(default)]
    pub age_days: u64,
    /// Disposition.
    pub action: PruneAction,
}

// ─────────────────────────────────────────────────────────────────────────
// The migrate pass.
// ─────────────────────────────────────────────────────────────────────────

/// Run the five-case migration pass over one MobKit state directory.
/// Blocking (async callers should wrap it in `spawn_blocking`); the caller
/// renders the report and maps [`MobKitMigrateReport::has_errors`] to a
/// nonzero exit.
///
/// `adopt` is the case-3 twin resolution: the twin copy to adopt as
/// authority (only meaningful with [`MigrateMode::Apply`]).
pub fn migrate_state_dir(
    state_dir: &Path,
    mode: MigrateMode,
    adopt: Option<&Path>,
) -> MobKitMigrateReport {
    let mut report = MobKitMigrateReport::new(mode, state_dir);
    if !state_dir.is_dir() {
        report.errors.push(format!(
            "state directory {} does not exist",
            state_dir.display()
        ));
        return report;
    }
    let apply = mode == MigrateMode::Apply;
    let layout = MobKitStorageLayout::with_injected_roots(state_dir.to_path_buf(), None);

    // Exclusive maintenance fence over every materialized database (apply
    // only; dry-run is read-only and reports what an apply run would fence).
    let mut fence = if apply {
        match MobKitMaintenanceFence::acquire(state_dir, FENCE_DRAIN_DEADLINE) {
            Ok(fence) => {
                report.fenced_databases = fence.fenced_databases().to_vec();
                Some(fence)
            }
            Err(error) => {
                report
                    .errors
                    .push(format!("maintenance fence not acquirable: {error}"));
                return report;
            }
        }
    } else {
        report.fenced_databases = enumerate_state_dir_databases(state_dir);
        None
    };

    // Canonical paths whose post-rename fence could not be re-acquired:
    // fail-closed — the fence failure is recorded as an error and the
    // remaining cases never mutate these databases.
    let mut unfenced: Vec<PathBuf> = Vec::new();

    // ── Case 3: twin reconciliation (fail-closed). ───────────────────────
    let mut unresolved_twin = false;
    for slot in DatabaseSlot::ALL {
        let Err(StorageLayoutError::FileNameTwins { paths, .. }) = layout.resolve_database(slot)
        else {
            continue;
        };
        let twin = reconcile_twin(
            slot,
            &paths,
            mode,
            adopt,
            &mut report.renames,
            fence.as_mut(),
            &mut report.errors,
            &mut unfenced,
        );
        if let TwinResolution::Refused { reason } = &twin.resolution {
            unresolved_twin = true;
            report
                .errors
                .push(format!("file-name twins for the {slot} store: {reason}"));
        }
        report.twins.push(twin);
    }
    if unresolved_twin {
        // Fail-closed: the divergence report is the whole output — every
        // remaining case would have to pick a spelling.
        return report;
    }

    // ── Case 2: lone-legacy-spelling renames (auto-safe, rename-only). ───
    for slot in DatabaseSlot::ALL {
        let Ok(resolved) = layout.resolve_database(slot) else {
            continue; // twins were resolved (or absent) above
        };
        let DatabaseProvenance::LegacySpelling(_) = resolved.provenance else {
            continue;
        };
        let to = state_dir.join(slot.canonical_name());
        let entry = rename_to_canonical(slot, &resolved.path, &to, apply, &mut report.errors);
        if entry.action == RenameAction::Renamed
            && slot != DatabaseSlot::AgentMemory
            && let Some(fence) = fence.as_mut()
            && let Err(error) = fence.cover_renamed(&to)
        {
            report.errors.push(format!(
                "renamed {slot} store is not fenced at its canonical path {}: {error}; \
                 skipping further mutation of this database",
                to.display()
            ));
            unfenced.push(to.clone());
        }
        report.renames.push(entry);
    }

    // ── Case 1: ledger baseline. ─────────────────────────────────────────
    let before = read_ledger_matrix(state_dir);
    if apply {
        open_stores_through_ledgered_constructors(&layout, &unfenced, &mut report.errors);
        let after = read_ledger_matrix(state_dir);
        let before_of = |database: &Path, domain: &str| -> Option<i64> {
            before
                .iter()
                .find(|(db, dom, _)| db == database && dom == domain)
                .and_then(|(_, _, version)| *version)
        };
        for (database, domain, after_version) in after {
            let class = ledger_domain_class(&domain);
            let before_version = before_of(&database, &domain);
            let (action, after_field) = match class {
                LedgerDomainClass::Stampable => {
                    if after_version == before_version {
                        (LedgerBaselineAction::AlreadyCurrent, after_version)
                    } else {
                        (LedgerBaselineAction::Stamped, after_version)
                    }
                }
                LedgerDomainClass::Exempt => (LedgerBaselineAction::Exempt, None),
                LedgerDomainClass::ReportOnly => (LedgerBaselineAction::ReportOnly, None),
            };
            report.ledger.push(LedgerBaselineEntry {
                database,
                domain,
                before: before_version,
                after: after_field,
                action,
            });
        }
    } else {
        for (database, domain, version) in before {
            let action = match ledger_domain_class(&domain) {
                LedgerDomainClass::Stampable => {
                    if version.is_some() {
                        LedgerBaselineAction::Recorded
                    } else {
                        LedgerBaselineAction::WouldStamp
                    }
                }
                LedgerDomainClass::Exempt => LedgerBaselineAction::Exempt,
                LedgerDomainClass::ReportOnly => LedgerBaselineAction::ReportOnly,
            };
            report.ledger.push(LedgerBaselineEntry {
                database,
                domain,
                before: version,
                after: None,
                action,
            });
        }
    }
    // The head-canonical continuity bump is the one migration a gateway open
    // deliberately never commits (it locks previous releases out of the
    // file), so the operator has to see it coming.
    if report.ledger.iter().any(|entry| {
        entry.domain == "mobkit-continuity"
            && entry.before.is_some_and(|version| {
                version < crate::identity_first::HEAD_CANONICAL_CONTINUITY_SCHEMA_VERSION
            })
    }) {
        report.notes.push(format!(
            "continuity database is below the head-canonical schema version \
             (mobkit-continuity v{}); {} — this bump is ONE-WAY: binaries older than this \
             release refuse a stamped file at open (SchemaFromTheFuture). Back up continuity.* \
             before applying.",
            crate::identity_first::HEAD_CANONICAL_CONTINUITY_SCHEMA_VERSION,
            if apply {
                "--apply committed it under the exclusive maintenance fence"
            } else {
                "--apply would commit it under the exclusive maintenance fence, and the first \
                 incremental session write would otherwise commit it lazily"
            }
        ));
    }
    if state_dir.join(WORKGRAPH_ADMISSION_SIDECAR_FILE).is_file() {
        report.notes.push(
            "workgraph admission sidecar is ledger-exempt by design (M3): the lock database \
             deliberately carries no tables; stamping a ledger row on open would contend the \
             cross-process admission lock"
                .to_string(),
        );
    }
    if report
        .ledger
        .iter()
        .any(|entry| entry.action == LedgerBaselineAction::ReportOnly)
    {
        report.notes.push(
            "sessions / runtime / schedule / workgraph are meerkat-owned stores: their ledgers \
             are reported read-only here and converge through the owning meerkat store's next \
             open (normal gateway boot)"
                .to_string(),
        );
    }

    // ── Case 4: continuity checkpoint-evidence adoption (H3 machinery). ──
    run_checkpoint_adoption(&layout, mode, &unfenced, &mut report);

    // ── Case 6: digest-format marker stamping over every session-document
    // store (runs after adoption so freshly adopted rows — which adoption's
    // re-serialization already marker-stamps — classify already-current
    // here, and remaining legacy rows stay adoption's population). ────────
    run_marker_stamping(&layout, mode, &unfenced, &mut report);

    // ── Case 5: deprecated leftovers (report-only). ──────────────────────
    let scope = DiagnoseScope::new(vec![state_dir.to_path_buf()]);
    let diagnosis = storage_doctor::diagnose_state_dir_blocking(&scope, None);
    report.findings.extend(
        diagnosis
            .findings
            .into_iter()
            .filter(|finding| LEFTOVER_FINDING_CODES.contains(&finding.code.as_str())),
    );
    sweep_runtime_registry(state_dir, &mut report.findings, &mut report.notes);

    drop(fence);
    report
}

/// Ledger-domain classes for case 1.
enum LedgerDomainClass {
    /// A MobKit-owned domain whose M3 constructor this verb runs.
    Stampable,
    /// The admission sidecar (no ledger by design).
    Exempt,
    /// A meerkat-owned domain (reported read-only).
    ReportOnly,
}

fn ledger_domain_class(domain: &str) -> LedgerDomainClass {
    match domain {
        "mobkit-continuity" | "mobkit-metadata" | "mobkit-console" | "jobs" => {
            LedgerDomainClass::Stampable
        }
        "mobkit-workgraph-admission" => LedgerDomainClass::Exempt,
        _ if domain == MEMORY_LEDGER_DOMAIN => LedgerDomainClass::Stampable,
        _ => LedgerDomainClass::ReportOnly,
    }
}

/// Read the schema-ledger rows of one database, read-only. `Ok(None)` = no
/// `meerkat_schema` table (pre-ledger file).
fn read_domain_versions(db_path: &Path) -> Result<Option<Vec<(String, i64)>>, String> {
    let conn = meerkat_sqlite::open(db_path, meerkat_sqlite::ConnectionProfile::ReadOnly)
        .map_err(|error| error.to_string())?;
    if !table_exists(&conn, "meerkat_schema").map_err(|error| error.to_string())? {
        return Ok(None);
    }
    let result = (|| -> Result<Vec<(String, i64)>, rusqlite::Error> {
        let mut statement =
            conn.prepare("SELECT domain, version FROM meerkat_schema ORDER BY domain")?;
        let rows = statement
            .query_map([], |row| {
                Ok((row.get::<_, String>(0)?, row.get::<_, i64>(1)?))
            })?
            .collect::<Result<Vec<_>, _>>()?;
        Ok(rows)
    })();
    result.map(Some).map_err(|error| error.to_string())
}

fn table_exists(conn: &rusqlite::Connection, table: &str) -> Result<bool, rusqlite::Error> {
    use rusqlite::OptionalExtension;
    Ok(conn
        .query_row(
            "SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = ?1",
            [table],
            |_| Ok(()),
        )
        .optional()?
        .is_some())
}

/// Ledger versions for every inventoried database currently on disk:
/// `(database, domain, version)` for ledger rows, `None` for expected
/// domains without a row. Read-only.
fn read_ledger_matrix(state_dir: &Path) -> Vec<(PathBuf, String, Option<i64>)> {
    let mut matrix = Vec::new();
    let mut push_db = |db_path: PathBuf, expected: &[&str]| {
        let rows = read_domain_versions(&db_path)
            .ok()
            .flatten()
            .unwrap_or_default();
        for (domain, version) in &rows {
            matrix.push((db_path.clone(), domain.clone(), Some(*version)));
        }
        for domain in expected {
            if !rows.iter().any(|(name, _)| name == domain) {
                matrix.push((db_path.clone(), (*domain).to_string(), None));
            }
        }
    };
    for family in DATABASE_FAMILIES {
        for spelling in family.spellings {
            let db_path = state_dir.join(spelling);
            if db_path.is_file() {
                push_db(db_path, family.ledger_domains);
            }
        }
    }
    for spelling in MEMORY_ROOT_SPELLINGS {
        let root = state_dir.join(spelling);
        let Ok(entries) = fs::read_dir(&root) else {
            continue;
        };
        let mut realm_dbs: Vec<PathBuf> = entries
            .filter_map(Result::ok)
            .map(|entry| entry.path())
            .filter(|path| {
                path.is_file() && path.extension().and_then(|ext| ext.to_str()) == Some("sqlite3")
            })
            .collect();
        realm_dbs.sort();
        for db_path in realm_dbs {
            push_db(db_path, &[MEMORY_LEDGER_DOMAIN]);
        }
    }
    let jobs = MobKitStorageLayout::with_injected_roots(state_dir.to_path_buf(), None).jobs_db();
    if jobs.is_file() {
        push_db(jobs, &["jobs"]);
    }
    matrix
}

/// Case 1 apply: run each existing MobKit-owned database through its normal
/// M3 constructor. The guarded ledger migrations are the structural
/// verification; a constructor that fails is a per-store error, never an
/// abort of the pass. Only files that already exist are opened — migrate
/// converges existing state, it never materializes new stores. Databases in
/// `unfenced` (canonical fence lost after a rename) are skipped fail-closed;
/// the fence failure is already recorded as an error.
fn open_stores_through_ledgered_constructors(
    layout: &MobKitStorageLayout,
    unfenced: &[PathBuf],
    errors: &mut Vec<String>,
) {
    match layout.continuity_db() {
        Ok(resolved) if resolved.path.is_file() && !unfenced.contains(&resolved.path) => {
            if let Err(error) = LocalContinuityStore::open(&resolved.path) {
                errors.push(format!("continuity store open failed: {error}"));
            } else if let Err(error) =
                LocalContinuityStore::apply_head_canonical_schema_at(&resolved.path)
            {
                // The head-canonical (`mobkit-continuity` v2) bump is the one
                // migration a plain gateway open deliberately does NOT
                // commit: it locks previous releases out of the file, so it
                // is committed only here, under the exclusive maintenance
                // fence, as an explicit operator action — or lazily, in the
                // transaction of a delta write that actually creates a head
                // row (a refused write leaves the file at v1).
                errors.push(format!(
                    "continuity head-canonical schema migration failed: {error}"
                ));
            }
        }
        Ok(_) => {}
        Err(error) => errors.push(format!("continuity locator unresolved: {error}")),
    }
    match layout.metadata_db() {
        Ok(resolved) if resolved.path.is_file() && !unfenced.contains(&resolved.path) => {
            if let Err(error) = SqliteMetadataStore::open(&resolved.path) {
                errors.push(format!("metadata store open failed: {error}"));
            }
        }
        Ok(_) => {}
        Err(error) => errors.push(format!("metadata locator unresolved: {error}")),
    }
    match layout.console_db() {
        Ok(resolved) if resolved.path.is_file() && !unfenced.contains(&resolved.path) => {
            if let Err(error) = SqliteConsoleLogStore::open(&resolved.path) {
                errors.push(format!("console store open failed: {error}"));
            }
        }
        Ok(_) => {}
        Err(error) => errors.push(format!("console locator unresolved: {error}")),
    }
    match layout.agent_memory_root() {
        Ok(resolved) if resolved.path.is_dir() => {
            match SqliteAgentMemoryStore::open(&resolved.path) {
                Ok(store) => match store.known_realms() {
                    Ok(realms) => {
                        for realm in realms {
                            if let Err(error) = store.open_realm_ledgered(&realm) {
                                errors.push(format!(
                                    "agent-memory realm '{realm}' open failed: {error}"
                                ));
                            }
                        }
                    }
                    Err(error) => {
                        errors.push(format!("agent-memory realm listing failed: {error}"));
                    }
                },
                Err(error) => errors.push(format!("agent-memory store open failed: {error}")),
            }
        }
        Ok(_) => {}
        Err(error) => errors.push(format!("agent-memory locator unresolved: {error}")),
    }
    let jobs = layout.jobs_db();
    if jobs.is_file()
        && !unfenced.contains(&jobs)
        && let Err(error) = meerkat::SqliteDetachedJobStore::open(&jobs)
    {
        errors.push(format!("detached-job store open failed: {error}"));
    }
}

/// Case 4: the H3 adoption walk over the canonical-resolved continuity
/// database. Apply composes with the already-held pass fence; dry-run takes
/// H3's own short-lived fence (nothing else is held). A continuity database
/// in `unfenced` (canonical fence lost after a rename) is skipped
/// fail-closed; the fence failure is already recorded as an error.
fn run_checkpoint_adoption(
    layout: &MobKitStorageLayout,
    mode: MigrateMode,
    unfenced: &[PathBuf],
    report: &mut MobKitMigrateReport,
) {
    let resolved = match layout.continuity_db() {
        Ok(resolved) => resolved,
        Err(error) => {
            report.errors.push(format!(
                "continuity locator unresolved for adoption: {error}"
            ));
            return;
        }
    };
    if unfenced.contains(&resolved.path) {
        report.adoption = Some(CheckpointAdoptionOutcome {
            database: resolved.path,
            report: None,
            skipped: Some(
                "continuity database is not fenced at its canonical path after rename; \
                 adoption skipped (fail-closed)"
                    .to_string(),
            ),
        });
        return;
    }
    if !resolved.path.is_file() {
        report.adoption = Some(CheckpointAdoptionOutcome {
            database: resolved.path,
            report: None,
            skipped: Some("no continuity database materialized; nothing to adopt".to_string()),
        });
        return;
    }
    let result = match mode {
        MigrateMode::Apply => {
            adopt_continuity_snapshots_already_fenced(&resolved.path, AdoptionMode::Apply)
        }
        _ => adopt_continuity_snapshots_blocking(&resolved.path, AdoptionMode::DryRun),
    };
    match result {
        Ok(walk) => {
            if !walk.is_clean() {
                report.errors.push(format!(
                    "{} continuity snapshot row(s) refused checkpoint adoption; see the \
                     adoption report",
                    walk.refused.len()
                ));
            }
            report.adoption = Some(CheckpointAdoptionOutcome {
                database: resolved.path,
                report: Some(walk),
                skipped: None,
            });
        }
        Err(error) => report
            .errors
            .push(format!("continuity checkpoint adoption failed: {error}")),
    }
}

/// Case 6: digest-format marker stamping over every session-document store
/// (continuity `session_snapshots`, meerkat runtime
/// `runtime_session_snapshots`, meerkat legacy headless `sessions` rows).
/// Same fence discipline as case 4: apply composes with the already-held
/// pass fence; dry-run takes the walk's own short-lived fence and read-only
/// connection. A database in `unfenced` (canonical fence lost after a
/// rename) is skipped fail-closed. Refusals surface in
/// [`MobKitMigrateReport::errors`] ⇒ nonzero exit.
fn run_marker_stamping(
    layout: &MobKitStorageLayout,
    mode: MigrateMode,
    unfenced: &[PathBuf],
    report: &mut MobKitMigrateReport,
) {
    let stamp_mode = match mode {
        MigrateMode::Apply => MarkerStampMode::Apply,
        _ => MarkerStampMode::DryRun,
    };
    for (slot, store) in [
        (DatabaseSlot::Continuity, SessionDocumentStore::Continuity),
        (
            DatabaseSlot::Runtime,
            SessionDocumentStore::RuntimeSnapshots,
        ),
        (DatabaseSlot::Sessions, SessionDocumentStore::SessionStore),
    ] {
        let resolved = match layout.resolve_database(slot) {
            Ok(resolved) => resolved,
            Err(error) => {
                report.errors.push(format!(
                    "{} locator unresolved for marker stamping: {error}",
                    store.label()
                ));
                continue;
            }
        };
        if unfenced.contains(&resolved.path) {
            report.marker_stamping.push(MarkerStampStoreOutcome {
                store: store.label().to_string(),
                database: resolved.path,
                report: None,
                skipped: Some(
                    "database is not fenced at its canonical path after rename; \
                     marker stamping skipped (fail-closed)"
                        .to_string(),
                ),
            });
            continue;
        }
        if !resolved.path.is_file() {
            report.marker_stamping.push(MarkerStampStoreOutcome {
                store: store.label().to_string(),
                database: resolved.path,
                report: None,
                skipped: Some("no database materialized; nothing to stamp".to_string()),
            });
            continue;
        }
        let result = match stamp_mode {
            MarkerStampMode::Apply => {
                stamp_session_document_markers_already_fenced(&resolved.path, store, stamp_mode)
            }
            MarkerStampMode::DryRun => {
                stamp_session_document_markers_blocking(&resolved.path, store, stamp_mode)
            }
        };
        match result {
            Ok(walk) => {
                if !walk.is_clean() {
                    report.errors.push(format!(
                        "{} session document row(s) in the {} store refused digest-format \
                         marker stamping; see the marker-stamping report",
                        walk.refused.len(),
                        store.label()
                    ));
                }
                report.marker_stamping.push(MarkerStampStoreOutcome {
                    store: store.label().to_string(),
                    database: resolved.path,
                    report: Some(walk),
                    skipped: None,
                });
            }
            // A file without this store's table is not this store (for
            // example a runtime.sqlite created before runtime snapshots
            // existed): skipped, never an error.
            Err(MarkerStampError::MissingTable { table, .. }) => {
                report.marker_stamping.push(MarkerStampStoreOutcome {
                    store: store.label().to_string(),
                    database: resolved.path,
                    report: None,
                    skipped: Some(format!(
                        "database carries no {table} table; not this session-document store"
                    )),
                });
            }
            Err(error) => report.errors.push(format!(
                "{} digest-format marker stamping failed: {error}",
                store.label()
            )),
        }
    }
}

// ─────────────────────────────────────────────────────────────────────────
// Case 2 mechanics: WAL-safe database moves + registered rename markers.
// ─────────────────────────────────────────────────────────────────────────

fn path_with_suffix(path: &Path, suffix: &str) -> PathBuf {
    let mut os = path.as_os_str().to_os_string();
    os.push(suffix);
    PathBuf::from(os)
}

/// Checkpoint a non-empty WAL (`TRUNCATE`) through a maintenance
/// connection so a subsequent move carries no live frames. Returns whether
/// a checkpoint ran. Refuses (Err) when the WAL cannot be fully
/// checkpointed — moving a database away from live WAL frames would lose
/// them.
fn checkpoint_wal_if_nonempty(db: &Path) -> Result<bool, String> {
    let wal = path_with_suffix(db, "-wal");
    let wal_len = fs::metadata(&wal).map(|meta| meta.len()).unwrap_or(0);
    if wal_len == 0 {
        return Ok(false);
    }
    let conn = meerkat_sqlite::open(
        db,
        meerkat_sqlite::ConnectionProfile::Maintenance { write: true },
    )
    .map_err(|error| format!("open for WAL checkpoint failed: {error}"))?;
    let busy: i64 = conn
        .query_row("PRAGMA wal_checkpoint(TRUNCATE)", [], |row| row.get(0))
        .map_err(|error| format!("wal_checkpoint failed: {error}"))?;
    if busy != 0 {
        return Err(format!(
            "WAL for {} cannot be checkpointed (readers active); refusing to move it",
            db.display()
        ));
    }
    Ok(true)
}

/// Move one database file to a new name, WAL-safely: checkpoint a non-empty
/// WAL first (refusing when impossible), rename the database, then move any
/// remaining `-wal` / `-shm` siblings so they stay paired with their
/// database. Returns the sibling moves and whether a checkpoint ran.
fn move_database_file(from: &Path, to: &Path) -> Result<(Vec<SiblingRename>, bool), String> {
    if to.exists() {
        return Err(format!(
            "cannot move {} to {}: target already exists",
            from.display(),
            to.display()
        ));
    }
    let checkpointed = checkpoint_wal_if_nonempty(from)?;
    fs::rename(from, to)
        .map_err(|error| format!("rename {} -> {}: {error}", from.display(), to.display()))?;
    let mut siblings = Vec::new();
    for suffix in ["-wal", "-shm"] {
        let src = path_with_suffix(from, suffix);
        if !src.exists() {
            continue;
        }
        let dst = path_with_suffix(to, suffix);
        fs::rename(&src, &dst)
            .map_err(|error| format!("rename {} -> {}: {error}", src.display(), dst.display()))?;
        siblings.push(SiblingRename { from: src, to: dst });
    }
    Ok((siblings, checkpointed))
}

/// Write the registered rename marker next to where the legacy file was:
/// `<legacy name>.pre-<version>-<ts>.renamed`, holding a small JSON record.
/// The `.pre-` segment makes it a registered backup artifact — doctor lists
/// it, prune owns its lifecycle, retention tooling recognizes it.
fn write_rename_marker(from: &Path, to: &Path) -> Result<PathBuf, String> {
    let from_name = from
        .file_name()
        .and_then(|name| name.to_str())
        .ok_or_else(|| format!("{} has no UTF-8 file name", from.display()))?;
    let to_name = to
        .file_name()
        .and_then(|name| name.to_str())
        .unwrap_or_default();
    let marker = from.with_file_name(backup_artifact_name(from_name, "renamed"));
    let body = serde_json::json!({
        "renamed_from": from_name,
        "renamed_to": to_name,
    });
    let bytes = serde_json::to_vec_pretty(&body).map_err(|error| error.to_string())?;
    fs::write(&marker, bytes)
        .map_err(|error| format!("write rename marker {}: {error}", marker.display()))?;
    Ok(marker)
}

/// Case 2 for one slot: rename the lone legacy spelling to its canonical
/// name (dry-run reports [`RenameAction::WouldRename`]).
fn rename_to_canonical(
    slot: DatabaseSlot,
    from: &Path,
    to: &Path,
    apply: bool,
    errors: &mut Vec<String>,
) -> FileRenameEntry {
    let mut entry = FileRenameEntry {
        slot: slot.to_string(),
        from: from.to_path_buf(),
        to: to.to_path_buf(),
        siblings: Vec::new(),
        wal_checkpointed: false,
        marker: None,
        action: RenameAction::WouldRename,
    };
    if !apply {
        return entry;
    }
    let moved = if slot == DatabaseSlot::AgentMemory {
        // A directory slot: the realm databases (and their WAL siblings)
        // move together with the directory.
        if to.exists() {
            Err(format!(
                "cannot move {} to {}: target already exists",
                from.display(),
                to.display()
            ))
        } else {
            fs::rename(from, to)
                .map_err(|error| format!("rename {} -> {}: {error}", from.display(), to.display()))
                .map(|()| (Vec::new(), false))
        }
    } else {
        move_database_file(from, to)
    };
    match moved {
        Ok((siblings, checkpointed)) => {
            entry.siblings = siblings;
            entry.wal_checkpointed = checkpointed;
            entry.action = RenameAction::Renamed;
            match write_rename_marker(from, to) {
                Ok(marker) => entry.marker = Some(marker),
                Err(error) => errors.push(error),
            }
        }
        Err(error) => {
            entry.action = RenameAction::Refused;
            errors.push(format!("rename of the {slot} store refused: {error}"));
        }
    }
    entry
}

// ─────────────────────────────────────────────────────────────────────────
// Case 3 mechanics: divergence + fail-closed reconciliation.
// ─────────────────────────────────────────────────────────────────────────

/// Digest of one file's bytes; for SQLite databases the `-wal` sidecar (if
/// present) is folded in, so uncheckpointed frames register as divergence
/// (conservative: never reports "equal" for possibly-different content).
fn file_digest(path: &Path) -> Result<[u8; 32], String> {
    let mut hasher = Sha256::new();
    hasher.update(fs::read(path).map_err(|error| format!("{}: {error}", path.display()))?);
    let wal = path_with_suffix(path, "-wal");
    if wal.is_file() {
        hasher.update(fs::read(&wal).map_err(|error| format!("{}: {error}", wal.display()))?);
    }
    Ok(hasher.finalize().into())
}

fn quote_ident(name: &str) -> String {
    format!("\"{}\"", name.replace('"', "\"\""))
}

type RowDigests = BTreeMap<String, [u8; 32]>;

/// Per-row content digests over every user table of one database, keyed
/// `table/<primary key>` (rowid when a table declares no primary key), fed
/// in declared column order so digests are comparable across copies. The
/// `meerkat_schema` ledger table is excluded — twins routinely differ only
/// in whether a copy was ever opened by an M3 binary, and that is not data
/// divergence (the byte-identity check still sees it).
fn row_digests(db_path: &Path) -> Result<RowDigests, String> {
    let conn = meerkat_sqlite::open(db_path, meerkat_sqlite::ConnectionProfile::ReadOnly)
        .map_err(|error| error.to_string())?;
    let inner = || -> Result<RowDigests, rusqlite::Error> {
        let tables: Vec<String> = {
            let mut statement = conn.prepare(
                "SELECT name FROM sqlite_master WHERE type = 'table' \
                 AND name NOT LIKE 'sqlite_%' AND name <> 'meerkat_schema' ORDER BY name",
            )?;
            statement
                .query_map([], |row| row.get::<_, String>(0))?
                .collect::<Result<Vec<_>, _>>()?
        };
        let mut digests = RowDigests::new();
        for table in tables {
            // Primary-key column positions, in key order.
            let mut pk_columns: Vec<(i64, usize)> = Vec::new();
            conn.pragma(None, "table_info", &table, |row| {
                let cid: i64 = row.get(0)?;
                let pk: i64 = row.get(5)?;
                if pk > 0 {
                    pk_columns.push((pk, cid as usize));
                }
                Ok(())
            })?;
            pk_columns.sort_unstable();

            let (sql, key_indexes, data_start): (String, Vec<usize>, usize) =
                if pk_columns.is_empty() {
                    (
                        format!(
                            "SELECT rowid, * FROM {} ORDER BY rowid",
                            quote_ident(&table)
                        ),
                        vec![0],
                        1,
                    )
                } else {
                    let order: Vec<String> = pk_columns
                        .iter()
                        .map(|(_, cid)| format!("{}", cid + 1))
                        .collect();
                    (
                        format!(
                            "SELECT * FROM {} ORDER BY {}",
                            quote_ident(&table),
                            order.join(", ")
                        ),
                        pk_columns.iter().map(|(_, cid)| *cid).collect(),
                        0,
                    )
                };
            let mut statement = conn.prepare(&sql)?;
            let column_count = statement.column_count();
            let mut rows = statement.query([])?;
            while let Some(row) = rows.next()? {
                let mut key = format!("{table}/");
                for (position, index) in key_indexes.iter().enumerate() {
                    if position > 0 {
                        key.push('|');
                    }
                    key.push_str(&value_key(row.get_ref(*index)?));
                }
                let mut hasher = Sha256::new();
                for index in data_start..column_count {
                    feed_value(&mut hasher, row.get_ref(index)?);
                }
                digests.insert(key, hasher.finalize().into());
            }
        }
        Ok(digests)
    };
    inner().map_err(|error| format!("{}: {error}", db_path.display()))
}

fn value_key(value: rusqlite::types::ValueRef<'_>) -> String {
    use rusqlite::types::ValueRef;
    match value {
        ValueRef::Null => "null".to_string(),
        ValueRef::Integer(int) => int.to_string(),
        ValueRef::Real(real) => real.to_string(),
        ValueRef::Text(text) => String::from_utf8_lossy(text).into_owned(),
        ValueRef::Blob(blob) => {
            use std::fmt::Write as _;
            blob.iter().fold(String::new(), |mut hex, byte| {
                let _ = write!(hex, "{byte:02x}");
                hex
            })
        }
    }
}

fn feed_value(hasher: &mut Sha256, value: rusqlite::types::ValueRef<'_>) {
    use rusqlite::types::ValueRef;
    let (tag, bytes): (u8, Vec<u8>) = match value {
        ValueRef::Null => (0, Vec::new()),
        ValueRef::Integer(int) => (1, int.to_be_bytes().to_vec()),
        ValueRef::Real(real) => (2, real.to_be_bytes().to_vec()),
        ValueRef::Text(text) => (3, text.to_vec()),
        ValueRef::Blob(blob) => (4, blob.to_vec()),
    };
    hasher.update([tag]);
    hasher.update((bytes.len() as u64).to_be_bytes());
    hasher.update(&bytes);
}

/// Per-file digests inside one directory twin (the agent-memory roots),
/// keyed by relative file name.
fn dir_file_digests(dir: &Path) -> Result<RowDigests, String> {
    let mut digests = RowDigests::new();
    let entries = fs::read_dir(dir).map_err(|error| format!("{}: {error}", dir.display()))?;
    let mut files: Vec<PathBuf> = entries
        .filter_map(Result::ok)
        .map(|entry| entry.path())
        .filter(|path| path.is_file())
        .collect();
    files.sort();
    for file in files {
        let Some(name) = file.file_name().and_then(|name| name.to_str()) else {
            continue;
        };
        // WAL/SHM siblings fold into their database's digest.
        if name.ends_with("-wal") || name.ends_with("-shm") {
            continue;
        }
        digests.insert(name.to_string(), file_digest(&file)?);
    }
    Ok(digests)
}

/// Classify per-key divergence across every twin copy: equal keys are
/// counted, divergent and single-copy keys are listed.
fn classify_divergence(
    per_copy: &[(PathBuf, RowDigests)],
    rows_equal: &mut usize,
    rows: &mut Vec<RowDivergenceEntry>,
) {
    let mut all_keys: Vec<String> = per_copy
        .iter()
        .flat_map(|(_, digests)| digests.keys().cloned())
        .collect();
    all_keys.sort();
    all_keys.dedup();
    for key in all_keys {
        let holders: Vec<(&PathBuf, &[u8; 32])> = per_copy
            .iter()
            .filter_map(|(path, digests)| digests.get(&key).map(|digest| (path, digest)))
            .collect();
        if holders.len() == 1 {
            rows.push(RowDivergenceEntry {
                key,
                status: DivergenceStatus::OnlyIn {
                    location: holders[0].0.clone(),
                },
            });
        } else if holders.len() == per_copy.len()
            && holders.iter().all(|(_, digest)| *digest == holders[0].1)
        {
            *rows_equal += 1;
        } else {
            rows.push(RowDivergenceEntry {
                key,
                status: DivergenceStatus::Divergent,
            });
        }
    }
}

/// Archive one twin copy read-only under the registered backup naming,
/// folding a non-empty WAL into the database first so the archive is
/// self-contained; any leftover siblings are archived alongside (their
/// names inherit the `.pre-` segment). Directories archive as a unit.
fn archive_twin_copy(
    path: &Path,
    purpose: &str,
    apply_checkpoint: bool,
) -> Result<Vec<PathBuf>, String> {
    if path.is_file() && apply_checkpoint {
        checkpoint_wal_if_nonempty(path)?;
    }
    let siblings: Vec<PathBuf> = ["-wal", "-shm"]
        .iter()
        .map(|suffix| path_with_suffix(path, suffix))
        .filter(|sibling| sibling.exists())
        .collect();
    let archive = archive_path_read_only(path, purpose).map_err(|error| error.to_string())?;
    let mut archived = vec![archive.clone()];
    for sibling in siblings {
        let Some(suffix) = sibling
            .file_name()
            .and_then(|name| name.to_str())
            .and_then(|name| name.rsplit_once('-').map(|(_, tail)| format!("-{tail}")))
        else {
            continue;
        };
        let dst = path_with_suffix(&archive, &suffix);
        fs::rename(&sibling, &dst)
            .map_err(|error| format!("archive sibling {}: {error}", sibling.display()))?;
        if let Ok(metadata) = fs::symlink_metadata(&dst) {
            let mut permissions = metadata.permissions();
            permissions.set_readonly(true);
            let _ = fs::set_permissions(&dst, permissions);
        }
        archived.push(dst);
    }
    Ok(archived)
}

/// Case 3 for one slot: compute the per-domain divergence report, then
/// refuse, dedup (exact equality), or adopt-and-archive. A canonicalization
/// rename that cannot be re-fenced at its new path records a pass error in
/// `errors` and the path in `unfenced` (the later cases skip it fail-closed).
#[allow(clippy::too_many_arguments)]
fn reconcile_twin(
    slot: DatabaseSlot,
    paths: &[PathBuf],
    mode: MigrateMode,
    adopt: Option<&Path>,
    renames: &mut Vec<FileRenameEntry>,
    fence: Option<&mut MobKitMaintenanceFence>,
    errors: &mut Vec<String>,
    unfenced: &mut Vec<PathBuf>,
) -> TwinReport {
    let mut twin = TwinReport {
        slot: slot.to_string(),
        paths: paths.to_vec(),
        byte_identical: false,
        rows_equal: 0,
        rows: Vec::new(),
        resolution: TwinResolution::Refused {
            reason: "unresolved".to_string(),
        },
        notes: Vec::new(),
        errors: Vec::new(),
    };
    match slot {
        DatabaseSlot::Continuity => twin.notes.push(
            "no synthesis: continuity fencing tokens are per-database sequences; merging twin \
             histories would corrupt CAS — adopt one copy, archive the rest"
                .to_string(),
        ),
        DatabaseSlot::Console => twin.notes.push(
            "no synthesis: console cursor_seq is a per-database AUTOINCREMENT sequence; merging \
             twin timelines would corrupt cursor replay — adopt one copy, archive the rest"
                .to_string(),
        ),
        _ => {}
    }

    // Divergence: row-level for the three row-compared domains, per-file for
    // the memory directory twin, file-digest for everything else.
    let row_level = matches!(
        slot,
        DatabaseSlot::Continuity | DatabaseSlot::Metadata | DatabaseSlot::Console
    );
    if row_level || slot == DatabaseSlot::AgentMemory {
        let mut per_copy: Vec<(PathBuf, RowDigests)> = Vec::new();
        for path in paths {
            let digests = if row_level {
                row_digests(path)
            } else {
                dir_file_digests(path)
            };
            match digests {
                Ok(digests) => per_copy.push((path.clone(), digests)),
                Err(error) => {
                    twin.errors.push(format!(
                        "divergence unavailable for {}: {error}",
                        path.display()
                    ));
                    per_copy.push((path.clone(), RowDigests::new()));
                }
            }
        }
        classify_divergence(&per_copy, &mut twin.rows_equal, &mut twin.rows);
    } else {
        twin.notes
            .push("divergence computed at file-digest level for this slot".to_string());
    }

    // Byte identity (exact-equality dedup precondition): file digests with
    // the WAL folded in; directory twins compare their full file maps.
    let byte_identical = if slot == DatabaseSlot::AgentMemory {
        let mut maps = Vec::new();
        let mut readable = true;
        for path in paths {
            match dir_file_digests(path) {
                Ok(map) => maps.push(map),
                Err(_) => {
                    readable = false;
                    break;
                }
            }
        }
        readable && maps.windows(2).all(|pair| pair[0] == pair[1])
    } else {
        let mut digests = Vec::new();
        let mut readable = true;
        for path in paths {
            match file_digest(path) {
                Ok(digest) => digests.push(digest),
                Err(error) => {
                    twin.errors.push(format!(
                        "file digest unavailable for {}: {error}",
                        path.display()
                    ));
                    readable = false;
                    break;
                }
            }
        }
        readable && digests.windows(2).all(|pair| pair[0] == pair[1])
    };
    twin.byte_identical = byte_identical;

    if mode != MigrateMode::Apply {
        twin.resolution = TwinResolution::Refused {
            reason: if byte_identical {
                "twin copies are byte-identical; rerun with `--apply` to dedup (keep the \
                 canonical spelling, archive the redundant copy read-only)"
                    .to_string()
            } else {
                "divergent twin copies; rerun with `--apply --adopt <path>` to adopt one copy \
                 and archive the rest read-only (no synthesis)"
                    .to_string()
            },
        };
        return twin;
    }

    // Apply: --adopt wins; byte-identical twins dedup; anything else refuses.
    let canonical_of = |path: &Path| fs::canonicalize(path).unwrap_or_else(|_| path.to_path_buf());
    let adopted_member = adopt.and_then(|adopt_path| {
        let adopt_canonical = canonical_of(adopt_path);
        paths
            .iter()
            .find(|path| canonical_of(path) == adopt_canonical)
            .cloned()
    });
    let (kept, purpose, dedup) = if let Some(adopted) = adopted_member {
        (adopted, "twin", false)
    } else if byte_identical {
        let canonical_name = slot.canonical_name();
        let kept = paths
            .iter()
            .find(|path| path.file_name().and_then(|name| name.to_str()) == Some(canonical_name))
            .unwrap_or(&paths[0])
            .clone();
        (kept, "twin-dedup", true)
    } else {
        twin.resolution = TwinResolution::Refused {
            reason: if adopt.is_some() {
                format!(
                    "--adopt does not name one of this slot's twin copies (candidates: {})",
                    paths
                        .iter()
                        .map(|path| path.display().to_string())
                        .collect::<Vec<_>>()
                        .join(", ")
                )
            } else {
                "divergent twin copies; rerun with `--apply --adopt <path>` to adopt one copy \
                 and archive the rest read-only (no synthesis)"
                    .to_string()
            },
        };
        return twin;
    };

    // Archive the non-kept copies first (freeing the canonical name when the
    // kept copy is legacy-named), then canonicalize the kept copy.
    let mut archived = Vec::new();
    for path in paths {
        if *path == kept {
            continue;
        }
        match archive_twin_copy(path, purpose, true) {
            Ok(mut paths) => archived.append(&mut paths),
            Err(error) => {
                twin.resolution = TwinResolution::Refused {
                    reason: format!("archive of {} failed: {error}", path.display()),
                };
                return twin;
            }
        }
    }
    let canonical_path = kept.parent().map_or_else(
        || PathBuf::from(slot.canonical_name()),
        |parent| parent.join(slot.canonical_name()),
    );
    let final_path = if kept == canonical_path {
        kept
    } else {
        let mut rename_errors = Vec::new();
        let entry = rename_to_canonical(slot, &kept, &canonical_path, true, &mut rename_errors);
        let renamed = entry.action == RenameAction::Renamed;
        renames.push(entry);
        if renamed {
            if slot != DatabaseSlot::AgentMemory
                && let Some(fence) = fence
                && let Err(error) = fence.cover_renamed(&canonical_path)
            {
                errors.push(format!(
                    "adopted {slot} store is not fenced at its canonical path {}: {error}; \
                     skipping further mutation of this database",
                    canonical_path.display()
                ));
                unfenced.push(canonical_path.clone());
            }
            canonical_path
        } else {
            twin.resolution = TwinResolution::Refused {
                reason: format!(
                    "adopted copy could not be renamed to its canonical name: {}",
                    rename_errors.join("; ")
                ),
            };
            return twin;
        }
    };
    twin.resolution = if dedup {
        TwinResolution::Deduped {
            kept: final_path,
            archived,
        }
    } else {
        TwinResolution::Adopted {
            adopted: final_path,
            archived,
        }
    };
    twin
}

// ─────────────────────────────────────────────────────────────────────────
// Case 5: dead runtime-registry entries.
// ─────────────────────────────────────────────────────────────────────────

/// True/false when pid liveness can be determined on this platform, `None`
/// otherwise. Report-only input; a pid owned by another user still probes
/// correctly through `ps`.
fn pid_alive(pid: u64) -> Option<bool> {
    #[cfg(unix)]
    {
        std::process::Command::new("ps")
            .args(["-p", &pid.to_string()])
            .stdout(std::process::Stdio::null())
            .stderr(std::process::Stdio::null())
            .status()
            .ok()
            .map(|status| status.success())
    }
    #[cfg(not(unix))]
    {
        let _ = pid;
        None
    }
}

/// Report `tux-runtimes.json` entries whose recorded pid is no longer alive
/// (report-only; migrate never edits the registry — the gateway prunes dead
/// entries itself on next boot).
fn sweep_runtime_registry(
    state_dir: &Path,
    findings: &mut Vec<StorageFinding>,
    notes: &mut Vec<String>,
) {
    let registry_path = state_dir.join(RUNTIME_REGISTRY_FILE_NAME);
    if !registry_path.is_file() {
        return;
    }
    let Ok(bytes) = fs::read(&registry_path) else {
        return;
    };
    let Ok(value) = serde_json::from_slice::<serde_json::Value>(&bytes) else {
        notes.push(format!(
            "runtime registry {} is unparseable; dead-entry census skipped",
            registry_path.display()
        ));
        return;
    };
    let Some(entries) = value.get("entries").and_then(serde_json::Value::as_array) else {
        return;
    };
    for entry in entries {
        let runtime_id = entry
            .get("runtime_id")
            .and_then(serde_json::Value::as_str)
            .unwrap_or("<unknown>");
        let Some(pid) = entry.get("pid").and_then(serde_json::Value::as_u64) else {
            continue;
        };
        match pid_alive(pid) {
            Some(false) => findings.push(
                StorageFinding::new(
                    FindingSeverity::Info,
                    FINDING_DEAD_RUNTIME_REGISTRY_ENTRY,
                    format!(
                        "runtime registry entry '{runtime_id}' records pid {pid}, which is no \
                         longer alive (report-only; the gateway prunes dead entries on next boot)"
                    ),
                )
                .with_path(registry_path.clone()),
            ),
            Some(true) => {}
            None => notes.push(format!(
                "runtime registry entry '{runtime_id}': pid liveness not determinable on this \
                 platform"
            )),
        }
    }
}

// ─────────────────────────────────────────────────────────────────────────
// Prune (registered backup-artifact lifecycle).
// ─────────────────────────────────────────────────────────────────────────

fn recursive_size(path: &Path) -> u64 {
    let Ok(metadata) = fs::symlink_metadata(path) else {
        return 0;
    };
    if metadata.is_dir() {
        let Ok(entries) = fs::read_dir(path) else {
            return 0;
        };
        entries
            .filter_map(Result::ok)
            .map(|entry| recursive_size(&entry.path()))
            .sum()
    } else {
        metadata.len()
    }
}

fn age_days(path: &Path) -> u64 {
    fs::symlink_metadata(path)
        .and_then(|metadata| metadata.modified())
        .ok()
        .and_then(|modified| std::time::SystemTime::now().duration_since(modified).ok())
        .map(|age| age.as_secs() / 86_400)
        .unwrap_or(0)
}

fn artifact_kind(name: &str) -> Option<PruneArtifactKind> {
    if is_registered_backup_artifact_name(name) {
        Some(PruneArtifactKind::BackupArtifact)
    } else if is_registered_quarantine_artifact_name(name) {
        Some(PruneArtifactKind::QuarantinedIndex)
    } else {
        None
    }
}

fn push_artifacts_in(dir: &Path, dirs_too: bool, artifacts: &mut Vec<MobKitPruneArtifact>) {
    let Ok(entries) = fs::read_dir(dir) else {
        return;
    };
    let mut paths: Vec<PathBuf> = entries
        .filter_map(Result::ok)
        .map(|entry| entry.path())
        .collect();
    paths.sort();
    for path in paths {
        if path.is_dir() && !dirs_too {
            continue;
        }
        let Some(name) = path.file_name().and_then(|name| name.to_str()) else {
            continue;
        };
        let Some(kind) = artifact_kind(name) else {
            continue;
        };
        artifacts.push(MobKitPruneArtifact {
            bytes: recursive_size(&path),
            age_days: age_days(&path),
            path,
            kind,
            action: PruneAction::Kept,
        });
    }
}

/// Enumerate every registered maintenance artifact under one MobKit state
/// directory: `*.pre-*` backups (files or archived memory-root directories)
/// and `*.corrupt-*` quarantines at the state-dir root and inside the live
/// agent-memory roots. Artifacts inside an archived directory belong to the
/// archive's own entry. Nothing outside these naming patterns is ever
/// returned — prune's deletion authority is exactly this listing.
pub fn enumerate_state_dir_artifacts(state_dir: &Path) -> Vec<MobKitPruneArtifact> {
    let mut artifacts = Vec::new();
    push_artifacts_in(state_dir, true, &mut artifacts);
    for spelling in MEMORY_ROOT_SPELLINGS {
        let root = state_dir.join(spelling);
        if root.is_dir() {
            push_artifacts_in(&root, false, &mut artifacts);
        }
    }
    artifacts
}

/// Registered maintenance-artifact lifecycle over one state directory:
/// enumerate `*.pre-*` / `*.corrupt-*` artifacts; with
/// [`MigrateMode::Apply`], delete those at least `older_than_days` old
/// (deletion goes through
/// [`meerkat_store::migrate::remove_maintenance_artifact`], which refuses
/// anything outside the registered naming). Blocking.
pub fn prune_state_dir(
    state_dir: &Path,
    older_than_days: u64,
    mode: MigrateMode,
) -> MobKitPruneReport {
    let mut report = MobKitPruneReport::new(mode, state_dir, older_than_days);
    let mut artifacts = enumerate_state_dir_artifacts(state_dir);
    for artifact in &mut artifacts {
        if artifact.age_days < older_than_days {
            artifact.action = PruneAction::Kept;
            continue;
        }
        if mode != MigrateMode::Apply {
            artifact.action = PruneAction::WouldDelete;
            continue;
        }
        match remove_maintenance_artifact(&artifact.path) {
            Ok(()) => artifact.action = PruneAction::Deleted,
            Err(error) => {
                artifact.action = PruneAction::DeleteFailed;
                report.errors.push(format!(
                    "failed to delete {}: {error}",
                    artifact.path.display()
                ));
            }
        }
    }
    report.artifacts = artifacts;
    report
}

#[cfg(test)]
#[allow(clippy::expect_used, clippy::unwrap_used, clippy::panic)]
mod tests {
    use super::*;
    use rusqlite::Connection;

    /// The exact 0.7.x-era continuity DDL (no `meerkat_schema` ledger).
    const LEGACY_CONTINUITY_DDL: &str = "CREATE TABLE continuity_records (
            identity       TEXT PRIMARY KEY,
            agent_runtime_id TEXT NOT NULL,
            session_id     TEXT NOT NULL,
            generation     INTEGER NOT NULL,
            checkpoint_version INTEGER NOT NULL,
            fencing_token  INTEGER NOT NULL
        );
        CREATE TABLE session_snapshots (
            session_id     TEXT PRIMARY KEY,
            identity       TEXT NOT NULL,
            generation     INTEGER NOT NULL,
            checkpoint_version INTEGER NOT NULL,
            fencing_token  INTEGER NOT NULL,
            data           BLOB NOT NULL
        );";

    fn create_legacy_continuity(path: &Path) -> Connection {
        let conn = Connection::open(path).expect("create fixture db");
        conn.execute_batch(LEGACY_CONTINUITY_DDL)
            .expect("apply legacy ddl");
        conn
    }

    fn insert_snapshot(conn: &Connection, session_id: &str, identity: &str, data: &[u8]) {
        conn.execute(
            "INSERT INTO session_snapshots \
             (session_id, identity, generation, checkpoint_version, fencing_token, data) \
             VALUES (?1, ?2, 3, 4, 7, ?3)",
            rusqlite::params![session_id, identity, data],
        )
        .expect("insert snapshot");
    }

    fn insert_record(conn: &Connection, identity: &str, session_id: &str) {
        conn.execute(
            "INSERT INTO continuity_records \
             (identity, agent_runtime_id, session_id, generation, checkpoint_version, \
              fencing_token) VALUES (?1, 'rt-1', ?2, 3, 4, 7)",
            rusqlite::params![identity, session_id],
        )
        .expect("insert record");
    }

    fn legacy_session_bytes() -> (String, Vec<u8>) {
        let session = meerkat_core::Session::new();
        let id = session.id().to_string();
        let bytes = serde_json::to_vec(&session).expect("serialize legacy session");
        (id, bytes)
    }

    /// Build a store through its normal constructor, then strip the ledger —
    /// the exact shape of a pre-M3 file with the historical schema.
    fn drop_ledger(path: &Path) {
        let conn = Connection::open(path).expect("open for ledger drop");
        conn.execute_batch("DROP TABLE meerkat_schema")
            .expect("drop ledger");
    }

    fn file_digest_hex(path: &Path) -> String {
        format!("{:x}", Sha256::digest(fs::read(path).expect("read db")))
    }

    fn ledger_entry<'a>(
        report: &'a MobKitMigrateReport,
        file_name: &str,
        domain: &str,
    ) -> &'a LedgerBaselineEntry {
        report
            .ledger
            .iter()
            .find(|entry| entry.database.ends_with(file_name) && entry.domain == domain)
            .unwrap_or_else(|| panic!("no ledger entry for {file_name} [{domain}]"))
    }

    #[test]
    fn fence_enumerates_layout_slots_and_memory_realms_sorted() {
        let temp = tempfile::tempdir().expect("tempdir");
        let state = temp.path();
        drop(create_legacy_continuity(&state.join("continuity.db")));
        drop(Connection::open(state.join("sessions.sqlite3")).expect("sessions"));
        drop(Connection::open(state.join("runtime.sqlite")).expect("runtime"));
        fs::create_dir_all(state.join("agent-memory")).expect("memory root");
        drop(Connection::open(state.join("agent-memory/alpha.sqlite3")).expect("realm"));
        let jobs = MobKitStorageLayout::with_injected_roots(state.to_path_buf(), None).jobs_db();
        drop(meerkat::SqliteDetachedJobStore::open(&jobs).expect("jobs"));
        // Excluded: the admission sidecar (no fence guard by design) and
        // non-database files.
        drop(Connection::open(state.join(WORKGRAPH_ADMISSION_SIDECAR_FILE)).expect("sidecar"));
        fs::write(state.join("notes.txt"), b"x").expect("notes");

        let fence = MobKitMaintenanceFence::acquire(state, Duration::from_secs(1)).expect("fence");
        let mut expected = vec![
            state.join("agent-memory/alpha.sqlite3"),
            state.join("continuity.db"),
            jobs,
            state.join("runtime.sqlite"),
            state.join("sessions.sqlite3"),
        ];
        expected.sort();
        assert_eq!(fence.fenced_databases(), expected.as_slice());
        assert_eq!(fence.len(), 5);
        assert!(!fence.is_empty());
        for database in fence.fenced_databases() {
            assert!(meerkat_sqlite::fence_lock_path(database).is_file());
        }
    }

    #[test]
    fn fence_foreign_holder_fails_typed_and_releases_partial_acquisition() {
        let temp = tempfile::tempdir().expect("tempdir");
        let state = temp.path();
        drop(create_legacy_continuity(&state.join("continuity.db")));
        drop(Connection::open(state.join("sessions.sqlite3")).expect("sessions"));

        // A FOREIGN process holding the second fence: a raw exclusive lock
        // without this process's holder registry entry.
        let foreign_lock = meerkat_sqlite::fence_lock_path(&state.join("sessions.sqlite3"));
        let foreign = fs::OpenOptions::new()
            .read(true)
            .write(true)
            .create(true)
            .truncate(false)
            .open(&foreign_lock)
            .expect("open foreign lock");
        foreign.try_lock().expect("foreign exclusive lock");

        let error = MobKitMaintenanceFence::acquire(state, Duration::from_millis(100))
            .expect_err("foreign holder must refuse acquisition");
        assert!(
            matches!(error, meerkat_sqlite::SqliteStoreError::MaintenanceFenceHeld { ref path }
                if path.ends_with("sessions.sqlite3")),
            "{error:?}"
        );

        // RAII: the continuity fence was released on failure.
        let reacquired = meerkat_sqlite::ExclusiveFence::try_acquire(&state.join("continuity.db"))
            .expect("try acquire");
        assert!(reacquired.is_some(), "partial acquisition must be released");
        drop(reacquired);
        drop(foreign);

        // And a fenced apply run surfaces the same typed refusal as an error.
        let foreign = fs::OpenOptions::new()
            .read(true)
            .write(true)
            .create(true)
            .truncate(false)
            .open(&foreign_lock)
            .expect("reopen foreign lock");
        foreign.try_lock().expect("foreign exclusive lock");
        let report = migrate_state_dir(state, MigrateMode::Apply, None);
        assert!(report.has_errors());
        assert!(
            report.errors[0].contains("maintenance fence"),
            "{:?}",
            report.errors
        );
        drop(foreign);
    }

    #[test]
    fn failed_canonical_fence_after_rename_fails_closed() {
        let temp = tempfile::tempdir().expect("tempdir");
        let state = temp.path();
        let legacy_db = state.join("continuity.db");
        let (sid, bytes) = legacy_session_bytes();
        {
            let conn = create_legacy_continuity(&legacy_db);
            insert_record(&conn, "test:alice", &sid);
            insert_snapshot(&conn, &sid, "test:alice", &bytes);
        }

        // A FOREIGN holder on the CANONICAL path's fence lock: the pass
        // fence rides the legacy lock path, so acquisition and the rename
        // itself succeed — but the post-rename re-fence must fail, and the
        // pass must not keep mutating the now-unfenced database.
        let canonical = state.join("continuity.sqlite3");
        let foreign_lock = meerkat_sqlite::fence_lock_path(&canonical);
        let foreign = fs::OpenOptions::new()
            .read(true)
            .write(true)
            .create(true)
            .truncate(false)
            .open(&foreign_lock)
            .expect("open foreign lock");
        foreign.try_lock().expect("foreign exclusive lock");

        let report = migrate_state_dir(state, MigrateMode::Apply, None);

        assert!(canonical.is_file(), "rename itself must have happened");
        assert!(report.has_errors(), "{:?}", report.errors);
        assert!(
            report
                .errors
                .iter()
                .any(|error| error.contains("not fenced at its canonical path")),
            "{:?}",
            report.errors
        );

        // Fail-closed: the ledgered constructor never ran against the
        // unfenced database (no meerkat_schema table materialized) ...
        {
            let conn = Connection::open(&canonical).expect("open canonical");
            let ledger_tables: i64 = conn
                .query_row(
                    "SELECT COUNT(*) FROM sqlite_master \
                     WHERE type = 'table' AND name = 'meerkat_schema'",
                    [],
                    |row| row.get(0),
                )
                .expect("probe ledger");
            assert_eq!(
                ledger_tables, 0,
                "constructor must not run against an unfenced database"
            );
        }
        // ... and the adoption walk was skipped, not run unfenced.
        let adoption = report.adoption.as_ref().expect("adoption outcome");
        assert!(adoption.report.is_none());
        assert!(
            adoption
                .skipped
                .as_deref()
                .is_some_and(|reason| reason.contains("fence")),
            "{adoption:?}"
        );
        drop(foreign);
    }

    /// M4b: the one-way head-canonical continuity bump is an OPERATOR
    /// action. Opening the store (what a gateway launch does) must leave the
    /// file at the rollback-safe baseline; dry-run must announce the pending
    /// bump without touching the file; `--apply` commits it under the fence.
    #[test]
    fn head_canonical_continuity_bump_is_operator_gated() {
        let temp = tempfile::tempdir().expect("tempdir");
        let state = temp.path();
        let db = state.join("continuity.sqlite3");
        drop(create_legacy_continuity(&db));

        // A gateway launch: opening the store converges the baseline only.
        LocalContinuityStore::open(&db).expect("gateway-style open");
        let probe = Connection::open(&db).expect("probe");
        assert_eq!(
            meerkat_sqlite::domain_version(&probe, "mobkit-continuity").expect("ledger"),
            Some(1),
            "launching a new gateway must not lock the previous release out of the file"
        );
        drop(probe);

        let before = file_digest_hex(&db);
        let dry = migrate_state_dir(state, MigrateMode::DryRun, None);
        assert!(!dry.has_errors(), "{:?}", dry.errors);
        assert!(
            dry.notes
                .iter()
                .any(|note| note.contains("head-canonical") && note.contains("ONE-WAY")),
            "dry-run must announce the pending one-way bump: {:?}",
            dry.notes
        );
        assert_eq!(
            file_digest_hex(&db),
            before,
            "dry-run must not mutate the database"
        );

        let applied = migrate_state_dir(state, MigrateMode::Apply, None);
        assert!(!applied.has_errors(), "{:?}", applied.errors);
        let entry = ledger_entry(&applied, "continuity.sqlite3", "mobkit-continuity");
        assert_eq!(entry.action, LedgerBaselineAction::Stamped);
        assert_eq!(entry.before, Some(1));
        assert_eq!(
            entry.after,
            Some(crate::identity_first::HEAD_CANONICAL_CONTINUITY_SCHEMA_VERSION)
        );

        // Idempotent: a second apply is a no-op.
        let second = migrate_state_dir(state, MigrateMode::Apply, None);
        assert!(!second.has_errors(), "{:?}", second.errors);
        assert_eq!(
            ledger_entry(&second, "continuity.sqlite3", "mobkit-continuity").action,
            LedgerBaselineAction::AlreadyCurrent
        );
    }

    #[test]
    fn dry_run_is_a_read_only_version_matrix() {
        let temp = tempfile::tempdir().expect("tempdir");
        let state = temp.path();
        drop(create_legacy_continuity(&state.join("continuity.db")));
        {
            let conn = Connection::open(state.join("sessions.db")).expect("sessions");
            conn.execute_batch("CREATE TABLE sessions (session_id TEXT PRIMARY KEY)")
                .expect("sessions ddl");
        }
        drop(Connection::open(state.join(WORKGRAPH_ADMISSION_SIDECAR_FILE)).expect("sidecar"));
        let jobs = MobKitStorageLayout::with_injected_roots(state.to_path_buf(), None).jobs_db();
        drop(meerkat::SqliteDetachedJobStore::open(&jobs).expect("jobs"));
        let before = file_digest_hex(&state.join("continuity.db"));
        let jobs_before = file_digest_hex(&jobs);

        let report = migrate_state_dir(state, MigrateMode::DryRun, None);
        assert!(!report.has_errors(), "{:?}", report.errors);
        assert_eq!(
            ledger_entry(&report, "continuity.db", "mobkit-continuity").action,
            LedgerBaselineAction::WouldStamp
        );
        assert_eq!(
            ledger_entry(&report, "sessions.db", "session-store").action,
            LedgerBaselineAction::ReportOnly
        );
        assert_eq!(
            ledger_entry(&report, "jobs.sqlite3", "jobs").action,
            LedgerBaselineAction::Recorded
        );
        assert_eq!(
            ledger_entry(
                &report,
                WORKGRAPH_ADMISSION_SIDECAR_FILE,
                "mobkit-workgraph-admission"
            )
            .action,
            LedgerBaselineAction::Exempt
        );
        assert!(
            report
                .notes
                .iter()
                .any(|note| note.contains("ledger-exempt")),
            "{:?}",
            report.notes
        );
        assert!(
            report
                .renames
                .iter()
                .any(|entry| entry.action == RenameAction::WouldRename
                    && entry.from.ends_with("continuity.db")),
            "dry-run must report the pending rename: {:?}",
            report.renames
        );
        assert_eq!(
            file_digest_hex(&state.join("continuity.db")),
            before,
            "dry-run must leave the database byte-identical"
        );
        assert_eq!(
            file_digest_hex(&jobs),
            jobs_before,
            "dry-run must leave the inherited jobs database byte-identical"
        );
        assert!(
            state.join("continuity.db").is_file(),
            "dry-run must not rename"
        );
    }

    #[test]
    fn apply_renames_with_wal_stamps_ledgers_adopts_and_is_idempotent() {
        let temp = tempfile::tempdir().expect("tempdir");
        let state = temp.path();

        // Legacy-named continuity db in WAL mode with a leaked reader so a
        // non-empty -wal survives on disk at migrate time, holding one
        // adoptable legacy snapshot.
        let legacy_db = state.join("continuity.db");
        let (sid, legacy_bytes) = legacy_session_bytes();
        {
            let conn = Connection::open(&legacy_db).expect("create fixture");
            conn.pragma_update(None, "journal_mode", "wal")
                .expect("wal mode");
            conn.execute_batch(LEGACY_CONTINUITY_DDL).expect("ddl");
            insert_record(&conn, "test:alice", &sid);
            insert_snapshot(&conn, &sid, "test:alice", &legacy_bytes);
            // Leak the connection: the -wal file persists on disk.
            std::mem::forget(conn);
        }
        let wal = path_with_suffix(&legacy_db, "-wal");
        assert!(
            fs::metadata(&wal).map(|meta| meta.len()).unwrap_or(0) > 0,
            "fixture must leave a non-empty WAL"
        );

        // Legacy metadata spelling with the real historical schema, no ledger.
        let legacy_metadata = state.join("mobkit_metadata.sqlite");
        drop(SqliteMetadataStore::open(&legacy_metadata).expect("metadata fixture"));
        drop_ledger(&legacy_metadata);

        // Legacy memory root with one pre-ledger realm database.
        let legacy_memory = state.join("agent-memory-sqlite");
        {
            let store = SqliteAgentMemoryStore::open(&legacy_memory).expect("memory fixture");
            store.open_realm_ledgered("alpha").expect("realm fixture");
        }
        drop_ledger(&legacy_memory.join("alpha.sqlite3"));

        let report = migrate_state_dir(state, MigrateMode::Apply, None);
        assert!(!report.has_errors(), "{:?}", report.errors);

        // Case 2: db + WAL moved together, marker registered.
        let continuity_rename = report
            .renames
            .iter()
            .find(|entry| entry.slot == "continuity")
            .expect("continuity rename entry");
        assert_eq!(continuity_rename.action, RenameAction::Renamed);
        assert!(continuity_rename.wal_checkpointed);
        assert!(
            continuity_rename
                .siblings
                .iter()
                .any(|sibling| sibling.to.ends_with("continuity.sqlite3-wal")),
            "{:?}",
            continuity_rename.siblings
        );
        let marker = continuity_rename.marker.as_ref().expect("rename marker");
        assert!(marker.is_file());
        assert!(is_registered_backup_artifact_name(
            marker.file_name().and_then(|name| name.to_str()).unwrap()
        ));
        assert!(!legacy_db.exists());
        let canonical = state.join("continuity.sqlite3");
        assert!(canonical.is_file());
        let memory_rename = report
            .renames
            .iter()
            .find(|entry| entry.slot == "agent-memory")
            .expect("memory rename entry");
        assert_eq!(memory_rename.action, RenameAction::Renamed);
        assert!(state.join("agent-memory/alpha.sqlite3").is_file());
        assert!(!legacy_memory.exists());

        // Case 1: mobkit domains stamped through the normal constructors.
        for (file, domain) in [
            ("continuity.sqlite3", "mobkit-continuity"),
            ("mobkit_metadata.sqlite3", "mobkit-metadata"),
            ("alpha.sqlite3", MEMORY_LEDGER_DOMAIN),
        ] {
            let entry = ledger_entry(&report, file, domain);
            assert_eq!(entry.action, LedgerBaselineAction::Stamped, "{file}");
            assert!(entry.after.is_some(), "{file}");
        }

        // Case 4: the H3 walk merged; the observed cursor was adopted.
        let adoption = report.adoption.as_ref().expect("adoption outcome");
        let walk = adoption.report.as_ref().expect("adoption walk");
        assert_eq!(walk.scanned, 1);
        assert_eq!(walk.adopted, 1);
        assert!(walk.is_clean());
        {
            let conn = Connection::open(&canonical).expect("reopen canonical");
            let data: Vec<u8> = conn
                .query_row(
                    "SELECT data FROM session_snapshots WHERE session_id = ?1",
                    [&sid],
                    |row| row.get(0),
                )
                .expect("adopted row");
            let session: meerkat_core::Session =
                serde_json::from_slice(&data).expect("decode adopted");
            assert!(matches!(
                session.try_checkpoint_state().expect("state"),
                meerkat_core::SessionCheckpointState::Verified(_)
            ));
        }

        // The layout now resolves everything canonically.
        let layout = MobKitStorageLayout::with_injected_roots(state.to_path_buf(), None);
        assert_eq!(
            layout.continuity_db().expect("resolve").provenance,
            DatabaseProvenance::Canonical
        );

        // Idempotence: a second apply renames nothing, stamps nothing new,
        // re-adopts nothing.
        let second = migrate_state_dir(state, MigrateMode::Apply, None);
        assert!(!second.has_errors(), "{:?}", second.errors);
        assert!(
            second
                .renames
                .iter()
                .all(|entry| entry.action != RenameAction::Renamed),
            "{:?}",
            second.renames
        );
        assert_eq!(
            ledger_entry(&second, "continuity.sqlite3", "mobkit-continuity").action,
            LedgerBaselineAction::AlreadyCurrent
        );
        let second_walk = second
            .adoption
            .as_ref()
            .and_then(|outcome| outcome.report.as_ref())
            .expect("second walk");
        assert_eq!(second_walk.already_stamped, 1);
        assert_eq!(second_walk.adopted, 0);
    }

    /// A verified, history-bearing session serialized the way a pre-marker
    /// (0.8.4-class) writer persisted it: current-format digests, verified
    /// checkpoint stamp, `digest_format` marker absent.
    fn verified_marker_less_session_bytes(tag: &str) -> (String, Vec<u8>) {
        use meerkat_core::types::UserMessage;
        let mut session = meerkat_core::Session::new();
        session.push(meerkat_core::Message::User(UserMessage::text(format!(
            "{tag} old context"
        ))));
        session.push(meerkat_core::Message::User(UserMessage::text(format!(
            "{tag} old context two"
        ))));
        session
            .commit_transcript_rewrite(
                meerkat_core::TranscriptRewriteSelection::MessageRange { start: 0, end: 2 },
                vec![meerkat_core::Message::User(UserMessage::text(format!(
                    "{tag} replacement"
                )))],
                meerkat_core::TranscriptRewriteReason::new("edit"),
                None,
                None,
            )
            .expect("commit fixture rewrite");
        let stamp = meerkat_core::SessionCheckpointStamp::root(
            &session,
            meerkat_core::SessionCheckpointProvenance::SessionCreated,
        )
        .expect("mint root stamp");
        session
            .install_checkpoint_stamp(stamp)
            .expect("install root stamp");
        let mut document = serde_json::to_value(&session).expect("serialize session");
        document["metadata"][meerkat_core::SESSION_TRANSCRIPT_HISTORY_STATE_KEY]
            .as_object_mut()
            .expect("history state object")
            .remove("digest_format")
            .expect("current writers stamp the marker");
        (
            session.id().to_string(),
            serde_json::to_vec(&document).expect("serialize marker-less document"),
        )
    }

    fn marker_stamp_outcome<'a>(
        report: &'a MobKitMigrateReport,
        store: &str,
    ) -> &'a MarkerStampStoreOutcome {
        report
            .marker_stamping
            .iter()
            .find(|outcome| outcome.store == store)
            .unwrap_or_else(|| panic!("no marker-stamping outcome for {store}"))
    }

    fn document_carries_current_marker(bytes: &[u8]) -> bool {
        let document: serde_json::Value = serde_json::from_slice(bytes).expect("decode document");
        document["metadata"][meerkat_core::SESSION_TRANSCRIPT_HISTORY_STATE_KEY]["digest_format"]
            .as_u64()
            .is_some_and(|format| format >= 2)
    }

    /// Case 6 over the full pass: verified marker-less documents in every
    /// session-document store (continuity, runtime, sessions) are respelled
    /// with the digest-format marker; dry-run is a byte-identical census;
    /// a second apply reports everything already current (HomeCore ask (a)).
    #[test]
    fn marker_stamping_case_stamps_verified_marker_less_rows_across_stores() {
        let temp = tempfile::tempdir().expect("tempdir");
        let state = temp.path();

        let continuity_db = state.join("continuity.sqlite3");
        let (continuity_sid, continuity_bytes) =
            verified_marker_less_session_bytes("case6-continuity");
        {
            let conn = create_legacy_continuity(&continuity_db);
            insert_record(&conn, "test:alice", &continuity_sid);
            insert_snapshot(&conn, &continuity_sid, "test:alice", &continuity_bytes);
        }

        let runtime_db = state.join("runtime.sqlite");
        let (runtime_sid, runtime_bytes) = verified_marker_less_session_bytes("case6-runtime");
        {
            let conn = Connection::open(&runtime_db).expect("create runtime fixture");
            conn.execute_batch(
                "CREATE TABLE runtime_session_snapshots (
                    runtime_id TEXT PRIMARY KEY,
                    session_snapshot BLOB NOT NULL
                );",
            )
            .expect("runtime ddl");
            conn.execute(
                "INSERT INTO runtime_session_snapshots (runtime_id, session_snapshot) \
                 VALUES (?1, ?2)",
                rusqlite::params![format!("session-runtime:{runtime_sid}"), runtime_bytes],
            )
            .expect("insert runtime snapshot");
        }

        let sessions_db = state.join("sessions.sqlite3");
        let (sessions_sid, sessions_bytes) = verified_marker_less_session_bytes("case6-sessions");
        {
            let conn = Connection::open(&sessions_db).expect("create sessions fixture");
            conn.execute_batch(
                "CREATE TABLE sessions (
                    session_id TEXT PRIMARY KEY,
                    created_at_ms INTEGER NOT NULL,
                    updated_at_ms INTEGER NOT NULL,
                    message_count INTEGER NOT NULL,
                    total_tokens INTEGER NOT NULL,
                    metadata_json TEXT NOT NULL,
                    session_json BLOB NOT NULL
                );",
            )
            .expect("sessions ddl");
            conn.execute(
                "INSERT INTO sessions (session_id, created_at_ms, updated_at_ms, message_count, \
                 total_tokens, metadata_json, session_json) VALUES (?1, 0, 0, 1, 0, '{}', ?2)",
                rusqlite::params![sessions_sid, sessions_bytes],
            )
            .expect("insert session row");
        }

        // Dry-run: census only, every database byte-identical.
        let digests_before: Vec<String> = [&continuity_db, &runtime_db, &sessions_db]
            .iter()
            .map(|path| file_digest_hex(path))
            .collect();
        let dry = migrate_state_dir(state, MigrateMode::DryRun, None);
        assert!(!dry.has_errors(), "{:?}", dry.errors);
        for store in ["continuity", "runtime", "sessions"] {
            let walk = marker_stamp_outcome(&dry, store)
                .report
                .as_ref()
                .unwrap_or_else(|| panic!("{store} walk must run"));
            assert_eq!(walk.stamped, 1, "{store} dry-run must report the candidate");
            assert!(walk.is_clean(), "{store}: {:?}", walk.refused);
        }
        let digests_after_dry: Vec<String> = [&continuity_db, &runtime_db, &sessions_db]
            .iter()
            .map(|path| file_digest_hex(path))
            .collect();
        assert_eq!(
            digests_before, digests_after_dry,
            "dry-run must leave every store byte-identical"
        );

        // Apply: every store's candidate row is respelled in place and the
        // rewritten document still verifies.
        let apply = migrate_state_dir(state, MigrateMode::Apply, None);
        assert!(!apply.has_errors(), "{:?}", apply.errors);
        for store in ["continuity", "runtime", "sessions"] {
            let walk = marker_stamp_outcome(&apply, store)
                .report
                .as_ref()
                .unwrap_or_else(|| panic!("{store} walk must run"));
            assert_eq!(walk.stamped, 1, "{store} apply must stamp");
            assert!(walk.is_clean(), "{store}: {:?}", walk.refused);
        }
        for (db, table, column, key, sid) in [
            (
                &continuity_db,
                "session_snapshots",
                "data",
                "session_id",
                &continuity_sid,
            ),
            (
                &runtime_db,
                "runtime_session_snapshots",
                "session_snapshot",
                "runtime_id",
                &format!("session-runtime:{runtime_sid}"),
            ),
            (
                &sessions_db,
                "sessions",
                "session_json",
                "session_id",
                &sessions_sid,
            ),
        ] {
            let conn = Connection::open(db).expect("reopen store");
            let bytes: Vec<u8> = conn
                .query_row(
                    &format!("SELECT {column} FROM {table} WHERE {key} = ?1"),
                    [sid],
                    |row| row.get(0),
                )
                .expect("rewritten row");
            assert!(
                document_carries_current_marker(&bytes),
                "{table} row must carry the digest-format marker after apply"
            );
            let session: meerkat_core::Session =
                serde_json::from_slice(&bytes).expect("decode rewritten document");
            assert!(
                matches!(
                    session.try_checkpoint_state().expect("checkpoint state"),
                    meerkat_core::SessionCheckpointState::Verified(_)
                ),
                "{table} row must keep verifying after the respell"
            );
        }

        // Idempotence: a second apply stamps nothing new.
        let second = migrate_state_dir(state, MigrateMode::Apply, None);
        assert!(!second.has_errors(), "{:?}", second.errors);
        for store in ["continuity", "runtime", "sessions"] {
            let walk = marker_stamp_outcome(&second, store)
                .report
                .as_ref()
                .unwrap_or_else(|| panic!("{store} walk must run"));
            assert_eq!(walk.stamped, 0, "{store} second apply must be a no-op");
            assert_eq!(walk.already_current, 1, "{store}");
        }
    }

    /// HomeCore ask (a), adoption half: checkpoint adoption (case 4)
    /// re-serializes what it adopts, so an adopted legacy history-bearing
    /// document comes out marker-stamped — and case 6 (running later in the
    /// same pass) classifies it already current instead of respelling it a
    /// second time.
    #[test]
    fn adoption_rewrites_carry_the_digest_format_marker() {
        let temp = tempfile::tempdir().expect("tempdir");
        let state = temp.path();
        let continuity_db = state.join("continuity.sqlite3");

        // A legacy (unstamped) history-bearing marker-less document: strip
        // both the checkpoint stamp and the digest-format marker.
        let (sid, verified_bytes) = verified_marker_less_session_bytes("case4-adopt");
        let mut document: serde_json::Value =
            serde_json::from_slice(&verified_bytes).expect("decode fixture");
        document["metadata"]
            .as_object_mut()
            .expect("metadata")
            .remove(meerkat_core::SESSION_CHECKPOINT_STAMP_KEY)
            .expect("stamped fixture");
        let legacy_bytes = serde_json::to_vec(&document).expect("legacy bytes");
        assert!(!document_carries_current_marker(&legacy_bytes));
        {
            let conn = create_legacy_continuity(&continuity_db);
            insert_record(&conn, "test:alice", &sid);
            insert_snapshot(&conn, &sid, "test:alice", &legacy_bytes);
        }

        let report = migrate_state_dir(state, MigrateMode::Apply, None);
        assert!(!report.has_errors(), "{:?}", report.errors);
        let adoption = report
            .adoption
            .as_ref()
            .and_then(|outcome| outcome.report.as_ref())
            .expect("adoption walk");
        assert_eq!(adoption.adopted, 1);
        let stamping = marker_stamp_outcome(&report, "continuity")
            .report
            .as_ref()
            .expect("marker-stamping walk");
        assert_eq!(
            stamping.already_current, 1,
            "adoption's rewrite already carries the marker; case 6 must not respell it"
        );
        assert_eq!(stamping.stamped, 0);

        let conn = Connection::open(&continuity_db).expect("reopen");
        let bytes: Vec<u8> = conn
            .query_row(
                "SELECT data FROM session_snapshots WHERE session_id = ?1",
                [&sid],
                |row| row.get(0),
            )
            .expect("adopted row");
        assert!(
            document_carries_current_marker(&bytes),
            "adopted documents must carry the digest-format marker"
        );
        let session: meerkat_core::Session =
            serde_json::from_slice(&bytes).expect("decode adopted document");
        assert!(matches!(
            session.try_checkpoint_state().expect("checkpoint state"),
            meerkat_core::SessionCheckpointState::Verified(_)
        ));
    }

    #[test]
    fn twin_dry_run_reports_row_divergence_and_refuses() {
        let temp = tempfile::tempdir().expect("tempdir");
        let state = temp.path();
        let db_a = state.join("continuity.db");
        let db_b = state.join("continuity.sqlite3");
        {
            let conn = create_legacy_continuity(&db_a);
            insert_snapshot(&conn, "s-shared", "test:alice", b"same");
            insert_snapshot(&conn, "s-divergent", "test:alice", b"version-a");
            insert_snapshot(&conn, "s-only-a", "test:alice", b"solo");
        }
        {
            let conn = create_legacy_continuity(&db_b);
            insert_snapshot(&conn, "s-shared", "test:alice", b"same");
            insert_snapshot(&conn, "s-divergent", "test:alice", b"version-b");
        }

        let report = migrate_state_dir(state, MigrateMode::DryRun, None);
        assert!(report.has_errors(), "twins must fail closed");
        assert!(
            report.errors[0].contains("continuity"),
            "{:?}",
            report.errors
        );
        // Fail-closed: the divergence report is the whole output.
        assert!(report.ledger.is_empty());
        assert!(report.renames.is_empty());
        assert!(report.adoption.is_none());

        let twin = &report.twins[0];
        assert_eq!(twin.slot, "continuity");
        assert!(!twin.byte_identical);
        assert_eq!(twin.rows_equal, 1, "{:?}", twin.rows);
        let status_of = |key: &str| {
            twin.rows
                .iter()
                .find(|row| row.key == format!("session_snapshots/{key}"))
                .map(|row| row.status.clone())
        };
        assert_eq!(status_of("s-divergent"), Some(DivergenceStatus::Divergent));
        assert_eq!(
            status_of("s-only-a"),
            Some(DivergenceStatus::OnlyIn { location: db_a })
        );
        assert!(matches!(
            twin.resolution,
            TwinResolution::Refused { ref reason } if reason.contains("--adopt")
        ));
        assert!(
            twin.notes
                .iter()
                .any(|note| note.contains("fencing tokens")),
            "{:?}",
            twin.notes
        );
        // Both copies untouched.
        assert!(db_b.is_file());
    }

    #[test]
    fn twin_apply_with_adopt_archives_read_only_and_canonicalizes() {
        let temp = tempfile::tempdir().expect("tempdir");
        let state = temp.path();
        let legacy = state.join("continuity.db");
        let canonical = state.join("continuity.sqlite3");
        {
            let conn = create_legacy_continuity(&legacy);
            insert_snapshot(&conn, "s-keep", "test:alice", b"keep-me");
        }
        {
            let conn = create_legacy_continuity(&canonical);
            insert_snapshot(&conn, "s-lose", "test:alice", b"archive-me");
        }
        let other_digest = file_digest_hex(&canonical);

        // Apply without --adopt still refuses divergent twins.
        let refused = migrate_state_dir(state, MigrateMode::Apply, None);
        assert!(refused.has_errors());

        let report = migrate_state_dir(state, MigrateMode::Apply, Some(&legacy));
        assert!(!report.has_errors(), "{:?}", report.errors);
        let twin = &report.twins[0];
        let TwinResolution::Adopted { adopted, archived } = &twin.resolution else {
            panic!("expected adoption, got {:?}", twin.resolution);
        };
        assert_eq!(
            adopted, &canonical,
            "adopted copy lands at the canonical name"
        );
        assert_eq!(archived.len(), 1);
        let archive = &archived[0];
        assert!(archive.is_file());
        let archive_name = archive.file_name().and_then(|name| name.to_str()).unwrap();
        assert!(
            is_registered_backup_artifact_name(archive_name),
            "{archive_name}"
        );
        assert!(archive_name.ends_with(".twin"), "{archive_name}");
        assert!(
            fs::metadata(archive)
                .expect("archive meta")
                .permissions()
                .readonly(),
            "archive must be read-only"
        );
        assert_eq!(
            file_digest_hex(archive),
            other_digest,
            "archived content preserved exactly"
        );

        // The adopted content lives at the canonical path; the resolver and
        // a fresh migrate see no twins.
        let conn = Connection::open(&canonical).expect("reopen canonical");
        let data: Vec<u8> = conn
            .query_row(
                "SELECT data FROM session_snapshots WHERE session_id = 's-keep'",
                [],
                |row| row.get(0),
            )
            .expect("adopted row");
        assert_eq!(data, b"keep-me");
        drop(conn);
        assert!(!legacy.exists());
        let layout = MobKitStorageLayout::with_injected_roots(state.to_path_buf(), None);
        assert!(layout.continuity_db().is_ok());
        let again = migrate_state_dir(state, MigrateMode::DryRun, None);
        assert!(again.twins.is_empty(), "{:?}", again.twins);
    }

    #[test]
    fn byte_identical_twins_dedup_under_plain_apply() {
        let temp = tempfile::tempdir().expect("tempdir");
        let state = temp.path();
        let canonical = state.join("mobkit_metadata.sqlite3");
        drop(SqliteMetadataStore::open(&canonical).expect("metadata fixture"));
        let legacy = state.join("mobkit_metadata.sqlite");
        fs::copy(&canonical, &legacy).expect("copy twin");

        let report = migrate_state_dir(state, MigrateMode::Apply, None);
        assert!(!report.has_errors(), "{:?}", report.errors);
        let twin = report
            .twins
            .iter()
            .find(|twin| twin.slot == "metadata")
            .expect("metadata twin");
        assert!(twin.byte_identical);
        let TwinResolution::Deduped { kept, archived } = &twin.resolution else {
            panic!("expected dedup, got {:?}", twin.resolution);
        };
        assert_eq!(kept, &canonical);
        // The archive is the database plus any `-wal` / `-shm` sidecars that
        // existed at archive time (read-only divergence opens materialize
        // empty ones on a WAL-mode database) — sidecars always travel with
        // their database.
        assert!(
            archived[0]
                .file_name()
                .and_then(|name| name.to_str())
                .unwrap()
                .ends_with(".twin-dedup")
        );
        assert!(
            archived.iter().skip(1).all(|path| {
                let name = path.file_name().and_then(|name| name.to_str()).unwrap();
                name.ends_with("-wal") || name.ends_with("-shm")
            }),
            "{archived:?}"
        );
        assert!(canonical.is_file());
        assert!(!legacy.exists());
    }

    #[test]
    fn adopt_path_outside_the_twin_refuses() {
        let temp = tempfile::tempdir().expect("tempdir");
        let state = temp.path();
        {
            let conn = create_legacy_continuity(&state.join("continuity.db"));
            insert_snapshot(&conn, "s-a", "test:alice", b"a");
        }
        {
            let conn = create_legacy_continuity(&state.join("continuity.sqlite3"));
            insert_snapshot(&conn, "s-b", "test:alice", b"b");
        }
        let elsewhere = state.join("unrelated.db");
        fs::write(&elsewhere, b"x").expect("unrelated");

        let report = migrate_state_dir(state, MigrateMode::Apply, Some(&elsewhere));
        assert!(report.has_errors());
        assert!(matches!(
            report.twins[0].resolution,
            TwinResolution::Refused { ref reason } if reason.contains("candidates")
        ));
        assert!(state.join("continuity.db").is_file(), "nothing moved");
        assert!(state.join("continuity.sqlite3").is_file());
    }

    #[test]
    fn leftovers_census_reports_blobs_sidecar_artifacts_and_dead_registry_entries() {
        let temp = tempfile::tempdir().expect("tempdir");
        let state = temp.path();
        // Legacy sharded-FS blob layout beside the current object layout.
        let shard = state.join("blobs").join("aa");
        fs::create_dir_all(&shard).expect("shard");
        fs::write(shard.join(format!("{}.json", "a".repeat(64))), b"{}").expect("legacy blob");
        drop(Connection::open(state.join(WORKGRAPH_ADMISSION_SIDECAR_FILE)).expect("sidecar"));
        fs::write(state.join("continuity.db.pre-0.0.1-1700000000"), b"backup")
            .expect("backup artifact");
        // A dead pid: a reaped child. A live pid: this process.
        let mut child = std::process::Command::new("true").spawn().expect("spawn");
        let dead_pid = child.id();
        child.wait().expect("reap");
        fs::write(
            state.join(RUNTIME_REGISTRY_FILE_NAME),
            serde_json::to_vec(&serde_json::json!({
                "entries": [
                    { "key": "k1", "runtime_id": "tux-dead", "http_base_url": "http://127.0.0.1:1",
                      "pid": dead_pid, "updated_at_ms": 0 },
                    { "key": "k2", "runtime_id": "tux-live", "http_base_url": "http://127.0.0.1:2",
                      "pid": std::process::id(), "updated_at_ms": 0 },
                ]
            }))
            .expect("registry json"),
        )
        .expect("write registry");

        let report = migrate_state_dir(state, MigrateMode::DryRun, None);
        assert!(!report.has_errors(), "{:?}", report.errors);
        let codes: Vec<&str> = report
            .findings
            .iter()
            .map(|finding| finding.code.as_str())
            .collect();
        for expected in [
            storage_doctor::FINDING_LEGACY_FS_BLOBS,
            storage_doctor::FINDING_WORKGRAPH_ADMISSION_SIDECAR,
            storage_doctor::FINDING_BACKUP_ARTIFACT,
            FINDING_DEAD_RUNTIME_REGISTRY_ENTRY,
        ] {
            assert!(codes.contains(&expected), "missing {expected}: {codes:?}");
        }
        let dead: Vec<_> = report
            .findings
            .iter()
            .filter(|finding| finding.code == FINDING_DEAD_RUNTIME_REGISTRY_ENTRY)
            .collect();
        assert_eq!(dead.len(), 1, "only the dead entry is flagged: {dead:?}");
        assert!(dead[0].message.contains("tux-dead"));
    }

    #[test]
    fn strict_artifact_name_validation_rejects_lookalikes() {
        // Everything this run generates validates, including sibling
        // archives and purposeless names.
        for valid in [
            backup_artifact_name("sessions.sqlite3", "").as_str(),
            backup_artifact_name("continuity.db", "renamed").as_str(),
            backup_artifact_name("team", "split-brain").as_str(),
            "sessions.sqlite3.pre-0.8.3-1700000000",
            "continuity.db.pre-0.8.2-1700000000.renamed",
            "mobkit_metadata.sqlite.pre-0.8.3-123.twin-dedup-wal",
            "agent-memory-sqlite.pre-0.0.1-1700000000.twin",
        ] {
            assert!(is_registered_backup_artifact_name(valid), "{valid}");
        }
        // Loose `.pre-` substrings are NOT registered artifacts — prune's
        // deletion authority must never claim user files.
        for invalid in [
            "notes.pre-release",
            "data.pre-view.txt",
            ".pre-0.8.3-1700000000",
            "foo.pre-1-1700000000",
            "foo.pre-0..3-1700000000",
            "foo.pre-0.8.3-17000x",
            "foo.pre-0.8.3-",
            "foo.pre-0.8.3-1700000000.",
            "sessions.sqlite3",
        ] {
            assert!(!is_registered_backup_artifact_name(invalid), "{invalid}");
        }
        for valid in [
            "alpha.sqlite3.corrupt-42",
            "session_index.sqlite3.corrupt-1700000000",
        ] {
            assert!(is_registered_quarantine_artifact_name(valid), "{valid}");
        }
        for invalid in [
            "report.corrupt-12a",
            "x.corrupt-",
            ".corrupt-42",
            "a.corrupt-1.bak",
            "notes.corrupted-1",
        ] {
            assert!(
                !is_registered_quarantine_artifact_name(invalid),
                "{invalid}"
            );
        }
    }

    #[test]
    fn prune_respects_threshold_and_registered_patterns_only() {
        let temp = tempfile::tempdir().expect("tempdir");
        let state = temp.path();
        let old_backup = state.join("continuity.db.pre-0.0.1-1700000000.renamed");
        fs::write(&old_backup, b"old").expect("old backup");
        let old_time = std::time::SystemTime::now() - Duration::from_hours(40 * 24);
        fs::File::options()
            .write(true)
            .open(&old_backup)
            .expect("open for mtime")
            .set_modified(old_time)
            .expect("set mtime");
        // An old archived memory-root DIRECTORY.
        let old_dir = state.join("agent-memory-sqlite.pre-0.0.1-1700000000.twin");
        fs::create_dir_all(&old_dir).expect("old dir");
        fs::write(old_dir.join("alpha.sqlite3"), b"realm").expect("realm bytes");
        fs::File::options()
            .write(true)
            .open(old_dir.join("alpha.sqlite3"))
            .expect("open for mtime")
            .set_modified(old_time)
            .expect("set inner mtime");
        // Directory mtimes: setting via the contained file is not enough on
        // all platforms; force the dir mtime through a fresh handle.
        fs::File::open(&old_dir)
            .expect("open dir")
            .set_modified(old_time)
            .expect("set dir mtime");
        let young_quarantine = state.join("alpha.sqlite3.corrupt-42");
        fs::write(&young_quarantine, b"young").expect("young quarantine");
        let distractor = state.join("notes.txt");
        fs::write(&distractor, b"keep me").expect("distractor");
        // Lookalikes with a `.pre-` / `.corrupt-` substring but not the
        // registered full shape: old enough to delete, never enumerated.
        let lookalike_backup = state.join("notes.pre-release");
        let lookalike_quarantine = state.join("report.corrupt-12a");
        for lookalike in [&lookalike_backup, &lookalike_quarantine] {
            fs::write(lookalike, b"user file").expect("lookalike");
            fs::File::options()
                .write(true)
                .open(lookalike)
                .expect("open for mtime")
                .set_modified(old_time)
                .expect("set lookalike mtime");
        }

        let dry = prune_state_dir(state, 30, MigrateMode::DryRun);
        assert!(!dry.has_errors());
        assert_eq!(dry.artifacts.len(), 3, "{:?}", dry.artifacts);
        let action_of = |report: &MobKitPruneReport, path: &Path| {
            report
                .artifacts
                .iter()
                .find(|artifact| artifact.path == path)
                .map(|artifact| artifact.action)
                .unwrap_or_else(|| panic!("no artifact entry for {}", path.display()))
        };
        assert_eq!(action_of(&dry, &old_backup), PruneAction::WouldDelete);
        assert_eq!(action_of(&dry, &old_dir), PruneAction::WouldDelete);
        assert_eq!(action_of(&dry, &young_quarantine), PruneAction::Kept);
        assert!(old_backup.is_file(), "dry-run deletes nothing");

        let applied = prune_state_dir(state, 30, MigrateMode::Apply);
        assert!(!applied.has_errors(), "{:?}", applied.errors);
        assert_eq!(action_of(&applied, &old_backup), PruneAction::Deleted);
        assert_eq!(action_of(&applied, &old_dir), PruneAction::Deleted);
        assert!(!old_backup.exists());
        assert!(!old_dir.exists());
        assert!(young_quarantine.is_file(), "young artifacts are kept");
        assert!(distractor.is_file(), "unregistered names are never touched");

        // Threshold 0 = everything registered; lookalikes still survive.
        let sweep = prune_state_dir(state, 0, MigrateMode::Apply);
        assert!(!sweep.has_errors());
        assert!(!young_quarantine.exists());
        assert!(
            lookalike_backup.is_file() && lookalike_quarantine.is_file(),
            "names outside the registered full shape are never enumerated or deleted"
        );
    }

    #[test]
    fn report_shapes_round_trip_through_json() {
        let mut report = MobKitMigrateReport::new(MigrateMode::Apply, Path::new("/state"));
        report.ledger.push(LedgerBaselineEntry {
            database: PathBuf::from("/state/continuity.sqlite3"),
            domain: "mobkit-continuity".to_string(),
            before: None,
            after: Some(1),
            action: LedgerBaselineAction::Stamped,
        });
        report.renames.push(FileRenameEntry {
            slot: "continuity".to_string(),
            from: PathBuf::from("/state/continuity.db"),
            to: PathBuf::from("/state/continuity.sqlite3"),
            siblings: vec![SiblingRename {
                from: PathBuf::from("/state/continuity.db-wal"),
                to: PathBuf::from("/state/continuity.sqlite3-wal"),
            }],
            wal_checkpointed: true,
            marker: Some(PathBuf::from("/state/continuity.db.pre-0.8.2-1.renamed")),
            action: RenameAction::Renamed,
        });
        report.twins.push(TwinReport {
            slot: "console".to_string(),
            paths: vec![PathBuf::from("/state/mobkit_console.sqlite")],
            byte_identical: false,
            rows_equal: 2,
            rows: vec![RowDivergenceEntry {
                key: "console_frames/7".to_string(),
                status: DivergenceStatus::Divergent,
            }],
            resolution: TwinResolution::Adopted {
                adopted: PathBuf::from("/state/mobkit_console.sqlite3"),
                archived: vec![PathBuf::from(
                    "/state/mobkit_console.sqlite.pre-0.8.2-1.twin",
                )],
            },
            notes: vec![],
            errors: vec![],
        });
        report.adoption = Some(CheckpointAdoptionOutcome {
            database: PathBuf::from("/state/continuity.sqlite3"),
            report: Some(ContinuityAdoptionReport::default()),
            skipped: None,
        });
        let json = serde_json::to_string(&report).expect("serialize");
        let parsed: MobKitMigrateReport = serde_json::from_str(&json).expect("deserialize");
        assert!(matches!(parsed.mode, MigrateMode::Apply));
        assert_eq!(parsed.ledger[0].action, LedgerBaselineAction::Stamped);
        assert_eq!(parsed.renames[0].action, RenameAction::Renamed);
        assert!(matches!(
            parsed.twins[0].resolution,
            TwinResolution::Adopted { .. }
        ));
        assert!(!parsed.has_errors());

        // Forward compatibility: unknown fields tolerated, defaults fill in.
        let sparse: MobKitMigrateReport =
            serde_json::from_str(r#"{"future_field":1}"#).expect("sparse migrate");
        assert!(matches!(sparse.mode, MigrateMode::DryRun));
        let sparse: MobKitPruneReport =
            serde_json::from_str(r#"{"future_field":1}"#).expect("sparse prune");
        assert!(sparse.artifacts.is_empty());
    }

    #[test]
    fn missing_state_dir_errors_typed() {
        let temp = tempfile::tempdir().expect("tempdir");
        let missing = temp.path().join("nope");
        let report = migrate_state_dir(&missing, MigrateMode::DryRun, None);
        assert!(report.has_errors());
        assert!(report.errors[0].contains("does not exist"));
    }
}