alkgit-substrate 0.1.0

//! The w9 end-to-end rows: real `git` + real `git-lfs` (3.7.0, the
//! capture-session clients) driven against the composition — the
//! substrate assembly (the wave-8 entry's complete backend) plus the
//! LFS surface, over the two test doors: a real `git push` commits the
//! pointer-file blobs as ordinary git blobs through the wave-8 git
//! door; the media rides the LFS door's batch/transfer surface
//! ([`lfs_common`] — the POC-L reference door's shape; the alkhttp
//! `lfs` feature stays family infrastructure). The POC-L
//! `pocl-findings.md` gate inventory re-landed on the real code.
//!
//! Discipline: git/git-lfs are test clients only (the serving path
//! spawns nothing — the crate convention); the rows skip cleanly (with
//! a note) when the `git` or `git-lfs` binary is absent; the live-pg
//! row skips when the harness container is down (the
//! `ALKGIT_SUBSTRATE_PG_HOST` convention). The skip notes ride
//! libtest's capture — they print under `--nocapture` only (recorded
//! posture: the note's audience is a human running with `--nocapture`
//! or reading the source; converting the skip into an assertion is not
//! possible and a second mechanism buys nothing). Rows are names +
//! `Contract stamp:` versions per `substrate/tests/common` §The
//! row/stamp discipline.
//!
//! The binding matrix is pinned in-task (the task's Notes): the door
//! binds the **DirectOps** shape — the adapter *is* the substrate's
//! in-process ops surface ([`LfsAdapter`] invokes the op registry
//! directly). The loopback binding belongs to the fleet/door byte-path
//! (alkblobs' own wire ops — "the wire ops remain the fleet/remote
//! consumer's form; the DirectOps-vs-loopback binding choice is
//! door-side"); riding it here would exercise alkblobs' loopback
//! machinery, not a substrate-side contract — the battery + liveness
//! rows carry that side. POC-L validated both bindings; this e2e's
//! breadth is the pinned call, flagged for the gate.

#![cfg(feature = "lfs")]

mod common;
mod lfs_common;

use lfs_common::{OUTSIDER_PASSWORD, WRITER_PASSWORD};

use std::net::SocketAddr;
use std::path::Path;
use std::sync::atomic::{AtomicU64, Ordering};
use std::time::Duration;

use alkgit_substrate::lfs_media_key;
use gix_hash::ObjectId;
use serde_json::{json, Value};
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::TcpStream;

use common::Harness;

/// The rows' repo id (one two-segment namespace; the route grammar is
/// a full path parse).
const REPO: &str = "owner/real-lfs";

/// The second repo of the cross-reference sweep row.
const REPO_B: &str = "owner/real-lfs-shared";

/// The row bases' unique temp-root suffix (per-process + per-row tag).
fn row_base(tag: &str) -> std::path::PathBuf {
    static COUNTER: AtomicU64 = AtomicU64::new(0);
    let n = COUNTER.fetch_add(1, Ordering::Relaxed);
    let dir =
        std::env::temp_dir().join(format!("alkgit-real-lfs-{tag}-{}-{n}", std::process::id()));
    let _ = std::fs::remove_dir_all(&dir);
    std::fs::create_dir_all(&dir).expect("the row's base creates");
    dir
}

/// The rows' git-lfs skip signal: both binaries resolved or the row
/// prints the note and its caller returns.
fn client_binaries() -> bool {
    if common::git_command().is_none() {
        eprintln!("skipping row: no `git` binary on PATH");
        return false;
    }
    if lfs_common::lfs_binary().is_none() {
        eprintln!("skipping row: no `git-lfs` binary on PATH");
        return false;
    }
    true
}

/// Real `git rev-parse` (hex, trimmed).
async fn rev_parse(cwd: &Path, what: &str) -> Option<String> {
    let run = common::git(Some(cwd), &["rev-parse", what]).await?;
    Some(run.stdout.trim().to_owned())
}

// ——— the raw HTTP client (the raw rows' shape; the POC raw_http.rs
// posture) ———

/// One raw http request's reply: the status, the head's headers, the
/// Content-Length-framed body.
struct RawReply {
    status: u16,
    headers: Vec<(String, String)>,
    body: Vec<u8>,
}

impl RawReply {
    fn header(&self, name: &str) -> Option<String> {
        self.headers
            .iter()
            .rev()
            .find(|(candidate, _)| candidate.eq_ignore_ascii_case(name))
            .map(|(_, value)| value.clone())
    }

    fn json(&self) -> Value {
        serde_json::from_slice(&self.body).expect("the reply's body parses as json")
    }
}

/// One raw http/1.1 exchange (Content-Length-framed, single request —
/// the door serves keep-alive, the harness drops after the reply).
async fn raw_http(
    addr: SocketAddr,
    method: &str,
    path: &str,
    headers: &[(&str, &str)],
    body: &[u8],
) -> RawReply {
    let mut wire = Vec::new();
    wire.extend_from_slice(
        format!("{method} {path} HTTP/1.1\r\nHost: 127.0.0.1\r\nConnection: close\r\n").as_bytes(),
    );
    for (name, value) in headers {
        wire.extend_from_slice(format!("{name}: {value}\r\n").as_bytes());
    }
    wire.extend_from_slice(format!("Content-Length: {}\r\n\r\n", body.len()).as_bytes());
    wire.extend_from_slice(body);

    let work = async {
        let mut conn = TcpStream::connect(addr)
            .await
            .expect("the raw request connects");
        conn.write_all(&wire).await.expect("the raw request writes");
        let mut buffer = Vec::new();
        let mut chunk = [0u8; 8192];
        let head_end = loop {
            if let Some(position) = buffer.windows(4).position(|w| w == b"\r\n\r\n") {
                break position;
            }
            let read = conn.read(&mut chunk).await.expect("the raw reply reads");
            if read == 0 {
                panic!("the door closed before the reply head");
            }
            buffer.extend_from_slice(&chunk[..read]);
        };
        let head = String::from_utf8_lossy(&buffer[..head_end]).to_string();
        let mut lines = head.split("\r\n");
        let status: u16 = lines
            .next()
            .and_then(|line| line.split_whitespace().nth(1))
            .and_then(|token| token.parse().ok())
            .expect("the reply's status parses");
        let mut reply_headers = Vec::new();
        for line in lines {
            if let Some((name, value)) = line.split_once(':') {
                reply_headers.push((name.trim().to_owned(), value.trim().to_owned()));
            }
        }
        let content_length = reply_headers
            .iter()
            .find(|(name, _)| name.eq_ignore_ascii_case("content-length"))
            .and_then(|(_, value)| value.parse::<usize>().ok())
            .unwrap_or(0);
        let body_start = head_end + 4;
        while buffer.len() < body_start + content_length {
            let read = conn
                .read(&mut chunk)
                .await
                .expect("the raw reply's body reads");
            if read == 0 {
                break;
            }
            buffer.extend_from_slice(&chunk[..read]);
        }
        RawReply {
            status,
            headers: reply_headers,
            body: buffer[body_start..].to_vec(),
        }
    };
    tokio::time::timeout(Duration::from_secs(30), work)
        .await
        .expect("the raw exchange completes inside the deadline")
}

fn batch_body(operation: &str, objects: &[(&str, u64)], hash_algo: Option<&str>) -> Vec<u8> {
    let mut request = json!({
        "operation": operation,
        "transfers": ["basic"],
        "objects": objects
            .iter()
            .map(|(oid, size)| json!({ "oid": oid, "size": size }))
            .collect::<Vec<_>>(),
    });
    if let Some(hash_algo) = hash_algo {
        request["hash_algo"] = json!(hash_algo);
    }
    serde_json::to_vec(&request).expect("the batch body serializes")
}

/// RFC 4648 base64 (standard alphabet) — the raw rows' Basic encoder.
fn basic64(input: &str) -> String {
    const ALPHABET: &[u8; 64] = b"ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/";
    let mut output = String::new();
    for chunk in input.as_bytes().chunks(3) {
        let mut triple = [0u8; 3];
        triple[..chunk.len()].copy_from_slice(chunk);
        let packed = ((triple[0] as u32) << 16) | ((triple[1] as u32) << 8) | triple[2] as u32;
        for shift in [18u32, 12, 6, 0] {
            output.push(ALPHABET[((packed >> shift) & 0x3f) as usize] as char);
        }
        for _ in 0..(3 - chunk.len()) {
            output.pop();
        }
        for _ in 0..(3 - chunk.len()) {
            output.push('=');
        }
    }
    output
}

