alkgit-substrate 0.1.0

//! The substrate-backed real-git harness — the two wave-8 end-to-end
//! tasks' shared fixture code (substrate/w8-e2e-real-git,
//! substrate/w8-e2e-replication). The core crate's `tests/common` is not
//! importable across the crate seam (the two-member workspace's
//! seam-separated test posture, the w7-packgen byte-identity note), so
//! this module re-lands the door shapes the core's `tests/real_client.rs`
//! pins — against the *real substrate backend*: every machine consumes
//! the composition entry's complete backend (the registry pair, refs
//! rows, packgen, the w8 ingest), not doubles.
//!
//! # The row/stamp discipline
//!
//! Assertions are **rows**: named, independently-runnable properties,
//! `snake_case`, the name stating the property. Every row is
//! **version-stamped** with a doc-comment list item beginning
//! `Contract stamp:` citing the spec § / ADR § that pins the behavior
//! tested; stamps accumulate. `grep -rn "Contract stamp:" substrate/tests/`
//! is the inventory. (The core's `tests/common` §The row/stamp
//! discipline is the reference text.)
//!
//! # The doors (the recorded postures — the core's real-client shapes
//! over the substrate backend)
//!
//! - **The git-daemon-framed duplex door** — the first pkt-line is the
//!   daemon request line (`git-upload-pack <repo>\0host=…\0\0version=2\0`
//!   for fetch; `git-receive-pack <repo>\0host=…\0` for push, ADR-016 §2
//!   — a push request line carries no `version=2`, the push path is
//!   V0-framed). The door parses the line, resolves the repo id against
//!   the harness's registry, authorizes the action the service selects
//!   (the ADR-007 front's posture, as the core's real-client doors run
//!   it per connection), and relays the framed transport into
//!   [`alkgit::serve_duplex`].
//! - **The door's identity injection**: `git` carries no identity over
//!   `git://` or smart-http, and push is always authenticated (the
//!   visible-surface invariant) — the duplex door injects the harness's
//!   writer identity, the door simulating the authenticated
//!   alkcall/alkssh peer whose session it fronts. The http door injects
//!   nothing: its repo is public and the fetch session is the anonymous
//!   public read (ADR-011 §3), the core's real-client door posture.
//! - **The smart-http-framed stateless door** — the core's service-stub
//!   shape verbatim, upload-pack routes:
//!   `GET /<repo>/info/refs?service=git-upload-pack` serves
//!   [`alkgit::serve_stateless_advertisement`], `POST
//!   /<repo>/git-upload-pack` serves [`alkgit::serve_stateless_post`]
//!   (one POST body one command invocation), Content-Length-framed
//!   bodies, keep-alive across the GET and the round POSTs.
//! - **The capture tee** — optional; both relay directions are
//!   accumulated as the row's wire evidence. The client→server tee is
//!   how a row proves what real `git` actually sent (a thin pack's
//!   delta bases; a delete-only push's packlessness).
//! - **The gated hold** — optional; the deterministic clashing-client
//!   orchestration (the atomic-CAS row): the door forwards the client's
//!   command pkt-lines live and *holds* everything from the command
//!   flush onward (the pack rides in the hold) until the row releases,
//!   first signaling `held` when the flush is held. The real client's
//!   session — its advertisement served, its request transmitted — then
//!   races exactly as the protocol intends; the hold freezes the race
//!   point at the door so the CAS orders the two clients
//!   deterministically.
//!
//! Hygiene: every harness lives under its own unique temp directory
//! (removed when the harness drops); every real-git invocation runs
//! under an isolated env (no system/user config, no prompts) with a
//! wall-clock timeout so a deadlocked session fails the row instead of
//! hanging it; the doors bound each connection's work (the machine's
//! session wall clock plus a door-level hygiene timeout).

#![allow(dead_code)]

use std::collections::{BTreeMap, BTreeSet};
use std::path::{Path, PathBuf};
use std::process::Stdio;
use std::sync::atomic::{AtomicBool, Ordering as AtomicOrdering};
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::Arc;
use std::time::{Duration, SystemTime, UNIX_EPOCH};

use alkcall::core::Identity;
use alkgit::{
    serve_duplex, serve_stateless_advertisement, serve_stateless_post, AccessAction,
    AuthorizedRepo, BodyReader, GitPackGen, GitPackIngest, GitRefs, GitRegistry, Limits,
    RepoRecord, Service, SubstrateError, Visibility,
};
#[cfg(feature = "sqlite")]
use alkgit_substrate::PackServing;
use alkgit_substrate::{AssemblyOptions, MaintenanceOpts, SubstrateAssembly};
use bytes::Bytes;
use futures::io::AsyncWrite as FuturesAsyncWrite;
use futures::stream;
use parking_lot::Mutex;
use tokio::io::{duplex, AsyncReadExt, AsyncWriteExt};
use tokio::net::{TcpListener, TcpStream};
use tokio::sync::Semaphore;
use tokio::task::JoinHandle;

/// The door's connection-level hygiene timeout (a deadlocked session
/// fails the row instead of parking it out to the machine's wall clock).
pub(crate) const DOOR_CONN_TIMEOUT: Duration = Duration::from_secs(120);

/// The relay's chunk size.
const RELAY_CHUNK_BYTES: usize = 16384;

/// The git invocations' wall clock (process hygiene — a real-git
/// session against a healthy server completes inside this).
pub const GIT_RUN_TIMEOUT: Duration = Duration::from_secs(90);

/// The default repo id a harness registers.
pub const DEFAULT_REPO_ID: &str = "owner/real-git";

/// The harness's writer identity — the peer the push doors inject (the
/// identity the grants table names).
pub fn writer_identity(id: &str) -> Identity {
    Identity {
        id: id.to_owned(),
        scopes: Vec::new(),
        resources: Default::default(),
    }
}

