alkgit 0.1.0

//! The wave-2 duplex serving entry (core/w2-duplex-serving): the
//! advertisement-once + command-loop session the native transport
//! drives — the marker-gated constructor, the wall clock, the error
//! mapping, and the resolve→authorize front's generic denial — over
//! in-memory duplex pairs with the wave-2 backend doubles. Extended by
//! core/w3-duplex-serving: the `Service::ReceivePack` branch serves the
//! real push session (the V0 advertisement, one request per stream, the
//! duplex tail discipline, the rows below the push block).
//!
//! Rows are names + `Contract stamp:` versions, per
//! `tests/common` §The row/stamp discipline.

mod common;

use std::collections::{BTreeMap, BTreeSet};
use std::sync::Arc;
use std::time::Duration;

use gix_hash::ObjectId;
use tokio::io::{duplex, AsyncReadExt, AsyncWriteExt};
use tokio::sync::Semaphore;
use tokio::task::JoinHandle;

use alkcall::core::Identity;
use alkgit::{
    serve_duplex, serve_duplex_resolved, AccessAction, AuthorizedRepo, GitPackGen, GitPackIngest,
    GitRefs, GitRegistry, GitRegistryStore, Limits, PreparedPush, PushOptions, RefUpdate,
    RepoRecord, Service, StorageError, SubstrateError, Visibility,
};
use common::{
    blob_fixture, chain_pack, commit_fixture, create_command, decode_pack, delete_command,
    push_commands, sideband_report_lines, tag_fixture, tree_fixture, MemObjects, MemPackGen,
    MemPackIngest, MemRefs, MemRegistry,
};

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

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: std::collections::HashMap::new(),
    }
}

/// The marker the adapter hands the entries — constructible only
/// through the checked path (`AuthorizedRepo::authorize`); here the
/// holder identity is granted (at least) the action.
fn marker(record: RepoRecord, identity: &Identity, action: AccessAction) -> AuthorizedRepo {
    AuthorizedRepo::authorize(record, Some(identity), action).expect("the adapter's check passed")
}

/// Encodes one pkt-line, client-side framing style.
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
}

/// A V2 command request, client-side framing: the `command=` line, the
/// `0001` delimiter, the args, the terminating flush.
fn command_request(command: &[u8], pre: &[&str], args: &[&str]) -> Vec<u8> {
    [
        pkt(command),
        pre.iter()
            .flat_map(|l| pkt(l.as_bytes()))
            .collect::<Vec<u8>>(),
        DELIMITER.to_vec(),
        args.iter()
            .flat_map(|l| pkt(l.as_bytes()))
            .collect::<Vec<u8>>(),
        FLUSH.to_vec(),
    ]
    .concat()
}

/// Reads pkt-lines from the client end until (and including) a flush —
/// one advertisement or one command response.
async fn read_pkts(reader: &mut tokio::io::DuplexStream) -> Vec<Vec<u8>> {
    let mut packets = Vec::new();
    loop {
        let mut header = [0u8; 4];
        reader
            .read_exact(&mut header)
            .await
            .expect("pkt-line header");
        let len = usize::from_str_radix(std::str::from_utf8(&header).expect("hex header"), 16)
            .expect("hex length");
        if len < 4 {
            packets.push(Vec::new());
            break;
        }
        let mut payload = vec![0u8; len - 4];
        reader
            .read_exact(&mut payload)
            .await
            .expect("pkt-line payload");
        packets.push(payload);
    }
    packets
}

/// Reads exactly one pkt-line — the session-end error band's shape (an
/// error band is written without a trailing flush; the session closes
/// right after it).
async fn read_band(reader: &mut tokio::io::DuplexStream) -> Vec<u8> {
    let mut header = [0u8; 4];
    reader.read_exact(&mut header).await.expect("band header");
    let len = usize::from_str_radix(std::str::from_utf8(&header).expect("hex header"), 16)
        .expect("hex length");
    let mut payload = vec![0u8; len - 4];
    reader.read_exact(&mut payload).await.expect("band payload");
    payload
}

fn text(payload: &[u8]) -> String {
    String::from_utf8_lossy(payload)
        .trim_end_matches('\n')
        .to_owned()
}

fn text_lines(pkts: &[Vec<u8>]) -> Vec<String> {
    pkts.iter().map(|p| text(p)).collect()
}

/// The advertisement's expected line sequence (feature-aware — the
/// sha256 rider rides the feature per OQ-05's passthrough policy).
fn advertisement_lines() -> Vec<String> {
    #[cfg(not(feature = "sha256"))]
    let lines: [&str; 4] = [
        "version 2",
        "ls-refs=unborn",
        "fetch=wait-for-done",
        "object-format=sha1",
    ];
    #[cfg(feature = "sha256")]
    let lines: [&str; 5] = [
        "version 2",
        "ls-refs=unborn",
        "fetch=wait-for-done",
        "object-format=sha1",
        "object-format=sha256",
    ];
    lines.iter().map(|l| l.to_string()).collect()
}

/// The advertisement's expected packet sequence: the lines plus the
/// terminating flush (read_pkts's trailing empty entry).
fn advertisement_pkts() -> Vec<String> {
    let mut lines = advertisement_lines();
    lines.push(String::new());
    lines
}

/// The adapter's per-session assembly: the markers (read- and
/// write-granted identity), the backend handles keyed on the resolved
/// repo, the wave-2 doubles + the wave-3 ingest double (one object
/// namespace shared by packgen and ingest).
struct Fixture {
    identity: Identity,
    read_marker: AuthorizedRepo,
    write_marker: AuthorizedRepo,
    refs: Arc<MemRefs>,
    gen: Arc<MemPackGen>,
    ingest: Arc<MemPackIngest>,
    objects: MemObjects,
}

fn fixture() -> Fixture {
    let record = repo("owner/doubled");
    let mut read_record = record.clone();
    read_record
        .grants
        .insert("u-7731".to_string(), vec![AccessAction::Read]);
    let mut write_record = record;
    write_record
        .grants
        .insert("u-7731".to_string(), vec![AccessAction::Write]);
    let holder = identity("u-7731");
    let objects = MemObjects::default();
    Fixture {
        identity: holder.clone(),
        read_marker: marker(read_record, &holder, AccessAction::Read),
        write_marker: marker(write_record, &holder, AccessAction::Write),
        refs: Arc::new(MemRefs::default()),
        gen: Arc::new(MemPackGen {
            objects: objects.clone(),
        }),
        ingest: Arc::new(MemPackIngest {
            objects: objects.clone(),
        }),
        objects,
    }
}

