alkgit_substrate/

compose.rs

//! The single composition entry (wave 8) — the complete backend
//! assembly (substrate.md §public API: "construction wired by a single
//! composition entry that takes the substrate engine handles (the
//! assembly-level one-statement engine choice)"; substrate.md §assembly
//! shapes: "Engine choice is one assembly-level statement — three
//! shapes, zero assembly code").
//!
//! [`SubstrateAssembly`] is the wired whole: the five backend-trait
//! implementations as `Arc<dyn ...>` (the core's seam types —
//! `GitRegistry`, `GitRegistryStore`, `GitRefs`, `GitPackGen`,
//! `GitPackIngest`) plus the machinery's handles the substrate doc's
//! members list names (the object front, the reactor, the maintenance
//! chain, the applied watermark). One fallible constructor per
//! assembly, per the [`OpenError`] posture:
//!
//! - [`SubstrateAssembly::mem`] — the completeness shape. No engine
//!   dependency: the registry adapter, the refs table, and every
//!   wiring piece are the crate's own base types, and the constructor
//!   *takes the substrate engine handles* — the alkstore engine the
//!   reactive halves ride and the alkblobs object pool the git-object
//!   family lives under (`alkstore-mem` + `Store::mem_only()` in the
//!   tests' column; any `Store` pairing for an embedder — the reactive
//!   composition is engine-generic, POC-R's posture). Nothing opens,
//!   so the failure set is only the wiring legs (the liveness
//!   registration).
//! - [`SubstrateAssembly::sqlite`] — the default single-host assembly
//!   per substrate.md §assembly shapes row one: one sqlite file
//!   co-tenant by the registry/ref tables
//!   ([`crate::sqlite::SqliteRegistryAssembly`]'s constructor), the
//!   alkstore-sqlite engine, and the alkblobs dual-tier sqlite object
//!   store (`dual_tier_sqlite_local` — the large tier the through-store
//!   serving shape and over-threshold objects need; a sqlite fleet is a
//!   recorded non-option, alkblobs ADR-010 §3). `:memory:` never
//!   reaches an engine — the DQ-7 ruling holds by construction: every
//!   open rides the same file, so the ephemeral shape is the mem
//!   assembly, never sqlite-in-memory.
//! - the `postgres` constructor — the fleet/durable-shared
//!   assembly (the `postgres` feature; substrate.md §assembly shapes
//!   row two): the DSN every open shares — the co-tenant connection
//!   (`CoTenantPg`, the registry's own connection,
//!   the fourth pool of the assembly budget), the alkstore-postgres
//!   engine, and the alkblobs dual-tier postgres object store
//!   (`dual_tier_postgres_local` — pg kv + a local large tier; the
//!   consolidation pg-lo posture is a deployment choice the entry does
//!   not make for the embedder).
//!
//! Wave 6's per-piece constructors are unified behind the entry, not
//! duplicated: the engine-twin entries call the twin assemblies'
//! constructors and consume their exposed wiring; the complete-backend
//! wiring legs (the object front, the machinery custody share, the
//! pack serving shapes, the liveness registration, the maintenance
//! chain) run once in `wire` for all three shapes.
//!
//! # The entry's construction decisions (the task's spec-level call,
//! recorded for the OQ-03 inventory)
//!
//! - **What the engine handles are** (the opened-assembly-handle vs
//!   config question): the sqlite/pg entries open with *config* — the
//!   assembly-level engine statement is the constructor call itself
//!   (alkstore ADR-016's deployment honesty; co-tenancy on one file /
//!   one DSN is guaranteed by construction, never an embedder duty) —
//!   and the mem entry takes the opened handles (no engine dependency,
//!   nothing to open).
//! - **What the machinery returns**: handles, not run tasks — the
//!   reactor/chain runtime legs are *returned for the embedder to
//!   drive*. [`MaintenanceChain::run`](crate::maintenance::MaintenanceChain::run)
//!   owns the supervision (the sweep schedule + the scheduler leg, the
//!   supervised wake runner, the reconcile tick, the worker claim
//!   loop) under one caller-owned stop token. No task spawns at
//!   construction; the entry constructs and wires.
//! - **Both pack serving shapes at the entry** (the wave-7 posture,
//!   assembly-level): [`AssemblyOptions::pack_serving`] picks
//!   [`PackServing::Direct`] (the default — no storage round-trip) or
//!   [`PackServing::ThroughStore`] (put-once + ranged windows with
//!   mid-flight digest verification; the request-digest cache is
//!   shared with the maintenance chain's pack-housekeeping release
//!   path). The large tier the through-store shape requires is part of
//!   the engine-twin object stores by construction; a mem object pool
//!   without one rejects over-threshold packs through the store's own
//!   put-time posture.
//!
//! # The receive-pack flow, as wired
//!
//! The entry's refs leg is the w8 custody/commit side: the refs table
//! wrapped in the liveness write counter
//! (`VersionedRefTable`)) and then in
//! the composition-internal `CommittedRefs` — stale-prune → CAS →
//! drain-only-after-commit, then the per-push mutation-marker write +
//! post-commit announce (the `RegistryMutations` legs surface: the mem
//! shape's fused legs or the engine-twin assemblies' fused
//! `record_mutation`). The [`PackIngest`]
//! shares the pending-pin registry the wrapped refs impl drains —
//! constructed together, the custody seam's one share (substrate.md
//! §object storage, "Pin custody across the frozen trait seam"). The
//! wire layer reaches the backend only through the trait objects, so
//! this assembly is the one place the receive flow's legs exist. The
//! ref-table-closure liveness source is constructed over the front +
//! the wrapped table's `RefRows` face + the same write counter and
//! registered on the object pool — zero registered sources make every
//! sweep abort (`GcAbortCause::NoLivenessSources`). The LFS media
//! source (`lfs`, the wave-9 wiring) registers beside it — the second
//! leg over the wrapped LFS rows surface and its assert counter, the
//! fail-safe default (ADR-022 §6: the `lfs` feature gates the
//! registration itself, so the source rides by construction and a
//! sweep never eats referenced media on a mixed deployment).
//! The twins' record-transaction legs (the unborn-HEAD seed at record
//! creation, the repo's ref rows leaving with its removal — substrate.md
//! §refs) ride the composition on the mem shape too now that the entry
//! wires it a refs table (`MemSeededBackend`; the twins carry their
//! own in-SQL copies).
//!
//! # Names pinned here (the OQ-03 substrate-side inventory for this task)
//!
//! `SubstrateAssembly` (`mem`/`sqlite`/`postgres` + the accessors),
//! `AssemblyOptions`, `PackServing` — plus the wave-9 rows task's
//! accessor: `lfs_rows()` (the `lfs` feature's row-op surface, the
//! `LfsRows` trait object whatever the shape — the rows surface the
//! entry wraps in the assert counter) and the wave-9 liveness task's
//! `lfs_registration()` (the second registration leg's token). The
//! receive-pack composition pieces themselves are deliberately not in
//! the inventory (substrate.md §public API: internal to the trait
//! impls).

#[cfg(feature = "sqlite")]
use std::path::Path;
use std::path::PathBuf;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::Arc;

use alkstore::Store;
use async_trait::async_trait;

use crate::error::OpenError;
use crate::ingest::PackIngest;
#[cfg(feature = "lfs")]
use crate::lfs::LfsRows as _;
use crate::liveness::RefClosure;
use crate::maintenance::{MaintenanceChain, MaintenanceOpts};
use crate::odb::ObjectFront;
use crate::packgen::{PackCache, PackGen};
use crate::pins::{CommittedRefs, MemMutationLegs, PendingPinRegistry, RegistryMutations};
use crate::reactor::{AppliedWatermark, Reactor};
use crate::refs::InMemoryRefTable;
use crate::registry::{
    CachedRegistryProvider, InMemoryRegistry, RegistryBackend, RegistryWrite, HEAD_SEED_TARGET,
};
use alkgit::{RegistryError, RepoRecord};

use alkblobs::Store as BlobStore;

#[cfg(feature = "lfs")]
use crate::liveness::LfsMedia;

use alkgit::{GitPackGen, GitPackIngest, GitRefs, GitRegistry, GitRegistryStore};

/// The pack generation's serving shape at the entry (substrate.md
/// §object storage "Packs": "Two serving shapes, both validated").
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
pub enum PackServing {
    /// The generated pack streams through sideband, no storage
    /// round-trip. The default.
    #[default]
    Direct,
    /// The pack is put once, then served through ranged windows with
    /// mid-flight digest verification; the request-digest cache lets
    /// repeat fetches ride the stored pack. Requires the large tier
    /// (over-threshold packs stage into it — the engine-twin object
    /// stores carry it; a mem object pool without one rejects
    /// over-threshold puts through the store's own posture).
    ThroughStore,
}

/// The entry's assembly-level options — the knobs the entry's
/// constructors carry. Everything else is the shipped default posture
/// (the object stores' thresholds and window delays are the twin
/// constructors' own defaults; constructor knobs are additive later —
/// the two-way-door rule the HEAD seed pinned).
#[derive(Clone, Debug)]
pub struct AssemblyOptions {
    /// The pack generation's serving shape (the default is
    /// [`PackServing::Direct`]).
    pub pack_serving: PackServing,
    /// The receive-pack working area's root — the ingest implement's
    /// budgeted host scratch ([`PackIngest`]'s constructor knob; a
    /// fresh per-push directory under it, discarded after unpack).
    pub work_root: PathBuf,
    /// The object store's local large tier root — the over-threshold
    /// staging directory the sqlite/pg entries' object stores construct
    /// (alkblobs' dual-tier shapes need one). The mem entry's object
    /// pool rides the handle the embedder supplies instead.
    pub large_root: PathBuf,
    /// The maintenance chain's tuning datums (the sweep cadence, the
    /// reconcile tick, the worker poll, the wake-restart backoff, the
    /// leader-lock TTL). [`MaintenanceOpts::default()`] is the shipped
    /// posture.
    pub maintenance: MaintenanceOpts,
}

impl AssemblyOptions {
    /// The options with the given working-area and large-tier roots
    /// and the shipped default posture everywhere else.
    pub fn new(work_root: impl Into<PathBuf>, large_root: impl Into<PathBuf>) -> Self {
        AssemblyOptions {
            pack_serving: PackServing::default(),
            work_root: work_root.into(),
            large_root: large_root.into(),
            maintenance: MaintenanceOpts::default(),
        }
    }
}

/// The assembly-level pieces the per-assembly constructors wire before
/// the shared complete-backend wiring runs — the registry machinery
/// (the wave-6 assembly shapes, unified) plus the refs table and
/// mutation legs the entry's refs leg wraps.
struct RegistryLegs {
    /// The reactive halves' substrate (the maintenance chain rides the
    /// same handle).
    store: Arc<dyn Store>,
    /// The coherently cached hot read.
    provider: Arc<CachedRegistryProvider>,
    /// The registry sides, pre-coerced (the record mutations announce
    /// post-commit and write through; the mem binding carries the
    /// seeding legs).
    registry: Arc<dyn GitRegistry>,
    registry_store: Arc<dyn GitRegistryStore>,
    /// The announce/wake/reconcile machinery.
    reactor: Arc<Reactor>,
    /// The refs table under the write counter — the wrapped refs impl's
    /// inner and the liveness snapshot's rows.
    versioned: Arc<dyn GitRefs>,
    /// The same instance's `RefRows` face (the snapshot source).
    rows: Arc<dyn crate::liveness::RefRows>,
    /// The refs writes' counter (the liveness arbitration signal).
    refs_version: Arc<AtomicU64>,
    /// The per-push mutation legs (the fused marker + announce).
    mutations: Arc<dyn RegistryMutations>,
    /// The LFS pointer-lifecycle row store (`lfs`) under the assert
    /// counter — the wrapped surface (the batch adapter's assertion
    /// surface and the wave-9 liveness source's row read; the one
    /// handle both consume, the counter bump riding every asserted
    /// row).
    #[cfg(feature = "lfs")]
    lfs: Arc<dyn crate::lfs::LfsRows>,
    /// The LFS rows' counter (the LFS liveness source's arbitration
    /// signal — the wrapped surface bumps it on every `assert`; the
    /// source reads it in `contains`).
    #[cfg(feature = "lfs")]
    lfs_version: Arc<AtomicU64>,
}

