alkgit 0.1.0

//! The wave-4 GitSession fetch client (core/w4-session-fetch): the
//! consumer half's ADR-017 §1 operations riding the wave-4
//! construction — `ls_refs` (the `ls-refs` argument grammar our own
//! server machine parses) and `fetch` (the full ack loop as full
//! stateless commands per ADR-014, the done round's sideband pack
//! pumped to the caller's sink) — rows driven in-process,
//! client-vs-our-server, both halves in one test over the wave-2/3
//! backend doubles (the POC-X gate shape re-landed on the real code:
//! findings 2/4/5 of pocx-findings.md). The direct-mode rows pair the
//! session with the real `GitAdapter` (the request line is the
//! adapter's consumption, ADR-016 §2); the machinery rows pair
//! channels-mode sessions with `serve_duplex` (zero preamble — the
//! preamble rode the open op).
//!
//! Rows are names + `Contract stamp:` versions, per `tests/common`
//! §The row/stamp discipline.

mod common;

use std::collections::{BTreeMap, BTreeSet, HashMap};
use std::pin::Pin;
use std::sync::{Arc, Mutex};
use std::task::{Context, Poll};
use std::time::Duration;

use alkcall::core::auth::Identity;
use alkcall::core::types::{BiStream, Connection, ProtocolHandler};
use tokio::io::{duplex, AsyncRead, AsyncWriteExt, ReadBuf};
use tokio::sync::Semaphore;

use gix_hash::ObjectId;

use alkgit::{
    serve_duplex, AccessAction, AuthorizedRepo, FetchRound, GitAdapter, GitPackGen, GitPackIngest,
    GitRefs, GitRegistry, GitRegistryStore, GitSession, Limits, LsRefsArgs, PreparedPush, RefLine,
    RefUpdate, RepoRecord, Service, SessionError, SubstrateError, Visibility,
};
use common::{
    blob_fixture, commit_fixture, decode_pack, tree_fixture, MemPackGen, MemPackIngest, MemRefs,
    MemRegistry,
};

type Handles = (
    Arc<MemRegistry>,
    Arc<MemRefs>,
    Arc<MemPackGen>,
    Arc<MemPackIngest>,
);

fn handles() -> Handles {
    let objects = common::MemObjects::default();
    (
        Arc::new(MemRegistry::default()),
        Arc::new(MemRefs::default()),
        Arc::new(MemPackGen {
            objects: objects.clone(),
        }),
        Arc::new(MemPackIngest { objects }),
    )
}

fn repo(id: &str) -> RepoRecord {
    RepoRecord {
        repo_id: id.to_string(),
        visibility: Visibility::Private,
        grants: BTreeMap::new(),
    }
}

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

type WireRecord = Arc<Mutex<Vec<u8>>>;

type StreamJoiner = tokio::task::JoinHandle<Result<(), SubstrateError>>;

/// Records every byte read off the wire — the client-sent side (what
/// the server RECEIVED), for the request-grammar byte pins.
struct RecordRead<S> {
    inner: S,
    record: WireRecord,
}

impl<S: AsyncRead + Unpin> AsyncRead for RecordRead<S> {
    fn poll_read(
        mut self: Pin<&mut Self>,
        cx: &mut Context<'_>,
        buf: &mut ReadBuf<'_>,
    ) -> Poll<std::io::Result<()>> {
        let this = self.as_mut().get_mut();
        let before = buf.filled().len();
        match Pin::new(&mut this.inner).poll_read(cx, buf) {
            Poll::Ready(Ok(())) => {
                this.record
                    .lock()
                    .expect("spy record")
                    .extend_from_slice(&buf.filled()[before..]);
                Poll::Ready(Ok(()))
            }
            other => other,
        }
    }
}

/// Records every byte written to the wire (the server-sent side).
struct RecordWrite<S> {
    inner: S,
    record: WireRecord,
}

impl<S: tokio::io::AsyncWrite + Unpin> tokio::io::AsyncWrite for RecordWrite<S> {
    fn poll_write(
        mut self: Pin<&mut Self>,
        cx: &mut Context<'_>,
        buf: &[u8],
    ) -> Poll<std::io::Result<usize>> {
        let this = self.as_mut().get_mut();
        this.record
            .lock()
            .expect("spy record")
            .extend_from_slice(buf);
        Pin::new(&mut this.inner).poll_write(cx, buf)
    }
    fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
        let this = self.as_mut().get_mut();
        Pin::new(&mut this.inner).poll_flush(cx)
    }
    fn poll_shutdown(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
        let this = self.as_mut().get_mut();
        Pin::new(&mut this.inner).poll_shutdown(cx)
    }
}

/// The fixture: the doubles' registry double plus the granted repo row.
async fn fetch_fixture(holder: &str, action: AccessAction) -> (Handles, RepoRecord, Identity) {
    let handles = handles();
    let mut granted = repo("owner/doubled");
    granted.grants.insert(holder.to_owned(), vec![action]);
    let registry = Arc::clone(&handles.0);
    registry.put_repo(&granted).await.expect("the repo row");
    (handles, granted, identity(holder))
}

/// A two-commit chain fixture (fetch_machines' rows' shape): a child
/// commit on its root, both trees/blobs seeded in one namespace.
struct Chain {
    commit_a: ObjectId,
    commit_b: ObjectId,
    tree_a: ObjectId,
    tree_b: ObjectId,
    blob_a: ObjectId,
    blob_b: ObjectId,
}

