alkgit-substrate 0.1.0

//! The sqlite registry twin — the family adapter pattern's default
//! engine assembly (ADR-019 §3, substrate.md §registry: "one trait
//! pair, two implements — the sqlite adapter (`sqlite` feature,
//! `alkstore-sqlite` + the co-tenant rusqlite connection)").
//!
//! - [`SqliteRegistry`] — the durable write side ([`RegistryWrite`])
//!   over the co-tenant `git_repos` rows. Every mutation bumps the
//!   durable `registry_seq` singleton **inside the row-write
//!   transaction** (transactionally exact — the watermark's anchor,
//!   ADR-019 §5; failed mutations consume none, the tx rolls back
//!   whole), and the twins' row commits also carry the refs-table legs
//!   the composition owes: **put_repo seeds `HEAD → refs/heads/main`**
//!   (substrate.md §refs — git parity: `git init --bare` creates the
//!   dangling HEAD at init, not at first push; git's push grammar
//!   carries no HEAD. The dangling symref is exactly the unborn-HEAD
//!   shape `ls-refs=unborn` serves — the OQ-19 rider (a) fixture — and
//!   is CAS-unreachable from a push per git parity; the row is
//!   CAS-rewriteable, so the v1-pinned target is a two-way door, the
//!   constructor knob additive later) and **remove_repo removes the
//!   registry row, the repo's ref rows, and the repo's mutation-marker
//!   rows together** (backend.md — deletion is a GC consequence; objects
//!   stay for the sweep, no synchronous object deletion exists — the
//!   removal only touches co-tenant rows). Plus
//!   `SqliteRegistry::record_mutation` — the per-push mutation-marker
//!   write (ADR-019 §7, composition-internal: bumps the
//!   seq/watermark legs without touching the record; the receive-pack
//!   composition is the consumer).
//! - [`SqliteRegistryAssembly`] — the adapter constructor: one
//!   fallible assembly call (loud failure — ADR-024 §1) that opens the
//!   co-tenant connection ([`CoTenantSqlite::open`] — WAL, the pinned
//!   `busy_timeout` 5000 ms, the migrations run at construction) beside
//!   the engine on one file, opens the `alkstore-sqlite` engine, loads
//!   the cached read provider's initial snapshot, wires the reactive
//!   halves ([`Reactor`]) plus the announced write side
//!   ([`AnnouncingRegistry`]) and the refs twin, and hands back the
//!   pieces. The wave-8 single composition entry unifies construction;
//!   this is the per-piece constructor (the recorded interpretation).
//!   Post-commit announce (publish + notify + enqueue — each
//!   independent, fail-soft) is wired by default: mutations ride
//!   [`AnnouncingRegistry`], never a non-reactive shipped path
//!   (ADR-019 §5).
//!
//! # The sqlite asymmetry carried (substrate.md §the reactive halves)
//!
//! The co-tenant file is itself the invalidation bus: the watcher's
//! file-wide `data_version` bump fans a wake to listeners on *any*
//! commit to the file — including alkgit's own registry writes on
//! their own connection, including cross-process (~3.2 ms measured,
//! POC-R; the probe is test-pinned here). A lost hint cannot exist
//! while any write lands; the reconcile sweep is the correctness
//! backstop, not a latency bound. The post-commit notify leg stays
//! redundant but kept — for semantics and for postgres.
//!
//! Concurrency model: rusqlite is blocking — the trait methods bridge
//! their SQL off the async executor (`spawn_blocking`) under the
//! co-tenant handle's short-held lock, exactly the refs twin's shape;
//! the pinned busy timeout absorbs the cross-process write-lock waits.

use std::path::Path;
use std::sync::atomic::AtomicU64;
use std::sync::Arc;

use alkgit::{RegistryError, RepoRecord};
use async_trait::async_trait;

use crate::error::OpenError;
use crate::migrations::CoTenantSqlite;
use crate::reactor::{AnnouncingRegistry, Reactor, RegistryEvent};
use crate::refs::{SqliteRefTable, VersionedRefTable};
use crate::registry::{
    row_to_record, validate_record, visibility_str, CachedRegistryProvider, RegistryBackend,
    RegistryRow, RegistryWrite,
};

/// The HEAD seed target moved engine-neutral ([`crate::registry`]) —
/// the sqlite twin re-exports it under its own name for the twin's
/// consumers.
pub use crate::registry::HEAD_SEED_TARGET;

/// The durable write side of the family pair over the co-tenant
/// `git_repos` rows (the pair's sqlite implement). Reads ride the
/// cached provider in the composed assemblies — the hot path never
/// touches this side except the refresh reads.
///
/// Mount over an already-migrated co-tenant handle (the refs twin's
/// shape — the same handle the composition opens once for both twins;
/// the migrations run at the handle's construction, loud, ADR-024 §1).
pub struct SqliteRegistry {
    handle: CoTenantSqlite,
}

impl SqliteRegistry {
    /// Mount the twin over a co-tenant handle.
    pub fn new(handle: CoTenantSqlite) -> Self {
        SqliteRegistry { handle }
    }

    /// The marker write's transaction rides its own `BEGIN IMMEDIATE`
    /// (`durable_record_mutation`): the marker row upserted with a
    /// `registry_seq` bump inside one transaction — the watermark's
    /// anchor moves on every push while the record row and its schema
    /// stay untouched (`rows_since` never carries markers; the seq is
    /// the reconcile watermark's currency). The record must exist (the
    /// receive flow resolved it) — a removed repo's marker is refused
    /// ([`RegistryError::NotFound`]). Consuming composition: the
    /// assembly's fused mutation legs ([`crate::pins::RegistryMutations`]
    /// — the receive flow's post-CAS legs), which announce the write
    /// post-commit.
    pub(crate) async fn record_mutation(&self, repo_id: &str) -> Result<i64, RegistryError> {
        let handle = self.handle.clone();
        let repo_id = repo_id.to_owned();
        blocking(move || durable_record_mutation(&handle, &repo_id)).await
    }
}

#[async_trait]
impl RegistryWrite for SqliteRegistry {
    async fn put(&self, record: &RepoRecord) -> Result<i64, RegistryError> {
        validate_record(record)?;
        let handle = self.handle.clone();
        let record = record.clone();
        blocking(move || durable_put(&handle, &record)).await
    }

    async fn update(&self, record: &RepoRecord) -> Result<i64, RegistryError> {
        validate_record(record)?;
        let handle = self.handle.clone();
        let record = record.clone();
        blocking(move || durable_update(&handle, &record)).await
    }

    async fn remove(&self, repo_id: &str) -> Result<i64, RegistryError> {
        let handle = self.handle.clone();
        let repo_id = repo_id.to_owned();
        blocking(move || durable_remove(&handle, &repo_id)).await
    }

