alkgit-substrate 0.1.0

//! The substrate object front (wave 7): alkblobs as the git-object
//! store under the derivation family, and the sync Find the pack
//! pipelines consume ([`gix_object::Find`] + [`gix_object::Exists`],
//! [`gix_pack::Find`]) — substrate.md §object storage; ADR-019 §1; the
//! POC-O `SubstrateOdb` pattern.
//!
//! # The derivation family (alkblobs ADR-018)
//!
//! Loose objects live under the git-object derivation family (registry
//! byte `0x03`; `0x04` is the sha256 member, registered but untouched
//! here — OQ-05 keeps every serving-path oid sha1). The stored preimage
//! is git's loose-object form — `<type> <len>\0` + content, the header
//! *inside* the value — so the digest is byte-identically git's oid of
//! the object, every kind included, and external-oid addressing is
//! map-free: the key is `[0x03] + oid bytes` ([`oid_key`]), built by
//! `Key::from_bytes` over the physical encoding (family-explicit,
//! rejects-never-guesses).
//!
//! The odb layer owns the header the family stores: it composes the
//! loose form ([`loose_form`]) it puts through the known-length
//! derivation seam (`put_as(Algorithm::GitObjectSha1, len, stream)` —
//! alkblobs 0.2's post-ADR-020 shape) and parses the header back out
//! through `gix_object::decode::loose_header` on read. Verification-on-
//! put is the oid re-derivation (the store derives the digest from the
//! stored bytes; a put whose returned key does not re-derive to the
//! object's oid is rejected — corruption cannot enter), and the front
//! re-derives on every read as well, so entries the store's kv tier
//! does not cover (large-tier payloads are not re-derived at open) fail
//! the read too. `stat` returns the *preimage* length — the stored
//! value's length, git's header included, never the content length (the
//! family semantic, alkblobs ADR-018 §Decision 3).
//!
//! # The Find front and the one-time construction
//!
//! [`ObjectFront`] is constructed once over the store (the
//! handle-sharing shape — backend.md §crate and feature model; ADR-004's
//! status clause: the retired disk-odb handle sharing is realized as
//! this front, constructed once with per-session views).
//! [`ObjectFront::session`] mints a per-session [`SessionFind`]: a view
//! sharing the front's store + runtime handles that carries its own
//! lookup-miss slot, so one session's failed walk cannot contaminate
//! another's. Clones of a view share that session's slot (the pipeline
//! clones the find — the core crate's `MemFind` double's sharing within
//! one session).
//!
//! `MemFind` is the core crate's test-double form of this exact shape
//! (`tests/common/mod.rs`). The location/entry/pack-offset
//! ([`gix_pack::Find::location_by_oid`],
//! [`gix_pack::Find::pack_offsets_and_oid`],
//! [`gix_pack::Find::entry_by_location`]) legs answer "not in any
//! pack": the substrate has no packed objects — the pipeline's `FindExt`
//! blanket impls pick up the rest. No disk odb exists anywhere on the
//! serving path (gix-odb is deliberately absent from the manifest).
//!
//! A lookup miss is recorded on the view ([`SessionFind::missed_object`])
//! so a failed walk/count stage maps its not-found error back to the
//! seam's `StorageError::ObjectMissing` abort (ADR-004), naming the
//! missing object rather than the want.
//!
//! # The sync Find / async store bridge — owned once, in one place
//!
//! gitoxide's Find traits are sync by design; the substrate store is
//! async. The bridge is this module's [`ObjectFront::block_on`]
//! (`Handle::block_on` against the runtime handle captured at
//! construction) — consumed from blocking pipeline tasks
//! (`tokio::task::spawn_blocking`), never from async contexts, where it
//! panics (the POC-O panic shape, learned live; poco-findings
//! §trait posture). Async contexts that must reach the store bridge with
//! [`ObjectFront::block_in_place`] (`block_in_place` there — it parks
//! the worker and makes `block_on` legal). The eventual
//! `GitPackGen`/`GitPackIngest` impls and the liveness source consume
//! these two postures — no consumer constructs its own `Handle`.
//!
//! # Error model
//!
//! The async legs ([`ObjectFront::put`], [`ObjectFront::get`],
//! [`ObjectFront::stat`]) surface the seam's `StorageError` family:
//! backing-store failures are `StorageError::Io` (message-only, no
//! engine source types — the same discipline as the other backend
//! impls), corruption detected at the front (a header that does not
//! parse back, a body that does not match its header, a re-derivation
//! mismatch) is `StorageError::Invalid` ("structurally invalid backend
//! state"). The sync trait layer maps those into `gix_error::Error`;
//! misses stay `Ok(None)` — a miss is a recorded result, not an error.
//! A non-sha1 oid ([`oid_key`]) is `StorageError::Invalid`: OQ-05's
//! rejects-never-guesses posture at the front.

use std::sync::Arc;

use alkblobs::{Algorithm, EntryMeta, Pin, Store};
use gix_hash::ObjectId;
use gix_object::Kind;
use tokio::runtime::Handle;

use alkgit::StorageError;

/// The derivation family every serving-path object rides: the sha1
/// member of the git-object family (alkblobs ADR-018, registry byte
/// `0x03`). The sha256 member (`0x04`) is registered but untouched —
/// OQ-05 keeps every oid sha1 until ecosystem need re-opens it.
pub(crate) const GIT_OBJECT_ALGORITHM: Algorithm = Algorithm::GitObjectSha1;