/// Seeds the two-commit fixture: `old` (a, the common ancestor) →
/// `main` (b) plus an annotated tag, and refs for the tips; returns the
/// object ids the conversation rows assert against.
struct TwoCommits {
    commit_a: ObjectId,
    commit_b: ObjectId,
    tree_b: ObjectId,
    blob_b: ObjectId,
}

async fn seed_two_commits(fx: &Fixture) -> TwoCommits {
    let doubled = repo("owner/doubled");
    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, 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, tag_bytes),
    ] {
        fx.objects.put(&doubled, id, bytes);
    }
    fx.refs
        .apply_updates(
            &doubled,
            vec![
                RefUpdate {
                    name: "refs/heads/old".to_string(),
                    expected: None,
                    new: Some(commit_a),
                },
                RefUpdate {
                    name: "refs/heads/main".to_string(),
                    expected: None,
                    new: Some(commit_b),
                },
                RefUpdate {
                    name: "refs/tags/v1".to_string(),
                    expected: None,
                    new: Some(tag),
                },
            ],
            false,
        )
        .await
        .unwrap();
    TwoCommits {
        commit_a,
        commit_b,
        tree_b,
        blob_b,
    }
}

/// Serves the marker-gated entry against a fresh in-memory duplex pair;
/// the returned peer end is the client's, the join handle the session's
/// result.
#[allow(clippy::too_many_arguments)]
async fn serve(
    marker: &AuthorizedRepo,
    refs: Arc<dyn GitRefs>,
    gen: Arc<dyn GitPackGen>,
    ingest: Arc<dyn GitPackIngest>,
    push_options_negotiated: bool,
    limits: Limits,
    service: Service,
) -> (
    tokio::io::DuplexStream,
    JoinHandle<Result<(), SubstrateError>>,
) {
    let (peer, stream) = duplex(1 << 16);
    let joined = tokio::spawn({
        let marker = marker.clone();
        async move {
            serve_duplex(
                &marker,
                stream,
                limits,
                service,
                refs,
                gen,
                ingest,
                push_options_negotiated,
                Arc::new(Semaphore::new(limits.max_concurrent_blocking_tasks)),
            )
            .await
        }
    });
    (peer, joined)
}

/// A read-session fixture serve: the advertisement precedes every
/// conversation row.
async fn serve_fixture(
    fx: &Fixture,
    limits: Limits,
) -> (
    tokio::io::DuplexStream,
    JoinHandle<Result<(), SubstrateError>>,
) {
    let refs: Arc<dyn GitRefs> = fx.refs.clone();
    let gen: Arc<dyn GitPackGen> = fx.gen.clone();
    let ingest: Arc<dyn GitPackIngest> = fx.ingest.clone();
    serve(
        &fx.read_marker,
        refs,
        gen,
        ingest,
        false,
        limits,
        Service::UploadPack,
    )
    .await
}

/// A push-session fixture serve: the write marker, the ingest handle,
/// the `push-options` negotiated-gate flag alongside.
async fn serve_push(
    fx: &Fixture,
    ingest: Arc<dyn GitPackIngest>,
    push_options_negotiated: bool,
    limits: Limits,
) -> (
    tokio::io::DuplexStream,
    JoinHandle<Result<(), SubstrateError>>,
) {
    let refs: Arc<dyn GitRefs> = fx.refs.clone();
    let gen: Arc<dyn GitPackGen> = fx.gen.clone();
    serve(
        &fx.write_marker,
        refs,
        gen,
        ingest,
        push_options_negotiated,
        limits,
        Service::ReceivePack,
    )
    .await
}

/// The advertisement is emitted exactly once per session at session
/// start, commands dispatch per request, and the reset discipline holds
/// between commands — three commands in one session (two ls-refs, one
/// fetch's no-done round) are all heard and answered, never swallowed
/// and never re-advertised.
///
/// Contract stamp: ADR-003 (advertisement-once per duplex session —
/// real git hangs on re-advertisement); ADR-005 §1 (the command loop
/// over the request reader, the reset discipline); core/w2-duplex-serving.
#[tokio::test]
async fn advertisement_once_with_three_commands_dispatching_over_the_reset_discipline() {
    let fx = fixture();
    let commits = seed_two_commits(&fx).await;

    let (mut peer, joined) = serve_fixture(&fx, Limits::default()).await;
    let advertisement = read_pkts(&mut peer).await;
    let mut conversation = text_lines(&advertisement);
    assert_eq!(
        conversation.as_slice(),
        advertisement_pkts().as_slice(),
        "the advertisement, emitted once at session start"
    );

    for round in 0..2 {
        peer.write_all(&command_request(
            b"command=ls-refs\n",
            &["agent=git/2.43"],
            &["ref-prefix refs/heads/"],
        ))
        .await
        .unwrap();
        let response = read_pkts(&mut peer).await;
        let lines = text_lines(&response);
        conversation.extend(lines.iter().cloned());
        assert_eq!(
            lines.last(),
            Some(&String::new()),
            "round {round}'s response ends at flush"
        );
        assert_eq!(
            lines[..lines.len() - 1]
                .iter()
                .map(|line| line
                    .split_once(' ')
                    .map_or(line.to_owned(), |(_, name)| name.to_owned()))
                .collect::<Vec<_>>(),
            vec!["refs/heads/main", "refs/heads/old"],
            "round {round}'s ref-prefix-filtered ls-refs response"
        );
    }

    peer.write_all(&command_request(
        b"command=fetch\n",
        &["object-format=sha1"],
        &[
            &format!("want {}", commits.commit_b),
            &format!("have {}", commits.commit_a),
        ],
    ))
    .await
    .unwrap();
    let response = read_pkts(&mut peer).await;
    let lines = text_lines(&response);
    conversation.extend(lines.clone());
    assert_eq!(lines[0], "acknowledgments", "the no-done round's shape");
    assert_eq!(lines[1], format!("ACK {}", commits.commit_a));
    assert_eq!(lines.last(), Some(&String::new()));

    drop(peer);
    joined
        .await
        .expect("the session task joins")
        .expect("the clean close ends the session quietly");

    assert_eq!(
        conversation
            .iter()
            .filter(|line| line == &"version 2")
            .count(),
        1,
        "the advertisement-once rule across the whole conversation"
    );
}

