alkgit 0.1.0

//! The wave-2 native transport (core/w2-native-transport): the
//! direct-ALPN `GitAdapter` over the ADR-016 §2 request-line preamble,
//! the resolve→authorize gate's collapsed denial, the V0/V1 fetch
//! decline, and the `channels/git/sub` open op + establisher — wire
//! rows over in-memory connections with the wave-2 backend doubles.
//!
//! Rows are names + `Contract stamp:` versions, per
//! `tests/common` §The row/stamp discipline.

mod common;

use std::collections::{HashMap, VecDeque};
use std::sync::Arc;
use std::time::Duration;

use alkcall::channels::adapter::{ChannelsAdapter, InstallChannelZero};
use alkcall::channels::client::{ChannelClient, ChannelOpenError};
use alkcall::channels::operations::ChannelCore;
use alkcall::channels::policy::{default_policy, NoCap};
use alkcall::core::auth::{AuthContext, Identity, IdentityProvider};
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 serde_json::json;
use tokio::io::{duplex, AsyncReadExt, AsyncWriteExt};
use tokio::sync::Semaphore;

use alkgit::{
    register_openable, AccessAction, GitAdapter, GitPackGen, GitPackIngest, GitRefs, GitRegistry,
    Limits, RepoRecord, GIT_ALPN, OP_GIT_OPEN,
};
use common::{MemPackGen, MemPackIngest, MemRefs, MemRegistry};

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

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

/// The adapter drives a serving session against these — the shared
/// handles the accept loop clones per row (the ingest handle rides the
/// wave-3 push threading).
fn handles(
    registry: MemRegistry,
    refs: Arc<MemRefs>,
    packgen: Arc<MemPackGen>,
    ingest: Arc<MemPackIngest>,
) -> SessionHandlesTuple {
    (Arc::new(registry), refs, packgen, ingest)
}

/// The doubles' defaults as owned handles.
type SessionHandlesTuple = (
    Arc<dyn GitRegistry>,
    Arc<dyn GitRefs>,
    Arc<dyn GitPackGen>,
    Arc<dyn GitPackIngest>,
);

#[allow(clippy::type_complexity)]
fn default_handles() -> SessionHandlesTuple {
    let objects = common::MemObjects::default();
    handles(
        MemRegistry::default(),
        Arc::new(MemRefs::default()),
        Arc::new(MemPackGen {
            objects: objects.clone(),
        }),
        Arc::new(MemPackIngest { objects }),
    )
}

/// Encodes the ADR-016 §2 request-line preamble, client framing: one
/// pkt-line — `<service> <repo>\0<internal extras>\0`.
fn request_line(service: &str, repo_id: &str, extras: &[&str]) -> Vec<u8> {
    let mut payload = format!("{service} {repo_id}");
    for extra in extras {
        payload.push('\0');
        payload.push_str(extra);
    }
    let total = payload.len() + 4;
    let mut out = format!("{total:04x}").into_bytes();
    out.extend_from_slice(payload.as_bytes());
    out
}

fn fetch_line(repo_id: &str) -> Vec<u8> {
    request_line(
        "git-upload-pack",
        repo_id,
        &["host=example.com", "", "version=2"],
    )
}

fn push_line(repo_id: &str) -> Vec<u8> {
    request_line("git-receive-pack", repo_id, &["host=example.com"])
}

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
}

fn command_request(command: &[u8], pre: &[&str], args: &[&str]) -> Vec<u8> {
    let mut out = pkt(command);
    for line in pre {
        out.extend(pkt(line.as_bytes()));
    }
    out.extend_from_slice(b"0001");
    for line in args {
        out.extend(pkt(line.as_bytes()));
    }
    out.extend_from_slice(b"0000");
    out
}

/// Reads pkt-lines from the client end until (and including) a flush —
/// one response block.
async fn read_pkts(reader: &mut tokio::io::DuplexStream) -> Vec<String> {
    let mut lines = 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 {
            lines.push(String::new());
            break;
        }
        let mut payload = vec![0u8; len - 4];
        reader
            .read_exact(&mut payload)
            .await
            .expect("pkt-line payload");
        lines.push(
            String::from_utf8_lossy(&payload)
                .trim_end_matches('\n')
                .to_owned(),
        );
    }
    lines
}