    async fn watermark(&self) -> Result<i64, RegistryError> {
        let handle = self.handle.clone();
        blocking(move || durable_watermark(&handle)).await
    }

    async fn rows_since(&self, seq: i64) -> Result<Vec<RegistryRow>, RegistryError> {
        let handle = self.handle.clone();
        blocking(move || durable_rows_since(&handle, seq)).await
    }
}

/// The per-mutation blocking-task bridge — the trait methods move their
/// owned inputs into the spawn_blocking task; join failure is a
/// substrate failure (message-only, the family posture).
async fn blocking<R, T>(work: T) -> Result<R, RegistryError>
where
    T: FnOnce() -> Result<R, RegistryError> + Send + 'static,
    R: Send + 'static,
{
    tokio::task::spawn_blocking(work)
        .await
        .map_err(|e| RegistryError::Io(format!("registry blocking task: {e}")))?
}

fn io_err(context: &str, error: rusqlite::Error) -> RegistryError {
    RegistryError::Io(format!("{context}: {error}"))
}

/// The `put`'s transaction: the existence check first (a taken id is
/// [`RegistryError::AlreadyExists`], no seq consumed), then the seq
/// bump, the record row, and the HEAD seed row — commit or roll back
/// whole.
fn durable_put(handle: &CoTenantSqlite, record: &RepoRecord) -> Result<i64, RegistryError> {
    let mut conn = handle.lock();
    let tx = conn
        .transaction_with_behavior(rusqlite::TransactionBehavior::Immediate)
        .map_err(|e| io_err("registry put BEGIN IMMEDIATE", e))?;
    let taken: i64 = tx
        .query_row(
            "SELECT count(*) FROM git_repos WHERE repo_id = ?1",
            rusqlite::params![record.repo_id],
            |r| r.get(0),
        )
        .map_err(|e| io_err("registry put existence read", e))?;
    if taken > 0 {
        return Err(RegistryError::AlreadyExists);
    }
    let grants = serde_json::to_string(&record.grants).map_err(|_| RegistryError::Invalid)?;
    let seq: i64 = tx
        .query_row(
            "UPDATE registry_seq SET value = value + 1 WHERE id = 1 RETURNING value",
            [],
            |r| r.get(0),
        )
        .map_err(|e| io_err("registry put seq bump", e))?;
    tx.execute(
        "INSERT INTO git_repos (repo_id, visibility, grants, seq) VALUES (?1, ?2, ?3, ?4)",
        rusqlite::params![
            record.repo_id,
            visibility_str(record.visibility),
            grants,
            seq
        ],
    )
    .map_err(|e| io_err("registry put record row", e))?;
    tx.execute(
        "INSERT INTO git_refs (repo_id, name, oid_hex, symref_target) \
         VALUES (?1, 'HEAD', NULL, ?2)",
        rusqlite::params![record.repo_id, HEAD_SEED_TARGET],
    )
    .map_err(|e| io_err("registry put HEAD seed", e))?;
    tx.commit().map_err(|e| io_err("registry put commit", e))?;
    Ok(seq)
}

/// The `update`'s transaction: existence first
/// ([`RegistryError::NotFound`]), the seq bump, the record row's
/// replace. The refs-table rows are untouched (HEAD belongs to the
/// record's creation).
fn durable_update(handle: &CoTenantSqlite, record: &RepoRecord) -> Result<i64, RegistryError> {
    let mut conn = handle.lock();
    let tx = conn
        .transaction_with_behavior(rusqlite::TransactionBehavior::Immediate)
        .map_err(|e| io_err("registry update BEGIN IMMEDIATE", e))?;
    let exists: i64 = tx
        .query_row(
            "SELECT count(*) FROM git_repos WHERE repo_id = ?1",
            rusqlite::params![record.repo_id],
            |r| r.get(0),
        )
        .map_err(|e| io_err("registry update existence read", e))?;
    if exists == 0 {
        return Err(RegistryError::NotFound);
    }
    let grants = serde_json::to_string(&record.grants).map_err(|_| RegistryError::Invalid)?;
    let seq: i64 = tx
        .query_row(
            "UPDATE registry_seq SET value = value + 1 WHERE id = 1 RETURNING value",
            [],
            |r| r.get(0),
        )
        .map_err(|e| io_err("registry update seq bump", e))?;
    tx.execute(
        "UPDATE git_repos SET visibility = ?2, grants = ?3, seq = ?4 WHERE repo_id = ?1",
        rusqlite::params![
            record.repo_id,
            visibility_str(record.visibility),
            grants,
            seq
        ],
    )
    .map_err(|e| io_err("registry update record row", e))?;
    tx.commit()
        .map_err(|e| io_err("registry update commit", e))?;
    Ok(seq)
}

/// The `remove`'s transaction: existence first
/// ([`RegistryError::NotFound`]), the seq bump, then the record row,
/// the repo's ref rows (the HEAD seed included), the repo's
/// mutation-marker rows, and the repo's LFS liveness rows — the whole
/// repo's co-tenant state in one removal. The `lfs_objects` leg is
/// unconditional (the table ships unconditionally in the step set;
/// the rows only exist when the `lfs` surface wrote them, and a mixed
/// deployment that turned the feature off must not leak), followed by
/// the shared-oids' refcount recompute (`substrate/src/lfs.rs`'s
/// semantics — every surviving row of an oid reads the same count at
/// commit). Objects and packs are not touched (deletion is a GC
/// consequence).
fn durable_remove(handle: &CoTenantSqlite, repo_id: &str) -> Result<i64, RegistryError> {
    let mut conn = handle.lock();
    let tx = conn
        .transaction_with_behavior(rusqlite::TransactionBehavior::Immediate)
        .map_err(|e| io_err("registry remove BEGIN IMMEDIATE", e))?;
    let exists: i64 = tx
        .query_row(
            "SELECT count(*) FROM git_repos WHERE repo_id = ?1",
            rusqlite::params![repo_id],
            |r| r.get(0),
        )
        .map_err(|e| io_err("registry remove existence read", e))?;
    if exists == 0 {
        return Err(RegistryError::NotFound);
    }
    let seq: i64 = tx
        .query_row(
            "UPDATE registry_seq SET value = value + 1 WHERE id = 1 RETURNING value",
            [],
            |r| r.get(0),
        )
        .map_err(|e| io_err("registry remove seq bump", e))?;
    tx.execute(
        "DELETE FROM git_repos WHERE repo_id = ?1",
        rusqlite::params![repo_id],
    )
    .map_err(|e| io_err("registry remove record row", e))?;
    tx.execute(
        "DELETE FROM git_refs WHERE repo_id = ?1",
        rusqlite::params![repo_id],
    )
    .map_err(|e| io_err("registry remove ref rows", e))?;
    tx.execute(
        "DELETE FROM repo_mutation_markers WHERE repo_id = ?1",
        rusqlite::params![repo_id],
    )
    .map_err(|e| io_err("registry remove marker rows", e))?;
    tx.execute(
        "DELETE FROM lfs_objects WHERE repo_id = ?1",
        rusqlite::params![repo_id],
    )
    .map_err(|e| io_err("registry remove lfs rows", e))?;
    tx.execute(
        "UPDATE lfs_objects \
         SET refcount = (SELECT count(*) FROM lfs_objects AS c WHERE c.oid = lfs_objects.oid)",
        [],
    )
    .map_err(|e| io_err("registry remove lfs recount", e))?;
    tx.commit()
        .map_err(|e| io_err("registry remove commit", e))?;
    Ok(seq)
}