/// The batch's upload-action href + token, from a served response
/// object's JSON (the door-minted absolute URL and the per-action
/// `Authorization`).
fn action_of(object: &Value, action: &str) -> (String, Option<String>) {
    let actions = object
        .get("actions")
        .map(|a| a.get(action))
        .unwrap_or(None)
        .unwrap_or(&Value::Null)
        .clone();
    let href = actions
        .get("href")
        .and_then(Value::as_str)
        .expect("the action's href")
        .to_owned();
    let token = actions
        .get("header")
        .and_then(|header| header.get("Authorization"))
        .and_then(Value::as_str)
        .map(str::to_owned);
    (href, token)
}

/// The minted href's path (the raw transfer requests ride the door's
/// absolute refs verbatim).
fn href_path(href: &str) -> String {
    let position = href.find('/').expect("the href's host boundary");
    let after_scheme = href[position + 2..].to_owned();
    let position = after_scheme.find('/').expect("the href's path");
    after_scheme[position..].to_owned()
}

/// The round trip (the push/pull rows' shared body): the real `git
/// push` rides both doors — the media through the LFS door (the
/// pre-push hook's batch + PUT; the Basic batch → the minted
/// per-action token, the `lfsurl` configured first) and the
/// pointer-file blobs as ordinary git blobs through the git door; the
/// content lands under the canonical `[0x04] + oid` key byte-exact
/// with the reference row asserted; the wiped local store's real
/// `git lfs pull` restores exact bytes.
async fn push_pull_round_trip(
    harness: &Harness,
    git_door: SocketAddr,
    lfs: &lfs_common::LfsDoor,
    repo_id: &str,
    client: &lfs_common::LfsClient,
) {
    // The named remote (the real-user shape): the pre-push hook the
    // push runs resolves the LFS endpoint from `remote.origin.lfsurl`
    // (a raw-URL push's hook derives it from the git:// url instead —
    // the real client's endpoint rules).
    let url_git = format!("git://{git_door}/{repo_id}");
    common::git(Some(client.path()), &["remote", "add", "origin", &url_git])
        .await
        .expect("the row rides a resolved git binary")
        .expect_ok();
    let url = lfs.lfs_url(repo_id, Some(lfs_common::WRITER_PASSWORD));
    common::git(
        Some(client.path()),
        &["config", "remote.origin.lfsurl", &url],
    )
    .await
    .expect("the row rides a resolved git binary")
    .expect_ok();
    common::git(Some(client.path()), &["push", "origin", "main"])
        .await
        .expect("the row rides a resolved git binary")
        .expect_ok();

    // The pointer-file blob landed as an ordinary git blob through
    // the wave-8 git door: the committed blob's content is the pointer
    // text (the clean filter's act) and the blob is in the store
    // under its git oid (the working tree keeps the media bytes).
    let pointer = rev_parse(client.path(), "HEAD:media.bin")
        .await
        .expect("the pointer blob resolves");
    let pointer_blob = common::git(Some(client.path()), &["cat-file", "blob", "HEAD:media.bin"])
        .await
        .expect("the row rides a resolved git binary")
        .expect_ok();
    assert!(
        pointer_blob.stdout.starts_with("version https://git-lfs"),
        "the committed media.bin blob is the pointer text: {}",
        pointer_blob.stdout
    );
    let pointer_oid =
        ObjectId::from_hex(pointer.as_bytes()).expect("the pointer blob's oid parses");
    assert!(
        harness
            .assembly
            .front()
            .stat(&pointer_oid)
            .await
            .expect("the front stats")
            .is_some(),
        "the pointer-file blob landed under its git oid through the git door"
    );

    // The reference row asserted (repo_id, oid, size — refcount 1).
    let rows = harness
        .assembly
        .lfs_rows()
        .repo_rows(repo_id)
        .await
        .expect("the lfs rows read");
    assert_eq!(
        rows,
        vec![alkgit_substrate::LfsRow {
            repo_id: repo_id.to_owned(),
            oid: client.oid.clone(),
            size: client.media.len() as u64,
            refcount: 1,
        }],
        "the reference row asserted at check/upload time"
    );

    // The content under the canonical key: `[0x04] + oid`, byte-exact
    // (the tag byte is the media family; the digest half is the oid).
    let key = lfs_media_key(&client.oid).expect("the canonical key forms");
    let key_bytes = key.key_bytes();
    assert_eq!(key_bytes[0], 0x04, "the media family's tag byte");
    assert_eq!(
        hex::encode(&key_bytes[1..]),
        client.oid,
        "the digest half is the oid, byte-exact"
    );
    let meta = harness
        .assembly
        .blobs()
        .stat(key_bytes.as_slice())
        .await
        .expect("the store stats")
        .expect("the content landed under the canonical key");
    assert_eq!(meta.len, client.media.len() as u64, "the stored length");

    // The transfer rode the minted per-action token (the real client
    // applied the batch-minted `Authorization`; the door's PUT class
    // says so).
    let puts = lfs.requests_of("PUT", &format!("/{repo_id}/lfs/objects/{}", client.oid));
    assert_eq!(puts.len(), 1, "exactly one PUT rode the transfer route");
    assert_eq!(puts[0].status, 200, "the PUT was the accepted one");
    assert_eq!(
        puts[0].auth, "token",
        "the PUT's auth was the minted per-action token"
    );

    // Pull after wiping the client store: exact bytes restored.
    lfs_common::wipe_local_store(client);
    common::git(Some(client.path()), &["lfs", "pull", "origin"])
        .await
        .expect("the row rides a resolved git binary")
        .expect_ok();
    let restored = std::fs::read(client.dir.join("media.bin")).expect("the pull restored");
    assert_eq!(restored, client.media, "the pull restored exact bytes");
}

/// The full real-client round trip over both doors on one assembly
/// (the mem shape): the push (pointer blobs through the git door), the
/// media upload (batch + PUT through the LFS door, the minted
/// per-action token applied), the canonical-key/row asserts, and the
/// wiped-store pull restoring exact bytes.
///
/// Contract stamp: ADR-022 §1 (the batch/checking mapping; sha256
/// only; the row assertion at check time), §2 (the pre-keyed PUT; the
/// oid byte-exact the media family's digest), §4 (the pointer
/// lifecycle rows); lfs.md §1–3; doors.md §alkhttp `lfs` (the route +
/// auth facts the test door mirrors); ADR-013 §7–8/ADR-016 §2 (the
/// git door's push framing); ADR-011 (the grants the provider
/// composes); pocl-findings (the gate rows re-landed);
/// substrate/w9-e2e-real-lfs.
#[tokio::test(flavor = "multi_thread")]
async fn a_real_git_lfs_push_and_pull_round_trip_over_the_two_doors() {
    if !client_binaries() {
        return;
    }
    let harness = Harness::mem("lfs-round-trip", REPO).await;
    let git_door = common::spawn_daemon_door(&harness, common::DoorOptions::default()).await;
    let lfs = lfs_common::spawn_lfs_door(&harness).await;
    let base = row_base("round-trip");
    let client = lfs_common::init_lfs_client(
        &base,
        "origin",
        lfs_common::pseudo_bytes(0x51fc, 384 * 1024),
    )
    .await
    .expect("the lfs client inits");

    push_pull_round_trip(&harness, git_door.addr, &lfs, REPO, &client).await;

    // The batches' shapes: the push's upload batch carried the mint
    // input; the pull's download batch the download one.
    let batches = lfs.requests_of("POST", "/objects/batch");
    assert_eq!(batches.len(), 2, "one batch per direction");
    let shaped = |request: &lfs_common::LfsDoorRequest, action: &str| {
        request
            .batch_objects
            .as_ref()
            .expect("the batch's objects logged")[0]
            .get("actions")
            .and_then(|actions| actions.get(action))
            .is_some()
    };
    assert!(shaped(&batches[0], "upload"), "the push's minted upload");
    assert!(
        shaped(&batches[1], "download"),
        "the pull's minted download"
    );
    assert!(
        !shaped(&batches[1], "upload"),
        "the download batch minted no upload action"
    );

    git_door.accept.abort();
    std::fs::remove_dir_all(&base).ok();
}

