diff --git a/Cargo.lock b/Cargo.lock
index 34e4ec484..41c7dd408 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -1045,6 +1045,7 @@ name = "mc-store"
 version = "0.1.0"
 dependencies = [
  "cortexkit-cache-core",
+ "cortexkit-lease",
  "cortexkit-store",
  "cortexkit-store-types",
  "flate2",
diff --git a/crates/mc-store/Cargo.toml b/crates/mc-store/Cargo.toml
index 51ac1e085..f91e5e6d8 100644
--- a/crates/mc-store/Cargo.toml
+++ b/crates/mc-store/Cargo.toml
@@ -14,6 +14,7 @@ mc-core = { workspace = true }
 cortexkit-cache-core = { workspace = true }
 cortexkit-store = { workspace = true }
 cortexkit-store-types = { workspace = true }
+cortexkit-lease = { workspace = true }
 rusqlite = { workspace = true, features = ["functions"] }
 serde = { workspace = true }
 serde_json = { workspace = true }
diff --git a/crates/mc-store/src/kernel/mod.rs b/crates/mc-store/src/kernel/mod.rs
index 1ce7e1766..a869c8539 100644
--- a/crates/mc-store/src/kernel/mod.rs
+++ b/crates/mc-store/src/kernel/mod.rs
@@ -1 +1,4 @@
+mod open;
 pub mod schema;
+
+pub use open::{KernelError, KernelStore};
diff --git a/crates/mc-store/src/kernel/open.rs b/crates/mc-store/src/kernel/open.rs
new file mode 100644
index 000000000..8f9778fcd
--- /dev/null
+++ b/crates/mc-store/src/kernel/open.rs
@@ -0,0 +1,859 @@
+use cortexkit_lease::{
+    protect_file, FileLeaseStore, LeaseError, LeaseHandle, LeaseKey, LeaseStore,
+};
+use mc_core::claim_operation::is_lower_hex;
+use rusqlite::{Connection, OpenFlags, OptionalExtension, Transaction, TransactionBehavior};
+use serde::{Deserialize, Serialize};
+use sha2::{Digest, Sha256};
+use std::fmt;
+use std::fs::{self, File, OpenOptions};
+use std::io::{ErrorKind, Read, Write};
+use std::path::{Path, PathBuf};
+use std::sync::atomic::{AtomicUsize, Ordering};
+use std::sync::{LazyLock, Mutex, PoisonError};
+
+use super::schema::{
+    apply_kernel_schema, kernel_schema_digest, kernel_schema_object_inventory,
+    KERNEL_APPLICATION_ID, KERNEL_FORMAT_EPOCH,
+};
+use crate::current_time_ms;
+use crate::sqlite_runtime::{
+    compute_marker_digest_for_application_id, evaluate_sqlite_runtime_gate,
+    probe_sqlite_engine_identity_off_path, SqliteEngineIdentity,
+};
+
+const BUSY_TIMEOUT_MS: i64 = 5_000;
+const READ_POOL_SIZE: usize = 2;
+const RESET_MARKER_PROTOCOL: &str = "mc-kernel-reset-marker-v1";
+const RESET_MARKER_SUFFIX: &str = ".mc-reset";
+const RESET_MARKER_STAGING_SUFFIX: &str = ".staging";
+const QUARANTINE_INFIX: &str = ".mc-quarantine-";
+const SQLITE_HEADER: &[u8; 16] = b"SQLite format 3\0";
+
+/// Limit marker reads to 64 KiB so invalid marker data cannot control
+/// allocation size.
+const RESET_MARKER_MAX_BYTES: u64 = 64 * 1024;
+
+#[derive(Clone, Copy, PartialEq, Eq)]
+pub enum KernelError {
+    Held,
+    EngineUnsupported,
+    Foreign,
+    Inconclusive,
+    Io,
+    IdentityMismatch,
+    FenceLost,
+}
+
+impl fmt::Display for KernelError {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.write_str(match self {
+            Self::Held => "kernel store is held by another writer",
+            Self::EngineUnsupported => "SQLite engine is unsupported for the kernel store",
+            Self::Foreign => "kernel store path contains a foreign database family",
+            Self::Inconclusive => "kernel store identity could not be established safely",
+            Self::Io => "kernel store I/O failed",
+            Self::IdentityMismatch => "kernel store identity does not match this build",
+            Self::FenceLost => "kernel store writer fence was lost",
+        })
+    }
+}
+
+impl fmt::Debug for KernelError {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        fmt::Display::fmt(self, f)
+    }
+}
+
+impl std::error::Error for KernelError {}
+
+pub struct KernelStore {
+    writer: Mutex<Connection>,
+    readers: Vec<Mutex<Connection>>,
+    next_reader: AtomicUsize,
+    lease_epoch: u64,
+    _lease: Box<dyn LeaseHandle>,
+}
+
+impl fmt::Debug for KernelStore {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.debug_struct("KernelStore")
+            .field("lease_epoch", &self.lease_epoch)
+            .field("read_pool_size", &self.readers.len())
+            .finish_non_exhaustive()
+    }
+}
+
+impl KernelStore {
+    pub fn open(root: impl AsRef<Path>) -> Result<Self, KernelError> {
+        let identity =
+            probe_sqlite_engine_identity_off_path().map_err(|_| KernelError::EngineUnsupported)?;
+        Self::open_with_engine_identity(root, &identity)
+    }
+
+    fn open_with_engine_identity(
+        root: impl AsRef<Path>,
+        identity: &SqliteEngineIdentity,
+    ) -> Result<Self, KernelError> {
+        if !evaluate_sqlite_runtime_gate(identity).is_empty() {
+            return Err(KernelError::EngineUnsupported);
+        }
+        Self::open_supported(root)
+    }
+
+    #[cfg(feature = "test-support")]
+    pub fn open_with_engine_identity_for_test(
+        root: impl AsRef<Path>,
+        identity: &SqliteEngineIdentity,
+    ) -> Result<Self, KernelError> {
+        Self::open_with_engine_identity(root, identity)
+    }
+
+    fn open_supported(root: impl AsRef<Path>) -> Result<Self, KernelError> {
+        let root = prepare_root(root.as_ref())?;
+        let db_path = root.join("core.sqlite");
+        let lease_store = FileLeaseStore::new(root.join("leases"));
+        let lease_key = LeaseKey::new("magic-context-kernel", "sqlite", "core");
+        let lease = lease_store.acquire(&lease_key).map_err(map_lease_error)?;
+        let lease_epoch = lease.epoch();
+
+        if entry_exists(&reset_marker_path(&db_path))? {
+            resume_quarantine(&db_path)?;
+        }
+
+        let header = inspect_header(&db_path)?;
+        let mut writer = match header {
+            HeaderState::Pristine => bootstrap(&db_path)?,
+            HeaderState::Kernel => match classify_existing_family(&db_path)? {
+                OpenIdentity::Exact => {
+                    let conn = open_writer(&db_path).map_err(|_| KernelError::Inconclusive)?;
+                    apply_preclassification_profile(&conn)
+                        .map_err(|_| KernelError::Inconclusive)?;
+                    conn
+                }
+                OpenIdentity::Mismatch { incarnation } => {
+                    quarantine(&db_path, &incarnation, lease_epoch)?;
+                    bootstrap(&db_path)?
+                }
+            },
+        };
+
+        activate_wal(&writer)?;
+        stamp_writer_fence(&mut writer, lease_epoch)?;
+        // Two hardening passes are required because the family grows between
+        // them. WAL activation creates `-wal` and `-shm` under the process umask,
+        // so the first pass restricts them as early as possible; a read-only open
+        // recreates `-shm` when it is missing, so the second pass restricts that
+        // one too.
+        harden_family(&db_path)?;
+        let readers = open_read_pool(&db_path)?;
+        harden_family(&db_path)?;
+
+        Ok(Self {
+            writer: Mutex::new(writer),
+            readers,
+            next_reader: AtomicUsize::new(0),
+            lease_epoch,
+            _lease: lease,
+        })
+    }
+
+    pub fn lease_epoch(&self) -> u64 {
+        self.lease_epoch
+    }
+
+    #[allow(
+        dead_code,
+        reason = "connection ownership is restricted to kernel mutation modules"
+    )]
+    pub(crate) fn with_writer<T>(
+        &self,
+        operation: impl FnOnce(&Transaction<'_>) -> rusqlite::Result<T>,
+    ) -> Result<T, KernelError> {
+        let mut writer = self.writer.lock().unwrap_or_else(PoisonError::into_inner);
+        // `BEGIN IMMEDIATE` blocks a competing writer until this transaction ends.
+        // The fence check and the mutation are therefore atomic.
+        let tx = writer
+            .transaction_with_behavior(TransactionBehavior::Immediate)
+            .map_err(|_| KernelError::Io)?;
+        let durable_epoch: i64 = tx
+            .query_row(
+                "SELECT writer_epoch FROM writer_fence WHERE id=0",
+                [],
+                |row| row.get(0),
+            )
+            .map_err(|_| KernelError::FenceLost)?;
+        if u64::try_from(durable_epoch).ok() != Some(self.lease_epoch) {
+            return Err(KernelError::FenceLost);
+        }
+        let value = operation(&tx).map_err(|_| KernelError::Io)?;
+        tx.commit().map_err(|_| KernelError::Io)?;
+        Ok(value)
+    }
+
+    #[allow(
+        dead_code,
+        reason = "connection ownership is restricted to kernel query modules"
+    )]
+    pub(crate) fn with_reader<T>(
+        &self,
+        operation: impl FnOnce(&Connection) -> rusqlite::Result<T>,
+    ) -> Result<T, KernelError> {
+        let index = self.next_reader.fetch_add(1, Ordering::Relaxed) % self.readers.len();
+        // Recovering a poisoned reader guard keeps its pool slot usable.
+        // `Transaction::Drop` rolls back after a closure panic.
+        let mut reader = self.readers[index]
+            .lock()
+            .unwrap_or_else(PoisonError::into_inner);
+        // After its first read, the transaction keeps one snapshot for the
+        // closure. Several reads therefore observe one consistent state.
+        let tx = reader.transaction().map_err(|_| KernelError::Io)?;
+        let value = operation(&tx).map_err(|_| KernelError::Io)?;
+        tx.commit().map_err(|_| KernelError::Io)?;
+        Ok(value)
+    }
+}
+
+enum HeaderState {
+    Pristine,
+    Kernel,
+}
+
+/// `Path::exists` maps every error to `false`; only `NotFound` counts as absence.
+fn entry_exists(path: &Path) -> Result<bool, KernelError> {
+    match fs::symlink_metadata(path) {
+        Ok(_) => Ok(true),
+        Err(error) if error.kind() == ErrorKind::NotFound => Ok(false),
+        Err(_) => Err(KernelError::Io),
+    }
+}
+
+/// A zero-length main file has never held a committed page.
+///
+/// A `-journal` does not prevent treating an empty main file as pristine.
+fn classify_empty_family(path: &Path) -> Result<HeaderState, KernelError> {
+    if entry_exists(&suffix_path(path, "-wal"))? || entry_exists(&suffix_path(path, "-shm"))? {
+        return Err(KernelError::Inconclusive);
+    }
+    Ok(HeaderState::Pristine)
+}
+
+fn inspect_header(path: &Path) -> Result<HeaderState, KernelError> {
+    let metadata = match fs::symlink_metadata(path) {
+        Ok(metadata) => metadata,
+        Err(error) if error.kind() == ErrorKind::NotFound => return classify_empty_family(path),
+        Err(_) => return Err(KernelError::Io),
+    };
+    if !metadata.is_file() {
+        return Err(KernelError::Inconclusive);
+    }
+    if metadata.len() == 0 {
+        return classify_empty_family(path);
+    }
+    if metadata.len() < 100 {
+        return Err(KernelError::Inconclusive);
+    }
+    let mut header = [0_u8; 100];
+    File::open(path)
+        .and_then(|mut file| file.read_exact(&mut header))
+        .map_err(|_| KernelError::Inconclusive)?;
+    if &header[..16] != SQLITE_HEADER {
+        return Err(KernelError::Foreign);
+    }
+    let application_id =
+        u32::from_be_bytes(header[68..72].try_into().map_err(|_| KernelError::Io)?);
+    if application_id != KERNEL_APPLICATION_ID {
+        return Err(KernelError::Foreign);
+    }
+    Ok(HeaderState::Kernel)
+}
+
+struct ExpectedIdentity {
+    digest: String,
+    inventory: Vec<(String, String)>,
+}
+
+static EXPECTED_IDENTITY: LazyLock<Option<ExpectedIdentity>> = LazyLock::new(|| {
+    let mut conn = Connection::open_in_memory().ok()?;
+    apply_kernel_schema(&mut conn, "00000000000000000000000000000000", 0).ok()?;
+    Some(ExpectedIdentity {
+        digest: kernel_schema_digest(&conn).ok()?,
+        inventory: kernel_schema_object_inventory(&conn).ok()?,
+    })
+});
+
+fn expected_identity() -> Result<&'static ExpectedIdentity, KernelError> {
+    EXPECTED_IDENTITY.as_ref().ok_or(KernelError::Io)
+}
+
+/// Classification must leave the database and its `-wal` byte-identical.
+///
+/// The `Foreign` and `Inconclusive` outcomes promise untouched durable content.
+/// Journal recovery and WAL checkpointing would both break that promise.
+///
+/// A read-only open may recreate a missing `-shm`; it contains no durable data.
+/// SQLite rebuilds the `-shm` from the `-wal` on demand.
+fn classify_existing_family(path: &Path) -> Result<OpenIdentity, KernelError> {
+    let expected = expected_identity()?;
+    let mut conn = Connection::open_with_flags(
+        path,
+        OpenFlags::SQLITE_OPEN_READ_ONLY | OpenFlags::SQLITE_OPEN_NO_MUTEX,
+    )
+    .map_err(|_| KernelError::Inconclusive)?;
+    conn.pragma_update(None, "query_only", "ON")
+        .map_err(|_| KernelError::Inconclusive)?;
+    conn.pragma_update(None, "trusted_schema", "OFF")
+        .map_err(|_| KernelError::Inconclusive)?;
+    conn.pragma_update(None, "busy_timeout", BUSY_TIMEOUT_MS)
+        .map_err(|_| KernelError::Inconclusive)?;
+    classify_open_kernel(&mut conn, expected)
+}
+
+enum OpenIdentity {
+    Exact,
+    Mismatch { incarnation: String },
+}
+
+struct FormatMarker {
+    epoch: i64,
+    incarnation: String,
+    schema_digest: String,
+    created_at: i64,
+    marker_digest: String,
+}
+
+fn classify_open_kernel(
+    conn: &mut Connection,
+    expected: &ExpectedIdentity,
+) -> Result<OpenIdentity, KernelError> {
+    let tx = conn.transaction().map_err(|_| KernelError::Inconclusive)?;
+    let quick_check: String = tx
+        .query_row("PRAGMA quick_check(1)", [], |row| row.get(0))
+        .map_err(|_| KernelError::Inconclusive)?;
+    if quick_check != "ok" {
+        return Err(KernelError::Inconclusive);
+    }
+    let marker = read_valid_marker(&tx)?;
+    let application_id: u32 = tx
+        .query_row("PRAGMA application_id", [], |row| row.get(0))
+        .map_err(|_| KernelError::Inconclusive)?;
+    let user_version: i64 = tx
+        .query_row("PRAGMA user_version", [], |row| row.get(0))
+        .map_err(|_| KernelError::Inconclusive)?;
+    let inventory = kernel_schema_object_inventory(&tx).map_err(|_| KernelError::Inconclusive)?;
+    let digest = kernel_schema_digest(&tx).map_err(|_| KernelError::Inconclusive)?;
+    let exact = application_id == KERNEL_APPLICATION_ID
+        && user_version == KERNEL_FORMAT_EPOCH
+        && marker.epoch == KERNEL_FORMAT_EPOCH
+        && marker.schema_digest == expected.digest
+        && digest == expected.digest
+        && inventory == expected.inventory;
+    tx.commit().map_err(|_| KernelError::Inconclusive)?;
+    if exact {
+        Ok(OpenIdentity::Exact)
+    } else {
+        Ok(OpenIdentity::Mismatch {
+            incarnation: marker.incarnation,
+        })
+    }
+}
+
+fn read_valid_marker(conn: &Connection) -> Result<FormatMarker, KernelError> {
+    let present: Option<i64> = conn
+        .query_row(
+            "SELECT 1 FROM sqlite_schema WHERE type='table' AND name='mc_kernel_format_marker'",
+            [],
+            |row| row.get(0),
+        )
+        .optional()
+        .map_err(|_| KernelError::Inconclusive)?;
+    if present.is_none() {
+        return Err(KernelError::Inconclusive);
+    }
+    // A lookalike table has no singleton constraint, so `LIMIT 2` detects multiple
+    // rows.
+    let mut statement = conn
+        .prepare(
+            "SELECT format_epoch,database_incarnation_id,schema_digest,created_at,marker_digest
+             FROM mc_kernel_format_marker
+             WHERE length(database_incarnation_id)=32
+               AND length(schema_digest)=64
+               AND length(marker_digest)=64
+             LIMIT 2",
+        )
+        .map_err(|_| KernelError::Inconclusive)?;
+    let rows = statement
+        .query_map([], |row| {
+            Ok(FormatMarker {
+                epoch: row.get(0)?,
+                incarnation: row.get(1)?,
+                schema_digest: row.get(2)?,
+                created_at: row.get(3)?,
+                marker_digest: row.get(4)?,
+            })
+        })
+        .map_err(|_| KernelError::Inconclusive)?
+        .collect::<rusqlite::Result<Vec<_>>>()
+        .map_err(|_| KernelError::Inconclusive)?;
+    let [marker] = rows.as_slice() else {
+        return Err(KernelError::Inconclusive);
+    };
+    if !is_lower_hex(&marker.incarnation, 32)
+        || !is_lower_hex(&marker.schema_digest, 64)
+        || !is_lower_hex(&marker.marker_digest, 64)
+        || marker.epoch < 1
+        || marker.created_at < 0
+    {
+        return Err(KernelError::Inconclusive);
+    }
+    let expected_marker_digest = compute_marker_digest_for_application_id(
+        KERNEL_APPLICATION_ID,
+        marker.epoch,
+        &marker.incarnation,
+        &marker.schema_digest,
+        marker.created_at,
+    );
+    if marker.marker_digest != expected_marker_digest {
+        return Err(KernelError::Inconclusive);
+    }
+    Ok(FormatMarker {
+        epoch: marker.epoch,
+        incarnation: marker.incarnation.clone(),
+        schema_digest: marker.schema_digest.clone(),
+        created_at: marker.created_at,
+        marker_digest: marker.marker_digest.clone(),
+    })
+}
+
+fn open_writer(path: &Path) -> rusqlite::Result<Connection> {
+    Connection::open_with_flags(
+        path,
+        OpenFlags::SQLITE_OPEN_READ_WRITE
+            | OpenFlags::SQLITE_OPEN_CREATE
+            | OpenFlags::SQLITE_OPEN_NO_MUTEX,
+    )
+}
+
+fn bootstrap(path: &Path) -> Result<Connection, KernelError> {
+    let mut conn = open_writer(path).map_err(|_| KernelError::Io)?;
+    apply_preclassification_profile(&conn).map_err(|_| KernelError::Io)?;
+    let incarnation: String = conn
+        .query_row("SELECT lower(hex(randomblob(16)))", [], |row| row.get(0))
+        .map_err(|_| KernelError::Io)?;
+    apply_kernel_schema(&mut conn, &incarnation, current_time_ms()).map_err(|_| KernelError::Io)?;
+    Ok(conn)
+}
+
+fn apply_preclassification_profile(conn: &Connection) -> rusqlite::Result<()> {
+    conn.pragma_update(None, "foreign_keys", "ON")?;
+    conn.pragma_update(None, "trusted_schema", "OFF")?;
+    conn.pragma_update(None, "busy_timeout", BUSY_TIMEOUT_MS)?;
+    // REPLACE deletes the conflicting row, which fires its BEFORE DELETE trigger
+    // only when recursive_triggers is ON.
+    conn.pragma_update(None, "recursive_triggers", "ON")?;
+    Ok(())
+}
+
+/// `PRAGMA journal_mode` returns the mode now in effect rather than failing.
+///
+/// Readers depend on WAL for snapshot isolation, so the returned mode is checked.
+fn activate_wal(conn: &Connection) -> Result<(), KernelError> {
+    let mode: String = conn
+        .query_row("PRAGMA journal_mode=WAL", [], |row| row.get(0))
+        .map_err(|_| KernelError::Io)?;
+    if !mode.eq_ignore_ascii_case("wal") {
+        return Err(KernelError::Io);
+    }
+    conn.pragma_update(None, "synchronous", "FULL")
+        .map_err(|_| KernelError::Io)
+}
+
+fn stamp_writer_fence(conn: &mut Connection, epoch: u64) -> Result<(), KernelError> {
+    let epoch = i64::try_from(epoch).map_err(|_| KernelError::IdentityMismatch)?;
+    let tx = conn.transaction().map_err(|_| KernelError::Io)?;
+    if tx
+        .execute(
+            "UPDATE writer_fence SET writer_epoch=?1 WHERE id=0",
+            [epoch],
+        )
+        .map_err(|_| KernelError::Io)?
+        != 1
+    {
+        return Err(KernelError::IdentityMismatch);
+    }
+    tx.commit().map_err(|_| KernelError::Io)
+}
+
+fn open_read_pool(path: &Path) -> Result<Vec<Mutex<Connection>>, KernelError> {
+    (0..READ_POOL_SIZE)
+        .map(|_| {
+            let conn = Connection::open_with_flags(
+                path,
+                OpenFlags::SQLITE_OPEN_READ_ONLY | OpenFlags::SQLITE_OPEN_NO_MUTEX,
+            )
+            .map_err(|_| KernelError::Io)?;
+            conn.pragma_update(None, "query_only", "ON")
+                .map_err(|_| KernelError::Io)?;
+            apply_preclassification_profile(&conn).map_err(|_| KernelError::Io)?;
+            Ok(Mutex::new(conn))
+        })
+        .collect()
+}
+
+fn family_sidecars(path: &Path) -> [PathBuf; 3] {
+    [
+        suffix_path(path, "-wal"),
+        suffix_path(path, "-shm"),
+        suffix_path(path, "-journal"),
+    ]
+}
+
+fn harden_family(path: &Path) -> Result<(), KernelError> {
+    protect_file(path).map_err(|_| KernelError::Io)?;
+    for sidecar in family_sidecars(path) {
+        protect_file(&sidecar).map_err(|_| KernelError::Io)?;
+    }
+    Ok(())
+}
+
+#[derive(Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+struct ResetMarker {
+    protocol: String,
+    db_path: PathBuf,
+    database_incarnation_id: String,
+    quarantine_dir: PathBuf,
+    marker_digest: String,
+}
+
+fn quarantine(path: &Path, incarnation: &str, lease_epoch: u64) -> Result<(), KernelError> {
+    let quarantine_dir = allocate_quarantine_dir(path, lease_epoch)?;
+    let mut marker = ResetMarker {
+        protocol: RESET_MARKER_PROTOCOL.to_string(),
+        db_path: path.to_path_buf(),
+        database_incarnation_id: incarnation.to_string(),
+        quarantine_dir,
+        marker_digest: String::new(),
+    };
+    marker.marker_digest = reset_marker_digest(&marker);
+    publish_reset_marker(path, &marker)?;
+    move_family(path, &marker)
+}
+
+fn resume_quarantine(path: &Path) -> Result<(), KernelError> {
+    let marker_path = reset_marker_path(path);
+    let metadata = fs::symlink_metadata(&marker_path).map_err(|_| KernelError::Inconclusive)?;
+    if !metadata.is_file() || metadata.len() > RESET_MARKER_MAX_BYTES {
+        return Err(KernelError::Inconclusive);
+    }
+    let bytes = fs::read(&marker_path).map_err(|_| KernelError::Inconclusive)?;
+    let marker: ResetMarker =
+        serde_json::from_slice(&bytes).map_err(|_| KernelError::Inconclusive)?;
+    if marker.protocol != RESET_MARKER_PROTOCOL
+        || marker.db_path != path
+        || !is_lower_hex(&marker.database_incarnation_id, 32)
+        || marker.marker_digest != reset_marker_digest(&marker)
+        || !valid_quarantine_path(path, &marker.quarantine_dir)
+    {
+        return Err(KernelError::Inconclusive);
+    }
+    move_family(path, &marker)
+}
+
+/// The live marker name appears only once its bytes are durable.
+fn publish_reset_marker(path: &Path, marker: &ResetMarker) -> Result<(), KernelError> {
+    let marker_path = reset_marker_path(path);
+    let staging = suffix_path(&marker_path, RESET_MARKER_STAGING_SUFFIX);
+    let bytes = serde_json::to_vec(marker).map_err(|_| KernelError::Io)?;
+    write_private_file(&staging, &bytes)?;
+    fs::rename(&staging, &marker_path).map_err(|_| KernelError::Io)?;
+    protect_file(&marker_path).map_err(|_| KernelError::Io)?;
+    sync_parent(path)?;
+    Ok(())
+}
+
+fn write_private_file(path: &Path, bytes: &[u8]) -> Result<(), KernelError> {
+    let mut options = OpenOptions::new();
+    options.write(true).create(true).truncate(true);
+    #[cfg(unix)]
+    {
+        use std::os::unix::fs::OpenOptionsExt;
+        options.mode(0o600);
+    }
+    let outcome = options.open(path).and_then(|mut file| {
+        file.write_all(bytes)?;
+        file.sync_all()
+    });
+    if outcome.is_err() {
+        let _ = fs::remove_file(path);
+        return Err(KernelError::Io);
+    }
+    Ok(())
+}
+
+/// The marker's absence declares the reset complete.
+///
+/// Move the marker only after both directories are durable.
+fn move_family(path: &Path, marker: &ResetMarker) -> Result<(), KernelError> {
+    prepare_private_dir(&marker.quarantine_dir)?;
+    for source in family_sidecars(path)
+        .into_iter()
+        .chain([path.to_path_buf()])
+    {
+        move_one(&source, &marker.quarantine_dir)?;
+    }
+    sync_directory(&marker.quarantine_dir)?;
+    sync_parent(path)?;
+    move_one(&reset_marker_path(path), &marker.quarantine_dir)?;
+    sync_directory(&marker.quarantine_dir)?;
+    sync_parent(path)
+}
+
+fn move_one(source: &Path, destination_dir: &Path) -> Result<(), KernelError> {
+    let name = source.file_name().ok_or(KernelError::Inconclusive)?;
+    let destination = destination_dir.join(name);
+    match (entry_exists(source)?, entry_exists(&destination)?) {
+        (true, false) => {
+            fs::rename(source, &destination).map_err(|_| KernelError::Io)?;
+            protect_file(&destination).map_err(|_| KernelError::Io)
+        }
+        (false, true) => protect_file(&destination).map_err(|_| KernelError::Io),
+        (false, false) => Ok(()),
+        (true, true) => Err(KernelError::Inconclusive),
+    }
+}
+
+fn reset_marker_digest(marker: &ResetMarker) -> String {
+    let canonical = format!(
+        "{RESET_MARKER_PROTOCOL}\ndb_path={}\ndatabase_incarnation_id={}\nquarantine_dir={}",
+        marker.db_path.display(),
+        marker.database_incarnation_id,
+        marker.quarantine_dir.display()
+    );
+    format!("{:x}", Sha256::digest(canonical.as_bytes()))
+}
+
+fn allocate_quarantine_dir(path: &Path, lease_epoch: u64) -> Result<PathBuf, KernelError> {
+    let base = suffix_path(path, &format!("{QUARANTINE_INFIX}{lease_epoch}"));
+    for suffix in 0..10_000_u32 {
+        let candidate = if suffix == 0 {
+            base.clone()
+        } else {
+            suffix_path(&base, &format!("-{suffix}"))
+        };
+        if !entry_exists(&candidate)? {
+            return Ok(candidate);
+        }
+    }
+    Err(KernelError::Io)
+}
+
+fn valid_quarantine_path(path: &Path, quarantine: &Path) -> bool {
+    let Some(file_name) = path.file_name() else {
+        return false;
+    };
+    let mut prefix = file_name.to_os_string();
+    prefix.push(QUARANTINE_INFIX);
+    quarantine.parent() == path.parent()
+        && quarantine.file_name().is_some_and(|name| {
+            name.as_encoded_bytes()
+                .starts_with(prefix.as_encoded_bytes())
+        })
+}
+
+fn reset_marker_path(path: &Path) -> PathBuf {
+    suffix_path(path, RESET_MARKER_SUFFIX)
+}
+
+/// `Path::display` replaces non-UTF-8 bytes, which would name a different file.
+fn suffix_path(path: &Path, suffix: &str) -> PathBuf {
+    let mut name = path.as_os_str().to_os_string();
+    name.push(suffix);
+    PathBuf::from(name)
+}
+
+/// `resume_quarantine` compares the marker's `db_path` against this path by value.
+///
+/// Canonicalizing after creation gives every spelling of one directory one name.
+fn prepare_root(root: &Path) -> Result<PathBuf, KernelError> {
+    fs::create_dir_all(root).map_err(|_| KernelError::Io)?;
+    prepare_private_dir(root)?;
+    fs::canonicalize(root).map_err(|_| KernelError::Io)
+}
+
+fn prepare_private_dir(path: &Path) -> Result<(), KernelError> {
+    // The umask would leave a readable window between `create_dir` and `chmod`.
+    #[cfg(unix)]
+    let created = {
+        use std::os::unix::fs::DirBuilderExt;
+        fs::DirBuilder::new().mode(0o700).create(path)
+    };
+    #[cfg(not(unix))]
+    let created = fs::create_dir(path);
+    match created {
+        Ok(()) => {}
+        Err(error) if error.kind() == ErrorKind::AlreadyExists => {}
+        Err(_) => return Err(KernelError::Io),
+    }
+    let metadata = fs::symlink_metadata(path).map_err(|_| KernelError::Io)?;
+    if !metadata.is_dir() {
+        return Err(KernelError::Io);
+    }
+    #[cfg(unix)]
+    {
+        use std::os::unix::fs::PermissionsExt;
+        if metadata.permissions().mode() & 0o777 != 0o700 {
+            fs::set_permissions(path, fs::Permissions::from_mode(0o700))
+                .map_err(|_| KernelError::Io)?;
+        }
+    }
+    Ok(())
+}
+
+fn sync_parent(path: &Path) -> Result<(), KernelError> {
+    let parent = path.parent().ok_or(KernelError::Io)?;
+    sync_directory(parent)
+}
+
+fn sync_directory(path: &Path) -> Result<(), KernelError> {
+    File::open(path)
+        .and_then(|directory| directory.sync_all())
+        .map_err(|_| KernelError::Io)
+}
+
+fn map_lease_error(error: LeaseError) -> KernelError {
+    match error {
+        LeaseError::Held { .. } => KernelError::Held,
+        LeaseError::Io(_) => KernelError::Io,
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn owned_read_connections_are_query_only() {
+        let directory = tempfile::tempdir().unwrap();
+        let store = KernelStore::open(directory.path()).unwrap();
+        store
+            .with_reader(|connection| {
+                assert_eq!(
+                    connection.query_row("PRAGMA query_only", [], |row| row.get::<_, i64>(0))?,
+                    1
+                );
+                Ok(())
+            })
+            .unwrap();
+    }
+
+    #[test]
+    fn stale_writer_fence_blocks_the_operation() {
+        let directory = tempfile::tempdir().unwrap();
+        let store = KernelStore::open(directory.path()).unwrap();
+        store
+            .with_writer(|connection| {
+                connection.execute(
+                    "UPDATE writer_fence SET writer_epoch=writer_epoch+1 WHERE id=0",
+                    [],
+                )?;
+                Ok(())
+            })
+            .unwrap();
+        let mut called = false;
+        let error = store
+            .with_writer(|_| {
+                called = true;
+                Ok(())
+            })
+            .unwrap_err();
+        assert_eq!(error, KernelError::FenceLost);
+        assert!(!called);
+    }
+
+    #[test]
+    fn failed_writer_operation_leaves_no_partial_write() {
+        let directory = tempfile::tempdir().unwrap();
+        let store = KernelStore::open(directory.path()).unwrap();
+        let error = store
+            .with_writer(|tx| -> rusqlite::Result<()> {
+                tx.execute(
+                    "INSERT INTO commit_log(transaction_id,writer_epoch,recorded_at,actor,cause)
+                     VALUES('t1',1,1,'actor','cause')",
+                    [],
+                )?;
+                Err(rusqlite::Error::InvalidQuery)
+            })
+            .unwrap_err();
+        assert_eq!(error, KernelError::Io);
+        store
+            .with_reader(|tx| {
+                assert_eq!(
+                    tx.query_row("SELECT COUNT(*) FROM commit_log", [], |row| row
+                        .get::<_, i64>(0))?,
+                    0
+                );
+                Ok(())
+            })
+            .unwrap();
+    }
+
+    #[test]
+    fn store_connections_run_delete_triggers_for_replace() {
+        // The schema-level guard test drives a connection built by
+        // `apply_kernel_connection_profile`, so it cannot observe the pragmas
+        // `KernelStore` sets on its own writer.
+        let directory = tempfile::tempdir().unwrap();
+        let store = KernelStore::open(directory.path()).unwrap();
+        store
+            .with_writer(|tx| {
+                assert_eq!(
+                    tx.query_row("PRAGMA recursive_triggers", [], |row| row.get::<_, i64>(0))?,
+                    1
+                );
+                tx.execute(
+                    "INSERT INTO commit_log(transaction_id,writer_epoch,recorded_at,actor,cause)
+                     VALUES('t1',1,1,'actor','cause')",
+                    [],
+                )?;
+                Ok(())
+            })
+            .unwrap();
+
+        let commit_seq = store
+            .with_reader(|tx| {
+                tx.query_row("SELECT commit_seq FROM commit_log", [], |row| {
+                    row.get::<_, i64>(0)
+                })
+            })
+            .unwrap();
+
+        for verb in ["INSERT OR REPLACE", "REPLACE"] {
+            let error = store
+                .with_writer(|tx| {
+                    tx.execute(
+                        &format!(
+                            "{verb} INTO commit_log(
+                                 commit_seq,transaction_id,writer_epoch,recorded_at,actor,cause
+                             ) VALUES(?1,'hijack',1,1,'attacker','rewrite')"
+                        ),
+                        [commit_seq],
+                    )?;
+                    Ok(())
+                })
+                .unwrap_err();
+            assert_eq!(error, KernelError::Io, "{verb} must be refused");
+        }
+
+        store
+            .with_reader(|tx| {
+                assert_eq!(
+                    tx.query_row("SELECT actor FROM commit_log", [], |row| row
+                        .get::<_, String>(0))?,
+                    "actor"
+                );
+                Ok(())
+            })
+            .unwrap();
+    }
+}
diff --git a/crates/mc-store/src/kernel/schema.rs b/crates/mc-store/src/kernel/schema.rs
index 1b0556fd3..31c3dad9f 100644
--- a/crates/mc-store/src/kernel/schema.rs
+++ b/crates/mc-store/src/kernel/schema.rs
@@ -1,9 +1,12 @@
-use crate::sqlite_runtime::{DIRECT_FORMAT_EPOCH, MC_APPLICATION_ID};
+use crate::sqlite_runtime::{
+    compute_marker_digest_for_application_id, DIRECT_FORMAT_EPOCH, MC_APPLICATION_ID,
+};
 use rusqlite::{params, Connection, TransactionBehavior};
 use sha2::{Digest, Sha256};
 use std::collections::BTreeSet;
 
 pub const KERNEL_APPLICATION_ID: u32 = MC_APPLICATION_ID;
