alkgit 0.1.0

//! The wave-4 GitSession construction (core/w4-session-construction):
//! the consumer half's wrapper over gix-transport's git-daemon
//! `Connection` on the alkcall `BiStream` compat halves — both ADR-016
//! preamble modes (the direct mode's request-line bytes byte-pinned
//! against the grammar the server's `parse_request_line` accepts; the
//! channels mode's zero preamble bytes), the client-side `Limits`
//! wiring (fail-closed, typed), the error-break discipline, the
//! version pins, and the channels dial contract — rows driven against
//! the in-crate server entries (`serve_duplex`, the channels open op)
//! over in-memory pairs with the wave-2/3 backend doubles, per the
//! POC-X-validated shape. Extended by the R-1 manifest drift pin.
//!
//! Rows are names + `Contract stamp:` versions, per `tests/common`
//! §The row/stamp discipline.

mod common;

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

use alkcall::channels::adapter::{ChannelsAdapter, InstallChannelZero};
use alkcall::channels::client::ChannelClient;
use alkcall::channels::operations::ChannelCore;
use alkcall::channels::policy::{default_policy, NoCap};
use alkcall::core::auth::{AuthContext, Identity};
use alkcall::core::types::{BiStream, Connection, ProtocolHandler};
use alkcall::protocol::connection::{split_single_stream, CallConnection};
use alkcall::protocol::dispatch::Dispatcher;
use alkcall::registry::registration::OperationRegistry;
use bstr::ByteSlice as _;
use serde_json::json;
use tokio::io::{duplex, AsyncRead, AsyncWrite, AsyncWriteExt, ReadBuf};
use tokio::sync::Semaphore;

use gix_transport::Protocol;

use alkgit::{
    register_openable, serve_duplex, AccessAction, AuthorizedRepo, GitPackGen, GitPackIngest,
    GitRefs, GitRegistry, GitRegistryStore, GitSession, Limits, RefUpdate, RepoRecord, Service,
    SubstrateError, Visibility, GIT_ALPN, OP_GIT_OPEN,
};
use common::{MemPackGen, MemPackIngest, MemRefs, MemRegistry};

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(),
    }
}

fn marker(record: RepoRecord, holder: &Identity, action: AccessAction) -> AuthorizedRepo {
    AuthorizedRepo::authorize(record, Some(holder), action).expect("the checked marker")
}

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

/// The pinned V2 capability list minus its `version 2` line — what the
/// handshake's capabilities parse out of our server's advertisement.
fn pinned_v2_caps() -> Vec<String> {
    #[cfg(not(feature = "sha256"))]
    let caps: [&str; 3] = [
        "ls-refs=unborn",
        "fetch=wait-for-done",
        "object-format=sha1",
    ];
    #[cfg(feature = "sha256")]
    let caps: [&str; 4] = [
        "ls-refs=unborn",
        "fetch=wait-for-done",
        "object-format=sha1",
        "object-format=sha256",
    ];
    caps.iter().map(|s| s.to_string()).collect()
}

/// The handshake's capabilities as `name[=value]` tokens — the
/// advertisement's readable form (gix's parse state).
fn capability_tokens(caps: &gix_transport::client::Capabilities) -> Vec<String> {
    caps.iter()
        .map(|c| match c.value() {
            Some(value) => format!(
                "{}={}",
                String::from_utf8_lossy(c.name()),
                String::from_utf8_lossy(value)
            ),
            None => String::from_utf8_lossy(c.name()).into_owned(),
        })
        .collect()
}

// --- the wire spies --------------------------------------------------------

/// Records every byte read off the wire — behind the zero-preamble
/// assertion (what the server RECEIVED from the client) and the
/// request-line byte pins.
struct RecordRead<S> {
    inner: S,
    record: Arc<Mutex<Vec<u8>>>,
}

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: Arc<Mutex<Vec<u8>>>,
}

impl<S: AsyncWrite + Unpin> 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)
    }
}

type WireRecord = Arc<Mutex<Vec<u8>>>;
type SessionJoiner = tokio::task::JoinHandle<Result<(), SubstrateError>>;

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 }),
    )
}

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