/// The key the derivation family derives an object's oid to: `[0x03] +
/// the oid bytes`, decoded through `Key::from_bytes` over the physical
/// encoding — family-explicit, rejects-never-guesses (alkblobs ADR-018;
/// ADR-020 §Decision 5's external-oid shape). A non-sha1 oid is
/// `StorageError::Invalid` (OQ-05: never silently re-addressed).
pub fn oid_key(oid: &gix_hash::oid) -> Result<alkblobs::Key, StorageError> {
    if oid.kind() != gix_hash::Kind::Sha1 {
        return Err(StorageError::Invalid(format!(
            "oid {oid} is not a sha1 oid; the object front addresses only the sha1 family (OQ-05)"
        )));
    }
    let mut bytes = Vec::with_capacity(1 + oid.as_bytes().len());
    bytes.push(GIT_OBJECT_ALGORITHM.byte());
    bytes.extend_from_slice(oid.as_bytes());
    alkblobs::Key::from_bytes(&bytes)
        .map_err(|e| StorageError::Invalid(format!("the oid-tagged key does not decode: {e}")))
}

/// The bounded prefix read behind [`ObjectFront::kind`]: the loose
/// header (`<type> <len>\0`) is never longer than ~30 bytes, so a kind
/// determination never touches the body — the is-commit half of the
/// ack predicate (ADR-014 §5) does not read the object's content.
pub(crate) const LOOSE_HEADER_PEEK_MAX: u64 = 128;

/// The object id behind a family key — `Some` only for a
/// `GitObjectSha1` key (the family the front serves; ADR-019 §1's
/// loose-form keying, alkblobs ADR-018); any other
/// algorithm is `None` (the `0x01`/`0x02` blob members and the `0x04`
/// member are not oid-addressable from here).
pub fn key_to_oid(key: &alkblobs::Key) -> Option<ObjectId> {
    if key.algorithm() != GIT_OBJECT_ALGORITHM {
        return None;
    }
    let raw = key.digest().bytes();
    if raw.len() != gix_hash::Kind::Sha1.len_in_bytes() {
        return None;
    }
    let mut bytes = [0u8; 20];
    bytes.copy_from_slice(raw);
    Some(ObjectId::from(bytes))
}

/// The stored preimage of a git object: git's loose-object form —
/// `<type> <len>\0` + content, the header *inside* the value (alkblobs
/// ADR-018; the digest over exactly these bytes is git's oid). The odb
/// layer owns this composition — [`ObjectFront::put`] puts it, the read
/// legs parse the header back out through
/// `gix_object::decode::loose_header`.
pub fn loose_form(kind: Kind, content: &[u8]) -> Vec<u8> {
    let mut form = gix_object::encode::loose_header(kind, content.len() as u64).into_vec();
    form.extend_from_slice(content);
    form
}

/// The async store legs' error mapping (crate-shared — the packgen
/// module's store legs compose through the same closure): backing-store
/// failure vs corruption-classified, no engine source types on the
/// surface. See the module's error-model section.
pub(crate) fn store_leg_error(context: &str, error: alkblobs::Error) -> StorageError {
    match error {
        alkblobs::Error::Verification | alkblobs::Error::KeyInvalid => {
            StorageError::Invalid(format!("{context}: {error}"))
        }
        alkblobs::Error::Missing
        | alkblobs::Error::Io(_)
        | alkblobs::Error::GcRefuse
        | alkblobs::Error::RegistryFull => StorageError::Io(format!("{context}: {error}")),
    }
}

/// The one-time object front over the substrate's alkblobs store: the
/// handle-sharing shape's constructed-once half (backend.md §crate and
/// feature model). The object pool is global, not per-repo-namespaced —
/// repo scoping is the refs rows' and liveness sources' business.
///
/// Construct within the running runtime (the assembly does): the
/// constructor captures `Handle::current()` for the sync/async bridge.
#[derive(Clone)]
pub struct ObjectFront {
    store: Arc<Store>,
    runtime: Handle,
}

impl ObjectFront {
    /// The front over the substrate store. The store handle is shared,
    /// not owned: pools are process-wide, the front is a view over one.
    pub fn new(store: Arc<Store>) -> Self {
        ObjectFront {
            store,
            runtime: Handle::current(),
        }
    }

    /// A per-session view: carries its own lookup-miss slot, sharing the
    /// front's store and runtime handles. Mint one per protocol session.
    pub fn session(&self) -> SessionFind {
        SessionFind {
            front: self.clone(),
            missed: Arc::default(),
        }
    }

    /// The shared store handle — the substrate composition's object
    /// pool. Batch puts (`put_in_batch`), range serving (`read_range`),
    /// and the GC wiring ride this handle; the front's legs are the
    /// loose-object seam over it.
    pub fn store(&self) -> &Arc<Store> {
        &self.store
    }

    /// The one sync Find / async store bridge, blocking-context half:
    /// the call pack pipelines own (`Handle::block_on` against the
    /// front's captured runtime handle — poco-findings §trait posture).
    /// Call it only from a blocking context (`tokio::task::spawn_blocking`
    /// pipeline tasks); in async contexts it panics (the POC-O panic
    /// shape — regression-tested) and [`ObjectFront::block_in_place`] is
    /// the bridge instead.
    pub fn block_on<F: std::future::Future>(&self, fut: F) -> F::Output {
        self.runtime.block_on(fut)
    }

    /// The bridge's async-context half (the POC-O liveness-source
    /// posture): parks the worker and serves the store call through the
    /// blocking pool, making the bridge legal on an async worker.
    /// Requires a multi-thread runtime.
    pub fn block_in_place<F>(&self, fut: F) -> F::Output
    where
        F: std::future::Future + Send,
        F::Output: Send,
    {
        tokio::task::block_in_place(move || self.runtime.block_on(fut))
    }