/// A public repo record granting the identity read + write (the
/// anonymous public read for the http fetch door; the write grant for
/// the injected push peer).
pub fn public_granted_record(repo_id: &str, identity: &Identity) -> RepoRecord {
    RepoRecord {
        repo_id: repo_id.to_owned(),
        visibility: Visibility::Public,
        grants: BTreeMap::from([(
            identity.id.clone(),
            vec![AccessAction::Read, AccessAction::Write],
        )]),
    }
}

/// A fresh unique harness root under the temp dir (pid + nanos + the
/// process-wide counter), created empty.
pub fn unique_root(tag: &str) -> PathBuf {
    static COUNTER: AtomicU64 = AtomicU64::new(0);
    let n = COUNTER.fetch_add(1, Ordering::Relaxed);
    let nanos = SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .map(|d| d.as_nanos())
        .unwrap_or_default();
    let dir = std::env::temp_dir().join(format!(
        "alkgit-substrate-real-git-{tag}-{}-{n}-{nanos}",
        std::process::id()
    ));
    std::fs::create_dir_all(&dir).expect("the harness root creates");
    dir
}

/// The substrate-backed harness: the complete backend (one composition
/// entry) plus the registered repo record and the push doors' injected
/// peer identity. Dropping it removes the root directory.
pub struct Harness {
    /// The wired whole (the composition entry's product).
    pub assembly: SubstrateAssembly,
    /// The registered repo the doors route to.
    pub record: RepoRecord,
    /// The injected peer identity (the push doors' authenticated shape).
    pub identity: Identity,
    /// The harness root (temporary; removed at drop) — public so the
    /// engine-matrix rows can assemble variant shapes around it.
    pub root: PathBuf,
}

impl Drop for Harness {
    fn drop(&mut self) {
        let _ = std::fs::remove_dir_all(&self.root);
    }
}

impl Harness {
    /// The mem assembly's completeness shape (no engine dependency): a
    /// mem alkstore engine + a mem object pool, the assembly options'
    /// working-area and large-tier roots under the harness root.
    pub async fn mem(tag: &str, repo_id: &str) -> Harness {
        Harness::mem_tuned(tag, repo_id, Default::default()).await
    }

    /// [`Harness::mem`] with tuned maintenance datums (the rows'
    /// fast chains).
    pub async fn mem_tuned(tag: &str, repo_id: &str, maintenance: MaintenanceOpts) -> Harness {
        let root = unique_root(tag);
        let options = AssemblyOptions {
            maintenance,
            ..AssemblyOptions::new(root.join("work"), root.join("large"))
        };
        let assembly = SubstrateAssembly::mem(
            Arc::new(alkstore_mem::MemStore::new()),
            Arc::new(alkblobs::Store::mem_only()),
            options,
        )
        .await
        .unwrap_or_else(|error| panic!("the {tag} mem assembly composes: {error:?}"));
        Harness {
            assembly,
            record: public_granted_record(repo_id, &writer_identity("u-writer")),
            identity: writer_identity("u-writer"),
            root,
        }
        .with_registered_repo()
        .await
    }

    /// The sqlite assembly (the default single-host shape, the
    /// `sqlite` feature; the co-tenant file under the harness root)
    /// with the row's pack-serving shape.
    #[cfg(feature = "sqlite")]
    pub async fn sqlite(tag: &str, repo_id: &str, pack_serving: PackServing) -> Harness {
        let root = unique_root(tag);
        let options = AssemblyOptions {
            pack_serving,
            ..AssemblyOptions::new(root.join("work"), root.join("large"))
        };
        let assembly = SubstrateAssembly::sqlite(root.join("assembly.db"), options)
            .await
            .unwrap_or_else(|error| panic!("the {tag} sqlite assembly composes: {error:?}"));
        Harness {
            assembly,
            record: public_granted_record(repo_id, &writer_identity("u-writer")),
            identity: writer_identity("u-writer"),
            root,
        }
        .with_registered_repo()
        .await
    }

    async fn with_registered_repo(self) -> Harness {
        self.assembly
            .registry_store()
            .put_repo(&self.record)
            .await
            .unwrap_or_else(|error| panic!("the harness's repo registers: {error:?}"));
        self
    }

    /// The git:// daemon URL for the harness's repo.
    pub fn git_url(&self, addr: std::net::SocketAddr) -> String {
        format!("git://{addr}/{}", self.record.repo_id)
    }

    /// The working-area root the assembly options opened (the rows'
    /// residue assertions read it after their pushes).
    pub fn work_root(&self) -> PathBuf {
        self.root.join("work")
    }

    /// The harness root (temporary; removed at drop).
    pub fn root(&self) -> &Path {
        &self.root
    }

    /// The smart-http URL for the harness's repo.
    pub fn http_url(&self, addr: std::net::SocketAddr) -> String {
        format!("http://{addr}/{}", self.record.repo_id)
    }
}

/// The doors' byte capture (the tee) — both relay directions, order
/// preserved, `to_server` the client's sent bytes and `to_client` the
/// server's responses.
#[derive(Clone, Default)]
pub struct StreamCapture {
    pub to_server: Arc<Mutex<Vec<u8>>>,
    pub to_client: Arc<Mutex<Vec<u8>>>,
}

impl StreamCapture {
    pub fn new() -> StreamCapture {
        StreamCapture::default()
    }

    pub fn to_server(&self) -> Vec<u8> {
        self.to_server.lock().clone()
    }

    pub fn to_client(&self) -> Vec<u8> {
        self.to_client.lock().clone()
    }
}