/// Spawns the in-crate duplex serving entry (`serve_duplex`) over a
/// pair whose server end is spy-wrapped; returns the client end plus
/// both wire records and the serving task's joiner.
fn serve_twin(
    handles: &Handles,
    record: RepoRecord,
    holder: &Identity,
    action: AccessAction,
    service: Service,
    limits: Limits,
) -> (BiStream, WireRecord, WireRecord, SessionJoiner) {
    let checked = marker(record, holder, action);
    let _registry: Arc<dyn GitRegistry> = handles.0.clone();
    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,
            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 (the native_transport harness shape):
/// yields each server-side stream once, then reports the connection
/// closed — the adapter's accept loop.
struct YieldQueued {
    streams: tokio::sync::Mutex<std::collections::VecDeque<BiStream>>,
    received: Arc<Mutex<Vec<u8>>>,
}

#[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)?;
        // The server end wraps the wire spies: the denial rows' wire and
        // the request-line's server-side reachability ride the same
        // records.
        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) {}
}

struct DenialPeer {
    client: BiStream,
    received: Arc<Mutex<Vec<u8>>>,
}

/// Runs the direct-ALPN adapter over one stream — the request-line
/// read happens at the adapter before the gate (the deny can only
/// arrive after the session's line rode out).
async fn denial_peer(handles: &Handles, opener: Option<Identity>) -> DenialPeer {
    let (client_end, server_end) = duplex(1 << 16);
    let received = Arc::new(Mutex::new(Vec::new()));
    let (server_read, server_write) = tokio::io::split(server_end);
    let served = BiStream::from_joined(server_read, server_write);
    let adapter = alkgit::GitAdapter::new(
        handles.0.clone(),
        handles.1.clone(),
        handles.2.clone(),
        handles.3.clone(),
        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 = AuthContext::anonymous(b"alk/git");
    auth.identity = opener;
    tokio::spawn(async move {
        let _ = ProtocolHandler::handle(&adapter, connection, &auth).await;
    });
    let (client_read, client_write) = tokio::io::split(client_end);
    DenialPeer {
        client: BiStream::from_joined(client_read, client_write),
        received,
    }
}

/// Spawns the fake wire peer: writes the scripted bytes, then holds the
/// stream open (the client's drop closes it) — the hostile-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)
}

// --- the direct mode: the request-line grammar, byte-pinned ----------------

/// Contract stamp: ADR-017 §2 (the direct constructor), ADR-016 §2 (the
/// fetch request line — `git-upload-pack <repo>\0host=<host>\0\0version=2\0`,
/// POC-1-verbatim), ADR-003 (the V2 capability list follows), ADR-017
/// §5 (nothing else rode the line — no credentials, no extras);
/// core/w4-session-construction.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn direct_fetch_sends_the_pinned_request_line_and_consumes_the_v2_advertisement() {
    let (handles, record, holder) = fixture_fresh("u-7731", AccessAction::Read).await;
    let (client, received, _sent, _joiner) = serve_twin(
        &handles,
        record.clone(),
        &holder,
        AccessAction::Read,
        Service::UploadPack,
        Limits::default(),
    );

    let mut session = GitSession::connect_direct(
        client,
        "owner/doubled",
        Service::UploadPack,
        Some("example.net"),
        Limits::default(),
    )
    .expect("the direct construction");
    let response = tokio::time::timeout(Duration::from_secs(5), session.handshake())
        .await
        .expect("the handshake is timely")
        .expect("the fetch handshake parses the advertisement");

    assert_eq!(
        response.protocol,
        Protocol::V2,
        "the fetch session negotiated V2"
    );
    assert_eq!(
        capability_tokens(&response.capabilities),
        pinned_v2_caps(),
        "the V2 capability list arrives as the handshake state"
    );
    assert!(
        response.refs.is_empty(),
        "a V2 handshake carries no ref lines"
    );
    assert_eq!(session.service(), Service::UploadPack);
    assert_eq!(session.repo(), "owner/doubled");
    assert_eq!(
        session.limits(),
        &Limits::default(),
        "the session carries the Limits it was constructed with"
    );

    assert_eq!(
        *received.lock().expect("spy record"),
        pkt(b"git-upload-pack owner/doubled\0host=example.net\0\0version=2\0"),
        "the direct mode's wire bytes are the ADR-016 §2 request-line grammar"
    );

    // The handshake runs once per session: a second call re-reports the
    // cached response without wire re-entry (the request line's bytes
    // do not repeat).
    let before = received.lock().expect("spy record").len();
    let again = session.handshake().await.expect("the cached handshake");
    assert!(
        again
            .capabilities
            .iter()
            .map(|c| c.name())
            .eq(response.capabilities.iter().map(|c| c.name())),
        "the cached response carries the same capabilities state"
    );
    assert_eq!(
        received.lock().expect("spy record").len(),
        before,
        "no wire re-entry behind the cached handshake"
    );
}