/// The mem binding's record-transaction legs — the twins' seeding and
/// removal legs realized at the composition (substrate.md §refs: the
/// substrate's `put_repo` seeds the unborn HEAD in the record
/// transaction; repo deletion takes the repo's ref rows with it — and
/// the repo's LFS liveness rows, the wave-9 rows task's mem-shape
/// leg). The wave-6 degenerate case had no refs table, so the twins
/// carried the legs; the entry wires the mem shape a refs table, and
/// the legs ride here. Not in the OQ-03 inventory (composition-internal).
struct MemSeededBackend {
    inner: Arc<RegistryBackend>,
    rows: Arc<InMemoryRefTable>,
    writes: Arc<AtomicU64>,
    #[cfg(feature = "lfs")]
    lfs: Arc<crate::lfs::VersionedLfsRows<crate::lfs::InMemoryLfsRows>>,
}

impl RegistryLegs {
    /// The mem shape's legs: the in-memory registry + table + the
    /// fused marker legs + the record-transaction seeding legs (the
    /// base types; the provider write-through).
    fn mem(store: Arc<dyn Store>) -> Self {
        let durable = Arc::new(InMemoryRegistry::new());
        let provider = Arc::new(CachedRegistryProvider::new());
        let reactor = Arc::new(Reactor::new(
            Arc::clone(&store),
            durable.clone(),
            provider.clone(),
        ));
        let announced: Arc<dyn RegistryWrite> = Arc::new(crate::reactor::AnnouncingRegistry::new(
            durable.clone(),
            reactor.clone(),
        ));
        let refs_version = Arc::new(AtomicU64::new(0));
        let table = Arc::new(InMemoryRefTable::new());
        #[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::InMemoryLfsRows::new()),
            Arc::clone(&lfs_version),
        ));
        let inner = Arc::new(RegistryBackend::with_cached_provider(
            announced,
            provider.clone(),
        ));
        let seeded = Arc::new(MemSeededBackend {
            inner,
            rows: Arc::clone(&table),
            writes: Arc::clone(&refs_version),
            #[cfg(feature = "lfs")]
            lfs: Arc::clone(&lfs),
        });
        let versioned = Arc::new(crate::refs::VersionedRefTable::new(
            Arc::clone(&table),
            Arc::clone(&refs_version),
        ));
        let mutations = Arc::new(MemMutationLegs::new(
            Arc::clone(&durable),
            Arc::clone(&reactor),
        ));
        RegistryLegs {
            store,
            provider,
            registry: Arc::clone(&seeded) as Arc<dyn GitRegistry>,
            registry_store: seeded as Arc<dyn GitRegistryStore>,
            reactor,
            versioned: Arc::clone(&versioned) as Arc<dyn GitRefs>,
            rows: versioned,
            refs_version,
            mutations,
            #[cfg(feature = "lfs")]
            lfs: lfs as Arc<dyn crate::lfs::LfsRows>,
            #[cfg(feature = "lfs")]
            lfs_version,
        }
    }
}

#[async_trait]
impl GitRegistryStore for MemSeededBackend {
    async fn put_repo(&self, record: &RepoRecord) -> Result<(), RegistryError> {
        self.inner.put_repo(record).await?;
        self.rows.seed_head(&record.repo_id, HEAD_SEED_TARGET).await;
        self.writes.fetch_add(1, Ordering::Relaxed);
        Ok(())
    }

    async fn update_repo(&self, record: &RepoRecord) -> Result<(), RegistryError> {
        self.inner.update_repo(record).await
    }

    async fn remove_repo(&self, repo_id: &str) -> Result<(), RegistryError> {
        self.inner.remove_repo(repo_id).await?;
        self.rows.drop_repo_rows(repo_id).await;
        #[cfg(feature = "lfs")]
        self.lfs
            .remove_repo(repo_id)
            .await
            .map_err(|e| RegistryError::Io(e.to_string()))?;
        self.writes.fetch_add(1, Ordering::Relaxed);
        Ok(())
    }
}

#[async_trait]
impl GitRegistry for MemSeededBackend {
    async fn resolve(&self, repo_id: &str) -> Result<RepoRecord, RegistryError> {
        self.inner.resolve(repo_id).await
    }
}

/// The complete-backend wiring — the legs every shape's constructor
/// runs once (the wave-6 per-piece wiring, unified here). Fallible on
/// the liveness registration alone (the construction's one loud leg).
async fn wire(
    legs: RegistryLegs,
    blobs: Arc<BlobStore>,
    options: AssemblyOptions,
) -> Result<SubstrateAssembly, OpenError> {
    let front = ObjectFront::new(Arc::clone(&blobs));
    let pins = Arc::new(PendingPinRegistry::new());

    let (packgen, packs) = match options.pack_serving {
        PackServing::Direct => (PackGen::direct(front.clone()), None),
        PackServing::ThroughStore => {
            let cache = Arc::new(PackCache::new(crate::packgen::DEFAULT_CACHE_ENTRIES));
            (
                PackGen::through_store_with(front.clone(), Arc::clone(&cache)),
                Some(cache),
            )
        }
    };

    let committed = CommittedRefs::new(
        Arc::clone(&legs.versioned),
        Arc::clone(&pins),
        Arc::clone(&legs.mutations),
    );
    let refs: Arc<dyn GitRefs> = Arc::new(committed);

    let ingest: Arc<dyn GitPackIngest> = Arc::new(PackIngest::new(
        front.clone(),
        Arc::clone(&pins),
        options.work_root.clone(),
    ));

    let closure = RefClosure::new(front.clone(), legs.rows.clone(), legs.refs_version.clone());
    let liveness_registration = closure
        .register(&blobs)
        .map_err(|e| OpenError::Engine(format!("the liveness source registration: {e}")))?;

    #[cfg(feature = "lfs")]
    let lfs_registration = {
        let media = LfsMedia::new(legs.lfs.clone(), legs.lfs_version.clone());
        media
            .register(&blobs)
            .map_err(|e| OpenError::Engine(format!("the LFS liveness source registration: {e}")))?
    };

    let applied = AppliedWatermark::new();
    let chain = MaintenanceChain::new(
        Arc::clone(&legs.store),
        Arc::clone(&blobs),
        legs.reactor.clone(),
        applied.clone(),
        packs,
        options.maintenance.clone(),
    );

    Ok(SubstrateAssembly {
        registry: legs.registry,
        registry_store: legs.registry_store,
        refs,
        packgen: Arc::new(packgen),
        ingest,
        front,
        reactor: legs.reactor,
        chain,
        applied,
        liveness_registration,
        #[cfg(feature = "lfs")]
        lfs_registration,
        blobs,
        provider: legs.provider,
        refs_version: legs.refs_version,
        #[cfg(feature = "lfs")]
        lfs_rows: legs.lfs,
    })
}

/// The complete wired backend — the single composition entry's product
/// (substrate.md §public API's members): the five backend-trait
/// implementations as `Arc<dyn ...>` plus the machinery's handles.
///
/// The runtime legs are the embedder's to drive:
/// [`self.chain().run(worker_id, stop, on_event)`](crate::maintenance::MaintenanceChain::run)
/// runs the supervision (the sweep schedule, the scheduler leg, the
/// supervised wake runner — the provider invalidation's delivery, the
/// reconcile tick, the worker claim loop) under one caller-owned stop
/// token. Nothing spawns at construction.
///
/// The public API = authorized-surface rule: the op and session layers
/// reach the backend only through the core's traits (the returned
/// trait objects); the receive-pack composition — the pending-pin
/// registry and the wrapped refs impl — is internal to the trait
/// impls (substrate.md §public API's internal-members rule).
pub struct SubstrateAssembly {
    registry: Arc<dyn GitRegistry>,
    registry_store: Arc<dyn GitRegistryStore>,
    refs: Arc<dyn GitRefs>,
    packgen: Arc<dyn GitPackGen>,
    ingest: Arc<dyn GitPackIngest>,
    front: ObjectFront,
    reactor: Arc<Reactor>,
    chain: MaintenanceChain,
    applied: Arc<AppliedWatermark>,
    liveness_registration: alkblobs::Registration,
    #[cfg(feature = "lfs")]
    lfs_registration: alkblobs::Registration,
    blobs: Arc<BlobStore>,
    provider: Arc<CachedRegistryProvider>,
    refs_version: Arc<AtomicU64>,
    #[cfg(feature = "lfs")]
    lfs_rows: Arc<dyn crate::lfs::LfsRows>,
}

impl SubstrateAssembly {
    /// The mem assembly — the completeness shape (the batteries-
    /// included degenerate case: single-process, the in-memory
    /// registry adapter and refs table, no feature gates, ADR-024 §6).
    ///
    /// `store` is the alkstore engine the reactive halves and the
    /// maintenance chain ride; `blobs` is the object pool the
    /// git-object family lives under — the substrate engine handles,
    /// supplied (the mem entry opens nothing; "no engine dependency"
    /// realized honestly). See the module doc for the failure set.
    pub async fn mem(
        store: Arc<dyn Store>,
        blobs: Arc<BlobStore>,
        options: AssemblyOptions,
    ) -> Result<Self, OpenError> {
        wire(RegistryLegs::mem(store), blobs, options).await
    }

    /// The sqlite assembly — the default single-host shape
    /// (substrate.md §assembly shapes row one): one co-tenant file.
    ///
    /// `path` is the sqlite file the registry/ref tables, the
    /// alkstore-sqlite engine, and the alkblobs kv engine share (the
    /// co-tenancy composition posture; the migrations run at the twin
    /// assembly's construction, loud — a half-migrated store refuses
    /// service).
    #[cfg(feature = "sqlite")]
    pub async fn sqlite(
        path: impl AsRef<Path>,
        options: AssemblyOptions,
    ) -> Result<Self, OpenError> {
        let path = path.as_ref();
        let assembly = crate::sqlite::SqliteRegistryAssembly::open(path).await?;
        let blobs = Arc::new(
            BlobStore::dual_tier_sqlite_local(
                path,
                options.large_root.clone(),
                alkblobs::DEFAULT_DISPATCH_THRESHOLD,
                alkblobs::DEFAULT_DELETE_WINDOW_DELAY,
            )
            .await
            .map_err(|e| OpenError::Engine(format!("alkblobs sqlite object store: {e}")))?,
        );

        let versioned: Arc<dyn GitRefs> = assembly.refs().clone();
        let rows: Arc<dyn crate::liveness::RefRows> = assembly.refs().clone();
        #[cfg(feature = "lfs")]
        let lfs_version = assembly.lfs_version().clone();
        #[cfg(feature = "lfs")]
        let lfs_rows: Arc<dyn crate::lfs::LfsRows> = assembly.lfs_rows().clone();
        let legs = RegistryLegs {
            store: assembly.store().clone(),
            provider: assembly.provider().clone(),
            registry: assembly.backend().clone() as Arc<dyn GitRegistry>,
            registry_store: assembly.backend().clone() as Arc<dyn GitRegistryStore>,
            reactor: assembly.reactor().clone(),
            versioned,
            rows,
            refs_version: assembly.refs_version().clone(),
            mutations: Arc::new(assembly),
            #[cfg(feature = "lfs")]
            lfs: lfs_rows.clone(),
            #[cfg(feature = "lfs")]
            lfs_version,
        };
        wire(legs, blobs, options).await
    }