/// The gated hold's rendezvous: `held` fires once the door is holding
/// the client's command flush onward; `release` tells it to forward
/// (a flag the waiter polls + wakes on — the ordered state change).
#[derive(Clone, Default)]
pub struct GateRendezvous {
    pub held: Arc<tokio::sync::Notify>,
    release_notify: Arc<tokio::sync::Notify>,
    released: Arc<AtomicBool>,
}

impl GateRendezvous {
    pub fn new() -> GateRendezvous {
        GateRendezvous::default()
    }

    /// The row's release (after the clashing push landed).
    pub fn release(&self) {
        self.released.store(true, AtomicOrdering::SeqCst);
        self.release_notify.notify_one();
    }

    /// Waits for the release (the door's side; the flag is queried
    /// first so an already-fired release never blocks).
    async fn wait_release(&self) {
        while !self.released.load(AtomicOrdering::SeqCst) {
            self.release_notify.notified().await;
        }
    }

    /// Whether the release already fired.
    pub fn is_released(&self) -> bool {
        self.released.load(AtomicOrdering::SeqCst)
    }
}

/// The door options: the tee and the gate, per door instance.
#[derive(Clone, Default)]
pub struct DoorOptions {
    pub capture: Option<StreamCapture>,
    pub gate: Option<GateRendezvous>,
}

/// A running door's handle: its bound address and the accept loop's
/// task (aborted by the row at the end).
pub struct DoorHandle {
    pub addr: std::net::SocketAddr,
    pub accept: JoinHandle<()>,
}

/// The git-daemon-framed duplex door (the core's real-client shape over
/// the substrate backend): accepts git:// connections, reads the
/// daemon request line, resolves + authorizes (the injected writer
/// identity is the door's authenticated-peer shape), and relays the
/// framed transport into the landed [`serve_duplex`] machine.
pub async fn spawn_daemon_door(harness: &Harness, options: DoorOptions) -> DoorHandle {
    let listener = TcpListener::bind("127.0.0.1:0")
        .await
        .unwrap_or_else(|error| panic!("the duplex door binds: {error}"));
    let addr = listener.local_addr().expect("the door's local addr");
    let registry: Arc<dyn GitRegistry> = harness.assembly.registry().clone();
    let refs: Arc<dyn GitRefs> = harness.assembly.refs().clone();
    let packgen: Arc<dyn GitPackGen> = harness.assembly.packgen().clone();
    let ingest: Arc<dyn GitPackIngest> = harness.assembly.ingest().clone();
    let identity = harness.identity.clone();
    let limits = Limits::default();

    let accept = tokio::spawn(async move {
        loop {
            let Ok((conn, _peer)) = listener.accept().await else {
                break;
            };
            let registry = Arc::clone(&registry);
            let refs = Arc::clone(&refs);
            let packgen = Arc::clone(&packgen);
            let ingest = Arc::clone(&ingest);
            let identity = identity.clone();
            let options = options.clone();
            tokio::spawn(serve_daemon_conn(
                conn, identity, registry, refs, packgen, ingest, limits, options,
            ));
        }
    });
    DoorHandle { addr, accept }
}

/// One connection's work: the daemon frame, the door's resolve→authorize
/// front, the relay tee (with the armed gate when present), and the
/// serving machine under the door-level hygiene timeout.
#[allow(clippy::too_many_arguments)]
async fn serve_daemon_conn(
    conn: TcpStream,
    identity: Identity,
    registry: Arc<dyn GitRegistry>,
    refs: Arc<dyn GitRefs>,
    packgen: Arc<dyn GitPackGen>,
    ingest: Arc<dyn GitPackIngest>,
    limits: Limits,
    options: DoorOptions,
) {
    let session = serve_daemon_conn_inner(
        conn, identity, registry, refs, packgen, ingest, limits, options,
    );
    match tokio::time::timeout(DOOR_CONN_TIMEOUT, session).await {
        Ok(Ok(())) => {}
        Ok(Err(error)) => eprintln!("the duplex door session ended: {error:?}"),
        Err(_elapsed) => eprintln!("the duplex door session exceeded the hygiene timeout"),
    }
}

#[allow(clippy::too_many_arguments)]
async fn serve_daemon_conn_inner(
    mut conn: TcpStream,
    identity: Identity,
    registry: Arc<dyn GitRegistry>,
    refs: Arc<dyn GitRefs>,
    packgen: Arc<dyn GitPackGen>,
    ingest: Arc<dyn GitPackIngest>,
    limits: Limits,
    options: DoorOptions,
) -> Result<(), SubstrateError> {
    let Some(frame) = read_daemon_frame(&mut conn).await else {
        return Ok(());
    };
    let Some((service, repo_id)) = parse_daemon_frame(&frame) else {
        return Ok(());
    };
    let action = if service.is_write() {
        AccessAction::Write
    } else {
        AccessAction::Read
    };
    let Ok(record) = registry.resolve(&repo_id).await else {
        return Ok(());
    };
    let Some(marker) = AuthorizedRepo::authorize(record, Some(&identity), action) else {
        return Ok(());
    };
    let (pair_client, pair_server) = duplex(1 << 20);
    let pipeline = Arc::new(Semaphore::new(limits.max_concurrent_blocking_tasks));

    let (mut client_read, mut client_write) = tokio::io::split(conn);
    let (mut server_read, mut server_write) = tokio::io::split(pair_client);

    let up = relay_up(
        &mut client_read,
        &mut server_write,
        options.capture.clone(),
        options.gate.clone(),
    );
    let down = telemetry_copy(
        &mut server_read,
        &mut client_write,
        options
            .capture
            .as_ref()
            .map(|capture| capture.to_client.clone()),
    );
    let serve = serve_duplex(
        &marker,
        pair_server,
        limits,
        service,
        refs,
        packgen,
        ingest,
        false,
        pipeline,
    );
    let (up, down, serve) = tokio::join!(up, down, serve);
    down?;
    up?;
    serve
}