/// The fresh consumer: `GIT_LFS_SKIP_SMUDGE=1 git clone` (over the
/// anonymous http git door) inherits `.gitattributes` but not the
/// filter config — `git lfs install --local` first, then `git lfs
/// pull` over the LFS door rides the anonymous-public batch (the
/// synthesized `anon` identity) and restores the exact bytes.
///
/// Contract stamp: doors.md §alkhttp `lfs` (the anonymous-public
/// synthesized `anon` identity; no header minted for it);
/// pocl-findings (the fresh-consumer gate: install --local's quirk;
/// the skip-smudge clone shape); lfs.md §1; ADR-011 §3 (the
/// anonymous public read the http git door carries);
/// substrate/w9-e2e-real-lfs.
#[tokio::test(flavor = "multi_thread")]
async fn a_fresh_consumer_clones_then_lfs_pulls_the_exact_media() {
    if !client_binaries() {
        return;
    }
    let harness = Harness::mem("lfs-fresh-consumer", REPO).await;
    let git_door = common::spawn_daemon_door(&harness, common::DoorOptions::default()).await;
    let http_door = common::spawn_http_door(&harness).await;
    let lfs = lfs_common::spawn_lfs_door(&harness).await;
    let base = row_base("fresh-consumer");
    let client = lfs_common::init_lfs_client(
        &base,
        "origin",
        lfs_common::pseudo_bytes(0x9ea5, 256 * 1024),
    )
    .await
    .expect("the lfs client inits");

    // The named remote (the real-user shape — the hook's endpoint
    // resolves from the lfsurl url), then the push rides both doors:
    // the media through the hook's batch + PUT (the LFS door), the
    // pointer blobs through the git door.
    let url = lfs.lfs_url(REPO, Some(lfs_common::WRITER_PASSWORD));
    common::git(
        Some(client.path()),
        &["remote", "add", "origin", &harness.git_url(git_door.addr)],
    )
    .await
    .expect("the row rides a resolved git binary")
    .expect_ok();
    common::git(
        Some(client.path()),
        &["config", "remote.origin.lfsurl", &url],
    )
    .await
    .expect("the row rides a resolved git binary")
    .expect_ok();
    common::git(Some(client.path()), &["push", "origin", "main"])
        .await
        .expect("the row rides a resolved git binary")
        .expect_ok();
    common::git(
        Some(client.path()),
        &["lfs", "push", "--object-id", "origin", &client.oid],
    )
    .await
    .expect("the row rides a resolved git binary")
    .expect_ok();

    // The fresh consumer: skip-smudge clone over the anonymous http
    // git door; the pointer rides the clone (the .gitattributes
    // inherited), the filter config does not.
    let dst = base.join("consumer");
    common::git_with(
        None,
        &[
            "-c",
            "protocol.version=2",
            "clone",
            "-q",
            &harness.http_url(http_door.addr),
            dst.to_str().expect("utf8"),
        ],
        &[("GIT_LFS_SKIP_SMUDGE", "1")],
    )
    .await
    .expect("the row rides a resolved git binary")
    .expect_ok();
    let skipped = std::fs::read(dst.join("media.bin")).expect("the pointer checkout");
    assert!(
        skipped.starts_with(b"version https://git-lfs"),
        "the skip-smudge checkout wrote the pointer, not the media"
    );

    common::git(Some(&dst), &["lfs", "install", "--local"])
        .await
        .expect("the row rides a resolved git binary")
        .expect_ok();
    let url = lfs.lfs_url(REPO, None);
    common::git(Some(&dst), &["config", "remote.origin.lfsurl", &url])
        .await
        .expect("the row rides a resolved git binary")
        .expect_ok();
    common::git(Some(&dst), &["lfs", "pull", "origin"])
        .await
        .expect("the row rides a resolved git binary")
        .expect_ok();
    let restored = std::fs::read(dst.join("media.bin")).expect("the consumer's media");
    assert_eq!(
        restored, client.media,
        "the fresh consumer restored exact bytes"
    );

    // The consumer's rides were the anonymous-public composition: the
    // synthesized `anon` identity on the batch and the transfer GET.
    let gets = lfs.requests_of("GET", &format!("/{REPO}/lfs/objects/{}", client.oid));
    assert!(
        !gets.is_empty(),
        "the consumer's download rode the transfer route"
    );
    assert!(
        gets.iter()
            .all(|request| request.auth == "anonymous" && request.status == 200),
        "the fresh consumer's actions rode the anonymous-public synthesis: {gets:?}"
    );

    git_door.accept.abort();
    http_door.accept.abort();
    std::fs::remove_dir_all(&base).ok();
}

/// The engine matrix's sqlite row: the same real-client round trip
/// (both doors, the durable single-host assembly — the media rides the
/// dual-tier object pool over the co-tenant file).
///
/// Contract stamp: lfs.md §Constraints (LFS ⇒ a dual-tier engine — the
/// sqlite assembly's object store is `dual_tier_sqlite_local`);
/// substrate.md §assembly shapes (the sqlite row); ADR-022 §1–2; the
/// round-trip row's stamps (the same protocol facts,
/// engine-swapped); substrate/w9-e2e-real-lfs.
#[cfg(feature = "sqlite")]
#[tokio::test(flavor = "multi_thread")]
async fn the_lfs_round_trip_rides_the_sqlite_assembly() {
    if !client_binaries() {
        return;
    }
    let harness = Harness::sqlite("lfs-sqlite", REPO, alkgit_substrate::PackServing::Direct).await;
    let git_door = common::spawn_daemon_door(&harness, common::DoorOptions::default()).await;
    let lfs = lfs_common::spawn_lfs_door(&harness).await;
    let base = row_base("sqlite");
    let client = lfs_common::init_lfs_client(
        &base,
        "origin",
        lfs_common::pseudo_bytes(0x0b3d, 256 * 1024),
    )
    .await
    .expect("the lfs client inits");

    push_pull_round_trip(&harness, git_door.addr, &lfs, REPO, &client).await;

    git_door.accept.abort();
    std::fs::remove_dir_all(&base).ok();
}

// ——— the postgres engine matrix row (the live-pg harness) ———

/// The live-pg harness DSN (the `ALKGIT_SUBSTRATE_PG_HOST` convention,
/// the `ALKBLOBS_PG_*` fallback family; the host's absence is the
/// clean-skip signal).
#[cfg(feature = "postgres")]
fn live_dsn() -> Option<String> {
    let pick = |alkgit_key: &str, alkblobs_key: &str, default: &str| {
        std::env::var(alkgit_key)
            .ok()
            .filter(|value| !value.is_empty())
            .or_else(|| {
                std::env::var(alkblobs_key)
                    .ok()
                    .filter(|value| !value.is_empty())
            })
            .unwrap_or_else(|| default.to_owned())
    };
    let host = std::env::var("ALKGIT_SUBSTRATE_PG_HOST")
        .ok()
        .filter(|value| !value.is_empty())
        .or_else(|| {
            std::env::var("ALKBLOBS_PG_HOST")
                .ok()
                .filter(|value| !value.is_empty())
        })?;
    Some(format!(
        "host={host} port={} user={} password={} dbname={}",
        pick("ALKGIT_SUBSTRATE_PG_PORT", "ALKBLOBS_PG_PORT", "15432"),
        pick("ALKGIT_SUBSTRATE_PG_USER", "ALKBLOBS_PG_USER", "postgres"),
        pick(
            "ALKGIT_SUBSTRATE_PG_PASSWORD",
            "ALKBLOBS_PG_PASSWORD",
            "poc"
        ),
        pick("ALKGIT_SUBSTRATE_PG_DB", "ALKBLOBS_PG_DB", "blobs"),
    ))
}

#[cfg(feature = "postgres")]
fn unique_schema(tag: &str) -> String {
    static COUNTER: AtomicU64 = AtomicU64::new(0);
    format!(
        "alkgit_substrate_pg_lfs_{}_{}_{}",
        std::process::id(),
        COUNTER.fetch_add(1, Ordering::Relaxed),
        tag
    )
}

#[cfg(feature = "postgres")]
async fn drop_schema(dsn: &str, schema: &str) {
    let (client, connection) = tokio_postgres::connect(dsn, tokio_postgres::NoTls)
        .await
        .expect("the schema drop connects");
    tokio::spawn(async move {
        let _ = connection.await;
    });
    client
        .batch_execute(&format!("DROP SCHEMA IF EXISTS \"{schema}\" CASCADE"))
        .await
        .ok();
}

#[cfg(feature = "postgres")]
fn pg_options(schema: &str) -> alkgit_substrate::PgRegistryOptions {
    alkgit_substrate::PgRegistryOptions {
        schema: schema.to_owned(),
        engine: alkstore_postgres::PgOpts {
            schema: format!("{schema}_engine"),
            ..alkstore_postgres::PgOpts::default()
        },
        blobs: alkblobs::PgOpts {
            schema: format!("{schema}_blobs"),
            ..alkblobs::PgOpts::default()
        },
    }
}