    /// The postgres assembly — the fleet/durable-shared shape (the
    /// `postgres` feature; substrate.md §assembly shapes row two).
    ///
    /// `config` is the DSN every open shares. `registry` carries the
    /// co-tenant schema + the alkstore-postgres engine's options
    /// ([`crate::postgres::PgRegistryOptions`]); the alkblobs kv
    /// engine rides its own shipped-default `PgOpts` (knobs additive
    /// later, the pinned posture).
    #[cfg(feature = "postgres")]
    pub async fn postgres(
        config: &str,
        registry: crate::postgres::PgRegistryOptions,
        options: AssemblyOptions,
    ) -> Result<Self, OpenError> {
        let assembly = crate::postgres::PgRegistryAssembly::open(config, registry.clone()).await?;
        let blobs = Arc::new(
            BlobStore::dual_tier_postgres_local(
                config,
                registry.blobs,
                options.large_root.clone(),
                alkblobs::DEFAULT_DISPATCH_THRESHOLD,
                alkblobs::DEFAULT_DELETE_WINDOW_DELAY,
            )
            .await
            .map_err(|e| OpenError::Engine(format!("alkblobs postgres object store: {e}")))?,
        );

        let versioned: Arc<dyn GitRefs> = assembly.refs().clone();
        let rows: Arc<dyn crate::liveness::RefRows> = assembly.refs().clone();
        #[cfg(feature = "lfs")]
        let lfs_version = assembly.lfs_version().clone();
        #[cfg(feature = "lfs")]
        let lfs_rows: Arc<dyn crate::lfs::LfsRows> = assembly.lfs_rows().clone();
        let legs = RegistryLegs {
            store: assembly.store().clone(),
            provider: assembly.provider().clone(),
            registry: assembly.backend().clone() as Arc<dyn GitRegistry>,
            registry_store: assembly.backend().clone() as Arc<dyn GitRegistryStore>,
            reactor: assembly.reactor().clone(),
            versioned,
            rows,
            refs_version: assembly.refs_version().clone(),
            mutations: Arc::new(assembly),
            #[cfg(feature = "lfs")]
            lfs: lfs_rows.clone(),
            #[cfg(feature = "lfs")]
            lfs_version,
        };
        wire(legs, blobs, options).await
    }
}

// The assembly's accessors (the OQ-03 inventory's member functions).

impl SubstrateAssembly {
    /// The registry read side ([`GitRegistry`] — the resolve hot
    /// path, served through the cached provider).
    pub fn registry(&self) -> &Arc<dyn GitRegistry> {
        &self.registry
    }

    /// The registry store side ([`GitRegistryStore`] — the record
    /// mutations, announcing post-commit and writing through).
    pub fn registry_store(&self) -> &Arc<dyn GitRegistryStore> {
        &self.registry_store
    }

    /// The refs side ([`GitRefs`]) — the receive flow's committed-ref
    /// side: stale-prune → CAS → drain-only-after-commit, then the
    /// per-push marker + announce legs.
    pub fn refs(&self) -> &Arc<dyn GitRefs> {
        &self.refs
    }

    /// The pack generation side ([`GitPackGen`], assembled in the
    /// configured serving shape).
    pub fn packgen(&self) -> &Arc<dyn GitPackGen> {
        &self.packgen
    }

    /// The pack ingest side ([`GitPackIngest`] — the receive path's
    /// prepare leg, custody via the shared pending-pin registry).
    pub fn ingest(&self) -> &Arc<dyn GitPackIngest> {
        &self.ingest
    }

    /// The object front (constructed once; per-session
    /// [`SessionFind`](crate::odb::ObjectFront::session) views).
    pub fn front(&self) -> &ObjectFront {
        &self.front
    }

    /// The reactor (the announce machinery; the stream consumer and
    /// the reconcile sweep's home for replication-style embedders).
    pub fn reactor(&self) -> &Arc<Reactor> {
        &self.reactor
    }

    /// The maintenance chain (the embedder drives
    /// [`MaintenanceChain::run`]).
    pub fn chain(&self) -> &MaintenanceChain {
        &self.chain
    }

    /// The composition's applied watermark — the same handle the
    /// reconcile tick advances; hand it to
    /// [`Reactor::run_stream_consumer`] alongside.
    pub fn applied_watermark(&self) -> &Arc<AppliedWatermark> {
        &self.applied
    }

    /// The cached read provider (the hot read's home — replication
    /// and observation consumers).
    pub fn provider(&self) -> &Arc<CachedRegistryProvider> {
        &self.provider
    }

    /// The object pool (the git-object family's home; the LFS
    /// surface's base).
    pub fn blobs(&self) -> &Arc<BlobStore> {
        &self.blobs
    }

    /// The liveness source's registration token (the composition holds
    /// it for the store's lifetime; deregister/re-register is the
    /// embedder's ADR-012 §6.4 correction flow).
    pub fn liveness_registration(&self) -> &alkblobs::Registration {
        &self.liveness_registration
    }

    /// The LFS media source's registration token (`lfs` — the second
    /// registration leg, registered beside the ref-table closure's at
    /// the composition entry; the fail-safe default ADR-022 §6). The
    /// same lifetime posture: the composition holds it, the
    /// deregister/re-register correction flow is the embedder's.
    #[cfg(feature = "lfs")]
    pub fn lfs_registration(&self) -> &alkblobs::Registration {
        &self.lfs_registration
    }

    /// The LFS pointer-lifecycle row store (the `lfs` feature's
    /// row-op surface): the batch adapter's assertion surface (rows
    /// asserted at check/upload time — lfs.md §3) and the LFS
    /// liveness source's read (all rows / per-repo rows — the
    /// re-protection contract's refcount data; the wave-9 liveness
    /// task consumes this accessor). The surface is the wrapped shape
    /// on every source ([`crate::lfs::VersionedLfsRows`] — every
    /// `assert` bumps [`Self::lfs_registration`]'s counter): on the
    /// engine assemblies the wrap is the twins' own (they carry it at
    /// construction; the entry composes it unchanged, no double wrap).
    #[cfg(feature = "lfs")]
    pub fn lfs_rows(&self) -> &Arc<dyn crate::lfs::LfsRows> {
        &self.lfs_rows
    }
}