fn durable_watermark(handle: &CoTenantSqlite) -> Result<i64, RegistryError> {
    let conn = handle.lock();
    conn.query_row("SELECT value FROM registry_seq WHERE id = 1", [], |r| {
        r.get(0)
    })
    .map_err(|e| io_err("registry watermark read", e))
}

fn durable_rows_since(
    handle: &CoTenantSqlite,
    seq: i64,
) -> Result<Vec<RegistryRow>, RegistryError> {
    let conn = handle.lock();
    let mut stmt = conn
        .prepare(
            "SELECT repo_id, visibility, grants, seq FROM git_repos \
             WHERE seq > ?1 ORDER BY seq",
        )
        .map_err(|e| io_err("registry rows_since prepare", e))?;
    let rows = stmt
        .query_map(rusqlite::params![seq], |row| {
            Ok((
                row.get::<_, String>(0)?,
                row.get::<_, String>(1)?,
                row.get::<_, String>(2)?,
                row.get::<_, i64>(3)?,
            ))
        })
        .map_err(|e| io_err("registry rows_since query", e))?;
    let mut since = Vec::new();
    for row in rows {
        let (repo_id, visibility, grants, row_seq) =
            row.map_err(|e| io_err("registry rows_since row", e))?;
        since.push(row_to_record(&repo_id, &visibility, &grants, row_seq)?);
    }
    Ok(since)
}

/// The marker row upsert inside its one transaction: existence first
/// (the receive flow resolved the record), the seq bump, the marker
/// row.
fn durable_record_mutation(handle: &CoTenantSqlite, repo_id: &str) -> Result<i64, RegistryError> {
    let mut conn = handle.lock();
    let tx = conn
        .transaction_with_behavior(rusqlite::TransactionBehavior::Immediate)
        .map_err(|e| io_err("mutation marker BEGIN IMMEDIATE", e))?;
    let exists: i64 = tx
        .query_row(
            "SELECT count(*) FROM git_repos WHERE repo_id = ?1",
            rusqlite::params![repo_id],
            |r| r.get(0),
        )
        .map_err(|e| io_err("mutation marker existence read", e))?;
    if exists == 0 {
        return Err(RegistryError::NotFound);
    }
    let seq: i64 = tx
        .query_row(
            "UPDATE registry_seq SET value = value + 1 WHERE id = 1 RETURNING value",
            [],
            |r| r.get(0),
        )
        .map_err(|e| io_err("mutation marker seq bump", e))?;
    tx.execute(
        "INSERT INTO repo_mutation_markers (repo_id, seq) VALUES (?1, ?2) \
         ON CONFLICT(repo_id) DO UPDATE SET seq = excluded.seq",
        rusqlite::params![repo_id, seq],
    )
    .map_err(|e| io_err("mutation marker row", e))?;
    tx.commit()
        .map_err(|e| io_err("mutation marker commit", e))?;
    Ok(seq)
}

/// The sqlite twin assembly — the pieces one adapter constructor wires
/// over the default engine assembly (single-host shape, substrate.md
/// §assembly: `alkstore-sqlite` + the co-tenant twin on one file):
///
/// - [`SqliteRegistryAssembly::store`] — the opened alkstore engine
///   handle (the reactive halves' substrate).
/// - [`SqliteRegistryAssembly::write`] — the **announced write side**
///   ([`AnnouncingRegistry`]): the pair with the post-commit announce
///   wired by default (ADR-019 §5 — no non-reactive shipped path).
/// - [`SqliteRegistryAssembly::durable`] — the bare twin (announce
///   free — the probe legs run here).
/// - [`SqliteRegistryAssembly::provider`] — the cached read provider,
///   loaded once from the durable side at construction (loud — a
///   registry serving an empty snapshot over a populated table would
///   contradict the definite-miss read contract).
/// - [`SqliteRegistryAssembly::reactor`] — the wake runner/stream
///   consumer/reconcile machinery, holding the durable side.
/// - [`SqliteRegistryAssembly::refs`] — the refs row CAS over the same
///   co-tenant handle, write-counter-wrapped ([`VersionedRefTable`];
///   the seeded HEAD row is its unborn shape) — the counter
///   [`SqliteRegistryAssembly::refs_version`] exposes is the liveness
///   source's arbitration-window signal.
/// - [`SqliteRegistryAssembly::backend`] — the core-traits binding
///   ([`RegistryBackend::with_cached_provider`]) over the announced
///   write side: the store methods announce post-commit and write
///   through.
/// - [`SqliteRegistryAssembly::lfs_rows`] — the LFS pointer-lifecycle
///   row CAS over the same co-tenant handle, write-counter-wrapped
///   ([`VersionedLfsRows`](crate::lfs::VersionedLfsRows) — the
///   refs-twins' wrap posture; the composition entry composes it
///   unchanged) — the counter [`SqliteRegistryAssembly::lfs_version`]
///   exposes is the LFS liveness source's arbitration-window signal.
///
/// The wave-8 single composition entry unifies construction; this is
/// the per-piece constructor (the recorded interpretation, flagged for
/// the gate).
pub struct SqliteRegistryAssembly {
    store: Arc<dyn alkstore::Store>,
    durable: Arc<SqliteRegistry>,
    registry: Arc<dyn RegistryWrite>,
    provider: Arc<CachedRegistryProvider>,
    reactor: Arc<Reactor>,
    refs: Arc<VersionedRefTable<SqliteRefTable>>,
    refs_version: Arc<AtomicU64>,
    backend: Arc<RegistryBackend>,
    #[cfg(feature = "lfs")]
    lfs: Arc<crate::lfs::VersionedLfsRows<crate::lfs::SqliteLfsRows>>,
    #[cfg(feature = "lfs")]
    lfs_version: Arc<AtomicU64>,
}