/// The advertisement's expected line sequence (feature-aware — the
/// sha256 rider rides the feature per OQ-05's passthrough policy).
fn advertisement_block() -> 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",
    ];
    let mut lines: Vec<String> = lines.iter().map(|l| l.to_string()).collect();
    lines.push(String::new());
    lines
}

/// The `accept_bi` queue source: yields each server-side stream once,
/// then reports the connection closed (the adapter's accept loop is the
/// unit under test).
struct YieldQueued {
    streams: tokio::sync::Mutex<VecDeque<BiStream>>,
}

#[async_trait::async_trait]
impl alkcall::core::BidiStreamSource for YieldQueued {
    async fn accept_bi(&self) -> Result<BiStream, alkcall::core::StreamError> {
        self.streams
            .lock()
            .await
            .pop_front()
            .ok_or(alkcall::core::StreamError::ConnectionClosed)
    }
    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) {}
}

/// Joins the server end of a fresh duplex pair into a `BiStream` — an
/// `accept_bi` yield — and returns the client side.
fn wire_pair() -> (tokio::io::DuplexStream, BiStream) {
    let (client, server) = duplex(1 << 16);
    let (server_read, server_write) = tokio::io::split(server);
    (client, BiStream::from_joined(server_read, server_write))
}

/// Runs the adapter's `handle` over a connection whose `accept_bi`
/// yields the given server-side streams. One stream becomes one session.
async fn run_adapter(
    (registry, refs, packgen, ingest): SessionHandlesTuple,
    server_streams: Vec<BiStream>,
    identity: Option<Identity>,
) {
    let adapter = GitAdapter::new(
        registry,
        refs,
        packgen,
        ingest,
        false,
        Limits::default(),
        Arc::new(Semaphore::new(
            Limits::default().max_concurrent_blocking_tasks,
        )),
    );
    let connection = Connection::from_source(
        YieldQueued {
            streams: tokio::sync::Mutex::new(server_streams.into()),
        },
        b"alk/git".to_vec(),
    );
    let mut auth = AuthContext::anonymous(b"alk/git");
    auth.identity = identity;
    let _ = adapter.handle(connection, &auth).await;
}

/// Starts one adapter session and returns (client side, accept-loop
/// joiner) — the caller writes the preamble and reads the wire.
async fn start_session(
    session_handles: SessionHandlesTuple,
    identity: Option<Identity>,
) -> (tokio::io::DuplexStream, tokio::task::JoinHandle<()>) {
    let (client, stream) = wire_pair();
    let joiner = tokio::spawn({
        let stream = stream;
        async move { run_adapter(session_handles, vec![stream], identity).await }
    });
    (client, joiner)
}

/// Runs one adapter session to its close from the client's view: sends
/// the preamble, drains the session's whole output to EOF (the session
/// closes after every pre-session decline), and joins the accept loop.
async fn preamble_session(
    session_handles: SessionHandlesTuple,
    preamble: &[u8],
    identity: Option<Identity>,
) -> Vec<u8> {
    let (mut client, joiner) = start_session(session_handles, identity).await;
    client
        .write_all(preamble)
        .await
        .expect("the preamble rides out");
    client
        .shutdown()
        .await
        .expect("the client signals end-of-preamble");
    let wire = drain_to_eof(client).await;
    joiner.await.expect("the accept loop joins");
    wire
}

/// Drains the client side to EOF — the session's whole wire output.
async fn drain_to_eof(mut client: tokio::io::DuplexStream) -> Vec<u8> {
    let mut wire = Vec::new();
    let mut chunk = [0u8; 512];
    loop {
        let read = tokio::time::timeout(Duration::from_secs(5), client.read(&mut chunk)).await;
        let Ok(read) = read else {
            panic!("the session never closed the client side");
        };
        let n = read.expect("the client read succeeds");
        if n == 0 {
            return wire;
        }
        wire.extend_from_slice(&chunk[..n]);
    }
}

/// Parses a raw wire byte sequence into pkt-line payloads (the
/// decline/denial capture 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
}