    /// The write leg: put one loose object under the family, riding the
    /// known-length derivation seam
    /// (`put_as(Algorithm::GitObjectSha1, len, stream)`) with the
    /// preimage length declared. Verification-on-put is the oid
    /// re-derivation: the store derives the digest from the stored
    /// bytes, and the front rejects a put whose returned key does not
    /// re-derive to the object's oid — corruption cannot enter.
    ///
    /// Returns the family key and the put's pin guard (the wave-8 pin
    /// bridge holds these across the ingest → CAS span; dropping the
    /// guard releases the entry to the sweep window).
    pub async fn put(
        &self,
        kind: Kind,
        content: &[u8],
    ) -> Result<(alkblobs::Key, Pin), StorageError> {
        let form = loose_form(kind, content);
        let (key, pin) = self
            .store
            .put_as(
                GIT_OBJECT_ALGORITHM,
                Some(form.len() as u64),
                std::io::Cursor::new(&form),
            )
            .await
            .map_err(|e| store_leg_error("alkblobs put_as", e))?;
        let oid = key_to_oid(&key).ok_or_else(|| {
            StorageError::Invalid(format!(
                "the put returned key {:?} is not the sha1 git-object family",
                key.digest().bytes()
            ))
        })?;
        let derived = GIT_OBJECT_ALGORITHM.derive(&form, form.len() as u64);
        if oid.as_bytes() != derived.bytes() {
            return Err(StorageError::Invalid(format!(
                "the put returned key {oid:?} does not re-derive to the object's oid"
            )));
        }
        Ok((key, pin))
    }

    /// The read leg: the loose object behind `oid`, its kind decoded
    /// from the header the odb layer owns. Verification-on-get is the
    /// oid re-derivation run over the delivered bytes (the store's kv
    /// tier verifies at open; the front re-verifies the full value so
    /// large-tier deliveries — not re-derived there — fail the read
    /// too): a corrupted entry is `StorageError::Invalid`.
    pub async fn get(&self, oid: &gix_hash::oid) -> Result<Option<(Kind, Vec<u8>)>, StorageError> {
        let key = oid_key(oid)?;
        let Some((len, blob)) = self
            .store
            .get(&key.as_bytes())
            .await
            .map_err(|e| store_leg_error("alkblobs get", e))?
        else {
            return Ok(None);
        };
        let value = blob
            .read_at(0, len)
            .await
            .map_err(|e| store_leg_error("alkblobs read_at", e))?;
        let (kind, header_len) = parse_loose_form(&value)?;
        let derived = GIT_OBJECT_ALGORITHM.derive(&value, value.len() as u64);
        if oid.as_bytes() != derived.bytes() {
            return Err(StorageError::Invalid(format!(
                "corruption detected at {oid}: the delivered value does not re-derive to the oid"
            )));
        }
        Ok(Some((kind, value[header_len..].to_vec())))
    }

    /// The object's kind behind `oid`, read from the loose header the
    /// odb layer owns — a bounded prefix read (`gix_object`'s
    /// `loose_header` parses
    /// a scan-to-NUL header; the body is never touched). `None` is a
    /// definitive miss. The kv tier's own get-time verification still
    /// fails a corrupted entry (mapped `Invalid`); the large tier's
    /// header parse is shape-level — body corruption fails the serving
    /// reads that consume the body instead.
    pub async fn kind(&self, oid: &gix_hash::oid) -> Result<Option<Kind>, StorageError> {
        let key = oid_key(oid)?;
        let Some((len, blob)) = self
            .store
            .get(&key.as_bytes())
            .await
            .map_err(|e| store_leg_error("alkblobs get", e))?
        else {
            return Ok(None);
        };
        let peeked = blob
            .read_at(0, LOOSE_HEADER_PEEK_MAX.min(len))
            .await
            .map_err(|e| store_leg_error("alkblobs header peek", e))?;
        let (kind, _size, _header_len) =
            gix_object::decode::loose_header(&peeked).map_err(|e| {
                StorageError::Invalid(format!(
                    "the stored loose-object form's header does not parse: {e}"
                ))
            })?;
        Ok(Some(kind))
    }

    /// The stat leg: the family entry's metadata — `len` is the
    /// *preimage* length (the stored loose form, git's header included),
    /// the family semantic (alkblobs ADR-018 §Decision 3), never the
    /// content length. `None` is a definitive miss.
    pub async fn stat(&self, oid: &gix_hash::oid) -> Result<Option<EntryMeta>, StorageError> {
        let key = oid_key(oid)?;
        let meta = self
            .store
            .stat(&key.as_bytes())
            .await
            .map_err(|e| store_leg_error("alkblobs stat", e))?;
        match meta {
            Some(meta) if meta.algorithm == GIT_OBJECT_ALGORITHM => Ok(Some(meta)),
            Some(meta) => Err(StorageError::Invalid(format!(
                "the family entry behind {oid} carries algorithm {:?}, not the sha1 git-object family",
                meta.algorithm
            ))),
            None => Ok(None),
        }
    }
}

/// A per-session view of the [`ObjectFront`]: the object lookup Find
/// the pack pipelines consume, carrying the session's lookup-miss slot
/// (one session's failed walk cannot leak another's miss — mint one
/// with [`ObjectFront::session`]). Clones share the slot, the shape the
/// pipeline's `find.clone()` calls assume within one session.
///
/// A lookup miss is recorded — the last missed oid — so a failed
/// walk/count stage maps its not-found error back to the seam's
/// `StorageError::ObjectMissing` abort ([`SessionFind::missed_object`];
/// the `MemFind` double's shape). Misses stay `Ok(None)`; corrupt
/// entries and backing failures surface as `Err`.
#[derive(Clone)]
pub struct SessionFind {
    front: ObjectFront,
    missed: Arc<parking_lot::Mutex<Option<ObjectId>>>,
}