/// The engine matrix's postgres row: the same real-client round trip
/// over the fleet/durable-shared assembly (the live-pg harness
/// container; 0 skips when the container is up).
///
/// Contract stamp: lfs.md §Constraints (the dual-tier postgres shape);
/// substrate.md §assembly shapes (the postgres row); ADR-022 §1–2 (the
/// same protocol facts, engine-swapped); the live-pg harness env (the
/// ALKGIT_SUBSTRATE_PG convention); substrate/w9-e2e-real-lfs.
#[cfg(feature = "postgres")]
#[tokio::test(flavor = "multi_thread")]
async fn the_lfs_round_trip_rides_the_postgres_assembly() {
    let Some(dsn) = live_dsn() else {
        eprintln!(
            "skipping (no postgres harness server): set \
             ALKGIT_SUBSTRATE_PG_HOST (or the ALKBLOBS_PG_* harness env)"
        );
        return;
    };
    if !client_binaries() {
        return;
    }
    let schema = unique_schema("e2e");
    let options = pg_options(&schema);
    let root = common::unique_root("lfs-pg-e2e");
    let assembly_options =
        alkgit_substrate::AssemblyOptions::new(root.join("work"), root.join("large"));
    let assembly = match alkgit_substrate::SubstrateAssembly::postgres(
        &dsn,
        options,
        assembly_options,
    )
    .await
    {
        Ok(assembly) => assembly,
        Err(error) => {
            drop_schema(&dsn, &schema).await;
            drop_schema(&dsn, &format!("{schema}_engine")).await;
            drop_schema(&dsn, &format!("{schema}_blobs")).await;
            panic!("the pg assembly composes on the harness: {error}");
        }
    };
    let identity = common::writer_identity("u-writer");
    let record = common::public_granted_record(REPO, &identity);
    assembly
        .registry_store()
        .put_repo(&record)
        .await
        .expect("the row's repo registers");
    let harness = Harness {
        assembly,
        record,
        identity,
        root: root.clone(),
    };

    let git_door = common::spawn_daemon_door(&harness, common::DoorOptions::default()).await;
    let lfs = lfs_common::spawn_lfs_door(&harness).await;
    let base = row_base("pg");
    let client = lfs_common::init_lfs_client(
        &base,
        "origin",
        lfs_common::pseudo_bytes(0xfeed, 256 * 1024),
    )
    .await
    .expect("the lfs client inits");

    push_pull_round_trip(&harness, git_door.addr, &lfs, REPO, &client).await;

    git_door.accept.abort();
    drop(harness);
    drop_schema(&dsn, &schema).await;
    drop_schema(&dsn, &format!("{schema}_engine")).await;
    drop_schema(&dsn, &format!("{schema}_blobs")).await;
    std::fs::remove_dir_all(&root).ok();
    std::fs::remove_dir_all(&base).ok();
}

/// The re-push idempotence: a real `git lfs push` of a present object
/// — the batch answers no actions and no error for it (the client says
/// nothing to do; no PUT rides), the reference row stays the upserted
/// one row.
///
/// Contract stamp: ADR-022 §1 (the batch's idempotent posture — the
/// present object answers no upload action); pocl-findings (the
/// re-push datum: batch answers no actions);
/// substrate/w9-e2e-real-lfs.
#[tokio::test(flavor = "multi_thread")]
async fn a_real_repush_of_present_media_answers_no_upload_actions() {
    if !client_binaries() {
        return;
    }
    let harness = Harness::mem("lfs-repush", REPO).await;
    let git_door = common::spawn_daemon_door(&harness, common::DoorOptions::default()).await;
    let lfs = lfs_common::spawn_lfs_door(&harness).await;
    let base = row_base("repush");
    let client =
        lfs_common::init_lfs_client(&base, "origin", lfs_common::pseudo_bytes(0x3cc, 192 * 1024))
            .await
            .expect("the lfs client inits");
    let url = lfs.lfs_url(REPO, Some(lfs_common::WRITER_PASSWORD));
    common::git(
        Some(client.path()),
        &["config", "remote.origin.lfsurl", &url],
    )
    .await
    .expect("the row rides a resolved git binary")
    .expect_ok();

    for _ in 0..2 {
        common::git(
            Some(client.path()),
            &["lfs", "push", "--object-id", "origin", &client.oid],
        )
        .await
        .expect("the row rides a resolved git binary")
        .expect_ok();
    }

    let puts = lfs.requests_of("PUT", &format!("/{REPO}/lfs/objects/{}", client.oid));
    assert_eq!(
        puts.len(),
        1,
        "the re-push rode no transfer PUT — the present object has nothing to do"
    );
    let batches = lfs.requests_of("POST", "/objects/batch");
    let re_pushed = batches
        .last()
        .expect("the re-push's batch")
        .batch_objects
        .as_ref()
        .expect("the batch's objects logged");
    assert_eq!(re_pushed.len(), 1);
    assert_eq!(
        re_pushed[0].get("oid"),
        Some(&json!(client.oid)),
        "the re-push's answered object"
    );
    assert!(
        re_pushed[0].get("actions").is_none(),
        "the present object answers no upload action"
    );
    assert!(
        re_pushed[0].get("error").is_none(),
        "the present object answers no error"
    );

    git_door.accept.abort();
    std::fs::remove_dir_all(&base).ok();
}

/// The token hand-off's refusal shapes (the raw transfer surface): an
/// unknown bearer refuses 401 (invalid is never anonymous), a
/// download-action token is refused on the PUT route (the per-action
/// minted scope), a wrong-repo token refuses, and the accepted shape
/// (the minted download token serves the exact bytes) rides beside
/// them. The raw door rows run without the client binaries (they
/// never invoke git/git-lfs — the real-client rows skip, these serve).
///
/// Contract stamp: lfs.md §1 (hrefs door-minted absolute with
/// per-action `Authorization` — client-verified); pocl-findings (the
/// token hand-off's accepted/refused datum); ADR-022 §2 (the byte
/// paths the actions mint onto);
/// substrate/w9-e2e-real-lfs.
#[tokio::test(flavor = "multi_thread")]
async fn the_transfer_token_handoff_refuses_the_unmatched_or_unknown_tokens() {
    let harness = Harness::mem("lfs-tokens", REPO).await;
    let lfs = lfs_common::spawn_lfs_door(&harness).await;
    let base = row_base("tokens");
    let media = lfs_common::pseudo_bytes(0xf00d, 160 * 1024);
    let oid = lfs_common::oid_of(&media);

    // The placement: the raw batch upload's minted write token rides
    // the raw PUT.
    let reply = raw_http(
        lfs.addr,
        "POST",
        &format!("/{REPO}/objects/batch"),
        &[
            (
                "Authorization",
                &format!("Basic {}", basic64(&format!("lfs:{WRITER_PASSWORD}"))),
            ),
            ("Content-Type", "application/vnd.git-lfs+json"),
        ],
        &batch_body("upload", &[(&oid, media.len() as u64)], None),
    )
    .await;
    assert_eq!(
        reply.status,
        200,
        "the seeded upload batch: {}",
        String::from_utf8_lossy(&reply.body)
    );
    let object = reply.json()["objects"][0].clone();
    assert!(object.get("error").is_none(), "the seeded object: {object}");
    let (href, token) = action_of(&object, "upload");
    assert!(
        href.starts_with("http://"),
        "the minted href is an absolute URL: {href}"
    );
    let token = token.expect("the minted per-action Authorization");
    let put = raw_http(
        lfs.addr,
        "PUT",
        &href_path(&href),
        &[("Authorization", &token)],
        &media,
    )
    .await;
    assert_eq!(
        put.status,
        200,
        "the seeded PUT: {}",
        String::from_utf8_lossy(&put.body)
    );

    // The download batch's minted read token + the whole GET it serves.
    let reply = raw_http(
        lfs.addr,
        "POST",
        &format!("/{REPO}/objects/batch"),
        &[
            (
                "Authorization",
                &format!("Basic {}", basic64(&format!("lfs:{WRITER_PASSWORD}"))),
            ),
            ("Content-Type", "application/vnd.git-lfs+json"),
        ],
        &batch_body("download", &[(&oid, media.len() as u64)], None),
    )
    .await;
    assert_eq!(reply.status, 200);
    let (href, token) = action_of(&reply.json()["objects"][0], "download");
    let download_token = token.expect("the download action's minted Authorization");

    let whole = raw_http(
        lfs.addr,
        "GET",
        &href_path(&href),
        &[("Authorization", &download_token)],
        &[],
    )
    .await;
    assert_eq!(
        whole.status,
        200,
        "the minted download token is accepted: {}",
        String::from_utf8_lossy(&whole.body)
    );
    assert_eq!(whole.body, media, "the accepted shape serves exact bytes");

    // An unknown bearer refuses (invalid is never anonymous).
    let refused = raw_http(
        lfs.addr,
        "GET",
        &href_path(&href),
        &[(
            "Authorization",
            "Bearer lfs:00000000000000000000000000000000",
        )],
        &[],
    )
    .await;
    assert_eq!(refused.status, 401, "an unknown bearer refuses");

    // A download-action token on the PUT route is the per-action
    // refusal (403).
    let refused = raw_http(
        lfs.addr,
        "PUT",
        &href_path(&href),
        &[("Authorization", &download_token)],
        &media,
    )
    .await;
    assert_eq!(refused.status, 403, "a GET-scoped token PUTs nothing");

    // An unknown token on the PUT route refuses 401.
    let refused = raw_http(
        lfs.addr,
        "PUT",
        &href_path(&href),
        &[(
            "Authorization",
            "Bearer lfs:11111111111111111111111111111111",
        )],
        &media,
    )
    .await;
    assert_eq!(refused.status, 401, "an unknown PUT bearer refuses");

    // A wrong Basic on the batch refuses (invalid credentials are
    // refused, not degraded to anonymous).
    let refused = raw_http(
        lfs.addr,
        "POST",
        &format!("/{REPO}/objects/batch"),
        &[(
            "Authorization",
            &format!("Basic {}", basic64("lfs:no-such-pass")),
        )],
        &batch_body("download", &[(&oid, media.len() as u64)], None),
    )
    .await;
    assert_eq!(refused.status, 401, "an unknown Basic refuses");

    std::fs::remove_dir_all(&base).ok();
}