/// Contract stamp: ADR-017 §2 (the direct constructor), ADR-016 §2 (the
/// push request line — `git-receive-pack <repo>\0host=<host>\0`, no
/// version token — push-captures git:// framing), ADR-013 §1–2 (the V0
/// advertisement the handshake consumes: the caps ride the first
/// line's NUL suffix, the empty repo's sentinel drains);
/// core/w4-session-construction.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn direct_push_sends_the_pinned_request_line_and_drains_the_v0_advertisement() {
    let (handles, record, holder) = fixture_fresh("u-write", AccessAction::Write).await;
    let (client, received, _sent, _joiner) = serve_twin(
        &handles,
        record.clone(),
        &holder,
        AccessAction::Write,
        Service::ReceivePack,
        Limits::default(),
    );

    let mut session = GitSession::connect_direct(
        client,
        "owner/doubled",
        Service::ReceivePack,
        Some("example.net"),
        Limits::default(),
    )
    .expect("the direct construction");
    let response = tokio::time::timeout(Duration::from_secs(5), session.handshake())
        .await
        .expect("the handshake is timely")
        .expect("the push handshake consumes the V0 advertisement");

    // gix's version detection reads a V0-framed, caps-NUL-attached
    // advertisement as its V1 shape — the normalization the push
    // machine (core/w4-session-push) composes against; the session's
    // own version refusal only fires on a V2 reply.
    assert_eq!(
        response.protocol,
        Protocol::V1,
        "the V0 advertisement parses as gix's V1 shape (caps NUL-attached)"
    );
    let agent = response
        .capabilities
        .capability("agent")
        .expect("the agent capability rides the first ref line");
    assert!(
        String::from_utf8_lossy(agent.value().expect("the agent value")).starts_with("alkgit/"),
        "the V0 advertisement's caps parse into the capabilities state"
    );
    let mut expected_caps: Vec<String> = [
        "report-status",
        "report-status-v2",
        "delete-refs",
        "side-band-64k",
        "atomic",
        "ofs-delta",
    ]
    .iter()
    .map(|s| s.to_string())
    .collect();
    expected_caps.push("object-format=sha1".to_owned());
    #[cfg(feature = "sha256")]
    expected_caps.push("object-format=sha256".to_owned());
    expected_caps.push(concat!("agent=alkgit/", env!("CARGO_PKG_VERSION")).to_owned());
    assert_eq!(
        capability_tokens(&response.capabilities),
        expected_caps,
        "the served capability set, feature riders included"
    );
    assert_eq!(
        response.refs,
        vec![bstr::BString::from(format!(
            "{} capabilities^{{}}\n",
            "0".repeat(40)
        ))],
        "the empty repo's sentinel is the drained advertisement (its caps ride capabilities)"
    );
    assert_eq!(
        *received.lock().expect("spy record"),
        pkt(b"git-receive-pack owner/doubled\0host=example.net\0"),
        "the direct push's request line carries no version token (ADR-016 §2's push grammar)"
    );
}

// --- the channels mode: zero preamble bytes --------------------------------

