alkgit/tests/real_client.rs

//! The OQ-19 rider-3 real-client sessions (`core/w5-oq19-real-client-tokens`):
//! real `git` (the 2.43.0 capture-session shape — the OQ-02/OQ-04
//! capture posture) driven against the landed serving entries over both
//! substrate doors — the git-daemon-framed duplex door (POC-1's tcp
//! bridge shape, the frame `git-upload-pack <repo>\0host=…\0\0version=2\0`)
//! and the smart-http-framed stateless door (POC-3's service-stub shape).
//!
//! The rows assert the rider's two spec-level calls against the real
//! client:
//!
//! - **`include-tag` — accept-and-honor-by-omission**: every configured
//!   fetch sends the token (transport.c v2.43.0 sets `args.include_tag`
//!   from the follow-tags transport option whenever the refspec carries
//!   tracking destinations; the V2 send path has no advertisement gate),
//!   and there is no `fetch=` feature slot in V2's grammar to advertise
//!   or retract it — the decline could not inform a conforming client,
//!   so the machine accepts it and changes nothing (the `thin-pack`
//!   shape). The session: a real clone, then an incremental real fetch —
//!   the include-tag sender — completes over each door, with the sent
//!   token asserted in the captured packet trace.
//! - **`session-id=` — honest retraction (the decline stays)**: git's V2
//!   client sends the pre-line only when the server advertises the
//!   `session-id` capability (`fetch-pack.c`:
//!   `advertise_sid && server_supports_v2("session-id")`); this
//!   advertisement never declares it, so the decline is unreachable by
//!   a conforming client — the row runs the real client with
//!   `transfer.advertiseSid=true` and asserts the session completes
//!   with `session-id` absent from the capture (the client honors the
//!   advertisement's silence).
//!
//! The rows skip quietly (with a note) when the `git` binary is absent —
//! the sessions need a real client by definition; every other row in
//! this corpus runs in-process.
//!
//! Rows are names + `Contract stamp:` versions, per
//! `tests/common` §The row/stamp discipline.

mod common;

use std::collections::BTreeMap;
use std::path::PathBuf;
use std::pin::Pin;
use std::process::Stdio;
use std::sync::Arc;
use std::sync::Mutex as StdMutex;
use std::time::Duration;

use bytes::Bytes;
use futures::io::AsyncWrite as FuturesAsyncWrite;
use gix_hash::ObjectId;
use tokio::io::{copy_bidirectional, duplex, AsyncReadExt, AsyncWriteExt};
use tokio::net::{TcpListener, TcpStream};
use tokio::sync::Semaphore;
use tokio::task::JoinHandle;

use alkgit::{
    serve_duplex_resolved, serve_stateless_advertisement, serve_stateless_post, AccessAction,
    AuthorizedRepo, BodyReader, GitPackGen, GitPackIngest, GitRefs, GitRegistry, GitRegistryStore,
    Limits, RefUpdate, RepoRecord, Service, SubstrateError, Visibility,
};
use common::{
    blob_fixture, commit_fixture, tag_fixture, tree_fixture, MemObjects, MemPackGen, MemPackIngest,
    MemRefs, MemRegistry,
};

/// The rider-3 fixture: a public repo (the anonymous first-class read
/// path, ADR-011 §3 — the git:// and http doors carry no identity beyond
/// the daemon frame) over the doubles, seeded with the two-commit chain
/// plus an annotated tag (the full clone shape: heads + tag ref), and
/// the increment objects for the configured-fetch sessions.
struct Fixture {
    registry: MemRegistry,
    refs: MemRefs,
    gen: Arc<MemPackGen>,
    ingest: Arc<MemPackIngest>,
    commit_b: ObjectId,
    tag_v1: ObjectId,
    /// The pre-built increment: `commit_c` (parent `commit_b`) — the
    /// configured fetch's new tip, applied to `refs/heads/main` by the
    /// row before its fetch session starts.
    commit_c: ObjectId,
    /// The rider-1 unborn repo (`owner/unborn`, the N-5/OQ-19 rider-1
    /// fixture): registered, zero ref rows, the HEAD symref row
    /// unborn (`oid: None`) at `refs/heads/main` — the fixture the
    /// unborn ls-refs sessions drive.
    unborn_repo_id: &'static str,
}

