alkgit-substrate 0.1.0

//! The LFS batch/transfer adapter (wave 9) — the storage-side logic of
//! the LFS protocol's batch/checking and byte paths, as a substrate
//! surface under the `lfs` feature (ADR-022 §1–2; lfs.md §1–2; the
//! substrate-side piece of ADR-025's LFS-surface member). The alkgit
//! core has no native LFS op set (git-lfs speaks HTTP only) — the door
//! is the alkhttp `lfs` feature (family infrastructure); this surface
//! is what that door consumes, front-door-blind: it carries the
//! storage-side mapping and never touches HTTP.
//!
//! # The adapter/door split (the wave's spec-level call, pinned here
//! for the gate)
//!
//! ADR-025's neutral record names the door-side batch mapping as
//! riding alkblobs directly; the row assertion is the
//! composition-owned piece only this surface carries. Drawn exactly:
//!
//! **The door's side (never here):** the JSON batch protocol parsing
//! and the HTTP route surface; the auth plumbing (alkhttp ADR-073's
//! Basic-userinfo tolerance on the batch); the **anonymous-public
//! `anon` identity synthesis** (the adapter refuses an absent
//! identity — the door synthesizes the identity, the grant
//! short-circuits on the public-repo flag); **href/token minting**
//! ([`LfsHrefMint`] is the mint-input carrier — the door mints the
//! absolute URLs with per-action `Authorization`; the adapter never
//! mints a URL or a door token); HTTP `Range`-header parsing and the
//! 206/416 responses; the object-size ceiling choice (the budget per
//! assembly, the door's [`LfsAdapterConfig::max_object_bytes`]); and
//! the batch request's `transfers` negotiation (the adapter carries
//! no negotiation input and answers `basic` — the door reads the
//! client's list and refuses a request that excludes `basic` before
//! the batch reaches the adapter; the wave-9 gate's recorded door
//! duty, lfs.md §1's pinned shape).
//!
//! **The adapter's side (this surface):** the batch's per-object
//! existence answered over the ops control plane — `blobs/stat`
//! through an
//! [`OperationContext`]
//! built with the caller's identity and the *wired* ownership
//! provider (the POC-L door's DirectOps binding shape as a substrate
//! surface); the response objects' shapes (`{oid, size,
//! authenticated}` + the per-object action grant-carriers); **the row
//! assertion at check time** (the moment the server learns the pair —
//! lfs.md §3, over the [`LfsRows`] surface); the
//! byte paths: the **pre-keyed put** (upload) and the **ranged fetch
//! with per-window slice-digest verification** (download); the
//! wired-provider requirement (never the static fallback).
//!
//! **What rides alkblobs directly (no substrate surface):** the JSON
//! batch protocol itself is the door's; the facade seam
//! (`Store::put_as(GitObjectSha256, len, stream)` / `read_range`) is
//! the in-process embedder alternative with no ops plane at all —
//! deliberately not re-exposed here (the embedder binds
//! [`SubstrateAssembly::blobs`](crate::compose::SubstrateAssembly::blobs)
//! directly; the adapter is the ops-plane surface). The wire ops
//! (`blobs/put` / `blobs/fetch`, the channel data plane) remain the
//! fleet/remote consumer's form — the DirectOps-vs-loopback binding
//! choice sits door-side.
//!
//! # The pinned protocol facts (lfs.md §1, riding the shapes)
//!
//! - `hash_algo` answers `sha256` only: a batch carrying any other
//!   declared algorithm is refused ([`LfsAdapterError::HashAlgo`] —
//!   the door's 409). `verify` is not served: no batch path produces
//!   a verify action.
//! - Transfer is `basic`; the download action's mint input is the
//!   whole media oid (range negotiation never rides the batch — the
//!   client's `Range` rides the transfer GET).
//! - The upload key is **formable in advance** (ADR-022 §1: under
//!   `Algorithm::GitObjectSha256` the oid *is* the digest — the key is
//!   `[0x04] + oid bytes`): existence is a direct `blobs/stat` on the
//!   formed key (the translate-then-stat is direct; there is no
//!   mapping row, and no `have` use — that is the replication diff
//!   verb).
//!
//! # The byte paths (ADR-022 §2)
//!
//! - **Upload — the pre-keyed put.** The offer carries the claimed
//!   oid and the batch's size; a size-less offer is a protocol error
//!   ([`LfsAdapterError::SizelessOffer`] — LFS clients always send
//!   size; the door refuses before the body). The put rides the put
//!   handler's contract minus the channel: the one-read tee (a sha256
//!   pass alongside a bounded pipe the facade put consumes
//!   concurrently via `put_as(GitObjectSha256, Some(len), …)`), the
//!   claim cross-check **before the facade's EOF** — a body the
//!   derivation contradicts kills the put pre-commit
//!   ([`LfsAdapterError::MislabeledBody`]: no stored content, no held
//!   pin — the zero-residue discipline the put's stage hygiene
//!   carries on both tiers) — then the custody hand-over through the
//!   [`OpsContext`] verbs: fleet
//!   (`mint_fleet`, then the immediate `confirm` — the upload is
//!   complete, so the token's ticket retires at the confirmation and
//!   the mint boundary's admission cap stays live per upload, ADR-022
//!   §5; the fleet pin row's TTL is the durable protection — a
//!   dropped unconfirmed ticket would hold an admission slot until
//!   process end, the wave-9 gate's fix) or non-fleet
//!   (`mint_custodial` + `confirm`, the LFS pattern POC-L validated;
//!   a mint refusal at
//!   the admission cap is [`LfsAdapterError::MintRefused`], answering
//!   the already-committed put honestly). The reference row is
//!   asserted at upload time too (batch check or upload — lfs.md §3).
//! - **Download — the ranged fetch.** Every download serves verified
//!   windows: the requested range (or the whole entry) is split
//!   internally at [`LfsAdapterConfig::window_bytes`] and each window
//!   rides `Store::read_range`, whose ADR-006 slice digest is checked
//!   mid-flight — a window's bytes are yielded only after the
//!   independently computed sha256 of exactly the returned window
//!   bytes matches the read's slice digest; a mismatch aborts the
//!   stream mid-flight (the caller re-fetches — the ranged arm's
//!   answer is the mid-flight-verified shape). **The resolved span
//!   is budget-gated** by [`LfsAdapterConfig::max_object_bytes`] —
//!   every form the adapter will actually produce counts (the
//!   range-less whole serve, the open-ended resume, any resolved
//!   span; `bytes=0-` is a whole download and refuses over the
//!   ceiling — the wave-9 gate's fix, the whole-arm budget's
//!   bypass closed). **Verification, stated precisely** (the
//!   gate's fix): a whole-span download additionally rides the
//!   end-to-end oid digest check — the windows' bytes accumulate
//!   into one sha256 pass, and the finished digest is compared
//!   against the key's derivation digest (the identity family's
//!   digest is the raw content's hash, so the comparison is the oid
//!   verification the large tier's reads carry no re-derivation
//!   for) **before the last window yields**. A partial span's
//!   windows carry the length + slice checks only — the store
//!   interior's ADR-006 digests, not an oid conclusion. **The
//!   pinned window-splitting call** (drawn against lfs.md's "the
//!   door owns window splitting", flagged for the gate): that fact
//!   governs the door binding the *ops byte-path* (the
//!   loopback/remote forms, where `fetch_max_total_bytes` bounds
//!   the op's ranged arm). This side of the HTTP boundary the
//!   adapter owns the arithmetic: a download call splits internally
//!   at `window_bytes` per verified window; the door's HTTP work is
//!   Range parsing and the budget, not window counting.
//!
//! # The wired ownership provider (alkcall ADR-048's posture, composed)
//!
//! The adapter takes the ownership provider **as a required
//! constructor argument** — there is no Option, no static path, and
//! every context it builds wires
//! `ownership: Some(provider)`: the static `Identity.resources`
//! fallback is unreachable through this surface (the POC-L
//! gateway-gap regression's posture; the battery pins it — an
//! identity carrying a static resources claim is refused unless the
//! wired provider grants). The shipped provider is
//! [`LfsGrantProvider`]: alkgit's registry composed onto the ops ACL
//! — the same per-repo grants the git surface composes (ADR-011's
//! posture), served through the cached read provider's sync resolve.
//! alkcall ADR-048's loud default guards the gateway dispatch spine
//! (the wire path); here the guard is the constructor requirement
//! plus the battery. The `anon` composition is door-side (above).

use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::Arc;
use std::task::{Context as PollContext, Poll};
use std::time::{SystemTime, UNIX_EPOCH};

use alkblobs::ops::{stat_pair, OpsConfig, OpsContext, PinConverter, BLOB_MISSING};
use alkblobs::{Key, Pin, Store as BlobStore};
use alkcall::core::auth::Identity;
use alkcall::core::ownership::OwnershipProvider;
use alkcall::core::types::Capabilities;
use alkcall::registry::context::{AbortPolicy, OperationContext, ScopedPeerEnv};
use alkcall::registry::env::{LocalOperationEnv, OperationEnv};
use alkcall::registry::registration::OperationRegistry;
use alkcall::OperationRegistryBuilder;
use sha2::{Digest as _, Sha256};
use tokio::io::{AsyncRead, AsyncReadExt, ReadBuf};
use tokio::sync::mpsc;

use crate::error::OpenError;
use crate::lfs::LfsRows;
use crate::liveness::lfs_media_key;
use crate::registry::RegistryRead;

/// The batch/transfer adapter's knobs (the door budgets the ceiling;
/// the adapter splits at the window). Defaults are the POC-L datums:
/// the 32 MiB split window and the `fetch_max_total_bytes`-shaped
/// 64 MiB object ceiling (raised per assembly to trade opens for
/// windows — lfs.md §2).
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct LfsAdapterConfig {
    /// The download windows' boundary — the size each verified
    /// `read_range` window serves before the next (the POC-L split
    /// datum, 32 MiB). Must be non-zero.
    pub window_bytes: u64,
    /// The download span ceiling (the door's budget per assembly —
    /// lfs.md §2's "the door budgets the whole arm", realized
    /// adapter-side): the **resolved span** of any download form is
    /// the budgeted quantity — a range-less whole serve, an
    /// open-ended resume (`bytes=0-`), or any resolved span over the
    /// ceiling refuses [`LfsAdapterError::TooLarge`]; a partial span
    /// under it rides the window arithmetic (a resume's legitimate
    /// use).
    pub max_object_bytes: u64,
}