/// The direct-ALPN fixture: `owner/doubled` is a private repo with one
/// granted holder per row; the handles are the doubles' (the packgen +
/// ingest pair share one object namespace).
struct Fixture {
    registry: MemRegistry,
    refs: Arc<MemRefs>,
    packgen: Arc<MemPackGen>,
    ingest: Arc<MemPackIngest>,
}

impl Fixture {
    fn fresh() -> Fixture {
        let objects = common::MemObjects::default();
        Fixture {
            registry: MemRegistry::default(),
            refs: Arc::new(MemRefs::default()),
            packgen: Arc::new(MemPackGen {
                objects: objects.clone(),
            }),
            ingest: Arc::new(MemPackIngest { objects }),
        }
    }

    fn session_tuple(&self) -> SessionHandlesTuple {
        handles(
            self.registry.clone(),
            Arc::clone(&self.refs),
            Arc::clone(&self.packgen),
            Arc::clone(&self.ingest),
        )
    }

    async fn put_private(&self, id: &str, holder: &str, grant: AccessAction) {
        use alkgit::GitRegistryStore;
        let mut record = repo(id);
        record.grants.insert(holder.to_string(), vec![grant]);
        self.registry.put_repo(&record).await.unwrap();
    }

    /// Runs one preamble-to-close session against this fixture's
    /// assembled handles.
    async fn preamble_capture(&self, preamble: &[u8], identity: Option<Identity>) -> Vec<u8> {
        preamble_session(self.session_tuple(), preamble, identity).await
    }
}

/// Each accepted `BiStream` becomes a duplex fetch session: the
/// preamble gates, the advertisement is the session's first byte, and
/// `ls-refs` dispatches — the adapter's full happy path.
///
/// Contract stamp: ADR-010 §producer (the direct-ALPN adapter shape);
/// ADR-016 §2 (the request line selects the service at establishment);
/// ADR-007 steps 2–3 (resolve→authorize precedes any protocol byte);
/// ADR-003 (advertisement-once at session start);
/// core/w2-native-transport.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn the_git_adapter_turns_each_accepted_bi_stream_into_a_duplex_fetch_session() {
    let fixture = Fixture::fresh();
    fixture
        .put_private("owner/doubled", "u-7731", AccessAction::Read)
        .await;

    let (mut client, joiner) =
        start_session(fixture.session_tuple(), Some(identity("u-7731"))).await;
    client
        .write_all(&fetch_line("owner/doubled"))
        .await
        .expect("the preamble rides out");
    let advertisement = tokio::time::timeout(Duration::from_secs(5), read_pkts(&mut client))
        .await
        .expect("the advertisement follows the gating preamble");
    assert_eq!(advertisement, advertisement_block());

    client
        .write_all(&command_request(
            b"command=ls-refs\n",
            &["agent=git/2.43"],
            &["ref-prefix refs/heads/"],
        ))
        .await
        .expect("the command rides out");
    let reply = tokio::time::timeout(Duration::from_secs(5), read_pkts(&mut client))
        .await
        .expect("ls-refs dispatches");
    assert_eq!(reply.last(), Some(&String::new()), "flush-terminated reply");

    drop(client);
    joiner.await.expect("the accept loop joins");
}

/// Contract stamp: ADR-016 §2 (`host=` and `version=2` extras are
/// parsed and accepted verbatim — the captured fetch shape, and a
/// further unknown extra, still gate through to the advertisement);
/// core/w2-native-transport.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn the_fetch_preamble_accepted_extras_do_not_disturb_the_gate() {
    let fixture = Fixture::fresh();
    fixture
        .put_private("owner/doubled", "u-7731", AccessAction::Read)
        .await;

    let (mut client, joiner) = fixture
        .capture_open_session(
            &request_line(
                "git-upload-pack",
                "owner/doubled",
                &["host=example.com", "", "version=2", "extra=whatever"],
            ),
            Some(identity("u-7731")),
        )
        .await;
    let advertisement = tokio::time::timeout(Duration::from_secs(5), read_pkts(&mut client))
        .await
        .expect("the advertisement follows the preamble");
    assert_eq!(
        advertisement.first().map(String::as_str),
        Some("version 2"),
        "the extras were accepted verbatim, session established"
    );
    drop(client);
    joiner.await.expect("the accept loop joins");
}