async fn fixture() -> Fixture {
    let repo = RepoRecord {
        repo_id: "owner/doubled".to_owned(),
        visibility: Visibility::Public,
        grants: BTreeMap::new(),
    };
    let registry = MemRegistry::default();
    registry
        .put_repo(&repo)
        .await
        .expect("the fixture repo registers");
    let unborn_repo_id = "owner/unborn";
    registry
        .put_repo(&RepoRecord {
            repo_id: unborn_repo_id.to_owned(),
            visibility: Visibility::Public,
            grants: BTreeMap::new(),
        })
        .await
        .expect("the unborn fixture repo registers");
    let refs = MemRefs::default();
    refs.set_head_symref(
        &RepoRecord {
            repo_id: unborn_repo_id.to_owned(),
            visibility: Visibility::Public,
            grants: BTreeMap::new(),
        },
        Some("refs/heads/main".to_owned()),
    );
    let objects = MemObjects::default();
    let gen = Arc::new(MemPackGen {
        objects: objects.clone(),
    });
    let ingest = Arc::new(MemPackIngest {
        objects: objects.clone(),
    });

    let (blob_a, blob_a_bytes) = blob_fixture(b"alpha");
    let (blob_b, blob_b_bytes) = blob_fixture(b"beta");
    let (tree_a, tree_a_bytes) =
        tree_fixture(&[(gix_object::tree::EntryKind::Blob, "a.txt", blob_a)]);
    let (tree_b, tree_b_bytes) =
        tree_fixture(&[(gix_object::tree::EntryKind::Blob, "b.txt", blob_b)]);
    let (commit_a, commit_a_bytes) = commit_fixture(tree_a, &[], "alpha");
    let (commit_b, commit_b_bytes) = commit_fixture(tree_b, &[commit_a], "beta");
    let (tag_v1, tag_bytes) = tag_fixture(commit_b, gix_object::Kind::Commit, "v1");
    for (id, bytes) in [
        (blob_a, blob_a_bytes),
        (blob_b, blob_b_bytes),
        (tree_a, tree_a_bytes),
        (tree_b, tree_b_bytes),
        (commit_a, commit_a_bytes),
        (commit_b, commit_b_bytes),
        (tag_v1, tag_bytes),
    ] {
        objects.put(&repo, id, bytes);
    }
    refs.apply_updates(
        &repo,
        vec![
            RefUpdate {
                name: "refs/heads/main".to_owned(),
                expected: None,
                new: Some(commit_b),
            },
            RefUpdate {
                name: "refs/heads/old".to_owned(),
                expected: None,
                new: Some(commit_a),
            },
            RefUpdate {
                name: "refs/tags/v1".to_owned(),
                expected: None,
                new: Some(tag_v1),
            },
        ],
        false,
    )
    .await
    .expect("the fixture refs apply");

    let (blob_c, blob_c_bytes) = blob_fixture(b"gamma");
    let (tree_c, tree_c_bytes) =
        tree_fixture(&[(gix_object::tree::EntryKind::Blob, "c.txt", blob_c)]);
    let (commit_c, commit_c_bytes) = commit_fixture(tree_c, &[commit_b], "gamma");
    for (id, bytes) in [
        (blob_c, blob_c_bytes),
        (tree_c, tree_c_bytes),
        (commit_c, commit_c_bytes),
    ] {
        objects.put(&repo, id, bytes);
    }

    Fixture {
        registry,
        refs,
        gen,
        ingest,
        commit_b,
        tag_v1,
        commit_c,
        unborn_repo_id,
    }
}

/// The fixture's refs double fronted by the born-HEAD line the ls-refs
/// grammar serves (a real backend lists the symref-backed HEAD row;
/// without it the real client's default-branch heuristic checks out an
/// unborn fallback — the empty-checkout the real clone rows observed).
/// The rider-1 unborn repo is exempt: the inner [`MemRefs`]'s own
/// unborn-HEAD row (`oid: None` at the symref target, zero ref rows)
/// serves as the sole listing.
struct HeadRefs {
    inner: MemRefs,
    head_oid: ObjectId,
    head_target: String,
    unborn_repo: Option<String>,
}

#[async_trait::async_trait]
impl GitRefs for HeadRefs {
    async fn list_refs(
        &self,
        repo: &RepoRecord,
    ) -> Result<Vec<alkgit::RefLine>, alkgit::StorageError> {
        let mut listed = self.inner.list_refs(repo).await?;
        if self
            .unborn_repo
            .as_deref()
            .is_some_and(|id| id == repo.repo_id)
        {
            return Ok(listed);
        }
        listed.insert(
            0,
            alkgit::RefLine {
                name: "HEAD".to_owned(),
                oid: Some(self.head_oid),
                peeled: None,
                symref_target: Some(self.head_target.clone()),
            },
        );
        Ok(listed)
    }

    async fn apply_updates(
        &self,
        repo: &RepoRecord,
        updates: Vec<RefUpdate>,
        atomic: bool,
    ) -> Result<Vec<alkgit::RefOutcome>, alkgit::StorageError> {
        self.inner.apply_updates(repo, updates, atomic).await
    }
}