/// Contract stamp: ADR-017 §2 as amended by ADR-023 §2 (the channels
/// mode — the transport writes nothing), ADR-016 §1 (the preamble rode
/// the open-op's params), ADR-023 §2 (the server's first bytes are the
/// handshake response); core/w4-session-construction.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn channels_mode_fetch_writes_zero_preamble_bytes_and_consumes_the_advertisement() {
    let (handles, record, holder) = fixture_fresh("u-7731", AccessAction::Read).await;
    let (client, received, _sent, _joiner) = serve_twin(
        &handles,
        record.clone(),
        &holder,
        AccessAction::Read,
        Service::UploadPack,
        Limits::default(),
    );

    let mut session = GitSession::open_via_channels(
        client,
        "owner/doubled",
        Service::UploadPack,
        Limits::default(),
    )
    .expect("the channels-mode construction");
    let response = tokio::time::timeout(Duration::from_secs(5), session.handshake())
        .await
        .expect("the handshake is timely")
        .expect("the server's first bytes consumed as the handshake response");
    assert_eq!(response.protocol, Protocol::V2);
    assert_eq!(
        capability_tokens(&response.capabilities),
        pinned_v2_caps(),
        "the V2 advertisement consumed"
    );
    assert!(
        received.lock().expect("spy record").is_empty(),
        "the channels mode wrote zero preamble bytes (the server heard nothing)"
    );
}

/// The push twin of the zero-preamble row: the push session's handshake
/// consumes the V0 advertisement without a single client byte.
///
/// Contract stamp: ADR-023 §2 (the Process handshake), ADR-013 §2 (the
/// V0 advertisement, server speaks first); core/w4-session-construction.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn channels_mode_push_writes_zero_preamble_bytes_and_drains_the_v0_advertisement() {
    let (handles, record, holder) = fixture_fresh("u-write", AccessAction::Write).await;
    let (client, received, _sent, _joiner) = serve_twin(
        &handles,
        record.clone(),
        &holder,
        AccessAction::Write,
        Service::ReceivePack,
        Limits::default(),
    );

    let mut session = GitSession::open_via_channels(
        client,
        "owner/doubled",
        Service::ReceivePack,
        Limits::default(),
    )
    .expect("the channels-mode construction");
    let response = tokio::time::timeout(Duration::from_secs(5), session.handshake())
        .await
        .expect("the handshake is timely")
        .expect("the pushed advertisement consumed");
    assert_eq!(response.protocol, Protocol::V1);
    assert_eq!(response.refs.len(), 1, "the drained sentinel");
    assert!(
        received.lock().expect("spy record").is_empty(),
        "zero preamble bytes"
    );
}

/// The seeded repo's advertisement drains ref lines (the push machine's
/// parse input), one pkt-line payload per element, NUL-stripped on the
/// first line (the caps ride `capabilities`).
///
/// Contract stamp: ADR-017 §2 (the handshake's drain), ADR-013 §2 (the
/// ref lines over the duplex session); core/w4-session-construction.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_non_empty_push_advertisement_drains_its_ref_lines() {
    let (handles, record, holder) = fixture_fresh("u-write", AccessAction::Write).await;
    let refs = Arc::clone(&handles.1);
    let seeded = gix_hash::ObjectId::from([7u8; 20]);
    refs.apply_updates(
        &record,
        vec![RefUpdate {
            name: "refs/heads/main".to_owned(),
            expected: None,
            new: Some(seeded),
        }],
        false,
    )
    .await
    .expect("the seed");
    let (client, _received, _sent, _joiner) = serve_twin(
        &handles,
        record.clone(),
        &holder,
        AccessAction::Write,
        Service::ReceivePack,
        Limits::default(),
    );

    let mut session = GitSession::open_via_channels(
        client,
        "owner/doubled",
        Service::ReceivePack,
        Limits::default(),
    )
    .expect("the channels-mode construction");
    let response = tokio::time::timeout(Duration::from_secs(5), session.handshake())
        .await
        .expect("the handshake is timely")
        .expect("the push handshake drains the advertisement");
    assert_eq!(
        response.refs.len(),
        1,
        "one ref line (the caps line moved to capabilities)"
    );
    assert!(
        response.refs[0].starts_with(format!("{seeded} refs/heads/main").as_bytes())
            || response.refs[0].starts_with("capabilities^{}".as_bytes()),
        "the ref line rides as the wire's payload shape: {:?}",
        response.refs[0]
    );
    assert_eq!(
        response.refs[0].trim_end(),
        format!("{seeded} refs/heads/main").as_bytes(),
        "the drained line is the wire's ref row, caps NUL-attachment moved to capabilities"
    );
}