/// Reads the daemon request line (the one pkt-line; bounded window).
/// A non-pkt-frame connection (a misrouted client — nothing of the
/// git protocol's shape) closes quietly.
async fn read_daemon_frame(conn: &mut TcpStream) -> Option<Vec<u8>> {
    let mut header = [0u8; 4];
    tokio::time::timeout(DOOR_CONN_TIMEOUT, conn.read_exact(&mut header))
        .await
        .ok()?
        .ok()?;
    let length_text = std::str::from_utf8(&header).ok()?;
    if !length_text.bytes().all(|byte| byte.is_ascii_hexdigit()) {
        return None;
    }
    let len = usize::from_str_radix(length_text, 16).ok()?;
    if !(4..=(1 << 16)).contains(&len) {
        return None;
    }
    let mut payload = vec![0u8; len - 4];
    tokio::time::timeout(DOOR_CONN_TIMEOUT, conn.read_exact(&mut payload))
        .await
        .ok()?
        .ok()?;
    Some(payload)
}

/// Parses the daemon request line (ADR-016 §2's grammar, both
/// services): the `<service> <repo>` first segment, the repo id
/// stripped of its leading slash; the upload-pack sessions must carry
/// `version=2` (fetch is V2-only, ADR-003), the receive-pack sessions
/// carry none (the push path is V0-framed, ADR-013 §1).
pub fn parse_daemon_frame(frame: &[u8]) -> Option<(Service, String)> {
    let text = String::from_utf8(frame.to_vec()).ok()?;
    let mut segments = text.split('\0');
    let (service_text, repo_path) = segments.next()?.split_once(' ')?;
    let service = service_text.parse::<Service>().ok()?;
    let repo_id = repo_path.strip_prefix('/').unwrap_or(repo_path);
    if repo_id.is_empty() {
        return None;
    }
    let version_2 = segments.any(|segment| segment == "version=2");
    if service == Service::UploadPack && !version_2 {
        return None;
    }
    Some((service, repo_id.to_owned()))
}

/// The client→server relay: the plain tee copy when unarmed; when a
/// gate is armed, the command pkt-lines forward live and everything
/// from the command flush onward (which includes the pack) is staged
/// until the gate releases — the deterministic clashing-client hold.
async fn relay_up(
    src: &mut (impl tokio::io::AsyncRead + Unpin),
    dst: &mut (impl tokio::io::AsyncWrite + Unpin),
    capture: Option<StreamCapture>,
    gate: Option<GateRendezvous>,
) -> std::io::Result<()> {
    let to_server = capture.as_ref().map(|capture| capture.to_server.clone());
    let push_capture = |bytes: &[u8]| {
        if let Some(to_server) = &to_server {
            to_server.lock().extend_from_slice(bytes);
        }
    };
    if let Some(gate) = &gate {
        let mut staged = Vec::new();
        let mut held = false;
        while !held {
            let mut header = [0u8; 4];
            if src.read_exact(&mut header).await.is_err() {
                return Ok(());
            }
            let Ok(len) =
                usize::from_str_radix(std::str::from_utf8(&header).expect("hex header"), 16)
            else {
                return Ok(());
            };
            let mut payload = vec![0u8; len.saturating_sub(4)];
            if len >= 4 && src.read_exact(&mut payload).await.is_err() {
                return Ok(());
            }
            if len < 4 {
                staged.extend_from_slice(&header);
                held = true;
            } else {
                let chunk = [&header[..], &payload].concat();
                push_capture(&chunk);
                dst.write_all(&chunk).await?;
            }
        }
        gate.held.notify_one();
        gate.wait_release().await;
        push_capture(&staged);
        dst.write_all(&staged).await?;
    }
    let mut chunk = vec![0u8; RELAY_CHUNK_BYTES];
    loop {
        match src.read(&mut chunk).await {
            Ok(0) => return Ok(()),
            Ok(n) => {
                push_capture(&chunk[..n]);
                dst.write_all(&chunk[..n]).await?;
            }
            Err(error) => return Err(error),
        }
    }
}

/// The plain one-direction relay (the server→git copy with the tee).
async fn telemetry_copy(
    src: &mut (impl tokio::io::AsyncRead + Unpin),
    dst: &mut (impl tokio::io::AsyncWrite + Unpin),
    capture: Option<Arc<Mutex<Vec<u8>>>>,
) -> std::io::Result<()> {
    let mut chunk = vec![0u8; RELAY_CHUNK_BYTES];
    loop {
        match src.read(&mut chunk).await {
            Ok(0) => return Ok(()),
            Ok(n) => {
                if let Some(capture) = &capture {
                    capture.lock().extend_from_slice(&chunk[..n]);
                }
                dst.write_all(&chunk[..n]).await?;
            }
            Err(error) => return Err(error),
        }
    }
}

/// The smart-http-framed stateless door (the core's service-stub shape,
/// upload-pack routes over the substrate backend): the anonymous public
/// read, keep-alive across the GET and the round POSTs, each request's
/// route resolving + authorizing (the ADR-007 front's per-request
/// posture).
pub async fn spawn_http_door(harness: &Harness) -> DoorHandle {
    let listener = TcpListener::bind("127.0.0.1:0")
        .await
        .unwrap_or_else(|error| panic!("the http door binds: {error}"));
    let addr = listener.local_addr().expect("the door's local addr");
    let registry: Arc<dyn GitRegistry> = harness.assembly.registry().clone();
    let refs: Arc<dyn GitRefs> = harness.assembly.refs().clone();
    let packgen: Arc<dyn GitPackGen> = harness.assembly.packgen().clone();
    let ingest: Arc<dyn GitPackIngest> = harness.assembly.ingest().clone();
    let limits = Limits::default();

    let accept = tokio::spawn(async move {
        loop {
            let Ok((conn, _peer)) = listener.accept().await else {
                break;
            };
            let registry = Arc::clone(&registry);
            let refs = Arc::clone(&refs);
            let packgen = Arc::clone(&packgen);
            let ingest = Arc::clone(&ingest);
            tokio::spawn(serve_http_conn(
                conn, registry, refs, packgen, ingest, limits,
            ));
        }
    });
    DoorHandle { addr, accept }
}