/// The git-daemon-framed duplex door (POC-1's validated tcp bridge
/// shape): the first pkt-line is the daemon handshake
/// `git-upload-pack <repo>\0host=…\0\0version=2\0` — the door resolves
/// the in-band repo id against the registry (the ADR-007
/// resolve→authorize front before any protocol byte) and hands the
/// stream to [`serve_duplex_resolved`], relaying the transport through
/// a bounded in-memory pipe. The half-duplex shutdown real git performs
/// while reading the pack (`close(fd[1])`, fetch-pack.c's GET_PACK
/// state) propagates cleanly through the bidirectional relay.
///
/// This door is the test's own transport shape (doors are family
/// infrastructure, convention 15) — the serving entry below it is the
/// landed machine. The door serves only `git-upload-pack` sessions (the
/// rider's fetch sessions); any other frame closes quietly.
async fn spawn_daemon_door(fx: &Fixture) -> (std::net::SocketAddr, JoinHandle<()>) {
    let listener = TcpListener::bind("127.0.0.1:0")
        .await
        .expect("the door binds");
    let addr = listener.local_addr().expect("the door's local addr");
    let registry: Arc<dyn GitRegistry> = Arc::new(MemRegistry::clone(&fx.registry));
    let refs: Arc<dyn GitRefs> = Arc::new(HeadRefs {
        inner: MemRefs::clone(&fx.refs),
        head_oid: fx.commit_b,
        head_target: "refs/heads/main".to_owned(),
        unborn_repo: Some(fx.unborn_repo_id.to_owned()),
    });
    let gen: Arc<dyn GitPackGen> = fx.gen.clone();
    let ingest: Arc<dyn GitPackIngest> = fx.ingest.clone();
    let limits = Limits::default();

    let handle = tokio::spawn(async move {
        loop {
            let Ok((mut conn, _peer)) = listener.accept().await else {
                break;
            };
            let registry = Arc::clone(&registry);
            let refs = Arc::clone(&refs);
            let gen = Arc::clone(&gen);
            let ingest = Arc::clone(&ingest);
            tokio::spawn(async move {
                let Some(frame) = read_daemon_frame(&mut conn).await else {
                    return;
                };
                let Some(repo_id) = parse_daemon_frame(&frame) else {
                    return;
                };
                // No door-side marker: the resolve→authorize front is
                // the entry's (ADR-007) — a denied resolution closes
                // quietly, the generic denial precedes any protocol byte.
                let (mut pair_client, pair_server) = duplex(1 << 20);
                let (relay, session) = tokio::join!(
                    copy_bidirectional(&mut conn, &mut pair_client),
                    serve_duplex_resolved(
                        &repo_id,
                        None,
                        Service::UploadPack,
                        pair_server,
                        limits,
                        registry,
                        refs,
                        gen,
                        ingest,
                        false,
                        Arc::new(Semaphore::new(limits.max_concurrent_blocking_tasks)),
                    )
                );
                let _ = relay;
                if let Err(error) = session {
                    eprintln!("the duplex session returned: {error:?}");
                }
            });
        }
    });
    (addr, handle)
}

/// Reads the daemon handshake's one pkt-line (bounded window); `None`
/// when the peer closed or the frame never arrived.
async fn read_daemon_frame(conn: &mut TcpStream) -> Option<Vec<u8>> {
    let mut header = [0u8; 4];
    tokio::time::timeout(Duration::from_secs(10), conn.read_exact(&mut header))
        .await
        .ok()?
        .ok()?;
    let len = usize::from_str_radix(std::str::from_utf8(&header).expect("hex header"), 16)
        .expect("hex length");
    if !(4..=1 << 16).contains(&len) {
        return None;
    }
    let mut payload = vec![0u8; len - 4];
    tokio::time::timeout(Duration::from_secs(10), conn.read_exact(&mut payload))
        .await
        .ok()?
        .ok()?;
    Some(payload)
}

/// Parses the daemon handshake frame (POC-1's observed shape): the
/// `<service> <repo>` first segment (the repo id stripped of its
/// leading slash), and the `version=2` request riding the NUL-separated
/// extras. `None` for any other service, shape, or protocol request.
fn parse_daemon_frame(frame: &[u8]) -> Option<String> {
    let text = String::from_utf8(frame.to_vec()).ok()?;
    let mut segments = text.split('\0');
    let (service, repo_path) = segments.next()?.split_once(' ')?;
    if service != "git-upload-pack" {
        return None;
    }
    let repo_id = repo_path.strip_prefix('/').unwrap_or(repo_path);
    if repo_id.is_empty() || !segments.any(|segment| segment == "version=2") {
        return None;
    }
    Some(repo_id.to_owned())
}