/// The channels dial contract end-to-end: the caller (the dialing
/// decision's documented contract) opens `channels/git/sub` with the
/// `{repo, service}` params through alkcall's client path and hands the
/// opened stream to the session — the full open-op path (establisher
/// gate → handler → serving entry) serves the session's handshake on
/// both services.
///
/// Contract stamp: ADR-016 §1 (the open-op params the caller sent —
/// the schema the gate fail-closes), ADR-016 §4 (open_via_channels'
/// shape — the preamble rode the open op), alkcall ADR-047/ADR-049
/// (the open op + its gate); core/w4-session-construction.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn the_channels_dial_serves_the_session_on_both_services() {
    // fetch: the open gate authorizes read; the handshake consumes the
    // V2 advertisement.
    let (handles, _record, holder) = fixture_fresh("u-7731", AccessAction::Read).await;
    let client = git_channels_client(&handles, Some(holder.clone())).await;
    let (_channel_id, send, recv) = tokio::time::timeout(
        Duration::from_secs(5),
        client.open_channel(
            OP_GIT_OPEN,
            json!({ "repo": "owner/doubled", "service": "git-upload-pack" }),
            GIT_ALPN,
        ),
    )
    .await
    .expect("the open op timed out")
    .expect("the open op succeeds");
    let mut session = GitSession::open_via_channels(
        BiStream::from_joined(recv, send),
        "owner/doubled",
        Service::UploadPack,
        Limits::default(),
    )
    .expect("the channels-mode construction");
    let response = tokio::time::timeout(Duration::from_secs(5), session.handshake())
        .await
        .expect("the handshake is timely")
        .expect("the open-op path serves the session");
    assert_eq!(response.protocol, Protocol::V2);
    assert_eq!(capability_tokens(&response.capabilities), pinned_v2_caps());
    drop(session);

    // push: the write tier opens (write grant); the handshake drains
    // the V0 advertisement.
    let (handles, _record, holder) = fixture_fresh("u-write", AccessAction::Write).await;
    let client = git_channels_client(&handles, Some(holder.clone())).await;
    let (_channel_id, send, recv) = tokio::time::timeout(
        Duration::from_secs(5),
        client.open_channel(
            OP_GIT_OPEN,
            json!({ "repo": "owner/doubled", "service": "git-receive-pack" }),
            GIT_ALPN,
        ),
    )
    .await
    .expect("the open op timed out")
    .expect("the open op succeeds");
    let mut session = GitSession::open_via_channels(
        BiStream::from_joined(recv, send),
        "owner/doubled",
        Service::ReceivePack,
        Limits::default(),
    )
    .expect("the channels-mode construction");
    let response = tokio::time::timeout(Duration::from_secs(5), session.handshake())
        .await
        .expect("the handshake is timely")
        .expect("the drained advertisement");
    assert_eq!(response.protocol, Protocol::V1);
    assert_eq!(response.refs.len(), 1, "the drained sentinel");
}

struct NoopIdProvider;
impl alkcall::core::auth::IdentityProvider for NoopIdProvider {
    fn resolve_from_fingerprint(&self, _: &str) -> Option<Identity> {
        None
    }
    fn resolve_from_token(&self, _: &alkcall::core::auth::AuthToken) -> Option<Identity> {
        None
    }
}