+pub const KERNEL_FORMAT_EPOCH: i64 = DIRECT_FORMAT_EPOCH;
 pub const KERNEL_SCHEMA_COMPONENT_NAMES: &[&str] = &[
     "commit_log",
     "change_event",
@@ -161,7 +164,7 @@ const COMPONENTS: &[(&str, &str)] = &[
     ),
     (
         "mc_kernel_format_marker",
-        r#"CREATE TABLE mc_kernel_format_marker(singleton INTEGER PRIMARY KEY CHECK(singleton=1),format_epoch INTEGER NOT NULL,database_incarnation_id TEXT NOT NULL CHECK(length(database_incarnation_id)=32),schema_digest TEXT NOT NULL CHECK(length(schema_digest)=64),created_at INTEGER NOT NULL) STRICT; CREATE TRIGGER mc_kernel_format_marker_no_update BEFORE UPDATE ON mc_kernel_format_marker BEGIN SELECT RAISE(ABORT,'mc_kernel_format_marker is immutable'); END; CREATE TRIGGER mc_kernel_format_marker_no_delete BEFORE DELETE ON mc_kernel_format_marker BEGIN SELECT RAISE(ABORT,'mc_kernel_format_marker is immutable'); END;"#,
+        r#"CREATE TABLE mc_kernel_format_marker(singleton INTEGER PRIMARY KEY CHECK(singleton=1),format_epoch INTEGER NOT NULL,database_incarnation_id TEXT NOT NULL CHECK(length(database_incarnation_id)=32),schema_digest TEXT NOT NULL CHECK(length(schema_digest)=64),created_at INTEGER NOT NULL,marker_digest TEXT NOT NULL CHECK(length(marker_digest)=64)) STRICT; CREATE TRIGGER mc_kernel_format_marker_no_update BEFORE UPDATE ON mc_kernel_format_marker BEGIN SELECT RAISE(ABORT, 'mc_kernel_format_marker is immutable'); END; CREATE TRIGGER mc_kernel_format_marker_no_delete BEFORE DELETE ON mc_kernel_format_marker BEGIN SELECT RAISE(ABORT, 'mc_kernel_format_marker is immutable'); END; CREATE TRIGGER mc_kernel_format_marker_no_replace BEFORE INSERT ON mc_kernel_format_marker WHEN EXISTS(SELECT 1 FROM mc_kernel_format_marker) BEGIN SELECT RAISE(ABORT, 'mc_kernel_format_marker is immutable'); END;"#,
     ),
 ];
 