/// The batch's auth refusals + the wired-provider regression (the
/// POC-L gateway-gap seam evidence's posture, re-landed): the
/// unauthenticated upload batch refuses 401 (push is always
/// authenticated — the anonymous synthesis rides nothing on write), a
/// caller carrying a *static* resources claim is refused through the
/// wired provider, the wrong-credential caller is refused, and the
/// refused batches asserted no rows. Binary-independent (raw HTTP
/// only — no git/git-lfs invocation).
///
/// Contract stamp: alkcall ADR-048 (the wired provider is the
/// authority — the static fallback rides nothing through the
/// adapter's contexts); lfs.md §1 (the door's Basic tolerance + the
/// anonymous-public read's shape); ADR-022 §1 (the batch's
/// authorization posture); pocl-findings (the seam evidence's
/// posture); substrate/w9-e2e-real-lfs.
#[tokio::test(flavor = "multi_thread")]
async fn the_batch_refuses_unauthenticated_and_ungranted_callers() {
    let harness = Harness::mem("lfs-batch-auth", REPO).await;
    let lfs = lfs_common::spawn_lfs_door(&harness).await;
    let base = row_base("batch-auth");
    let media = lfs_common::pseudo_bytes(0x5eed, 64 * 1024);
    let oid = lfs_common::oid_of(&media);
    let route = format!("/{REPO}/objects/batch");

    // No credential on an upload batch: 401 (push is always
    // authenticated).
    let refused = raw_http(
        lfs.addr,
        "POST",
        &route,
        &[],
        &batch_body("upload", &[(&oid, media.len() as u64)], None),
    )
    .await;
    assert_eq!(
        refused.status, 401,
        "the unauthenticated upload batch refuses"
    );

    // A wrong credential: 401.
    let refused = raw_http(
        lfs.addr,
        "POST",
        &route,
        &[("Authorization", &format!("Basic {}", basic64("lfs:bogus")))],
        &batch_body("upload", &[(&oid, media.len() as u64)], None),
    )
    .await;
    assert_eq!(refused.status, 401, "the wrong credential refuses");

    // The static-claim caller (the outsider the door seeds): the wired
    // provider decides — the claim rides nothing; 401.
    let refused = raw_http(
        lfs.addr,
        "POST",
        &route,
        &[(
            "Authorization",
            &format!("Basic {}", basic64(&format!("lfs:{OUTSIDER_PASSWORD}"))),
        )],
        &batch_body("upload", &[(&oid, media.len() as u64)], None),
    )
    .await;
    assert_eq!(refused.status, 401, "the static-claim caller rides nothing");

    // The refusals asserted no rows.
    let rows = harness
        .assembly
        .lfs_rows()
        .rows()
        .await
        .expect("the rows read");
    assert!(
        rows.is_empty(),
        "the refused batches asserted nothing: {rows:?}"
    );

    // A non-sha256 batch is the pinned 409 (through the wired creds).
    let refused = raw_http(
        lfs.addr,
        "POST",
        &route,
        &[(
            "Authorization",
            &format!("Basic {}", basic64(&format!("lfs:{WRITER_PASSWORD}"))),
        )],
        &batch_body("upload", &[(&oid, media.len() as u64)], Some("sha1")),
    )
    .await;
    assert_eq!(refused.status, 409, "a non-sha256 batch is refused");

    // The anonymous-public read composes (the synthesized `anon` on
    // the public repo's download batch; the absent object answers the
    // per-object 404, authenticated by the grant it rode).
    let answer = raw_http(
        lfs.addr,
        "POST",
        &route,
        &[],
        &batch_body("download", &[(&oid, media.len() as u64)], None),
    )
    .await;
    assert_eq!(
        answer.status, 200,
        "the anonymous-public download batch composes"
    );
    let object = answer.json()["objects"][0].clone();
    assert_eq!(
        object["error"]["code"],
        json!(404),
        "the absent object answers per-object at the anonymous-public composition"
    );

    std::fs::remove_dir_all(&base).ok();
}

/// The mislabeled transfer PUT (the ADR-022 §2 contract at the
/// real-client surface level — a real client never mislabels; the raw
/// client shape is POC-L's raw_http posture): the body's digest
/// contradicts the claimed oid — the door refuses 400, no content
/// stored under the claimed key, no held pin (zero residue — the
/// sweep deletes nothing), and the check-time reference row stands.
/// Binary-independent (raw HTTP only).
///
/// Contract stamp: ADR-022 §2 (the mislabeled-body refusal — no
/// stored content, no held pin); lfs.md §2 (the transfer's
/// verification-on-put); pocl-findings (the mislabeled-PUT gate,
/// raw-client shape); substrate/w9-e2e-real-lfs.
#[tokio::test(flavor = "multi_thread")]
async fn a_mislabeled_transfer_put_refuses_with_zero_residue() {
    let harness = Harness::mem("lfs-mislabeled", REPO).await;
    let lfs = lfs_common::spawn_lfs_door(&harness).await;
    let base = row_base("mislabeled");
    let honest = lfs_common::pseudo_bytes(0xaaaa, 64 * 1024);
    let dishonest = lfs_common::pseudo_bytes(0xbbbb, honest.len());
    let claimed = lfs_common::oid_of(&honest);

    let reply = raw_http(
        lfs.addr,
        "POST",
        &format!("/{REPO}/objects/batch"),
        &[(
            "Authorization",
            &format!("Basic {}", basic64(&format!("lfs:{WRITER_PASSWORD}"))),
        )],
        &batch_body("upload", &[(&claimed, honest.len() as u64)], None),
    )
    .await;
    assert_eq!(reply.status, 200);
    let (href, token) = action_of(&reply.json()["objects"][0], "upload");
    let token = token.expect("the minted upload Authorization");

    let refused = raw_http(
        lfs.addr,
        "PUT",
        &href_path(&href),
        &[("Authorization", &token)],
        &dishonest,
    )
    .await;
    assert_eq!(
        refused.status, 400,
        "the mislabeled body refuses at the door"
    );

    // No stored content under the claimed key; zero residue; no held
    // pin (a sweep reclaims nothing — the content never landed).
    let key = lfs_media_key(&claimed).expect("the canonical key forms");
    assert!(
        harness
            .assembly
            .blobs()
            .stat(key.key_bytes().as_slice())
            .await
            .expect("the store stats")
            .is_none(),
        "nothing stored under the claimed oid"
    );
    assert!(
        harness
            .assembly
            .blobs()
            .list()
            .await
            .expect("the store lists")
            .is_empty(),
        "the refusal left zero residue in the pool"
    );
    let report = harness
        .assembly
        .blobs()
        .sweep()
        .await
        .expect("the sweep runs");
    assert!(
        report.aborted.is_none() && report.deleted == 0,
        "nothing of the refused body is GC-relevant: {report:?}"
    );
    let rows = harness
        .assembly
        .lfs_rows()
        .repo_rows(REPO)
        .await
        .expect("the rows read");
    assert_eq!(
        rows.iter().map(|row| row.oid.clone()).collect::<Vec<_>>(),
        vec![claimed],
        "the check-time reference row stands (the pair was learned; the byte path refused)"
    );

    std::fs::remove_dir_all(&base).ok();
}