impl SqliteRegistryAssembly {
    /// The adapter constructor: open the co-tenant connection at
    /// `path` (migrations at construction, loud — a half-migrated
    /// store refuses service; `:memory:` confers no co-tenancy and is
    /// rejected before the engine opens), open the alkstore-sqlite
    /// engine at the same path, load the provider's initial snapshot,
    /// and wire the reactive composition. Every failure is
    /// [`OpenError`].
    pub async fn open(path: impl AsRef<Path>) -> Result<Self, OpenError> {
        let path = path.as_ref();
        let handle = CoTenantSqlite::open(path)?;
        let path_str = path.to_str().ok_or_else(|| {
            OpenError::Invalid(format!(
                "sqlite registry path is not valid utf-8: {}",
                path.display()
            ))
        })?;
        let store: Arc<dyn alkstore::Store> = Arc::from(
            alkstore_sqlite::open(path_str, Default::default()).map_err(|e| {
                OpenError::Engine(format!("alkstore-sqlite open ({}): {e}", path.display()))
            })?,
        );
        let durable: Arc<SqliteRegistry> = Arc::new(SqliteRegistry::new(handle));

        let provider = Arc::new(CachedRegistryProvider::new());
        let rows = durable
            .rows_since(0)
            .await
            .map_err(|e| OpenError::Engine(format!("cached provider initial load: {e}")))?;
        provider.install(rows);

        let durable_side: Arc<dyn RegistryWrite> = durable.clone();
        let reactor = Arc::new(Reactor::new(
            store.clone(),
            durable_side.clone(),
            provider.clone(),
        ));
        let registry: Arc<dyn RegistryWrite> = Arc::new(AnnouncingRegistry::new(
            durable_side.clone(),
            reactor.clone(),
        ));
        let backend = Arc::new(RegistryBackend::with_cached_provider(
            registry.clone(),
            provider.clone(),
        ));
        let refs_version = Arc::new(AtomicU64::new(0));
        let refs = Arc::new(VersionedRefTable::new(
            Arc::new(SqliteRefTable::new(durable.handle.clone())),
            refs_version.clone(),
        ));
        #[cfg(feature = "lfs")]
        let lfs_version = Arc::new(AtomicU64::new(0));
        #[cfg(feature = "lfs")]
        let lfs = Arc::new(crate::lfs::VersionedLfsRows::new(
            Arc::new(crate::lfs::SqliteLfsRows::new(durable.handle.clone())),
            lfs_version.clone(),
        ));
        Ok(SqliteRegistryAssembly {
            store,
            durable,
            registry,
            provider,
            reactor,
            refs,
            refs_version,
            backend,
            #[cfg(feature = "lfs")]
            lfs,
            #[cfg(feature = "lfs")]
            lfs_version,
        })
    }

    /// The opened alkstore engine handle (the reactive halves'
    /// substrate; the wave-7 maintenance chain rides the same).
    pub fn store(&self) -> &Arc<dyn alkstore::Store> {
        &self.store
    }

    /// The announced write side — the pair with the post-commit
    /// announce wired (the write surface the composition's callers
    /// use).
    pub fn write(&self) -> &Arc<dyn RegistryWrite> {
        &self.registry
    }

    /// The bare twin (announce-free — the transactional legs' probe
    /// surface).
    pub fn durable(&self) -> &Arc<SqliteRegistry> {
        &self.durable
    }

    /// The cached read provider (the hot read).
    pub fn provider(&self) -> &Arc<CachedRegistryProvider> {
        &self.provider
    }

    /// The reactor (the wake runner / stream consumer / reconcile
    /// sweep run through it).
    pub fn reactor(&self) -> &Arc<Reactor> {
        &self.reactor
    }

    /// The refs row CAS over the same co-tenant handle — the
    /// write-counter-wrapped composition refs ([`VersionedRefTable`]):
    /// the CAS semantics are unchanged (verbatim delegation), every
    /// CAS call bumps the shared counter the liveness source's
    /// arbitration consult reads (crate::liveness — the composition's
    /// invalidation half). Pushes route through this impl.
    pub fn refs(&self) -> &Arc<VersionedRefTable<SqliteRefTable>> {
        &self.refs
    }

    /// The refs writes' shared counter — the same handle
    /// [`Self::refs`] bumps and the [`crate::liveness::RefClosure`]
    /// source consults; hand it to the closure's constructor.
    pub fn refs_version(&self) -> &Arc<AtomicU64> {
        &self.refs_version
    }

    /// The core-traits binding (the engine-twin binding announced and
    /// write-through wired).
    pub fn backend(&self) -> &Arc<RegistryBackend> {
        &self.backend
    }

    /// The LFS pointer-lifecycle row CAS over the same co-tenant
    /// handle — the write-counter-wrapped rows
    /// ([`crate::lfs::VersionedLfsRows`]; the refs twins' wrap posture,
    /// the per-piece parity posture the per-piece-path embedder
    /// composes): the row semantics are unchanged (verbatim
    /// delegation), every `assert` bumps the shared counter the LFS
    /// liveness source's arbitration consult reads (`crate::liveness`
    /// — the composition's invalidation half). The record-removal leg
    /// takes a removed repo's rows with the record transactionally
    /// (the twins' unconditional in-SQL copy).
    #[cfg(feature = "lfs")]
    pub fn lfs_rows(&self) -> &Arc<crate::lfs::VersionedLfsRows<crate::lfs::SqliteLfsRows>> {
        &self.lfs
    }

    /// The LFS rows' asserts' shared counter — the same handle
    /// [`Self::lfs_rows`] bumps and the
    /// [`crate::liveness::LfsMedia`] source consults; hand it to the
    /// media source's constructor.
    #[cfg(feature = "lfs")]
    pub fn lfs_version(&self) -> &Arc<AtomicU64> {
        &self.lfs_version
    }
}

/// The per-push mutation-marker write, announced
/// ([`RegistryEvent::Mutation`] post-commit, fail-soft) — the receive
/// flow's watermark leg (ADR-019 §7; the composition-internal
/// [`RegistryMutations`](crate::pins::RegistryMutations) legs the
/// wrapped refs implement rides — [`crate::pins::CommittedRefs`]; the
/// record schema untouched). The durable leg carries no marker-driven
/// record change: `rows_since` never carries markers.
#[async_trait]
impl crate::pins::RegistryMutations for SqliteRegistryAssembly {
    async fn record_mutation(&self, repo_id: &str) -> Result<i64, RegistryError> {
        let seq = self.durable.record_mutation(repo_id).await?;
        self.reactor
            .announce(&RegistryEvent::Mutation {
                seq,
                repo_id: repo_id.to_owned(),
            })
            .await;
        Ok(seq)
    }
}

#[cfg(all(test, feature = "sqlite"))]
mod tests {
    use std::time::Duration;

    use crate::registry::RegistryRead;
    use crate::testutil::{
        private_record, record, spawn_wake_runner, stop_and_join, unique_db_path, wait_until,
    };