impl Default for LfsAdapterConfig {
    fn default() -> Self {
        LfsAdapterConfig {
            window_bytes: 32 * 1024 * 1024,
            max_object_bytes: 64 * 1024 * 1024,
        }
    }
}

/// The batch API's operation half (the door maps the wire's
/// `"upload"` / `"download"` onto it).
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum LfsBatchOperation {
    /// The client means to push media (the absent response objects
    /// carry upload-action mint inputs).
    Upload,
    /// The client means to pull media (the present response objects
    /// carry download-action mint inputs).
    Download,
}

impl LfsBatchOperation {
    /// The wire form's parse (the door's mapping input); anything
    /// else is not a batch operation.
    pub fn from_wire(wire: &str) -> Option<Self> {
        match wire {
            "upload" => Some(LfsBatchOperation::Upload),
            "download" => Some(LfsBatchOperation::Download),
            _ => None,
        }
    }
}

/// One batch request object: the pointer's oid and its mandatory
/// size (the batch's size is mandatory — LFS clients always send it;
/// the door refuses the size-less form before this surface).
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct LfsBatchObject {
    /// The LFS media oid (the raw sha256 of the content — 64 hex, the
    /// media family's digest; case as the client sent it, the
    /// validated form is lowercased).
    pub oid: String,
    /// The mandatory asserted media size.
    pub size: u64,
}

/// The batch/checking request as the door hands it over (post-JSON,
/// post-auth plumbing — the identity resolved, the operation mapped).
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct LfsBatchRequest {
    /// The batch's operation half.
    pub operation: LfsBatchOperation,
    /// The client's declared hash algorithm (`None` = the default).
    /// Anything but `"sha256"` is refused (the pinned protocol fact).
    pub hash_algo: Option<String>,
    /// The requested objects.
    pub objects: Vec<LfsBatchObject>,
}

/// The action-grant carrier the response objects hold — the **mint
/// inputs**, never a URL: the door mints the absolute href and the
/// per-action `Authorization` from these (front-door-blind — no URL
/// or door token is formed here).
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct LfsHrefMint {
    /// The repo id (the ops namespace — the href's repo-path input).
    pub repo_id: String,
    /// The object's oid (the href's object-path input).
    pub oid: String,
}

/// The per-object action grants (the LFS shape: `upload` rides the
/// absent objects' answers for an upload batch, `download` the
/// present ones for a download batch; at most one side is populated
/// per response object).
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct LfsActions {
    /// The upload-action mint input (present when the media is
    /// absent and the caller means to upload).
    pub upload: Option<LfsHrefMint>,
    /// The download-action mint input (present when the media is
    /// present and the caller means to download).
    pub download: Option<LfsHrefMint>,
}

/// A per-object error, carried in the response object (the LFS
/// per-object error shape — the object's `error` field; the HTTP
/// mapping stays the door's).
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct LfsObjectError {
    /// The LFS error code (403 refused, 404 not found, 422
    /// malformed / disagreement, 500 server).
    pub code: u16,
    /// The stable message text.
    pub message: &'static str,
}

impl LfsObjectError {
    /// The refused-per-grant object.
    fn refused() -> Self {
        LfsObjectError {
            code: 403,
            message: "refused by grant",
        }
    }

    /// The absent-object error (a download batch's miss).
    fn not_found() -> Self {
        LfsObjectError {
            code: 404,
            message: "object not found",
        }
    }

    /// The malformed-oid error (not a 64-lowercase-hex sha256 form).
    fn malformed() -> Self {
        LfsObjectError {
            code: 422,
            message: "malformed oid",
        }
    }

    /// The size-disagreement error (the store entry's length and the
    /// asserted size disagree — the pointer contradicts the content).
    fn disagree() -> Self {
        LfsObjectError {
            code: 422,
            message: "size disagreement",
        }
    }

    /// The storage-side failure error.
    fn server() -> Self {
        LfsObjectError {
            code: 500,
            message: "server error",
        }
    }
}

/// One batch response object — the pinned shape `{oid, size,
/// authenticated}` plus the optional per-object error/actions
/// branches. `authenticated` is `true` whenever the object is
/// answered: the request rode the wired ownership provider's grant
/// (the ops ACL path), so every served action is server-authorized —
/// the anonymous-public composition (the door's synthesized `anon`
/// identity on a public repo's read) answers the same way.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct LfsObjectAnswer {
    /// The object's oid, echoed in the validated lowercase form.
    pub oid: String,
    /// The object's asserted size.
    pub size: u64,
    /// The server-authorized flag (see the type doc).
    pub authenticated: bool,
    /// The per-object error branch (`None` for answered objects).
    pub error: Option<LfsObjectError>,
    /// The per-object action branch (`None` when there is nothing to
    /// do — the already-present object of an upload batch, the
    /// idempotent re-push's posture).
    pub actions: Option<LfsActions>,
}

/// The batch/checking response — the transfer is `basic` and the
/// served algorithm is `sha256` (the pinned protocol facts; the door
/// serializes the JSON shape).
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct LfsBatchResponse {
    /// The negotiated transfer adapter (`"basic"`).
    pub transfer: &'static str,
    /// The served hash algorithm (`"sha256"` — the only one).
    pub hash_algo: &'static str,
    /// One answer per request object, request order.
    pub objects: Vec<LfsObjectAnswer>,
}

/// The byte range a download call serves: `[offset, offset + len)`
/// with `len = None` meaning to the entry's end (the door maps the
/// HTTP `bytes=a-b` / `bytes=a-` forms onto the explicit count).
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct LfsRange {
    /// The range's first byte offset.
    pub offset: u64,
    /// The range's byte count (`None` = through the entry's end).
    pub len: Option<u64>,
}

/// The adapter's error surface (thiserror family; message-only —
/// the engine source types stay off the surface, the row store's
/// [`StorageError`](alkgit::StorageError) text is carried). The
/// whole-batch failures (forbidden, hash-algo) and the transfer-path
/// failures are errors; per-object failures ride the batch response's
/// error objects, not this surface.
#[derive(Clone, Debug, PartialEq, Eq, thiserror::Error)]
pub enum LfsAdapterError {
    /// The caller is unauthorized (no identity, or the wired provider
    /// does not grant the operation's action on the namespace). The
    /// door's mapping is the 401 / generic denial (unknown ≡
    /// unauthorized).
    #[error("forbidden: {0}")]
    Forbidden(String),
    /// The batch declared a hash algorithm other than `sha256` (the
    /// pinned 409 fact).
    #[error("unsupported hash_algo {0}: the server serves sha256 only")]
    HashAlgo(String),
    /// The transfer payload is malformed (a non-oid form).
    #[error("invalid LFS input: {0}")]
    Invalid(String),
    /// The whole download's object exceeds the configured ceiling.
    #[error("object of {0} bytes exceeds the configured download ceiling")]
    TooLarge(u64),
    /// The download's byte range names nothing the entry serves
    /// (a past-EOF offset — the door's 416).
    #[error("the requested range is not satisfiable")]
    NotSatisfiable,
    /// The storage read or the serving machinery failed.
    #[error("storage: {0}")]
    Storage(String),
    /// The row assertion failed.
    #[error("lfs rows: {0}")]
    Rows(String),
    /// The upload's offer carried no size (the pre-keyed put's
    /// mandatory offer — the pinned protocol refusal).
    #[error("size-less offer refused: the pre-keyed put requires the batch's size")]
    SizelessOffer,
    /// The upload's body failed its claim: the stream's sha256 is
    /// not the claimed oid (the put's contract — nothing stored, no
    /// held pin; killed pre-EOF, zero residue).
    #[error("mislabeled body: the stream's digest contradicts the claimed oid {0}")]
    MislabeledBody(String),
    /// The custody hand-over failed post-commit (the confirm verb
    /// refused, the mint failed operationally — the put landed; the
    /// custody question is the operator's).
    #[error("custody: {0}")]
    Custody(String),
    /// The token mint was refused at the registry's admission cap
    /// (alkblobs ADR-019): the put already committed and verified;
    /// no token issued, no custody taken — the re-put is
    /// CAS-idempotent.
    #[error("mint refused at the token registry's admission cap: {0}")]
    MintRefused(String),
}

/// The wired ownership provider — alkgit's registry composed onto the
/// ops ACL (the batch's authorization posture: lfs.md §1 — the
/// per-repo grant governs batch access, the ADR-011 posture composed
/// onto the ops ACL). The alkblobs ops specs gate on `blob-namespace`
/// resources whose id is the payload's namespace — the repo id; this
/// provider answers with the same [`authorize`](alkgit::authorize)
/// policy function the git surface composes: `read` granted (or
/// anonymous-public — the door's synthesized `anon` identity rides
/// the public-repo flag here), `write` per the grant monotonicity
/// (`manage` ⊃ `write` ⊃ `read`). Unknown repos deny by default —
/// the unknown ≡ unauthorized posture. Served through the cached
/// read provider's sync resolve (the hot path — no await, no round
/// trip; a stale cache delays revocation until the next refresh, the
/// same posture the git serving path takes).
pub struct LfsGrantProvider {
    provider: Arc<crate::registry::CachedRegistryProvider>,
}

impl LfsGrantProvider {
    /// The provider over a cached read provider (the assembly's —
    /// [`SubstrateAssembly::provider`](crate::compose::SubstrateAssembly::provider)).
    pub fn new(provider: Arc<crate::registry::CachedRegistryProvider>) -> Self {
        LfsGrantProvider { provider }
    }
}

impl OwnershipProvider for LfsGrantProvider {
    fn owns(
        &self,
        identity: &Identity,
        resource_type: &str,
        resource_id: &str,
        action: &str,
    ) -> bool {
        if resource_type != "blob-namespace" {
            return false;
        }
        let action = match action {
            "read" => alkgit::AccessAction::Read,
            "write" => alkgit::AccessAction::Write,
            _ => return false,
        };
        match self.provider.resolve(resource_id) {
            Some(record) => alkgit::authorize(&record, Some(identity), action),
            None => false,
        }
    }