/// Wires a `ChannelClient` against a server whose per-connection
/// registry carries `channels/git/sub` — registered through alkgit's
/// `register_openable` (the in-crate open op) over the git doubles —
/// the alkcall open-op harness pattern (native_transport's).
async fn git_channels_client(handles: &Handles, opener: Option<Identity>) -> ChannelClient {
    let registry: Arc<dyn GitRegistry> = handles.0.clone();
    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 identity_for_hook = opener.clone();

    let install_hook: InstallChannelZero = Arc::new(move |manager, channel0_conn, auth| {
        let registry = Arc::clone(&registry);
        let refs = Arc::clone(&refs);
        let packgen = Arc::clone(&packgen);
        let ingest = Arc::clone(&ingest);
        let identity_for_conn = identity_for_hook.clone();
        tokio::spawn(async move {
            let channel0_bidi = match channel0_conn.accept_bi().await {
                Ok(s) => s,
                Err(_) => return,
            };
            let (writer, reader) = split_single_stream(channel0_bidi);
            if let Some(id) = identity_for_conn {
                let _ = channel0_conn.set_identity(id);
            }
            let core = ChannelCore::new(manager, default_policy());
            let mut registry_ops = OperationRegistry::new();
            register_openable(
                &core,
                registry,
                refs,
                packgen,
                ingest,
                false,
                Limits::default(),
                Arc::new(Semaphore::new(
                    Limits::default().max_concurrent_blocking_tasks,
                )),
                &mut registry_ops,
                auth.clone(),
            )
            .expect("register_openable");
            let call_connection = Arc::new(CallConnection::new_single_stream(
                channel0_conn,
                Arc::clone(&writer),
            ));
            let dp = Dispatcher::new(Arc::new(registry_ops), Arc::new(NoopIdProvider));
            dp.run_loop_single_stream(call_connection, reader, writer)
                .await;
        })
    });

    let (client_end, server_end) = duplex(64 * 1024);
    let client_conn = Connection::from_bidi(client_end, b"alk/channels".to_vec(), None);
    let server_conn = Connection::from_bidi(server_end, b"alk/channels".to_vec(), None);
    if let Some(id) = opener.clone() {
        let _ = server_conn.set_identity(id);
    }

    let adapter = ChannelsAdapter::new(install_hook, Arc::new(NoCap));
    let auth = AuthContext::anonymous(b"alk/channels");
    tokio::spawn(async move {
        let _ = ProtocolHandler::handle(&adapter, server_conn, &auth).await;
    });

    ChannelClient::from_connection(client_conn)
        .await
        .expect("channel client init")
}

// --- version pins, budgets, invalid inputs ----------------------------------