impl Fixture {
    /// Starts one adapter session and sends the preamble — the client
    /// end stays open (the advertisement/conversation rows).
    async fn capture_open_session(
        &self,
        preamble: &[u8],
        identity: Option<Identity>,
    ) -> (tokio::io::DuplexStream, tokio::task::JoinHandle<()>) {
        let (mut client, joiner) = start_session(self.session_tuple(), identity).await;
        client
            .write_all(preamble)
            .await
            .expect("the preamble rides out");
        (client, joiner)
    }
}

/// The preamble refusal set is the wire before any advertisement: an
/// unknown service (git-upload-archive included — not in the ADR-016
/// enum) and a malformed frame fail with a clear, repo-blind decline.
///
/// Contract stamp: ADR-016 §1 (git-upload-archive is not in the enum —
/// rejected at the gate, one layer earlier than the doors' fixed
/// refusal); ADR-016 §2 (malformed lines fail before any
/// advertisement); core/w2-native-transport.
#[tokio::test]
async fn the_preamble_refusal_set_declines_with_a_clear_error_before_any_advertisement() {
    let fixture = Fixture::fresh();
    fixture
        .put_private("owner/doubled", "u-7731", AccessAction::Read)
        .await;

    for (preamble, why) in [
        (
            request_line("git-upload-archive", "owner/doubled", &["host=x"]),
            "git-upload-archive is not in the enum",
        ),
        (
            request_line("nope", "owner/doubled", &[]),
            "an unknown service",
        ),
        (b"zzzz".to_vec(), "a malformed frame"),
    ] {
        let wire = fixture
            .preamble_capture(&preamble, Some(identity("u-7731")))
            .await;
        let payloads = pkt_payloads(&wire);
        assert_eq!(
            payloads.len(),
            1,
            "{why}: the decline is the only wire shape"
        );
        let band = String::from_utf8_lossy(&payloads[0]);
        assert!(
            band.starts_with("ERR "),
            "{why}: the decline is the band: {band}"
        );
        assert!(
            !band.contains("version "),
            "{why}: no advertisement precedes the decline: {band}"
        );
    }
}

/// Contract stamp: ADR-003 (the V0/V1 decline — fetch clients that
/// cannot speak V2 get a clear pkt-line error, no advertisement);
/// ADR-016 §2 (a fetch request line without `version=2`, or with an
/// unrecognized version token, is the same V0/V1-decline path, before
/// any advertisement); core/w2-native-transport.
#[tokio::test]
async fn the_v0_v1_fetch_decline_takes_the_clear_error_path() {
    let fixture = Fixture::fresh();
    fixture
        .put_private("owner/doubled", "u-7731", AccessAction::Read)
        .await;

    for (extras, why) in [
        (vec!["host=example.com", ""], "no version token"),
        (
            vec!["host=example.com", "version=1"],
            "an unrecognized token",
        ),
        (vec![], "no extras at all"),
    ] {
        let preamble = request_line("git-upload-pack", "owner/doubled", &extras);
        let wire = fixture
            .preamble_capture(&preamble, Some(identity("u-7731")))
            .await;
        let payloads = pkt_payloads(&wire);
        assert_eq!(payloads.len(), 1, "{why}: one decline shape");
        let band = String::from_utf8_lossy(&payloads[0]);
        assert!(band.starts_with("ERR "), "{why}: {band}");
        assert!(
            band.contains("version=2"),
            "{why}: the decline states the requirement: {band}"
        );
        assert!(
            !band.contains("version 2\n"),
            "{why}: no advertisement rode ahead: {band}"
        );
    }
}