/// The smart-http-framed stateless door (POC-3's service-stub shape):
/// `GET /<repo>/info/refs?service=git-upload-pack` serves
/// [`serve_stateless_advertisement`] (the smart prefix + the V2 dump),
/// `POST /<repo>/git-upload-pack` serves [`serve_stateless_post`] — one
/// POST body is one command invocation, the request body rides
/// [`BodyReader`], the response body is the collected [`VecSink`] bytes
/// with the door's Content-Type and a Content-Length (real curl
/// keep-alives across the GET and the round POSTs). The resolve→
/// authorize front is the door's ADR-007 check (anonymous public read,
/// ADR-011 §3).
async fn spawn_http_door(fx: &Fixture) -> (std::net::SocketAddr, JoinHandle<()>) {
    let listener = TcpListener::bind("127.0.0.1:0")
        .await
        .expect("the door binds");
    let addr = listener.local_addr().expect("the door's local addr");
    let registry: Arc<dyn GitRegistry> = Arc::new(MemRegistry::clone(&fx.registry));
    let repo_ids = ["owner/doubled".to_owned(), fx.unborn_repo_id.to_owned()]
        .into_iter()
        .collect::<std::collections::BTreeSet<String>>();
    let refs: Arc<dyn GitRefs> = Arc::new(HeadRefs {
        inner: MemRefs::clone(&fx.refs),
        head_oid: fx.commit_b,
        head_target: "refs/heads/main".to_owned(),
        unborn_repo: Some(fx.unborn_repo_id.to_owned()),
    });
    let gen: Arc<dyn GitPackGen> = fx.gen.clone();
    let ingest: Arc<dyn GitPackIngest> = fx.ingest.clone();
    let limits = Limits::default();

    let handle = tokio::spawn(async move {
        loop {
            let Ok((mut conn, _peer)) = listener.accept().await else {
                break;
            };
            let registry = Arc::clone(&registry);
            let refs = Arc::clone(&refs);
            let gen = Arc::clone(&gen);
            let ingest = Arc::clone(&ingest);
            let repo_ids = repo_ids.clone();
            tokio::spawn(async move {
                // 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. The routed repo resolves+authorizes per
                // request (the stateless POSTs are one-shot rounds —
                // one resolve→authorize front per request, ADR-007).
                let mut buffer: Vec<u8> = Vec::new();
                loop {
                    let Some(mut request) = read_http_request(&mut conn, &mut buffer).await else {
                        return;
                    };
                    let Some(repo_id) = request
                        .path
                        .as_deref()
                        .and_then(repo_id_from_path)
                        .filter(|id| repo_ids.contains(id))
                    else {
                        return;
                    };
                    let Ok(record) = registry.resolve(&repo_id).await else {
                        return;
                    };
                    let Some(marker) = AuthorizedRepo::authorize(record, None, AccessAction::Read)
                    else {
                        return;
                    };
                    match serve_http_request(
                        &mut conn,
                        &mut request,
                        &marker,
                        &refs,
                        &gen,
                        &ingest,
                        limits,
                    )
                    .await
                    {
                        Some(true) => {}
                        _ => return,
                    }
                }
            });
        }
    });
    (addr, handle)
}

/// The request path's repo id (the routed repo): the git smart path is
/// `/<repo>/info/refs` (GET) and `/<repo>/git-upload-pack` (POST); the
/// repo is 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())
}

/// One http/1.1 request: the request line, the bounded headers, and the
/// Content-Length body (the rows' real curl requests are
/// Content-Length-framed; a chunked upload would fail loudly in the
/// route's `None` legs).
struct HttpRequest {
    method: String,
    path: Option<String>,
    query: Option<String>,
    body: Vec<u8>,
}

/// Reads one full request (head + Content-Length body) from the
/// connection's persistent read buffer; the bytes beyond the body stay
/// in the buffer for the next keep-alive request. `None` when the peer
/// closed, the head is oversized, or the body never arrives.
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;
    for header in lines {
        let Some((name, value)) = header.split_once(':') else {
            continue;
        };
        let name = name.trim().to_ascii_lowercase();
        let value = value.trim();
        if name == "content-length" {
            content_length = value.parse().ok();
        }
    }
    let body_start = head_end + 4;
    let body_len = content_length.unwrap_or(0) as usize;
    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,
    })
}

/// Routes one http request (the door's POC-3 shapes); `Some(keep_alive)`
/// 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>,
    gen: &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 = VecSink::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(futures::stream::iter(vec![Ok::<
            Bytes,
            std::convert::Infallible,
        >(Bytes::from(body))]));
        let sink = VecSink::default();
        serve_stateless_post(
            marker,
            reader,
            sink.clone(),
            limits,
            Service::UploadPack,
            Arc::clone(refs),
            Arc::clone(gen),
            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]) {
    let head = 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()
    );
    conn.write_all(head.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");
}

/// An async sink collecting the response's framed bytes — the http
/// response body's surface shape (the stateless rows' VecSink).
#[derive(Clone, Default)]
struct VecSink(Arc<StdMutex<Vec<u8>>>);

impl VecSink {
    fn body(&self) -> Vec<u8> {
        self.0.lock().expect("sink is not poisoned").clone()
    }
}

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

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

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