#[allow(clippy::too_many_arguments)]
async fn serve_http_conn(
    mut conn: TcpStream,
    registry: Arc<dyn GitRegistry>,
    refs: Arc<dyn GitRefs>,
    packgen: Arc<dyn GitPackGen>,
    ingest: Arc<dyn GitPackIngest>,
    limits: Limits,
) {
    let session = serve_http_conn_inner(&mut conn, registry, refs, packgen, ingest, limits);
    tokio::pin!(session);
    match tokio::time::timeout(DOOR_CONN_TIMEOUT, &mut session).await {
        Ok(Ok(())) | Ok(Err(())) | Err(_) => {}
    }
}

async fn serve_http_conn_inner(
    conn: &mut TcpStream,
    registry: Arc<dyn GitRegistry>,
    refs: Arc<dyn GitRefs>,
    packgen: Arc<dyn GitPackGen>,
    ingest: Arc<dyn GitPackIngest>,
    limits: Limits,
) -> Result<(), ()> {
    // Keep-alive: real curl reuses the connection across the GET and
    // the round POSTs; the per-connection read buffer carries any
    // pipelined remainder across requests.
    let mut buffer: Vec<u8> = Vec::new();
    loop {
        let Some(mut request) = read_http_request(conn, &mut buffer).await else {
            return Ok(());
        };
        let Some(repo_id) = request.path.as_deref().and_then(repo_id_from_path) else {
            return Ok(());
        };
        let Ok(record) = registry.resolve(&repo_id).await else {
            return Ok(());
        };
        let Some(marker) = AuthorizedRepo::authorize(record, None, AccessAction::Read) else {
            return Ok(());
        };
        match serve_http_request(
            conn,
            &mut request,
            &marker,
            &refs,
            &packgen,
            &ingest,
            limits,
        )
        .await
        {
            Some(true) => {}
            _ => return Ok(()),
        }
    }
}

/// Routes one http request over the landed serving entries (the GET
/// advertisement route and the one-command POST route); `Some(true)`
/// after a served response, `None` on a non-served shape (the rows'
/// clients only speak the served routes).
async fn serve_http_request(
    conn: &mut TcpStream,
    request: &mut HttpRequest,
    marker: &AuthorizedRepo,
    refs: &Arc<dyn GitRefs>,
    packgen: &Arc<dyn GitPackGen>,
    ingest: &Arc<dyn GitPackIngest>,
    limits: Limits,
) -> Option<bool> {
    let path = request.path.clone()?;
    if request.method == "GET" {
        if !path.ends_with("/info/refs")
            || request.query.as_deref() != Some("service=git-upload-pack")
        {
            return None;
        }
        let sink = Sink::default();
        serve_stateless_advertisement(
            marker,
            sink.clone(),
            Service::UploadPack,
            Arc::clone(refs),
            &limits,
            false,
        )
        .await
        .ok()?;
        respond(
            conn,
            "application/x-git-upload-pack-advertisement",
            &sink.body(),
        )
        .await;
        Some(true)
    } else if request.method == "POST" {
        if !path.ends_with("/git-upload-pack") {
            return None;
        }
        let body = std::mem::take(&mut request.body);
        let reader = BodyReader::new(stream::iter(vec![Ok::<Bytes, std::convert::Infallible>(
            Bytes::from(body),
        )]));
        let sink = Sink::default();
        serve_stateless_post(
            marker,
            reader,
            sink.clone(),
            limits,
            Service::UploadPack,
            Arc::clone(refs),
            Arc::clone(packgen),
            Arc::clone(ingest),
            false,
            Arc::new(Semaphore::new(limits.max_concurrent_blocking_tasks)),
        )
        .await
        .ok()?;
        respond(conn, "application/x-git-upload-pack-result", &sink.body()).await;
        Some(true)
    } else {
        None
    }
}

async fn respond(conn: &mut TcpStream, content_type: &str, body: &[u8]) {
    conn.write_all(
        format!(
            "HTTP/1.1 200 OK\r\nContent-Type: {content_type}\r\nCache-Control: no-cache\r\nContent-Length: {}\r\n\r\n",
            body.len()
        )
        .as_bytes(),
    )
    .await
    .expect("the door writes the head");
    conn.write_all(body)
        .await
        .expect("the door writes the body");
    conn.flush().await.expect("the door flushes the body");
}

/// One http/1.1 request head + Content-Length body (the rows' real curl
/// requests are Content-Length-framed; keep-alive pipelining keeps the
/// remainder in the connection's read buffer).
pub struct HttpRequest {
    pub method: String,
    pub path: Option<String>,
    pub query: Option<String>,
    pub body: Vec<u8>,
    /// The request's `Content-Length` when the head carried one (the
    /// LFS door's size-less-offer detection).
    pub content_length: Option<u64>,
    /// The request's `Authorization` header (the LFS door's
    /// batch/transfer credential resolution).
    pub authorization: Option<String>,
    /// The request's `Range` header (the LFS door's response framing).
    pub range: Option<String>,
}