/// A second advertisement between commands never occurs: two ls-refs
/// rounds ride plain ref listings — no capability dump rides between
/// them.
///
/// Contract stamp: ADR-003 (emitted once per duplex session, never
/// between commands — a re-advertisement hangs real git);
/// core/w2-duplex-serving.
#[tokio::test]
async fn a_second_advertisement_never_occurs_between_commands() {
    let fx = fixture();
    seed_two_commits(&fx).await;

    let (mut peer, joined) = serve_fixture(&fx, Limits::default()).await;
    let advertisement = read_pkts(&mut peer).await;
    assert_eq!(text_lines(&advertisement), advertisement_pkts());

    peer.write_all(&command_request(b"command=ls-refs\n", &[], &[]))
        .await
        .unwrap();
    let first = text_lines(&read_pkts(&mut peer).await);
    peer.write_all(&command_request(b"command=ls-refs\n", &[], &[]))
        .await
        .unwrap();
    let second = text_lines(&read_pkts(&mut peer).await);

    assert_eq!(first, second, "both rounds list the same refs");
    let listing: Vec<String> = first
        .iter()
        .filter(|line| !line.is_empty())
        .map(|line| {
            line.split_once(' ')
                .map_or(line.to_owned(), |(_, name)| name.to_owned())
        })
        .collect();
    assert_eq!(
        listing,
        vec!["refs/heads/main", "refs/heads/old", "refs/tags/v1"],
        "the unfiltered ls-refs listing"
    );
    assert!(
        first
            .iter()
            .chain(second.iter())
            .all(|line| line != "version 2"),
        "no capability dump rode between the commands"
    );

    drop(peer);
    joined
        .await
        .expect("the session task joins")
        .expect("the clean close ends the session quietly");
}

/// A full fetch conversation — advertisement → ls-refs → the fetch done
/// round with a real pack from the doubles — passes end-to-end over an
/// in-memory duplex pair; the client's clean close ends the session
/// quietly.
///
/// Contract stamp: ADR-002 (the duplex session entry, the running V2
/// fetch session); ADR-005 §1 (the command loop); ADR-014 §2 (the done
/// round's packfile section); core/w2-duplex-serving (the full
/// conversation over the entry).
#[tokio::test]
async fn a_full_fetch_conversation_passes_end_to_end() {
    let fx = fixture();
    let commits = seed_two_commits(&fx).await;

    let (mut peer, joined) = serve_fixture(&fx, Limits::default()).await;

    let advertisement = read_pkts(&mut peer).await;
    assert_eq!(text_lines(&advertisement), advertisement_pkts());

    peer.write_all(&command_request(
        b"command=ls-refs\n",
        &["symrefs", "unborn"],
        &["ref-prefix refs/heads/"],
    ))
    .await
    .unwrap();
    let ref_lines = text_lines(&read_pkts(&mut peer).await);
    assert_eq!(
        ref_lines[0],
        format!("{} refs/heads/main", commits.commit_b)
    );
    assert_eq!(ref_lines[1], format!("{} refs/heads/old", commits.commit_a));
    assert_eq!(ref_lines.last(), Some(&String::new()));

    peer.write_all(&command_request(
        b"command=fetch\n",
        &["object-format=sha1", "agent=git/2.43"],
        &[
            "thin-pack",
            "no-progress",
            &format!("want {}", commits.commit_b),
            &format!("have {}", commits.commit_a),
            "done",
        ],
    ))
    .await
    .unwrap();
    let fetch_response = read_pkts(&mut peer).await;
    assert_eq!(text(&fetch_response[0]), "packfile");
    let band_chunks: Vec<&[u8]> = fetch_response[1..fetch_response.len() - 1]
        .iter()
        .map(|p| &p[1..])
        .collect();
    let pack: Vec<u8> = band_chunks.concat();
    let decoded = decode_pack(&pack);
    let ids: BTreeSet<ObjectId> = decoded.objects.iter().map(|(id, _, _)| *id).collect();
    assert_eq!(
        ids,
        BTreeSet::from([commits.commit_b, commits.tree_b, commits.blob_b]),
        "closure(want) − closure(recognized have) over the entry"
    );
    let _ = decoded.trailer;
    assert!(
        fetch_response.last().unwrap().is_empty(),
        "flush-terminated"
    );

    drop(peer);
    joined
        .await
        .expect("the session task joins")
        .expect("the clean close ends the session quietly");
}

/// The OQ-19 rider-2 positive-ack capture over the duplex session
/// (the machinery-side run): a multi-have negotiation session where the
/// no-`done` round acks **each** recognized have — per-have `ACK <oid>`
/// lines in the request's order, only the seam-recognized subset (the
/// missing and the exists-but-not-a-commit haves stay unacked), `NAK`
/// absent — and the done round's boundary set is the recognized subset:
/// the unrecognized haves subtract nothing, both commons bound the walk.
///
/// Contract stamp: OQ-19 rider 2 (the positive-ACK capture,
/// tasks/architecture/oq-19-wire-validations.md §2 — the
/// machinery-side run; the substrate-backed re-run is track B wave 7);
/// ADR-014 §1 (ACK per recognized have, NAK only when no ACKs);
/// ADR-014 §2 as amended by review 001 A-4 (the boundary set is the
/// recognized subset); ADR-014 §5 (exists + is-commit — the blob have
/// drives the negative leg); ADR-005 §1 (both rounds ride one session
/// over the command loop).
#[tokio::test]
async fn a_multi_have_session_acks_each_recognized_have_and_the_done_boundary_is_the_recognized_subset(
) {
    let fx = fixture();
    let doubled = repo("owner/doubled");
    let (blob_a, blob_a_bytes) = blob_fixture(b"alpha");
    let (blob_b, blob_b_bytes) = blob_fixture(b"beta");
    let (blob_c, blob_c_bytes) = blob_fixture(b"gamma");
    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 (tree_c, tree_c_bytes) =
        tree_fixture(&[(gix_object::tree::EntryKind::Blob, "c.txt", blob_c)]);
    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 (commit_c, commit_c_bytes) = commit_fixture(tree_c, &[commit_b], "gamma");
    for (id, bytes) in [
        (blob_a, blob_a_bytes),
        (blob_b, blob_b_bytes),
        (blob_c, blob_c_bytes),
        (tree_a, tree_a_bytes),
        (tree_b, tree_b_bytes),
        (tree_c, tree_c_bytes),
        (commit_a, commit_a_bytes),
        (commit_b, commit_b_bytes),
        (commit_c, commit_c_bytes),
    ] {
        fx.objects.put(&doubled, id, bytes);
    }

    let missing = ObjectId::from([9u8; 20]);
    let (mut peer, joined) = serve_fixture(&fx, Limits::default()).await;
    let advertisement = read_pkts(&mut peer).await;
    assert_eq!(text_lines(&advertisement), advertisement_pkts());

    peer.write_all(&command_request(
        b"command=fetch\n",
        &["object-format=sha1", "agent=git/2.43"],
        &[
            &format!("want {}", commit_c),
            &format!("have {}", commit_a),
            &format!("have {}", missing),
            &format!("have {}", commit_b),
        ],
    ))
    .await
    .unwrap();
    let ack_round = text_lines(&read_pkts(&mut peer).await);
    assert_eq!(
        ack_round,
        vec![
            "acknowledgments".to_owned(),
            format!("ACK {}", commit_a),
            format!("ACK {}", commit_b),
            String::new(),
        ],
        "per-have ACKs in request order, unrecognized haves silent, flush-terminated"
    );

    peer.write_all(&command_request(
        b"command=fetch\n",
        &["object-format=sha1", "agent=git/2.43"],
        &[
            "thin-pack",
            "no-progress",
            &format!("want {}", commit_c),
            &format!("have {}", commit_a),
            &format!("have {}", missing),
            &format!("have {}", blob_a),
            &format!("have {}", commit_b),
            "done",
        ],
    ))
    .await
    .unwrap();
    let done_round = read_pkts(&mut peer).await;
    assert_eq!(text(&done_round[0]), "packfile");
    let band_chunks: Vec<&[u8]> = done_round[1..done_round.len() - 1]
        .iter()
        .map(|p| &p[1..])
        .collect();
    let pack: Vec<u8> = band_chunks.concat();
    let decoded = decode_pack(&pack);
    let ids: BTreeSet<ObjectId> = decoded.objects.iter().map(|(id, _, _)| *id).collect();
    assert_eq!(
        ids,
        BTreeSet::from([commit_c, tree_c, blob_c]),
        "the boundary set is the recognized subset: both commons bounded the \
         walk, the missing and the non-commit have subtracted nothing"
    );
    let _ = decoded.trailer;

    drop(peer);
    joined
        .await
        .expect("the session task joins")
        .expect("the clean close ends the session quietly");
}