/// The typed session error the doors' entries return (unused in the
/// happy-path rows — the doors close quietly on failures; the rows'
/// git-exit assertions are the visible surface).
#[allow(unused)]
fn session_error_note(error: SubstrateError) -> SubstrateError {
    error
}

/// The isolated workspace for one real-git session (per-row temp dir,
/// cleaned before use; leftovers from crashed runs are removed).
fn workspace(tag: &str) -> PathBuf {
    let dir =
        std::env::temp_dir().join(format!("alkgit-real-client-{}-{}", std::process::id(), tag));
    let _ = std::fs::remove_dir_all(&dir);
    std::fs::create_dir_all(&dir).expect("the workspace creates");
    dir
}

/// The isolation env every real-git session runs under (no system or
/// user config, no prompts).
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()),
    ]
}

fn git_command() -> Option<tokio::process::Command> {
    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)
}

/// The fresh git command for a row's follow-up assertions (the binary
/// resolved earlier in the row).
fn expect_git_command() -> tokio::process::Command {
    git_command().expect("the git binary resolved earlier in the row")
}

/// The clone's post-state: HEAD is the expected branch tip, the
/// annotated tag object resolves locally (the tag chain rode the pack),
/// and the clone's own pack ingestion validated the stream.
async fn assert_clone_completed(dst: &PathBuf, expected_head: ObjectId, tag: ObjectId) {
    let output = expect_git_command()
        .arg("-C")
        .arg(dst)
        .args(["rev-parse", "HEAD"])
        .stdin(Stdio::null())
        .output()
        .await
        .expect("rev-parse runs");
    assert!(
        output.status.success(),
        "rev-parse HEAD in the clone failed: {}",
        String::from_utf8_lossy(&output.stderr)
    );
    assert_eq!(
        String::from_utf8_lossy(&output.stdout).trim(),
        expected_head.to_string(),
        "the clone's HEAD is the expected branch tip"
    );
    let output = expect_git_command()
        .arg("-C")
        .arg(dst)
        .args(["cat-file", "-t"])
        .arg(tag.to_string())
        .stdin(Stdio::null())
        .output()
        .await
        .expect("cat-file runs");
    assert!(
        output.status.success(),
        "the annotated tag did not resolve in the clone: {}",
        String::from_utf8_lossy(&output.stderr)
    );
    assert_eq!(
        String::from_utf8_lossy(&output.stdout).trim(),
        "tag",
        "the tag resolves as an annotated tag object"
    );
}

/// The configured fetch's post-state: `refs/remotes/origin/main`
/// advanced to the new tip — the fetch's connectivity check passed, the
/// include-tag round completed, the client's ref updates applied.
async fn assert_fetch_completed(dst: &PathBuf, expected: ObjectId) {
    let mut command = expect_git_command();
    let output = command
        .arg("-C")
        .arg(dst)
        .args(["rev-parse", "refs/remotes/origin/main"])
        .stdin(Stdio::null())
        .output()
        .await
        .expect("rev-parse runs");
    assert!(
        output.status.success(),
        "the tracking ref did not resolve: {}",
        String::from_utf8_lossy(&output.stderr)
    );
    assert_eq!(
        String::from_utf8_lossy(&output.stdout).trim(),
        expected.to_string(),
        "the tracking ref advanced to the new tip"
    );
}

/// A real `git -c protocol.version=2 clone` (V2) completes over the
/// git-daemon-framed duplex door against the landed serving entries:
/// the daemon frame, the advertisement-once, the ls-refs round, the
/// done round's packfile section, and the client's own pack validation
/// — HEAD at the expected branch tip, the annotated tag object
/// resolving locally (`git cat-file -t`).
///
/// Contract stamp: ADR-002 (the duplex session boundary, the running V2
/// fetch session); ADR-005 §1 (the pkt-line session machinery the
/// real client drives); ADR-011 §3 (anonymous public read — the door's
/// marker path); POC-1 (the git:// frame shape + real-git validation);
/// ADR-014 (the negotiation grammar); OQ-19 rider 3 (the
/// capture-session shape); core/w5-oq19-real-client-tokens.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_real_git_clone_completes_over_the_duplex_git_daemon_door() {
    let git = git_command();
    if git.is_none() {
        eprintln!("skipping row: no `git` binary on PATH");
        return;
    }
    let _ = git;
    let fx = fixture().await;
    let (addr, door) = spawn_daemon_door(&fx).await;
    let dir = workspace("duplex-clone");
    let mut command = expect_git_command();
    let output = command
        .args(["-c", "protocol.version=2", "clone"])
        .arg(format!("git://{addr}/owner/doubled"))
        .arg(dir.join("dst"))
        .stdin(Stdio::null())
        .output()
        .await
        .expect("the clone runs");
    assert!(
        output.status.success(),
        "the real clone failed over the duplex door: {}{}",
        String::from_utf8_lossy(&output.stdout),
        String::from_utf8_lossy(&output.stderr)
    );
    assert_clone_completed(&dir.join("dst"), fx.commit_b, fx.tag_v1).await;
    door.abort();
}