/// One http/1.1 request head + Content-Length body (the rows' real curl
/// requests are Content-Length-framed; keep-alive pipelining keeps the
/// remainder in the connection's read buffer). Shared by the git and
/// LFS test doors; the `Expect: 100-continue` tolerance is the real
/// transfer clients' interim-response posture (an unanswered Expect
/// would stall an upload for the continue timeout).
pub(crate) async fn read_http_request(
    conn: &mut TcpStream,
    buffer: &mut Vec<u8>,
) -> Option<HttpRequest> {
    let mut chunk = [0u8; 4096];
    let head_end = loop {
        if let Some(position) = buffer.windows(4).position(|window| window == b"\r\n\r\n") {
            break position;
        }
        let read = conn.read(&mut chunk).await.ok()?;
        if read == 0 {
            return None;
        }
        buffer.extend_from_slice(&chunk[..read]);
        if buffer.len() > 64 * 1024 {
            return None;
        }
    };
    let head_text = String::from_utf8(buffer[..head_end].to_vec()).ok()?;
    let mut lines = head_text.split("\r\n");
    let request_line = lines.next()?;
    let mut parts = request_line.split(' ');
    let method = parts.next()?.to_owned();
    let path = parts.next()?;
    let (path, query) = match path.split_once('?') {
        Some((path, query)) => (path.to_owned(), Some(query.to_owned())),
        None => (path.to_owned(), None),
    };
    let mut content_length: Option<u64> = None;
    let mut expects_continue = false;
    let mut authorization: Option<String> = None;
    let mut range: Option<String> = None;
    for header in lines {
        let Some((name, value)) = header.split_once(':') else {
            continue;
        };
        if name.trim().eq_ignore_ascii_case("content-length") {
            content_length = value.trim().parse().ok();
        } else if name.trim().eq_ignore_ascii_case("expect")
            && value.trim().eq_ignore_ascii_case("100-continue")
        {
            expects_continue = true;
        } else if name.trim().eq_ignore_ascii_case("authorization") {
            authorization = Some(value.trim().to_owned());
        } else if name.trim().eq_ignore_ascii_case("range") {
            range = Some(value.trim().to_owned());
        }
    }
    let body_start = head_end + 4;
    let body_len = content_length.unwrap_or(0) as usize;
    if expects_continue && body_len > 0 {
        conn.write_all(b"HTTP/1.1 100 Continue\r\n\r\n")
            .await
            .expect("the interim continue writes");
    }
    while buffer.len() < body_start + body_len {
        let read = conn.read(&mut chunk).await.ok()?;
        if read == 0 {
            return None;
        }
        buffer.extend_from_slice(&chunk[..read]);
    }
    let body = buffer[body_start..body_start + body_len].to_vec();
    buffer.drain(..body_start + body_len);
    Some(HttpRequest {
        method,
        path: Some(path),
        query,
        body,
        content_length,
        authorization,
        range,
    })
}

/// The routed repo id: `/<repo>/info/refs` (GET) or
/// `/<repo>/git-upload-pack` (POST) — everything between the leading
/// slash and the suffix.
fn repo_id_from_path(path: &str) -> Option<String> {
    let trimmed = path.strip_prefix('/')?;
    let rest = trimmed
        .strip_suffix("/info/refs")
        .or_else(|| trimmed.strip_suffix("/git-upload-pack"))?;
    if rest.is_empty() {
        return None;
    }
    Some(rest.to_owned())
}

/// An async sink collecting an http response body (the stateless
/// serving entries' writer surface in these door shapes).
#[derive(Clone, Default)]
pub struct Sink(Arc<Mutex<Vec<u8>>>);

impl Sink {
    fn body(&self) -> Vec<u8> {
        self.0.lock().clone()
    }
}

impl FuturesAsyncWrite for Sink {
    fn poll_write(
        self: std::pin::Pin<&mut Self>,
        _cx: &mut std::task::Context<'_>,
        buf: &[u8],
    ) -> std::task::Poll<std::io::Result<usize>> {
        self.0.lock().extend_from_slice(buf);
        std::task::Poll::Ready(Ok(buf.len()))
    }

    fn poll_flush(
        self: std::pin::Pin<&mut Self>,
        _cx: &mut std::task::Context<'_>,
    ) -> std::task::Poll<std::io::Result<()>> {
        std::task::Poll::Ready(Ok(()))
    }

    fn poll_close(
        self: std::pin::Pin<&mut Self>,
        cx: &mut std::task::Context<'_>,
    ) -> std::task::Poll<std::io::Result<()>> {
        self.poll_flush(cx)
    }
}

// ——— the real git client (the rows' client; a test client only — the
// serving path spawns nothing, the crate convention) ———

/// The isolation env every real-git invocation runs under.
fn git_env() -> Vec<(String, String)> {
    vec![
        ("GIT_CONFIG_NOSYSTEM".to_owned(), "1".to_owned()),
        ("GIT_CONFIG_GLOBAL".to_owned(), "/dev/null".to_owned()),
        ("GIT_TERMINAL_PROMPT".to_owned(), "0".to_owned()),
    ]
}

/// The git command, or `None` when the binary is absent (the rows'
/// skip signal): the lookup scans `PATH` for an executable `git` (the
/// harness's env is otherwise fully isolated, so no other source can
/// resolve the binary).
pub fn git_command() -> Option<tokio::process::Command> {
    git_on_path()?;
    let mut command = tokio::process::Command::new("git");
    command
        .env_clear()
        .envs(git_env())
        .env("PATH", std::env::var("PATH").unwrap_or_default());
    Some(command)
}

fn git_on_path() -> Option<PathBuf> {
    let path = std::env::var_os("PATH")?;
    for dir in std::env::split_paths(&path) {
        let candidate = dir.join("git");
        if !candidate.is_file() {
            continue;
        }
        #[cfg(unix)]
        {
            use std::os::unix::fs::PermissionsExt;
            let executable = std::fs::metadata(&candidate)
                .map(|meta| meta.permissions().mode() & 0o111 != 0)
                .unwrap_or(false);
            if !executable {
                continue;
            }
        }
        return Some(candidate);
    }
    None
}