/// Contract stamp: ADR-007 steps 2–3; ADR-008/N-2 (unknown ≡
/// unauthorized — no existence oracle: the direct-ALPN denial is
/// byte-identical and no protocol byte precedes it);
/// core/w2-native-transport.
#[tokio::test]
async fn unknown_repo_and_unauthorized_denials_are_indistinguishable_on_the_direct_path() {
    let grantless = Fixture::fresh();
    grantless
        .put_private("owner/doubled", "someone-else", AccessAction::Read)
        .await;

    let unknown = default_handles();
    let unknown_wire = preamble_session(
        unknown,
        &fetch_line("owner/doubled"),
        Some(identity("u-7731")),
    )
    .await;
    let unauthorized = grantless
        .preamble_capture(&fetch_line("owner/doubled"), Some(identity("u-7731")))
        .await;

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

/// Contract stamp: backend.md §repo-id grammar (a grammar-invalid id is
/// `unauthorized` on the serving paths — collapsed with unknown-repo,
/// the id shape leaks nothing; `invalid` is the op paths' mapping);
/// ADR-008; core/w2-native-transport.
#[tokio::test]
async fn a_grammar_invalid_repo_id_collapses_with_the_unknown_repo_denial() {
    let id = "UPPER/x";
    let invalid = preamble_session(
        default_handles(),
        &request_line("git-upload-pack", id, &["", "version=2"]),
        Some(identity("u-7731")),
    )
    .await;
    let unknown = preamble_session(
        default_handles(),
        &request_line("git-upload-pack", id, &["", "version=2"]),
        Some(identity("u-7731")),
    )
    .await;
    assert_eq!(
        invalid, unknown,
        "the grammar-invalid denial is byte-identical to the unknown-repo denial for the same id"
    );
    assert_eq!(
        pkt_payloads(&invalid),
        vec![format!("ERR access denied: {id}").into_bytes()],
        "the collapsed shape, naming only what the client sent"
    );
}

/// The push request line gates on the write tier (ADR-015 §2): an
/// authorized push session proceeds to the V0 ref advertisement — the
/// server speaks first, no service prefix, the empty repo's
/// `capabilities^{}` sentinel with the caps NUL-attached — and a
/// grantless identity collapses to the generic denial, nothing before
/// it.
///
/// Contract stamp: ADR-015 §2 (write requires identity + a write grant
/// — the push tier); ADR-016 §2 (the request line selects the action —
/// read for fetch, write for push); ADR-008 (the unauthorized push
/// collapses to the generic denial, no existence oracle); ADR-013 §1–2
/// (the V0 advertisement, server speaks first, no service prefix);
/// core/w3-duplex-serving (the authorized session serves V0 now — the
/// wave-2 decline is gone).
#[tokio::test]
async fn the_push_request_line_gates_on_the_write_tier_and_serves_the_v0_advertisement() {
    let fixture = Fixture::fresh();
    fixture
        .put_private("owner/doubled", "u-write", AccessAction::Write)
        .await;

    let (mut client, joiner) = fixture
        .capture_open_session(&push_line("owner/doubled"), Some(identity("u-write")))
        .await;
    let advertisement = tokio::time::timeout(Duration::from_secs(5), read_pkts(&mut client))
        .await
        .expect("the advertisement follows the gating preamble");
    assert_eq!(advertisement.len(), 2, "an empty repo's sentinel + flush");
    let sentinel = &advertisement[0];
    assert!(
        sentinel.contains("capabilities^{}"),
        "the empty repo's sentinel: {sentinel}"
    );
    assert!(
        sentinel.contains('\0') && sentinel.contains("agent=alkgit/"),
        "the caps ride NUL-attached: {sentinel}"
    );
    assert!(
        advertisement
            .iter()
            .all(|line| !line.contains("# service=")),
        "no service prefix on the duplex push path"
    );
    drop(client);
    joiner.await.expect("the accept loop joins");

    let grantless = fixture
        .preamble_capture(&push_line("owner/doubled"), Some(identity("grantless")))
        .await;
    assert_eq!(
        pkt_payloads(&grantless),
        vec![b"ERR access denied: owner/doubled".to_vec()],
        "a read-less identity never passes the write gate"
    );
}

/// The assembled native path serves push end to end (ADR-010's
/// producer shape, the wave-3 push threading): the preamble's
/// `git-receive-pack` request line gates on the write grant, the V0
/// advertisement follows (no service prefix), one request — the create
/// command + the pack — ingests through the adapter's ingest handle,
/// the report streams back sideband-framed, and the pushed refs +
/// objects are visible in the backend.
///
/// Contract stamp: ADR-013 §2–3, §7–8 (the V0 advertisement, one
/// request per stream, the CAS, the sideband report); ADR-020 §1 (the
/// duplex tail discipline); ADR-010 §producer (the adapter drives the
/// session); ADR-009 (the pipeline semaphore is adapter-wide);
/// core/w3-duplex-serving (the assembled native path serves push).
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn the_git_adapter_serves_a_push_session_end_to_end() {
    let fixture = Fixture::fresh();
    fixture
        .put_private("owner/doubled", "u-write", AccessAction::Write)
        .await;
    let record = repo("owner/doubled");

    let blob = common::blob_fix(b"pushed");
    let tree = common::tree_fix(&[(gix_object::tree::EntryKind::Blob, "p.txt", blob.oid)]);
    let commit = common::commit_fix(tree.oid, &[], "pushed root");
    let e1 = common::pack_entry_base(blob.kind, &blob.body, 0);
    let e2 = common::pack_entry_base(tree.kind, &tree.body, e1.bytes_in_pack());
    let e3 = common::pack_entry_base(commit.kind, &commit.body, e2.bytes_in_pack());
    let pack = common::pack_bytes(&[e1, e2, e3]);

    let (mut client, joiner) = fixture
        .capture_open_session(&push_line("owner/doubled"), Some(identity("u-write")))
        .await;

    let advertisement = tokio::time::timeout(Duration::from_secs(5), read_pkts(&mut client))
        .await
        .expect("the advertisement follows the gating preamble");
    assert_eq!(advertisement.len(), 2, "the empty repo's sentinel + flush");
    assert!(
        advertisement[0].contains("capabilities^{}"),
        "the empty repo advertises exactly the sentinel: {:?}",
        advertisement[0]
    );
    assert!(
        advertisement
            .iter()
            .all(|line| !line.contains("# service=")),
        "no service prefix"
    );

    let create = format!(
        "{} {} refs/heads/main\0report-status side-band-64k",
        "0".repeat(40),
        commit.oid
    );
    let request = [
        pkt(create.as_bytes()),
        b"0000".to_vec(),
        pack[..1].to_vec(),
        pack[1..40].to_vec(),
        pack[40..].to_vec(),
    ]
    .concat();
    client
        .write_all(&request)
        .await
        .expect("the push request rides out");

    let report_body = tokio::time::timeout(Duration::from_secs(5), read_until_flush(&mut client))
        .await
        .expect("the report follows the push");
    let report = common::sideband_report_lines(&report_body);
    assert_eq!(
        report,
        vec!["unpack ok".to_owned(), "ok refs/heads/main".to_owned()],
        "the report is unpack + per-ref"
    );

    drop(client);
    joiner.await.expect("the accept loop joins");

    let listed = tokio::time::timeout(Duration::from_secs(5), fixture.refs.list_refs(&record))
        .await
        .expect("the listing")
        .expect("the listing succeeds");
    assert_eq!(
        listed
            .iter()
            .map(|line| (line.name.clone(), line.oid))
            .collect::<Vec<_>>(),
        vec![("refs/heads/main".to_owned(), Some(commit.oid))],
        "the CAS applied the pushed update"
    );
    assert!(
        fixture.ingest.objects().contains(&record, &commit.oid),
        "the pack's objects landed via the ingest handle"
    );
}

/// Reads raw bytes from the client end until (and including) the
/// terminating flush — one report body.
async fn read_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);
    }
}