fn two_commit_chain(objects: &common::MemObjects, doubled: &RepoRecord) -> Chain {
    let (blob_a, blob_a_bytes) = blob_fixture(b"root content");
    let (blob_b, blob_b_bytes) = blob_fixture(b"child content");
    let (tree_a, tree_a_bytes) =
        tree_fixture(&[(gix_object::tree::EntryKind::Blob, "file", blob_a)]);
    let (tree_b, tree_b_bytes) =
        tree_fixture(&[(gix_object::tree::EntryKind::Blob, "file", 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");
    for (oid, 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),
    ] {
        objects.put(doubled, oid, bytes);
    }
    Chain {
        commit_a,
        commit_b,
        tree_a,
        tree_b,
        blob_a,
        blob_b,
    }
}

/// Seeds the chain's commits as two ref rows (base + main) for the
/// listing rows.
async fn seed_chain_refs(handles: &Handles, record: &RepoRecord, chain: &Chain) {
    Arc::clone(&handles.1)
        .apply_updates(
            record,
            vec![
                RefUpdate {
                    name: "refs/heads/base".to_owned(),
                    expected: None,
                    new: Some(chain.commit_a),
                },
                RefUpdate {
                    name: "refs/heads/main".to_owned(),
                    expected: None,
                    new: Some(chain.commit_b),
                },
            ],
            false,
        )
        .await
        .expect("the seed");
}

/// The channels-mode serving twin: spawns the in-crate duplex serving
/// entry (`serve_duplex`) over a spy-wrapped pair whose server records
/// both directions. Pairs with `open_via_channels` sessions (the
/// preamble rode the open op — zero preamble bytes here).
async fn serve_twin(
    handles: &Handles,
    record: RepoRecord,
    holder: &Identity,
    action: AccessAction,
    limits: Limits,
) -> (BiStream, WireRecord, WireRecord, StreamJoiner) {
    let checked =
        AuthorizedRepo::authorize(record, Some(holder), action).expect("the checked marker");
    let refs: Arc<dyn GitRefs> = handles.1.clone();
    let packgen: Arc<dyn GitPackGen> = handles.2.clone();
    let ingest: Arc<dyn GitPackIngest> = handles.3.clone();
    let (client_end, server_end) = duplex(1 << 16);
    let received = Arc::new(Mutex::new(Vec::new()));
    let sent = Arc::new(Mutex::new(Vec::new()));
    let (server_read, server_write) = tokio::io::split(server_end);
    let server = BiStream::from_joined(
        RecordRead {
            inner: server_read,
            record: Arc::clone(&received),
        },
        RecordWrite {
            inner: server_write,
            record: Arc::clone(&sent),
        },
    );
    let joiner = tokio::spawn(async move {
        serve_duplex(
            &checked,
            server,
            limits,
            Service::UploadPack,
            refs,
            packgen,
            ingest,
            false,
            Arc::new(Semaphore::new(
                Limits::default().max_concurrent_blocking_tasks,
            )),
        )
        .await
    });
    let (client_read, client_write) = tokio::io::split(client_end);
    (
        BiStream::from_joined(client_read, client_write),
        received,
        sent,
        joiner,
    )
}

/// The `accept_bi` queue source (session_construction's shape): yields
/// the server-side stream (spy-wrapped on the client-sent side) once.
struct YieldQueued {
    streams: tokio::sync::Mutex<std::collections::VecDeque<BiStream>>,
    received: WireRecord,
}

#[async_trait::async_trait]
impl alkcall::core::BidiStreamSource for YieldQueued {
    async fn accept_bi(&self) -> Result<BiStream, alkcall::core::StreamError> {
        let stream = self
            .streams
            .lock()
            .await
            .pop_front()
            .ok_or(alkcall::core::StreamError::ConnectionClosed)?;
        let (server_read, server_write) = tokio::io::split(stream);
        Ok(BiStream::from_joined(
            RecordRead {
                inner: server_read,
                record: Arc::clone(&self.received),
            },
            server_write,
        ))
    }
    async fn open_bi(&self) -> Result<BiStream, alkcall::core::StreamError> {
        Err(alkcall::core::StreamError::StreamClosed)
    }
    fn remote_addr(&self) -> Option<std::net::SocketAddr> {
        None
    }
    fn close(&self, _code: u32, _reason: &str) {}
}

/// The direct-mode serving twin: the real `GitAdapter` (the request
/// line is the adapter's consumption — the preamble rides out from the
/// session's `connect_direct` handshake) over a one-stream source.
/// Pairs with `connect_direct` sessions; the holder rides the
/// `AuthContext` identity (the adapter resolve→authorizes through the
/// doubles' registry row).
async fn adapter_twin(
    handles: &Handles,
    holder: &Identity,
) -> (
    BiStream,
    WireRecord,
    WireRecord,
    tokio::task::JoinHandle<()>,
) {
    let (client_end, server_end) = duplex(1 << 16);
    let received = Arc::new(Mutex::new(Vec::new()));
    let sent = Arc::new(Mutex::new(Vec::new()));
    // The served stream carries the sent-side spy; the adapter's gate
    // reads the request line off the received-side spy.
    let (server_read, server_write) = tokio::io::split(server_end);
    let served = BiStream::from_joined(
        server_read,
        RecordWrite {
            inner: server_write,
            record: Arc::clone(&sent),
        },
    );
    let adapter = GitAdapter::new(
        Arc::clone(&handles.0) as Arc<dyn GitRegistry>,
        Arc::clone(&handles.1) as Arc<dyn GitRefs>,
        Arc::clone(&handles.2) as Arc<dyn GitPackGen>,
        Arc::clone(&handles.3) as Arc<dyn GitPackIngest>,
        false,
        Limits::default(),
        Arc::new(Semaphore::new(
            Limits::default().max_concurrent_blocking_tasks,
        )),
    );
    let connection = Connection::from_source(
        YieldQueued {
            streams: tokio::sync::Mutex::new([served].into()),
            received: Arc::clone(&received),
        },
        b"alk/git".to_vec(),
    );
    let mut auth = alkcall::core::auth::AuthContext::anonymous(b"alk/git");
    auth.identity = Some(holder.clone());
    let joiner = tokio::spawn(async move {
        let _ = ProtocolHandler::handle(&adapter, connection, &auth).await;
    });
    let (client_read, client_write) = tokio::io::split(client_end);
    (
        BiStream::from_joined(client_read, client_write),
        received,
        sent,
        joiner,
    )
}

/// The fake wire peer: writes the scripted bytes, then holds the
/// stream open (the client's drop closes it) — the hostile/stalling
/// peer rows' serving side.
async fn spawn_fake_peer(script: Vec<Vec<u8>>) -> BiStream {
    let (client_end, server_end) = duplex(1 << 16);
    let mut server = server_end;
    let _ = server.write_all(script.concat().as_slice()).await;
    let _ = server.flush().await;
    tokio::spawn(async move {
        tokio::time::sleep(Duration::from_secs(30)).await;
        let _ = server.shutdown().await;
    });
    let (client_read, client_write) = tokio::io::split(client_end);
    BiStream::from_joined(client_read, client_write)
}

/// Encodes one pkt-line (client framing).
fn pkt(payload: &[u8]) -> Vec<u8> {
    let total = payload.len() + 4;
    let mut out = Vec::with_capacity(total);
    out.extend_from_slice(format!("{total:04x}").as_bytes());
    out.extend_from_slice(payload);
    out
}

/// One text command line (the client's text codec appends the newline
/// the wire grammar carries).
fn text_line(text: &[u8]) -> Vec<u8> {
    let mut payload = text.to_vec();
    payload.push(b'\n');
    pkt(&payload)
}

fn want_line(oid: &ObjectId) -> Vec<u8> {
    text_line(format!("want {oid}").as_bytes())
}

fn have_line(oid: &ObjectId) -> Vec<u8> {
    text_line(format!("have {oid}").as_bytes())
}

const DELIM: &[u8] = b"0001";
const FLUSH: &[u8] = b"0000";

/// The commands' capability line set (the agent + the object-format
/// echo, negotiation-captures' grammar).
fn capability_lines() -> Vec<Vec<u8>> {
    vec![
        text_line(
            concat!(
                "agent=",
                env!("CARGO_PKG_NAME"),
                "/",
                env!("CARGO_PKG_VERSION")
            )
            .as_bytes(),
        ),
        text_line(b"object-format=sha1"),
    ]
}

// --- ls-refs ----------------------------------------------------------------

/// Contract stamp: ADR-017 §1 (ls_refs), ADR-016 §2 (the request
/// line is the adapter's — byte-pinned), the negotiation-captures
/// §Round grammar (caps then `0001` then args then flush) on the
/// `ls-refs` command; core/w4-session-fetch.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn ls_refs_direct_returns_the_ref_list_and_the_request_is_the_pinned_grammar() {
    let (handles, record, holder) = fetch_fixture("u-7731", AccessAction::Read).await;
    let seeded_a = ObjectId::from([7u8; 20]);
    let seeded_b = ObjectId::from([8u8; 20]);
    Arc::clone(&handles.1)
        .apply_updates(
            &record,
            vec![
                RefUpdate {
                    name: "refs/heads/main".to_owned(),
                    expected: None,
                    new: Some(seeded_a),
                },
                RefUpdate {
                    name: "refs/tags/v1".to_owned(),
                    expected: None,
                    new: Some(seeded_b),
                },
            ],
            false,
        )
        .await
        .expect("the seed");
    handles
        .1
        .set_head_symref(&record, Some("refs/heads/main".to_owned()));
    let (client, received, _sent, _joiner) = adapter_twin(&handles, &holder).await;

    let mut session = GitSession::connect_direct(
        client,
        "owner/doubled",
        Service::UploadPack,
        Some("example.net"),
        Limits::default(),
    )
    .expect("the direct construction");
    let listed = tokio::time::timeout(
        Duration::from_secs(5),
        session.ls_refs(&LsRefsArgs {
            symrefs: true,
            peel: true,
            ref_prefixes: vec!["refs/heads/".to_owned()],
            unborn: false,
        }),
    )
    .await
    .expect("ls-refs is timely")
    .expect("the ref list returns");

    assert_eq!(
        listed,
        vec![RefLine {
            name: "refs/heads/main".to_owned(),
            oid: Some(seeded_a),
            peeled: None,
            symref_target: None,
        }],
        "the ref-prefix filter honors the argument (the unborn HEAD row is not requested, so the \
         row is omitted there); the tags row filters out"
    );

    assert_eq!(
        *received.lock().expect("spy record"),
        [
            pkt(b"git-upload-pack owner/doubled\0host=example.net\0\0version=2\0"),
            text_line(b"command=ls-refs"),
            agent_line_placeholder(),
            text_line(b"object-format=sha1"),
            DELIM.to_vec(),
            text_line(b"symrefs"),
            text_line(b"peel"),
            text_line(b"ref-prefix refs/heads/"),
            FLUSH.to_vec(),
        ]
        .concat(),
        "the request bytes are the ADR-016 §2 preamble line plus the capture grammar's command shape"
    );
    let _ = seeded_b;
}

fn agent_line_placeholder() -> Vec<u8> {
    text_line(
        concat!(
            "agent=",
            env!("CARGO_PKG_NAME"),
            "/",
            env!("CARGO_PKG_VERSION")
        )
        .as_bytes(),
    )
}

/// Contract stamp: ADR-017 §1 (ls_refs — the `unborn` argument
/// honored), ADR-016 §2's V2 ref grammar (the unborn symref line
/// parses oid-less — both the literal `unborn` emission and the
/// zero-oid/attribute-only peer shapes), review 001 N-5 (the unborn
/// HEAD fixture); core/w4-session-fetch.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn ls_refs_honors_the_unborn_argument_and_the_bare_command_shape() {
    let (handles, record, holder) = fetch_fixture("u-7731", AccessAction::Read).await;
    handles
        .1
        .set_head_symref(&record, Some("refs/heads/unborn".to_owned()));
    let (client, _received, _sent, _joiner) = serve_twin(
        &handles,
        record.clone(),
        &holder,
        AccessAction::Read,
        Limits::default(),
    )
    .await;

    let mut session = GitSession::open_via_channels(
        client,
        "owner/doubled",
        Service::UploadPack,
        Limits::default(),
    )
    .expect("the channels-mode construction");
    let listed = tokio::time::timeout(
        Duration::from_secs(5),
        session.ls_refs(&LsRefsArgs {
            symrefs: true,
            unborn: true,
            ..LsRefsArgs::default()
        }),
    )
    .await
    .expect("ls-refs is timely")
    .expect("the argument-bearing ls-refs runs against the serving twin");
    assert_eq!(
        listed,
        vec![RefLine {
            name: "HEAD".to_owned(),
            oid: None,
            peeled: None,
            symref_target: Some("refs/heads/unborn".to_owned()),
        }],
        "the unborn HEAD row rides the grammar line and parses oid-less"
    );

    let bare = tokio::time::timeout(
        Duration::from_secs(5),
        session.ls_refs(&LsRefsArgs::default()),
    )
    .await
    .expect("ls-refs is timely")
    .expect("the bare command shape (no argument lines, no delimiter)");
    assert!(
        bare.is_empty(),
        "the empty-argument command's response is refs-less (the unborn row only on the request)"
    );
}