/// The Range semantics (the raw download surface): a Range-bearing GET
/// answers 206 with the exact slice and the correct `Content-Range`;
/// an out-of-range offset answers 416 with `bytes */len`; the
/// open-ended and byte-suffix (resume) forms answer the tails; the
/// multi-window slice rides the adapter's verified-window split
/// (byte-exact). Binary-independent (raw HTTP only).
///
/// Contract stamp: lfs.md §2 (the transfer download's Range serving —
/// 206/416, Content-Range); ADR-022 §2 (the verified-window ranged
/// fetch); pocl-findings (the range gate: `bytes=0-99` → 206 exact,
/// `bytes=<len+>-` → 416 `bytes */len`);
/// substrate/w9-e2e-real-lfs.
#[tokio::test(flavor = "multi_thread")]
async fn range_bearing_gets_serve_206_and_416_with_exact_framing() {
    let harness = Harness::mem("lfs-range", REPO).await;
    let lfs = lfs_common::spawn_lfs_door(&harness).await;
    let base = row_base("range");
    let media = lfs_common::pseudo_bytes(0xf17e, 300 * 1024);
    let oid = lfs_common::oid_of(&media);
    let total = media.len() as u64;

    // The placement rides the raw batch + PUT (with the minted token).
    let reply = raw_http(
        lfs.addr,
        "POST",
        &format!("/{REPO}/objects/batch"),
        &[(
            "Authorization",
            &format!("Basic {}", basic64(&format!("lfs:{WRITER_PASSWORD}"))),
        )],
        &batch_body("upload", &[(&oid, total)], None),
    )
    .await;
    assert_eq!(reply.status, 200);
    let (href, token) = action_of(&reply.json()["objects"][0], "upload");
    let token = token.expect("the minted upload Authorization");
    let put = raw_http(
        lfs.addr,
        "PUT",
        &href_path(&href),
        &[("Authorization", &token)],
        &media,
    )
    .await;
    assert_eq!(
        put.status,
        200,
        "the seeded content lands: {}",
        String::from_utf8_lossy(&put.body)
    );

    // The download batch mints the GET's token.
    let reply = raw_http(
        lfs.addr,
        "POST",
        &format!("/{REPO}/objects/batch"),
        &[(
            "Authorization",
            &format!("Basic {}", basic64(&format!("lfs:{WRITER_PASSWORD}"))),
        )],
        &batch_body("download", &[(&oid, total)], None),
    )
    .await;
    assert_eq!(reply.status, 200);
    let (href, token) = action_of(&reply.json()["objects"][0], "download");
    let token = token.expect("the minted download Authorization");
    let path = href_path(&href);

    let get = |range: String| {
        let token = token.clone();
        let path = path.clone();
        let lfs_addr = lfs.addr;
        async move {
            raw_http(
                lfs_addr,
                "GET",
                &path,
                &[("Authorization", &token), ("Range", &range)],
                &[],
            )
            .await
        }
    };
    let whole = raw_http(lfs.addr, "GET", &path, &[("Authorization", &token)], &[]).await;
    assert_eq!(whole.status, 200, "the whole GET");
    assert_eq!(whole.body, media, "the whole GET is byte-exact");

    // The exact slice + its Content-Range.
    let slice = get("bytes=0-99".to_owned()).await;
    assert_eq!(slice.status, 206, "the ranged GET answers partial content");
    assert_eq!(slice.body, &media[..100], "the served slice is exact");
    let head = slice.header("content-range").expect("the Content-Range");
    assert_eq!(
        head,
        format!("bytes 0-99/{total}"),
        "the Content-Range frames the slice"
    );

    // A multi-window slice (the verified-window split, byte-exact).
    let slice = get("bytes=0-199999".to_owned()).await;
    assert_eq!(slice.status, 206);
    assert_eq!(
        slice.body,
        &media[..200_000],
        "the multi-window slice is byte-exact"
    );
    assert_eq!(
        slice.header("content-range").expect("the Content-Range"),
        format!("bytes 0-199999/{total}"),
        "the clamped multi-window framing"
    );

    // The open-ended (resume) form.
    let tail = get("bytes=100-".to_owned()).await;
    assert_eq!(tail.status, 206);
    assert_eq!(tail.body, &media[100..], "the open-ended tail");
    assert_eq!(
        tail.header("content-range").expect("the Content-Range"),
        format!("bytes 100-{}/{total}", total - 1),
    );

    // The byte-suffix (resume) form.
    let suffix = get("bytes=-50".to_owned()).await;
    assert_eq!(suffix.status, 206);
    assert_eq!(
        suffix.body,
        &media[media.len() - 50..],
        "the suffix form's tail"
    );

    // The out-of-range offset: 416 with `Content-Range: bytes */len`.
    let refused = get(format!("bytes={}-", total + 5)).await;
    assert_eq!(refused.status, 416, "the past-EOF offset refuses");
    assert_eq!(
        refused.header("content-range").expect("the Content-Range"),
        format!("bytes */{total}"),
        "the 416's Content-Range carries the total"
    );

    // The past-EOF-end span: the door clamps the resolved span to the
    // entry (the resolved-span arithmetic the adapter re-checks) — a
    // whole serve as 206 with the true-end framing.
    let clamped = get(format!("bytes=0-{}", total + 12_345)).await;
    assert_eq!(
        clamped.status, 206,
        "the past-EOF-end span clamps to the entry"
    );
    assert_eq!(clamped.body, media, "the clamped span serves the whole");
    assert_eq!(
        clamped.header("content-range").expect("the Content-Range"),
        format!("bytes 0-{}/{total}", total - 1),
        "the clamped framing carries the true end"
    );

    // The inverted span refuses.
    let inverted = get("bytes=99-0".to_owned()).await;
    assert_eq!(inverted.status, 416, "the inverted span refuses");

    // The multi-range header refuses: the door's grammar is the
    // single-range surface — a multi-range never degenerates into an
    // open-ended serve.
    let multi = get("bytes=0-99,200-299".to_owned()).await;
    assert_eq!(multi.status, 416, "the multi-range header refuses");
    assert_eq!(
        multi.header("content-range").expect("the Content-Range"),
        format!("bytes */{total}"),
        "the multi-range refusal carries the total"
    );

    std::fs::remove_dir_all(&base).ok();
}