// --- the channels open op (channels/git/sub) -------------------------------

struct NoopIdProvider;
impl 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 code under test) over the git doubles —
/// the alkcall open-op harness pattern.
///
/// The opener identity rides the server connection (the dispatcher's
/// per-call context derives the establisher's identity view, CF-006).
async fn git_channels_client(fixture: &Fixture, opener: Option<Identity>) -> ChannelClient {
    let (registry, refs, packgen, ingest) = fixture.session_tuple();
    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")
}

/// The channels open op end-to-end: `channels/git/sub` with the
/// `{repo, service}` params opens a channel whose stream is a running
/// duplex fetch session — the establisher gated, the handler drives the
/// serving entry inline (the R-02 lifetime contract).
///
/// Contract stamp: ADR-016 §1 (the channels path's open-time
/// resolve→authorize; the establisher pattern — no phantom channel);
/// ADR-047 §5 (the channel adoption); core/w2-native-transport.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn the_channels_open_op_serves_a_fetch_session_on_the_opened_channel() {
    let fixture = Fixture::fresh();
    fixture
        .put_private("owner/doubled", "u-7731", AccessAction::Read)
        .await;
    let client = git_channels_client(&fixture, Some(identity("u-7731"))).await;

    let (channel_id, mut send, mut 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");
    assert!(channel_id > 0, "channel 0 is the call channel (ADR-036)");

    tokio::time::timeout(Duration::from_secs(5), async {
        let advertisement = read_channel_pkts(&mut recv).await;
        assert_eq!(advertisement, advertisement_block());

        let request = command_request(
            b"command=ls-refs\n",
            &["agent=git/2.43"],
            &["ref-prefix refs/heads/"],
        );
        send.write_all(&request)
            .await
            .expect("the command rides out");
        let reply = read_channel_pkts(&mut recv).await;
        assert_eq!(reply.last(), Some(&String::new()), "flush-terminated reply");
    })
    .await
    .expect("the fetch session runs on the channel");

    drop(send);
    drop(recv);
}