// --- the ack loop and the done round ----------------------------------------

/// Contract stamp: ADR-017 §1 (fetch — the client side of the loop we
/// serve), ADR-014 §1/§5 (the acknowledgments grammar + the ack rule —
/// the exists + is-commit predicate), ADR-023 §6 (rounds are full
/// stateless commands); core/w4-session-fetch.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn the_ack_loop_rounds_parse_both_acknowledgment_shapes() {
    let (handles, record, holder) = fetch_fixture("u-7731", AccessAction::Read).await;
    let chain = two_commit_chain(&handles.2.objects, &record);
    let (client, received, _sent, _joiner) = serve_twin(
        &handles,
        record.clone(),
        &holder,
        AccessAction::Read,
        Limits::default(),
    )
    .await;

    let mut session = GitSession::open_via_channels(
        client,
        "owner/doubled",
        Service::UploadPack,
        Limits::default(),
    )
    .expect("the channels-mode construction");
    let reports = tokio::time::timeout(
        Duration::from_secs(5),
        session.fetch(
            vec![
                // A recognized common (the root commit exists and is a
                // commit) — the positive ack.
                FetchRound {
                    wants: vec![chain.commit_b],
                    haves: vec![chain.commit_a],
                    done: false,
                },
                // An unknown and a non-commit have — NAK alone.
                FetchRound {
                    wants: vec![chain.commit_b],
                    haves: vec![chain.blob_a, ObjectId::from([9u8; 20])],
                    done: false,
                },
            ],
            &mut Vec::new(),
        ),
    )
    .await
    .expect("the loop is timely")
    .expect("both rounds acknowledged");

    assert_eq!(
        reports,
        vec![
            alkgit::FetchRoundReport {
                acks: vec![chain.commit_a],
                pack_bytes: None,
            },
            alkgit::FetchRoundReport {
                acks: Vec::new(),
                pack_bytes: None,
            },
        ],
        "the positive ACK shape and the NAK-only shape both parsed, one report per round"
    );
    let wire = received.lock().expect("spy record").clone();
    assert_eq!(
        wire,
        [
            // Round 1's full stateless command (caps, delim, the
            // thin-pack/ofs-delta initial args, then the caller's wants
            // and haves).
            text_line(b"command=fetch"),
            capability_lines().concat(),
            DELIM.to_vec(),
            text_line(b"thin-pack"),
            text_line(b"ofs-delta"),
            want_line(&chain.commit_b),
            have_line(&chain.commit_a),
            FLUSH.to_vec(),
            // Round 2 re-sends the want, new haves.
            text_line(b"command=fetch"),
            capability_lines().concat(),
            DELIM.to_vec(),
            text_line(b"thin-pack"),
            text_line(b"ofs-delta"),
            want_line(&chain.commit_b),
            have_line(&chain.blob_a),
            have_line(&ObjectId::from([9u8; 20])),
            FLUSH.to_vec(),
        ]
        .concat(),
        "each round's request bytes are the capture grammar: wants re-sent verbatim, the \
         acknowledgments ride gix's send machinery"
    );
}