/// A real configured `git fetch` — the include-tag sender — completes
/// over the duplex door: after the clone, `refs/heads/main` advances
/// server-side; the configured fetch (transport.c v2.43.0 sets
/// `args.include_tag` from the follow-tags option — un-gated in V2's
/// send path) negotiates, its packet trace carries `include-tag`, and
/// the session completes with the tracking ref advanced. The decline
/// the machine served before the decision broke this exact session.
///
/// Contract stamp: OQ-19 rider 3 (the include-tag decision —
/// accept-and-honor-by-omission, validated end-to-end; the rider's
/// "survivable" leg); ADR-014 (the negotiation loop — acknowledgments,
/// then the done round's packfile section); ADR-011 §3 (the anonymous
/// public read path); core/w5-oq19-real-client-tokens.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_real_configured_fetch_sends_include_tag_and_completes_over_the_duplex_door() {
    if git_command().is_none() {
        eprintln!("skipping row: no `git` binary on PATH");
        return;
    }
    let fx = fixture().await;
    let (addr, door) = spawn_daemon_door(&fx).await;
    let dir = workspace("duplex-fetch");

    let mut command = expect_git_command();
    let output = command
        .args(["-c", "protocol.version=2", "clone"])
        .arg(format!("git://{addr}/owner/doubled"))
        .arg(dir.join("dst"))
        .stdin(Stdio::null())
        .output()
        .await
        .expect("the clone runs");
    assert!(
        output.status.success(),
        "the real clone failed over the duplex door: {}{}",
        String::from_utf8_lossy(&output.stdout),
        String::from_utf8_lossy(&output.stderr)
    );

    fx.refs
        .apply_updates(
            &RepoRecord {
                repo_id: "owner/doubled".to_owned(),
                visibility: Visibility::Public,
                grants: BTreeMap::new(),
            },
            vec![RefUpdate {
                name: "refs/heads/main".to_owned(),
                expected: Some(fx.commit_b),
                new: Some(fx.commit_c),
            }],
            false,
        )
        .await
        .expect("the fixture's main advances");

    let mut command = expect_git_command();
    let output = command
        .env("GIT_TRACE_PACKET", "1")
        .args([
            "-C",
            dir.join("dst").to_str().unwrap_or_default(),
            "fetch",
            "origin",
        ])
        .stdin(Stdio::null())
        .output()
        .await
        .expect("the fetch runs");
    let trace = String::from_utf8_lossy(&output.stderr).to_string();
    assert!(
        output.status.success(),
        "the real configured fetch failed over the duplex door: {trace}"
    );
    assert!(
        trace.contains("include-tag"),
        "the real client's negotiation carried include-tag (the trace was captured)"
    );
    assert!(
        !trace.contains("session-id"),
        "the capable client sent no session-id against the honest advertisement"
    );
    assert_clone_completed(&dir.join("dst"), fx.commit_b, fx.tag_v1).await;
    door.abort();
}

/// A real client configured with `transfer.advertiseSid=true` still
/// sends no `session-id=` against the honest advertisement: the
/// capability is never declared, and git's V2 client sends the pre-line
/// only when the server advertises it — the honest-retraction posture
/// makes the decline unreachable by a conforming client, and the
/// session completes.
///
/// Contract stamp: OQ-19 rider 3 (the session-id decision — the honest
/// retraction; the decline stays, fail closed); ADR-003 (never advertise
/// what we don't serve — the `session-id` capability is absent from the
/// served list); fetch-pack.c v2.43.0 (`advertise_sid &&
/// server_supports_v2("session-id")` — the client-side gate the
/// capture-session asserts); core/w5-oq19-real-client-tokens.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_real_client_with_advertise_sid_sends_no_session_id_over_the_duplex_door() {
    if git_command().is_none() {
        eprintln!("skipping row: no `git` binary on PATH");
        return;
    }
    let fx = fixture().await;
    let (addr, door) = spawn_daemon_door(&fx).await;
    let dir = workspace("duplex-sid");

    let mut command = expect_git_command();
    let output = command
        .args([
            "-c",
            "protocol.version=2",
            "-c",
            "transfer.advertiseSID=true",
            "clone",
        ])
        .arg(format!("git://{addr}/owner/doubled"))
        .arg(dir.join("dst"))
        .env("GIT_TRACE_PACKET", "1")
        .stdin(Stdio::null())
        .output()
        .await
        .expect("the clone runs");
    let trace = String::from_utf8_lossy(&output.stderr).to_string();
    assert!(
        output.status.success(),
        "the real clone failed over the duplex door: {trace}"
    );
    assert!(
        !trace.contains("session-id"),
        "the capable client sent no session-id against the honest advertisement"
    );
    assert_clone_completed(&dir.join("dst"), fx.commit_b, fx.tag_v1).await;
    door.abort();
}