/// One real-git invocation's outcome.
pub struct GitRun {
    pub code: i32,
    pub stdout: String,
    pub stderr: String,
}

impl GitRun {
    /// Asserts success and returns the run.
    pub fn expect_ok(self) -> GitRun {
        assert_eq!(
            self.code, 0,
            "the git invocation failed: {}{}",
            self.stdout, self.stderr
        );
        self
    }

    /// The run's success.
    pub fn ok(&self) -> bool {
        self.code == 0
    }

    /// The stderr's text.
    pub fn stderr(&self) -> &str {
        &self.stderr
    }
}

/// Runs one real-git invocation (isolated env, wall clock); `None`
/// when the `git` binary is absent (the rows' skip signal). A timeout
/// panics with the argument list (process hygiene: a deadlocked client
/// fails the row instead of hanging it).
pub async fn git_run(cwd: Option<&Path>, args: &[&str]) -> Option<GitRun> {
    if git_command().is_none() {
        eprintln!("skipping row: no `git` binary on PATH");
        return None;
    }
    let mut command = git_command().expect("the git binary the row resolved");
    if let Some(cwd) = cwd {
        command.current_dir(cwd);
    }
    command.args(args).stdin(Stdio::null());
    let output = match tokio::time::timeout(GIT_RUN_TIMEOUT, command.output()).await {
        Ok(output) => output.expect("the git invocation spawns"),
        Err(_elapsed) => panic!("the git invocation exceeded {GIT_RUN_TIMEOUT:?}: {args:?}"),
    };
    Some(GitRun {
        code: output.status.code().unwrap_or(1),
        stdout: String::from_utf8_lossy(&output.stdout).to_string(),
        stderr: String::from_utf8_lossy(&output.stderr).to_string(),
    })
}

/// Runs a row's git invocation expecting success; `None` propagates as
/// the row's clean skip (the caller returns immediately).
pub async fn git(cwd: Option<&Path>, args: &[&str]) -> Option<GitRun> {
    git_run(cwd, args).await.map(GitRun::expect_ok)
}

/// Runs a git invocation with extra env (the git-lfs rows' skip-smudge
/// shape); the same isolation and wall clock.
pub async fn git_with(
    cwd: Option<&Path>,
    args: &[&str],
    extra_env: &[(&str, &str)],
) -> Option<GitRun> {
    if git_command().is_none() {
        eprintln!("skipping row: no `git` binary on PATH");
        return None;
    }
    let mut command = git_command().expect("the git binary the row resolved");
    if let Some(cwd) = cwd {
        command.current_dir(cwd);
    }
    for (name, value) in extra_env {
        command.env(name, value);
    }
    command.args(args).stdin(Stdio::null());
    let output = match tokio::time::timeout(GIT_RUN_TIMEOUT, command.output()).await {
        Ok(output) => output.expect("the git invocation spawns"),
        Err(_elapsed) => panic!("the git invocation exceeded {GIT_RUN_TIMEOUT:?}: {args:?}"),
    };
    let run = GitRun {
        code: output.status.code().unwrap_or(1),
        stdout: String::from_utf8_lossy(&output.stdout).to_string(),
        stderr: String::from_utf8_lossy(&output.stderr).to_string(),
    };
    Some(run.expect_ok())
}

// ——— the wire-decode helpers (the captured streams' evidence) ———

/// A captured pkt-line: a data line (its payload) or the flush/control
/// frame (`len < 4`).
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Pkt {
    /// A data pkt-line (its payload).
    Line(Vec<u8>),
    /// The flush packet (`0000`) or the delimiter class (`0001`–`0003`).
    Control,
}

/// Splits a captured pkt-line stream (a trailing partial line — a
/// capture that races a still-writing peer — is carried as-is).
pub fn pkt_stream(body: &[u8]) -> Vec<Pkt> {
    let mut lines = Vec::new();
    let mut cursor = body;
    while cursor.len() >= 4 {
        let len = usize::from_str_radix(std::str::from_utf8(&cursor[..4]).expect("hex header"), 16)
            .expect("hex length");
        if len < 4 {
            lines.push(Pkt::Control);
            cursor = &cursor[4..];
            continue;
        }
        if cursor.len() < len {
            lines.push(Pkt::Line(cursor[4..].to_vec()));
            break;
        }
        lines.push(Pkt::Line(cursor[4..len].to_vec()));
        cursor = &cursor[len..];
    }
    lines
}

/// The sideband status report's lines, decoded from a captured
/// server→client stream (the ADR-013 §8 shape): all band-1 payloads
/// concatenate into the report's inner pkt-line framing (the lines and
/// the inner flush); the V0 advertisement lines and the outer flushes
/// skip. The band-1 chunks' presence is the selected sideband framing.
pub fn sideband_report_lines(to_client: &[u8]) -> Vec<String> {
    let mut payload = Vec::new();
    for pkt in pkt_stream(to_client) {
        if let Pkt::Line(line) = &pkt {
            let (band, rest) = line.split_first().expect("band chunk nonempty");
            if *band == 1 {
                payload.extend_from_slice(rest);
            }
        }
    }
    let mut inner = pkt_stream(&payload);
    let last = inner.pop().expect("the report's framing is non-empty");
    assert!(
        matches!(last, Pkt::Control),
        "the report's inner flush terminates it"
    );
    inner
        .into_iter()
        .map(|pkt| match pkt {
            Pkt::Line(line) => String::from_utf8_lossy(&line)
                .trim_end_matches('\n')
                .to_owned(),
            Pkt::Control => String::new(),
        })
        .collect()
}