/// Contract stamp: ADR-017 §1 (the done round), ADR-017 §4 (the
/// storage-agnostic pack delivery — the caller's sink, byte-counted),
/// ADR-014 §2 (the done round's packfile section, sideband + flush),
/// pocx-findings finding 4 (the pump); core/w4-session-fetch.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn the_done_round_terminates_fetch_and_delivers_the_packed_bytes_to_the_consumer() {
    let (handles, record, holder) = fetch_fixture("u-7731", AccessAction::Read).await;
    let chain = two_commit_chain(&handles.2.objects, &record);
    let (client, received, _sent, _joiner) = serve_twin(
        &handles,
        record.clone(),
        &holder,
        AccessAction::Read,
        Limits::default(),
    )
    .await;

    let mut session = GitSession::open_via_channels(
        client,
        "owner/doubled",
        Service::UploadPack,
        Limits::default(),
    )
    .expect("the channels-mode construction");
    let mut pack = Vec::new();
    let reports = tokio::time::timeout(
        Duration::from_secs(5),
        session.fetch(
            vec![FetchRound {
                wants: vec![chain.commit_b],
                haves: Vec::new(),
                done: true,
            }],
            &mut pack,
        ),
    )
    .await
    .expect("the fetch is timely")
    .expect("the done round delivered the pack");

    assert_eq!(
        reports,
        vec![alkgit::FetchRoundReport {
            acks: Vec::new(),
            pack_bytes: Some(pack.len() as u64),
        }],
        "the report's byte count is exactly the consumer's byte count"
    );
    assert!(
        !pack.is_empty(),
        "the sideband-unwrapped pack reached the consumer intact"
    );

    let decoded = decode_pack(&pack);
    assert_eq!(
        decoded
            .objects
            .iter()
            .map(|(oid, _kind, _body)| *oid)
            .collect::<BTreeSet<_>>(),
        BTreeSet::from([
            chain.commit_a,
            chain.commit_b,
            chain.tree_a,
            chain.tree_b,
            chain.blob_a,
            chain.blob_b,
        ]),
        "the full closure of commit_b was delivered (no boundary haves), the trailer verified by \
         the decode"
    );

    // The request byte pin (the capture grammar's done-round shape).
    assert_eq!(
        *received.lock().expect("spy record"),
        [
            text_line(b"command=fetch"),
            capability_lines().concat(),
            DELIM.to_vec(),
            text_line(b"thin-pack"),
            text_line(b"ofs-delta"),
            want_line(&chain.commit_b),
            text_line(b"done"),
            FLUSH.to_vec(),
        ]
        .concat(),
        "the fetch request bytes are the capture grammar (command, caps, delim, args, done, flush)"
    );

    // The pack round-trips through the doubles' ingest path: the
    // objects land, the connectivity walk closes from the tips.
    let prepared: PreparedPush = handles
        .3
        .prepare(
            &record,
            vec![RefUpdate {
                name: "refs/heads/main".to_owned(),
                expected: None,
                new: Some(chain.commit_b),
            }],
            Box::new(std::io::Cursor::new(pack.clone())),
            None,
            Limits::default(),
        )
        .await
        .expect("the ingestion accepts the delivered pack");
    assert!(
        prepared.unpack.is_ok(),
        "the delivered pack ingests cleanly: {:?}",
        prepared.unpack
    );
    for object in [chain.commit_b, chain.tree_b, chain.blob_b] {
        assert!(
            handles.3.objects().contains(&record, &object),
            "the ingested landed row holds {object}"
        );
    }
}