/// An unknown command declines via the error band and the session ends:
/// the band is the response, the entry returns the typed failure, and
/// nothing follows.
///
/// Contract stamp: ADR-005 §4 (protocol-shaped failures map to the
/// pkt-line error band before the session ends); ADR-003 (honest
/// surface — unimplemented features declined); core/w2-duplex-serving.
#[tokio::test]
async fn an_unknown_command_declines_via_the_error_band() {
    let fx = fixture();

    let (mut peer, joined) = serve_fixture(&fx, Limits::default()).await;
    let advertisement = read_pkts(&mut peer).await;
    assert_eq!(text_lines(&advertisement), advertisement_pkts());

    peer.write_all(&command_request(
        b"command=object-info\n",
        &[],
        &["size 0000"],
    ))
    .await
    .unwrap();

    let band = text(&read_band(&mut peer).await);
    assert!(
        band.starts_with("ERR unknown command \"object-info\""),
        "the decline is the band: {band}"
    );

    let result = joined
        .await
        .expect("the session task joins")
        .expect_err("the decline is a typed failure");
    match &result {
        SubstrateError::Protocol(message) => {
            assert!(message.contains("unknown command"), "{message}");
            assert!(message.contains("object-info"), "{message}");
        }
        other => panic!("an unknown command is protocol-shaped: {other:?}"),
    }

    let mut remaining = [0u8; 1];
    let n = peer.read(&mut remaining).await.expect("read after close");
    assert_eq!(n, 0, "nothing follows the decline band");
}

/// A malformed client request (the machinery's protocol failure)
/// band-maps and ends the session — the entry's error mapping covers the
/// substrate's failures, not just the command dispatch's.
///
/// Contract stamp: ADR-005 §4 (machinery protocol errors band-map);
/// ADR-005 §1 (the unconditional error-break rule — the session ends,
/// never resets-and-retries); core/w2-duplex-serving.
#[tokio::test]
async fn a_malformed_request_bands_and_ends_the_session() {
    let fx = fixture();

    let (mut peer, joined) = serve_fixture(&fx, Limits::default()).await;
    read_pkts(&mut peer).await; // the advertisement

    peer.write_all(b"zzzz").await.unwrap(); // a malformed pkt-line

    let band = text(&read_band(&mut peer).await);
    assert!(band.starts_with("ERR "), "the failure bands");
    assert!(
        band.contains("malformed pkt-line"),
        "the machinery's protocol failure bands: {band}"
    );

    let result = joined
        .await
        .expect("the session task joins")
        .expect_err("a machinery protocol failure is a typed failure");
    assert!(
        matches!(result, SubstrateError::Protocol(_)),
        "the typed failure returns to the caller: {result:?}"
    );
}

/// The wall-clock budget (ADR-009) ends a session that exceeds it —
/// here with a shrunken `Limits`: the stalled read breaches the budget,
/// the entry bands + closes (the duplex mapping), and the typed
/// server/session failure returns.
///
/// Contract stamp: ADR-009 (the session wall clock — enforced by
/// transport's session loop; breach is the server/session class, error
/// band + close on duplex); core/w2-duplex-serving.
#[tokio::test]
async fn the_wall_clock_budget_ends_a_session_that_exceeds_it() {
    let fx = fixture();
    let limits = Limits {
        session_wall_clock: Duration::from_millis(100),
        ..Limits::default()
    };

    let (mut peer, joined) = serve_fixture(&fx, limits).await;
    let advertisement = read_pkts(&mut peer).await;
    assert_eq!(text_lines(&advertisement), advertisement_pkts());

    // No command ever arrives; the stalled read breaches the budget.
    let band = text(&read_band(&mut peer).await);
    assert!(
        band.starts_with("ERR ") && band.contains("session_wall_clock"),
        "the breach bands the session: {band}"
    );

    let mut remaining = [0u8; 1];
    let n = peer.read(&mut remaining).await.expect("read after close");
    assert_eq!(n, 0, "the breached session closed");

    let result = joined
        .await
        .expect("the session task joins")
        .expect_err("the breach is a typed failure");
    match &result {
        SubstrateError::Protocol(message) => {
            assert!(message.contains("session_wall_clock"), "{message}");
        }
        other => panic!("a wall-clock breach is protocol-shaped: {other:?}"),
    }
}