/// The captured client→server stream's pack shape — what the real
/// client actually sent (the thin-pack and packless rows' evidence).
pub struct SentPack {
    /// The pack header's declared object count.
    pub object_count: u32,
    /// The entry count the walk consumed.
    pub entry_count: usize,
    /// The full (non-delta) entries' oids — the objects sent whole.
    pub full_oids: Vec<gix_hash::ObjectId>,
    /// The delta entries, as `(base, kind)` — `kind` is
    /// `ThinBase` when the base is not among the pack's own sent
    /// objects (the thin shape: the base exists server-side only) and
    /// `InPack` otherwise.
    pub deltas: Vec<(gix_hash::ObjectId, DeltaKind)>,
}

/// A delta entry's base kind (the pack-objects encoding's two shapes).
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum DeltaKind {
    /// The base rides in this pack.
    InPack,
    /// The base is not in this pack — the thin shape.
    ThinBase,
}

/// Walks the captured client→server bytes for the pack stream (the raw
/// tail from the `PACK` magic onward): `None` when no pack was ever
/// sent — the delete-only push's captured shape.
pub fn extract_and_walk_client_pack(to_server: &[u8]) -> Option<SentPack> {
    let magic = to_server.windows(4).position(|window| window == b"PACK")?;
    parse_client_pack(&to_server[magic..])
}

/// Walks a V2 sha1 pack stream's entries (the captured client pack):
/// full entries inflate + re-hash to their oids; delta entries record
/// their bases (a captured ref-delta base is a thin candidate by
/// construction — its base object does not ride this pack; an
/// ofs-delta's base offset resolves against the pack's own entries).
/// A delta entry's own oid is not re-derived (it exists only as the
/// post-unpack server-side object) — the rows' land-asserts run through
/// the substrate's own front, not this walk.
pub fn parse_client_pack(pack: &[u8]) -> Option<SentPack> {
    let header: [u8; 12] = pack[..12].try_into().expect("the pack header present");
    let (version, object_count) =
        gix_pack::data::header::decode(&header).expect("the captured pack's header decodes");
    assert_eq!(version, gix_pack::data::Version::V2, "the client's pack");
    let entries = gix_pack::data::input::BytesToEntriesIter::new_from_header(
        std::io::Cursor::new(pack),
        gix_pack::data::input::Mode::Verify,
        gix_pack::data::input::EntryDataMode::Keep,
        gix_hash::Kind::Sha1,
    )
    .expect("the captured pack's stream starts");
    let mut sent = SentPack {
        object_count,
        entry_count: 0,
        full_oids: Vec::new(),
        deltas: Vec::new(),
    };
    let mut full_oids_by_offset: BTreeMap<u64, gix_hash::ObjectId> = BTreeMap::new();
    let mut pending_ofs_bases: Vec<(u64, usize)> = Vec::new();
    for entry in entries {
        let entry = entry.ok()?;
        sent.entry_count += 1;
        match &entry.header {
            gix_pack::data::entry::Header::Blob
            | gix_pack::data::entry::Header::Tree
            | gix_pack::data::entry::Header::Commit
            | gix_pack::data::entry::Header::Tag => {
                let kind = match entry.header {
                    gix_pack::data::entry::Header::Blob => gix_object::Kind::Blob,
                    gix_pack::data::entry::Header::Tree => gix_object::Kind::Tree,
                    gix_pack::data::entry::Header::Commit => gix_object::Kind::Commit,
                    _ => gix_object::Kind::Tag,
                };
                let body = inflate_entry_body(&entry)?;
                let oid = gix_object::compute_hash(gix_hash::Kind::Sha1, kind, &body)
                    .expect("the captured entry hashes");
                full_oids_by_offset.insert(entry.pack_offset, oid);
                sent.full_oids.push(oid);
            }
            gix_pack::data::entry::Header::OfsDelta { base_distance } => {
                let base_offset = entry.pack_offset.checked_sub(*base_distance)?;
                pending_ofs_bases.push((base_offset, sent.deltas.len()));
                sent.deltas.push((
                    gix_hash::ObjectId::null(gix_hash::Kind::Sha1),
                    DeltaKind::InPack,
                ));
            }
            gix_pack::data::entry::Header::RefDelta { base_id } => {
                sent.deltas.push((*base_id, DeltaKind::ThinBase));
            }
        }
    }
    for (base_offset, delta_index) in pending_ofs_bases {
        let base = full_oids_by_offset
            .get(&base_offset)
            .copied()
            .expect("the in-pack ofs-delta base resolves within the pack");
        sent.deltas[delta_index] = (base, DeltaKind::InPack);
    }
    Some(sent)
}

/// Inflates one captured entry's compressed payload (its zlib stream).
fn inflate_entry_body(entry: &gix_pack::data::input::Entry) -> Option<Vec<u8>> {
    let compressed = entry.compressed.as_ref()?;
    let mut out: Vec<u8> = Vec::new();
    let mut decompressor = gix_zlib::Decompress::new();
    let mut reader = std::io::BufReader::new(std::io::Cursor::new(compressed.as_slice()));
    let mut chunk = vec![0u8; 65536];
    loop {
        let written =
            gix_zlib::stream::inflate::read(&mut reader, &mut decompressor, &mut chunk).ok()?;
        out.extend_from_slice(&chunk[..written]);
        if written < chunk.len() {
            break;
        }
    }
    Some(out)
}

/// The pushed tips' ancestry exclusion helper for the row asserts:
/// `rev-list --objects` output, as an oid set.
pub fn rev_list_oids(stdout: &str) -> BTreeSet<gix_hash::ObjectId> {
    stdout
        .lines()
        .filter_map(|line| line.split(' ').next())
        .filter_map(|hex| gix_hash::ObjectId::from_hex(hex.as_bytes()).ok())
        .collect()
}