@@ -291,14 +294,21 @@ fn apply_schema<F: FnOnce() -> rusqlite::Result<()>>(
         return Err(rusqlite::Error::InvalidQuery);
     }
     tx.pragma_update(None, "application_id", KERNEL_APPLICATION_ID)?;
-    tx.pragma_update(None, "user_version", DIRECT_FORMAT_EPOCH)?;
+    tx.pragma_update(None, "user_version", KERNEL_FORMAT_EPOCH)?;
     for (_, sql) in COMPONENTS {
         tx.execute_batch(sql)?;
     }
     tx.execute("INSERT INTO writer_fence(id) VALUES(0)", [])?;
     hook()?;
     let digest = kernel_schema_digest(&tx)?;
-    tx.execute("INSERT INTO mc_kernel_format_marker(singleton,format_epoch,database_incarnation_id,schema_digest,created_at) VALUES(1,1,?1,?2,?3)", params![incarnation,digest,created_at])?;
+    let marker_digest = compute_marker_digest_for_application_id(
+        KERNEL_APPLICATION_ID,
+        KERNEL_FORMAT_EPOCH,
+        incarnation,
+        &digest,
+        created_at,
+    );
+    tx.execute("INSERT INTO mc_kernel_format_marker(singleton,format_epoch,database_incarnation_id,schema_digest,created_at,marker_digest) VALUES(1,?1,?2,?3,?4,?5)", params![KERNEL_FORMAT_EPOCH,incarnation,digest,created_at,marker_digest])?;
     tx.commit()
 }
 