/// The same three sessions over the smart-http-framed stateless door
/// (POC-3's service-stub shape): the clone, the configured
/// include-tag-sending fetch, and the advertiseSid client — one POST
/// body one command invocation, one-shot flush responses, real curl
/// keep-alive across the GET and the round POSTs.
///
/// Contract stamp: OQ-19 rider 3 (both substrates per the rider house
/// style); ADR-005 §2 (the stateless request/response rules the real
/// client drives); ADR-005 §3 (one machine core over two substrates);
/// POC-3 (the smart-http door shape); ADR-011 §3; ADR-014; OQ-19 rider
/// 3; core/w5-oq19-real-client-tokens.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn the_real_client_sessions_complete_over_the_stateless_http_door() {
    if git_command().is_none() {
        eprintln!("skipping row: no `git` binary on PATH");
        return;
    }
    let fx = fixture().await;
    let (addr, door) = spawn_http_door(&fx).await;
    let url = format!("http://{addr}/owner/doubled");

    let dir = workspace("http-clone");
    let mut command = expect_git_command();
    let output = command
        .args(["-c", "protocol.version=2", "clone", &url])
        .arg(dir.join("dst"))
        .stdin(Stdio::null())
        .output()
        .await
        .expect("the clone runs");
    assert!(
        output.status.success(),
        "the real clone failed over the stateless door: {}{}",
        String::from_utf8_lossy(&output.stdout),
        String::from_utf8_lossy(&output.stderr)
    );
    assert_clone_completed(&dir.join("dst"), fx.commit_b, fx.tag_v1).await;

    fx.refs
        .apply_updates(
            &RepoRecord {
                repo_id: "owner/doubled".to_owned(),
                visibility: Visibility::Public,
                grants: BTreeMap::new(),
            },
            vec![RefUpdate {
                name: "refs/heads/main".to_owned(),
                expected: Some(fx.commit_b),
                new: Some(fx.commit_c),
            }],
            false,
        )
        .await
        .expect("the fixture's main advances");
    // The include-tag sender: the configured fetch runs inside the
    // first session's clone (which has `commit_b`, so the advance is
    // real negotiation — the stateless rounds POST per round).
    let mut command = expect_git_command();
    let output = command
        .env("GIT_TRACE_PACKET", "1")
        .args([
            "-C",
            dir.join("dst").to_str().unwrap_or_default(),
            "fetch",
            "origin",
        ])
        .stdin(Stdio::null())
        .output()
        .await
        .expect("the fetch runs");
    let trace = String::from_utf8_lossy(&output.stderr).to_string();
    assert!(
        output.status.success(),
        "the real configured fetch failed over the stateless door: {trace}"
    );
    assert!(
        trace.contains("include-tag"),
        "the real client's negotiation carried include-tag over the stateless door"
    );
    assert!(
        !trace.contains("session-id"),
        "the client sent no session-id against the honest advertisement"
    );
    assert_fetch_completed(&dir.join("dst"), fx.commit_c).await;

    let dir = workspace("http-sid");
    let mut command = expect_git_command();
    let output = command
        .args([
            "-c",
            "protocol.version=2",
            "-c",
            "transfer.advertiseSID=true",
            "clone",
            &url,
        ])
        .arg(dir.join("dst"))
        .env("GIT_TRACE_PACKET", "1")
        .stdin(Stdio::null())
        .output()
        .await
        .expect("the clone runs");
    let trace = String::from_utf8_lossy(&output.stderr).to_string();
    assert!(
        output.status.success(),
        "the real clone failed over the stateless door: {trace}"
    );
    assert!(
        !trace.contains("session-id"),
        "the capable client sent no session-id against the honest advertisement"
    );
    // This clone ran after the fetch scenario advanced `main` — the
    // fixture's tip is now `commit_c`.
    assert_clone_completed(&dir.join("dst"), fx.commit_c, fx.tag_v1).await;
    door.abort();
}