    use alkgit::{AccessAction, GitRefs, GitRegistry, GitRegistryStore, Visibility};
    use alkstore::{QueueOpts, StopToken, Store, StreamEvent, WakeReceiver};
    use serde_json::Value;

    use super::*;

    /// One co-tenant file with the twin mounted — the durable-side
    /// probe without the reactive wiring (the assembly is over it).
    async fn durable_twin(tag: &str) -> (CoTenantSqlite, SqliteRegistry) {
        let path = unique_db_path("sqlite", tag);
        let handle = CoTenantSqlite::open(&path).unwrap();
        let twin = SqliteRegistry::new(handle.clone());
        (handle, twin)
    }

    fn count(handle: &CoTenantSqlite, sql: &str, repo_id: &str) -> i64 {
        handle
            .lock()
            .query_row(sql, rusqlite::params![repo_id], |r| r.get(0))
            .unwrap()
    }

    #[tokio::test]
    async fn durable_twin_carries_the_family_write_semantics() {
        let (_, twin) = durable_twin("family").await;

        let seq_put = twin.put(&record("owner/r")).await.unwrap();
        assert_eq!(seq_put, 1, "first mutation carries seq 1");

        let mut granted = record("owner/r");
        granted.visibility = Visibility::Private;
        granted
            .grants
            .insert("u-7731".to_owned(), vec![AccessAction::Write]);
        let seq_update = twin.update(&granted).await.unwrap();
        assert_eq!(seq_update, 2);
        let rows = twin.rows_since(0).await.unwrap();
        assert_eq!(rows.len(), 1);
        assert_eq!(rows[0].seq, seq_update);
        assert_eq!(
            rows[0].record, granted,
            "the record round-trips the columns"
        );

        assert_eq!(twin.watermark().await.unwrap(), 2);
        let seq_remove = twin.remove("owner/r").await.unwrap();
        assert_eq!(seq_remove, 3);
        assert_eq!(twin.watermark().await.unwrap(), 3);
        assert!(
            twin.rows_since(0).await.unwrap().is_empty(),
            "the committed removal leaves no row"
        );
    }

    #[tokio::test]
    async fn durable_twin_error_map_matches_the_family() {
        let (_, twin) = durable_twin("errors").await;

        assert_eq!(
            twin.update(&record("ghost")).await.unwrap_err(),
            RegistryError::NotFound
        );
        assert_eq!(
            twin.remove("ghost").await.unwrap_err(),
            RegistryError::NotFound
        );
        assert_eq!(
            twin.record_mutation("ghost").await.unwrap_err(),
            RegistryError::NotFound,
            "a marker write is refused for a removed repo"
        );

        assert_eq!(twin.put(&record("owner/r")).await.unwrap(), 1);
        assert_eq!(
            twin.put(&record("owner/r")).await.unwrap_err(),
            RegistryError::AlreadyExists
        );
        assert_eq!(
            twin.watermark().await.unwrap(),
            1,
            "the failed duplicate put consumed no seq"
        );

        let malformed = record("Owner/name");
        assert_eq!(
            twin.put(&malformed).await.unwrap_err(),
            RegistryError::Invalid
        );
        let mut empty_key = record("empty");
        empty_key
            .grants
            .insert(String::new(), vec![AccessAction::Read]);
        assert_eq!(
            twin.put(&empty_key).await.unwrap_err(),
            RegistryError::Invalid
        );
        assert_eq!(
            twin.watermark().await.unwrap(),
            1,
            "the rejections consumed no seq"
        );
    }

    /// The seq is transactionally exact (the watermark's anchor,
    /// ADR-019 §5): the bump lives in the row-write transaction — a
    /// read on a second connection right after the commit sees the
    /// bumped seq and the row, before any announce could fire.
    #[tokio::test]
    async fn the_seq_bump_is_transactionally_exact() {
        let path = unique_db_path("sqlite", "tx-seq");
        let handle = CoTenantSqlite::open(&path).unwrap();
        let twin = SqliteRegistry::new(handle.clone());
        let second = CoTenantSqlite::open(&path).unwrap();

        let seq = twin.put(&record("owner/tx")).await.unwrap();

        {
            let db = second.lock();
            assert_eq!(
                db.query_row("SELECT value FROM registry_seq WHERE id = 1", [], |r| r
                    .get::<_, i64>(0))
                    .unwrap(),
                seq,
                "the second connection sees the bumped seq — it committed with the row"
            );
            assert_eq!(
                db.query_row(
                    "SELECT seq FROM git_repos WHERE repo_id = 'owner/tx'",
                    [],
                    |r| r.get::<_, i64>(0)
                )
                .unwrap(),
                seq,
            );
            assert_eq!(
                db.query_row(
                    "SELECT symref_target FROM git_refs WHERE repo_id = 'owner/tx' AND name = 'HEAD'",
                    [],
                    |r| r.get::<_, String>(0)
                )
                .unwrap(),
                HEAD_SEED_TARGET,
                "the refs-table leg rode the same transaction"
            );
        }

        twin.update(&private_record("owner/tx")).await.unwrap();
        twin.remove("owner/tx").await.unwrap();

        {
            let db = second.lock();
            assert_eq!(
                db.query_row("SELECT value FROM registry_seq WHERE id = 1", [], |r| r
                    .get::<_, i64>(0))
                    .unwrap(),
                3,
            );
        }

        let reopened = CoTenantSqlite::open(&path).unwrap();
        let fresh = SqliteRegistry::new(reopened);
        assert_eq!(
            fresh.watermark().await.unwrap(),
            3,
            "the seq singleton is durable"
        );
        assert!(
            fresh.put(&record("owner/tx")).await.unwrap() > 3,
            "the bump continues from the durable watermark"
        );
        drop(fresh);
        drop(handle);
        drop(second);
        let _ = std::fs::remove_file(&path);
    }

    /// put_repo seeds `HEAD → refs/heads/main` in the record
    /// transaction (substrate.md §refs): a fresh repo lists the unborn
    /// HEAD line via the refs twin; update leaves the row alone;
    /// remove clears it with the record.
    #[tokio::test]
    async fn put_repo_seeds_the_unborn_head_line() {
        let path = unique_db_path("sqlite", "seed-head");
        let assembly = SqliteRegistryAssembly::open(&path).await.unwrap();
        let repo = record("owner/seeded");

        assembly.durable().put(&repo).await.unwrap();
        let listed = assembly.refs().list_refs(&repo).await.unwrap();
        assert_eq!(
            listed,
            vec![alkgit::RefLine {
                name: "HEAD".to_owned(),
                oid: None,
                peeled: None,
                symref_target: Some(HEAD_SEED_TARGET.to_owned()),
            }],
            "the fresh repo lists the unborn HEAD line (the OQ-19 rider (a) fixture)"
        );

        assembly
            .durable()
            .update(&private_record("owner/seeded"))
            .await
            .unwrap();
        assert_eq!(
            assembly.refs().list_refs(&repo).await.unwrap().len(),
            1,
            "the update's transaction does not touch the seeded row"
        );

        assembly.durable().remove("owner/seeded").await.unwrap();
        assert!(
            assembly.refs().list_refs(&repo).await.unwrap().is_empty(),
            "the removal cleared the seeded row with the record"
        );
    }