    fn owned_resources(&self, identity: &Identity, resource_type: &str) -> Vec<String> {
        if resource_type != "blob-namespace" {
            return Vec::new();
        }
        let snapshot = self.provider.snapshot();
        let mut owned = Vec::new();
        for (repo_id, record) in snapshot.iter() {
            if alkgit::authorize(record, Some(identity), alkgit::AccessAction::Read) {
                owned.push(repo_id.clone());
            }
        }
        owned.sort();
        owned
    }

    fn owns_any(&self, identity: &Identity, resource_type: &str) -> bool {
        !self.owned_resources(identity, resource_type).is_empty()
    }
}

/// The LFS put's converter — the substrate's `PinConverter`
/// registration act (ADR-022 §4's pinned sentence, realized): the
/// converter stays what alkblobs pins it to be (sync-only,
/// failure-free, `Key`-only) and registers **nothing** — the rich
/// reference metadata lives in alkgit's rows (asserted by this
/// surface before the confirm), and the entry's store-side
/// protection is the LFS media liveness source registered at the
/// composition entry (the fail-safe default), which consults the
/// rows. The confirm verb demands *a* converter per namespace (an
/// unbound converter refuses, mapping intact — alkblobs ADR-010 §4);
/// this is the honest act.
struct RowsAssertedConverter;

impl PinConverter for RowsAssertedConverter {
    fn convert(&self, _key: &Key) {}
}

/// The batch/transfer adapter — the storage-side surface the alkhttp
/// `lfs` door consumes (ADR-022 §1–3, lfs.md §1–2 — the
/// adapter/door split is the module doc). Construct over the object
/// pool, the rows surface, the **wired** ownership provider
/// (required — see the module doc), and the config knobs.
pub struct LfsAdapter {
    ops: Arc<OpsContext>,
    registry: Arc<OperationRegistry>,
    env: Arc<LocalOperationEnv>,
    rows: Arc<dyn LfsRows>,
    ownership: Arc<dyn OwnershipProvider>,
    config: LfsAdapterConfig,
    requests: AtomicU64,
}

impl LfsAdapter {
    /// Construct the adapter. The registry assembled here is the ops
    /// control plane the batch answers over (the POC-L DirectOps
    /// binding shape): `blobs/stat` through the [`OpsContext`], the
    /// namespace ACL riding the wired provider — the one op this
    /// surface registers, so the ops fetch op's own knobs
    /// (`fetch_max_total_bytes` and friends) stay at their defaults
    /// and are not this surface's budget: downloads ride
    /// `read_range` directly under the resolved-span ceiling
    /// ([`LfsAdapterConfig::max_object_bytes`], enforced adapter-side).
    /// Fallible on the config's shape (the fallible-constructor
    /// posture).
    pub fn new(
        store: Arc<BlobStore>,
        rows: Arc<dyn LfsRows>,
        ownership: Arc<dyn OwnershipProvider>,
        config: LfsAdapterConfig,
    ) -> Result<Self, OpenError> {
        if config.window_bytes == 0 {
            return Err(OpenError::Invalid(
                "the LFS adapter's window_bytes must be non-zero".to_owned(),
            ));
        }
        let ops = Arc::new(OpsContext::new(Arc::clone(&store), OpsConfig::default()));
        let (spec, handler) = stat_pair(Arc::clone(&ops));
        let registry = Arc::new(
            OperationRegistryBuilder::new()
                .with_local(spec, handler, None, None, Capabilities::new())
                .map_err(|e| OpenError::Invalid(format!("the ops registry assembly: {e}")))?
                .build(),
        );
        let env = Arc::new(LocalOperationEnv::new(Arc::clone(&registry)));
        Ok(LfsAdapter {
            ops,
            registry,
            env,
            rows,
            ownership,
            config,
            requests: AtomicU64::new(0),
        })
    }

    /// The adapter's config (the door reads the budget knobs).
    pub fn config(&self) -> &LfsAdapterConfig {
        &self.config
    }

    /// The per-request `OperationContext` — the POC-L door's DirectOps
    /// posture: the caller's identity, the wired ownership provider
    /// (always `Some` — never the static fallback), the local
    /// operation env, abort defaults. A request id forms from a
    /// per-adapter counter (correlation data; the wire never sees it).
    fn context(&self, identity: Option<Identity>) -> OperationContext {
        let nanos = SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .map(|d| d.as_nanos())
            .unwrap_or_default();
        OperationContext {
            request_id: format!(
                "lfs-adapter-{}-{nanos}",
                self.requests.fetch_add(1, Ordering::Relaxed)
            ),
            parent_request_id: None,
            identity,
            handler_identity: None,
            forwarded_for: None,
            capabilities: Capabilities::new(),
            metadata: std::collections::HashMap::new(),
            scoped_env: ScopedPeerEnv::empty(),
            env: Arc::clone(&self.env) as Arc<dyn OperationEnv>,
            abort_policy: AbortPolicy::default(),
            deadline: None,
            internal: false,
            ownership: Some(Arc::clone(&self.ownership)),
        }
    }

    /// The batch/checking answer (lfs.md §1): the formable-key
    /// existence over the ops control plane (`blobs/stat` through the
    /// context), the response objects' shapes, and the row assertion
    /// at check time — every validated object's `(repo_id, oid, size)`
    /// pair is asserted **before** the probes (the moment the server
    /// learns the pair; a malformed oid cannot form a row and its
    /// answer is the per-object 422). The batch's own gates run
    /// first: no identity, or the wired provider not granting the
    /// operation's action (`write` for upload batches, `read` for
    /// download batches), is the forbidden refusal; a declared
    /// `hash_algo` other than `sha256` is the pinned 409.
    pub async fn check(
        &self,
        identity: Option<&Identity>,
        repo_id: &str,
        request: LfsBatchRequest,
    ) -> Result<LfsBatchResponse, LfsAdapterError> {
        let identity = require_identity(identity)?;
        if let Some(algo) = &request.hash_algo {
            if algo != "sha256" {
                return Err(LfsAdapterError::HashAlgo(algo.clone()));
            }
        }
        let action = match request.operation {
            LfsBatchOperation::Upload => "write",
            LfsBatchOperation::Download => "read",
        };
        if !self
            .ownership
            .owns(&identity, "blob-namespace", repo_id, action)
        {
            return Err(LfsAdapterError::Forbidden(format!(
                "not granted: blob-namespace/{repo_id} {action}"
            )));
        }

        for object in &request.objects {
            if lfs_media_key(&object.oid).is_ok() {
                self.rows
                    .assert(repo_id, &lower_hex(&object.oid), object.size)
                    .await
                    .map_err(|e| LfsAdapterError::Rows(e.to_string()))?;
            }
        }

        let mut objects = Vec::with_capacity(request.objects.len());
        for object in &request.objects {
            let oid = lower_hex(&object.oid);
            let answer = match lfs_media_key(&oid) {
                Err(_) => LfsObjectAnswer {
                    oid,
                    size: object.size,
                    authenticated: true,
                    error: Some(LfsObjectError::malformed()),
                    actions: None,
                },
                Ok(key) => {
                    self.answers(
                        &identity,
                        repo_id,
                        request.operation,
                        &oid,
                        object.size,
                        &key,
                    )
                    .await
                }
            };
            objects.push(answer);
        }
        Ok(LfsBatchResponse {
            transfer: "basic",
            hash_algo: "sha256",
            objects,
        })
    }

    /// One validated request object's answer: the stat probe over the
    /// control plane, then the operation's branch (the mint inputs —
    /// never URLs — for the to-be-uploaded absent object or the
    /// downloadable present one, the per-object error shapes).
    async fn answers(
        &self,
        identity: &Identity,
        repo_id: &str,
        operation: LfsBatchOperation,
        oid: &str,
        size: u64,
        key: &Key,
    ) -> LfsObjectAnswer {
        match self.stat(identity, repo_id, key).await {
            StatProbe::Missing => match operation {
                LfsBatchOperation::Upload => LfsObjectAnswer {
                    oid: oid.to_owned(),
                    size,
                    authenticated: true,
                    error: None,
                    actions: Some(LfsActions {
                        upload: Some(LfsHrefMint {
                            repo_id: repo_id.to_owned(),
                            oid: oid.to_owned(),
                        }),
                        download: None,
                    }),
                },
                LfsBatchOperation::Download => LfsObjectAnswer {
                    oid: oid.to_owned(),
                    size,
                    authenticated: true,
                    error: Some(LfsObjectError::not_found()),
                    actions: None,
                },
            },
            StatProbe::Length(len) if len == size => match operation {
                LfsBatchOperation::Upload => LfsObjectAnswer {
                    oid: oid.to_owned(),
                    size,
                    authenticated: true,
                    error: None,
                    actions: None,
                },
                LfsBatchOperation::Download => LfsObjectAnswer {
                    oid: oid.to_owned(),
                    size,
                    authenticated: true,
                    error: None,
                    actions: Some(LfsActions {
                        upload: None,
                        download: Some(LfsHrefMint {
                            repo_id: repo_id.to_owned(),
                            oid: oid.to_owned(),
                        }),
                    }),
                },
            },
            StatProbe::Length(_) => LfsObjectAnswer {
                oid: oid.to_owned(),
                size,
                authenticated: true,
                error: Some(LfsObjectError::disagree()),
                actions: None,
            },
            StatProbe::Forbidden => LfsObjectAnswer {
                oid: oid.to_owned(),
                size,
                authenticated: false,
                error: Some(LfsObjectError::refused()),
                actions: None,
            },
            StatProbe::Failed => LfsObjectAnswer {
                oid: oid.to_owned(),
                size,
                authenticated: true,
                error: Some(LfsObjectError::server()),
                actions: None,
            },
        }
    }

    /// The ops control plane's stat probe (the formable key's
    /// existence): `blobs/stat` invoked directly on the assembled
    /// registry through a per-object context — the ACL runs exactly
    /// as on the wire (the `resource_id_path` selects the namespace).
    async fn stat(&self, identity: &Identity, namespace: &str, key: &Key) -> StatProbe {
        let envelope = self
            .registry
            .invoke(
                "blobs/stat",
                serde_json::json!({
                    "namespace": namespace,
                    "digest": key_hex(key),
                }),
                self.context(Some(identity.clone())),
            )
            .await;
        match envelope.result {
            Ok(payload) => {
                let len = payload
                    .get("len")
                    .and_then(serde_json::Value::as_u64)
                    .unwrap_or_default();
                StatProbe::Length(len)
            }
            Err(error) if error.code == BLOB_MISSING => StatProbe::Missing,
            Err(error) if error.code == "FORBIDDEN" => StatProbe::Forbidden,
            Err(_) => StatProbe::Failed,
        }
    }