/// Unknown-repo ≡ unauthorized: the resolve→authorize front's denial is
/// byte-identical for a `RegistryError::NotFound` resolution and a
/// failed `authorize` evaluation, and no protocol byte precedes the
/// denial.
///
/// Contract stamp: ADR-007 steps 2–3 (resolve, then authorize, before
/// any protocol output); ADR-008/N-2 (unknown ≡ unauthorized — no
/// existence oracle); core/w2-duplex-serving (the front's collapse).
#[tokio::test]
async fn unknown_repo_and_unauthorized_denials_are_indistinguishable() {
    let fx = fixture();
    let holder = fx.identity.clone();

    let unknown_registry: Arc<dyn GitRegistry> = Arc::new(MemRegistry::default());

    let unauthorized_registry = MemRegistry::default();
    unauthorized_registry
        .put_repo(&repo("owner/doubled"))
        .await
        .unwrap();
    assert_eq!(
        AuthorizedRepo::authorize(
            unauthorized_registry
                .resolve("owner/doubled")
                .await
                .unwrap(),
            Some(&holder),
            AccessAction::Read
        ),
        None,
        "the grantless identity on the private repo denies"
    );
    let unauthorized_registry: Arc<dyn GitRegistry> = Arc::new(unauthorized_registry);

    let (unknown, unknown_wire) = denial_wire(unknown_registry, Service::UploadPack).await;
    let (unauthorized, unauthorized_wire) =
        denial_wire(unauthorized_registry, Service::UploadPack).await;

    match (&unknown, &unauthorized) {
        (Err(SubstrateError::AccessDenied), Err(SubstrateError::AccessDenied)) => {}
        other => panic!("both denials collapse to AccessDenied: {other:?}"),
    }

    assert!(!unknown_wire.is_empty());
    assert_eq!(
        unknown_wire, unauthorized_wire,
        "the denials are byte-identical"
    );
    let denial = pkt_payloads(&unknown_wire);
    assert_eq!(
        denial,
        vec![b"ERR access denied: owner/doubled".to_vec()],
        "the collapsed band names only what the client sent, and no protocol byte precedes it"
    );
    assert!(
        !String::from_utf8_lossy(&unknown_wire).contains("not found")
            && !String::from_utf8_lossy(&unknown_wire).contains("grant"),
        "no cause leaks through the denial band"
    );
}

/// Parses a raw captured wire byte sequence into pkt-line payloads (the
/// denial capture's shape: frames only, no trailing flush).
fn pkt_payloads(wire: &[u8]) -> Vec<Vec<u8>> {
    let mut payloads = Vec::new();
    let mut i = 0;
    while i < wire.len() {
        assert!(i + 4 <= wire.len(), "trailing partial header");
        let len = usize::from_str_radix(
            std::str::from_utf8(&wire[i..i + 4]).expect("hex header"),
            16,
        )
        .expect("hex length");
        assert!(len >= 4 && i + len <= wire.len(), "partial pkt-line");
        payloads.push(wire[i + 4..i + len].to_vec());
        i += len;
    }
    payloads
}

/// Drives the resolve→authorize front to its denial and captures the
/// denial's wire bytes through a live reader (the client's view: the
/// band is the first thing on the wire).
async fn denial_wire(
    registry: Arc<dyn GitRegistry>,
    service: Service,
) -> (Result<(), SubstrateError>, Vec<u8>) {
    let (mut peer, stream) = duplex(1 << 16);
    let holder = identity("u-7731");
    let refs: Arc<dyn GitRefs> = Arc::new(MemRefs::default());
    let gen: Arc<dyn GitPackGen> = Arc::new(MemPackGen::default());
    let ingest: Arc<dyn GitPackIngest> = Arc::new(MemPackIngest::default());
    let joined = tokio::spawn(async move {
        serve_duplex_resolved(
            "owner/doubled",
            Some(&holder),
            service,
            stream,
            Limits::default(),
            registry,
            refs,
            gen,
            ingest,
            false,
            Arc::new(Semaphore::new(
                Limits::default().max_concurrent_blocking_tasks,
            )),
        )
        .await
    });

    let mut wire = Vec::new();
    let mut chunk = [0u8; 256];
    let drained = tokio::time::timeout(Duration::from_secs(5), async {
        loop {
            let n = peer.read(&mut chunk).await.expect("the denial wire");
            if n == 0 {
                break;
            }
            wire.extend_from_slice(&chunk[..n]);
        }
    })
    .await;
    if drained.is_err() {
        panic!("the denial never closed the stream");
    }
    (joined.await.expect("the front task joins"), wire)
}

/// Reads the raw response bytes up to and including the terminating
/// flush — one advertisement or one report body.
async fn read_body_until_flush(reader: &mut tokio::io::DuplexStream) -> Vec<u8> {
    let mut body = Vec::new();
    loop {
        let mut header = [0u8; 4];
        reader.read_exact(&mut header).await.expect("header");
        body.extend_from_slice(&header);
        let len = usize::from_str_radix(std::str::from_utf8(&header).expect("hex header"), 16)
            .expect("hex length");
        if len < 4 {
            return body;
        }
        let mut payload = vec![0u8; len - 4];
        reader.read_exact(&mut payload).await.expect("payload");
        body.extend_from_slice(&payload);
    }
}

/// A prepare double that counts its invocations — the observable for
/// the packless peek's path (no prepare call) and for fail-closed
/// breaches (no backend call).
struct CountedIngest {
    inner: Arc<MemPackIngest>,
    runs: Arc<std::sync::atomic::AtomicUsize>,
}

fn counting_ingest(
    inner: Arc<MemPackIngest>,
    runs: Arc<std::sync::atomic::AtomicUsize>,
) -> Arc<dyn GitPackIngest> {
    Arc::new(CountedIngest { inner, runs })
}

#[async_trait::async_trait]
impl GitPackIngest for CountedIngest {
    async fn prepare(
        &self,
        record: &RepoRecord,
        updates: Vec<RefUpdate>,
        pack: Box<dyn std::io::Read + Send>,
        push_options: Option<PushOptions>,
        limits: Limits,
    ) -> Result<PreparedPush, StorageError> {
        use std::sync::atomic::Ordering::SeqCst;
        self.runs.fetch_add(1, SeqCst);
        self.inner
            .prepare(record, updates, pack, push_options, limits)
            .await
    }
}

/// A prepare double that records the push-options it received and
/// delegates — the observable for the duplex options-section routing.
struct OptionsRecorder {
    inner: Arc<MemPackIngest>,
    seen: Arc<std::sync::Mutex<Option<PushOptions>>>,
}

#[async_trait::async_trait]
impl GitPackIngest for OptionsRecorder {
    async fn prepare(
        &self,
        record: &RepoRecord,
        updates: Vec<RefUpdate>,
        pack: Box<dyn std::io::Read + Send>,
        push_options: Option<PushOptions>,
        limits: Limits,
    ) -> Result<PreparedPush, StorageError> {
        *self.seen.lock().expect("seen is not poisoned") = push_options.clone();
        self.inner
            .prepare(record, updates, pack, push_options, limits)
            .await
    }
}