    /// remove_repo removes the registry row + the repo's ref rows +
    /// the repo's marker rows in the same removal (backend.md:
    /// deletion is a GC consequence — objects stay for the sweep; the
    /// only durable changes are the co-tenant rows).
    #[tokio::test]
    async fn remove_repo_drops_the_rows_the_removal_owes_together() {
        let path = unique_db_path("sqlite", "remove");
        let assembly = SqliteRegistryAssembly::open(&path).await.unwrap();

        assembly.durable().put(&record("survivor")).await.unwrap();
        assembly.durable().put(&record("victim")).await.unwrap();
        assembly.durable().record_mutation("victim").await.unwrap();
        assembly
            .refs()
            .apply_updates(
                &record("victim"),
                vec![alkgit::RefUpdate {
                    name: "refs/heads/main".to_owned(),
                    expected: None,
                    new: Some(gix_hash::ObjectId::from([7; 20])),
                }],
                false,
            )
            .await
            .unwrap();

        let engine_event = assembly
            .store()
            .stream(crate::reactor::EVENTS_STREAM)
            .await
            .unwrap()
            .publish(serde_json::json!("kept — no object deletion exists"))
            .await
            .unwrap();

        assembly.backend().remove_repo("victim").await.unwrap();

        let handle = CoTenantSqlite::open(&path).unwrap();
        assert_eq!(
            count(
                &handle,
                "SELECT count(*) FROM git_repos WHERE repo_id = ?1",
                "victim"
            ),
            0
        );
        assert_eq!(
            count(
                &handle,
                "SELECT count(*) FROM git_refs WHERE repo_id = ?1",
                "victim"
            ),
            0
        );
        assert_eq!(
            count(
                &handle,
                "SELECT count(*) FROM repo_mutation_markers WHERE repo_id = ?1",
                "victim"
            ),
            0
        );
        assert_eq!(
            count(
                &handle,
                "SELECT count(*) FROM git_repos WHERE repo_id = ?1",
                "survivor"
            ),
            1
        );
        drop(handle);

        let events = assembly
            .store()
            .stream(crate::reactor::EVENTS_STREAM)
            .await
            .unwrap()
            .read_since(0, 100)
            .await
            .unwrap();
        assert!(
            events.iter().any(|event: &StreamEvent| {
                event.payload_as::<Value>().is_ok_and(|payload| {
                    payload == serde_json::json!("kept — no object deletion exists")
                })
            }),
            "the engine-side data is untouched by the removal"
        );
        assert!(engine_event > 0);
    }

    /// The LFS liveness rows go with the removed repo — the
    /// unconditional `lfs_objects` leg inside the record-removal
    /// transaction (the rows only exist when the `lfs` surface wrote
    /// them; the leg is unconditional so a mixed deployment that
    /// turned the feature off must not leak). The rows seed directly
    /// here (the leg's SQL is unconditional — the test holds with the
    /// surface off too), a shared oid's surviving row recomputes, and
    /// the victim's oids leave no row at all.
    #[tokio::test]
    async fn remove_repo_takes_the_lfs_rows_and_recounts_the_survivors() {
        let path = unique_db_path("sqlite", "remove-lfs");
        let assembly = SqliteRegistryAssembly::open(&path).await.unwrap();
        assembly.durable().put(&record("victim")).await.unwrap();
        assembly.durable().put(&record("survivor")).await.unwrap();

        let shared_oid = "01".repeat(32);
        let victim_only_oid = "02".repeat(32);
        {
            let conn = assembly.durable().handle.lock();
            conn.execute(
                "INSERT INTO lfs_objects (repo_id, oid, size, refcount) \
                 VALUES ('victim', ?1, 8, 2), ('survivor', ?1, 8, 2), \
                         ('victim', ?2, 4, 1)",
                rusqlite::params![shared_oid, victim_only_oid],
            )
            .unwrap();
        }

        assembly.backend().remove_repo("victim").await.unwrap();

        {
            let conn = assembly.durable().handle.lock();
            let victim_only: i64 = conn
                .query_row(
                    "SELECT count(*) FROM lfs_objects WHERE oid = ?1",
                    rusqlite::params![victim_only_oid],
                    |r| r.get(0),
                )
                .unwrap();
            assert_eq!(victim_only, 0, "the victim's own oid left no row at all");
            let shared_count: i64 = conn
                .query_row(
                    "SELECT refcount FROM lfs_objects WHERE repo_id = 'survivor' AND oid = ?1",
                    rusqlite::params![shared_oid],
                    |r| r.get(0),
                )
                .unwrap();
            assert_eq!(
                shared_count, 1,
                "the surviving row carries the recomputed count (the re-protection contract's data)"
            );
        }

        #[cfg(feature = "lfs")]
        assert_eq!(
            crate::lfs::LfsRows::repo_rows(assembly.lfs_rows().as_ref(), "survivor")
                .await
                .unwrap(),
            vec![crate::lfs::LfsRow {
                repo_id: "survivor".to_owned(),
                oid: shared_oid,
                size: 8,
                refcount: 1,
            }],
            "the surface reads back the same state (the composition's row store)"
        );
    }

    /// The per-push mutation-marker write: the seq/watermark legs move
    /// without touching the record (ADR-019 §7 — the receive flow's
    /// watermark leg; rows_since never carries markers).
    #[tokio::test]
    async fn record_mutation_bumps_the_watermark_without_touching_the_record() {
        let (handle, twin) = durable_twin("marker").await;

        let put_seq = twin.put(&record("owner/m")).await.unwrap();
        assert_eq!(put_seq, 1);
        let seq = twin.record_mutation("owner/m").await.unwrap();

        assert_eq!(seq, 2);
        assert_eq!(twin.watermark().await.unwrap(), 2);
        let rows = twin.rows_since(0).await.unwrap();
        assert_eq!(
            rows.iter().map(|row| row.seq).collect::<Vec<_>>(),
            vec![put_seq],
            "the marker left no record row — the record schema is untouched"
        );
        let marker: i64 = handle
            .lock()
            .query_row(
                "SELECT seq FROM repo_mutation_markers WHERE repo_id = 'owner/m'",
                [],
                |r| r.get(0),
            )
            .unwrap();
        assert_eq!(marker, seq, "the marker row carries its seq");

        twin.remove("owner/m").await.unwrap();
        let left: i64 = handle
            .lock()
            .query_row(
                "SELECT count(*) FROM repo_mutation_markers WHERE repo_id = 'owner/m'",
                [],
                |r| r.get(0),
            )
            .unwrap();
        assert_eq!(left, 0, "the removal took the repo's marker rows with it");
    }