/// The big-media/ceiling door row (the POC-L big-media measurement
/// gates re-landed small-config, plus the door's refused paths): a
/// narrow-door config (128 KiB verified windows, a 1 MiB object
/// ceiling) serves the measurement shapes — a multi-window PUT
/// (accepted, verified-on-put, the pair's row asserted, the content
/// byte-exact under the canonical key), a multi-window ranged GET
/// (the window-split slice datum, byte-exact with the framing), and
/// a multi-window whole GET — and isolates the door paths the
/// booking's `TRANSFER_BODY_BUDGET` / `TooLarge`→413 / unrouted-404
/// rows assert: the whole serve over the configured ceiling answers
/// 413 with the partial span under it riding the windows (the
/// resolved-span ceiling's e2e shape), the PUT body over the door's
/// transfer-body budget refuses 413 **before the adapter** (zero
/// residue — no store entry, and the pair's row never asserted — the
/// door-side ordering), a non-oid transfer path is the door's
/// unrouted 404 (not the adapter's per-object answer), and the doubled
/// `info/lfs/lfs/…` transfer path is unrouted 404 (the suffix grammar
/// serves transfer only at the door-minted shape — the batch-only
/// `/info/lfs` accommodation). Binary-
/// independent (raw HTTP only). The upload arm carries no ceiling
/// (the put rides the pre-keyed put's contract) — the over-ceiling
/// media's PUT lands and its whole GET refuses (the ceiling is the
/// download span budget, per assembly).
///
/// Contract stamp: pocl-findings (the big-media measurement gates —
/// the 70 MiB PUT / 45 MiB two-window ranged / 70 MiB whole GET
/// datums, the shapes pinned at the small config); lfs.md §2 (the
/// door budgets the whole arm — the ceiling is the door's per
/// assembly choice, realized adapter-side); ADR-022 §2 (the byte
/// paths); substrate/w10-test-evidence (F-w10-B1).
#[tokio::test(flavor = "multi_thread")]
async fn the_small_config_door_serves_the_big_media_shapes_and_refuses_the_ceiling_and_budget_paths(
) {
    let harness = Harness::mem("lfs-bigmedia", REPO).await;
    let config = alkgit_substrate::LfsAdapterConfig {
        window_bytes: 128 * 1024,
        max_object_bytes: 1024 * 1024,
    };
    let lfs = lfs_common::spawn_lfs_door_with(&harness, config).await;
    let base = row_base("bigmedia");

    // The measurement media: 768 KiB — six verified windows.
    let media = lfs_common::pseudo_bytes(0xb17e, 768 * 1024);
    let oid = lfs_common::oid_of(&media);
    let total = media.len() as u64;

    // The multi-window PUT: the batch mints, the PUT lands byte-exact
    // (the POC-L big-PUT datum's shape — accepted, verified, attached).
    let reply = raw_http(
        lfs.addr,
        "POST",
        &format!("/{REPO}/objects/batch"),
        &[(
            "Authorization",
            &format!("Basic {}", basic64(&format!("lfs:{WRITER_PASSWORD}"))),
        )],
        &batch_body("upload", &[(&oid, total)], None),
    )
    .await;
    assert_eq!(reply.status, 200);
    let (href, token) = action_of(&reply.json()["objects"][0], "upload");
    let upload_token = token.expect("the minted upload Authorization");
    let put = raw_http(
        lfs.addr,
        "PUT",
        &href_path(&href),
        &[("Authorization", &upload_token)],
        &media,
    )
    .await;
    assert_eq!(
        put.status,
        200,
        "the multi-window media lands: {}",
        String::from_utf8_lossy(&put.body)
    );

    // The pair's row asserted (the batch-check time) + the content
    // under the canonical key.
    let rows = harness
        .assembly
        .lfs_rows()
        .repo_rows(REPO)
        .await
        .expect("the rows read");
    assert_eq!(
        rows.iter().map(|row| row.oid.clone()).collect::<Vec<_>>(),
        vec![oid.clone()],
        "the upload's pair asserted its row"
    );
    let key = lfs_media_key(&oid).expect("the canonical key forms");
    assert!(
        harness
            .assembly
            .blobs()
            .stat(key.key_bytes().as_slice())
            .await
            .expect("the store stats")
            .is_some(),
        "the multi-window content attached under the canonical key"
    );

    // The download batch mints the read token.
    let reply = raw_http(
        lfs.addr,
        "POST",
        &format!("/{REPO}/objects/batch"),
        &[(
            "Authorization",
            &format!("Basic {}", basic64(&format!("lfs:{WRITER_PASSWORD}"))),
        )],
        &batch_body("download", &[(&oid, total)], None),
    )
    .await;
    assert_eq!(reply.status, 200);
    let (href, token) = action_of(&reply.json()["objects"][0], "download");
    let download_token = token.expect("the minted download Authorization");
    let path = href_path(&href);

    // The multi-window whole GET (the POC-L big-GET datum's shape).
    let whole = raw_http(
        lfs.addr,
        "GET",
        &path,
        &[("Authorization", &download_token)],
        &[],
    )
    .await;
    assert_eq!(whole.status, 200, "the whole GET serves");
    assert_eq!(whole.body, media, "the whole GET is byte-exact");

    // The multi-window ranged slice (the two-window split datum's
    // shape): a span riding three window boundaries, byte-exact with
    // the clamped framing.
    let ranged = raw_http(
        lfs.addr,
        "GET",
        &path,
        &[
            ("Authorization", &download_token),
            ("Range", "bytes=0-299999"),
        ],
        &[],
    )
    .await;
    assert_eq!(ranged.status, 206, "the multi-window slice serves");
    assert_eq!(
        ranged.body,
        &media[..300_000],
        "the multi-window slice is byte-exact"
    );
    assert_eq!(
        ranged.header("content-range").expect("the Content-Range"),
        format!("bytes 0-299999/{total}"),
        "the multi-window framing"
    );

    // The over-ceiling media: the PUT lands (the upload arm carries no
    // ceiling — the pre-keyed put's contract), and the whole GET's
    // resolved span over the configured ceiling refuses `TooLarge`→413.
    let big = lfs_common::pseudo_bytes(0xc01a, 2 * 1024 * 1024);
    let big_oid = lfs_common::oid_of(&big);
    let reply = raw_http(
        lfs.addr,
        "POST",
        &format!("/{REPO}/objects/batch"),
        &[(
            "Authorization",
            &format!("Basic {}", basic64(&format!("lfs:{WRITER_PASSWORD}"))),
        )],
        &batch_body("upload", &[(&big_oid, big.len() as u64)], None),
    )
    .await;
    assert_eq!(reply.status, 200);
    let (big_href, token) = action_of(&reply.json()["objects"][0], "upload");
    let big_put = raw_http(
        lfs.addr,
        "PUT",
        &href_path(&big_href),
        &[(
            "Authorization",
            &token.expect("the big media's upload token"),
        )],
        &big,
    )
    .await;
    assert_eq!(
        big_put.status,
        200,
        "the upload arm carries no ceiling: {}",
        String::from_utf8_lossy(&big_put.body)
    );

    // The download batch for the big media mints its read token.
    let reply = raw_http(
        lfs.addr,
        "POST",
        &format!("/{REPO}/objects/batch"),
        &[(
            "Authorization",
            &format!("Basic {}", basic64(&format!("lfs:{WRITER_PASSWORD}"))),
        )],
        &batch_body("download", &[(&big_oid, big.len() as u64)], None),
    )
    .await;
    assert_eq!(reply.status, 200);
    let (big_path, big_token) = {
        let (href, token) = action_of(&reply.json()["objects"][0], "download");
        (
            href_path(&href),
            token.expect("the big media's download token"),
        )
    };

    // The whole GET over the ceiling refuses `TooLarge`→413.
    let refused = raw_http(
        lfs.addr,
        "GET",
        &big_path,
        &[("Authorization", &big_token)],
        &[],
    )
    .await;
    assert_eq!(
        refused.status, 413,
        "the resolved span over the ceiling refuses"
    );

    // The partial span under the ceiling rides the windows (the
    // resume shape the ceiling preserves) — byte-exact at the ceiling
    // edge.
    let partial = raw_http(
        lfs.addr,
        "GET",
        &big_path,
        &[("Authorization", &big_token), ("Range", "bytes=0-999999")],
        &[],
    )
    .await;
    assert_eq!(partial.status, 206, "the under-ceiling span serves");
    assert_eq!(
        partial.body,
        &big[..1_000_000],
        "the ceiling-edge slice is byte-exact"
    );

    // The door's transfer-body budget: a PUT body over
    // `TRANSFER_BODY_BUDGET` refuses 413 before the adapter — no
    // store entry, and the pair's row never asserted (the door-side
    // ordering; the minted upload token is repo+action scoped and
    // rides the route fine).
    let budget = lfs_common::pseudo_bytes(0x7bd9, lfs_common::TRANSFER_BODY_BUDGET + 1);
    let budget_oid = lfs_common::oid_of(&budget);
    let refused = raw_http(
        lfs.addr,
        "PUT",
        &format!("/{REPO}/lfs/objects/{budget_oid}"),
        &[("Authorization", &upload_token)],
        &budget,
    )
    .await;
    assert_eq!(
        refused.status, 413,
        "the transfer body over the door's budget refuses"
    );
    let budget_key = lfs_media_key(&budget_oid).expect("the canonical key forms");
    assert!(
        harness
            .assembly
            .blobs()
            .stat(budget_key.key_bytes().as_slice())
            .await
            .expect("the store stats")
            .is_none(),
        "the budget-refused body stored nothing"
    );

    // The unrouted shape: a non-oid transfer path is the door's 404
    // (the route grammar parses the oid shape — not the adapter's
    // per-object answer).
    let unrouted = raw_http(
        lfs.addr,
        "GET",
        &format!("/{REPO}/lfs/objects/{}", "0".repeat(63)),
        &[("Authorization", &download_token)],
        &[],
    )
    .await;
    assert_eq!(unrouted.status, 404, "the unrouted shape refuses");

    // The doubled route form: the gitea-embedded prefix before the
    // transfer marker (`/{repo}/info/lfs/lfs/objects/{oid}`) is
    // unrouted too — hrefs are door-minted and the one served transfer
    // shape is `/{repo}/lfs/objects/{oid}`; the `/info/lfs` embedding
    // is a batch-only accommodation (the door's suffix grammar
    // tightened against the doubled form — F-w10-B5).
    let doubled = raw_http(
        lfs.addr,
        "GET",
        &format!("/{REPO}/info/lfs/lfs/objects/{oid}"),
        &[("Authorization", &download_token)],
        &[],
    )
    .await;
    assert_eq!(doubled.status, 404, "the doubled route form is unrouted");

    // The final rows: exactly the landed medias' pairs (the
    // budget-refused PUT never wrote).
    let rows = harness
        .assembly
        .lfs_rows()
        .repo_rows(REPO)
        .await
        .expect("the rows read");
    let oids = rows.iter().map(|row| row.oid.clone()).collect::<Vec<_>>();
    assert!(
        oids.iter().all(|o| *o == oid || *o == big_oid) && oids.len() == 2,
        "the landed pairs are the served medias only: {oids:?}"
    );

    std::fs::remove_dir_all(&base).ok();
}