    /// The upload — the pre-keyed put (ADR-022 §2). The identity and
    /// the write grant are required (the wired provider; the door
    /// carries the Basic tolerance). The size is the put's mandatory
    /// claim: `None` is the size-less refusal. The row is asserted
    /// first (the moment the upload learns the pair), then the put
    /// rides the put handler's contract minus the channel: the
    /// one-read tee over a bounded pipe, the claim cross-check
    /// pre-EOF (a mislabeled body kills the put — zero residue), then
    /// the custody hand-over through the `OpsContext` verbs — mint
    /// and confirm (the fleet form mints and drops the guard; the
    /// non-fleet form mints custodial and confirms, the POC-L
    /// attach-reference pattern).
    pub async fn upload<R>(
        &self,
        identity: Option<&Identity>,
        repo_id: &str,
        oid: &str,
        size: Option<u64>,
        body: R,
    ) -> Result<(), LfsAdapterError>
    where
        R: AsyncRead + Unpin + Send,
    {
        let identity = require_identity(identity)?;
        if !self
            .ownership
            .owns(&identity, "blob-namespace", repo_id, "write")
        {
            return Err(LfsAdapterError::Forbidden(format!(
                "not granted: blob-namespace/{repo_id} write"
            )));
        }
        let size = size.ok_or(LfsAdapterError::SizelessOffer)?;
        let oid = lower_hex(oid);
        let expected = lfs_media_key(&oid).map_err(|e| LfsAdapterError::Invalid(e.to_string()))?;

        self.rows
            .assert(repo_id, &oid, size)
            .await
            .map_err(|e| LfsAdapterError::Rows(e.to_string()))?;

        let (key, pin) =
            pre_keyed_put(Arc::clone(self.ops.store()), &expected, &oid, size, body).await?;

        self.hand_over(repo_id, key, pin).await
    }

    /// The custody hand-over post-commit (the put handler's finish
    /// arm plus the LFS confirm): fleet — `mint_fleet` over the
    /// committed row (the owner is the serving instance's id, minted
    /// against the store's fleet declaration) then an immediate
    /// `confirm` through [`RowsAssertedConverter`]: the upload is
    /// complete before the hand-over, so the token's ticket retires
    /// at the confirmation (the mint boundary's admission cap stays
    /// live per upload — ADR-022 §5's hostile-loop posture — while a
    /// dropped unconfirmed ticket would hold an admission slot until
    /// process end, the wave-9 gate's fix); the fleet pin row's TTL
    /// remains the durable protection. Non-fleet — `mint_custodial`
    /// (the registry takes the guard) then `confirm`. A mint refused
    /// at the cap answers the already-committed put honestly
    /// ([`LfsAdapterError::MintRefused`]; the re-put is
    /// CAS-idempotent); an operational mint failure or a refused
    /// confirm is [`LfsAdapterError::Custody`].
    async fn hand_over(&self, namespace: &str, key: Key, pin: Pin) -> Result<(), LfsAdapterError> {
        let refused = |key: &Key| {
            LfsAdapterError::MintRefused(format!(
                "no token issued, no custody taken; the committed key {} stands — the re-put is \
                 CAS-idempotent",
                hex_of(key.digest().bytes()),
            ))
        };
        match self.ops.fleet_owner() {
            Some(fleet) => match self.ops.mint_fleet(namespace, key.clone(), &fleet) {
                Ok(token) => {
                    self.ops
                        .bind_converter(namespace, Arc::new(RowsAssertedConverter));
                    if self.ops.confirm(namespace, &token) {
                        drop(pin);
                        Ok(())
                    } else {
                        drop(pin);
                        Err(LfsAdapterError::Custody(
                            "the confirm conversion refused (the mapping stands intact)".to_owned(),
                        ))
                    }
                }
                Err(alkblobs::Error::RegistryFull) => Err(refused(&key)),
                Err(e) => {
                    drop(pin);
                    Err(LfsAdapterError::Custody(e.to_string()))
                }
            },
            None => {
                self.ops
                    .bind_converter(namespace, Arc::new(RowsAssertedConverter));
                match self.ops.mint_custodial(namespace, key.clone(), pin) {
                    Ok(token) if self.ops.confirm(namespace, &token) => Ok(()),
                    Ok(_token) => Err(LfsAdapterError::Custody(
                        "the confirm conversion refused (the mapping stands intact)".to_owned(),
                    )),
                    Err(alkblobs::Error::RegistryFull) => Err(refused(&key)),
                    Err(e) => Err(LfsAdapterError::Custody(e.to_string())),
                }
            }
        }
    }

    /// The download — the ranged fetch (ADR-022 §2): the read grant is
    /// required (the wired provider), the entry's existence rides
    /// `stat`, and the served stream verifies per-window slice
    /// digests mid-flight (the module doc's byte-path section). The
    /// resolved span is budget-gated by
    /// [`LfsAdapterConfig::max_object_bytes`] — every form (the
    /// range-less whole serve, the open-ended resume, any resolved
    /// span) the adapter will actually produce counts against the
    /// ceiling; a past-EOF offset is the not-satisfiable refusal (the
    /// door's 416). A whole-span download additionally verifies the
    /// served bytes' sha256 against the oid end-to-end at EOF (the
    /// large tier's reads carry no canonical re-derivation — the
    /// oid check is this path's job); a partial span's windows carry
    /// the length + store-interior slice checks only (the module
    /// doc's verification statement).
    pub async fn download(
        &self,
        identity: Option<&Identity>,
        repo_id: &str,
        oid: &str,
        range: Option<LfsRange>,
    ) -> Result<LfsDownload, LfsAdapterError> {
        let identity = require_identity(identity)?;
        if !self
            .ownership
            .owns(&identity, "blob-namespace", repo_id, "read")
        {
            return Err(LfsAdapterError::Forbidden(format!(
                "not granted: blob-namespace/{repo_id} read"
            )));
        }
        let oid = lower_hex(oid);
        let key = lfs_media_key(&oid).map_err(|e| LfsAdapterError::Invalid(e.to_string()))?;
        let meta = self
            .ops
            .store()
            .stat(key.key_bytes().as_slice())
            .await
            .map_err(|e| LfsAdapterError::Storage(e.to_string()))?;
        let entry_len = match meta {
            Some(meta) => meta.len,
            None => {
                return Err(LfsAdapterError::Storage(
                    "the media entry is absent".to_owned(),
                ))
            }
        };
        let (from, until) = match range {
            None => (0, entry_len),
            Some(range) => {
                if range.offset >= entry_len {
                    return Err(LfsAdapterError::NotSatisfiable);
                }
                let end = match range.len {
                    Some(len) => range.offset.saturating_add(len).min(entry_len),
                    None => entry_len,
                };
                (range.offset, end)
            }
        };
        if until - from > self.config.max_object_bytes {
            return Err(LfsAdapterError::TooLarge(entry_len));
        }
        let whole_span = from == 0 && until == entry_len;
        Ok(LfsDownload::start(
            Arc::clone(self.ops.store()),
            key,
            from,
            until,
            self.config.window_bytes,
            whole_span,
        ))
    }
}

/// The probe's outcome (the control plane's answer, folded to what
/// the batch branch consumes).
enum StatProbe {
    /// The formed key is absent from the pool (`BLOB_MISSING`).
    Missing,
    /// The entry's length (`blobs/stat`'s `len`).
    Length(u64),
    /// The ACL refused (a raced revocation mid-batch).
    Forbidden,
    /// Any other failure.
    Failed,
}

/// The lowercase-hex normalization (the validated oid form).
fn lower_hex(oid: &str) -> String {
    oid.chars().map(|c| c.to_ascii_lowercase()).collect()
}

/// The digest's hex form over the key's physical bytes (tag-prefixed
/// key hex — the ops wire form, ADR-020 §Decision 2).
fn key_hex(key: &Key) -> String {
    let bytes = key.as_bytes();
    let mut hex = String::with_capacity(bytes.len() * 2);
    for byte in &bytes {
        hex.push(char::from_digit((byte >> 4) as u32, 16).unwrap_or('0'));
        hex.push(char::from_digit((byte & 0xf) as u32, 16).unwrap_or('0'));
    }
    hex
}

/// The byte-slice's hex form (the digest comparisons).
fn hex_of(bytes: &[u8]) -> String {
    let mut hex = String::with_capacity(bytes.len() * 2);
    for byte in bytes {
        hex.push(char::from_digit((byte >> 4) as u32, 16).unwrap_or('0'));
        hex.push(char::from_digit((byte & 0xf) as u32, 16).unwrap_or('0'));
    }
    hex
}

/// The identity requirement — the adapter never synthesizes one (the
/// `anon` composition is the door's; lfs.md §1).
fn require_identity(identity: Option<&Identity>) -> Result<Identity, LfsAdapterError> {
    identity
        .cloned()
        .ok_or_else(|| LfsAdapterError::Forbidden("authentication required".to_owned()))
}