/// The refs writes' shared counter — bumped by every CAS call through
/// the refs side; the liveness source's arbitration epoch signal
/// (observation data).
impl SubstrateAssembly {
    /// The counter's current value.
    pub fn refs_version(&self) -> u64 {
        self.refs_version.load(Ordering::Relaxed)
    }
}
#[cfg(test)]
mod tests {
    use std::collections::BTreeMap;
    use std::sync::atomic::AtomicU64;
    use std::sync::Arc;
    use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};

    use alkcall::core::Identity;
    use alkstore::Store as AlkStore;
    use gix_hash::{hasher, Kind as HashKind, ObjectId};
    use gix_object::Kind;
    use gix_pack::data::entry::Header as PackHeader;
    use gix_pack::data::input::Entry as PackEntry;
    use tokio::io::{duplex, AsyncReadExt, AsyncWriteExt};
    use tokio::sync::Semaphore;

    use alkgit::{
        serve_duplex_resolved, GitPackGen, GitPackIngest, GitRefs, GitRegistry, Limits,
        RegistryError, Service, SubstrateError, Visibility,
    };
    use serde_json::Value;

    use alkgit::RepoRecord;

    use super::*;
    use crate::loose_form;
    use crate::registry::RegistryRead as _;

    fn unique_root(tag: &str) -> std::path::PathBuf {
        static COUNTER: AtomicU64 = AtomicU64::new(0);
        let n = COUNTER.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
        let nanos = SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .map(|d| d.as_nanos())
            .unwrap_or_default();
        std::env::temp_dir().join(format!(
            "alkgit-substrate-compose-{tag}-{}-{n}-{nanos}",
            std::process::id()
        ))
    }

    fn identity() -> Identity {
        Identity {
            id: "u-writer".to_owned(),
            scopes: Vec::new(),
            resources: Default::default(),
        }
    }

    fn granted_record(id: &str, identity: &Identity) -> RepoRecord {
        RepoRecord {
            repo_id: id.to_owned(),
            visibility: Visibility::Private,
            grants: BTreeMap::from([(
                identity.id.clone(),
                vec![alkgit::AccessAction::Read, alkgit::AccessAction::Write],
            )]),
        }
    }

    fn ungranted_record(id: &str) -> RepoRecord {
        RepoRecord {
            repo_id: id.to_owned(),
            visibility: Visibility::Private,
            grants: BTreeMap::new(),
        }
    }

    fn mem_object_pool() -> Arc<alkblobs::Store> {
        Arc::new(alkblobs::Store::mem_only())
    }

    // ——— the client-side session helpers (the core tests' shapes) ———

    fn pkt(payload: &[u8]) -> Vec<u8> {
        let total = payload.len() + 4;
        let mut out = Vec::with_capacity(total);
        out.extend_from_slice(format!("{total:04x}").as_bytes());
        out.extend_from_slice(payload);
        out
    }

    async fn read_pkts<R: tokio::io::AsyncRead + Unpin>(reader: &mut R) -> Vec<Vec<u8>> {
        let mut packets = Vec::new();
        loop {
            let mut header = [0u8; 4];
            reader.read_exact(&mut header).await.expect("pkt header");
            let len = usize::from_str_radix(std::str::from_utf8(&header).expect("hex header"), 16)
                .expect("hex length");
            if len < 4 {
                packets.push(Vec::new());
                break;
            }
            let mut payload = vec![0u8; len - 4];
            reader.read_exact(&mut payload).await.expect("pkt payload");
            packets.push(payload);
        }
        packets
    }

    fn text(payload: &[u8]) -> String {
        String::from_utf8_lossy(payload)
            .trim_end_matches('\n')
            .to_owned()
    }

    fn create_command(name: &str, new: ObjectId) -> String {
        format!("{} {new} {name}", "0".repeat(40))
    }

    /// Reads the response body until the outer flush, collecting the
    /// band-1 payloads (sideband framing).
    async fn read_band1_body<R: tokio::io::AsyncRead + Unpin>(reader: &mut R) -> Vec<u8> {
        let mut payload = Vec::new();
        loop {
            let mut header = [0u8; 4];
            reader.read_exact(&mut header).await.expect("pkt header");
            let len = usize::from_str_radix(std::str::from_utf8(&header).expect("hex header"), 16)
                .expect("hex length");
            if len < 4 {
                break;
            }
            let mut packet = vec![0u8; len - 4];
            reader.read_exact(&mut packet).await.expect("pkt payload");
            assert_eq!(packet[0], 1, "band-1 framing");
            payload.extend_from_slice(&packet[1..]);
        }
        payload
    }

    fn report_lines(body: &[u8]) -> Vec<String> {
        let inner = {
            let mut lines = Vec::new();
            let mut cursor = body;
            while !cursor.is_empty() {
                let len = usize::from_str_radix(
                    std::str::from_utf8(&cursor[..4]).expect("hex header"),
                    16,
                )
                .expect("hex length");
                cursor = &cursor[4..];
                if len < 4 {
                    lines.push(Vec::new());
                    continue;
                }
                lines.push(cursor[..len - 4].to_vec());
                cursor = &cursor[len - 4..];
            }
            let last = lines.pop().expect("inner framing non-empty");
            assert!(last.is_empty(), "the inner flush ends the report");
            lines
        };
        inner.iter().map(|p| text(p)).collect()
    }

    // ——— the fixture chain + its pack (the ingest tests' shapes) ———

    fn tree_bytes(entries: &[(&str, &ObjectId)]) -> Vec<u8> {
        let mut tree = Vec::new();
        for (name, oid) in entries {
            tree.extend_from_slice(format!("100644 {name}\0").as_bytes());
            tree.extend_from_slice(oid.as_bytes());
        }
        tree
    }

    fn commit_bytes(tree: &ObjectId, parents: &[&ObjectId]) -> Vec<u8> {
        let mut body = format!("tree {tree}\n");
        for parent in parents {
            body.push_str(&format!("parent {parent}\n"));
        }
        body.push_str(
            "author A <a@example.invalid> 1750000000 +0000\n\
             committer A <a@example.invalid> 1750000000 +0000\n\nthe message\n",
        );
        body.into_bytes()
    }

    fn oid_of(kind: Kind, content: &[u8]) -> ObjectId {
        let form = loose_form(kind, content);
        let mut h = hasher(HashKind::Sha1);
        h.update(&form);
        h.try_finalize().expect("fixture oid hashes")
    }

    fn deflate(bytes: &[u8]) -> Vec<u8> {
        let mut deflate =
            gix_zlib::stream::deflate::Write::new(Vec::new(), gix_zlib::Compression::DEFAULT);
        std::io::Write::write_all(&mut deflate, bytes).expect("fixture bytes deflate");
        std::io::Write::flush(&mut deflate).expect("the fixture deflate stream finishes");
        deflate.into_inner()
    }

    fn pack_entry(kind: Kind, body: &[u8], pack_offset: u64) -> PackEntry {
        let header = match kind {
            Kind::Blob => PackHeader::Blob,
            Kind::Tree => PackHeader::Tree,
            Kind::Commit => PackHeader::Commit,
            Kind::Tag => PackHeader::Tag,
        };
        let compressed = deflate(body);
        PackEntry {
            header,
            header_size: header.size(body.len() as u64) as u16,
            pack_offset,
            compressed: Some(compressed.clone()),
            compressed_size: compressed.len() as u64,
            crc32: None,
            decompressed_size: body.len() as u64,
            trailer: None,
        }
    }

    fn pack_bytes(entries: &[PackEntry]) -> Vec<u8> {
        let mut out = Vec::new();
        out.extend_from_slice(&gix_pack::data::header::encode(
            gix_pack::data::Version::V2,
            entries.len() as u32,
        ));
        for entry in entries {
            entry
                .header
                .write_to(entry.decompressed_size, &mut out)
                .expect("fixture entry headers write");
            out.extend_from_slice(
                entry
                    .compressed
                    .as_ref()
                    .expect("fixture entries keep bytes"),
            );
        }
        let mut h = hasher(HashKind::Sha1);
        h.update(&out);
        out.extend_from_slice(
            h.try_finalize()
                .expect("fixture pack trailer hashes")
                .as_slice(),
        );
        out
    }

    /// A two-commit chain fixture: (pack, tip, base, tree_tip, blob).
    fn chain_pack() -> (Vec<u8>, ObjectId, ObjectId, ObjectId, ObjectId) {
        let blob_body = b"the composed chain's blob\n".to_vec();
        let blob = oid_of(Kind::Blob, &blob_body);
        let tree_base_body = tree_bytes(&[("a.txt", &blob)]);
        let tree_base = oid_of(Kind::Tree, &tree_base_body);
        let base_body = commit_bytes(&tree_base, &[]);
        let base = oid_of(Kind::Commit, &base_body);
        let tree_tip_body = tree_bytes(&[("a.txt", &blob), ("b.txt", &blob)]);
        let tree_tip = oid_of(Kind::Tree, &tree_tip_body);
        let tip_body = commit_bytes(&tree_tip, &[&base]);
        let tip = oid_of(Kind::Commit, &tip_body);

        let mut offset = 0u64;
        let mut entries = Vec::new();
        for (kind, body) in [
            (Kind::Blob, blob_body.as_slice()),
            (Kind::Tree, tree_base_body.as_slice()),
            (Kind::Commit, base_body.as_slice()),
            (Kind::Tree, tree_tip_body.as_slice()),
            (Kind::Commit, tip_body.as_slice()),
        ] {
            let entry = pack_entry(kind, body, offset);
            offset += entry.header_size as u64 + entry.compressed_size;
            entries.push(entry);
        }
        (pack_bytes(&entries), tip, base, tree_tip, blob)
    }

    // ——— the in-process serving entry ———

    #[allow(clippy::too_many_arguments)]
    async fn run_push(
        repo_id: &str,
        identity: &Identity,
        registry: Arc<dyn GitRegistry>,
        refs: Arc<dyn GitRefs>,
        packgen: Arc<dyn GitPackGen>,
        ingest: Arc<dyn GitPackIngest>,
        commands: &[String],
        pack: &[u8],
    ) -> Vec<String> {
        let limits = Limits::default();
        let (mut peer, stream) = duplex(1 << 16);
        let joined = tokio::spawn({
            let registry = registry.clone();
            let refs = refs.clone();
            let packgen = packgen.clone();
            let ingest = ingest.clone();
            let identity = identity.clone();
            let repo_id = repo_id.to_owned();
            let pipeline = Arc::new(Semaphore::new(limits.max_concurrent_blocking_tasks));
            async move {
                serve_duplex_resolved(
                    &repo_id,
                    Some(&identity),
                    Service::ReceivePack,
                    stream,
                    limits,
                    registry,
                    refs,
                    packgen,
                    ingest,
                    false,
                    pipeline,
                )
                .await
            }
        });

        read_pkts(&mut peer).await;
        let mut chunks = Vec::new();
        for (i, command) in commands.iter().enumerate() {
            if i == 0 {
                chunks.push(pkt(
                    format!("{command}\0report-status side-band-64k").as_bytes()
                ));
            } else {
                chunks.push(pkt(command.as_bytes()));
            }
        }
        chunks.push(b"0000".to_vec());
        chunks.push(pack.to_vec());
        peer.write_all(&chunks.concat()).await.unwrap();

        let body = read_band1_body(&mut peer).await;
        drop(peer);
        joined
            .await
            .expect("the session task joins")
            .expect("the push session serves cleanly");
        report_lines(&body)
    }

    async fn feed_events(
        assembly: &SubstrateAssembly,
        since: i64,
    ) -> Vec<crate::reactor::RegistryEvent> {
        let stream = assembly
            .reactor()
            .store()
            .stream(crate::reactor::EVENTS_STREAM)
            .await
            .expect("the feed opens");
        stream
            .read_since(since, 100)
            .await
            .expect("the feed reads")
            .iter()
            .filter_map(crate::reactor::RegistryEvent::decode_event)
            .collect()
    }

    // ——— the mem shape ———

    async fn mem_assembly(tag: &str) -> SubstrateAssembly {
        let root = unique_root(tag);
        let options = AssemblyOptions::new(root.join("work"), root.join("large"));
        SubstrateAssembly::mem(
            Arc::new(alkstore_mem::MemStore::new()),
            mem_object_pool(),
            options,
        )
        .await
        .expect("the mem assembly composes")
    }

    /// The composition's completeness shape: the five trait impls and
    /// the machinery handles are wired; the registry pair resolves
    /// through the cached provider (write-through); the unborn HEAD
    /// line lists through the refs side.
    #[tokio::test(flavor = "multi_thread")]
    async fn the_mem_assembly_wires_the_complete_backend() {
        let assembly = mem_assembly("wired").await;
        let identity = identity();
        let record = granted_record("owner/composed", &identity);

        assembly.registry_store().put_repo(&record).await.unwrap();
        assert_eq!(
            assembly.registry().resolve("owner/composed").await.unwrap(),
            record,
            "the provider was written through by the store"
        );
        assert!(assembly.provider().resolve("owner/composed").is_some());
        assert_eq!(
            assembly.refs_version(),
            1,
            "the record-create seeding leg bumped the counter"
        );

        let lines = assembly
            .refs()
            .list_refs(&record)
            .await
            .expect("the refs side lists");
        assert_eq!(
            lines,
            vec![alkgit::RefLine {
                name: "HEAD".to_owned(),
                oid: None,
                peeled: None,
                symref_target: Some(crate::registry::HEAD_SEED_TARGET.to_owned()),
            }],
            "the unborn HEAD shape rides the composed refs — the record's seed row"
        );

        let unknown = assembly.registry().resolve("owner/ghost").await;
        assert_eq!(unknown.unwrap_err(), RegistryError::NotFound);
    }

    /// The reactive halves re-landed through the entry: the announced
    /// record write fires the post-commit legs — the durable feed
    /// event, the wake, and the maintenance-queue job.
    #[tokio::test(flavor = "multi_thread")]
    async fn the_mem_assemblys_reactive_legs_fire_post_commit() {
        let assembly = mem_assembly("legs").await;
        let identity = identity();
        let mut listener = assembly
            .reactor()
            .store()
            .listen(crate::reactor::WAKE_CHANNEL)
            .await
            .expect("the wake source opens");
        let record = granted_record("owner/announced", &identity);

        assembly.registry_store().put_repo(&record).await.unwrap();

        let events = feed_events(&assembly, 0).await;
        assert_eq!(
            events,
            vec![crate::reactor::RegistryEvent::Put {
                seq: 1,
                record: record.clone()
            }],
            "the announced put published post-commit"
        );
        let wake = listener
            .recv_timeout(Duration::from_secs(2))
            .await
            .expect("the wake delivers")
            .expect("wake source open");
        assert_eq!(wake.channel, crate::reactor::WAKE_CHANNEL);
        drop(listener);

        let queue = assembly
            .reactor()
            .store()
            .queue(
                crate::reactor::MAINTENANCE_QUEUE,
                alkstore::QueueOpts::default(),
            )
            .await
            .expect("the queue opens");
        let claimed = queue
            .claim_one("probe")
            .await
            .expect("claim checks")
            .expect("the enqueue leg landed the housekeeping job");
        assert_eq!(
            claimed
                .job()
                .payload_as::<crate::reactor::MaintenanceJob>()
                .unwrap(),
            crate::reactor::MaintenanceJob::PackHousekeeping {
                repo_id: "owner/announced".to_owned(),
            }
        );
    }

    /// The receive path composes through the entry end-to-end
    /// in-process: resolve → authorize (the core's types, the granted
    /// identity rides) → the serving machinery → the refs commit side
    /// (custody + marker + announce) → ingest. The custody observable:
    /// the pushed chain is liveness-covered post-commit (a sweep
    /// deletes nothing); the legs observable: the `Mutation` marker
    /// event anchors the watermark.
    #[tokio::test(flavor = "multi_thread")]
    async fn the_receive_path_composes_through_the_entry_end_to_end() {
        let assembly = mem_assembly("receipt").await;
        let identity = identity();
        let record = granted_record("owner/pushed", &identity);
        assembly.registry_store().put_repo(&record).await.unwrap();

        let (pack, tip, _base, _tree, _blob) = chain_pack();
        let report = run_push(
            "owner/pushed",
            &identity,
            assembly.registry().clone(),
            assembly.refs().clone(),
            assembly.packgen().clone(),
            assembly.ingest().clone(),
            &[create_command("refs/heads/main", tip)],
            &pack,
        )
        .await;
        assert_eq!(
            report,
            vec!["unpack ok".to_owned(), "ok refs/heads/main".to_owned(),],
            "the band-1 report equals the capture's shape"
        );

        let lines = assembly.refs().list_refs(&record).await.unwrap();
        assert_eq!(
            lines
                .iter()
                .find(|line| line.name == "refs/heads/main")
                .map(|line| line.oid),
            Some(Some(tip)),
            "the committed ref landed through the entry's refs side"
        );
        assert!(assembly.refs_version() >= 1, "the CAS bumped the counter");

        let events = feed_events(&assembly, 1).await;
        assert_eq!(
            events,
            vec![crate::reactor::RegistryEvent::Mutation {
                seq: 2,
                repo_id: "owner/pushed".to_owned(),
            }],
            "the marker write announced post-CAS — the watermark leg"
        );
        let watermark = assembly.reactor().registry().watermark().await.unwrap();
        assert_eq!(watermark, 2, "the put (1) + the push's marker (2)");

        let report = assembly.blobs().sweep().await.expect("the sweep runs");
        assert!(
            report.aborted.is_none() && report.deleted == 0,
            "the pushed chain is liveness-covered — the registered closure \
             and the committed custody hold it across the sweep"
        );
    }

    /// The unauthorized shape: the ungranted identity reads the generic
    /// denial — resolve succeeds (the record exists) and the
    /// authorization collapses with the refusal (the unknown ≡
    /// unauthorized posture rides the entry's impls untouched).
    #[tokio::test(flavor = "multi_thread")]
    async fn the_receive_path_enforces_the_visible_surface() {
        let assembly = mem_assembly("gate").await;
        let identity = identity();
        let record = ungranted_record("owner/gated");
        assembly.registry_store().put_repo(&record).await.unwrap();

        let (_pack, tip, _a, _b, _c) = chain_pack();
        let limits = Limits::default();
        let (peer, stream) = duplex(1 << 16);
        let joined = tokio::spawn({
            let registry = assembly.registry().clone();
            let refs = assembly.refs().clone();
            let packgen = assembly.packgen().clone();
            let ingest = assembly.ingest().clone();
            let identity = identity.clone();
            let pipeline = Arc::new(Semaphore::new(limits.max_concurrent_blocking_tasks));
            async move {
                serve_duplex_resolved(
                    "owner/gated",
                    Some(&identity),
                    Service::ReceivePack,
                    stream,
                    limits,
                    registry,
                    refs,
                    packgen,
                    ingest,
                    false,
                    pipeline,
                )
                .await
            }
        });

        // The pre-protocol gate denies before any protocol byte: the
        // client reads the one `ERR <message>` pkt-line and the close
        // (no advertisement, no existence oracle — writing first does
        // not change the collapse). The push bytes ride their own task
        // against the write half because the server's pinned posture is
        // deny-then-close without ever reading: it may complete the
        // whole session before the client's write drains, and that
        // write's `BrokenPipe` is the honest outcome of the losing
        // interleave (pinned alongside the drain — never retried,
        // never tolerating a third shape).
        let chunks = [
            pkt(format!("{} {tip} refs/heads/main\0report-status", "0".repeat(40)).as_bytes()),
            b"0000".to_vec(),
        ];
        let (mut peer_read, mut peer_write) = tokio::io::split(peer);
        let write_outcome =
            tokio::spawn(async move { peer_write.write_all(&chunks.concat()).await })
                .await
                .expect("the client write task joins");
        assert!(
            write_outcome.is_ok()
                || matches!(
                    &write_outcome,
                    Err(e) if e.kind() == std::io::ErrorKind::BrokenPipe
                ),
            "the push bytes drain against a live session, or the \
             deny-then-close beat them to the close — never another \
             shape: {write_outcome:?}"
        );

        let denial = {
            let mut buf = Vec::new();
            peer_read
                .read_to_end(&mut buf)
                .await
                .expect("the denial body reads to the close");
            buf
        };
        assert_eq!(
            &denial[4..],
            format!("ERR access denied: {}", "owner/gated").as_bytes(),
            "the denial is the one wire shape — its message the \
             client-requested id and nothing else"
        );

        drop(peer_read);
        let result = joined
            .await
            .expect("the session task joins")
            .expect_err("the denial ends the session denied");
        assert!(
            matches!(result, SubstrateError::AccessDenied),
            "the collapse is the one generic denial: {result:?}"
        );
        assert!(
            assembly.refs().list_refs(&record).await.unwrap().len() <= 1,
            "only the seeded HEAD row exists — no ungranted write landed"
        );
    }

    /// The maintenance chain rides the entry's wiring: the worker
    /// claims and dispatches the announced mutation's housekeeping job
    /// (the enqueue leg's consumer side, observable).
    #[tokio::test(flavor = "multi_thread")]
    async fn the_chain_rides_the_entry_and_drains_the_housekeeping_job() {
        let root = unique_root("chain");
        let options = AssemblyOptions {
            maintenance: crate::maintenance::MaintenanceOpts {
                worker_poll: Duration::from_millis(20),
                reconcile_tick: Duration::from_secs(60),
                sweep_spec: "@every 1h".to_owned(),
                ..Default::default()
            },
            ..AssemblyOptions::new(root.join("work"), root.join("large"))
        };
        let assembly = SubstrateAssembly::mem(
            Arc::new(alkstore_mem::MemStore::new()),
            mem_object_pool(),
            options,
        )
        .await
        .expect("the mem assembly composes");

        let identity = identity();
        let record = granted_record("owner/hk", &identity);
        assembly.registry_store().put_repo(&record).await.unwrap();

        let seen = Arc::new(parking_lot::Mutex::new(Vec::new()));
        let stop = alkstore::StopToken::new();
        let worker = {
            let seen = seen.clone();
            tokio::spawn({
                let stop = stop.clone();
                let seen_probe = seen.clone();
                async move {
                    let result = assembly
                        .chain()
                        .run("entry-worker", stop, move |event| {
                            seen_probe.lock().push(event);
                        })
                        .await;
                    result.expect("the chain ran clean");
                }
            })
        };

        let deadline = Instant::now() + Duration::from_secs(5);
        while Instant::now() < deadline {
            if seen
                .lock()
                .iter()
                .any(|event| matches!(event, crate::maintenance::WorkerEvent::PackHousekeeping(_)))
            {
                break;
            }
            tokio::time::sleep(Duration::from_millis(20)).await;
        }
        stop.cancel();

        tokio::time::timeout(Duration::from_secs(5), worker)
            .await
            .expect("the chain wound down inside the deadline")
            .expect("the chain task joins clean");

        assert!(
            seen.lock()
                .iter()
                .any(|event| matches!(event, crate::maintenance::WorkerEvent::PackHousekeeping(_))),
            "the chain drained the announced mutation's housekeeping job"
        );
    }

    /// A store double whose scheduler leg dies instantly — the O-b
    /// disposition's probe (the chain's legs wind down via the wired
    /// internal token; the run returns the substrate failure
    /// promptly).
    struct SchedulerDeadStore {
        inner: Box<dyn AlkStore>,
    }

    impl AlkStore for SchedulerDeadStore {
        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 alkstore::WakeReceiver>>> {
            self.inner.listen(channel)
        }

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

        fn queue<'a>(
            &'a self,
            name: &str,
            opts: alkstore::QueueOpts,
        ) -> alkstore::BoxedFuture<'a, alkstore::Result<Box<dyn alkstore::Queue>>> {
            self.inner.queue(name, opts)
        }

        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: alkstore::StopToken,
        ) -> alkstore::BoxedFuture<'a, alkstore::Result<()>> {
            Box::pin(async { Err(alkstore::Error::database("the scheduler leg died (probe)")) })
        }
    }

    /// The O-b disposition: the scheduler leg's substrate failure ends
    /// the chain — the run returns the failure promptly (the wake leg
    /// wound down by the wired internal token, not by the join's abort
    /// window) and the failure is the chain's return.
    #[tokio::test(flavor = "multi_thread")]
    async fn a_scheduler_leg_failure_ends_the_chain_with_the_legs_wound_down() {
        let root = unique_root("ob");
        let options = AssemblyOptions::new(root.join("work"), root.join("large"));
        let assembly = SubstrateAssembly::mem(
            Arc::new(SchedulerDeadStore {
                inner: Box::new(alkstore_mem::MemStore::new()),
            }) as Arc<dyn AlkStore>,
            mem_object_pool(),
            options,
        )
        .await
        .expect("the mem assembly composes");

        let stop = alkstore::StopToken::new();
        let run =
            tokio::spawn(async move { assembly.chain().run("ob-worker", stop, |_| {}).await });
        let outcome = tokio::time::timeout(Duration::from_millis(750), run)
            .await
            .expect(
                "the chain returned inside the bound — the legs were wound \
                     down by the wired token, not outwaited into the join's \
                     abort window",
            )
            .expect("the chain task joins");
        let _error = outcome.expect_err("the scheduler leg's failure is the chain's return");
    }

    /// The through-store serving shape at the entry: the configured
    /// `PackGen` puts the generated pack once and serves through the
    /// store's ranged windows — the generation through the entry packs
    /// exactly the pushed chain's closure (the serving half over the
    /// same wiring the push landed its objects on).
    #[tokio::test(flavor = "multi_thread")]
    async fn the_through_store_serving_shape_composes_at_the_entry() {
        let root = unique_root("through-store");
        let options = AssemblyOptions {
            pack_serving: PackServing::ThroughStore,
            ..AssemblyOptions::new(root.join("work"), root.join("large"))
        };
        let blobs = Arc::new(
            alkblobs::Store::dual_tier_mem_local(
                root.join("large"),
                alkblobs::DEFAULT_DISPATCH_THRESHOLD,
            )
            .await
            .expect("the dual-tier mem object pool constructs"),
        );
        let assembly =
            SubstrateAssembly::mem(Arc::new(alkstore_mem::MemStore::new()), blobs, options)
                .await
                .expect("the mem assembly composes");

        let identity = identity();
        let record = granted_record("owner/served", &identity);
        assembly.registry_store().put_repo(&record).await.unwrap();
        let (pack, tip, base, tree_tip, blob) = chain_pack();
        run_push(
            "owner/served",
            &identity,
            assembly.registry().clone(),
            assembly.refs().clone(),
            assembly.packgen().clone(),
            assembly.ingest().clone(),
            &[create_command("refs/heads/main", tip)],
            &pack,
        )
        .await;

        #[derive(Clone)]
        struct SinkBytes(Arc<parking_lot::Mutex<Vec<u8>>>);
        impl SinkBytes {
            fn bytes(&self) -> Vec<u8> {
                self.0.lock().clone()
            }
        }
        impl std::io::Write for SinkBytes {
            fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
                self.0.lock().extend_from_slice(buf);
                Ok(buf.len())
            }
            fn flush(&mut self) -> std::io::Result<()> {
                Ok(())
            }
        }
        let sink = SinkBytes(Arc::new(parking_lot::Mutex::new(Vec::new())));
        assembly
            .packgen()
            .generate(
                &record,
                vec![tip],
                Vec::new(),
                Limits::default(),
                Box::new(sink.clone()),
            )
            .await
            .expect("the fetch generates over the through-store shape");

        let pack = sink.bytes();
        let mut entries = gix_pack::data::input::BytesToEntriesIter::new_from_header(
            std::io::Cursor::new(&pack),
            gix_pack::data::input::Mode::Verify,
            gix_pack::data::input::EntryDataMode::Keep,
            HashKind::Sha1,
        )
        .expect("the pack stream starts with a decodable header");
        let mut ids = Vec::new();
        let mut decompressor = gix_zlib::Decompress::new();
        for entry in &mut entries {
            let entry = entry.expect("every pack entry decodes");
            let compressed = entry.compressed.expect("keep mode retains entry bytes");
            let mut object = vec![0; entry.decompressed_size as usize];
            gix_zlib::stream::inflate::read(
                &mut std::io::Cursor::new(compressed),
                &mut decompressor,
                &mut object,
            )
            .expect("the entry inflates");
            decompressor.reset();
            let kind = match entry.header {
                PackHeader::Blob => Kind::Blob,
                PackHeader::Tree => Kind::Tree,
                PackHeader::Commit => Kind::Commit,
                PackHeader::Tag => Kind::Tag,
                other => panic!("the loose closure packs no deltas: {other:?}"),
            };
            ids.push(oid_of(kind, &object));
        }
        let tree_base = oid_of(Kind::Tree, &tree_bytes(&[("a.txt", &blob)]));
        let expected = std::collections::BTreeSet::from([tip, base, tree_base, tree_tip, blob]);
        assert_eq!(
            ids.iter()
                .copied()
                .collect::<std::collections::BTreeSet<ObjectId>>(),
            expected,
            "the through-store generation packs exactly the pushed chain's closure"
        );

        let _ = std::fs::remove_dir_all(&root);
    }

    // ——— the sqlite shape ———

    /// The default assembly composes co-tenant: the receive path
    /// composes end-to-end through it (resolve → authorize → serve →
    /// commit side → ingest), the legs observable on the durable feed,
    /// the seeded HEAD rides the entry's refs side, and the pushed
    /// chain is liveness-covered across a sweep. (Reopen/persistence is
    /// the twin assembly's own probe; the entry pins the composed legs.)
    #[cfg(feature = "sqlite")]
    #[tokio::test(flavor = "multi_thread")]
    async fn the_sqlite_assembly_composes_default_and_carries_the_receive_path() {
        let root = unique_root("sqlite");
        std::fs::create_dir_all(&root).unwrap();
        let options = AssemblyOptions::new(root.join("work"), root.join("large"));
        let assembly = SubstrateAssembly::sqlite(root.join("co-tenant.db"), options)
            .await
            .expect("the default assembly composes");
        let identity = identity();
        let record = granted_record("owner/composed-sqlite", &identity);
        assembly.registry_store().put_repo(&record).await.unwrap();

        let events = feed_events(&assembly, 0).await;
        assert_eq!(
            events.len(),
            1,
            "the announced put fired through the twin assembly the entry unified"
        );

        let (pack, tip, _base, _tree, _blob) = chain_pack();
        let report = run_push(
            "owner/composed-sqlite",
            &identity,
            assembly.registry().clone(),
            assembly.refs().clone(),
            assembly.packgen().clone(),
            assembly.ingest().clone(),
            &[create_command("refs/heads/main", tip)],
            &pack,
        )
        .await;
        assert_eq!(
            report,
            vec!["unpack ok".to_owned(), "ok refs/heads/main".to_owned()]
        );

        let lines = assembly.refs().list_refs(&record).await.unwrap();
        assert_eq!(
            lines
                .iter()
                .find(|line| line.name == "refs/heads/main")
                .map(|line| line.oid),
            Some(Some(tip))
        );

        let events = feed_events(&assembly, 1).await;
        assert_eq!(
            events,
            vec![crate::reactor::RegistryEvent::Mutation {
                seq: 2,
                repo_id: "owner/composed-sqlite".to_owned(),
            }],
            "the marker + announce legs rode the committed CAS"
        );

        let report = assembly.blobs().sweep().await.expect("the sweep runs");
        assert!(
            report.aborted.is_none() && report.deleted == 0,
            "the pushed chain is covered (the seeded closure over the co-tenant rows)"
        );

        let _ = std::fs::remove_dir_all(&root);
    }

    /// DQ-7 by construction at the entry: `:memory:` never reaches an
    /// engine — the sqlite entry refuses it before the twin assembly
    /// opens.
    #[cfg(feature = "sqlite")]
    #[tokio::test]
    async fn the_sqlite_entry_refuses_memory_mode() {
        let root = unique_root("dq7");
        let options = AssemblyOptions::new(root.join("work"), root.join("large"));
        let outcome = SubstrateAssembly::sqlite(":memory:", options).await;
        assert!(matches!(outcome, Err(OpenError::Invalid(_))));
    }

    /// The sqlite shape's LFS row store through the entry (the `lfs`
    /// accessor): rows asserted through the composition surface, the
    /// shared oid's refcounts carried, and the record-transaction
    /// removal leg — the rows go with the repo through
    /// `registry_store` (the twin's unconditional in-SQL copy). The
    /// mem shape's leg drains [`crate::lfs::InMemoryLfsRows`] (the
    /// mem test below); the pg shape runs the same script on the live
    /// harness (the `live` module below).
    #[cfg(all(feature = "lfs", feature = "sqlite"))]
    #[tokio::test(flavor = "multi_thread")]
    async fn the_sqlite_row_store_asserts_and_the_rows_leave_with_the_repo() {
        let root = unique_root("lfs-rows");
        std::fs::create_dir_all(&root).unwrap();
        let options = AssemblyOptions::new(root.join("work"), root.join("large"));
        let assembly = SubstrateAssembly::sqlite(root.join("co-tenant.db"), options)
            .await
            .expect("the default assembly composes");
        run_lfs_rows_removal_leg(&assembly).await;
        let _ = std::fs::remove_dir_all(&root);
    }

    #[cfg(all(feature = "lfs", not(feature = "sqlite")))]
    #[tokio::test(flavor = "multi_thread")]
    async fn the_mem_row_store_asserts_and_the_rows_leave_with_the_repo() {
        let assembly = mem_assembly("lfs-rows").await;
        run_lfs_rows_removal_leg(&assembly).await;
    }

    #[cfg(all(feature = "lfs", feature = "sqlite"))]
    #[tokio::test(flavor = "multi_thread")]
    async fn the_mem_row_store_asserts_and_the_rows_leave_with_the_repo() {
        let assembly = mem_assembly("lfs-rows").await;
        run_lfs_rows_removal_leg(&assembly).await;
    }

    /// The leg's script shared by the mem and sqlite shapes: two
    /// repos, one shared oid (refcount 2 on both rows), repo deletion
    /// through `registry_store` — the victim's rows gone, the
    /// survivor's row recomputed.
    #[cfg(feature = "lfs")]
    async fn run_lfs_rows_removal_leg(assembly: &SubstrateAssembly) {
        let identity = identity();
        assembly
            .registry_store()
            .put_repo(&granted_record("owner/victim-lfs", &identity))
            .await
            .unwrap();
        assembly
            .registry_store()
            .put_repo(&granted_record("owner/survivor-lfs", &identity))
            .await
            .unwrap();

        let shared_oid = "01".repeat(32);
        assembly
            .lfs_rows()
            .assert("owner/victim-lfs", &shared_oid, 16)
            .await
            .unwrap();
        assembly
            .lfs_rows()
            .assert("owner/survivor-lfs", &shared_oid, 16)
            .await
            .unwrap();
        let rows = assembly.lfs_rows().rows().await.unwrap();
        assert_eq!(rows.len(), 2);
        for row in &rows {
            assert_eq!(row.refcount, 2);
        }

        assembly
            .registry_store()
            .remove_repo("owner/victim-lfs")
            .await
            .unwrap();
        let rows = assembly.lfs_rows().rows().await.unwrap();
        assert_eq!(
            rows,
            vec![crate::lfs::LfsRow {
                repo_id: "owner/survivor-lfs".to_owned(),
                oid: shared_oid,
                size: 16,
                refcount: 1,
            }],
            "the victim's rows went with the repo; the survivor carries the recomputed count"
        );
    }

    // ——— the LFS liveness source's sweep contract, entry-wired ———

    /// Put one LFS media entry under its own family (the facade seam's
    /// put — the media family the liveness source marks) and read its
    /// oid back off the derived key: the hex the row store asserts by.
    #[cfg(feature = "lfs")]
    async fn put_media(
        assembly: &SubstrateAssembly,
        content: &'static [u8],
    ) -> (String, alkblobs::Key) {
        let (key, _pin) = assembly
            .blobs()
            .put_as(
                alkblobs::Algorithm::GitObjectSha256,
                Some(content.len() as u64),
                std::io::Cursor::new(content),
            )
            .await
            .expect("the media put lands");
        let hex: String = key
            .digest()
            .bytes()
            .iter()
            .map(|b| format!("{b:02x}"))
            .collect();
        (hex, key)
    }

    #[cfg(feature = "lfs")]
    async fn pool_has(assembly: &SubstrateAssembly, key: &alkblobs::Key) -> bool {
        assembly
            .blobs()
            .stat(&key.as_bytes())
            .await
            .expect("the pool probe reads")
            .is_some()
    }

    /// The sweep contract through the mem entry (the `lfs` source's
    /// fail-safe registration observable: the re-protected media never
    /// stages — an unregistered source's missed mark would stage it,
    /// the dangerous default ADR-022 §6 pins against). Cross-reference
    /// re-protection (one key, two rows, removed one by one across
    /// sweeps), orphan reclaim (the never-asserted media), and the
    /// repo-deletion reclaim (the rows leave with the repo through the
    /// record-removal leg; the media sweeps — nothing LFS-specific).
    #[cfg(feature = "lfs")]
    #[tokio::test(flavor = "multi_thread")]
    async fn the_mem_entrys_lfs_source_reprotects_and_reclaims_media() {
        let assembly = mem_assembly("lfs-gc").await;
        let _token = assembly.lfs_registration();

        let (shared_oid, shared_key) = put_media(&assembly, b"lfs media shared\n").await;
        let (_, orphan_key) = put_media(&assembly, b"lfs media orphan\n").await;
        let identity = identity();
        assembly
            .registry_store()
            .put_repo(&granted_record("owner/repro-1", &identity))
            .await
            .unwrap();
        assembly
            .registry_store()
            .put_repo(&granted_record("owner/repro-2", &identity))
            .await
            .unwrap();
        assembly
            .lfs_rows()
            .assert("owner/repro-1", &shared_oid, 16)
            .await
            .unwrap();
        assembly
            .lfs_rows()
            .assert("owner/repro-2", &shared_oid, 16)
            .await
            .unwrap();

        let first = assembly.blobs().sweep().await.expect("the sweep runs");
        assert!(first.aborted.is_none());
        assert_eq!(
            first.candidates_staged, 1,
            "only the never-asserted orphan marks uncovered"
        );
        assert_eq!(first.deleted, 1, "the orphan reclaims (POC-L's datum)");
        assert!(pool_has(&assembly, &shared_key).await, "referenced survive");
        assert!(
            !pool_has(&assembly, &orphan_key).await,
            "the orphan is gone from the pool"
        );

        assembly
            .lfs_rows()
            .remove("owner/repro-1", &shared_oid)
            .await
            .unwrap();
        let second = assembly.blobs().sweep().await.expect("the sweep runs");
        assert!(
            second.deleted == 0 && second.candidates_staged == 0,
            "one row removed — the key stays protected while any row remains"
        );
        assert!(pool_has(&assembly, &shared_key).await);

        assembly
            .lfs_rows()
            .remove("owner/repro-2", &shared_oid)
            .await
            .unwrap();
        let third = assembly.blobs().sweep().await.expect("the sweep runs");
        assert_eq!(
            third.deleted, 1,
            "the last row's removal makes the media sweep-eligible"
        );
        assert!(!pool_has(&assembly, &shared_key).await);

        let (victim_oid, victim_key) = put_media(&assembly, b"lfs victim media\n").await;
        assembly
            .registry_store()
            .put_repo(&granted_record("owner/victim-lfs", &identity))
            .await
            .unwrap();
        assembly
            .lfs_rows()
            .assert("owner/victim-lfs", &victim_oid, 16)
            .await
            .unwrap();
        let fourth = assembly.blobs().sweep().await.expect("the sweep runs");
        assert!(
            fourth.deleted == 0 && fourth.candidates_staged == 0,
            "the row stands — the media re-protects"
        );
        assembly
            .registry_store()
            .remove_repo("owner/victim-lfs")
            .await
            .unwrap();
        assert!(
            assembly
                .lfs_rows()
                .repo_rows("owner/victim-lfs")
                .await
                .unwrap()
                .is_empty(),
            "the rows went with the repo (the record-removal leg)"
        );
        let fifth = assembly.blobs().sweep().await.expect("the sweep runs");
        assert_eq!(
            fifth.deleted, 1,
            "the repo-deletion reclaim — nothing LFS-specific (lfs.md §3)"
        );
        assert!(!pool_has(&assembly, &victim_key).await);
    }

    /// The same contract's sqlite-entry rows (the default assembly
    /// shape's fail-safe registration observable): the re-protected
    /// media never stages, the uncovering marks it into the durable
    /// window (the shipped window delay defers the reclaim — the
    /// stage-only observable at the entry; the full stage→
    /// executed-overdue timing runs on the zero-delay wiring in
    /// [`crate::liveness::media_tests`]).
    #[cfg(all(feature = "lfs", feature = "sqlite"))]
    #[tokio::test(flavor = "multi_thread")]
    async fn the_sqlite_entrys_lfs_source_reprotects_media_across_the_sweep() {
        let root = unique_root("lfs-gc");
        std::fs::create_dir_all(&root).unwrap();
        let options = AssemblyOptions::new(root.join("work"), root.join("large"));
        let assembly = SubstrateAssembly::sqlite(root.join("co-tenant.db"), options)
            .await
            .expect("the default assembly composes");

        let (shared_oid, shared_key) = put_media(&assembly, b"lfs media shared\n").await;
        let identity = identity();
        assembly
            .registry_store()
            .put_repo(&granted_record("owner/repro-1", &identity))
            .await
            .unwrap();
        assembly
            .registry_store()
            .put_repo(&granted_record("owner/repro-2", &identity))
            .await
            .unwrap();
        assembly
            .lfs_rows()
            .assert("owner/repro-1", &shared_oid, 16)
            .await
            .unwrap();
        assembly
            .lfs_rows()
            .assert("owner/repro-2", &shared_oid, 16)
            .await
            .unwrap();

        let first = assembly.blobs().sweep().await.expect("the sweep runs");
        assert!(first.aborted.is_none());
        assert_eq!(
            first.candidates_staged, 0,
            "the rows re-protect the media — the registered source's mark consult"
        );
        assert_eq!(first.deleted, 0);

        assembly
            .lfs_rows()
            .remove("owner/repro-1", &shared_oid)
            .await
            .unwrap();
        let second = assembly.blobs().sweep().await.expect("the sweep runs");
        assert!(
            second.deleted == 0 && second.candidates_staged == 0,
            "one row removed — the survivor re-protects across the durable window"
        );
        assert!(pool_has(&assembly, &shared_key).await);

        assembly
            .lfs_rows()
            .remove("owner/repro-2", &shared_oid)
            .await
            .unwrap();
        let third = assembly.blobs().sweep().await.expect("the sweep runs");
        assert_eq!(
            third.candidates_staged, 1,
            "the last row's removal marks the media — staged into the window"
        );
        assert_eq!(third.deleted, 0, "the window defers the reclaim");
        assert!(pool_has(&assembly, &shared_key).await);

        let _ = std::fs::remove_dir_all(&root);
    }

    // ——— the postgres shape (the live-pg harness) ———

    #[cfg(feature = "postgres")]
    mod live {
        use super::*;
        use crate::postgres::live_harness::{drop_schema, harness_dsn, unique_schema, LIVE_GATE};

        /// The fleet/durable-shared assembly composes over the harness
        /// container: the receive path composes end-to-end, the legs
        /// observable on the durable feed, and the object pool rides
        /// the shared DSN.
        ///
        /// The live gate serializes the announced writes against the
        /// harness's database-wide silence-assertion probe (the
        /// `LISTEN/NOTIFY` channel is per-database, not per-schema —
        /// [`crate::postgres::live_harness::LIVE_GATE`]).
        #[tokio::test(flavor = "multi_thread")]
        async fn the_postgres_assembly_composes_and_carries_the_receive_path() {
            let Some(dsn) = harness_dsn() else {
                eprintln!(
                    "skipping (no postgres harness server): set \
                     ALKGIT_SUBSTRATE_PG_HOST (or the ALKBLOBS_PG_* harness env)"
                );
                return;
            };
            let _gate = LIVE_GATE.lock().await;
            let schema = unique_schema("compose");
            let registry = crate::postgres::PgRegistryOptions {
                schema: schema.clone(),
                engine: alkstore_postgres::PgOpts {
                    schema: format!("{schema}_engine"),
                    ..alkstore_postgres::PgOpts::default()
                },
                blobs: alkblobs::PgOpts {
                    schema: format!("{schema}_blobs"),
                    ..alkblobs::PgOpts::default()
                },
            };
            let root = unique_root("pg-compose");
            let options = AssemblyOptions::new(root.join("work"), root.join("large"));
            let assembly = match SubstrateAssembly::postgres(&dsn, registry.clone(), options).await
            {
                Ok(assembly) => assembly,
                Err(error) => {
                    drop_schema(&dsn, &schema).await;
                    drop_schema(&dsn, &format!("{}_engine", schema)).await;
                    drop_schema(&dsn, &format!("{}_blobs", schema)).await;
                    panic!("the pg assembly composes on the harness: {error}");
                }
            };

            let identity = identity();
            let record = granted_record("owner/composed-pg", &identity);
            assembly.registry_store().put_repo(&record).await.unwrap();
            assert_eq!(
                assembly
                    .registry()
                    .resolve("owner/composed-pg")
                    .await
                    .unwrap(),
                record,
                "the provider carried the fleet record"
            );

            let (pack, tip, _base, _tree, _blob) = chain_pack();
            let report = run_push(
                "owner/composed-pg",
                &identity,
                assembly.registry().clone(),
                assembly.refs().clone(),
                assembly.packgen().clone(),
                assembly.ingest().clone(),
                &[create_command("refs/heads/main", tip)],
                &pack,
            )
            .await;
            assert_eq!(
                report,
                vec!["unpack ok".to_owned(), "ok refs/heads/main".to_owned()]
            );

            let lines = assembly.refs().list_refs(&record).await.unwrap();
            assert_eq!(
                lines
                    .iter()
                    .find(|line| line.name == "refs/heads/main")
                    .map(|line| line.oid),
                Some(Some(tip))
            );

            let events = feed_events(&assembly, 1).await;
            assert_eq!(
                events,
                vec![crate::reactor::RegistryEvent::Mutation {
                    seq: 2,
                    repo_id: "owner/composed-pg".to_owned(),
                }],
                "the marker + announce legs rode the committed CAS through the \
                 fleet assembly"
            );

            let report = assembly.blobs().sweep().await.expect("the sweep runs");
            assert!(
                report.aborted.is_none() && report.deleted == 0,
                "the pushed chain is covered (the in-process closure source over \
                 the co-tenant pg rows)"
            );

            drop(assembly);
            drop_schema(&dsn, &schema).await;
            drop_schema(&dsn, &format!("{}_engine", schema)).await;
            drop_schema(&dsn, &format!("{}_blobs", schema)).await;
            let _ = std::fs::remove_dir_all(&root);
        }

        /// The fleet liveness answer's materialization exercise (wave
        /// 7's O-a booking, riding wave 10): the shipped pg entry
        /// pushes a real chain; the pushed chain's git keys ride the
        /// embedder's materialized liveness table beside the
        /// co-tenant tables (the composition's fleet-wide answer); a
        /// second instance — a kv-postgres **fleet** store over the
        /// same DSN, **no in-process liveness source** — registers the
        /// `FleetTable` reference and the store core reads it at the
        /// delete-time SQL arbitration. Pinned: the mark stages every
        /// pool candidate (the table is never enumerated); the
        /// materialized key cancels at the re-arbitration while the
        /// unmaterialized keys reclaim at the same execution; once the
        /// table row leaves, the last protected key reclaims too (the
        /// pool empties); and the fleet form errors on the entry's own
        /// store (the non-fleet boundary — the honest surface the
        /// entry ships). The wiring inputs this completes the booking
        /// with are the composition-adjacent materialization itself
        /// (the wave-8 entry's pg refs legs are its in-process half,
        /// consumed by this row's push through the entry).
        #[tokio::test(flavor = "multi_thread")]
        async fn the_fleet_table_materialization_serves_the_shared_arbitration() {
            use std::time::Duration;

            use alkblobs::LivenessRegistration;

            use crate::odb::oid_key;
            use crate::postgres::live_harness::connect_direct;

            let Some(dsn) = harness_dsn() else {
                eprintln!(
                    "skipping (no postgres harness server): set \
                     ALKGIT_SUBSTRATE_PG_HOST (or the ALKBLOBS_PG_* harness env)"
                );
                return;
            };
            let _gate = LIVE_GATE.lock().await;
            let schema = unique_schema("compose-fleet-table");
            let registry = crate::postgres::PgRegistryOptions {
                schema: schema.clone(),
                engine: alkstore_postgres::PgOpts {
                    schema: format!("{schema}_engine"),
                    ..alkstore_postgres::PgOpts::default()
                },
                blobs: alkblobs::PgOpts {
                    schema: format!("{schema}_blobs"),
                    ..alkblobs::PgOpts::default()
                },
            };
            let root = unique_root("pg-fleet-table");
            let options = AssemblyOptions::new(root.join("work"), root.join("large"));
            let assembly = match SubstrateAssembly::postgres(&dsn, registry.clone(), options).await
            {
                Ok(assembly) => assembly,
                Err(error) => {
                    drop_schema(&dsn, &schema).await;
                    drop_schema(&dsn, &format!("{}_engine", schema)).await;
                    drop_schema(&dsn, &format!("{}_blobs", schema)).await;
                    panic!("the pg assembly composes on the harness: {error}");
                }
            };

            // The real push through the entry: the pool carries the
            // chain's five git objects.
            let identity = identity();
            let record = granted_record("owner/fleet-liveness", &identity);
            assembly.registry_store().put_repo(&record).await.unwrap();
            let (pack, tip, base, tree, blob) = chain_pack();
            let report = run_push(
                "owner/fleet-liveness",
                &identity,
                assembly.registry().clone(),
                assembly.refs().clone(),
                assembly.packgen().clone(),
                assembly.ingest().clone(),
                &[create_command("refs/heads/main", tip)],
                &pack,
            )
            .await;
            assert_eq!(
                report,
                vec!["unpack ok".to_owned(), "ok refs/heads/main".to_owned()]
            );

            let chain_keys: Vec<alkblobs::Key> = [&tip, &base, &tree, &blob]
                .iter()
                .map(|oid| oid_key(oid).expect("the git-object key forms"))
                .collect();
            for key in &chain_keys {
                assert!(
                    assembly
                        .blobs()
                        .stat(key.key_bytes().as_slice())
                        .await
                        .expect("the pool stats")
                        .is_some(),
                    "the pushed chain's object sits in the shared pool under its oid key"
                );
            }
            let pool = assembly.blobs().list().await.expect("the pool lists");
            let pool_len = pool.len() as u64;
            assert_eq!(
                pool_len, 5,
                "the pushed chain is the pool's whole inventory: {pool:?}"
            );

            // The embedder's materialized liveness table beside the
            // co-tenant tables — the store never owns it (the
            // schema-ownership rule); the composition's materializer
            // writes it (the row carries the composition's half: the
            // one pushed chain's keys, the tip commit materialized).
            let probe = connect_direct(&dsn).await;
            probe
                .batch_execute(&format!(
                    "CREATE TABLE \"{schema}\".git_live_keys (key bytea PRIMARY KEY)"
                ))
                .await
                .expect("the embedder table creates");
            let tip_key = oid_key(&tip).expect("the tip's key forms");
            probe
                .execute(
                    &format!("INSERT INTO \"{schema}\".git_live_keys (key) VALUES ($1)"),
                    &[&tip_key.key_bytes().as_slice()],
                )
                .await
                .expect("the tip's key materializes");

            // The second instance: the fleet store over the same DSN
            // (one pool), the fleet table its only liveness
            // registration — the fleet-wide answer is the table's, no
            // in-process sources ride this instance.
            let fleet = alkblobs::Store::kv_only_postgres_fleet(
                &dsn,
                alkblobs::PgOpts {
                    schema: format!("{schema}_blobs"),
                    ..alkblobs::PgOpts::default()
                },
                1024 * 1024,
                Duration::ZERO,
                Duration::from_secs(60),
            )
            .await
            .expect("the fleet store constructs");
            let registration = fleet
                .register_liveness_source(LivenessRegistration::FleetTable(format!(
                    "{schema}.git_live_keys"
                )))
                .expect("the embedder table reference registers");
            assert!(
                assembly
                    .blobs()
                    .register_liveness_source(LivenessRegistration::FleetTable(
                        "some.where".to_owned()
                    ))
                    .is_err(),
                "the fleet form errors on the non-fleet entry store (the form boundary)"
            );

            // Sweep 1: the mark stages every pool candidate; the table
            // is never enumerated; nothing executes.
            let first = fleet.sweep().await.expect("the sweep runs");
            assert_eq!(
                first.aborted, None,
                "a fleet-table-only registration admits the sweep"
            );
            assert_eq!(
                first.candidates_staged, pool_len,
                "the mark stages the chain — the table is never enumerated"
            );
            assert_eq!(first.deleted, 0, "nothing overdue at the staging mark");

            // Sweep 2: the materialized key cancels at the delete-time
            // SQL re-arbitration; the unmaterialized keys reclaim at
            // the same execution — the protect/reclaim split is the
            // table's content.
            let second = fleet.sweep().await.expect("the sweep executes");
            assert_eq!(second.aborted, None);
            assert_eq!(
                second.cancelled_by_arbitration, 1,
                "the materialized key cancelled at the SQL re-arbitration"
            );
            assert_eq!(
                second.deleted,
                pool_len - 1,
                "the unmaterialized keys reclaimed at the same execution"
            );
            assert!(
                assembly
                    .blobs()
                    .stat(tip_key.key_bytes().as_slice())
                    .await
                    .expect("the pool stats")
                    .is_some(),
                "the protected key rides the table's protection"
            );

            // The table row leaves: the last protected key reclaims.
            probe
                .batch_execute(&format!("DELETE FROM \"{schema}\".git_live_keys"))
                .await
                .expect("the materializer's removal");
            let third = fleet.sweep().await.expect("the sweep executes");
            assert_eq!(third.aborted, None);
            assert_eq!(
                (third.deleted, third.cancelled_by_arbitration),
                (1, 0),
                "the unprotected key reclaims once its table row left"
            );
            for key in &chain_keys {
                assert!(
                    assembly
                        .blobs()
                        .stat(key.key_bytes().as_slice())
                        .await
                        .expect("the pool stats")
                        .is_none(),
                    "the pool emptied once the materialization left"
                );
            }
            let _ = fleet.unregister_liveness_source(registration);

            drop(assembly);
            drop(probe);
            drop_schema(&dsn, &schema).await;
            drop_schema(&dsn, &format!("{}_engine", schema)).await;
            drop_schema(&dsn, &format!("{}_blobs", schema)).await;
            let _ = std::fs::remove_dir_all(&root);
        }

        /// The pg shape's LFS row store through the entry: the removal
        /// leg's script over the harness (the mem/sqlite shapes' script,
        /// engine-swapped — the rows go with the repo inside the
        /// record-removal transaction).
        #[cfg(feature = "lfs")]
        #[tokio::test(flavor = "multi_thread")]
        async fn the_postgres_row_store_asserts_and_the_rows_leave_with_the_repo() {
            let Some(dsn) = harness_dsn() else {
                eprintln!(
                    "skipping (no postgres harness server): set \
                     ALKGIT_SUBSTRATE_PG_HOST (or the ALKBLOBS_PG_* harness env)"
                );
                return;
            };
            let _gate = LIVE_GATE.lock().await;
            let schema = unique_schema("compose-lfs-rows");
            let registry = crate::postgres::PgRegistryOptions {
                schema: schema.clone(),
                engine: alkstore_postgres::PgOpts {
                    schema: format!("{schema}_engine"),
                    ..alkstore_postgres::PgOpts::default()
                },
                blobs: alkblobs::PgOpts {
                    schema: format!("{schema}_blobs"),
                    ..alkblobs::PgOpts::default()
                },
            };
            let root = unique_root("pg-compose-lfs");
            let options = AssemblyOptions::new(root.join("work"), root.join("large"));
            let assembly = match SubstrateAssembly::postgres(&dsn, registry.clone(), options).await
            {
                Ok(assembly) => assembly,
                Err(error) => {
                    drop_schema(&dsn, &schema).await;
                    drop_schema(&dsn, &format!("{}_engine", schema)).await;
                    drop_schema(&dsn, &format!("{}_blobs", schema)).await;
                    panic!("the pg assembly composes on the harness: {error}");
                }
            };

            let identity = identity();
            assembly
                .registry_store()
                .put_repo(&granted_record("owner/victim-lfs", &identity))
                .await
                .unwrap();
            assembly
                .registry_store()
                .put_repo(&granted_record("owner/survivor-lfs", &identity))
                .await
                .unwrap();

            let shared_oid = "01".repeat(32);
            assembly
                .lfs_rows()
                .assert("owner/victim-lfs", &shared_oid, 16)
                .await
                .unwrap();
            assembly
                .lfs_rows()
                .assert("owner/survivor-lfs", &shared_oid, 16)
                .await
                .unwrap();

            assembly
                .registry_store()
                .remove_repo("owner/victim-lfs")
                .await
                .unwrap();
            assert_eq!(
                assembly.lfs_rows().rows().await.unwrap(),
                vec![crate::lfs::LfsRow {
                    repo_id: "owner/survivor-lfs".to_owned(),
                    oid: shared_oid,
                    size: 16,
                    refcount: 1,
                }],
                "the victim's rows went with the repo; the survivor carries the recomputed count"
            );

            drop(assembly);
            drop_schema(&dsn, &schema).await;
            drop_schema(&dsn, &format!("{}_engine", schema)).await;
            drop_schema(&dsn, &format!("{}_blobs", schema)).await;
            let _ = std::fs::remove_dir_all(&root);
        }

        /// The LFS liveness source's sweep-contract row on the live
        /// harness (the entry wiring's third shape): the re-protected
        /// media never stages while any row stands — cross-reference
        /// re-protection through one removed row — and the
        /// repo-deletion's uncovering marks the media into the
        /// durable window (the shipped window delay defers the
        /// reclaim; the full stage→executed-overdue timing is the
        /// zero-delay wiring's, [`crate::liveness::media_tests`]).
        #[cfg(feature = "lfs")]
        #[tokio::test(flavor = "multi_thread")]
        async fn the_postgres_entrys_lfs_source_reprotects_media_across_the_sweep() {
            let Some(dsn) = harness_dsn() else {
                eprintln!(
                    "skipping (no postgres harness server): set \
                     ALKGIT_SUBSTRATE_PG_HOST (or the ALKBLOBS_PG_* harness env)"
                );
                return;
            };
            let _gate = LIVE_GATE.lock().await;
            let schema = unique_schema("compose-lfs-gc");
            let registry = crate::postgres::PgRegistryOptions {
                schema: schema.clone(),
                engine: alkstore_postgres::PgOpts {
                    schema: format!("{schema}_engine"),
                    ..alkstore_postgres::PgOpts::default()
                },
                blobs: alkblobs::PgOpts {
                    schema: format!("{schema}_blobs"),
                    ..alkblobs::PgOpts::default()
                },
            };
            let root = unique_root("pg-compose-lfs-gc");
            let options = AssemblyOptions::new(root.join("work"), root.join("large"));
            let assembly = match SubstrateAssembly::postgres(&dsn, registry.clone(), options).await
            {
                Ok(assembly) => assembly,
                Err(error) => {
                    drop_schema(&dsn, &schema).await;
                    drop_schema(&dsn, &format!("{}_engine", schema)).await;
                    drop_schema(&dsn, &format!("{}_blobs", schema)).await;
                    panic!("the pg assembly composes on the harness: {error}");
                }
            };

            let (shared_oid, shared_key) = put_media(&assembly, b"lfs media shared\n").await;
            let identity = identity();
            assembly
                .registry_store()
                .put_repo(&granted_record("owner/lfs-repro-1", &identity))
                .await
                .unwrap();
            assembly
                .registry_store()
                .put_repo(&granted_record("owner/lfs-repro-2", &identity))
                .await
                .unwrap();
            assembly
                .lfs_rows()
                .assert("owner/lfs-repro-1", &shared_oid, 16)
                .await
                .unwrap();
            assembly
                .lfs_rows()
                .assert("owner/lfs-repro-2", &shared_oid, 16)
                .await
                .unwrap();

            let first = assembly.blobs().sweep().await.expect("the sweep runs");
            assert!(
                first.aborted.is_none() && first.candidates_staged == 0 && first.deleted == 0,
                "the rows re-protect the media — the registered source consulted the \
                 co-tenant pg rows"
            );
            assert!(pool_has(&assembly, &shared_key).await);

            assembly
                .lfs_rows()
                .remove("owner/lfs-repro-1", &shared_oid)
                .await
                .unwrap();
            let second = assembly.blobs().sweep().await.expect("the sweep runs");
            assert!(
                second.deleted == 0 && second.candidates_staged == 0,
                "one row removed — the survivor re-protects across the durable window"
            );

            assembly
                .registry_store()
                .remove_repo("owner/lfs-repro-2")
                .await
                .unwrap();
            let third = assembly.blobs().sweep().await.expect("the sweep runs");
            assert_eq!(
                third.candidates_staged, 1,
                "the rows all left with the repo — the mark uncovered the media"
            );
            assert_eq!(third.deleted, 0, "the durable window defers the reclaim");
            assert!(pool_has(&assembly, &shared_key).await);

            drop(assembly);
            drop_schema(&dsn, &schema).await;
            drop_schema(&dsn, &format!("{}_engine", schema)).await;
            drop_schema(&dsn, &format!("{}_blobs", schema)).await;
            let _ = std::fs::remove_dir_all(&root);
        }
    }
}