/// The anonymous transfer-route refusal row (the composition's
/// anonymous-public posture's write half): a public repo's
/// synthesized `anon` identity rides the transfer PUT unauthenticated
/// (the door resolves it — the anonymous-public read posture) and the
/// **adapter refuses the write** (push is always authenticated — the
/// wired provider grants `anon` read-only): the door answers 401 with
/// the refused request's auth kind logged as the anonymous ride (the
/// adapter's refusal mapping, not the door's missing-resolution 401),
/// zero residue (no store entry), and the pair's row never asserted.
/// Binary-independent (raw HTTP only).
///
/// Contract stamp: doors.md §alkhttp `lfs` (the anonymous-public
/// synthesized `anon` identity — no header minted for it); lfs.md §1
/// (the anonymous ride is read-shaped); ADR-022 §1 (push is always
/// authenticated); substrate/w10-test-evidence (F-w10-B1).
#[tokio::test(flavor = "multi_thread")]
async fn a_public_repos_synthesized_anon_put_refuses_at_the_adapter() {
    let harness = Harness::mem("lfs-anon-put", REPO).await;
    let lfs = lfs_common::spawn_lfs_door(&harness).await;
    let base = row_base("anon-put");
    let media = lfs_common::pseudo_bytes(0xa000, 64 * 1024);
    let oid = lfs_common::oid_of(&media);

    // The unauthenticated transfer PUT on the public repo: the door
    // resolves the synthesized `anon` (the anonymous-public posture)
    // and the adapter refuses the write.
    let refused = raw_http(
        lfs.addr,
        "PUT",
        &format!("/{REPO}/lfs/objects/{oid}"),
        &[],
        &media,
    )
    .await;
    assert_eq!(
        refused.status, 401,
        "the synthesized anon's PUT refuses (push is always authenticated)"
    );

    // The refusal's provenance is the adapter: the request logged as
    // the anonymous ride (the door resolved `anon` — a private repo's
    // refusal would log the missing-resolution 401 instead).
    let puts = lfs.requests_of("PUT", &format!("/{REPO}/lfs/objects/{oid}"));
    assert_eq!(puts.len(), 1, "the refusal logged once");
    assert_eq!(puts[0].auth, "anonymous", "the anon ride resolved");
    assert_eq!(puts[0].status, 401, "the adapter refused it");

    // Zero residue: no store entry under the oid, the pair's row
    // never asserted (the refusal precedes the row write).
    let key = lfs_media_key(&oid).expect("the canonical key forms");
    assert!(
        harness
            .assembly
            .blobs()
            .stat(key.key_bytes().as_slice())
            .await
            .expect("the store stats")
            .is_none(),
        "the refused body stored nothing"
    );
    let rows = harness
        .assembly
        .lfs_rows()
        .repo_rows(REPO)
        .await
        .expect("the rows read");
    assert!(
        rows.is_empty(),
        "the refused PUT asserted no rows: {rows:?}"
    );

    std::fs::remove_dir_all(&base).ok();
}

/// The GC sweep contract at the real-client level: real-client media
/// survives the sweep (the rows protect it); the cross-reference
/// re-protection holds across a sweep window (a second repo's row
/// re-protects the shared key past the first repo's removal — the
/// surviving repo's real pull still serves the exact bytes); the last
/// removal makes the media sweep-eligible and the sweep reclaims it.
///
/// Contract stamp: ADR-022 §4 (the reference rows are the media's
/// liveness — the re-protection contract, repo-deletion's reclaim);
/// lfs.md §3 (the pointer lifecycle: rows written at check time,
/// consumed by the sweep chain; rows go with the repo);
/// substrate.md §GC (the registered second source);
/// pocl-findings (the sweep gate: orphaned entry deleted, referenced
/// survive); substrate/w9-e2e-real-lfs.
#[tokio::test(flavor = "multi_thread")]
async fn the_sweep_contract_holds_real_client_media_across_repo_removals() {
    if !client_binaries() {
        return;
    }
    let harness = Harness::mem("lfs-sweep", REPO).await;
    let lfs = lfs_common::spawn_lfs_door(&harness).await;
    let base = row_base("sweep");
    let media = lfs_common::pseudo_bytes(0xda7a, 192 * 1024);

    // Repo A: the real client uploads the media.
    let client_a = lfs_common::init_lfs_client(&base, "a", media.clone())
        .await
        .expect("the client inits");
    let url = lfs.lfs_url(REPO, Some(lfs_common::WRITER_PASSWORD));
    common::git(
        Some(client_a.path()),
        &["config", "remote.origin.lfsurl", &url],
    )
    .await
    .expect("the row rides a resolved git binary")
    .expect_ok();
    common::git(
        Some(client_a.path()),
        &["lfs", "push", "--object-id", "origin", &client_a.oid],
    )
    .await
    .expect("the row rides a resolved git binary")
    .expect_ok();
    let key = lfs_media_key(&client_a.oid).expect("the canonical key forms");
    assert!(
        harness
            .assembly
            .blobs()
            .stat(key.key_bytes().as_slice())
            .await
            .expect("the store stats")
            .is_some(),
        "the real-client media is in the pool"
    );

    // The sweep holds it: the reference row protects the key.
    let report = harness
        .assembly
        .blobs()
        .sweep()
        .await
        .expect("the sweep runs");
    assert!(
        report.aborted.is_none() && report.deleted == 0 && report.candidates_staged == 0,
        "the referenced media never stages: {report:?}"
    );

    // Repo B references the same content (the shared key): the real
    // client's batch asserts B's row — the content is present, so no
    // PUT rides at all.
    harness
        .assembly
        .registry_store()
        .put_repo(&common::public_granted_record(REPO_B, &harness.identity))
        .await
        .expect("repo B registers");
    let client_b = lfs_common::init_lfs_client(&base, "b", media.clone())
        .await
        .expect("the client inits");
    assert_eq!(client_b.oid, client_a.oid, "the shared content's oid");
    let url_b = lfs.lfs_url(REPO_B, Some(lfs_common::WRITER_PASSWORD));
    common::git(
        Some(client_b.path()),
        &["config", "remote.origin.lfsurl", &url_b],
    )
    .await
    .expect("the row rides a resolved git binary")
    .expect_ok();
    common::git(
        Some(client_b.path()),
        &["lfs", "push", "--object-id", "origin", &client_b.oid],
    )
    .await
    .expect("the row rides a resolved git binary")
    .expect_ok();
    assert!(
        lfs.requests_of("PUT", &format!("/{REPO_B}/lfs/objects/{}", client_b.oid))
            .is_empty(),
        "the shared content needed no second PUT"
    );
    let rows = harness
        .assembly
        .lfs_rows()
        .rows()
        .await
        .expect("the rows read")
        .into_iter()
        .filter(|row| row.oid == client_a.oid)
        .collect::<Vec<_>>();
    assert_eq!(rows.len(), 2, "two repos — one key, two rows");
    for row in &rows {
        assert_eq!(row.refcount, 2, "the shared key's count: {row:?}");
    }

    // Repo A's removal: the rows go with the repo — the sweep still
    // protects the media (B's row re-protects it; the re-protection
    // contract across a sweep window).
    harness
        .assembly
        .registry_store()
        .remove_repo(REPO)
        .await
        .expect("repo A removes");
    let report = harness
        .assembly
        .blobs()
        .sweep()
        .await
        .expect("the sweep runs");
    assert!(
        report.aborted.is_none() && report.deleted == 0 && report.candidates_staged == 0,
        "the surviving row re-protects the shared key: {report:?}"
    );
    assert!(
        harness
            .assembly
            .blobs()
            .stat(key.key_bytes().as_slice())
            .await
            .expect("the store stats")
            .is_some(),
        "the media survives repo A's removal"
    );

    // The surviving repo's real pull serves the exact bytes.
    lfs_common::wipe_local_store(&client_b);
    let url = lfs.lfs_url(REPO_B, None);
    common::git(
        Some(client_b.path()),
        &["config", "remote.origin.lfsurl", &url],
    )
    .await
    .expect("the row rides a resolved git binary")
    .expect_ok();
    common::git(Some(client_b.path()), &["lfs", "pull", "origin"])
        .await
        .expect("the row rides a resolved git binary")
        .expect_ok();
    let restored = std::fs::read(client_b.dir.join("media.bin")).expect("the pull");
    assert_eq!(restored, media, "the surviving repo pulls the exact bytes");

    // Repo B's removal: the last row leaves — the media sweeps.
    harness
        .assembly
        .registry_store()
        .remove_repo(REPO_B)
        .await
        .expect("repo B removes");
    let report = harness
        .assembly
        .blobs()
        .sweep()
        .await
        .expect("the sweep runs");
    assert!(
        report.aborted.is_none() && report.deleted == 1,
        "the unreferenced media reclaims: {report:?}"
    );
    assert!(
        harness
            .assembly
            .blobs()
            .stat(key.key_bytes().as_slice())
            .await
            .expect("the store stats")
            .is_none(),
        "the media is gone"
    );

    std::fs::remove_dir_all(&base).ok();
}