/// Reads pkt-lines from the channel stream end until (and including) a
/// flush — one response block.
///
/// On a malformed header the panic dumps the desync point and the next
/// bytes before exiting (the diagnostic that identified the upstream
/// alkcall early-arrival ordering race: chunks racing
/// `adopt_channel`'s install→drain window can reach this reader ahead
/// of earlier-parked chunks — finding recorded in
/// `core/review-wave-2`; the alkgit read itself is a plain byte
/// consumer).
async fn read_channel_pkts<R>(recv: &mut R) -> Vec<String>
where
    R: tokio::io::AsyncRead + Unpin,
{
    let mut lines = Vec::new();
    loop {
        let mut header = [0u8; 4];
        recv.read_exact(&mut header).await.expect("pkt-line header");
        let len = std::str::from_utf8(&header)
            .map_err(|e| format!("non-utf8 header {header:x?}: {e}"))
            .and_then(|h| {
                usize::from_str_radix(h, 16).map_err(|e| format!("bad hex header {header:x?}: {e}"))
            });
        let len = match len {
            Ok(l) => l,
            Err(detail) => {
                let mut tail = [0u8; 48];
                let n = recv.read(&mut tail).await.unwrap_or(0);
                panic!(
                    "read_channel_pkts desync after {} lines: {detail}; \
                     next {} bytes: {:x?} ({:?})",
                    lines.len(),
                    n,
                    &tail[..n],
                    String::from_utf8_lossy(&tail[..n]),
                );
            }
        };
        if len < 4 {
            lines.push(String::new());
            break;
        }
        let mut payload = vec![0u8; len - 4];
        recv.read_exact(&mut payload)
            .await
            .expect("pkt-line payload");
        lines.push(
            String::from_utf8_lossy(&payload)
                .trim_end_matches('\n')
                .to_owned(),
        );
    }
    lines
}

/// Contract stamp: ADR-016 §1 (git-upload-archive is not in the enum —
/// rejected at the gate, one layer earlier than the doors' fixed
/// refusal); ADR-049 §3 (the failure is the open reply — no phantom
/// channel); core/w2-native-transport.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn the_open_op_rejects_git_upload_archive_at_the_gate_without_a_phantom_channel() {
    let fixture = Fixture::fresh();
    fixture
        .put_private("owner/doubled", "u-7731", AccessAction::Read)
        .await;
    let client = git_channels_client(&fixture, Some(identity("u-7731"))).await;

    let error = tokio::time::timeout(
        Duration::from_secs(5),
        client.open_channel(
            OP_GIT_OPEN,
            json!({ "repo": "owner/doubled", "service": "git-upload-archive" }),
            GIT_ALPN,
        ),
    )
    .await
    .expect("the open op timed out");

    match error {
        Err(ChannelOpenError::CallFailed { error }) => {
            let schema_gate = error.code == "INVALID_INPUT";
            let establisher_gate = error.code == "channel:open_failed"
                && error
                    .details
                    .as_ref()
                    .map(|d| {
                        d["reason"] == "handler_error"
                            && d["message"]
                                .as_str()
                                .is_some_and(|m| m.contains("git-upload-archive"))
                    })
                    .unwrap_or(false);
            assert!(
                schema_gate || establisher_gate,
                "the gate refuses git-upload-archive (the schema enum at dispatch, \
                 the establisher's enum copy behind it), got code={} details={:?}",
                error.code,
                error.details
            );
        }
        Ok(_) => panic!("git-upload-archive must be rejected at the gate"),
        Err(other) => panic!("the open op must refuse, got {other:?}"),
    }
    assert!(
        client.manager().channel_ids().iter().all(|&id| id == 0),
        "no phantom channel survives a failed establishment"
    );
}