/// The pre-keyed put's machinery — the put handler's contract minus
/// the channel, replicated in-process: the facade put consumes a
/// bounded pipe whose feed is the caller body's one read; the sha256
/// pass rides the same read; a claim the derivation contradicts
/// kills the put pre-EOF (the kill's ordering — flag armed ahead of
/// the sender's drop — matches the put handler's pipe writer). The
/// facade arm is `put_as(GitObjectSha256, Some(len), …)` — the
/// identity family's derivation is the raw content's hash (the oid,
/// byte-exact), so the returned key equals the formed key on an
/// honest body and the cross-check is the same one-stream pass.
async fn pre_keyed_put<R>(
    store: Arc<BlobStore>,
    expected: &Key,
    oid: &str,
    size: u64,
    body: R,
) -> Result<(Key, Pin), LfsAdapterError>
where
    R: AsyncRead + Unpin + Send,
{
    const PIPE_BOUND: usize = 64;
    const READ_CHUNK: usize = 16 * 1024;

    enum StreamEnd {
        Eof,
        BodyFailed(tokio::io::Error),
        PutEnded,
    }

    let fail = Arc::new(AtomicBool::new(false));
    let (tx, rx) = mpsc::channel::<Arc<Vec<u8>>>(PIPE_BOUND);
    let put = {
        let store = Arc::clone(&store);
        let fail = Arc::clone(&fail);
        tokio::spawn(async move {
            store
                .put_as(
                    alkblobs::Algorithm::GitObjectSha256,
                    Some(size),
                    PutPipe {
                        rx,
                        fail,
                        buffer: None,
                        offset: 0,
                    },
                )
                .await
        })
    };

    let mut body = body;
    let mut hasher = Sha256::new();
    let mut window = vec![0u8; READ_CHUNK];
    let end = loop {
        if put.is_finished() {
            break StreamEnd::PutEnded;
        }
        match body.read(&mut window).await {
            Ok(0) => break StreamEnd::Eof,
            Ok(n) => {
                hasher.update(&window[..n]);
                if tx.send(Arc::new(window[..n].to_vec())).await.is_err() {
                    break StreamEnd::PutEnded;
                }
            }
            Err(e) => break StreamEnd::BodyFailed(e),
        }
    };

    match end {
        StreamEnd::Eof => {
            let derived = hex_of(&hasher.finalize());
            if derived != oid {
                fail.store(true, Ordering::Release);
                drop(tx);
                let _ = put.await;
                return Err(LfsAdapterError::MislabeledBody(format!(
                    "no stored content, no held pin — {oid}"
                )));
            }
            drop(tx);
            match put.await {
                Ok(Ok(landing)) => {
                    if landing.0 != *expected {
                        return Err(LfsAdapterError::MislabeledBody(oid.to_owned()));
                    }
                    Ok(landing)
                }
                Ok(Err(e)) => Err(LfsAdapterError::Storage(e.to_string())),
                Err(e) => Err(LfsAdapterError::Storage(format!(
                    "the put's store task failed: {e}"
                ))),
            }
        }
        StreamEnd::BodyFailed(e) => {
            fail.store(true, Ordering::Release);
            drop(tx);
            let _ = put.await;
            Err(LfsAdapterError::Storage(format!(
                "the upload body failed mid-stream: {e}"
            )))
        }
        StreamEnd::PutEnded => match put.await {
            Ok(Ok(_)) => Err(LfsAdapterError::Storage(
                "the put's store arm completed against a torn stream".to_owned(),
            )),
            Ok(Err(e)) => Err(LfsAdapterError::Storage(e.to_string())),
            Err(e) => Err(LfsAdapterError::Storage(format!(
                "the put's store task failed: {e}"
            ))),
        },
    }
}

/// The put pipe's reader half (the put handler's `PutPipeReader`
/// shape): the facade put consumes the tee's feed; the kill flag
/// arms the next read into an error — the facade's stream-error path
/// (no commit, zero residue).
struct PutPipe {
    rx: mpsc::Receiver<Arc<Vec<u8>>>,
    fail: Arc<AtomicBool>,
    buffer: Option<Arc<Vec<u8>>>,
    offset: usize,
}

impl AsyncRead for PutPipe {
    fn poll_read(
        self: std::pin::Pin<&mut Self>,
        cx: &mut PollContext<'_>,
        buf: &mut ReadBuf<'_>,
    ) -> Poll<tokio::io::Result<()>> {
        let this = self.get_mut();
        loop {
            if this.fail.load(Ordering::Acquire) {
                return Poll::Ready(Err(tokio::io::Error::new(
                    tokio::io::ErrorKind::BrokenPipe,
                    "the put's feed failed pre-commit (the stream errored)",
                )));
            }
            if let Some(chunk) = &this.buffer {
                if this.offset < chunk.len() {
                    let n = (chunk.len() - this.offset).min(buf.remaining());
                    buf.put_slice(&chunk[this.offset..this.offset + n]);
                    this.offset += n;
                    return Poll::Ready(Ok(()));
                }
                this.buffer = None;
                this.offset = 0;
            }
            match this.rx.poll_recv(cx) {
                Poll::Ready(Some(chunk)) => {
                    this.buffer = Some(chunk);
                    this.offset = 0;
                }
                Poll::Ready(None) => return Poll::Ready(Ok(())),
                Poll::Pending => return Poll::Pending,
            }
        }
    }
}

/// The download's verified-window stream — an [`AsyncRead`] fed by a
/// pump that splits the requested range at the configured window
/// size and yields each window only after its mid-flight slice-digest
/// check passes (the module doc's byte-path section) — ADR-022 §2
/// as amended (the download byte path's verification posture). A failed
/// window check or a failed read surfaces as an io error at the
/// yield point (already-delivered windows stand; the caller aborts
/// and re-fetches). The pump ends at EOF; a zero-length whole read
/// is an empty stream (the entry's zero-length entry served).
pub struct LfsDownload {
    rx: mpsc::Receiver<tokio::io::Result<Vec<u8>>>,
    buffer: Option<Vec<u8>>,
    position: usize,
}

impl LfsDownload {
    /// Start the pump (the download call's tail). The channel bound
    /// is two windows — the store's pacing backpressures the
    /// consumer through the pipe (the same discipline as the put's
    /// bounded pipe, read-swapped). A whole-span download
    /// (`whole_span`) rides the end-to-end oid digest check.
    fn start(
        store: Arc<BlobStore>,
        key: Key,
        from: u64,
        until: u64,
        window_bytes: u64,
        whole_span: bool,
    ) -> Self {
        let (tx, rx) = mpsc::channel(2);
        tokio::spawn(verified_windows(
            store,
            key,
            from,
            until,
            window_bytes,
            whole_span,
            tx,
        ));
        LfsDownload {
            rx,
            buffer: None,
            position: 0,
        }
    }
}

impl AsyncRead for LfsDownload {
    fn poll_read(
        self: std::pin::Pin<&mut Self>,
        cx: &mut PollContext<'_>,
        buf: &mut ReadBuf<'_>,
    ) -> Poll<tokio::io::Result<()>> {
        let this = self.get_mut();
        loop {
            if let Some(chunk) = this.buffer.as_ref() {
                if this.position < chunk.len() {
                    let n = (chunk.len() - this.position).min(buf.remaining());
                    buf.put_slice(&chunk[this.position..this.position + n]);
                    this.position += n;
                    return Poll::Ready(Ok(()));
                }
                this.buffer = None;
                this.position = 0;
            }
            match this.rx.poll_recv(cx) {
                Poll::Ready(Some(Ok(chunk))) => {
                    this.buffer = Some(chunk);
                    this.position = 0;
                }
                Poll::Ready(Some(Err(e))) => return Poll::Ready(Err(e)),
                Poll::Ready(None) => return Poll::Ready(Ok(())),
                Poll::Pending => return Poll::Pending,
            }
        }
    }
}

/// The download pump: one `read_range` per window (clamped at the
/// requested end), the slice digest checked against the sha256 of
/// exactly the returned bytes mid-flight, then the window handed on.
/// A whole-span download (`whole_span`) additionally accumulates the
/// windows' bytes into one sha256 pass and checks the finished
/// digest against the key's derivation digest **before the last
/// window yields** — the identity family's digest is the raw
/// content's hash, so the comparison is the end-to-end oid
/// verification over the large tier's unverified reads (the
/// module doc's verification statement; the wave-9 gate's fix —
/// the per-window slice pair was checked against the same in-process
/// bytes and could not catch a corrupted entry).
async fn verified_windows(
    store: Arc<BlobStore>,
    key: Key,
    from: u64,
    until: u64,
    window_bytes: u64,
    whole_span: bool,
    tx: mpsc::Sender<tokio::io::Result<Vec<u8>>>,
) {
    let key_bytes = key.key_bytes();
    let expected = whole_span.then(|| key.digest().bytes().to_vec());
    let mut hasher = whole_span.then(Sha256::new);
    let mut offset = from;
    while offset < until {
        let window_len = window_bytes.min(until - offset);
        let last = offset + window_len == until;
        match store
            .read_range(key_bytes.as_slice(), offset, window_len)
            .await
        {
            Ok((slice, digest)) => {
                let hash = Sha256::digest(&slice);
                if slice.len() as u64 != window_len || hex_of(hash.as_slice()) != hex_of(&digest) {
                    let _ = tx
                        .send(Err(tokio::io::Error::other(
                            "the verified window's slice digest mismatched mid-flight",
                        )))
                        .await;
                    return;
                }
                if let Some(h) = hasher.as_mut() {
                    h.update(&slice);
                }
                if last && whole_span {
                    if let (Some(h), Some(expected_bytes)) = (hasher.as_ref(), expected.as_deref())
                    {
                        let digest_now = h.clone().finalize();
                        if digest_now.as_slice() != expected_bytes {
                            let _ = tx
                                .send(Err(tokio::io::Error::other(
                                    "the served media's whole-content digest mismatched the oid",
                                )))
                                .await;
                            return;
                        }
                    }
                }
                offset += window_len;
                if tx.send(Ok(slice)).await.is_err() {
                    return;
                }
            }
            Err(e) => {
                let _ = tx
                    .send(Err(tokio::io::Error::other(format!(
                        "the media read failed: {e}"
                    ))))
                    .await;
                return;
            }
        }
    }
}

#[cfg(test)]
mod battery {
    use std::collections::HashMap;

    use tokio::io::AsyncReadExt;

    use alkgit::{AccessAction, RepoRecord, Visibility};

    use super::*;
    use crate::registry::{CachedRegistryProvider, RegistryRow};

    fn oid_of(content: &[u8]) -> String {
        let digest = Sha256::digest(content);
        hex_of(digest.as_slice())
    }

    fn identity(id: &str) -> Identity {
        Identity {
            id: id.to_owned(),
            scopes: Vec::new(),
            resources: HashMap::new(),
        }
    }

    /// An identity carrying a **static** `Identity.resources` claim on
    /// the ops resource — the shape the gateway-gap regression pins
    /// (admitted only if the wired provider grants; the static path
    /// is never consulted through the adapter).
    fn identity_with_static_claim(id: &str) -> Identity {
        let mut resources = HashMap::new();
        resources.insert(
            "blob-namespace".to_owned(),
            vec!["read".to_owned(), "write".to_owned()],
        );
        Identity {
            id: id.to_owned(),
            scopes: Vec::new(),
            resources,
        }
    }