/// Contract stamp: ADR-017 §1 (the done round as the negotiation
/// terminator), ADR-014 §2/§7 (the done request's haves are the
/// complete boundary view — the negotiation's payoff is the trimmed
/// pack), ADR-023 §6 (2 no-done rounds + done, the POC-X gate shape);
/// core/w4-session-fetch.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn multi_round_negotiation_converges_with_two_no_done_rounds_plus_the_done_round() {
    let (handles, record, holder) = fetch_fixture("u-7731", AccessAction::Read).await;
    let chain = two_commit_chain(&handles.2.objects, &record);
    let (client, received, _sent, _joiner) = serve_twin(
        &handles,
        record.clone(),
        &holder,
        AccessAction::Read,
        Limits::default(),
    )
    .await;

    let mut session = GitSession::open_via_channels(
        client,
        "owner/doubled",
        Service::UploadPack,
        Limits::default(),
    )
    .expect("the channels-mode construction");
    let mut pack = Vec::new();
    let reports = tokio::time::timeout(
        Duration::from_secs(5),
        session.fetch(
            vec![
                FetchRound {
                    wants: vec![chain.commit_b],
                    haves: vec![chain.commit_a],
                    done: false,
                },
                // The stateless re-send: wants + commons again, plus
                // one unrecognized extra.
                FetchRound {
                    wants: vec![chain.commit_b],
                    haves: vec![chain.commit_a, chain.blob_b],
                    done: false,
                },
                FetchRound {
                    wants: vec![chain.commit_b],
                    haves: vec![chain.commit_a],
                    done: true,
                },
            ],
            &mut pack,
        ),
    )
    .await
    .expect("the loop is timely")
    .expect("the negotiation converges on the done round");

    assert_eq!(
        reports[0].acks,
        vec![chain.commit_a],
        "the root commit acks on the first round"
    );
    assert_eq!(
        reports[1].acks,
        vec![chain.commit_a],
        "the common re-sent every round re-acks (the non-commit have does not)"
    );
    let trimmed = reports[2].pack_bytes.expect("the done round delivered");
    assert_eq!(trimmed, pack.len() as u64);

    // The negotiation's payoff: the boundary's closure (commit_a's
    // chain) subtracted from the pack — measured against the plain
    // full fetch of the same want.
    let (client_full, _received_full, _sent_full, _joiner_full) = serve_twin(
        &handles,
        record.clone(),
        &holder,
        AccessAction::Read,
        Limits::default(),
    )
    .await;
    let mut session_full = GitSession::open_via_channels(
        client_full,
        "owner/doubled",
        Service::UploadPack,
        Limits::default(),
    )
    .expect("the full-fetch construction");
    let full_packed = tokio::time::timeout(
        Duration::from_secs(5),
        session_full.fetch(
            vec![FetchRound {
                wants: vec![chain.commit_b],
                haves: Vec::new(),
                done: true,
            }],
            &mut Vec::new(),
        ),
    )
    .await
    .expect("the full fetch is timely")
    .expect("the full pack delivered");

    assert!(
        trimmed < full_packed[0].pack_bytes.expect("the full fetch delivered"),
        "the done round's haves trimmed the pack below the no-haves shape"
    );
    let decoded = decode_pack(&pack);
    assert_eq!(
        decoded
            .objects
            .iter()
            .map(|(oid, _kind, _body)| *oid)
            .collect::<BTreeSet<_>>(),
        BTreeSet::from([chain.commit_b, chain.tree_b, chain.blob_b]),
        "closure(commit_b) − closure(commit_a)"
    );
    let grammar = String::from_utf8(received.lock().expect("spy record").clone())
        .expect("the request grammar is text");
    assert_eq!(
        grammar.matches("want ").count(),
        3,
        "each stateless round re-sent the want (three rounds)"
    );
}