impl SessionFind {
    /// The oid recorded by the latest lookup miss, if any (the recorded
    /// id names the missing object rather than the want).
    pub fn missed_object(&self) -> Option<ObjectId> {
        self.missed.lock().take()
    }
}

fn find_error(leg: StorageError) -> gix_error::Error {
    match leg {
        StorageError::ObjectMissing { oid } => {
            gix_error::message!("substrate odb: object {oid} does not exist").not_found_error()
        }
        StorageError::Invalid(_) => gix_error::message!("substrate odb: {leg}").validation_error(),
        StorageError::Io(_) => gix_error::message!("substrate odb: {leg}").retryable_error(),
    }
}

impl gix_object::Exists for SessionFind {
    fn exists(&self, id: &gix_hash::oid) -> bool {
        <Self as gix_pack::Find>::contains(self, id)
    }
}

impl gix_object::Find for SessionFind {
    fn try_find<'a>(
        &self,
        id: &gix_hash::oid,
        buffer: &'a mut Vec<u8>,
    ) -> gix_error::Result<Option<gix_object::Data<'a>>> {
        let find = <Self as gix_pack::Find>::try_find_cached(
            self,
            id,
            buffer,
            &mut gix_pack::cache::Never,
        )?;
        Ok(find.map(|(data, _location)| data))
    }
}

impl gix_pack::Find for SessionFind {
    fn contains(&self, id: &gix_hash::oid) -> bool {
        let oid: ObjectId = (id as &gix_hash::oid).to_owned();
        let Ok(key) = oid_key(&oid) else {
            return false;
        };
        self.front
            .block_on(self.front.store.has(&key.as_bytes()))
            .unwrap_or(false)
    }

    fn try_find_cached<'a>(
        &self,
        id: &gix_hash::oid,
        buffer: &'a mut Vec<u8>,
        _pack_cache: &mut dyn gix_pack::cache::DecodeEntry,
    ) -> gix_error::Result<
        Option<(
            gix_object::Data<'a>,
            Option<gix_pack::data::entry::Location>,
        )>,
    > {
        let oid: ObjectId = (id as &gix_hash::oid).to_owned();
        let found = self.front.block_on(self.front.get(&oid));
        match found {
            Ok(None) => {
                self.missed.lock().replace(oid);
                Ok(None)
            }
            Ok(Some((kind, data))) => {
                buffer.clear();
                buffer.extend_from_slice(&data);
                Ok(Some((
                    gix_object::Data {
                        kind,
                        object_hash: id.kind(),
                        data: buffer,
                    },
                    None,
                )))
            }
            Err(leg) => Err(find_error(leg)),
        }
    }

    fn location_by_oid(
        &self,
        _id: &gix_hash::oid,
        _buf: &mut Vec<u8>,
    ) -> gix_error::Result<Option<gix_pack::data::entry::Location>> {
        Ok(None)
    }

    fn pack_offsets_and_oid(
        &self,
        _pack_id: u32,
    ) -> gix_error::Result<Option<Vec<(gix_pack::data::Offset, ObjectId)>>> {
        Ok(None)
    }

    fn entry_by_location(
        &self,
        _location: &gix_pack::data::entry::Location,
    ) -> Option<gix_pack::find::Entry> {
        None
    }
}

/// The header parser: back out the kind and the body offset, and
/// validate that the body's length matches the header's declared length
/// — the odb layer's parse-back of the header it owns.
fn parse_loose_form(value: &[u8]) -> Result<(Kind, usize), StorageError> {
    let (kind, size, header_len) = gix_object::decode::loose_header(value).map_err(|e| {
        StorageError::Invalid(format!("the stored loose-object form does not parse: {e}"))
    })?;
    let body = value.len().saturating_sub(header_len) as u64;
    if body != size {
        return Err(StorageError::Invalid(format!(
            "the stored loose-object form's body is {body} bytes against a header declaring {size}"
        )));
    }
    Ok((kind, header_len))
}
#[cfg(test)]
mod tests {
    use alkblobs::{Store, StoreMode};
    use gix_hash::ObjectId;
    use gix_object::Kind;
    use std::sync::atomic::AtomicBool;
    use std::sync::Arc;

    use super::{key_to_oid, loose_form, oid_key, ObjectFront, SessionFind};
    use crate::testutil::scratch_dir as scratch_root;
    use alkgit::StorageError;

    fn mem_front() -> ObjectFront {
        ObjectFront::new(Arc::new(Store::mem_only()))
    }

    async fn put_object(front: &ObjectFront, kind: Kind, content: &[u8]) -> ObjectId {
        let (key, _pin) = front.put(kind, content).await.unwrap();
        key_to_oid(&key).expect("the put's key is the family key")
    }