@@ -330,6 +340,20 @@ pub fn kernel_schema_inventory(conn: &Connection) -> rusqlite::Result<Vec<String
     Ok(result)
 }
 
+pub fn kernel_schema_object_inventory(
+    conn: &Connection,
+) -> rusqlite::Result<Vec<(String, String)>> {
+    let mut stmt = conn.prepare(
+        "SELECT type,name FROM sqlite_schema
+         WHERE name NOT LIKE 'sqlite\\_%' ESCAPE '\\'
+         ORDER BY type,name",
+    )?;
+    let inventory = stmt
+        .query_map([], |row| Ok((row.get(0)?, row.get(1)?)))?
+        .collect();
+    inventory
+}
+
 pub fn kernel_schema_digest(conn: &Connection) -> rusqlite::Result<String> {
     let mut stmt = conn.prepare("SELECT type,name,tbl_name,sql FROM sqlite_schema WHERE name NOT LIKE 'sqlite\\_%' ESCAPE '\\' AND sql IS NOT NULL ORDER BY type,name")?;
     let mut rows = stmt.query([])?;
diff --git a/crates/mc-store/src/lib.rs b/crates/mc-store/src/lib.rs
index fd668d073..c412b532f 100644
--- a/crates/mc-store/src/lib.rs
+++ b/crates/mc-store/src/lib.rs
@@ -423,7 +423,7 @@ pub const PASS_SCHEDULER_TELEMETRY_MAX_BYTES: usize = 1
     + PASS_SCHEDULER_INTERESTING_HISTORY_CAP
         * (MAX_INTERESTING_PASS_SCHEDULER_OBSERVATION_JSON_BYTES + 1);
 
-fn current_time_ms() -> i64 {
+pub(crate) fn current_time_ms() -> i64 {
     std::time::SystemTime::now()
         .duration_since(std::time::UNIX_EPOCH)
         .map(|duration| i64::try_from(duration.as_millis()).unwrap_or(i64::MAX))
diff --git a/crates/mc-store/src/sqlite_runtime.rs b/crates/mc-store/src/sqlite_runtime.rs
index dacdf3f34..b853f41ff 100644
--- a/crates/mc-store/src/sqlite_runtime.rs
+++ b/crates/mc-store/src/sqlite_runtime.rs
@@ -177,10 +177,26 @@ pub fn compute_marker_digest(
     database_incarnation_id: &str,
     component_manifest_digest: &str,
     created_at_ms: i64,
+) -> String {
+    compute_marker_digest_for_application_id(
+        MC_APPLICATION_ID,
+        format_epoch,
+        database_incarnation_id,
+        component_manifest_digest,
+        created_at_ms,
+    )
+}
+
+pub fn compute_marker_digest_for_application_id(
+    application_id: u32,
+    format_epoch: i64,
+    database_incarnation_id: &str,
+    component_manifest_digest: &str,
+    created_at_ms: i64,
 ) -> String {
     let lines = [
         FORMAT_MARKER_DIGEST_PROTOCOL.to_string(),
-        format!("application_id={MC_APPLICATION_ID}"),
+        format!("application_id={application_id}"),
         format!("format_epoch={format_epoch}"),
         format!("database_incarnation_id={database_incarnation_id}"),
         format!("component_manifest_digest={component_manifest_digest}"),
diff --git a/crates/mc-store/tests/kernel_open.rs b/crates/mc-store/tests/kernel_open.rs
new file mode 100644
index 000000000..782c4540b
--- /dev/null
+++ b/crates/mc-store/tests/kernel_open.rs
@@ -0,0 +1,382 @@
+use mc_store::kernel::schema::{
+    apply_kernel_connection_profile, apply_kernel_schema, kernel_schema_digest,
+    KERNEL_APPLICATION_ID,
+};
+use mc_store::kernel::{KernelError, KernelStore};
+use mc_store::sqlite_runtime::compute_marker_digest_for_application_id;
+#[cfg(feature = "test-support")]
+use mc_store::sqlite_runtime::SqliteEngineIdentity;
+use rusqlite::{Connection, OpenFlags};
+use sha2::{Digest, Sha256};
+use std::fs;
+use std::path::{Path, PathBuf};
+
+const INCARNATION: &str = "0123456789abcdef0123456789abcdef";
+
+fn core_path(root: &Path) -> PathBuf {
+    root.join("core.sqlite")
+}
+
+fn inspect<T>(root: &Path, query: impl FnOnce(&Connection) -> T) -> T {
+    let conn = Connection::open_with_flags(core_path(root), OpenFlags::SQLITE_OPEN_READ_ONLY)
+        .expect("open read-only inspection connection");
+    query(&conn)
+}
+
+fn seed_kernel(root: &Path) -> Connection {
+    fs::create_dir_all(root).unwrap();
+    let mut conn = Connection::open(core_path(root)).unwrap();
+    apply_kernel_connection_profile(&mut conn, 5_000).unwrap();
+    apply_kernel_schema(&mut conn, INCARNATION, 1_000).unwrap();
+    conn
+}
+
+fn marker_digest(epoch: i64, schema_digest: &str) -> String {
+    compute_marker_digest_for_application_id(
+        KERNEL_APPLICATION_ID,
+        epoch,
+        INCARNATION,
+        schema_digest,
+        1_000,
+    )
+}
+
+fn replace_marker(conn: &Connection, epoch: i64, schema_digest: &str) {
+    conn.execute_batch(
+        "DROP TRIGGER mc_kernel_format_marker_no_update;
+         DROP TRIGGER mc_kernel_format_marker_no_delete;",
+    )
+    .unwrap();
+    conn.execute(
+        "UPDATE mc_kernel_format_marker
+         SET format_epoch=?1, schema_digest=?2, marker_digest=?3",
+        (epoch, schema_digest, marker_digest(epoch, schema_digest)),
+    )
+    .unwrap();
+    conn.execute_batch(
+        "CREATE TRIGGER mc_kernel_format_marker_no_update
+         BEFORE UPDATE ON mc_kernel_format_marker BEGIN
+           SELECT RAISE(ABORT, 'mc_kernel_format_marker is immutable');
+         END;
+         CREATE TRIGGER mc_kernel_format_marker_no_delete
+         BEFORE DELETE ON mc_kernel_format_marker BEGIN
+           SELECT RAISE(ABORT, 'mc_kernel_format_marker is immutable');
+         END;",
+    )
+    .unwrap();
+}
+
+fn quarantine_dirs(root: &Path) -> Vec<PathBuf> {
+    let mut paths = fs::read_dir(root)
+        .unwrap()
+        .map(|entry| entry.unwrap().path())
+        .filter(|path| {
+            path.file_name()
+                .unwrap()
+                .to_string_lossy()
+                .starts_with("core.sqlite.mc-quarantine-")
+        })
+        .collect::<Vec<_>>();
+    paths.sort();
+    paths
+}
+
+#[test]
+fn fresh_open_and_exact_reopen_preserve_identity_and_advance_fence() {
+    let dir = tempfile::tempdir().unwrap();
+    let first = KernelStore::open(dir.path()).unwrap();
+    let first_epoch = first.lease_epoch();
+    let incarnation = inspect(dir.path(), |conn| {
+        conn.query_row(
+            "SELECT database_incarnation_id FROM mc_kernel_format_marker",
+            [],
+            |row| row.get::<_, String>(0),
+        )
+    })
+    .unwrap();
+    assert_eq!(
+        inspect(dir.path(), |conn| conn.query_row(
+            "SELECT writer_epoch FROM writer_fence WHERE id=0",
+            [],
+            |row| row.get::<_, i64>(0)
+        ))
+        .unwrap(),
+        i64::try_from(first_epoch).unwrap()
+    );
+    drop(first);
+
+    let second = KernelStore::open(dir.path()).unwrap();
+    assert!(second.lease_epoch() > first_epoch);
+    assert_eq!(
+        inspect(dir.path(), |conn| {
+            conn.query_row(
+                "SELECT database_incarnation_id FROM mc_kernel_format_marker",
+                [],
+                |row| row.get::<_, String>(0),
+            )
+        })
+        .unwrap(),
+        incarnation
+    );
+}
+
+#[test]
+fn second_opener_is_held_without_touching_database_family() {
+    let dir = tempfile::tempdir().unwrap();
+    let first = KernelStore::open(dir.path()).unwrap();
+    let before = fs::read(core_path(dir.path())).unwrap();
+    assert_eq!(
+        KernelStore::open(dir.path()).unwrap_err(),
+        KernelError::Held
+    );
+    assert_eq!(fs::read(core_path(dir.path())).unwrap(), before);
+    drop(first);
+}
+
+#[test]
+fn every_conclusive_kernel_mismatch_is_quarantined_and_rebuilt() {
+    for mismatch in ["epoch", "digest", "inventory"] {
+        let dir = tempfile::tempdir().unwrap();
+        let conn = seed_kernel(dir.path());
+        match mismatch {
+            "epoch" => {
+                let digest = kernel_schema_digest(&conn).unwrap();
+                replace_marker(&conn, 2, &digest);
+            }
+            "digest" => replace_marker(&conn, 1, &"a".repeat(64)),
+            "inventory" => {
+                conn.execute_batch("CREATE TABLE unexpected(value INTEGER) STRICT;")
+                    .unwrap();
+            }
+            _ => unreachable!(),
+        }
+        drop(conn);
+
+        let _store = KernelStore::open(dir.path()).unwrap();
+        assert_eq!(quarantine_dirs(dir.path()).len(), 1, "{mismatch}");
+        assert_eq!(
+            inspect(dir.path(), |conn| conn.query_row(
+                "SELECT COUNT(*) FROM mc_kernel_format_marker",
+                [],
+                |row| row.get::<_, i64>(0)
+            ))
+            .unwrap(),
+            1
+        );
+    }
+}
+
+#[test]
+fn foreign_family_is_refused_before_sqlite_can_touch_it() {
+    // `FOREIGN_APPLICATION_ID` must differ from `KERNEL_APPLICATION_ID` so the
+    // fixture exercises foreign-header classification.
+    const FOREIGN_APPLICATION_ID: u32 = 0x5A5A_5A5A;
+    assert_ne!(FOREIGN_APPLICATION_ID, KERNEL_APPLICATION_ID);
+
+    let dir = tempfile::tempdir().unwrap();
+    let path = core_path(dir.path());
+    let conn = Connection::open(&path).unwrap();
+    conn.pragma_update(None, "application_id", FOREIGN_APPLICATION_ID)
+        .unwrap();
+    conn.execute_batch("CREATE TABLE legacy(value TEXT);")
+        .unwrap();
+    drop(conn);
+    fs::write(format!("{}-wal", path.display()), b"foreign wal").unwrap();
+    let before_main = fs::read(&path).unwrap();
+    let before_wal = fs::read(format!("{}-wal", path.display())).unwrap();
+
+    assert_eq!(
+        KernelStore::open(dir.path()).unwrap_err(),
+        KernelError::Foreign
+    );
+    assert_eq!(fs::read(&path).unwrap(), before_main);
+    assert_eq!(
+        fs::read(format!("{}-wal", path.display())).unwrap(),
+        before_wal
+    );
+    assert!(quarantine_dirs(dir.path()).is_empty());
+}
+
+#[test]
+fn a_sibling_mc_family_is_refused_and_left_untouched() {
+    // `KERNEL_APPLICATION_ID` is shared across mc families; schema inspection
+    // distinguishes them.
+    let dir = tempfile::tempdir().unwrap();
+    let path = core_path(dir.path());
+    let conn = Connection::open(&path).unwrap();
+    conn.pragma_update(None, "application_id", KERNEL_APPLICATION_ID)
+        .unwrap();
+    conn.execute_batch("CREATE TABLE legacy(value TEXT);")
+        .unwrap();
+    drop(conn);
+    let before_main = fs::read(&path).unwrap();
+
+    assert_eq!(
+        KernelStore::open(dir.path()).unwrap_err(),
+        KernelError::Inconclusive
+    );
+    assert_eq!(fs::read(&path).unwrap(), before_main);
+    assert!(quarantine_dirs(dir.path()).is_empty());
+}
+
+#[test]
+fn malformed_marker_is_inconclusive_and_untouched() {
+    // "g" fails the lowercase-hex check; the all-zero digest fails digest comparison.
+    for digest in ["g".repeat(64), "0".repeat(64)] {
+        let dir = tempfile::tempdir().unwrap();
+        let conn = seed_kernel(dir.path());
+        conn.execute_batch("DROP TRIGGER mc_kernel_format_marker_no_update;")
+            .unwrap();
+        conn.execute(
+            "UPDATE mc_kernel_format_marker SET marker_digest=?1",
+            [&digest],
+        )
+        .unwrap();
+        drop(conn);
+        let path = core_path(dir.path());
+        let before = fs::read(&path).unwrap();
+
+        assert_eq!(
+            KernelStore::open(dir.path()).unwrap_err(),
+            KernelError::Inconclusive,
+            "{digest}"
+        );
+        assert_eq!(fs::read(&path).unwrap(), before, "{digest}");
+        assert!(quarantine_dirs(dir.path()).is_empty(), "{digest}");
+    }
+}
+
+#[test]
+fn valid_interrupted_reset_marker_resumes_without_opening_old_family() {
+    let dir = tempfile::tempdir().unwrap();
+    // Quarantine resume compares paths lexically. A symlinked root
+    // (`/var` -> `/private/var`) fails that comparison when spellings are mixed.
+    // commentlint: allow(JUDGE)
+    let root = dir.path().canonicalize().unwrap();
+    let conn = seed_kernel(&root);
+    conn.execute_batch("CREATE TABLE unexpected(value INTEGER) STRICT;")
+        .unwrap();
+    drop(conn);
+    let db_path = core_path(&root);
+    let quarantine = root.join("core.sqlite.mc-quarantine-resume");
+    fs::create_dir(&quarantine).unwrap();
+    fs::rename(&db_path, quarantine.join("core.sqlite")).unwrap();
+    for suffix in ["-journal", "-wal", "-shm"] {
+        fs::write(format!("{}{suffix}", db_path.display()), suffix.as_bytes()).unwrap();
+    }
+    let marker_without_digest = serde_json::json!({
+        "protocol": "mc-kernel-reset-marker-v1",
+        "db_path": db_path,
+        "database_incarnation_id": INCARNATION,
+        "quarantine_dir": quarantine,
+    });
+    let canonical = format!(
+        "mc-kernel-reset-marker-v1\ndb_path={}\ndatabase_incarnation_id={}\nquarantine_dir={}",
+        marker_without_digest["db_path"].as_str().unwrap(),
+        INCARNATION,
+        marker_without_digest["quarantine_dir"].as_str().unwrap(),
+    );
+    let digest = format!("{:x}", Sha256::digest(canonical.as_bytes()));
+    let mut marker = marker_without_digest;
+    marker["marker_digest"] = digest.into();
+    fs::write(
+        root.join("core.sqlite.mc-reset"),
+        serde_json::to_vec(&marker).unwrap(),
+    )
+    .unwrap();
+
+    let _store = KernelStore::open(&root).unwrap();
+    assert!(core_path(&root).is_file());
+    assert!(quarantine.join("core.sqlite.mc-reset").is_file());
+    assert_owner_only(&quarantine, 0o700);
+    for name in [
+        "core.sqlite",
+        "core.sqlite-journal",
+        "core.sqlite-wal",
+        "core.sqlite-shm",
+        "core.sqlite.mc-reset",
+    ] {
+        assert_owner_only(&quarantine.join(name), 0o600);
+    }
+}
+
+#[test]
+fn writer_and_sidecars_are_owner_only() {
+    let dir = tempfile::tempdir().unwrap();
+    let store = KernelStore::open(dir.path()).unwrap();
+    let path = core_path(dir.path());
+    assert_owner_only(&path, 0o600);
+    assert_owner_only(&PathBuf::from(format!("{}-wal", path.display())), 0o600);
+    assert_owner_only(&PathBuf::from(format!("{}-shm", path.display())), 0o600);
+    assert_eq!(
+        inspect(dir.path(), |conn| conn.query_row(
+            "PRAGMA journal_mode",
+            [],
+            |row| row.get::<_, String>(0)
+        ))
+        .unwrap(),
+        "wal"
+    );
+    drop(store);
+}
+
+#[cfg(feature = "test-support")]
+#[test]
+fn unsupported_engine_is_rejected_before_creating_lease_or_database_files() {
+    let dir = tempfile::tempdir().unwrap();
+    let root = dir.path().join("kernel");
+    let identity = SqliteEngineIdentity {
+        sqlite_version: "3.51.2".to_string(),
+        sqlite_source_id: "2026-01-01 00:00:00 0123456789abcdef0123456789abcdef01234567"
+            .to_string(),
+    };
+    assert_eq!(
+        KernelStore::open_with_engine_identity_for_test(&root, &identity).unwrap_err(),
+        KernelError::EngineUnsupported
+    );
+    assert!(!root.exists());
+}
+
+#[test]
+fn kernel_with_uncheckpointed_wal_opens_and_preserves_rows() {
+    let dir = tempfile::tempdir().unwrap();
+    drop(KernelStore::open(dir.path()).unwrap());
+
+    let conn = Connection::open(core_path(dir.path())).unwrap();
+    conn.pragma_update(None, "journal_mode", "WAL").unwrap();
+    conn.execute(
+        "INSERT INTO commit_log(transaction_id,writer_epoch,recorded_at,actor,cause)
+         VALUES('t1',1,1,'actor','cause')",
+        [],
+    )
+    .unwrap();
+    let wal = PathBuf::from(format!("{}-wal", core_path(dir.path()).display()));
+    assert!(wal.is_file(), "the row should still be in the WAL");
+    // Leaking skips the clean close that would checkpoint and remove the WAL.
+    // A crashed writer leaves the uncheckpointed WAL on disk.
+    std::mem::forget(conn);
+
+    let _store = KernelStore::open(dir.path()).unwrap();
+    assert_eq!(
+        inspect(dir.path(), |conn| conn.query_row(
+            "SELECT COUNT(*) FROM commit_log",
+            [],
+            |row| row.get::<_, i64>(0)
+        ))
+        .unwrap(),
+        1
+    );
+    assert!(quarantine_dirs(dir.path()).is_empty());
+}
+
+#[cfg(unix)]
+fn assert_owner_only(path: &Path, expected: u32) {
+    use std::os::unix::fs::PermissionsExt;
+    assert_eq!(
+        fs::metadata(path).unwrap().permissions().mode() & 0o777,
+        expected
+    );
+}
+
+#[cfg(not(unix))]
+fn assert_owner_only(_path: &Path, _expected: u32) {}
diff --git a/crates/mc-store/tests/kernel_schema.rs b/crates/mc-store/tests/kernel_schema.rs
index fc515e3b0..8a4985fda 100644
--- a/crates/mc-store/tests/kernel_schema.rs
+++ b/crates/mc-store/tests/kernel_schema.rs
@@ -132,7 +132,7 @@ fn kernel_schema_has_one_ordered_full_shape() {
 const INCARNATION: &str = "0123456789abcdef0123456789abcdef";
 
 const PINNED_SCHEMA_DIGEST: &str =
-    "edd21757592088f528212faa058a1c3dd0feb122dc9fa539ecbfa17b67ab6b01";
+    "92e7c76e51e721720c2123e0bba45ea41995545b308f9767c64c7aef0fe7a9e6";
 
 #[test]
 fn kernel_schema_digest_is_pinned_to_the_frozen_v1_shape() {
@@ -1223,15 +1223,21 @@ fn replace_cannot_bypass_the_append_only_guards() {
     let commit_seq = next_commit(&conn, "tx-replace");
 
     // REPLACE resolves a conflict by deleting the existing row; the delete
-    // trigger only runs when recursive_triggers is on.
-    assert!(conn
-        .execute(
-            "INSERT OR REPLACE INTO mc_kernel_format_marker(
-                 singleton, format_epoch, database_incarnation_id, schema_digest, created_at
-             ) VALUES (1, 99, '99999999999999999999999999999999', ?1, 2)",
-            [PINNED_SCHEMA_DIGEST],
-        )
-        .is_err());
+    // trigger only runs when recursive_triggers is on. Every value below
+    // satisfies the STRICT column types and the length CHECKs, so the BEFORE
+    // INSERT guard is the only thing left that can reject these statements.
+    let columns = "singleton, format_epoch, database_incarnation_id, schema_digest, created_at,
+                   marker_digest";
+    let values = format!(
+        "1, 99, '{}', '{}', 2, '{}'",
+        "9".repeat(32),
+        PINNED_SCHEMA_DIGEST,
+        "d".repeat(64)
+    );
+    for verb in ["INSERT OR REPLACE", "REPLACE"] {
+        let statement = format!("{verb} INTO mc_kernel_format_marker({columns}) VALUES({values})");
+        assert!(conn.execute(&statement, []).is_err(), "{statement}");
+    }
     assert_eq!(
         conn.query_row(
             "SELECT database_incarnation_id FROM mc_kernel_format_marker",