/// The full push session over the duplex entry (ADR-013 §3's one
/// request per stream): the V0 ref advertisement speaks first with no
/// service prefix (the empty repo's `capabilities^{}` sentinel with the
/// caps NUL-attached, the honest served set — `quiet` absent), then one
/// request — the create command + the pack, dribbled in chunks —
/// ingests and CAS-applies, the report streams back sideband-framed,
/// and the session ends cleanly after the report (the client's close
/// releases the relay's park; nothing rides after the report).
///
/// Contract stamp: ADR-013 §1–2 (the V0 advertisement, server speaks
/// first, no service prefix; the sentinel for the no-refs repo; the
/// honest set), §3 (one POST / one stream), §7 (the one-transaction CAS
/// home), §8 (the sideband report); ADR-020 §1 (the duplex tail
/// discipline — the bounded peek; rule (iv)'s clean close after the
/// report); ADR-005 §4 (the protocol-failure band mapping); ADR-009
/// (the wall clock over the whole push); core/w3-duplex-serving.
#[tokio::test]
async fn a_full_push_conversation_applies_and_reports_over_the_duplex_entry() {
    let fx = fixture();
    let record = repo("owner/doubled");
    let chain = chain_pack();

    let (mut peer, joined) = serve_push(&fx, fx.ingest.clone(), false, Limits::default()).await;

    let advertisement = read_pkts(&mut peer).await;
    let lines = text_lines(&advertisement);
    assert_eq!(
        lines.len(),
        2,
        "an empty repo advertises exactly the sentinel line + flush: {lines:?}"
    );
    let caps_line = &lines[0];
    assert!(
        caps_line.contains("capabilities^{}"),
        "the empty repo's sentinel: {caps_line}"
    );
    assert!(
        caps_line.contains('\0'),
        "the caps ride NUL-attached on the first emitted line: {caps_line}"
    );
    assert!(
        caps_line.contains("agent=alkgit/"),
        "the agent token is advertised: {caps_line}"
    );
    for served in ["report-status", "side-band-64k", "atomic"] {
        assert!(
            caps_line.contains(served),
            "{served} stays served: {caps_line}"
        );
    }
    assert!(
        !caps_line.contains("quiet"),
        "quiet is not advertised (ADR-013 §2): {caps_line}"
    );
    assert!(
        !lines.iter().all(|line| line.contains("# service=")),
        "no service prefix rides the duplex push advertisement"
    );
    assert_eq!(*lines.last().expect("the flush"), "", "flush-terminated");

    let request = [
        push_commands(
            "report-status side-band-64k",
            &[create_command("refs/heads/main", chain.commit_b)],
        ),
        vec![
            chain.pack[..1].to_vec(),
            chain.pack[1..24].to_vec(),
            chain.pack[24..].to_vec(),
        ],
    ]
    .concat()
    .concat();
    peer.write_all(&request).await.unwrap();

    let report_body = read_body_until_flush(&mut peer).await;
    assert_eq!(
        sideband_report_lines(&report_body),
        vec!["unpack ok".to_owned(), "ok refs/heads/main".to_owned()],
        "the report is unpack + per-ref in request order"
    );

    drop(peer);
    joined
        .await
        .expect("the session task joins")
        .expect("the clean close ends the session quietly after the report");

    assert_eq!(
        fx.refs
            .list_refs(&record)
            .await
            .unwrap()
            .iter()
            .filter_map(|line| Some((line.name.clone(), line.oid?)))
            .collect::<BTreeMap<String, ObjectId>>(),
        BTreeMap::from([("refs/heads/main".to_owned(), chain.commit_b)]),
        "the CAS applied the pushed update"
    );
    assert!(
        fx.ingest.objects().contains(&record, &chain.commit_b),
        "the pushed objects landed via the ingest handle"
    );
}

/// The packless delete-only push (ADR-020 §1 rule (ii)): the client
/// sends commands + flush and stays open — the bounded peek times out
/// with zero pack bytes, the packless CAS-only path runs with **no
/// prepare call**, and the same report shape (the ~40 ms-datum capture)
/// streams back while the client is still open. The client's close
/// after the report ends the drain cleanly (rule (iv) — no mid-stream
/// RST; the session returns quietly, never an error).
///
/// Contract stamp: ADR-020 §1 rules (ii)/(iv) (the packless path, the
/// clean-close discipline), §3 (the client's write-side-open waiting
/// shape); ADR-013 §7–8 (the CAS-only path, the same report);
/// ADR-009 (the peek window is the named `packless_peek` budget);
/// core/w3-duplex-serving.
#[tokio::test]
async fn a_packless_delete_push_takes_the_peek_packless_path_and_drains_cleanly() {
    use std::sync::atomic::Ordering::SeqCst;

    let fx = fixture();
    let record = repo("owner/doubled");
    let current = ObjectId::from([0x1b; 20]);
    fx.refs
        .apply_updates(
            &record,
            vec![RefUpdate {
                name: "refs/heads/gone".to_owned(),
                expected: None,
                new: Some(current),
            }],
            false,
        )
        .await
        .unwrap();

    let runs = Arc::new(std::sync::atomic::AtomicUsize::new(0));
    let (mut peer, joined) = serve_push(
        &fx,
        counting_ingest(fx.ingest.clone(), Arc::clone(&runs)),
        false,
        Limits::default(),
    )
    .await;

    read_pkts(&mut peer).await;
    peer.write_all(
        &push_commands(
            "report-status side-band-64k",
            &[delete_command("refs/heads/gone", current)],
        )
        .concat(),
    )
    .await
    .unwrap();

    let report_body = read_body_until_flush(&mut peer).await;
    assert_eq!(
        sideband_report_lines(&report_body),
        vec!["unpack ok".to_owned(), "ok refs/heads/gone".to_owned()],
        "the packless report equals the capture's shape, while the client stays open"
    );
    assert_eq!(
        runs.load(SeqCst),
        0,
        "the packless path runs no prepare call"
    );
    assert!(
        !fx.refs
            .list_refs(&record)
            .await
            .unwrap()
            .iter()
            .any(|line| line.name == "refs/heads/gone"),
        "the CAS-only path deleted the ref"
    );

    drop(peer);
    joined
        .await
        .expect("the session task joins")
        .expect("the packless tail's drain ends cleanly, no mid-stream RST");
}