// --- the session persistence and the budget rows -----------------------------

/// Contract stamp: ADR-023 §2 (one connection = one service machine —
/// the V2 session persists across ls-refs → fetch), ADR-003's
/// advertisement-once discipline (the duplex entry's serving shape);
/// core/w4-session-fetch.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn the_session_persists_across_ls_refs_fetch_on_one_stream() {
    let (handles, record, holder) = fetch_fixture("u-7731", AccessAction::Read).await;
    let chain = two_commit_chain(&handles.2.objects, &record);
    seed_chain_refs(&handles, &record, &chain).await;
    let (client, received, sent, _joiner) = serve_twin(
        &handles,
        record.clone(),
        &holder,
        AccessAction::Read,
        Limits::default(),
    )
    .await;

    let mut session = GitSession::open_via_channels(
        client,
        "owner/doubled",
        Service::UploadPack,
        Limits::default(),
    )
    .expect("the channels-mode construction");
    let listed = tokio::time::timeout(
        Duration::from_secs(5),
        session.ls_refs(&LsRefsArgs {
            symrefs: true,
            ..LsRefsArgs::default()
        }),
    )
    .await
    .expect("ls-refs is timely")
    .expect("the listing");
    assert_eq!(listed.len(), 2, "the chain's two commit refs listed");

    let mut pack = Vec::new();
    let reports = tokio::time::timeout(
        Duration::from_secs(5),
        session.fetch(
            vec![FetchRound {
                wants: vec![chain.commit_a],
                haves: Vec::new(),
                done: true,
            }],
            &mut pack,
        ),
    )
    .await
    .expect("the fetch is timely")
    .expect("the fetch on the same stream");
    assert_eq!(reports.len(), 1);
    assert!(
        reports[0].pack_bytes.is_some(),
        "the pack delivered on the persisted session"
    );

    let sent_bytes = sent.lock().expect("spy record").clone();
    assert_eq!(
        sent_bytes
            .windows(b"version 2".len())
            .filter(|window| window == b"version 2")
            .count(),
        1,
        "one advertisement was emitted for the session (advertisement-once)"
    );
    let received_bytes = received.lock().expect("spy record").clone();
    let grammar = String::from_utf8(received_bytes).expect("the request grammar is text");
    assert!(
        grammar.contains("command=ls-refs") && grammar.contains("command=fetch"),
        "both commands rode the one stream: {grammar}"
    );
}

/// Contract stamp: ADR-017 §6 / ADR-009 (the negotiation-rounds
/// budget, client-side; a breach ends the session with a typed error —
/// fail-closed and terminal); core/w4-session-fetch.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn the_negotiation_rounds_budget_fails_closed_typed() {
    let (handles, record, holder) = fetch_fixture("u-7731", AccessAction::Read).await;
    let (client, received, _sent, _joiner) = serve_twin(
        &handles,
        record.clone(),
        &holder,
        AccessAction::Read,
        Limits::default(),
    )
    .await;
    let limits = Limits {
        max_negotiation_rounds: 2,
        ..Limits::default()
    };
    let round = || FetchRound {
        wants: vec![ObjectId::from([1u8; 20])],
        haves: vec![ObjectId::from([2u8; 20])],
        done: false,
    };
    let mut session =
        GitSession::open_via_channels(client, "owner/doubled", Service::UploadPack, limits)
            .expect("the construction");
    let error = tokio::time::timeout(
        Duration::from_secs(5),
        session.fetch(vec![round(), round(), round()], &mut Vec::new()),
    )
    .await
    .expect("the refusal is immediate")
    .expect_err("the plan is over the rounds budget");
    match &error {
        SessionError::BudgetExceeded {
            what,
            budget_name,
            limit,
        } => {
            assert!(what.contains("negotiation"), "{what}");
            assert_eq!(*budget_name, "max_negotiation_rounds");
            assert_eq!(*limit, 2);
        }
        other => panic!("expected the typed rounds breach, got {other:?}"),
    }
    assert!(
        received.lock().expect("spy record").is_empty(),
        "the breach refused before anything touched the wire"
    );
    let again = session
        .fetch(Vec::new(), &mut Vec::new())
        .await
        .unwrap_err();
    assert_eq!(error, again, "the breach is terminal (error-break)");
}