    fn record(
        repo_id: &str,
        visibility: Visibility,
        grants: &[(&str, &[AccessAction])],
    ) -> RepoRecord {
        RepoRecord {
            repo_id: repo_id.to_owned(),
            visibility,
            grants: grants
                .iter()
                .map(|(id, actions)| ((*id).to_owned(), actions.to_vec()))
                .collect(),
        }
    }

    fn registry_provider() -> Arc<CachedRegistryProvider> {
        let provider = Arc::new(CachedRegistryProvider::new());
        provider.install(vec![
            RegistryRow {
                record: record(
                    "owner/granted",
                    Visibility::Private,
                    &[("u-writer", &[AccessAction::Read, AccessAction::Write])],
                ),
                seq: 1,
            },
            RegistryRow {
                record: record(
                    "owner/readonly",
                    Visibility::Private,
                    &[
                        ("u-writer", &[AccessAction::Read, AccessAction::Write]),
                        ("u-reader", &[AccessAction::Read]),
                    ],
                ),
                seq: 2,
            },
            RegistryRow {
                record: record(
                    "owner/public",
                    Visibility::Public,
                    &[("u-writer", &[AccessAction::Read, AccessAction::Write])],
                ),
                seq: 3,
            },
            RegistryRow {
                record: record("owner/closed", Visibility::Private, &[]),
                seq: 4,
            },
        ]);
        provider
    }

    fn adapter(
        provider: Arc<dyn OwnershipProvider>,
        config: LfsAdapterConfig,
    ) -> (LfsAdapter, Arc<BlobStore>, Arc<crate::lfs::InMemoryLfsRows>) {
        let rows = Arc::new(crate::lfs::InMemoryLfsRows::new());
        let store = Arc::new(BlobStore::mem_only());
        let adapter = LfsAdapter::new(Arc::clone(&store), rows.clone(), provider, config).unwrap();
        (adapter, store, rows)
    }

    fn grant_provider() -> Arc<dyn OwnershipProvider> {
        Arc::new(LfsGrantProvider::new(registry_provider()))
    }

    fn object(oid: &str, size: u64) -> LfsBatchObject {
        LfsBatchObject {
            oid: oid.to_owned(),
            size,
        }
    }

    fn upload_request(objects: Vec<LfsBatchObject>) -> LfsBatchRequest {
        LfsBatchRequest {
            operation: LfsBatchOperation::Upload,
            hash_algo: None,
            objects,
        }
    }

    fn download_request(objects: Vec<LfsBatchObject>) -> LfsBatchRequest {
        LfsBatchRequest {
            operation: LfsBatchOperation::Download,
            hash_algo: None,
            objects,
        }
    }

    async fn upload_media(
        adapter: &LfsAdapter,
        repo_id: &str,
        content: &[u8],
    ) -> Result<(), LfsAdapterError> {
        adapter
            .upload(
                Some(&identity("u-writer")),
                repo_id,
                &oid_of(content),
                Some(content.len() as u64),
                content,
            )
            .await
    }

    async fn read_all(mut download: LfsDownload) -> Vec<u8> {
        let mut out = Vec::new();
        download
            .read_to_end(&mut out)
            .await
            .expect("the download reads");
        out
    }

    async fn download_whole(adapter: &LfsAdapter, repo_id: &str, oid: &str) -> Vec<u8> {
        read_all(
            adapter
                .download(Some(&identity("u-writer")), repo_id, oid, None)
                .await
                .unwrap(),
        )
        .await
    }

    /// The writer identity's batch upload: the two absent objects
    /// answer with upload mint inputs (no URLs — the inputs are the
    /// repo/oid pair), the rows are asserted at check time, the byte
    /// path stores exact bytes, the re-push answers no actions, the
    /// dedup re-put is idempotent, and the custody consumes the
    /// minted mapping at confirm.
    #[tokio::test]
    async fn the_upload_batch_and_the_pre_keyed_put() {
        let (adapter, store, rows) = adapter(grant_provider(), LfsAdapterConfig::default());
        let writer = identity("u-writer");
        let content_a = b"the first LFS media\n".to_vec();
        let content_b = b"the second LFS media\n".to_vec();
        let (oid_a, oid_b) = (oid_of(&content_a), oid_of(&content_b));

        let answer = adapter
            .check(
                Some(&writer),
                "owner/granted",
                upload_request(vec![object(&oid_a, content_a.len() as u64)]),
            )
            .await
            .unwrap();
        assert_eq!(answer.transfer, "basic");
        assert_eq!(answer.hash_algo, "sha256");
        let first = &answer.objects[0];
        assert_eq!(first.oid, oid_a);
        assert!(first.authenticated);
        assert!(first.error.is_none());
        let mint = first
            .actions
            .as_ref()
            .and_then(|a| a.upload.as_ref())
            .expect("the upload mint input");
        assert_eq!(mint.repo_id, "owner/granted");
        assert_eq!(mint.oid, oid_a);
        assert!(first
            .actions
            .as_ref()
            .and_then(|a| a.download.as_ref())
            .is_none());
        // the row asserted at check time (the moment the server learns the pair)
        let asserted = rows.repo_rows("owner/granted").await.unwrap();
        assert_eq!(
            asserted,
            vec![crate::lfs::LfsRow {
                repo_id: "owner/granted".to_owned(),
                oid: oid_a.clone(),
                size: content_a.len() as u64,
                refcount: 1,
            }]
        );

        // the byte path: exact bytes in, exact bytes out.
        upload_media(&adapter, "owner/granted", &content_a)
            .await
            .unwrap();
        assert_eq!(
            download_whole(&adapter, "owner/granted", &oid_a).await,
            content_a
        );
        let stat = rows.repo_rows("owner/granted").await.unwrap();
        assert_eq!(
            stat.iter().find(|r| r.oid == oid_a).unwrap().size,
            content_a.len() as u64,
            "the upload refreshed the assertion"
        );

        // the re-push: the present object answers no actions (the
        // idempotent posture) — and the batch with both objects
        // still mints the absent one.
        let answer = adapter
            .check(
                Some(&writer),
                "owner/granted",
                upload_request(vec![
                    object(&oid_a, content_a.len() as u64),
                    object(&oid_b, content_b.len() as u64),
                ]),
            )
            .await
            .unwrap();
        assert!(
            answer.objects[0].actions.is_none(),
            "the present object: nothing to do"
        );
        assert!(answer.objects[0].error.is_none());
        assert!(
            answer.objects[1]
                .actions
                .as_ref()
                .and_then(|a| a.upload.as_ref())
                .is_some(),
            "the absent object still offers the upload mint"
        );

        // the dedup re-put lands (the CAS-idempotent posture).
        upload_media(&adapter, "owner/granted", &content_a)
            .await
            .unwrap();

        // the confirm retired the minted mapping (the custody consumed;
        // the entry stands, protected by the registered rows source).
        assert_eq!(adapter.ops.tokens().len(), 0);
        let key = crate::liveness::lfs_media_key(&oid_a).unwrap();
        assert!(
            store
                .stat(key.key_bytes().as_slice())
                .await
                .unwrap()
                .is_some(),
            "the committed content stands post-confirm"
        );
    }

    /// The download batch's shapes: the present object answers the
    /// download mint input, the absent object answers the 404 error
    /// object, the size-disagreeing request answers the 422, and the
    /// anonymous-public read composes through the door's synthesized
    /// `anon` identity (read-only on the public repo).
    #[tokio::test]
    async fn the_download_batch_shapes() {
        let (adapter, _store, _rows) = adapter(grant_provider(), LfsAdapterConfig::default());
        let content = b"the downloadable media\n".to_vec();
        let oid = oid_of(&content);
        upload_media(&adapter, "owner/public", &content)
            .await
            .unwrap();

        let answer = adapter
            .check(
                Some(&identity("anon")),
                "owner/public",
                download_request(vec![
                    object(&oid, content.len() as u64),
                    object(&oid_of(b"never stored"), 10),
                ]),
            )
            .await
            .unwrap();
        let present = &answer.objects[0];
        assert!(present.authenticated);
        assert!(present.error.is_none());
        let mint = present
            .actions
            .as_ref()
            .and_then(|a| a.download.as_ref())
            .expect("the download mint input");
        assert_eq!(mint.repo_id, "owner/public");
        assert_eq!(mint.oid, oid);

        let absent = &answer.objects[1];
        assert_eq!(absent.error.as_ref().unwrap().code, 404);
        assert!(absent.actions.is_none());

        // the anonymous upload batch is refused (push is always
        // authenticated) and the anonymous read of a private repo is
        // refused (unknown ≡ unauthorized).
        assert!(matches!(
            adapter
                .check(
                    Some(&identity("anon")),
                    "owner/public",
                    upload_request(vec![object(&oid, 10)])
                )
                .await,
            Err(LfsAdapterError::Forbidden(_))
        ));
        assert!(matches!(
            adapter
                .check(
                    Some(&identity("anon")),
                    "owner/closed",
                    download_request(vec![object(&oid, 10)])
                )
                .await,
            Err(LfsAdapterError::Forbidden(_))
        ));

        // the size disagreement (the pointer's size contradicts the
        // stored content's length) answers the 422 error object.
        let answer = adapter
            .check(
                Some(&identity("u-writer")),
                "owner/granted",
                download_request(vec![object(&oid, content.len() as u64 + 5)]),
            )
            .await
            .unwrap();
        assert_eq!(answer.objects[0].error.as_ref().unwrap().code, 422);
        assert_eq!(
            answer.objects[0].error.as_ref().unwrap().message,
            "size disagreement"
        );
    }