    /// The announcement through the composed store binding fires
    /// post-commit: the Put event is keyed in the feed, the wake lands
    /// on a listener, the pack-housekeeping job is on the maintenance
    /// queue, and the resolve rides the write-through provider.
    #[tokio::test]
    async fn announced_store_binding_fires_the_composition_post_commit() {
        use crate::reactor::MaintenanceJob;
        use crate::reactor::{RegistryEvent, MAINTENANCE_QUEUE};

        let path = unique_db_path("sqlite", "announce");
        let assembly = SqliteRegistryAssembly::open(&path).await.unwrap();
        let mut listener = assembly
            .store()
            .listen(crate::reactor::WAKE_CHANNEL)
            .await
            .unwrap();
        let mut granted = record("owner/a");
        granted
            .grants
            .insert("u-1".to_owned(), vec![AccessAction::Read]);

        assembly.backend().put_repo(&granted).await.unwrap();

        let events = assembly
            .store()
            .stream(crate::reactor::EVENTS_STREAM)
            .await
            .unwrap()
            .read_since(0, 10)
            .await
            .unwrap();
        assert_eq!(events.len(), 1, "one announced put = one feed event");
        assert_eq!(events[0].key.as_deref(), Some("owner/a"));
        assert_eq!(
            crate::reactor::RegistryEvent::decode_event(&events[0]),
            Some(RegistryEvent::Put {
                seq: 1,
                record: granted.clone(),
            })
        );
        let wake = listener
            .recv_timeout(Duration::from_secs(2))
            .await
            .unwrap()
            .expect("wake source open");
        assert_eq!(wake.channel, crate::reactor::WAKE_CHANNEL);

        let queue = assembly
            .store()
            .queue(MAINTENANCE_QUEUE, QueueOpts::default())
            .await
            .unwrap();
        let claimed = queue
            .claim_one("probe")
            .await
            .unwrap()
            .expect("the pack-housekeeping job");
        assert_eq!(
            claimed.job().payload_as::<MaintenanceJob>().unwrap(),
            MaintenanceJob::PackHousekeeping {
                repo_id: "owner/a".to_owned(),
            }
        );

        assert_eq!(
            assembly.backend().resolve("owner/a").await.unwrap(),
            granted,
            "the write-through provider serves the mutation"
        );

        assembly.backend().remove_repo("owner/a").await.unwrap();
        let events = assembly
            .store()
            .stream(crate::reactor::EVENTS_STREAM)
            .await
            .unwrap()
            .read_since(1, 10)
            .await
            .unwrap();
        assert_eq!(
            events.first().and_then(RegistryEvent::decode_event),
            Some(RegistryEvent::Removed {
                seq: 2,
                repo_id: "owner/a".to_owned(),
            })
        );
    }

    /// A store double with every announce leg's entry point failing
    /// (the rest rides the mem engine).
    struct LegsDeadStore {
        inner: Box<dyn Store>,
    }

    impl Store for LegsDeadStore {
        fn begin_tx(
            &self,
        ) -> alkstore::BoxedFuture<'_, alkstore::Result<Box<dyn alkstore::TxHandle + Send>>>
        {
            self.inner.begin_tx()
        }