/// Contract stamp: ADR-017 §6 / ADR-009 (the wall clock bounds every
/// operation, fetch included — the stalled peer is the client-side
/// breach); core/w4-session-fetch.
#[tokio::test]
async fn the_wall_clock_bounds_the_fetch_operation_and_the_breach_is_terminal() {
    // The peer answers the handshake's advertisement, then stalls (no
    // fetch response ever arrives).
    let script = vec![
        text_line(b"version 2"),
        text_line(b"ls-refs=unborn"),
        text_line(b"fetch=wait-for-done"),
        text_line(b"object-format=sha1"),
        FLUSH.to_vec(),
    ];
    let client = spawn_fake_peer(script).await;
    let limits = Limits {
        session_wall_clock: Duration::from_millis(300),
        ..Limits::default()
    };
    let mut session =
        GitSession::open_via_channels(client, "owner/doubled", Service::UploadPack, limits)
            .expect("the construction");
    tokio::time::timeout(Duration::from_secs(5), session.handshake())
        .await
        .expect("the handshake is timely")
        .expect("the advertisement consumed");
    let start = tokio::time::Instant::now();
    let done_round = || FetchRound {
        wants: vec![ObjectId::from([1u8; 20])],
        haves: vec![ObjectId::from([2u8; 20])],
        done: true,
    };
    let error = tokio::time::timeout(
        Duration::from_secs(5),
        session.fetch(vec![done_round()], &mut Vec::new()),
    )
    .await
    .expect("the fetch answers at the breach's scale")
    .expect_err("the stalled peer breached the wall clock");
    assert!(
        start.elapsed() < Duration::from_secs(5),
        "the breach answers at the budget's scale"
    );
    match &error {
        SessionError::WallClock { budget } => assert_eq!(*budget, Duration::from_millis(300)),
        other => panic!("expected the wall-clock breach, got {other:?}"),
    }
    let again = session
        .fetch(vec![done_round()], &mut Vec::new())
        .await
        .unwrap_err();
    assert_eq!(error, again, "the breach re-reports without re-entry");
}

/// Contract stamp: ADR-017 §6 / ADR-009 (the response-pack budget —
/// `max_pack_size`, the construction task's recorded mapping; the sink
/// never receives a byte past the budget); core/w4-session-fetch.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn the_response_pack_budget_fails_closed_typed() {
    let (handles, record, holder) = fetch_fixture("u-7731", AccessAction::Read).await;
    let chain = two_commit_chain(&handles.2.objects, &record);
    let (client, _received, _sent, _joiner) = serve_twin(
        &handles,
        record.clone(),
        &holder,
        AccessAction::Read,
        Limits::default(),
    )
    .await;
    let limits = Limits {
        max_pack_size: 32,
        ..Limits::default()
    };
    let mut session =
        GitSession::open_via_channels(client, "owner/doubled", Service::UploadPack, limits)
            .expect("the construction");
    let mut pack = Vec::new();
    let done_round = || FetchRound {
        wants: vec![chain.commit_b],
        haves: Vec::new(),
        done: true,
    };
    let error = tokio::time::timeout(
        Duration::from_secs(5),
        session.fetch(vec![done_round()], &mut pack),
    )
    .await
    .expect("the breach is timely")
    .expect_err("the pack is over the budget");
    match &error {
        SessionError::BudgetExceeded {
            what,
            budget_name,
            limit,
        } => {
            assert!(what.contains("pack"), "{what}");
            assert_eq!(*budget_name, "max_pack_size");
            assert_eq!(*limit, 32);
        }
        other => panic!("expected the typed pack budget breach, got {other:?}"),
    }
    assert!(
        pack.len() as u64 <= 32,
        "the sink never received a byte past the budget: {}",
        pack.len()
    );
    let again = session
        .fetch(vec![done_round()], &mut Vec::new())
        .await
        .unwrap_err();
    assert_eq!(error, again, "the breach is terminal (error-break)");
}

/// Contract stamp: ADR-017 §1 (the caller's negotiation shape), the
/// fail-before-the-wire posture (the `invalid` class, nothing on the
/// wire); core/w4-session-fetch.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_no_done_round_without_haves_and_a_fetch_on_a_push_session_refuse_locally() {
    let (handles, record, holder) = fetch_fixture("u-7731", AccessAction::Read).await;
    let (client, received, _sent, _joiner) = serve_twin(
        &handles,
        record.clone(),
        &holder,
        AccessAction::Read,
        Limits::default(),
    )
    .await;
    let mut session = GitSession::open_via_channels(
        client,
        "owner/doubled",
        Service::UploadPack,
        Limits::default(),
    )
    .expect("the construction");
    let error = session
        .fetch(
            vec![FetchRound {
                wants: vec![ObjectId::from([1u8; 20])],
                haves: Vec::new(),
                done: false,
            }],
            &mut Vec::new(),
        )
        .await
        .expect_err("a no-done round without haves is out of grammar (gix's assert shape)");
    match &error {
        SessionError::InvalidInput(message) => {
            assert!(message.contains("at least one have"), "{message}");
        }
        other => panic!("expected the local refusal, got {other:?}"),
    }
    assert!(
        received.lock().expect("spy record").is_empty(),
        "the refusal happened before anything touched the wire"
    );
    drop(session);

    // The done round terminates the negotiation: a round after it is
    // out of grammar.
    let (handles, record, holder) = fetch_fixture("u-7731", AccessAction::Read).await;
    let (client, _received, _sent, _joiner) = serve_twin(
        &handles,
        record.clone(),
        &holder,
        AccessAction::Read,
        Limits::default(),
    )
    .await;
    let mut session = GitSession::open_via_channels(
        client,
        "owner/doubled",
        Service::UploadPack,
        Limits::default(),
    )
    .expect("the construction");
    match session
        .fetch(
            vec![
                FetchRound {
                    wants: vec![ObjectId::from([1u8; 20])],
                    haves: vec![ObjectId::from([2u8; 20])],
                    done: true,
                },
                FetchRound {
                    wants: vec![ObjectId::from([1u8; 20])],
                    haves: vec![ObjectId::from([2u8; 20])],
                    done: false,
                },
            ],
            &mut Vec::new(),
        )
        .await
    {
        Err(SessionError::InvalidInput(message)) => {
            assert!(message.contains("terminates"), "{message}");
        }
        other => panic!("expected the post-done refusal, got {other:?}"),
    }
    drop(session);

    // Wrong-service misuse: the fetch machine refuses a push session.
    let (client_end, _server_end) = duplex(1 << 16);
    let (client_read, client_write) = tokio::io::split(client_end);
    let mut push_session = GitSession::connect_direct(
        BiStream::from_joined(client_read, client_write),
        "owner/doubled",
        Service::ReceivePack,
        None,
        Limits::default(),
    )
    .expect("the push construction");
    match push_session
        .fetch(
            vec![FetchRound {
                wants: vec![ObjectId::from([1u8; 20])],
                haves: vec![ObjectId::from([2u8; 20])],
                done: false,
            }],
            &mut Vec::new(),
        )
        .await
    {
        Err(SessionError::InvalidInput(message)) => {
            assert!(message.contains("upload-pack"), "{message}");
        }
        other => panic!("expected the wrong-service refusal, got {other:?}"),
    }
    match push_session.ls_refs(&LsRefsArgs::default()).await {
        Err(SessionError::InvalidInput(_)) => {}
        other => panic!("expected the wrong-service refusal for ls-refs, got {other:?}"),
    }
}