    /// The pinned protocol gates: `hash_algo` serves sha256 only
    /// (the 409 fact) and the size-less upload offer is refused.
    #[tokio::test]
    async fn the_hash_algo_gate_and_the_sizeless_refusal() {
        let (adapter, _store, rows) = adapter(grant_provider(), LfsAdapterConfig::default());
        let writer = identity("u-writer");
        let oid = oid_of(b"gate media");

        let mut request = upload_request(vec![object(&oid, 10)]);
        request.hash_algo = Some("sha1".to_owned());
        assert!(
            matches!(
                adapter.check(Some(&writer), "owner/granted", request).await,
                Err(LfsAdapterError::HashAlgo(algo)) if algo == "sha1"
            ),
            "a non-sha256 batch is the pinned 409"
        );
        assert!(
            rows.repo_rows("owner/granted").await.unwrap().is_empty(),
            "the refused batch asserted nothing"
        );

        // the default algorithm (`None`) and the explicit sha256 pass;
        // their checks assert the pair (the row assertion's own contract).
        let mut request = upload_request(vec![object(&oid, 10)]);
        request.hash_algo = Some("sha256".to_owned());
        adapter
            .check(Some(&writer), "owner/granted", request)
            .await
            .unwrap();
        adapter
            .check(
                Some(&writer),
                "owner/granted",
                upload_request(vec![object(&oid, 10)]),
            )
            .await
            .unwrap();
        assert_eq!(rows.repo_rows("owner/granted").await.unwrap().len(), 1);

        assert!(
            matches!(
                adapter
                    .upload(Some(&writer), "owner/granted", &oid, None, &b"media"[..])
                    .await,
                Err(LfsAdapterError::SizelessOffer)
            ),
            "the size-less offer is a protocol error"
        );
        assert_eq!(
            rows.repo_rows("owner/granted").await.unwrap().len(),
            1,
            "the size-less refusal asserted nothing"
        );
    }

    /// The mislabeled body: the put's contract — the claim
    /// contradicted by the derivation kills the put pre-EOF (zero
    /// residue: no stored content, no held pin), the row assertion
    /// stands, and a corrected re-upload lands.
    #[tokio::test]
    async fn the_mislabeled_body_refusal() {
        let (adapter, store, rows) = adapter(grant_provider(), LfsAdapterConfig::default());
        let claimed = oid_of(b"the honest content");
        let honest = b"the honest content".to_vec();

        upload_media(&adapter, "owner/granted", &honest)
            .await
            .unwrap();
        let answer = adapter
            .download(Some(&identity("u-writer")), "owner/granted", &claimed, None)
            .await
            .unwrap();
        assert_eq!(read_all(answer).await, honest);

        // a body whose sha256 is not the claimed oid — the put kills
        // pre-EOF and nothing stores.
        let bad = b"the dishonest content".to_vec();
        adapter
            .upload(
                Some(&identity("u-writer")),
                "owner/granted",
                &claimed,
                Some(bad.len() as u64),
                bad.as_slice(),
            )
            .await
            .unwrap_err();

        let key = crate::liveness::lfs_media_key(&claimed).unwrap();
        let key_borrow = key.key_bytes();
        assert_eq!(
            store.stat(&key_borrow).await.unwrap().unwrap().len,
            honest.len() as u64,
            "the honest entry is the committed one — nothing mislabeled landed"
        );
        let list = store.list().await.unwrap();
        assert_eq!(list.len(), 1, "the kill left zero residue");
        assert_eq!(
            rows.repo_rows("owner/granted").await.unwrap()[0].oid,
            claimed
        );
        assert_eq!(
            adapter.ops.tokens().len(),
            0,
            "no held pin, no mapping left"
        );

        // a short body against the offer's len — the facade's own
        // known-length refusal (zero residue), the put contract's
        // inherited posture.
        adapter
            .upload(
                Some(&identity("u-writer")),
                "owner/granted",
                &oid_of(&bad),
                Some(bad.len() as u64 + 10),
                &bad[..bad.len() - 5],
            )
            .await
            .unwrap_err();
        assert_eq!(store.list().await.unwrap().len(), 1);
    }

    /// The ranged download: windows split at the configured boundary,
    /// each verified mid-flight, byte-exact over the whole and every
    /// slice; the past-EOF offset is the not-satisfiable refusal and
    /// the whole download honors the object-size ceiling.
    #[tokio::test]
    async fn the_ranged_download_splits_and_verifies() {
        let config = LfsAdapterConfig {
            window_bytes: 1024,
            max_object_bytes: 8 * 1024,
        };
        let (adapter, _store, _rows) = adapter(grant_provider(), config);
        let mut content = Vec::new();
        for i in 0..100u32 {
            content.extend_from_slice(format!("{i:064x}").as_bytes());
        }
        let oid = oid_of(&content);
        upload_media(&adapter, "owner/granted", &content)
            .await
            .unwrap();

        // whole download (over the window boundary — the adapter's
        // window arithmetic; byte-exact across the verified windows).
        assert_eq!(
            download_whole(&adapter, "owner/granted", &oid).await,
            content
        );

        // a slice, byte-exact.
        let slice = adapter
            .download(
                Some(&identity("u-writer")),
                "owner/granted",
                &oid,
                Some(LfsRange {
                    offset: 500,
                    len: Some(1000),
                }),
            )
            .await
            .unwrap();
        assert_eq!(read_all(slice).await, content[500..1500]);

        // an open-ended slice (to the end).
        let tail = adapter
            .download(
                Some(&identity("u-writer")),
                "owner/granted",
                &oid,
                Some(LfsRange {
                    offset: content.len() as u64 - 100,
                    len: None,
                }),
            )
            .await
            .unwrap();
        assert_eq!(read_all(tail).await, content[content.len() - 100..]);

        // past-EOF offset: the not-satisfiable refusal.
        assert!(matches!(
            adapter
                .download(
                    Some(&identity("u-writer")),
                    "owner/granted",
                    &oid,
                    Some(LfsRange {
                        offset: content.len() as u64 + 1,
                        len: None,
                    })
                )
                .await,
            Err(LfsAdapterError::NotSatisfiable)
        ));

        // a whole download over the ceiling is refused; the
        // full-span ranged form of the same object is the same span
        // and refuses too (the ceiling counts the resolved span, not
        // the request's header form); a partial span under the
        // ceiling rides the window budget.
        let big = vec![0u8; 10 * 1024];
        let big_oid = oid_of(&big);
        upload_media(&adapter, "owner/granted", &big).await.unwrap();
        assert!(matches!(
            adapter
                .download(Some(&identity("u-writer")), "owner/granted", &big_oid, None)
                .await,
            Err(LfsAdapterError::TooLarge(_))
        ));
        assert!(matches!(
            adapter
                .download(
                    Some(&identity("u-writer")),
                    "owner/granted",
                    &big_oid,
                    Some(LfsRange {
                        offset: 0,
                        len: Some(10 * 1024),
                    }),
                )
                .await,
            Err(LfsAdapterError::TooLarge(_))
        ));
        let ranged = adapter
            .download(
                Some(&identity("u-writer")),
                "owner/granted",
                &big_oid,
                Some(LfsRange {
                    offset: 0,
                    len: Some(4 * 1024),
                }),
            )
            .await
            .unwrap();
        assert_eq!(read_all(ranged).await, big[..4 * 1024]);

        // the reader identity rides the read grant; the unknown oid
        // is the absent-entry refusal.
        assert!(matches!(
            adapter
                .download(
                    Some(&identity("u-writer")),
                    "owner/granted",
                    &oid_of(b"absent"),
                    None
                )
                .await,
            Err(LfsAdapterError::Storage(_))
        ));
    }

    /// The wired-provider posture (the POC-L gateway-gap regression,
    /// re-landed): the static `Identity.resources` fallback is never
    /// consulted through the adapter — a claim-carrying identity is
    /// admitted only through the wired provider's live grants; the
    /// read-only grantee downloads but never uploads; the grant
    /// monotonicity (manage ⊃ write ⊃ read) and the unknown-repo
    /// denial ride the same `authorize` policy the git surface
    /// composes.
    #[tokio::test]
    async fn the_wired_provider_posture() {
        let provider = grant_provider();
        let (adapter, _store, _rows) = adapter(provider.clone(), LfsAdapterConfig::default());
        let content = b"the gated media\n".to_vec();
        let oid = oid_of(&content);

        // no identity: the adapter never synthesizes one.
        assert!(matches!(
            adapter
                .check(
                    None,
                    "owner/granted",
                    download_request(vec![object(&oid, 10)])
                )
                .await,
            Err(LfsAdapterError::Forbidden(_))
        ));

        // the unrecognized identity: refused everywhere, regardless
        // of a static resources claim (the fallback is unreachable —
        // the contexts always wire the provider).
        for identity in [identity("u-ghost"), identity_with_static_claim("u-ghost")] {
            assert!(matches!(
                adapter
                    .check(
                        Some(&identity),
                        "owner/closed",
                        download_request(vec![object(&oid, 10)])
                    )
                    .await,
                Err(LfsAdapterError::Forbidden(_))
            ));
            assert!(matches!(
                adapter
                    .upload(Some(&identity), "owner/closed", &oid, Some(10), &b"x"[..])
                    .await,
                Err(LfsAdapterError::Forbidden(_))
            ));
            assert!(matches!(
                adapter
                    .download(Some(&identity), "owner/closed", &oid, None)
                    .await,
                Err(LfsAdapterError::Forbidden(_))
            ));
        }

        // the static claim does not admit either: the same
        // claim-carrying identity through the granting repo — the
        // live grant table (the wired provider's answer) decides.
        let claimed = identity_with_static_claim("u-nobody");
        assert!(
            matches!(
                adapter
                    .check(
                        Some(&claimed),
                        "owner/granted",
                        download_request(vec![object(&oid, 10)])
                    )
                    .await,
                Err(LfsAdapterError::Forbidden(_))
            ),
            "a resources claim rides nothing when the provider refuses"
        );

        // the read-only grantee: read granted, write refused — the
        // batch action mapping (lfs.md §1: read/write ↔
        // download/upload) enforced through the wired provider.
        let reader = identity("u-reader");
        upload_media(&adapter, "owner/readonly", &content)
            .await
            .unwrap();
        let answer = adapter
            .check(
                Some(&reader),
                "owner/readonly",
                download_request(vec![object(&oid, content.len() as u64)]),
            )
            .await
            .unwrap();
        assert!(answer.objects[0].error.is_none());
        assert!(answer.objects[0]
            .actions
            .as_ref()
            .unwrap()
            .download
            .is_some());
        assert!(matches!(
            adapter
                .check(
                    Some(&reader),
                    "owner/readonly",
                    upload_request(vec![object(&oid, 1)])
                )
                .await,
            Err(LfsAdapterError::Forbidden(_))
        ));
        let answer = adapter
            .download(Some(&reader), "owner/readonly", &oid, None)
            .await
            .unwrap();
        assert_eq!(read_all(answer).await, content);

        // the provider's own rows: the public read (anon rides the
        // visibility, through the batch's provider), the denied
        // public write (push is always authenticated); the manage ⊃
        // write ⊃ read monotonicity, the unrecognized identity and
        // the unknown repo on a fresh provider.
        let anon = identity("anon");
        assert!(provider.owns(&anon, "blob-namespace", "owner/public", "read"));
        assert!(!provider.owns(&anon, "blob-namespace", "owner/public", "write"));
        let manages = record(
            "repo/m",
            Visibility::Private,
            &[("u-m", &[AccessAction::Manage])],
        );
        assert!(!manages.grants.is_empty());
        let provider2 = Arc::new(CachedRegistryProvider::new());
        provider2.install(vec![RegistryRow {
            record: manages,
            seq: 1,
        }]);
        let p: Arc<dyn OwnershipProvider> = Arc::new(LfsGrantProvider::new(provider2));
        // the ops resource actions are read/write — a stored manage
        // grant satisfies both (the authorize monotonicity); unknown
        // action strings deny; the unrecognized identity and the
        // unknown repo deny.
        assert!(p.owns(&identity("u-m"), "blob-namespace", "repo/m", "read"));
        assert!(p.owns(&identity("u-m"), "blob-namespace", "repo/m", "write"));
        assert!(!p.owns(&identity("u-x"), "blob-namespace", "repo/m", "read"));
        assert!(!p.owns(&identity("u-x"), "blob-namespace", "repo/m", "write"));
        assert!(!p.owns(&identity("u-m"), "blob-namespace", "repo/m", "manage"));
        assert!(!p.owns(&identity("u-m"), "blob-namespace", "repo/unknown", "read"));
        assert!(!p.owns(&identity("u-m"), "other-type", "repo/m", "read"));
        assert_eq!(
            p.owned_resources(&identity("u-m"), "blob-namespace"),
            vec!["repo/m".to_owned()]
        );
    }