/// OQ-19 rider 1 (review 001 N-5) over the git-daemon-framed duplex
/// door: real `git ls-remote` (V2) against the unborn repo — zero ref
/// rows, the HEAD symref row unborn at `refs/heads/main` — requests
/// `unborn` (captured live: ls-remote's ls-refs round carries `peel`/
/// `symrefs`/`unborn`), and the served unborn HEAD symref line
/// completes the session with the client's exit code 0. The unborn
/// row is a resolution shape, not a listing entry: a real client
/// prints nothing for it (captured live against upstream's own server
/// — the ls-refs response is consumed for the default branch, not
/// listed).
///
/// Contract stamp: ADR-003 (the `ls-refs=unborn` advertisement is
/// served — validate or retract); ADR-018 §5 (`RefLine`'s `oid: None`
/// unborn leg); transport.md §ls-refs (the N-5 rider — the only
/// advertisement promise without real-client capture before this
/// session); core/w5-oq19-real-client-tokens's doors; architecture/
/// n5-unborn-head-rider.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_real_git_ls_remote_completes_against_the_unborn_repo_over_the_duplex_door() {
    if git_command().is_none() {
        eprintln!("skipping row: no `git` binary on PATH");
        return;
    }
    let fx = fixture().await;
    let (addr, door) = spawn_daemon_door(&fx).await;
    let mut command = expect_git_command();
    let output = command
        .env("GIT_TRACE_PACKET", "1")
        .args(["-c", "protocol.version=2", "ls-remote"])
        .arg(format!("git://{addr}/owner/unborn"))
        .stdin(Stdio::null())
        .output()
        .await
        .expect("the ls-remote runs");
    let trace = String::from_utf8_lossy(&output.stderr).to_string();
    assert!(
        output.status.success(),
        "the real ls-remote failed against the unborn repo over the duplex door: {trace}"
    );
    assert!(
        trace.contains("ls-remote> unborn"),
        "the real client requested the unborn argument (the capture-session fact)"
    );
    assert!(
        trace.contains("symref-target:refs/heads/main"),
        "the unborn HEAD symref line was served: {trace}"
    );
    door.abort();
}

/// The same rider-1 session over the smart-http-framed stateless door:
/// real `git ls-remote` (V2) against the unborn repo — the
/// advertisement GET, then the ls-refs POST — completes with exit 0
/// and the unborn HEAD symref line served per the ls-refs grammar.
///
/// Contract stamp: ADR-003; ADR-005 §2 (the stateless request/response
/// rules the real client drives); ADR-005 §3 (one machine core over
/// two substrates); ADR-018 §5; architecture/n5-unborn-head-rider
/// (both substrates per the rider's definition).
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_real_git_ls_remote_completes_against_the_unborn_repo_over_the_stateless_door() {
    if git_command().is_none() {
        eprintln!("skipping row: no `git` binary on PATH");
        return;
    }
    let fx = fixture().await;
    let (addr, door) = spawn_http_door(&fx).await;
    let url = format!("http://{addr}/owner/unborn");
    let mut command = expect_git_command();
    let output = command
        .env("GIT_TRACE_PACKET", "1")
        .args(["-c", "protocol.version=2", "ls-remote", &url])
        .stdin(Stdio::null())
        .output()
        .await
        .expect("the ls-remote runs");
    let trace = String::from_utf8_lossy(&output.stderr).to_string();
    assert!(
        output.status.success(),
        "the real ls-remote failed against the unborn repo over the stateless door: {trace}"
    );
    assert!(
        trace.contains("symref-target:refs/heads/main"),
        "the unborn HEAD symref line was served over the stateless door: {trace}"
    );
    door.abort();
}

/// The rider-1 session's full shape: a real `git clone` (V2) of the
/// unborn repo over the duplex door — the default-branch resolution
/// the unborn HEAD symref line exists for. The clone completes (the
/// empty-repository warning is the expected stdout), the clone's HEAD
/// points at the symref target's branch, and nothing is checked out
/// (no commit exists anywhere).
///
/// Contract stamp: OQ-19 rider 1 (the unborn-HEAD session — validate
/// or retract); ADR-003; ADR-014 (the ls-refs round, no fetch);
/// ADR-018 §5 (`RefLine`'s `oid: None`); architecture/
/// n5-unborn-head-rider.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_real_git_clone_of_the_unborn_repo_resolves_the_default_branch_over_the_duplex_door() {
    if git_command().is_none() {
        eprintln!("skipping row: no `git` binary on PATH");
        return;
    }
    let fx = fixture().await;
    let (addr, door) = spawn_daemon_door(&fx).await;
    let dir = workspace("duplex-unborn-clone");
    let mut command = expect_git_command();
    let output = command
        .env("GIT_TRACE_PACKET", "1")
        .args(["-c", "protocol.version=2", "clone"])
        .arg(format!("git://{addr}/owner/unborn"))
        .arg(dir.join("dst"))
        .stdin(Stdio::null())
        .output()
        .await
        .expect("the clone runs");
    let trace = String::from_utf8_lossy(&output.stderr).to_string();
    assert!(
        output.status.success(),
        "the real clone of the unborn repo failed over the duplex door: {trace}"
    );
    assert!(
        trace.contains("symref-target:refs/heads/main"),
        "the unborn HEAD symref line rode the ls-refs round: {trace}"
    );
    assert!(
        trace.contains("You appear to have cloned an empty repository"),
        "the real client recognized the empty repository"
    );
    let head_path = dir.join("dst/.git/HEAD");
    let head = std::fs::read_to_string(&head_path).expect("the clone's HEAD reads");
    assert!(
        head.contains("ref: refs/heads/main"),
        "the clone's HEAD points at the symref target's branch: {head}"
    );
    door.abort();
}