/// Contract stamp: ADR-008/N-2 (the collapse rule holds at the
/// establisher: one reason, one generic message — recorded
/// interpretation confirmed at the wave-2 review); ADR-049 §3 (the
/// failure is the open reply, no phantom channel);
/// core/w2-native-transport.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn the_open_op_collapses_unknown_repo_and_unauthorized_to_one_reason() {
    let holder = identity("grantless");

    let empty = Fixture::fresh();
    let unknown_client = git_channels_client(&empty, Some(holder.clone())).await;
    let unknown = tokio::time::timeout(
        Duration::from_secs(5),
        unknown_client.open_channel(
            OP_GIT_OPEN,
            json!({ "repo": "owner/doubled", "service": "git-upload-pack" }),
            GIT_ALPN,
        ),
    )
    .await
    .expect("the open op timed out");

    let granted = Fixture::fresh();
    granted
        .put_private("owner/doubled", "someone-else", AccessAction::Read)
        .await;
    let unauthorized_client = git_channels_client(&granted, Some(holder)).await;
    let unauthorized = tokio::time::timeout(
        Duration::from_secs(5),
        unauthorized_client.open_channel(
            OP_GIT_OPEN,
            json!({ "repo": "owner/doubled", "service": "git-upload-pack" }),
            GIT_ALPN,
        ),
    )
    .await
    .expect("the open op timed out");

    let unknown_error = match unknown {
        Err(ChannelOpenError::CallFailed { error }) => error,
        _ => panic!("an unknown repo denies"),
    };
    let unauthorized_error = match unauthorized {
        Err(ChannelOpenError::CallFailed { error }) => error,
        _ => panic!("an unauthorized identity denies"),
    };
    assert_eq!(unknown_error.code, "channel:open_failed");
    assert_eq!(unauthorized_error.code, "channel:open_failed");
    let unknown_details = unknown_error.details.expect("details carry the reason");
    let unauthorized_details = unauthorized_error
        .details
        .expect("details carry the reason");
    assert_eq!(
        unknown_details["reason"], unauthorized_details["reason"],
        "one reason"
    );
    assert_eq!(
        unknown_details["message"], unauthorized_details["message"],
        "one generic message — no cause distinction"
    );
    assert_eq!(
        unknown_details["message"], "access denied: owner/doubled",
        "the message carries only what the caller sent"
    );
    assert!(
        unknown_client
            .manager()
            .channel_ids()
            .iter()
            .all(|&id| id == 0)
            && unauthorized_client
                .manager()
                .channel_ids()
                .iter()
                .all(|&id| id == 0),
        "no phantom channel survives either denial"
    );
}

/// Contract stamp: ADR-011 §3 / ADR-015 §2 (the service-selected
/// action, end to end: fetch gates on read, push on write);
/// core/w2-native-transport.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn the_open_op_gates_the_service_selected_action_end_to_end() {
    let fixture = Fixture::fresh();
    fixture
        .put_private("owner/doubled", "u-read", AccessAction::Read)
        .await;
    let client = git_channels_client(&fixture, Some(identity("u-read"))).await;

    let fetch = 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");
    assert!(fetch.is_ok(), "a read grant satisfies fetch");

    let push = 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");
    match push {
        Err(ChannelOpenError::CallFailed { error }) => {
            assert_eq!(error.code, "channel:open_failed");
            let details = error.details.expect("details");
            assert_eq!(details["message"], "access denied: owner/doubled");
        }
        Ok(_) => panic!("a read grant never satisfies push"),
        Err(other) => panic!("expected CallFailed(channel:open_failed), got {other:?}"),
    }
}