    /// The row assertion at check time, composed end-to-end: both
    /// arms assert, the malformed oid asserts nothing, the
    /// cross-reference composition bumps the rows' refcounts (the
    /// same oid referenced by two repos — the re-protection contract's
    /// data riding the adapter's assertions), and the re-assert is
    /// one row (the refcount-aware upsert).
    #[tokio::test]
    async fn the_rows_are_asserted_at_check_time() {
        let (adapter, _store, _rows) = adapter(grant_provider(), LfsAdapterConfig::default());
        let a = oid_of(b"shared media a");
        let malformed = "not-an-oid";

        // upload arm, absent objects: asserted.
        adapter
            .check(
                Some(&identity("u-writer")),
                "owner/granted",
                upload_request(vec![object(&a, 15)]),
            )
            .await
            .unwrap();
        // download arm, absent objects: asserted too (the same posture).
        adapter
            .check(
                Some(&identity("u-writer")),
                "owner/readonly",
                download_request(vec![object(&a, 15), object(malformed, 15)]),
            )
            .await
            .unwrap();

        let all = _rows.rows().await.unwrap();
        assert_eq!(all.len(), 2, "the malformed oid asserted nothing");
        for row in &all {
            assert_eq!(row.refcount, 2, "two repos — one key, two rows: {row:?}");
        }

        // the re-assert refreshes the size; the refcount unmoved.
        adapter
            .check(
                Some(&identity("u-writer")),
                "owner/granted",
                upload_request(vec![object(&a, 15)]),
            )
            .await
            .unwrap();
        let all = _rows.rows().await.unwrap();
        let mine = all.iter().filter(|r| r.repo_id == "owner/granted").count();
        assert_eq!(mine, 1);
    }
}

/// The pg fleet battery (F-S5's test half, the wave-10 booking): the
/// live-pg fleet assembly row — a kv-postgres **fleet** store under
/// the harness container, the adapter over it riding the live pg rows
/// twin — proving the fleet custody arm's confirm-immediately posture
/// consumes no admission slot across repeated uploads. The other
/// battery adapters are all `mem_only()` (the fleet owner resolves
/// `None`; only the custodial arm ever ran) — this is the fleet arm's
/// coverage: the fleet owner resolves (the arm's precondition), every
/// upload of fresh media succeeds (no `MintRefused`), the token
/// registry ends **empty** after every upload (each mint retired at
/// its immediate confirm — a dropped-unconfirmed token would hold its
/// admission slot to process end, the wave-9 gate's leak), the mint
/// boundary stays live per upload (an explicitly-held token persists
/// across an upload whose own mint lands and retires around it —
/// ADR-022 §5's hostile-loop posture, zero slot leak), and the pair
/// rows assert identically to the custodial arm's. Live-pg only;
/// clean skip when the harness env is absent (the `live_pg!`
/// convention).
#[cfg(all(test, feature = "postgres"))]
mod pg_fleet_tests {
    use std::collections::{BTreeMap, HashMap};
    use std::io::Cursor;
    use std::time::Duration;

    use alkgit::{AccessAction, RepoRecord, Visibility};

    use super::*;
    use crate::postgres::live_harness::{drop_schema, live_pg, unique_schema};
    use crate::registry::{CachedRegistryProvider, RegistryRow};

    const REPO: &str = "owner/fleet-granted";

    /// One fresh media body — the seed's little-endian form repeated,
    /// so every seed carries its own distinct header (each upload a
    /// new oid; the CAS re-put idempotence never degenerates the
    /// mints).
    fn media_of(seed: u64, len: usize) -> Vec<u8> {
        let pattern = seed.to_le_bytes();
        let mut out = Vec::with_capacity(len);
        while out.len() < len {
            out.extend_from_slice(&pattern);
        }
        out.truncate(len);
        out
    }

    fn oid_of(content: &[u8]) -> String {
        hex_of(Sha256::digest(content).as_slice())
    }

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

    fn grant_provider() -> Arc<dyn OwnershipProvider> {
        let provider = Arc::new(CachedRegistryProvider::new());
        provider.install(vec![RegistryRow {
            record: RepoRecord {
                repo_id: REPO.to_owned(),
                visibility: Visibility::Private,
                grants: BTreeMap::from([(
                    "u-writer".to_owned(),
                    vec![AccessAction::Read, AccessAction::Write],
                )]),
            },
            seq: 1,
        }]);
        Arc::new(LfsGrantProvider::new(provider))
    }

    #[tokio::test(flavor = "multi_thread")]
    async fn the_fleet_handover_retires_every_admission_slot_across_repeated_uploads() {
        let dsn = live_pg!();
        let schema = unique_schema("lfs-fleet");
        let blobs_schema = format!("{schema}_blobs");

        let rows = Arc::new(crate::lfs::PgLfsRows::new(
            crate::postgres::CoTenantPg::open(&dsn, &schema)
                .await
                .unwrap(),
        ));
        let fleet_store = Arc::new(
            alkblobs::Store::kv_only_postgres_fleet(
                &dsn,
                alkblobs::PgOpts {
                    schema: blobs_schema.clone(),
                    ..alkblobs::PgOpts::default()
                },
                1024 * 1024,
                Duration::ZERO,
                Duration::from_secs(60),
            )
            .await
            .unwrap(),
        );
        let adapter = LfsAdapter::new(
            Arc::clone(&fleet_store),
            rows.clone(),
            grant_provider(),
            LfsAdapterConfig {
                window_bytes: 128 * 1024,
                ..LfsAdapterConfig::default()
            },
        )
        .unwrap();
        let identity = identity();

        // The fleet arm's precondition: the store's fleet declaration
        // resolves the serving instance's id (the mem battery never
        // reached this arm — every mint below is a fleet mint through
        // `hand_over`'s `Some(fleet)` branch).
        let owner = adapter
            .ops
            .fleet_owner()
            .expect("the fleet store carries the serving instance's id");

        // Repeated uploads of fresh media: each rides the row
        // assertion, the pre-keyed put, then the fleet hand-over —
        // `mint_fleet` and the immediate confirm. None refuses at the
        // admission cap, and the registry ends empty after every one.
        const UPLOADS: usize = 12;
        for i in 0..UPLOADS as u64 {
            let body = media_of(i, 16 * 1024);
            let oid = oid_of(&body);
            adapter
                .upload(
                    Some(&identity),
                    REPO,
                    &oid,
                    Some(body.len() as u64),
                    Cursor::new(&body),
                )
                .await
                .expect("the fleet upload lands");
            let pairs = rows.repo_rows(REPO).await.unwrap();
            assert!(
                pairs.iter().any(|row| row.oid == oid),
                "upload {i}'s pair asserted its row: {pairs:?}"
            );
            assert_eq!(
                adapter.ops.tokens().len(),
                0,
                "upload {i}'s mint retired at the confirm — no leaked slot"
            );
        }
        assert_eq!(
            rows.repo_rows(REPO).await.unwrap().len(),
            UPLOADS,
            "one row per upload pair"
        );

        // The mint boundary's liveness: an explicitly-held token (a
        // fleet mint without the confirm — the custody the pin row's
        // TTL durably protects) counts its slot, and the next upload's
        // own mint lands and retires around it — the admission cap
        // stays live per upload with zero leak (the post-upload
        // registry carries exactly the held slot, nothing of the
        // upload's).
        let hold_body = media_of(0xff, 16 * 1024);
        let hold_key = lfs_media_key(&oid_of(&hold_body)).expect("the canonical key");
        let held = adapter
            .ops
            .mint_fleet(REPO, hold_key, &owner)
            .expect("the boundary mint");
        assert_eq!(adapter.ops.tokens().len(), 1, "the held token counts");
        let next = media_of(0xfffe, 16 * 1024);
        adapter
            .upload(
                Some(&identity),
                REPO,
                &oid_of(&next),
                Some(next.len() as u64),
                Cursor::new(&next),
            )
            .await
            .expect("the upload around the held token lands");
        assert_eq!(
            adapter.ops.tokens().len(),
            1,
            "the upload's mint retired around the held one — no leak"
        );
        assert!(
            adapter.ops.tokens().holds(&held),
            "the held token's slot persists (its durability is the pin row's TTL)"
        );

        drop_schema(&dsn, &schema).await;
        drop_schema(&dsn, &blobs_schema).await;
    }
}