/// The version pins fail closed, both directions: a peer whose fetch
/// reply negotiates a V0/V1-shaped advertisement (the ADR-003
/// downgrade class) and a peer whose push reply negotiates V2 (ADR-013
/// §1's unconditional framing) each end the session with the typed
/// refusal — never a mis-framed machine.
///
/// Contract stamp: ADR-003 (fetch demands V2 — the client side of the
/// decline), ADR-013 §1 (push is V0-framed unconditionally),
/// ADR-017 §6 (fail-closed); core/w4-session-construction.
#[tokio::test]
async fn the_service_version_pins_refuse_off_dimension_replies_typed() {
    // A fake peer answering a FETCH with a V0-shaped advertisement
    // (caps riding the first line's NUL suffix — gix reads V1).
    let script = vec![
        pkt(b"0500000000000000000000000000000000000000 refs/heads/x\0side-band-64k atomic"),
        b"0000".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");
    let refused = session
        .handshake()
        .await
        .expect_err("a V0 fetch reply is off-dimension");
    match &refused {
        alkgit::SessionError::VersionRefused {
            service,
            negotiated,
        } => {
            assert_eq!(*service, Service::UploadPack);
            assert_eq!(*negotiated, Protocol::V1);
        }
        other => panic!("expected the typed version refusal, got {other:?}"),
    }
    assert_eq!(
        refused,
        session.handshake().await.expect_err("fail closed"),
        "the refusal is terminal"
    );

    // A fake peer answering a PUSH with a V2 capability advertisement
    // (the impossible direction for git — refused before any machine).
    let script = vec![
        pkt(b"version 2\n"),
        pkt(b"ls-refs=unborn\n"),
        b"0000".to_vec(),
    ];
    let client = spawn_fake_peer(script).await;
    let mut session = GitSession::open_via_channels(
        client,
        "owner/doubled",
        Service::ReceivePack,
        Limits::default(),
    )
    .expect("the construction");
    match session
        .handshake()
        .await
        .expect_err("a V2 push reply is off-dimension")
    {
        alkgit::SessionError::VersionRefused {
            service,
            negotiated,
        } => {
            assert_eq!(service, Service::ReceivePack);
            assert_eq!(negotiated, Protocol::V2);
        }
        other => panic!("expected the typed version refusal, got {other:?}"),
    }
}

/// Contract stamp: backend.md §repo-id grammar (the `invalid` class —
/// the caller's correction, nothing on the wire); ADR-016 §2 (the
/// request line's framing bytes make a NUL/LF host unfittable);
/// core/w4-session-construction.
#[tokio::test]
async fn local_input_grammar_failures_surfaces_before_anything_touches_the_wire() {
    let (client_end, _server_end) = duplex(1 << 16);
    let (client_read, client_write) = tokio::io::split(client_end);
    let refused = GitSession::connect_direct(
        BiStream::from_joined(client_read, client_write),
        "UPPER/x",
        Service::UploadPack,
        None,
        Limits::default(),
    );
    match refused {
        Err(alkgit::SessionError::InvalidInput(message)) => {
            assert!(
                message.contains("UPPER/x"),
                "the correction names the id: {message}"
            );
        }
        _ => panic!("a grammar-violating id refuses locally"),
    }

    let (client_end, _server_end) = duplex(1 << 16);
    let (client_read, client_write) = tokio::io::split(client_end);
    let refused = GitSession::connect_direct(
        BiStream::from_joined(client_read, client_write),
        "owner/doubled",
        Service::UploadPack,
        Some("evil\0host"),
        Limits::default(),
    );
    match refused {
        Err(alkgit::SessionError::InvalidInput(message)) => {
            assert!(
                message.contains("framing bytes"),
                "the NUL/LF refusal names the rule: {message}"
            );
        }
        _ => panic!("a host carrying framing bytes refuses locally"),
    }
}

/// Review 004 R-1's disposition pinned mechanically: the dead-until-now
/// `gix-protocol`/`gix-transport` declarations are live code whose pins
/// match the research doc's table (gitoxide.md, version alignment:
/// 0.67 / 0.61, `async-client` on both) and the crate's MSRV floor.
///
/// Contract stamp: gitoxide.md (the pin confirmation basis), review 004
/// R-1 (the first-commit disposition); core/w4-session-construction.
#[test]
fn the_manifest_pins_match_the_research_doc_table() {
    let manifest = include_str!("../Cargo.toml");
    for pin in [
        "gix-protocol = { version = \"0.67\", features = [\"async-client\"] }",
        "gix-transport = { version = \"0.61\", features = [\"async-client\"] }",
    ] {
        assert!(
            manifest.contains(pin),
            "the pin drifted from the confirmed research doc: {pin:?}"
        );
    }
    assert!(
        manifest.contains("rust-version = \"1.88\""),
        "the manifest MSRV floor matches the research doc's alignment fact"
    );
}

// --- peer refusals, budgets, version pins ----------------------------------

/// Contract stamp: ADR-008 (the peer's collapsed denial arrives at the
/// client verbatim — unknown ≡ unauthorized, one generic message),
/// ADR-017 §6 (fail-closed: the session is terminal after the breach
/// and re-reports without re-entering); core/w4-session-construction.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_denied_session_surfaces_the_collapse_and_terminates_fail_closed() {
    let (handles, _record, _granted_holder) =
        fixture_fresh("someone-else", AccessAction::Read).await;
    let peer = denial_peer(&handles, Some(identity("u-striker"))).await;

    let mut session = GitSession::connect_direct(
        peer.client,
        "owner/doubled",
        Service::UploadPack,
        None,
        Limits::default(),
    )
    .expect("the direct construction");
    let error = tokio::time::timeout(Duration::from_secs(5), session.handshake())
        .await
        .expect("the refusal responds promptly")
        .expect_err("the peer denied the session");

    match &error {
        alkgit::SessionError::PeerRefused(message) => {
            assert_eq!(
                message, "access denied: owner/doubled",
                "the ADR-008 collapse shape"
            );
        }
        other => panic!("expected the peer's refusal, got {other:?}"),
    }

    let wire = peer.received.lock().expect("spy record").clone();
    let again = session
        .handshake()
        .await
        .expect_err("the session is terminal");
    assert_eq!(
        error, again,
        "the error-break rule re-reports the cached refusal"
    );
    assert_eq!(
        *peer.received.lock().expect("spy record"),
        wire,
        "the re-report never re-entered the transport"
    );
}