/// The nothing-to-do shape (ADR-013 §3): an immediate flush with no
/// commands after the advertisement — the machine replies flush, no
/// report; the session ends cleanly.
///
/// Contract stamp: ADR-013 §3 (the nothing-to-do shape — reply flush,
/// no report); ADR-020 §1 (its duplex read shape); ADR-009 (no
/// prepare call, no peek); core/w3-duplex-serving.
#[tokio::test]
async fn a_push_nothing_to_do_flush_replies_flush_with_no_report() {
    use std::sync::atomic::Ordering::SeqCst;

    let fx = fixture();
    let runs = Arc::new(std::sync::atomic::AtomicUsize::new(0));
    let (mut peer, joined) = serve_push(
        &fx,
        counting_ingest(fx.ingest.clone(), Arc::clone(&runs)),
        false,
        Limits::default(),
    )
    .await;

    read_pkts(&mut peer).await;
    peer.write_all(FLUSH).await.unwrap();

    let reply = read_pkts(&mut peer).await;
    assert_eq!(
        reply,
        vec![Vec::<u8>::new()],
        "exactly one reply flush, no report"
    );

    drop(peer);
    joined
        .await
        .expect("the session task joins")
        .expect("the session ends quietly after the reply");
    assert_eq!(
        runs.load(SeqCst),
        0,
        "the nothing-to-do shape runs no prepare call"
    );
}

/// A partial pack header at the peek timeout is the malformed-push
/// session error (ADR-020 §1 rule (iii) — a half-sent pack cannot be
/// re-interpreted as absence): the error bands, no report, no ref
/// applies, no prepare call.
///
/// Contract stamp: ADR-020 §1 rule (iii); ADR-005 §4 (the
/// protocol-failure band); ADR-009 (the peek window bounds the gather,
/// observed bytes still classify);
/// core/w3-duplex-serving.
#[tokio::test]
async fn a_partial_pack_header_at_the_peek_timeout_is_a_malformed_push() {
    use std::sync::atomic::Ordering::SeqCst;

    let fx = fixture();
    let record = repo("owner/doubled");
    let runs = Arc::new(std::sync::atomic::AtomicUsize::new(0));
    let limits = Limits {
        packless_peek: Duration::from_millis(150),
        ..Limits::default()
    };
    let (mut peer, joined) = serve_push(
        &fx,
        counting_ingest(fx.ingest.clone(), Arc::clone(&runs)),
        false,
        limits,
    )
    .await;

    read_pkts(&mut peer).await;
    peer.write_all(
        &push_commands(
            "report-status",
            &[create_command(
                "refs/heads/main",
                ObjectId::from([0x2c; 20]),
            )],
        )
        .concat(),
    )
    .await
    .unwrap();
    peer.write_all(b"PA").await.unwrap();

    let band = text(&read_band(&mut peer).await);
    assert!(
        band.starts_with("ERR ") && band.contains("half-sent pack"),
        "the partial header is the malformed-push band: {band}"
    );
    let result = joined
        .await
        .expect("the session task joins")
        .expect_err("the malformed push is a typed failure");
    assert!(
        matches!(result, SubstrateError::Protocol(_)),
        "a malformed push is protocol-shaped (banded): {result:?}"
    );
    assert_eq!(runs.load(SeqCst), 0, "no prepare call ran");
    assert!(
        fx.refs.list_refs(&record).await.unwrap().is_empty(),
        "no ref applied"
    );
}

/// The negotiated `push-options` section is consumed before the pack
/// peek (ADR-020 §1 rule (i)) and routed into `prepare`: the
/// gate-on advertisement carries the `push-options` token; the client
/// selects the cap and sends the options section + its flush between
/// the command flush and the pack — the pack still ingests byte-exact
/// past the second flush.
///
/// Contract stamp: ADR-013 §11 (the section's grammar, the
/// negotiated gate); ADR-020 §1 rule (i) (consumed before the peek);
/// ADR-013 §2 (the gate rides the advertisement);
/// core/w3-duplex-serving.
#[tokio::test]
async fn the_negotiated_push_options_section_is_consumed_before_the_pack_peek() {
    let fx = fixture();
    let record = repo("owner/doubled");
    let chain = chain_pack();

    let seen = Arc::new(std::sync::Mutex::new(None));
    let ingest: Arc<dyn GitPackIngest> = Arc::new(OptionsRecorder {
        inner: Arc::clone(&fx.ingest),
        seen: Arc::clone(&seen),
    });
    let (mut peer, joined) = serve_push(&fx, ingest, true, Limits::default()).await;

    let advertisement = text_lines(&read_pkts(&mut peer).await);
    assert!(
        advertisement[0].contains("push-options"),
        "the gate's token rides the gate-on advertisement: {:?}",
        advertisement[0]
    );

    let request = [
        push_commands(
            "report-status side-band-64k push-options",
            &[create_command("refs/heads/main", chain.commit_b)],
        ),
        vec![pkt(b"deploy-key=abc"), FLUSH.to_vec(), chain.pack.clone()],
    ]
    .concat()
    .concat();
    peer.write_all(&request).await.unwrap();

    let report_body = read_body_until_flush(&mut peer).await;
    assert_eq!(
        sideband_report_lines(&report_body),
        vec!["unpack ok".to_owned(), "ok refs/heads/main".to_owned()],
    );
    assert_eq!(
        *seen.lock().expect("seen is not poisoned"),
        Some(PushOptions {
            pairs: vec![("deploy-key".to_owned(), "abc".to_owned())]
        }),
        "the negotiated options section routed into prepare"
    );

    drop(peer);
    joined
        .await
        .expect("the session task joins")
        .expect("the clean close after the report");
    assert!(
        fx.ingest.objects().contains(&record, &chain.commit_b),
        "the pack after the options flush ingested byte-exact"
    );
}

/// The denial path (ADR-007/ADR-008): an unauthorized push collapses to
/// the generic band *before any advertisement byte* — the resolve→
/// authorize gate precedes the V0 advertisement, unknown ≡ unauthorized.
///
/// Contract stamp: ADR-007 steps 2–3 (the gate precedes any protocol
/// byte — the advertisement included); ADR-008/N-2 (unknown ≡
/// unauthorized); ADR-011 §3/ADR-015 §2 (push is the write action);
/// core/w3-duplex-serving.
#[tokio::test]
async fn a_denied_push_is_the_generic_band_before_any_advertisement_byte() {
    let unauthorized_registry = MemRegistry::default();
    unauthorized_registry
        .put_repo(&repo("owner/doubled"))
        .await
        .unwrap();
    let unauthorized_registry: Arc<dyn GitRegistry> = Arc::new(unauthorized_registry);

    let (result, wire) = denial_wire(unauthorized_registry, Service::ReceivePack).await;
    match &result {
        Err(SubstrateError::AccessDenied) => {}
        other => panic!("the denial is the typed AccessDenied: {other:?}"),
    }
    assert_eq!(
        pkt_payloads(&wire),
        vec![b"ERR access denied: owner/doubled".to_vec()],
        "the generic band names only the client-sent id; no advertisement preceded it"
    );
}