/// The hostile-shape row: a peer that sends `ready` on an ack round and
/// a peer whose done round carries no packfile section each fail
/// closed with a typed protocol error — the wait-for-done posture has
/// no ready, and the done round's pack is its terminator.
///
/// Contract stamp: ADR-014 §4 (ready is never in the grammar), ADR-014
/// §2 (the working done round's response is the packfile section);
/// core/w4-session-fetch.
#[tokio::test]
async fn the_hostile_shape_rows_refuse_the_off_contract_responses_typed() {
    // ready on the ack round.
    let script = vec![
        text_line(b"version 2"),
        text_line(b"ls-refs=unborn"),
        text_line(b"fetch=wait-for-done"),
        text_line(b"object-format=sha1"),
        FLUSH.to_vec(),
        text_line(b"acknowledgments"),
        text_line(b"ready"),
        DELIM.to_vec(),
        text_line(b"packfile"),
        pkt(b"\x01packdata"),
        FLUSH.to_vec(),
    ];
    let client = spawn_fake_peer(script).await;
    let mut session = GitSession::open_via_channels(
        client,
        "owner/doubled",
        Service::UploadPack,
        Limits::default(),
    )
    .expect("the construction");
    tokio::time::timeout(Duration::from_secs(5), session.handshake())
        .await
        .expect("the handshake is timely")
        .expect("the advertisement consumed");
    let error = tokio::time::timeout(
        Duration::from_secs(5),
        session.fetch(
            vec![FetchRound {
                wants: vec![ObjectId::from([1u8; 20])],
                haves: vec![ObjectId::from([2u8; 20])],
                done: false,
            }],
            &mut Vec::new(),
        ),
    )
    .await
    .expect("the round is timely")
    .expect_err("ready is off grammar (the wait-for-done posture)");
    match &error {
        SessionError::Protocol(message) => assert!(message.contains("ready"), "{message}"),
        other => panic!("expected the typed protocol refusal, got {other:?}"),
    }
    assert!(
        session
            .fetch(
                vec![FetchRound {
                    wants: vec![ObjectId::from([1u8; 20])],
                    haves: vec![ObjectId::from([2u8; 20])],
                    done: true,
                }],
                &mut Vec::new(),
            )
            .await
            .is_err(),
        "the session is terminal after the refusal"
    );

    // A done round answered with acknowledgments only.
    let script = vec![
        text_line(b"version 2"),
        text_line(b"ls-refs=unborn"),
        text_line(b"fetch=wait-for-done"),
        text_line(b"object-format=sha1"),
        FLUSH.to_vec(),
        text_line(b"acknowledgments"),
        text_line(b"NAK"),
        FLUSH.to_vec(),
    ];
    let client = spawn_fake_peer(script).await;
    let mut session = GitSession::open_via_channels(
        client,
        "owner/doubled",
        Service::UploadPack,
        Limits::default(),
    )
    .expect("the construction");
    tokio::time::timeout(Duration::from_secs(5), session.handshake())
        .await
        .expect("the handshake is timely")
        .expect("the advertisement consumed");
    let error = tokio::time::timeout(
        Duration::from_secs(5),
        session.fetch(
            vec![FetchRound {
                wants: vec![ObjectId::from([1u8; 20])],
                haves: vec![ObjectId::from([2u8; 20])],
                done: true,
            }],
            &mut Vec::new(),
        ),
    )
    .await
    .expect("the round is timely")
    .expect_err("the done round's packfile section is mandatory");
    match &error {
        SessionError::Protocol(message) => {
            assert!(message.contains("packfile"), "{message}")
        }
        other => panic!("expected the typed protocol refusal, got {other:?}"),
    }
}

/// The session surface's honesty pin (ADR-017 §4's storage-agnostic
/// shape): the machine module imports and names no object-store or
/// pack machinery of any kind — the session is the transport + grammar
/// layer, the pack bytes stream to the caller's consumer.
///
/// Contract stamp: ADR-017 §4 (storage-agnostic), ADR-010 (the
/// odb-free protocol core); core/w4-session-fetch.
#[test]
fn the_session_machine_carries_no_object_store_surface() {
    let source = include_str!("../src/session.rs");
    for banned in [
        "gix_odb",
        "gix_object",
        "gix_pack::",
        "gix_traverse",
        "use gix_pack",
    ] {
        assert!(
            !source.contains(banned),
            "the session machine names the object-store/pack machinery {banned:?} — ADR-017 §4's \
             storage-agnostic shape forbids it"
        );
    }
}