/// Contract stamp: ADR-017 §6 / ADR-009 (the wall clock bounds every
/// operation — the handshake included; the breach is typed, fail-closed,
/// and re-reported); core/w4-session-construction.
#[tokio::test]
async fn the_wall_clock_bounds_the_handshake_and_the_breach_re_reports() {
    let (client_end, _server_end) = duplex(1 << 16);
    let (client_read, client_write) = tokio::io::split(client_end);
    let limits = Limits {
        session_wall_clock: Duration::from_millis(150),
        ..Limits::default()
    };
    let mut session = GitSession::connect_direct(
        BiStream::from_joined(client_read, client_write),
        "owner/doubled",
        Service::UploadPack,
        None,
        limits,
    )
    .expect("the construction (no server attached)");

    let start = tokio::time::Instant::now();
    let error = tokio::time::timeout(Duration::from_secs(5), session.handshake())
        .await
        .expect("the handshake returns promptly at the breach")
        .expect_err("the wall clock breached");
    assert!(
        start.elapsed() < Duration::from_secs(2),
        "the breach answers at the budget's scale"
    );

    match &error {
        alkgit::SessionError::WallClock { budget } => {
            assert_eq!(*budget, Duration::from_millis(150));
        }
        other => panic!("expected the wall-clock breach, got {other:?}"),
    }
    let again = session
        .handshake()
        .await
        .expect_err("the session is terminal");
    assert_eq!(error, again, "the breach re-reports without re-entry");
}

const FLOOD_FIRST_LINE: &[u8] =
    b"0000000000000000000000000000000000000000 capabilities^{}\0side-band-64k atomic";

/// Contract stamp: ADR-017 §6 / ADR-009 (the advertisement bounds —
/// the same budget kinds client-direction, no new fields; the
/// response-pack budget `max_pack_size` is the machines');
/// core/w4-session-construction.
#[tokio::test]
async fn the_advertisement_drain_bounds_fail_closed_typed() {
    for (budget_name, limit, why) in [
        (
            "max_advertisement_refs",
            3u64,
            "a line count over the refs budget",
        ),
        (
            "max_request_body",
            10,
            "a byte total over the request-body budget",
        ),
    ] {
        let mut script = vec![pkt(FLOOD_FIRST_LINE)];
        for i in 0..5 {
            script.push(pkt(format!(
                "0000000000000000000000000000000000000000 refs/heads/ref{i}"
            )
            .as_bytes()));
        }
        script.push(b"0000".to_vec());
        let client = spawn_fake_peer(script).await;
        let limits = match budget_name {
            "max_advertisement_refs" => Limits {
                max_advertisement_refs: 3,
                ..Limits::default()
            },
            _ => Limits {
                max_request_body: 10,
                ..Limits::default()
            },
        };
        let mut session =
            GitSession::open_via_channels(client, "owner/doubled", Service::ReceivePack, limits)
                .expect("the construction");
        let error = session
            .handshake()
            .await
            .expect_err("the drain refused the flood");

        match &error {
            alkgit::SessionError::BudgetExceeded {
                what,
                budget_name: breached,
                limit: breached_limit,
            } => {
                assert!(what.contains("advertisement"), "{why}: {what}");
                assert_eq!(*breached, budget_name, "{why}: the breached kind is named");
                assert_eq!(
                    *breached_limit, limit,
                    "{why}: the carried limit is the breaching one"
                );
            }
            other => panic!("{why}: expected the typed budget breach, got {other:?}"),
        }
        let again = session.handshake().await.expect_err("fail closed");
        assert_eq!(error, again, "the breach is terminal (error-break)");
    }
}