/// The wall clock bounds the whole push (ADR-009): a client that stalls
/// after the advertisement (no request ever read, the peek never even
/// reached) breaches the session budget — the error band + close, the
/// typed server/session failure.
///
/// Contract stamp: ADR-009 (the session wall clock over the whole
/// session, push included; breach is the server/session class — error
/// band + close on duplex); ADR-020 §1 (the peek window runs inside
/// the wall clock, never outside it); core/w3-duplex-serving.
#[tokio::test]
async fn the_wall_clock_bounds_the_push_session() {
    let fx = fixture();
    let limits = Limits {
        session_wall_clock: Duration::from_millis(200),
        ..Limits::default()
    };

    let (mut peer, joined) = serve_push(&fx, fx.ingest.clone(), false, limits).await;
    read_pkts(&mut peer).await;
    // Nothing is ever sent: the stalled request read breaches the budget.
    let band = text(&read_band(&mut peer).await);
    assert!(
        band.starts_with("ERR ") && band.contains("session_wall_clock"),
        "the breach bands the push session: {band}"
    );

    let mut remaining = [0u8; 1];
    assert_eq!(
        peer.read(&mut remaining).await.expect("read after close"),
        0,
        "the breached session closed"
    );
    let result = joined
        .await
        .expect("the session task joins")
        .expect_err("the breach is a typed failure");
    assert!(
        matches!(
            &result,
            SubstrateError::Protocol(message) if message.contains("session_wall_clock")
        ),
        "the typed failure returns: {result:?}"
    );
}

/// A push advertisement listing failure bands the session (the
/// backend-failure mapping — ADR-005 §4 through the
/// storage-to-session map, the ADR-008 posture at the advertisement):
/// nothing is emitted, never a partial ref list.
///
/// Contract stamp: ADR-013 §2 (the advertisement's listing — the one
/// backend call; failures map through the storage-to-session error
/// model, review 001 N-1's no-partial-list rule); ADR-005 §4
/// (backend failures band-map);
/// core/w3-duplex-serving.
#[tokio::test]
async fn a_push_advertisement_listing_failure_bands_the_session() {
    struct ListingFailure;
    #[async_trait::async_trait]
    impl GitRefs for ListingFailure {
        async fn list_refs(
            &self,
            _repo: &RepoRecord,
        ) -> Result<Vec<alkgit::RefLine>, StorageError> {
            Err(StorageError::Invalid("listing unavailable".to_string()))
        }

        async fn apply_updates(
            &self,
            _repo: &RepoRecord,
            _updates: Vec<RefUpdate>,
            _atomic: bool,
        ) -> Result<Vec<alkgit::RefOutcome>, StorageError> {
            Err(StorageError::Invalid("listing unavailable".to_string()))
        }
    }

    let fx = fixture();
    let refs: Arc<dyn GitRefs> = Arc::new(ListingFailure);
    let gen: Arc<dyn GitPackGen> = fx.gen.clone();
    let ingest: Arc<dyn GitPackIngest> = fx.ingest.clone();

    let (mut peer, joined) = serve(
        &fx.write_marker,
        refs,
        gen,
        ingest,
        false,
        Limits::default(),
        Service::ReceivePack,
    )
    .await;

    let band = text(&read_band(&mut peer).await);
    assert!(band.starts_with("ERR "), "a listing failure bands: {band}");
    assert!(
        band.contains("listing unavailable"),
        "the backend failure's own text bands through the map: {band}"
    );
    assert_eq!(
        peer.read(&mut [0u8; 1]).await.expect("read after close"),
        0,
        "the banded session closed"
    );
    let result = joined
        .await
        .expect("the session task joins")
        .expect_err("the failure is typed");
    assert!(
        matches!(result, SubstrateError::Protocol(_)),
        "a backend failure band-maps to a protocol-shaped typed failure: {result:?}"
    );
}

/// A resolved front with a passing gate reaches the session: an
/// anonymous caller on a public repo — the anonymous-first-class read
/// path — gets the advertisement, and unknown-repo/unauthorized callers
/// get only the generic denial (nothing in between).
///
/// Contract stamp: ADR-011 §3 (anonymous public read is first-class at
/// the gate); ADR-007 (the gate precedes the advertisement);
/// core/w2-duplex-serving.
#[tokio::test]
async fn the_resolved_front_serves_the_anonymous_public_read_session() {
    let fx = fixture();
    let mut public = repo("owner/doubled");
    public.visibility = Visibility::Public;
    let registry = MemRegistry::default();
    registry.put_repo(&public).await.unwrap();
    let registry: Arc<dyn GitRegistry> = Arc::new(registry);

    let (mut peer, stream) = duplex(1 << 16);
    let refs: Arc<dyn GitRefs> = fx.refs.clone();
    let gen: Arc<dyn GitPackGen> = fx.gen.clone();
    let ingest: Arc<dyn GitPackIngest> = fx.ingest.clone();
    let joined = tokio::spawn(async move {
        serve_duplex_resolved(
            "owner/doubled",
            None,
            Service::UploadPack,
            stream,
            Limits::default(),
            registry,
            refs,
            gen,
            ingest,
            false,
            Arc::new(Semaphore::new(
                Limits::default().max_concurrent_blocking_tasks,
            )),
        )
        .await
    });

    let advertisement = read_pkts(&mut peer).await;
    assert_eq!(text_lines(&advertisement), advertisement_pkts());

    peer.write_all(&command_request(b"command=ls-refs\n", &[], &[]))
        .await
        .unwrap();
    let _ = read_pkts(&mut peer).await;

    drop(peer);
    joined
        .await
        .expect("the front task joins")
        .expect("the clean close ends the session quietly");
}

/// The entry's marker is the type-level ACL ordering: the only
/// constructor path into the entries' input type runs the checked
/// policy — a denied evaluation yields no marker value, so an
/// unauthorized session cannot reach [`serve_duplex`] at any type level.
///
/// Contract stamp: ADR-007 step 3 (the marker is the only path in —
/// resolve → authorize precedes any protocol byte); ADR-011 §3 (the
/// policy evaluation producing it); core/w2-duplex-serving.
#[test]
fn only_the_checked_marker_path_feeds_the_entries() {
    let private = repo("owner/doubled");
    let holder = identity("u-7731");
    assert_eq!(
        AuthorizedRepo::authorize(private.clone(), None, AccessAction::Read),
        None,
        "anonymous on a private repo is no marker"
    );
    assert_eq!(
        AuthorizedRepo::authorize(private.clone(), Some(&holder), AccessAction::Read),
        None,
        "a grantless identity is no marker"
    );
    let mut granted = private;
    granted
        .grants
        .insert("u-7731".to_string(), vec![AccessAction::Read]);
    assert!(
        AuthorizedRepo::authorize(granted, Some(&holder), AccessAction::Read).is_some(),
        "the granted identity is the entry's input"
    );
}