    fn blob_content() -> &'static [u8] {
        b"hello\n"
    }

    fn tree_bytes(blob: &ObjectId) -> Vec<u8> {
        let mut tree = b"100644 hello\0".to_vec();
        tree.extend_from_slice(blob.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\n",
        );
        body.into_bytes()
    }

    async fn fixture_front() -> (ObjectFront, ObjectId) {
        let front = mem_front();
        let blob = put_object(&front, Kind::Blob, blob_content()).await;
        let tree = front
            .put(Kind::Tree, &tree_bytes(&blob))
            .await
            .map(|(key, _)| key_to_oid(&key).unwrap())
            .unwrap();
        let base = front
            .put(Kind::Commit, &commit_bytes(&tree, &[]))
            .await
            .map(|(key, _)| key_to_oid(&key).unwrap())
            .unwrap();
        let tip = front
            .put(Kind::Commit, &commit_bytes(&tree, &[&base]))
            .await
            .map(|(key, _)| key_to_oid(&key).unwrap())
            .unwrap();
        (front, tip)
    }

    #[tokio::test]
    async fn put_get_round_trips_through_the_family() {
        let front = mem_front();
        for (kind, content) in [
            (Kind::Blob, b"hello\n".as_slice()),
            (Kind::Blob, b"".as_slice()),
            (Kind::Blob, vec![0u8, 255, 0, 10, 13, 255].as_slice()),
        ] {
            let (key, _pin) = front.put(kind, content).await.unwrap();
            assert_eq!(
                key.algorithm(),
                alkblobs::Algorithm::GitObjectSha1,
                "the family byte the front writes is the sha1 git-object member"
            );
            let (got_kind, got) = front
                .get(&key_to_oid(&key).unwrap())
                .await
                .unwrap()
                .unwrap();
            assert_eq!(got_kind, kind);
            assert_eq!(got, content);
        }

        let blob = put_object(&front, Kind::Blob, b"hello\n").await;
        let tree = put_object(&front, Kind::Tree, &tree_bytes(&blob)).await;
        let commit = put_object(&front, Kind::Commit, &commit_bytes(&tree, &[])).await;
        let (got_kind, got) = front.get(&tree).await.unwrap().unwrap();
        assert_eq!(got_kind, Kind::Tree);
        assert_eq!(got, tree_bytes(&blob));
        let (got_kind, got) = front.get(&commit).await.unwrap().unwrap();
        assert_eq!(got_kind, Kind::Commit);
        assert_eq!(got, commit_bytes(&tree, &[]));
    }

    /// The load-bearing interop shape (alkblobs' `git_interop.rs`
    /// standard): the put's re-derived oid is byte-exact git's
    /// hash-object oid — hardcoded vectors plus, when a `git` is
    /// installed, a CLI cross-check in a detached scratch repo.
    #[tokio::test]
    async fn the_loose_preimage_hashes_to_git_s_oid() {
        let front = mem_front();
        let vectors: &[(&[u8], Kind, &str)] = &[
            (b"", Kind::Blob, "e69de29bb2d1d6434b8b29ae775ad8c2e48c5391"),
            (
                b"hello world\n",
                Kind::Blob,
                "3b18e512dba79e4c8300dd08aeb37f8e728b8dad",
            ),
            (b"", Kind::Tree, "4b825dc642cb6eb9a060e54bf8d69288fbee4904"),
        ];
        for (content, kind, oid_hex) in vectors {
            let oid = put_object(&front, *kind, content).await;
            assert_eq!(
                oid.to_string(),
                *oid_hex,
                "the derived oid is git's hash-object oid"
            );
        }
    }

    /// The CLI half of the interop shape (alkblobs' `git_interop.rs`
    /// harness detail): a real `git hash-object --stdin` in a detached
    /// scratch repo — skipped with a message when no `git` is installed.
    #[tokio::test]
    async fn the_put_oid_matches_the_git_cli() {
        if std::process::Command::new("git")
            .arg("--version")
            .output()
            .is_err()
        {
            eprintln!("skipping (no git binary installed)");
            return;
        }
        let dir = scratch_root("odb", "cli");
        let dir = dir.display().to_string();
        let out = std::process::Command::new("git")
            .args(["init", "-q"])
            .current_dir(&dir)
            .env("GIT_CONFIG_GLOBAL", "/dev/null")
            .env("GIT_CONFIG_SYSTEM", "/dev/null")
            .output()
            .unwrap();
        assert!(out.status.success(), "git init failed");
        for content in [b"hello\n".as_slice(), b"hello world\n".as_slice()] {
            let mut child = std::process::Command::new("git")
                .current_dir(&dir)
                .args(["hash-object", "--stdin"])
                .env("GIT_CONFIG_GLOBAL", "/dev/null")
                .env("GIT_CONFIG_SYSTEM", "/dev/null")
                .stdin(std::process::Stdio::piped())
                .stdout(std::process::Stdio::piped())
                .spawn()
                .unwrap();
            use std::io::Write as _;
            child.stdin.as_mut().unwrap().write_all(content).unwrap();
            let out = child.wait_with_output().unwrap();
            assert!(out.status.success(), "git hash-object failed");
            let oid_hex = String::from_utf8(out.stdout).unwrap().trim().to_owned();

            let front = mem_front();
            let oid = put_object(&front, Kind::Blob, content).await;
            assert_eq!(oid.to_string(), oid_hex, "the put's oid matches git's");
        }
        std::fs::remove_dir_all(&dir).unwrap_or_default();
    }

    /// `stat` returns the preimage length — the family semantic
    /// (alkblobs ADR-018 §Decision 3): the stored value's length, git's
    /// header included, never the content length.
    #[tokio::test]
    async fn stat_reports_the_preimage_length() {
        let front = mem_front();
        let content = b"hello".as_slice();
        let oid = put_object(&front, Kind::Blob, content).await;
        let form = loose_form(Kind::Blob, content);
        assert_ne!(form.len(), content.len(), "the header is inside the value");
        let meta = front.stat(&oid).await.unwrap().expect("the entry stats");
        assert_eq!(meta.len, form.len() as u64, "stat is the preimage length");
        assert_eq!(meta.len, 12, "header (7) + content (5) = the preimage");
        assert_eq!(
            meta.algorithm,
            alkblobs::Algorithm::GitObjectSha1,
            "stat's algorithm is the sha1 family member"
        );

        let missing = ObjectId::from([7u8; 20]);
        assert!(
            front.stat(&missing).await.unwrap().is_none(),
            "stat is a definitive miss, never a guessed zero"
        );
    }

    /// Verification-on-get over the kv tier: a corrupted stored value
    /// fails the read (the store derives at open; the front's own
    /// re-derivation backs it), riding the seeded sqlite store's `blobs`
    /// rows tampered in place.
    #[cfg(feature = "sqlite")]
    #[tokio::test]
    async fn corrupt_kvtier_entry_fails_the_read() {
        let dir = scratch_root("odb", "kv-corrupt");
        let db = dir.join("blobs.sqlite3");
        let store = Arc::new(
            Store::kv_only_sqlite(&db, 128 * 1024, alkblobs::DEFAULT_DELETE_WINDOW_DELAY)
                .await
                .unwrap(),
        );
        let front = ObjectFront::new(store);
        let oid = put_object(&front, Kind::Blob, blob_content()).await;
        assert!(front.get(&oid).await.is_ok(), "the pristine entry reads");

        let key = oid_key(&oid).unwrap();
        let conn = rusqlite::Connection::open(&db).unwrap();
        conn.execute_batch("PRAGMA journal_mode = WAL;").unwrap();
        let value = conn
            .query_row(
                "SELECT value FROM blobs WHERE key = ?",
                [key.as_bytes()],
                |row| row.get::<_, Vec<u8>>(0),
            )
            .unwrap();
        assert_eq!(
            value.len(),
            loose_form(Kind::Blob, blob_content()).len(),
            "the preimage is in the row"
        );
        let mut corrupted = value.clone();
        corrupted[2] = b'9';
        conn.execute(
            "UPDATE blobs SET value = ? WHERE key = ?",
            rusqlite::params![corrupted, key.as_bytes()],
        )
        .unwrap();

        match front.get(&oid).await {
            Err(StorageError::Invalid(reason)) => {
                assert!(
                    reason.contains("verification failed: content does not match key")
                        || reason.contains("re-derive")
                        || reason.contains("parse"),
                    "the failure names the corruption, not a backing failure: {reason}"
                );
            }
            Err(e) => panic!("corruption is a validation failure, not {e}"),
            Ok(_) => panic!("a corrupted entry cannot read back"),
        }
        let find = front.session();
        assert!(
            tokio::task::spawn_blocking(move || {
                gix_object::Find::try_find(&find, oid.as_ref(), &mut Vec::new()).is_err()
            })
            .await
            .unwrap(),
            "the Find leg fails the corrupted entry too"
        );
        std::fs::remove_dir_all(&dir).unwrap_or_default();
    }

    /// The front's own re-derivation: the large tier does not verify at
    /// open, so a corrupted over-threshold entry is caught by the front's
    /// oid re-derivation of the delivered bytes, not by the store.
    #[tokio::test]
    async fn corrupt_largetier_entry_fails_the_read() {
        let dir = scratch_root("odb", "large-corrupt");
        let store = Arc::new(Store::dual_tier_mem_local(&dir, 4).await.unwrap());
        const THRESHOLD: u64 = 4;
        assert_eq!(store.mode(), StoreMode::DualTier, "the large tier is live");
        let form_threshold = loose_form(Kind::Blob, b"!").len() as u64;
        assert!(form_threshold > THRESHOLD, "any loose form routes large");

        let front = ObjectFront::new(store);
        let oid = put_object(&front, Kind::Blob, blob_content()).await;
        assert!(front.get(&oid).await.is_ok(), "the pristine entry reads");

        let key = oid_key(&oid).unwrap();
        let hex = hex::encode(key.as_bytes());
        let entry_path = dir.join(&hex[..2]).join(&hex[2..4]).join(&hex);
        assert!(entry_path.is_file(), "the large tier's committed file");
        let mut value = std::fs::read(&entry_path).unwrap();
        let last = value.len() - 1;
        value[last] ^= 0xff;
        std::fs::write(&entry_path, &value).unwrap();

        match front.get(&oid).await {
            Err(StorageError::Invalid(reason)) => {
                assert!(
                    reason.contains("re-derive"),
                    "the failure names the corruption, not a backing failure: {reason}"
                );
            }
            Err(e) => panic!("corruption is a validation failure, not {e}"),
            Ok(_) => panic!("a corrupted entry cannot read back"),
        }
        std::fs::remove_dir_all(&dir).unwrap_or_default();
    }

    /// The kind leg: the header-derived kind without the body — every
    /// kind round-trips, a miss is `None`, and the peek never reads the
    /// content (a large-tier entry past the peek window still answers).
    #[tokio::test(flavor = "multi_thread")]
    async fn the_kind_leg_answer_from_the_header_not_the_body() {
        let dir = scratch_root("odb", "kind-leg");
        let store = Arc::new(Store::dual_tier_mem_local(&dir, 4).await.unwrap());
        let front = ObjectFront::new(store);
        for (kind, content) in [
            (Kind::Blob, b"hello\n".as_slice()),
            (Kind::Tree, tree_bytes(&ObjectId::from([3u8; 20])).as_slice()),
            (Kind::Commit, commit_bytes(&ObjectId::from([3u8; 20]), &[]).as_slice()),
            (Kind::Tag, b"object \0\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0\ntype commit\ntag v\ntagger A <a@example.invalid> 1750000000 +0000\n\nm\n".as_slice()),
        ] {
            let oid = put_object(&front, kind, content).await;
            assert_eq!(front.kind(&oid).await.unwrap(), Some(kind), "{kind:?}");
        }

        let missing = ObjectId::from([0xE2u8; 20]);
        assert!(
            front.kind(&missing).await.unwrap().is_none(),
            "a miss is None"
        );

        // A kind answer does not depend on the body: the peek window is
        // orders below a large-tier object's size and still answers.
        let big: Vec<u8> = std::iter::repeat_n(b'x', 4096).collect();
        let oid = put_object(&front, Kind::Blob, &big).await;
        assert_eq!(front.kind(&oid).await.unwrap(), Some(Kind::Blob));
        std::fs::remove_dir_all(&dir).unwrap_or_default();
    }

    /// OQ-05's posture at the front: every serving-path oid is sha1, the
    /// `0x04` family byte stays untouched, and every address the front
    /// builds decodes family-explicitly — never guessed.
    #[tokio::test]
    async fn the_front_serves_only_the_sha1_family() {
        let front = mem_front();
        put_object(&front, Kind::Blob, b"one\n").await;
        put_object(&front, Kind::Blob, b"two\n").await;

        for key in front.store().list().await.unwrap() {
            assert_eq!(key.algorithm(), alkblobs::Algorithm::GitObjectSha1);
            assert_eq!(key.as_bytes()[0], 0x03, "the sha1 family byte");
        }
        let (key, _pin) = front.put(Kind::Blob, b"three\n").await.unwrap();
        assert_eq!(
            key.as_bytes()[0],
            0x03,
            "the sha256 member (0x04) is never written"
        );
    }

    /// The three trait impls over the front — `gix_object::Find` +
    /// `Exists` + `gix_pack::Find` — with the location legs answering
    /// "not in any pack" (the `MemFind` double's exact shape). The sync
    /// Find legs run in a blocking task — the pipeline posture; from an
    /// async context the bridge panics.
    #[tokio::test]
    async fn the_find_legs_answer_with_no_packed_objects() {
        let (front, _tip) = fixture_front().await;
        let blob = put_object(&front, Kind::Blob, blob_content()).await;
        let find = front.session();

        tokio::task::spawn_blocking(move || {
            assert!(gix_object::Exists::exists(&find, blob.as_ref()));
            let missing = ObjectId::from([9u8; 20]);
            assert!(!gix_object::Exists::exists(&find, missing.as_ref()));

            let mut buffer = Vec::new();
            let data = gix_object::Find::try_find(&find, blob.as_ref(), &mut buffer)
                .unwrap()
                .expect("the blob is found");
            assert_eq!(data.kind, Kind::Blob);
            assert_eq!(data.data, blob_content());
            assert_eq!(data.object_hash, gix_hash::Kind::Sha1);

            use gix_object::FindExt as _;
            let mut buffer = Vec::new();
            let blob_ref = find.find_blob(blob.as_ref(), &mut buffer).unwrap();
            assert_eq!(
                blob_ref.data,
                blob_content(),
                "the FindExt blanket finds too"
            );

            use gix_pack::Find as _;
            let id = blob.as_ref() as &gix_hash::oid;
            assert!(
                find.contains(id),
                "the pack-level contains is the same lookup"
            );
            let locate = find.location_by_oid(id, &mut Vec::new()).unwrap();
            assert!(locate.is_none(), "no pack holds the object");
            let offsets = find.pack_offsets_and_oid(0).unwrap();
            assert!(offsets.is_none(), "no pack exists");
            let mut buffer = Vec::new();
            let resolved = find.try_find(id, &mut buffer).unwrap().expect("found");
            assert!(
                resolved.1.is_none(),
                "the location leg is none even on a resolved object"
            );
        })
        .await
        .unwrap();
    }

    /// `entry_by_location` answers none for every location shape — no
    /// packed objects anywhere on the serving path (no disk odb, gix-odb
    /// is deliberately absent from the manifest).
    #[tokio::test]
    async fn entry_by_location_is_unanswered() {
        let (front, _tip) = fixture_front().await;
        let find = front.session();
        let location = gix_pack::data::entry::Location {
            pack_id: 0,
            entry_size: 1,
            pack_offset: 0,
        };
        assert!(gix_pack::Find::entry_by_location(&find, &location).is_none());
    }

    /// Misses record per session: one session's failed walk cannot
    /// contaminate another's, the recorded id names the missing object
    /// (not the want), and the slot drains on take.
    #[tokio::test]
    async fn lookup_misses_record_per_session() {
        let (front, _tip) = fixture_front().await;
        let missing = ObjectId::from([11u8; 20]);
        let blob = put_object(&front, Kind::Blob, b"again\n").await;

        tokio::task::spawn_blocking(move || {
            let missing_id = missing.as_ref() as &gix_hash::oid;

            let one = front.session();
            let two = front.session();
            let mut buffer = Vec::new();
            assert_eq!(
                gix_object::Find::try_find(&one, missing_id, &mut buffer).unwrap(),
                None,
                "a miss is a recorded result, not an error"
            );
            assert_eq!(one.missed_object(), Some(missing));
            assert!(two.missed_object().is_none(), "the miss is session-private");
            assert!(one.missed_object().is_none(), "the slot drains on take");

            let find = front.session();
            assert!(
                gix_object::Find::try_find(&find, missing.as_ref(), &mut Vec::new())
                    .unwrap()
                    .is_none()
            );
            assert_eq!(find.missed_object(), Some(missing));
            assert!(
                gix_object::Find::try_find(&find, blob.as_ref(), &mut Vec::new())
                    .unwrap()
                    .is_some(),
                "a real lookup still succeeds after a recorded miss"
            );
        })
        .await
        .unwrap();
    }

    /// The POC-O panic shape, regression-tested: the blocking bridge
    /// (`Handle::block_on`) panics in async contexts — never call it
    /// there; `block_in_place` is the bridge instead.
    #[tokio::test]
    #[should_panic]
    async fn the_blocking_bridge_panics_in_async_contexts() {
        let (front, tip) = fixture_front().await;
        let _ = front.block_on(front.get(&tip));
    }

    /// The bridge's async-context half: `block_in_place` parks the
    /// worker and serves the store call through the blocking pool
    /// (multi-thread runtimes only) — the liveness-source posture.
    #[tokio::test(flavor = "multi_thread")]
    async fn the_async_context_bridges_with_block_in_place() {
        let (front, tip) = fixture_front().await;
        assert!(front.block_in_place(front.get(&tip)).unwrap().is_some());
    }

    type PipelineResult = Result<(Vec<u8>, gix_hash::ObjectId, u32), StorageError>;

    fn pipeline_error(find: &SessionFind, e: gix_error::Error, fallback: ObjectId) -> StorageError {
        if e.is_not_found() {
            StorageError::ObjectMissing {
                oid: find.missed_object().unwrap_or(fallback),
            }
        } else {
            StorageError::Io(e.to_string())
        }
    }

    fn closure_of(find: &SessionFind, tip: ObjectId) -> Result<Vec<ObjectId>, StorageError> {
        let mut closure = Vec::new();
        let mut stack = vec![tip];
        while let Some(current) = stack.pop() {
            if closure.contains(&current) {
                continue;
            }
            let mut buf = Vec::new();
            let commit = gix_object::FindExt::find_commit(find, &current, &mut buf)
                .map_err(|e| pipeline_error(find, e, tip))?;
            for parent in &commit.parents {
                let parent =
                    gix_hash::ObjectId::from_hex(parent).expect("the stored parent is hex");
                stack.push(parent);
            }
            closure.push(current);
        }
        Ok(closure)
    }

    /// The generation-pipeline smoke over the front (ADR-004's Find-
    /// generic stages): the commit-ancestry closure rides `FindExt`,
    /// the count stage runs over the pack Find, and the entry/byte
    /// stages stream a V2 sha1 pack through the front's lookup legs.
    #[tokio::test]
    async fn the_generation_pipeline_smokes_over_the_front() {
        let (front, tip) = fixture_front().await;
        let find = front.session();
        let pack = tokio::task::spawn_blocking(move || smoke_pack(&find, tip))
            .await
            .unwrap();
        let (pack_bytes, digest, _objects) =
            pack.unwrap_or_else(|e| panic!("the pipeline packs over the front: {e}"));
        assert!(pack_bytes.starts_with(b"PACK"), "the pack header wrote");
        let trailing = &pack_bytes[pack_bytes.len() - 20..];
        assert_eq!(
            trailing,
            digest.as_slice(),
            "the trailer is the pipeline's digest"
        );
    }

    fn smoke_pack(find: &SessionFind, want: ObjectId) -> PipelineResult {
        use gix_pack::data::output::bytes::FromEntriesIter;
        use gix_pack::data::output::count;
        use gix_pack::data::output::entry::iter_from_counts;
        use gix_pack::data::output::entry::iter_from_counts::{Mode, Options, Outcome};
        use gix_parallel::Finalize as _;

        let commits = closure_of(find, want)?;
        let counter = gix_utils::progress::Discard;
        let interrupt = AtomicBool::new(false);
        let (counts, _outcome) = count::objects_unthreaded(
            find,
            &mut commits.into_iter().map(Ok::<ObjectId, gix_error::Error>),
            &counter,
            &interrupt,
            count::objects::ObjectExpansion::TreeContents,
        )
        .map_err(|e| pipeline_error(find, e, want))?;
        let counts_len = counts.len() as u32;

        let sink = Vec::<u8>::new();
        let mut entries = gix_parallel::InOrderIter::from(
            iter_from_counts(
                counts,
                find.clone(),
                Box::new(gix_utils::progress::Discard),
                Options {
                    thread_limit: Some(1),
                    mode: Mode::PackCopyAndBaseObjects,
                    allow_thin_pack: false,
                    chunk_size: 10,
                    version: gix_pack::data::Version::V2,
                    compression: gix_zlib::Compression::DEFAULT,
                },
            )
            .map_err(|e| pipeline_error(find, e, want))?,
        );
        let mut from_entries = FromEntriesIter::new(
            entries.by_ref(),
            sink,
            counts_len,
            gix_pack::data::Version::V2,
            gix_hash::Kind::Sha1,
        );
        for step in from_entries.by_ref() {
            step.map_err(|e| pipeline_error(find, e, want))?;
        }
        let digest = from_entries
            .digest()
            .ok_or_else(|| StorageError::Io("the trailer was never written".to_string()))?;
        let pack_bytes = from_entries.into_write();
        let statistics: Outcome = entries
            .inner
            .finalize()
            .map_err(|e| pipeline_error(find, e, want))?;
        assert_eq!(
            statistics.missing_objects, 0,
            "ADR-004's abort posture: no missing objects in the smoke"
        );
        Ok((pack_bytes, digest, counts_len))
    }
}