        fn notify<'a>(
            &'a self,
            channel: &str,
            payload: Value,
        ) -> alkstore::BoxedFuture<'a, alkstore::Result<()>> {
            self.inner.notify(channel, payload)
        }

        fn listen<'a>(
            &'a self,
            channel: &str,
        ) -> alkstore::BoxedFuture<'a, alkstore::Result<Box<dyn WakeReceiver>>> {
            self.inner.listen(channel)
        }

        fn stream<'a>(
            &'a self,
            _name: &str,
        ) -> alkstore::BoxedFuture<'a, alkstore::Result<Box<dyn alkstore::StreamHandle>>> {
            Box::pin(async { Err(alkstore::Error::database("publish leg suppressed (probe)")) })
        }

        fn queue<'a>(
            &'a self,
            _name: &str,
            _opts: alkstore::QueueOpts,
        ) -> alkstore::BoxedFuture<'a, alkstore::Result<Box<dyn alkstore::Queue>>> {
            Box::pin(async { Err(alkstore::Error::database("enqueue leg suppressed (probe)")) })
        }

        fn outbox<'a>(
            &'a self,
            name: &str,
        ) -> alkstore::BoxedFuture<'a, alkstore::Result<Box<dyn alkstore::Outbox>>> {
            self.inner.outbox(name)
        }

        fn try_lock<'a>(
            &'a self,
            name: &str,
            owner: &str,
            ttl: i64,
        ) -> alkstore::BoxedFuture<'a, alkstore::Result<Option<Box<dyn alkstore::Lock>>>> {
            self.inner.try_lock(name, owner, ttl)
        }

        fn schedule<'a>(
            &'a self,
            name: &str,
            spec: &str,
            queue: &str,
            payload: Value,
            opts: alkstore::ScheduleOpts,
        ) -> alkstore::BoxedFuture<'a, alkstore::Result<alkstore::Schedule>> {
            self.inner.schedule(name, spec, queue, payload, opts)
        }

        fn unschedule<'a>(
            &'a self,
            name: &str,
        ) -> alkstore::BoxedFuture<'a, alkstore::Result<bool>> {
            self.inner.unschedule(name)
        }

        fn run_schedules<'a>(
            &'a self,
            stop: StopToken,
        ) -> alkstore::BoxedFuture<'a, alkstore::Result<()>> {
            self.inner.run_schedules(stop)
        }
    }

    /// The announce's legs are fail-soft and the write's seq is
    /// transactionally exact: every leg dead still lands the durable
    /// commit (a read between write and announce sees the bumped seq),
    /// and the reconcile sweep closes the window.
    #[tokio::test]
    async fn dead_announce_legs_fail_soft_and_the_seq_stays_exact() {
        let path = unique_db_path("sqlite", "fail-soft");
        let handle = CoTenantSqlite::open(&path).unwrap();
        let durable: Arc<dyn RegistryWrite> = Arc::new(SqliteRegistry::new(handle));
        let provider = Arc::new(CachedRegistryProvider::new());
        let reactor = Arc::new(Reactor::new(
            Arc::new(LegsDeadStore {
                inner: Box::new(alkstore_mem::MemStore::new()),
            }) as Arc<dyn Store>,
            durable.clone(),
            provider.clone(),
        ));
        let announced: Arc<dyn RegistryWrite> =
            Arc::new(AnnouncingRegistry::new(durable.clone(), reactor.clone()));
        let backend = RegistryBackend::with_cached_provider(announced.clone(), provider.clone());

        backend.put_repo(&record("owner/x")).await.unwrap();
        assert_eq!(
            durable.watermark().await.unwrap(),
            1,
            "the seq is transactionally exact — not announce-dependent"
        );
        assert_eq!(durable.rows_since(0).await.unwrap().len(), 1);
        assert_eq!(
            provider.resolve("owner/x"),
            Some(record("owner/x")),
            "the write-through install is separate machinery and still landed"
        );

        let stream = alkstore_sqlite::open(path.to_str().unwrap(), Default::default()).unwrap();
        let events: Vec<StreamEvent> = stream
            .stream(crate::reactor::EVENTS_STREAM)
            .await
            .unwrap()
            .read_since(0, 10)
            .await
            .unwrap();
        assert!(
            events.is_empty(),
            "the suppressed publish leg left no event — the crash window"
        );

        let applied = crate::reactor::AppliedWatermark::new();
        let caught = reactor.reconcile(&applied).await.unwrap();
        assert_eq!(caught.len(), 1, "the crash window closes in one pass");
        assert_eq!(provider.resolve("owner/x"), Some(record("owner/x")));
        drop(stream);
        drop(backend);
        drop(durable);
        drop(reactor);
        let _ = std::fs::remove_file(&path);
    }

    /// The co-tenant file is itself the invalidation bus (substrate.md
    /// §the reactive halves, POC-R's backstop): an unannounced
    /// registry write on its own connection wakes a listener on a
    /// second engine connection, and the wake runner's refresh
    /// surfaces the row in the provider — milliseconds, not the
    /// reconcile cadence.
    #[tokio::test]
    async fn the_co_tenant_file_is_the_invalidation_bus() {
        let path = unique_db_path("sqlite", "backstop");
        let assembly = SqliteRegistryAssembly::open(&path).await.unwrap();
        let mut listener = assembly
            .store()
            .listen(crate::reactor::WAKE_CHANNEL)
            .await
            .unwrap();
        let (stop, runner) = spawn_wake_runner(assembly.reactor().clone());
        tokio::time::sleep(Duration::from_millis(200)).await;

        let seq = assembly
            .durable()
            .put(&private_record("owner/silent"))
            .await
            .unwrap();

        let wake = listener
            .recv_timeout(Duration::from_secs(2))
            .await
            .unwrap()
            .expect(
                "the file-wide data_version bump fans the wake cross-connection — \
                 no notify involved",
            );
        assert_eq!(wake.channel, crate::reactor::WAKE_CHANNEL);
        drop(listener);

        wait_until(|| assembly.provider().resolve("owner/silent").is_some()).await;
        assert_eq!(
            assembly.provider().resolve("owner/silent"),
            Some(private_record("owner/silent")),
            "the unannounced write surfaced through the wake-driven refresh"
        );
        assert!(assembly.provider().refresh_count() >= 1);

        assert_eq!(
            assembly.durable().watermark().await.unwrap(),
            seq,
            "the watermark advanced though nothing announced"
        );
        stop_and_join(stop, runner).await;
    }

    /// The assembly constructor is loud and loads the persisted
    /// registry: a reopened assembly resolves the stored record with
    /// no refresh, the seq singleton persists, and `:memory:` is
    /// rejected before the engine opens (DQ-7).
    #[tokio::test]
    async fn assembly_open_is_loud_and_loads_the_persisted_registry() {
        let path = unique_db_path("sqlite", "reopen");
        {
            let assembly = SqliteRegistryAssembly::open(&path).await.unwrap();
            assembly
                .backend()
                .put_repo(&private_record("owner/kept"))
                .await
                .unwrap();
            assert_eq!(assembly.durable().watermark().await.unwrap(), 1);
        }

        let reopened = SqliteRegistryAssembly::open(&path).await.unwrap();
        assert_eq!(
            reopened.backend().resolve("owner/kept").await.unwrap(),
            private_record("owner/kept"),
            "the initial snapshot load carried the persisted table"
        );
        assert_eq!(reopened.durable().watermark().await.unwrap(), 1);
        assert_eq!(reopened.provider().refresh_count(), 1, "load only");

        // the initial load is loud: a broken durable side refuses
        // assembly (the twin's reads fail as the provider load maps
        // into OpenError)

        assert!(matches!(
            SqliteRegistryAssembly::open(":memory:").await,
            Err(OpenError::Invalid(_))
        ));

        let _ = std::fs::remove_file(&path);
    }

    /// The corrupt row defenses: a row the composition did not write
    /// (malformed grants JSON) is structurally invalid row state —
    /// `RegistryError::Invalid` on the catch-up read, never a poison
    /// row wedging the reconcile.
    #[tokio::test]
    async fn corrupt_record_rows_map_registry_invalid() {
        let path = unique_db_path("sqlite", "corrupt");
        let handle = CoTenantSqlite::open(&path).unwrap();
        let twin = SqliteRegistry::new(handle.clone());
        twin.put(&record("owner/ok")).await.unwrap();
        handle
            .lock()
            .execute(
                "INSERT INTO git_repos (repo_id, visibility, grants, seq) \
             VALUES ('owner/bent', 'public', 'not-json', 2)",
                [],
            )
            .unwrap();

        let err = twin.rows_since(0).await.unwrap_err();
        assert_eq!(err, RegistryError::Invalid, "{err:?}");

        let _ = std::fs::remove_file(&path);
    }

    /// The core-traits error map through the assembled binding — the
    /// closed registry family set exactly as the mem binding's.
    #[tokio::test]
    async fn the_assembled_binding_matches_the_family_error_map() {
        let path = unique_db_path("sqlite", "binding");
        let assembly = SqliteRegistryAssembly::open(&path).await.unwrap();
        let backend = assembly.backend().clone();

        assert_eq!(
            backend.update_repo(&record("ghost")).await.unwrap_err(),
            RegistryError::NotFound
        );
        assert_eq!(
            backend.remove_repo("ghost").await.unwrap_err(),
            RegistryError::NotFound
        );
        backend.put_repo(&record("owner/b")).await.unwrap();
        assert_eq!(
            backend.put_repo(&record("owner/b")).await.unwrap_err(),
            RegistryError::AlreadyExists
        );
        let malformed = record("Bad/id");
        assert_eq!(
            backend.put_repo(&malformed).await.unwrap_err(),
            RegistryError::Invalid
        );

        backend.remove_repo("owner/b").await.unwrap();
        assert_eq!(
            backend.resolve("owner/b").await.unwrap_err(),
            RegistryError::NotFound,
            "the unknown ≡ unauthorized collapse is at resolve's caller"
        );
    }
}