modelpipe 0.5.0

//! Tests for [`super`] — the authentication edge, end to end.
//!
//! Split out via `#[path]` so `exchange.rs` stays inside the file-size
//! budget.
//!
//! Every test here runs over `tokio::io::duplex()`. There is no socket, no
//! port and no peer, which is the whole reason this module is generic over
//! its streams — the security-critical path is fully exercised before the
//! transport exists.
//!
//! The backend is hand-written and counts how often it is asked for a
//! connection. That counter is the point: a 401 proves the client was
//! refused, and only the counter proves the backend never heard about it.

use std::cell::RefCell;
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};

use tokio::io::{AsyncReadExt, AsyncWriteExt, DuplexStream, duplex};
use tokio::sync::Mutex;
use tokio::task::JoinHandle;

use super::*;
use crate::token_policy::TokenPolicy;

const TOKEN: &str = "sk-zzq-the-credential";
/// The fingerprint the listener would have derived for the peer.
const TEST_PEER: &str = "3ca82708b995";
const OK_RESPONSE: &[u8] =
    b"HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: 2\r\n\r\n{}";

/// A backend that records how often it was connected to and what arrived.
///
/// Modelled on the counting stub gglib uses for the same job: a status code
/// says the request was refused, and only a counter says the work never
/// happened. Those are different promises and this crate sells the second.
struct CountingBackend {
    connects: Arc<AtomicUsize>,
    response: Vec<u8>,
    received: Arc<Mutex<Vec<JoinHandle<Vec<u8>>>>>,
}

impl CountingBackend {
    fn new(response: &[u8]) -> Self {
        Self {
            connects: Arc::new(AtomicUsize::new(0)),
            response: response.to_vec(),
            received: Arc::new(Mutex::new(Vec::new())),
        }
    }

    fn connects(&self) -> usize {
        self.connects.load(Ordering::SeqCst)
    }

    /// Everything the backend was sent, once the exchange has finished and
    /// dropped its end.
    async fn received(&self) -> Vec<u8> {
        // The guard is released before awaiting the tasks: holding a lock
        // across an await that waits on something which might want it is how
        // a test deadlocks intermittently.
        let taken: Vec<_> = self.received.lock().await.drain(..).collect();
        let mut out = Vec::new();
        for handle in taken {
            out.extend_from_slice(&handle.await.expect("backend task"));
        }
        out
    }
}

impl Backend for CountingBackend {
    type Stream = DuplexStream;

    fn authority(&self) -> &'static str {
        "127.0.0.1:11434"
    }

    async fn connect(&self) -> std::io::Result<DuplexStream> {
        self.connects.fetch_add(1, Ordering::SeqCst);
        let (mine, mut theirs) = duplex(64 * 1024);
        let response = self.response.clone();
        let handle = tokio::spawn(async move {
            // Written before reading: the duplex is buffered, so this sits
            // there until the edge asks for it, and nothing deadlocks.
            let _ = theirs.write_all(&response).await;
            let _ = theirs.flush().await;
            let mut seen = Vec::new();
            let _ = theirs.read_to_end(&mut seen).await;
            seen
        });
        self.received.lock().await.push(handle);
        Ok(mine)
    }
}

/// A backend that hands over one prepared stream, for the case where the
/// test needs to drive the backend side by hand. Declared at module scope
/// because an item after a statement inside a test body is a clippy error.
struct Fixed(Mutex<Option<DuplexStream>>);

impl Backend for Fixed {
    type Stream = DuplexStream;

    fn authority(&self) -> &'static str {
        "127.0.0.1:11434"
    }

    async fn connect(&self) -> std::io::Result<DuplexStream> {
        Ok(self.0.lock().await.take().expect("connected once"))
    }
}

/// Run one exchange against a fresh client stream, returning what the
/// client saw.
async fn exchange(
    request: &[u8],
    policy: &TokenPolicy,
    backend: &CountingBackend,
) -> (Outcome, Vec<u8>) {
    let (mut client, mut edge) = duplex(64 * 1024);
    client.write_all(request).await.unwrap();
    client.shutdown().await.unwrap();

    let (credential, _) = Credential::new(policy).expect("a usable policy");
    let outcome = tokio::time::timeout(
        std::time::Duration::from_secs(5),
        serve_exchange(&mut edge, &credential, backend, TEST_PEER),
    )
    .await
    .expect("the exchange must not hang")
    .expect("no transport failure");

    drop(edge);
    let mut seen = Vec::new();
    client.read_to_end(&mut seen).await.unwrap();
    (outcome, seen)
}

fn get(auth: Option<&str>) -> Vec<u8> {
    let mut req = b"GET /v1/models HTTP/1.1\r\nHost: 127.0.0.1:8080\r\n".to_vec();
    if let Some(value) = auth {
        req.extend_from_slice(format!("Authorization: {value}\r\n").as_bytes());
    }
    req.extend_from_slice(b"\r\n");
    req
}

fn supplied() -> TokenPolicy {
    TokenPolicy::Supplied(TOKEN.to_owned())
}

// ── Admitted ─────────────────────────────────────────────────────────────

#[tokio::test]
async fn an_authorized_request_reaches_the_backend_and_the_response_comes_back() {
    let backend = CountingBackend::new(OK_RESPONSE);
    let (outcome, seen) = exchange(
        &get(Some(&format!("Bearer {TOKEN}"))),
        &supplied(),
        &backend,
    )
    .await;

    assert_eq!(outcome, Outcome::Forwarded);
    assert_eq!(backend.connects(), 1);
    assert!(
        String::from_utf8_lossy(&seen).starts_with("HTTP/1.1 200 OK"),
        "the backend's response reaches the client: {}",
        String::from_utf8_lossy(&seen)
    );
}

/// Serving open is a configuration, not the absence of a check.
#[tokio::test]
async fn serving_open_forwards_a_request_with_no_credential() {
    let backend = CountingBackend::new(OK_RESPONSE);
    let (outcome, _) = exchange(&get(None), &TokenPolicy::InsecureNoAuth, &backend).await;
    assert_eq!(outcome, Outcome::Forwarded);
    assert_eq!(backend.connects(), 1);
}

// ── Refused, with the backend untouched ──────────────────────────────────

/// The assertion this whole module is shaped around.
///
/// "Returned 401" and "the backend never saw it" are different promises,
/// and only the second is what `lib.rs` means by "before a byte reaches
/// your backend". A status code cannot tell them apart; the counter can.
#[tokio::test]
async fn a_rejected_request_opens_no_backend_connection_at_all() {
    let backend = CountingBackend::new(OK_RESPONSE);
    let wrong = [
        None,
        Some("Bearer wrong-token"),
        Some(TOKEN),
        Some(&format!("Bearer {TOKEN}x")),
        Some(&format!("Basic {TOKEN}")),
    ];
    for (i, auth) in wrong.iter().enumerate() {
        let (outcome, seen) = exchange(&get(auth.as_deref()), &supplied(), &backend).await;
        assert_eq!(outcome, Outcome::Unauthorized, "case {i}");
        assert!(
            String::from_utf8_lossy(&seen).starts_with("HTTP/1.1 401"),
            "case {i} must be refused"
        );
    }
    assert_eq!(
        backend.connects(),
        0,
        "after {} unauthorized requests the backend was never contacted",
        wrong.len()
    );
    assert!(backend.received().await.is_empty(), "and sent nothing");
}

/// A 401 produced here cannot be confused with anything upstream said,
/// because upstream has not been spoken to.
#[tokio::test]
async fn the_401_is_synthesized_locally_and_advertises_the_scheme() {
    let backend = CountingBackend::new(OK_RESPONSE);
    let (_, seen) = exchange(&get(None), &supplied(), &backend).await;
    let text = String::from_utf8_lossy(&seen);

    assert!(text.starts_with("HTTP/1.1 401 Unauthorized"));
    assert!(text.contains("WWW-Authenticate: Bearer"));
    assert!(text.contains("invalid_api_key"));
    assert_eq!(backend.connects(), 0);
}

/// An ambiguously framed request is refused before the credential is even
/// consulted, and so also before the backend exists.
#[tokio::test]
async fn an_ambiguously_framed_request_is_refused_without_a_backend_connection() {
    let backend = CountingBackend::new(OK_RESPONSE);
    let smuggle = b"POST /v1/chat/completions HTTP/1.1\r\nHost: x\r\nContent-Length: 6\r\nTransfer-Encoding: chunked\r\n\r\n0\r\n\r\n";

    let (outcome, seen) = exchange(smuggle, &supplied(), &backend).await;
    assert_eq!(outcome, Outcome::BadRequest);
    assert!(String::from_utf8_lossy(&seen).starts_with("HTTP/1.1 400"));
    assert_eq!(backend.connects(), 0, "smuggling never reaches the backend");
}

/// Even carrying a valid credential. Framing is checked first because a
/// request the edge cannot read unambiguously is one it must not forward,
/// whoever sent it.
#[tokio::test]
async fn framing_is_refused_before_the_credential_is_consulted() {
    let backend = CountingBackend::new(OK_RESPONSE);
    let mut smuggle = b"POST /v1/chat/completions HTTP/1.1\r\nHost: x\r\n".to_vec();
    smuggle.extend_from_slice(format!("Authorization: Bearer {TOKEN}\r\n").as_bytes());
    smuggle.extend_from_slice(b"Content-Length: 6\r\nTransfer-Encoding: chunked\r\n\r\n0\r\n\r\n");

    let (outcome, _) = exchange(&smuggle, &supplied(), &backend).await;
    assert_eq!(outcome, Outcome::BadRequest);
    assert_eq!(backend.connects(), 0);
}

#[tokio::test]
async fn a_head_that_is_not_http_is_refused_without_a_backend_connection() {
    let backend = CountingBackend::new(OK_RESPONSE);
    let (outcome, seen) = exchange(b"this is not http\r\n\r\n", &supplied(), &backend).await;
    assert_eq!(outcome, Outcome::BadRequest);
    assert!(String::from_utf8_lossy(&seen).starts_with("HTTP/1.1 400"));
    assert_eq!(backend.connects(), 0);
}

// ── What the backend receives ────────────────────────────────────────────

/// Connection-scoped headers stop at the edge; the message survives.
/// `Authorization` is forwarded on purpose — that is what lets a `Supplied`
/// embedder's own backend check the same credential a second time.
#[tokio::test]
async fn the_backend_receives_the_message_headers_and_not_the_connection_ones() {
    let backend = CountingBackend::new(OK_RESPONSE);
    let mut req = b"GET /v1/models HTTP/1.1\r\nHost: 127.0.0.1:8080\r\n".to_vec();
    req.extend_from_slice(format!("Authorization: Bearer {TOKEN}\r\n").as_bytes());
    req.extend_from_slice(b"Connection: keep-alive, X-Hop\r\nX-Hop: 1\r\n");
    req.extend_from_slice(b"X-Forwarded-For: 203.0.113.1\r\nAccept: */*\r\n\r\n");

    let (outcome, _) = exchange(&req, &supplied(), &backend).await;
    assert_eq!(outcome, Outcome::Forwarded);

    let sent = String::from_utf8(backend.received().await).expect("ascii");
    let lower = sent.to_ascii_lowercase();
    assert!(
        sent.contains("Host: 127.0.0.1:11434"),
        "the backend's own authority: {sent}"
    );
    assert!(!sent.contains("127.0.0.1:8080"), "not the client's: {sent}");
    assert!(
        lower.contains("authorization: bearer"),
        "forwarded for the second check"
    );
    assert!(
        !lower.contains("x-hop"),
        "a nominated hop-by-hop header: {sent}"
    );
    // The client's `Connection` is gone; the edge's own is in its place.
    // Not a survival: this connection carries one exchange and the edge
    // drops it afterwards, so saying so is what keeps a keep-alive backend
    // from holding an `UntilClose` response open forever.
    assert!(
        !lower.contains("connection: keep-alive"),
        "the client's connection header must not survive: {sent}"
    );
    assert_eq!(
        lower.matches("connection:").count(),
        1,
        "exactly one, and it is the edge's: {sent}"
    );
    assert!(
        lower.contains("connection: close"),
        "the backend is told this connection carries one exchange: {sent}"
    );
    assert!(
        !lower.contains("x-forwarded-for"),
        "a forwarding chain: {sent}"
    );
    assert!(lower.contains("accept: */*"), "a message header survives");
}

/// The backend is told the request came through the tunnel, and from whom —
/// in the edge's words, not the client's. A client that arrives already
/// claiming a `Via` or a peer is describing a chain that does not exist,
/// and its claim is replaced rather than extended.
#[tokio::test]
async fn the_backend_sees_the_edges_tunnel_markers_and_not_the_clients() {
    let backend = CountingBackend::new(OK_RESPONSE);
    let mut req = b"GET /v1/models HTTP/1.1\r\nHost: 127.0.0.1:8080\r\n".to_vec();
    req.extend_from_slice(format!("Authorization: Bearer {TOKEN}\r\n").as_bytes());
    req.extend_from_slice(b"Via: 1.1 somebody-else\r\nVIA: 1.0 another\r\n");
    req.extend_from_slice(b"X-Modelpipe-Peer: 000000000000\r\n\r\n");

    let (outcome, _) = exchange(&req, &supplied(), &backend).await;
    assert_eq!(outcome, Outcome::Forwarded);

    let sent = String::from_utf8(backend.received().await).expect("ascii");
    let lower = sent.to_ascii_lowercase();
    assert_eq!(
        lower.matches("\r\nvia:").count(),
        1,
        "exactly one Via: {sent}"
    );
    assert!(
        sent.contains("Via: 1.1 modelpipe"),
        "and it is the edge's: {sent}"
    );
    assert!(
        !lower.contains("somebody-else") && !lower.contains("another"),
        "{sent}"
    );
    assert_eq!(
        lower.matches("\r\nx-modelpipe-peer:").count(),
        1,
        "exactly one peer marker: {sent}"
    );
    assert!(
        sent.contains(&format!("X-Modelpipe-Peer: {TEST_PEER}")),
        "and it names the peer the listener saw: {sent}"
    );
    assert!(
        !sent.contains("000000000000"),
        "not the one the client claimed: {sent}"
    );
}

/// A length-framed body arrives intact and is not truncated by the head
/// read having over-read into it.
#[tokio::test]
async fn a_request_body_reaches_the_backend_byte_for_byte() {
    let backend = CountingBackend::new(OK_RESPONSE);
    let payload = r#"{"model":"llama","messages":[{"role":"user","content":"hi"}]}"#;
    let mut req = b"POST /v1/chat/completions HTTP/1.1\r\nHost: x\r\n".to_vec();
    req.extend_from_slice(format!("Authorization: Bearer {TOKEN}\r\n").as_bytes());
    req.extend_from_slice(format!("Content-Length: {}\r\n\r\n", payload.len()).as_bytes());
    req.extend_from_slice(payload.as_bytes());

    let (outcome, _) = exchange(&req, &supplied(), &backend).await;
    assert_eq!(outcome, Outcome::Forwarded);
    assert!(
        String::from_utf8(backend.received().await)
            .expect("ascii")
            .ends_with(payload),
        "the body must arrive unaltered"
    );
}

// ── Streaming ────────────────────────────────────────────────────────────

/// The product is a token stream, and a `collect` anywhere in the response
/// path would still return 200 with the right bytes. This asserts the
/// frames leave the edge as they arrive.
#[tokio::test]
async fn a_streaming_response_reaches_the_client_frame_by_frame() {
    let backend = CountingBackend::new(OK_RESPONSE);
    let (mut client, mut edge) = duplex(64 * 1024);
    client
        .write_all(&get(Some(&format!("Bearer {TOKEN}"))))
        .await
        .unwrap();

    // A backend that sends a head, then one frame, then waits before the
    // last. If the edge buffers, the first read below times out.
    let (mine, mut theirs) = duplex(64 * 1024);
    let released = Arc::new(tokio::sync::Notify::new());
    let wait = released.clone();
    tokio::spawn(async move {
        theirs
            .write_all(b"HTTP/1.1 200 OK\r\nContent-Type: text/event-stream\r\n\r\n")
            .await
            .unwrap();
        theirs.write_all(b"data: first\n\n").await.unwrap();
        theirs.flush().await.unwrap();
        wait.notified().await;
        theirs.write_all(b"data: [DONE]\n\n").await.unwrap();
    });

    let fixed = Fixed(Mutex::new(Some(mine)));
    let (credential, _) = Credential::new(&supplied()).expect("a usable token");
    let pump = tokio::spawn(async move {
        serve_exchange(&mut edge, &credential, &fixed, TEST_PEER)
            .await
            .unwrap()
    });

    // Read until the frame appears rather than a fixed number of bytes.
    // What is under test is that it arrives while the backend is still
    // producing; a hardcoded length additionally asserts the response head's
    // size, which breaks the moment a header is added to it.
    let mut text = String::new();
    tokio::time::timeout(std::time::Duration::from_secs(5), async {
        let mut buf = [0u8; 512];
        loop {
            let n = client.read(&mut buf).await.expect("read");
            assert!(n > 0, "the stream ended early: {text}");
            text.push_str(&String::from_utf8_lossy(&buf[..n]));
            if text.contains("data: first") {
                return;
            }
        }
    })
    .await
    .unwrap_or_else(|_| panic!("the first frame must arrive before the backend finishes: {text}"));
    assert!(text.contains("200 OK"), "{text}");

    released.notify_one();
    assert_eq!(pump.await.unwrap(), Outcome::Forwarded);
    let _ = backend.connects();
}

// ── The backend's half of the framing rules ──────────────────────────────

/// A backend whose response head is written by hand, so a test can say
/// exactly what came back. Distinct from `Fixed` in taking the bytes rather
/// than a prepared stream, and in never closing: a real keep-alive backend
/// holds the socket open after answering, which is the condition under
/// which every framing mistake below becomes a hang rather than a wrong
/// answer.
struct KeepAlive(Mutex<Option<DuplexStream>>);

impl KeepAlive {
    fn new(response: &'static str) -> Self {
        let (mine, mut theirs) = duplex(64 * 1024);
        tokio::spawn(async move {
            let mut sink = Vec::new();
            // Answer, then hold the connection open exactly as Ollama and
            // llama-server do. Nothing here ever writes EOF.
            let _ = theirs.write_all(response.as_bytes()).await;
            let _ = theirs.flush().await;
            let _ = theirs.read_to_end(&mut sink).await;
        });
        Self(Mutex::new(Some(mine)))
    }
}

impl Backend for KeepAlive {
    type Stream = DuplexStream;

    fn authority(&self) -> &'static str {
        "127.0.0.1:11434"
    }

    async fn connect(&self) -> std::io::Result<DuplexStream> {
        Ok(self.0.lock().await.take().expect("connected once"))
    }
}

/// Drive one exchange against a keep-alive backend, failing rather than
/// hanging. The deadline is the assertion: every case below completes in
/// microseconds when the framing is right and never completes when it is
/// not.
async fn against_keepalive(request: &[u8], response: &'static str) -> (Outcome, String) {
    let backend = KeepAlive::new(response);
    let (mut client, mut edge) = duplex(64 * 1024);
    client.write_all(request).await.unwrap();

    let (credential, _) = Credential::new(&supplied()).expect("a usable token");
    let outcome = tokio::time::timeout(
        std::time::Duration::from_secs(5),
        serve_exchange(&mut edge, &credential, &backend, TEST_PEER),
    )
    .await
    .expect("a keep-alive backend must not hang the exchange")
    .expect("no transport failure");

    drop(edge);
    let mut seen = Vec::new();
    client.read_to_end(&mut seen).await.unwrap();
    (outcome, String::from_utf8_lossy(&seen).into_owned())
}

fn authed(method: &str) -> Vec<u8> {
    format!("{method} /v1/models HTTP/1.1\r\nHost: x\r\nAuthorization: Bearer {TOKEN}\r\n\r\n")
        .into_bytes()
}

/// RFC 9112 §6.3: the status code settles this before any header does.
/// Reading it from the headers alone is not a wrong answer, it is a hang —
/// `204` declares no framing, so the old rule resolved it to `UntilClose`
/// and waited for a close a keep-alive backend never sends.
#[tokio::test]
async fn a_bodyless_status_is_framed_by_its_status_and_not_by_its_headers() {
    for response in [
        "HTTP/1.1 204 No Content\r\n\r\n",
        "HTTP/1.1 304 Not Modified\r\nContent-Length: 42\r\n\r\n",
    ] {
        let (outcome, seen) = against_keepalive(&authed("GET"), response).await;
        assert_eq!(outcome, Outcome::Forwarded, "{response:?}");
        assert!(
            seen.starts_with("HTTP/1.1 3") || seen.starts_with("HTTP/1.1 2"),
            "{seen}"
        );
    }
}

/// The same rule from the request's side: a response to `HEAD` carries the
/// length a `GET` would have returned, and no body. Waiting for that many
/// bytes is a wait that cannot end.
#[tokio::test]
async fn a_head_response_is_not_a_body_to_wait_for() {
    let (outcome, seen) = against_keepalive(
        &authed("HEAD"),
        "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: 4096\r\n\r\n",
    )
    .await;
    assert_eq!(outcome, Outcome::Forwarded);
    assert!(seen.starts_with("HTTP/1.1 200"), "{seen}");
    assert!(
        seen.to_ascii_lowercase().contains("content-length: 4096"),
        "the declared length still describes what a GET would return: {seen}"
    );
}

/// An interim response is a head that precedes the real one. Delivered as
/// final it was not merely the wrong status: a `1xx` declares no framing,
/// so what followed was `UntilClose` and the client waited forever.
#[tokio::test]
async fn an_interim_response_is_skipped_and_the_real_one_forwarded() {
    let (outcome, seen) = against_keepalive(
        &authed("GET"),
        "HTTP/1.1 100 Continue\r\n\r\nHTTP/1.1 200 OK\r\nContent-Length: 2\r\n\r\n{}",
    )
    .await;
    assert_eq!(outcome, Outcome::Forwarded);
    assert!(
        seen.starts_with("HTTP/1.1 200"),
        "the interim head must not be the answer: {seen}"
    );
    assert!(seen.ends_with("{}"), "and the real body arrives: {seen}");
}

/// The rule the request path enforces, applied to the backend. A response
/// carrying both `Content-Length` and `Transfer-Encoding` is the
/// request-smuggling shape; refusing it inbound and resolving it outbound
/// is one rule applied on one side of the pipe only.
#[tokio::test]
async fn an_ambiguously_framed_backend_response_is_refused_rather_than_resolved() {
    let (outcome, seen) = against_keepalive(
        &authed("GET"),
        "HTTP/1.1 200 OK\r\nContent-Length: 2\r\nTransfer-Encoding: chunked\r\n\r\n{}",
    )
    .await;
    assert_eq!(outcome, Outcome::BadGateway);
    assert!(
        seen.starts_with("HTTP/1.1 502"),
        "the client did nothing wrong, and the backend's answer is unreadable: {seen}"
    );
}

/// A backend that will not take the connection owes the client an answer.
/// Before this the stream simply died: no status, no malformed response, no
/// bytes at all — indistinguishable from the tunnel being gone.
#[tokio::test]
async fn a_backend_that_refuses_the_connection_is_reported_as_a_gateway_failure() {
    struct Refusing;
    impl Backend for Refusing {
        type Stream = DuplexStream;
        fn authority(&self) -> &'static str {
            "127.0.0.1:11434"
        }
        async fn connect(&self) -> std::io::Result<DuplexStream> {
            Err(std::io::Error::new(
                std::io::ErrorKind::ConnectionRefused,
                "nothing is listening",
            ))
        }
    }

    let (mut client, mut edge) = duplex(64 * 1024);
    client.write_all(&authed("GET")).await.unwrap();
    client.shutdown().await.unwrap();

    let (credential, _) = Credential::new(&supplied()).expect("a usable token");
    let outcome = serve_exchange(&mut edge, &credential, &Refusing, TEST_PEER)
        .await
        .expect("a refused backend is an answer, not a transport failure");
    assert_eq!(outcome, Outcome::BadGateway);

    drop(edge);
    let mut seen = Vec::new();
    client.read_to_end(&mut seen).await.unwrap();
    let text = String::from_utf8_lossy(&seen);
    assert!(text.starts_with("HTTP/1.1 502"), "got: {text}");
}

// ── The backend that answers before it has finished listening ────────────

/// A backend that replies as soon as it has the head and then stops
/// reading, exactly as a server rejecting an oversized payload does. Its
/// buffer is small so the edge's write blocks well before the body is
/// through — which is the whole point: the answer is available while the
/// request is still going out.
struct AnswersEarly(Mutex<Option<DuplexStream>>);

impl AnswersEarly {
    fn new(response: &'static str) -> Self {
        let (mine, mut theirs) = duplex(1024);
        tokio::spawn(async move {
            let mut buf = [0u8; 256];
            let mut seen = Vec::new();
            while !seen.windows(4).any(|w| w == b"\r\n\r\n") {
                match theirs.read(&mut buf).await {
                    Ok(0) | Err(_) => return,
                    Ok(n) => seen.extend_from_slice(&buf[..n]),
                }
            }
            let _ = theirs.write_all(response.as_bytes()).await;
            let _ = theirs.flush().await;
            // And now it stops reading. The edge is mid-body.
            std::future::pending::<()>().await;
        });
        Self(Mutex::new(Some(mine)))
    }

    /// The same shape, except it hangs up after answering instead of
    /// stalling — the real-socket version, where the close arrives as an
    /// RST because the receive queue is not empty.
    ///
    /// Written out rather than folded into [`new`](Self::new) with a flag:
    /// that constructor is the fixture of the test that guards the
    /// overlapping read, and leaving it untouched is worth more than the
    /// dozen lines it saves.
    fn hanging_up(response: &'static str) -> Self {
        let (mine, mut theirs) = duplex(1024);
        tokio::spawn(async move {
            let mut buf = [0u8; 256];
            let mut seen = Vec::new();
            while !seen.windows(4).any(|w| w == b"\r\n\r\n") {
                match theirs.read(&mut buf).await {
                    Ok(0) | Err(_) => return,
                    Ok(n) => seen.extend_from_slice(&buf[..n]),
                }
            }
            let _ = theirs.write_all(response.as_bytes()).await;
            let _ = theirs.flush().await;
            drop(theirs);
        });
        Self(Mutex::new(Some(mine)))
    }
}

impl Backend for AnswersEarly {
    type Stream = DuplexStream;

    fn authority(&self) -> &'static str {
        "127.0.0.1:11434"
    }

    async fn connect(&self) -> std::io::Result<DuplexStream> {
        Ok(self.0.lock().await.take().expect("connected once"))
    }
}

/// A backend that answers before it has read the request must not cost the
/// client the answer.
///
/// This is the shape of every `413` on an oversized payload, every `400` on
/// bad JSON, every `429` — and SECURITY.md names multi-MiB vision payloads
/// as the expected traffic. Written sequentially, the edge was still inside
/// `write_all` when the backend stopped draining, so the write blocked
/// forever here and, against a real socket, died with `ECONNRESET` and took
/// the already-delivered response with it.
#[tokio::test]
async fn a_backend_that_answers_before_reading_the_body_is_still_heard() {
    let backend =
        AnswersEarly::new("HTTP/1.1 413 Payload Too Large\r\nContent-Length: 2\r\n\r\nno");
    let (mut client, mut edge) = duplex(256 * 1024);

    // A body far larger than the backend's buffer, so the pump cannot
    // finish and the answer is only reachable by reading while it stalls.
    let body = "x".repeat(64 * 1024);
    let request = format!(
        "POST /v1/chat/completions HTTP/1.1\r\nHost: x\r\n\
         Authorization: Bearer {TOKEN}\r\nContent-Length: {}\r\n\r\n{body}",
        body.len()
    );
    client.write_all(request.as_bytes()).await.unwrap();

    let (credential, _) = Credential::new(&supplied()).expect("a usable token");
    let outcome = tokio::time::timeout(
        std::time::Duration::from_secs(5),
        serve_exchange(&mut edge, &credential, &backend, TEST_PEER),
    )
    .await
    .expect("the answer is already in hand; waiting on the body is waiting forever")
    .expect("no transport failure");
    assert_eq!(outcome, Outcome::Forwarded);

    drop(edge);
    let mut seen = Vec::new();
    client.read_to_end(&mut seen).await.unwrap();
    let text = String::from_utf8_lossy(&seen);
    assert!(
        text.starts_with("HTTP/1.1 413"),
        "the backend's answer must reach the client: {text}"
    );
    assert!(text.ends_with("no"), "body included: {text}");
}

// ── Bounds before authentication ─────────────────────────────────────────

/// A stream opened and then left silent must not hold a task forever. This
/// is the third bound on what a leaked ticket is worth before it
/// authenticates, alongside the head's size and the per-peer stream cap.
#[tokio::test(start_paused = true)]
async fn a_peer_that_never_finishes_asking_is_timed_out() {
    let backend = CountingBackend::new(OK_RESPONSE);
    let (mut client, mut edge) = duplex(64 * 1024);

    // A head that begins and never ends: valid so far, so the parser keeps
    // asking for more.
    client
        .write_all(b"GET /v1/models HTTP/1.1\r\nHost: x\r\n")
        .await
        .unwrap();

    let (credential, _) = Credential::new(&supplied()).expect("a usable token");
    // `start_paused` advances the clock only when everything is idle, so
    // this resolves the moment the timeout is the only thing left to wait
    // on — no real thirty seconds pass.
    let outcome = serve_exchange(&mut edge, &credential, &backend, TEST_PEER)
        .await
        .expect("a timeout is not a transport failure");

    assert_eq!(outcome, Outcome::TimedOut);
    assert_eq!(backend.connects(), 0, "and the backend never heard of it");

    drop(edge);
    let mut seen = Vec::new();
    client.read_to_end(&mut seen).await.unwrap();
    assert!(
        seen.is_empty(),
        "a peer that never finished asking is owed no answer: {seen:?}"
    );
}

/// The timeout bounds the *head*, not the request. An admitted inference
/// call may run for many minutes, which is the product.
#[tokio::test(start_paused = true)]
async fn a_slow_head_that_arrives_in_time_is_served_normally() {
    let backend = CountingBackend::new(OK_RESPONSE);
    let (mut client, mut edge) = duplex(64 * 1024);
    let auth = format!("Bearer {TOKEN}");

    tokio::spawn(async move {
        client
            .write_all(b"GET /v1/models HTTP/1.1\r\nHost: x\r\n")
            .await
            .unwrap();
        tokio::time::sleep(std::time::Duration::from_secs(5)).await;
        client
            .write_all(format!("Authorization: {auth}\r\n\r\n").as_bytes())
            .await
            .unwrap();
        // Held open so the response has somewhere to go. Not a round
        // minute, which clippy reads as a unit that wants rewriting.
        tokio::time::sleep(std::time::Duration::from_secs(45)).await;
    });

    let (credential, _) = Credential::new(&supplied()).expect("a usable token");
    let outcome = serve_exchange(&mut edge, &credential, &backend, TEST_PEER)
        .await
        .expect("no transport failure");
    assert_eq!(outcome, Outcome::Forwarded);
    assert_eq!(backend.connects(), 1);
}

// ── The client that stops mid-body ───────────────────────────────────────

/// A complete 200, as text, for the stubs below that take one.
const OK_TEXT: &str = "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\n\
                       Content-Length: 2\r\n\r\n{}";

/// A backend that reads the whole declared body before it answers — what
/// every real inference server does, and what no other stub in this file
/// does. `CountingBackend`, `KeepAlive` and `AnswersEarly` all write first
/// and read afterwards, which is precisely why none of them could reproduce
/// a client that stops mid-body: the edge got its answer regardless.
///
/// `saw_eof` is the mechanism assertion. Over a duplex it can only become
/// true if the edge actually called `poll_shutdown` on its write half, so a
/// test asserting it cannot pass by the exchange having ended some other
/// way.
struct ReadsWholeBody {
    stream: Mutex<Option<DuplexStream>>,
    saw_eof: Arc<AtomicBool>,
}

/// What a [`ReadsWholeBody`] does when the body stops before its declared
/// end. Three real server behaviours, and the edge owes a different answer
/// to none of them — which is the point of testing all three.
#[derive(Clone, Copy)]
enum OnEarlyEnd {
    /// Close without a word. What most servers do once their read comes up
    /// short and their parser rejects what arrived.
    HangUp,
    /// Answer anyway, from what it did receive.
    Answer(&'static str),
    /// Say nothing and hold the socket open — the case the half-close alone
    /// cannot fix.
    Hold,
}

impl ReadsWholeBody {
    /// Answers `response` once the whole declared body has arrived, and
    /// does `after_eof` if it never does.
    fn new(response: &'static str, after_eof: OnEarlyEnd) -> Self {
        let (mine, mut theirs) = duplex(256 * 1024);
        let saw_eof = Arc::new(AtomicBool::new(false));
        let flag = saw_eof.clone();
        tokio::spawn(async move {
            let mut buf = [0u8; 4096];
            let mut seen = Vec::new();
            while !seen.windows(4).any(|w| w == b"\r\n\r\n") {
                match theirs.read(&mut buf).await {
                    Ok(0) | Err(_) => return,
                    Ok(n) => seen.extend_from_slice(&buf[..n]),
                }
            }
            let head_end = seen
                .windows(4)
                .position(|w| w == b"\r\n\r\n")
                .expect("head ends")
                + 4;
            while !body_complete(&seen, head_end) {
                match theirs.read(&mut buf).await {
                    Ok(0) => {
                        flag.store(true, Ordering::SeqCst);
                        match after_eof {
                            OnEarlyEnd::HangUp => {}
                            OnEarlyEnd::Answer(text) => {
                                let _ = theirs.write_all(text.as_bytes()).await;
                                let _ = theirs.flush().await;
                            }
                            // Holds the stream open by never returning, so
                            // the edge sees neither an answer nor an end.
                            OnEarlyEnd::Hold => std::future::pending::<()>().await,
                        }
                        return;
                    }
                    Err(_) => return,
                    Ok(n) => seen.extend_from_slice(&buf[..n]),
                }
            }
            let _ = theirs.write_all(response.as_bytes()).await;
            let _ = theirs.flush().await;
        });
        Self {
            stream: Mutex::new(Some(mine)),
            saw_eof,
        }
    }

    /// Whether the backend was ever told the body had stopped.
    fn saw_eof(&self) -> bool {
        self.saw_eof.load(Ordering::SeqCst)
    }
}

impl Backend for ReadsWholeBody {
    type Stream = DuplexStream;

    fn authority(&self) -> &'static str {
        "127.0.0.1:11434"
    }

    async fn connect(&self) -> std::io::Result<DuplexStream> {
        Ok(self.stream.lock().await.take().expect("connected once"))
    }
}

/// Whether everything the head promised has arrived, by whichever framing
/// it declared. Deliberately a hand-rolled read of the two cases rather
/// than a call into `framing`: a stub that shared the code under test could
/// agree with it about a body neither had read correctly.
fn body_complete(seen: &[u8], head_end: usize) -> bool {
    let head = String::from_utf8_lossy(&seen[..head_end]).to_ascii_lowercase();
    if head.contains("transfer-encoding: chunked") {
        return seen[head_end..].windows(5).any(|w| w == b"0\r\n\r\n");
    }
    let declared = head
        .split("content-length:")
        .nth(1)
        .and_then(|rest| rest.split("\r\n").next())
        .and_then(|value| value.trim().parse::<usize>().ok())
        .unwrap_or(0);
    seen.len() - head_end >= declared
}

/// A backend that takes the head and then vanishes without answering.
///
/// Its buffer is small on purpose, so the edge is still writing the body
/// when the peer goes — which is what makes the write, rather than the
/// read, the first thing to fail.
struct Vanishing(Mutex<Option<DuplexStream>>);

impl Vanishing {
    fn new() -> Self {
        let (mine, mut theirs) = duplex(1024);
        tokio::spawn(async move {
            let mut buf = [0u8; 256];
            let mut seen = Vec::new();
            while !seen.windows(4).any(|w| w == b"\r\n\r\n") {
                match theirs.read(&mut buf).await {
                    Ok(0) | Err(_) => return,
                    Ok(n) => seen.extend_from_slice(&buf[..n]),
                }
            }
            drop(theirs);
        });
        Self(Mutex::new(Some(mine)))
    }
}

impl Backend for Vanishing {
    type Stream = DuplexStream;

    fn authority(&self) -> &'static str {
        "127.0.0.1:11434"
    }

    async fn connect(&self) -> std::io::Result<DuplexStream> {
        Ok(self.0.lock().await.take().expect("connected once"))
    }
}

/// Drive one exchange against a backend of any shape, returning what the
/// client saw.
///
/// `hang_up` closes the client's send half, which is the difference between
/// an aborted upload and a slow one — and the reason it is a parameter is
/// that the distinction is the subject of half the tests below.
///
/// `patience` bounds the whole exchange. It is a parameter rather than a
/// constant because a test of something that is *meant* to take time would
/// otherwise fail against its own subject: under `start_paused` the clock
/// jumps to whichever timer is nearest, and a fixed five seconds is always
/// nearer than a ten-second grace.
async fn drive<B: Backend + Sync>(
    request: &[u8],
    backend: &B,
    hang_up: bool,
    patience: std::time::Duration,
) -> (Outcome, String) {
    let (mut client, mut edge) = duplex(256 * 1024);
    client.write_all(request).await.unwrap();
    if hang_up {
        client.shutdown().await.unwrap();
    }

    let (credential, _) = Credential::new(&supplied()).expect("a usable token");
    let outcome = tokio::time::timeout(
        patience,
        serve_exchange(&mut edge, &credential, backend, TEST_PEER),
    )
    .await
    .expect("the exchange must not hang")
    .expect("no transport failure");

    drop(edge);
    let mut seen = Vec::new();
    client.read_to_end(&mut seen).await.unwrap();
    (outcome, String::from_utf8_lossy(&seen).into_owned())
}

/// [`drive`] with the patience every test that is not about time wants.
async fn against<B: Backend + Sync>(
    request: &[u8],
    backend: &B,
    hang_up: bool,
) -> (Outcome, String) {
    drive(request, backend, hang_up, std::time::Duration::from_secs(5)).await
}

/// A POST whose head declares `declared` bytes and whose body carries
/// `body`. When the two disagree the request is a truncated upload.
fn post(declared: usize, body: &str) -> Vec<u8> {
    format!(
        "POST /v1/chat/completions HTTP/1.1\r\nHost: x\r\n\
         Authorization: Bearer {TOKEN}\r\nContent-Length: {declared}\r\n\r\n{body}"
    )
    .into_bytes()
}

/// The bug, at the edge. A client that declares a length, sends less and
/// hangs up used to leave the exchange waiting forever on a response the
/// backend could not send: its head was already upstream, so the backend
/// sat blocked against a length that would never arrive, and nothing shut
/// the write half to tell it otherwise.
///
/// Measured, before the halves were told apart: the client got nothing at
/// all, and because the exchange never returned it held its in-flight
/// guard, so `ServeHandle::shutdown` never returned either — the first
/// Ctrl-C on `modelpipe serve` hung past twenty seconds where ordinary
/// traffic took one.
#[tokio::test]
async fn a_truncated_request_body_is_answered_rather_than_waited_on() {
    let backend = ReadsWholeBody::new(OK_TEXT, OnEarlyEnd::HangUp);

    let (outcome, seen) = against(&post(1000, "{\"model\":\""), &backend, true).await;

    assert_eq!(outcome, Outcome::Unfinished);
    assert!(seen.starts_with("HTTP/1.1 400"), "got: {seen}");
    assert!(seen.contains("incomplete_request"), "got: {seen}");
    assert!(
        backend.saw_eof(),
        "the backend must be told where the body stopped, not merely abandoned"
    );
}

/// The control for the test above. The same backend, the same route, a body
/// that arrives whole — so the 400 cannot be this stub simply never
/// answering anything.
#[tokio::test]
async fn a_complete_request_body_reaches_the_same_backend_that_hangs_on_a_short_one() {
    let backend = ReadsWholeBody::new(OK_TEXT, OnEarlyEnd::HangUp);
    let body = "{\"model\":\"m\"}";

    let (outcome, seen) = against(&post(body.len(), body), &backend, true).await;

    assert_eq!(outcome, Outcome::Forwarded);
    assert!(seen.starts_with("HTTP/1.1 200"), "got: {seen}");
    assert!(!backend.saw_eof(), "a complete body is not an early end");
}

/// A backend that answers the short body anyway must be heard. The
/// synthesized 400 is what this edge says when there is nothing to relay,
/// never something it says over the top of a real answer.
#[tokio::test]
async fn a_backend_that_answers_the_short_body_is_heard_rather_than_overridden() {
    let backend = ReadsWholeBody::new(
        OK_TEXT,
        OnEarlyEnd::Answer("HTTP/1.1 400 Bad Request\r\nContent-Length: 9\r\n\r\ntruncated"),
    );

    let (outcome, seen) = against(&post(1000, "{\"model\":\""), &backend, true).await;

    assert_eq!(outcome, Outcome::Forwarded);
    assert!(
        seen.ends_with("truncated"),
        "the backend's own words, not ours: {seen}"
    );
}

/// The trap, and the reason the fault only decides what happens when
/// nothing came back. A backend that answers and *then* hangs up makes the
/// edge's next write fail — so a fix that charged any failed pump to the
/// client would answer 400 while a perfectly good 413 sat unread in the
/// buffer. That is the size-dependent phantom the overlapping read was
/// written for in the first place, and this is the test that would catch
/// its return.
#[tokio::test]
async fn an_answer_followed_by_a_hangup_is_relayed_rather_than_charged_to_the_client() {
    let backend =
        AnswersEarly::hanging_up("HTTP/1.1 413 Payload Too Large\r\nContent-Length: 2\r\n\r\nno");
    let body = "x".repeat(64 * 1024);

    let (outcome, seen) = against(&post(body.len(), &body), &backend, false).await;

    assert_eq!(outcome, Outcome::Forwarded);
    assert!(seen.starts_with("HTTP/1.1 413"), "got: {seen}");
}

/// The control for the fault verdict itself: not every failed pump is the
/// client's. A backend that goes away mid-body is a gateway failure, and
/// reporting it as a 400 would send whoever is debugging it to the wrong
/// machine entirely.
#[tokio::test]
async fn a_backend_that_hangs_up_mid_body_is_a_gateway_failure_rather_than_a_client_one() {
    let backend = Vanishing::new();
    let body = "x".repeat(64 * 1024);

    let (outcome, seen) = against(&post(body.len(), &body), &backend, false).await;

    assert_eq!(outcome, Outcome::BadGateway);
    assert!(seen.starts_with("HTTP/1.1 502"), "got: {seen}");
}

/// The other way a body stops, and the one where the client is still
/// there. An unreadable chunk size is not a hang-up — the socket is open
/// and the client is waiting — so the answer it is owed actually reaches
/// it. `body::forward` reports this as `ErrorKind::Other`, which is what a
/// sink failure would look like too; only watching the sink tells them
/// apart.
#[tokio::test]
async fn a_chunked_body_with_an_unreadable_size_is_refused_rather_than_relayed_on() {
    let backend = ReadsWholeBody::new(OK_TEXT, OnEarlyEnd::HangUp);
    let request = format!(
        "POST /v1/chat/completions HTTP/1.1\r\nHost: x\r\n\
         Authorization: Bearer {TOKEN}\r\nTransfer-Encoding: chunked\r\n\r\nzz\r\n"
    );

    let (outcome, seen) = against(request.as_bytes(), &backend, false).await;

    assert_eq!(outcome, Outcome::Unfinished);
    assert!(seen.starts_with("HTTP/1.1 400"), "got: {seen}");
    assert!(backend.saw_eof(), "the backend is told here too");
}

/// The control for the test above: a well-formed chunked body still gets
/// through. Nothing else in this file exercises chunked at the exchange
/// level, so without this the refusal could be the edge rejecting every
/// chunked request.
#[tokio::test]
async fn a_well_formed_chunked_body_is_forwarded_rather_than_refused() {
    let backend = ReadsWholeBody::new(OK_TEXT, OnEarlyEnd::HangUp);
    let request = format!(
        "POST /v1/chat/completions HTTP/1.1\r\nHost: x\r\n\
         Authorization: Bearer {TOKEN}\r\nTransfer-Encoding: chunked\r\n\r\n\
         a\r\n0123456789\r\n0\r\n\r\n"
    );

    let (outcome, seen) = against(request.as_bytes(), &backend, false).await;

    assert_eq!(outcome, Outcome::Forwarded);
    assert!(seen.starts_with("HTTP/1.1 200"), "got: {seen}");
}

/// The half-close is most of the answer, and this is the rest of it. A
/// backend that is told the body stopped and neither answers nor closes put
/// the exchange straight back where it started — waiting forever on a
/// response that was never coming, holding the in-flight guard that
/// `ServeHandle::shutdown` drains against, so one misbehaving server could
/// hold a teardown open indefinitely.
///
/// Measured against a real server blocked in `read`: the client got
/// nothing and `serve` would not shut down, with the half-close working
/// perfectly. Being told is not the same as acting on it.
///
/// Virtual time, so the ten-second grace costs the suite nothing.
#[tokio::test(start_paused = true)]
async fn a_backend_that_neither_answers_nor_closes_is_not_waited_on_forever() {
    let backend = ReadsWholeBody::new(OK_TEXT, OnEarlyEnd::Hold);

    let (outcome, seen) = drive(
        &post(1000, "{\"model\":\""),
        &backend,
        true,
        crate::request_body::ANSWER_GRACE * 100,
    )
    .await;

    assert_eq!(outcome, Outcome::Unfinished);
    assert!(seen.starts_with("HTTP/1.1 400"), "got: {seen}");
    assert!(backend.saw_eof(), "it was told, it simply did nothing");
}

/// The control for the grace, and the promise that it is not a request
/// timeout wearing a different name: a body that arrived whole leaves it
/// disarmed, so a backend taking far longer than the grace to think is
/// waited on exactly as before. Without this, shortening `ANSWER_GRACE` to
/// nothing would still pass every other test in this file.
#[tokio::test(start_paused = true)]
async fn a_slow_answer_to_a_complete_body_is_waited_for_however_long_it_takes() {
    let backend = Deliberating::new(OK_TEXT, crate::request_body::ANSWER_GRACE * 100);
    let body = "{\"model\":\"m\"}";

    let (outcome, seen) = drive(
        &post(body.len(), body),
        &backend,
        true,
        crate::request_body::ANSWER_GRACE * 1000,
    )
    .await;

    assert_eq!(outcome, Outcome::Forwarded);
    assert!(seen.starts_with("HTTP/1.1 200"), "got: {seen}");
}

/// A backend that reads the whole body and then thinks for a long time —
/// an inference call, which is the product.
struct Deliberating(Mutex<Option<DuplexStream>>);

impl Deliberating {
    fn new(response: &'static str, think_for: std::time::Duration) -> Self {
        let (mine, mut theirs) = duplex(256 * 1024);
        tokio::spawn(async move {
            let mut buf = [0u8; 4096];
            let mut seen = Vec::new();
            while !seen.windows(4).any(|w| w == b"\r\n\r\n") {
                match theirs.read(&mut buf).await {
                    Ok(0) | Err(_) => return,
                    Ok(n) => seen.extend_from_slice(&buf[..n]),
                }
            }
            let head_end = seen
                .windows(4)
                .position(|w| w == b"\r\n\r\n")
                .expect("head ends")
                + 4;
            while !body_complete(&seen, head_end) {
                match theirs.read(&mut buf).await {
                    Ok(0) | Err(_) => return,
                    Ok(n) => seen.extend_from_slice(&buf[..n]),
                }
            }
            tokio::time::sleep(think_for).await;
            let _ = theirs.write_all(response.as_bytes()).await;
            let _ = theirs.flush().await;
        });
        Self(Mutex::new(Some(mine)))
    }
}

impl Backend for Deliberating {
    type Stream = DuplexStream;

    fn authority(&self) -> &'static str {
        "127.0.0.1:11434"
    }

    async fn connect(&self) -> std::io::Result<DuplexStream> {
        Ok(self.0.lock().await.take().expect("connected once"))
    }
}

// ── What the diagnostics say, and what they must never say ───────────────

/// The credential, as a string no other part of this crate produces.
///
/// The fourth such sentinel here — `credential.rs`, `serve_tests.rs` and
/// `api_surface.rs` each have their own — and distinct from all of them on
/// purpose: a leak that turns up in a failure message should name the route
/// it escaped through rather than leaving three candidates.
const LOG_TOKEN: &str = "sk-zzq-tracing-sentinel";

/// A second sentinel, in the query string.
///
/// Azure's OpenAI-compatible endpoints take `?api-key=`, so this is not a
/// hypothetical shape: it is a real client putting a real credential in a
/// real request target, and the edge forwards that target verbatim. What it
/// must not do is repeat it in a log line.
const LOG_QUERY: &str = "sk-zzq-query-sentinel";

// Where a captured line goes, if anything on this thread is listening.
//
// `None` is the state every other test in this binary runs in: the
// subscriber below is installed process-wide, and its writer throws the
// bytes away unless a test has armed this buffer. That is deliberate on both
// counts — it keeps the capture tests from reading each other's output, and
// it means all two hundred-odd tests here exercise the instrumentation
// rather than compiling past it.
thread_local! {
    static CAPTURED: RefCell<Option<Vec<u8>>> = const { RefCell::new(None) };
}

/// The writer behind the process-wide subscriber.
#[derive(Clone, Copy)]
struct Sink;

impl std::io::Write for Sink {
    fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
        CAPTURED.with(|slot| {
            if let Some(into) = slot.borrow_mut().as_mut() {
                into.extend_from_slice(buf);
            }
        });
        Ok(buf.len())
    }

    fn flush(&mut self) -> std::io::Result<()> {
        Ok(())
    }
}

impl tracing_subscriber::fmt::MakeWriter<'_> for Sink {
    type Writer = Self;

    fn make_writer(&self) -> Self::Writer {
        *self
    }
}

/// Install the capturing subscriber once, for the whole test binary.
///
/// Global rather than attached per future, and that is a bug fix rather
/// than a preference. `tracing` keeps a process-wide maximum level which a
/// *scoped* subscriber raises on entry and drops again on exit; with these
/// tests running concurrently, one test's teardown switched the span macro
/// off in the middle of another's, and that one captured nothing at all.
/// Deterministically, and only when run beside its neighbours — which is
/// the shape of failure that reaches CI and not the developer.
///
/// The cost of a global is that "no subscriber at all" stops existing in
/// this binary. That case is not untested, it is tested elsewhere:
/// `tests/integration_pipe.rs` is a separate binary that installs nothing
/// and drives this edge over a real pipe, end to end. (`tests/api_surface.rs`
/// is a separate binary too, and also installs nothing, but it pins the
/// exported surface at compile time and never runs an exchange — so it is
/// not evidence for this and is not cited as such.)
fn capturing() {
    static ONCE: std::sync::Once = std::sync::Once::new();
    ONCE.call_once(|| {
        let subscriber = tracing_subscriber::fmt()
            .with_writer(Sink)
            .with_ansi(false)
            // TRACE because these tests assert about `debug` lines as well
            // as `info` ones, and the default maximum would silently drop
            // half of what they check.
            .with_max_level(tracing::Level::TRACE)
            .finish();
        tracing::subscriber::set_global_default(subscriber)
            .expect("nothing else in this binary may install a subscriber");
    });
}

/// Run one exchange with this thread's buffer armed, and hand back what was
/// written to it.
async fn logged(
    request: &[u8],
    policy: &TokenPolicy,
    backend: &CountingBackend,
) -> (Outcome, String) {
    capturing();
    CAPTURED.with(|slot| *slot.borrow_mut() = Some(Vec::new()));
    // `#[tokio::test]` builds a current-thread runtime, so everything the
    // exchange does — including the backend stub's spawned reader — runs on
    // this thread and writes into this thread's buffer. A multi-threaded
    // flavour here would capture whatever happened to stay put.
    let (outcome, _) = exchange(request, policy, backend).await;
    let written = CAPTURED
        .with(|slot| slot.borrow_mut().take())
        .expect("armed just above");
    (outcome, String::from_utf8(written).expect("utf-8"))
}

/// A request carrying both sentinels: one in the credential, one in the
/// query string.
fn sensitive_get() -> Vec<u8> {
    let mut req =
        format!("GET /v1/models?api-key={LOG_QUERY} HTTP/1.1\r\nHost: 127.0.0.1:8080\r\n")
            .into_bytes();
    req.extend_from_slice(format!("Authorization: Bearer {LOG_TOKEN}\r\n").as_bytes());
    req.extend_from_slice(b"\r\n");
    req
}

/// The line an operator actually wants: what was asked for, what came back,
/// and how long it took.
#[tokio::test]
async fn a_forwarded_exchange_reports_what_it_did() {
    let backend = CountingBackend::new(OK_RESPONSE);
    let (outcome, log) = logged(
        &sensitive_get(),
        &TokenPolicy::Supplied(LOG_TOKEN.to_owned()),
        &backend,
    )
    .await;

    assert_eq!(outcome, Outcome::Forwarded);
    for field in [
        "method=\"GET\"",
        "path=\"/v1/models\"",
        "status=200",
        "outcome=\"forwarded\"",
        "elapsed_ms=",
    ] {
        assert!(log.contains(field), "no {field} in:\n{log}");
    }
}

/// The property the crate documentation states: no event carries a
/// credential.
///
/// Both sentinels are in the request bytes — asserted below rather than
/// assumed, because a test that looks for a secret the request never
/// carried is a test that passes for the wrong reason and would go on
/// passing after the redaction was removed.
#[tokio::test]
async fn no_line_repeats_the_credential_or_the_query_string() {
    let request = sensitive_get();
    let raw = String::from_utf8(request.clone()).unwrap();
    assert!(raw.contains(LOG_TOKEN), "the request must carry the token");
    assert!(raw.contains(LOG_QUERY), "the request must carry the query");

    let backend = CountingBackend::new(OK_RESPONSE);
    let (outcome, log) = logged(
        &request,
        &TokenPolicy::Supplied(LOG_TOKEN.to_owned()),
        &backend,
    )
    .await;

    assert_eq!(outcome, Outcome::Forwarded);
    assert!(!log.contains(LOG_TOKEN), "the token is in the log:\n{log}");
    assert!(!log.contains(LOG_QUERY), "the query is in the log:\n{log}");
    // The positive half, so this cannot pass by the subscriber having
    // captured nothing at all — which is the way a redaction test rots.
    assert!(
        log.contains("path=\"/v1/models\""),
        "captured nothing:\n{log}"
    );
    assert!(
        !log.contains("api-key"),
        "even the parameter name is gone:\n{log}"
    );
}

/// A grant is a credential too, and the same rule covers it: the one
/// request it admits is logged like any other, and the value that admitted
/// it is not in the line.
#[tokio::test]
async fn a_grant_that_admits_is_logged_without_the_grant() {
    const GRANT: &str = "sk-zzq-grant-sentinel";
    let mut request = b"GET /v1/models HTTP/1.1\r\nHost: 127.0.0.1:8080\r\n".to_vec();
    request.extend_from_slice(format!("Authorization: Bearer {GRANT}\r\n\r\n").as_bytes());

    let backend = CountingBackend::new(OK_RESPONSE);
    let (credential, _) = Credential::new(&supplied()).expect("a usable policy");
    assert!(credential.grant(GRANT.to_owned(), std::time::Duration::from_mins(1), None));

    capturing();
    CAPTURED.with(|slot| *slot.borrow_mut() = Some(Vec::new()));
    let (mut client, mut edge) = duplex(64 * 1024);
    client.write_all(&request).await.unwrap();
    client.shutdown().await.unwrap();
    let outcome = serve_exchange(&mut edge, &credential, &backend, TEST_PEER)
        .await
        .expect("no transport failure");
    drop(edge);
    let log = CAPTURED
        .with(|slot| slot.borrow_mut().take())
        .map(|bytes| String::from_utf8(bytes).expect("utf-8"))
        .expect("armed just above");

    assert_eq!(outcome, Outcome::Forwarded, "the grant admitted");
    assert_eq!(backend.connects(), 1);
    assert!(
        log.contains("outcome=\"forwarded\""),
        "captured nothing:\n{log}"
    );
    assert!(!log.contains(GRANT), "the grant is in the log:\n{log}");
}

/// A refusal is reported as one, and is still attributable.
///
/// A 401 nobody can tie to a path tells an operator only that somebody,
/// somewhere, was wrong — which is why the method and path are recorded
/// before the credential is checked rather than after it.
#[tokio::test]
async fn a_refused_exchange_names_the_refusal_and_the_request() {
    let backend = CountingBackend::new(OK_RESPONSE);
    let (outcome, log) = logged(&sensitive_get(), &supplied(), &backend).await;

    assert_eq!(outcome, Outcome::Unauthorized);
    assert_eq!(backend.connects(), 0, "the backend must not be contacted");
    for field in [
        "outcome=\"unauthorized\"",
        "method=\"GET\"",
        "path=\"/v1/models\"",
    ] {
        assert!(log.contains(field), "no {field} in:\n{log}");
    }
    // A rejected credential is exactly as secret as an accepted one.
    assert!(!log.contains(LOG_TOKEN), "the token is in the log:\n{log}");
    assert!(!log.contains(LOG_QUERY), "the query is in the log:\n{log}");
    // No status: nothing upstream ever answered, and an empty `status=`
    // would be worse than its absence.
    assert!(
        !log.contains("status="),
        "there was no upstream status:\n{log}"
    );
}

/// Instrumentation changes what is *said* about an exchange and nothing
/// about what it does.
///
/// What this proves is narrower than it looks, and the narrowness is the
/// honest part: the subscriber is installed process-wide by `capturing`
/// above, so both runs below have one — the difference is only whether this
/// thread's buffer is armed. That still catches the mistake worth catching
/// here, which is a recording or a field evaluation that changes the
/// exchange under it.
///
/// The genuinely subscriber-free case is covered where it actually exists:
/// `tests/integration_pipe.rs` is a separate binary that installs no
/// subscriber and drives this edge over a live pipe, end to end.
#[tokio::test]
async fn an_exchange_is_unchanged_by_being_watched() {
    let watched = CountingBackend::new(OK_RESPONSE);
    let (with_capture, log) = logged(
        &sensitive_get(),
        &TokenPolicy::Supplied(LOG_TOKEN.to_owned()),
        &watched,
    )
    .await;
    assert!(
        !log.is_empty(),
        "the armed run must have captured something"
    );

    let unwatched = CountingBackend::new(OK_RESPONSE);
    let (without, seen) = exchange(
        &sensitive_get(),
        &TokenPolicy::Supplied(LOG_TOKEN.to_owned()),
        &unwatched,
    )
    .await;

    assert_eq!(with_capture, without);
    assert_eq!(watched.connects(), unwatched.connects());
    assert!(String::from_utf8_lossy(&seen).starts_with("HTTP/1.1 200"));
}

/// A backend that answers with a head declaring more body than it sends.
///
/// The head is complete and correctly framed, so the edge forwards it to
/// the client and starts on the body; the backend then closes with the body
/// short. `body::forward_exact` turns that into `unexpected_eof`, which
/// `run` propagates — and that is the one route to `serve_exchange`'s `Err`
/// arm which is entirely the backend's doing, with the client still
/// connected and owed nothing further.
struct Truncating(Mutex<Option<DuplexStream>>);

impl Truncating {
    fn new() -> Self {
        let (mine, mut theirs) = duplex(1024);
        tokio::spawn(async move {
            let mut buf = [0u8; 256];
            let mut seen = Vec::new();
            while !seen.windows(4).any(|w| w == b"\r\n\r\n") {
                match theirs.read(&mut buf).await {
                    Ok(0) | Err(_) => return,
                    Ok(n) => seen.extend_from_slice(&buf[..n]),
                }
            }
            let _ = theirs
                .write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 10\r\n\r\nfive!")
                .await;
            let _ = theirs.flush().await;
            drop(theirs);
        });
        Self(Mutex::new(Some(mine)))
    }
}

impl Backend for Truncating {
    type Stream = DuplexStream;

    fn authority(&self) -> &'static str {
        "127.0.0.1:11434"
    }

    async fn connect(&self) -> std::io::Result<DuplexStream> {
        Ok(self.0.lock().await.take().expect("connected once"))
    }
}

/// Drive one exchange to a transport failure, capturing what was logged.
///
/// Separate from [`logged`] because every other helper in this file
/// `.expect()`s the `Result`: the `Err` arm is the one outcome none of them
/// can reach, and it was the one arm nothing exercised.
async fn logged_failure<B: Backend + Sync>(
    request: &[u8],
    backend: &B,
) -> (std::io::Error, String) {
    capturing();
    CAPTURED.with(|slot| *slot.borrow_mut() = Some(Vec::new()));

    let (mut client, mut edge) = duplex(64 * 1024);
    client.write_all(request).await.unwrap();
    let (credential, _) = Credential::new(&supplied()).expect("a usable token");
    let error = tokio::time::timeout(
        std::time::Duration::from_secs(5),
        serve_exchange(&mut edge, &credential, backend, TEST_PEER),
    )
    .await
    .expect("the exchange must not hang")
    .expect_err("this exchange must fail at the transport");

    let written = CAPTURED
        .with(|slot| slot.borrow_mut().take())
        .expect("armed just above");
    (error, String::from_utf8(written).expect("utf-8"))
}

/// The `Err` arm, and the counterexample to what it must not claim.
///
/// A comment here used to say an `Err` means "the local stream itself
/// failing, the peer going away mid-exchange rather than anything the
/// request did". This test is the case that is false for: the backend
/// declared ten bytes, sent five and closed, and the client is still
/// connected with a 200 head already in hand. Nothing about it is the
/// peer's doing.
#[tokio::test]
async fn a_backend_that_under_delivers_its_body_is_reported_as_a_failure() {
    let backend = Truncating::new();
    let (error, log) = logged_failure(&get(Some(&format!("Bearer {TOKEN}"))), &backend).await;

    assert_eq!(
        error.kind(),
        std::io::ErrorKind::UnexpectedEof,
        "the body ended before its declared length: {error}"
    );
    assert!(
        log.contains("exchange failed"),
        "no failure line in:\n{log}"
    );
    // Still attributable, for the reason a refusal is: a failure nobody can
    // tie to a request tells an operator only that something, somewhere,
    // broke.
    assert!(log.contains("method=\"GET\""), "unattributable:\n{log}");
    assert!(
        log.contains("status=200"),
        "the backend's head did arrive, and the line should say so:\n{log}"
    );
}

/// The recording order that `span.record("status", …)` sits above the
/// response-framing check for.
///
/// A backend answering `200` and framing it ambiguously is refused. Without
/// the status on the line, that refusal reads as a bare gateway failure and
/// sends whoever is debugging it to the backend's logic; with it, the line
/// says the backend replied fine and framed it in a way this edge will not
/// resolve. The comment and the commit message both defended this ordering
/// and nothing tested it.
///
/// Its negative control is `a_refused_exchange_names_the_refusal_and_the_request`,
/// which asserts a 401 carries no `status=` at all — so this cannot pass by
/// the field being present unconditionally.
#[tokio::test]
async fn a_refused_backend_response_still_reports_the_status_it_sent() {
    let backend = CountingBackend::new(
        b"HTTP/1.1 200 OK\r\nContent-Length: 2\r\nTransfer-Encoding: chunked\r\n\r\n{}",
    );
    let (outcome, log) = logged(
        &get(Some(&format!("Bearer {TOKEN}"))),
        &supplied(),
        &backend,
    )
    .await;

    assert_eq!(outcome, Outcome::BadGateway);
    assert!(log.contains("outcome=\"bad_gateway\""), "{log}");
    assert!(
        log.contains("status=200"),
        "the status it refused is missing:\n{log}"
    );
}

// ── The expectation a client wants answered before it uploads ────────────

/// A GET carrying an arbitrary `Expect` value, so the header is exercised
/// without a body needing to exist.
fn expecting(value: &str) -> Vec<u8> {
    format!(
        "GET /v1/models HTTP/1.1\r\nHost: 127.0.0.1:8080\r\n\
         Authorization: Bearer {TOKEN}\r\nExpect: {value}\r\n\r\n"
    )
    .into_bytes()
}

/// The client asked to be told before sending its body, and is told.
///
/// Measured before this, with the real binary and a 2 MB POST that curl
/// sends with `Expect` of its own accord: 1.015s through the pipe against
/// 0.048s straight at the same backend. The whole second was curl waiting
/// out its own timeout for an interim response that never came.
#[tokio::test]
async fn an_expectation_of_continue_is_answered_before_the_response() {
    let backend = CountingBackend::new(OK_RESPONSE);
    let (outcome, seen) = exchange(&expecting("100-continue"), &supplied(), &backend).await;

    assert_eq!(outcome, Outcome::Forwarded);
    let text = String::from_utf8(seen).expect("ascii");
    assert!(
        text.starts_with("HTTP/1.1 100 Continue\r\n\r\n"),
        "the interim answer must come first: {text:?}"
    );
    // And it is interim, not final: the real response still follows it.
    assert!(
        text.contains("HTTP/1.1 200 OK"),
        "the final response is still sent: {text:?}"
    );
}

/// The negative control: a request that did not ask gets no interim answer.
///
/// Without this, always writing `100 Continue` would satisfy the test above
/// while putting an unasked-for interim response in front of every single
/// exchange — which a strict client is entitled to treat as a protocol
/// error.
#[tokio::test]
async fn a_request_that_expects_nothing_gets_no_interim_response() {
    let backend = CountingBackend::new(OK_RESPONSE);
    let (outcome, seen) = exchange(
        &get(Some(&format!("Bearer {TOKEN}"))),
        &supplied(),
        &backend,
    )
    .await;

    assert_eq!(outcome, Outcome::Forwarded);
    let text = String::from_utf8(seen).expect("ascii");
    assert!(text.starts_with("HTTP/1.1 200"), "{text:?}");
    assert!(!text.contains("100 Continue"), "{text:?}");
}

/// A refused request is refused, and the refusal is the first thing it sees.
///
/// The interim answer is written on the admitted path only. Getting this
/// wrong would tell a client with a bad key to go ahead and upload, and
/// then refuse it — which is the exact waste `Expect` exists to prevent,
/// arranged by the party that was supposed to prevent it.
#[tokio::test]
async fn an_expectation_is_not_answered_for_a_request_that_is_refused() {
    let backend = CountingBackend::new(OK_RESPONSE);
    let (outcome, seen) = exchange(&expecting("100-continue"), &supplied(), &backend).await;
    assert_eq!(outcome, Outcome::Forwarded, "the sanity case");

    // Now the same request with no credential at all.
    let refused = CountingBackend::new(OK_RESPONSE);
    let mut req = expecting("100-continue");
    req = String::from_utf8(req)
        .unwrap()
        .replace(&format!("Authorization: Bearer {TOKEN}\r\n"), "")
        .into_bytes();
    let (outcome, denied) = exchange(&req, &supplied(), &refused).await;

    assert_eq!(outcome, Outcome::Unauthorized);
    assert_eq!(refused.connects(), 0, "the backend was not contacted");
    let text = String::from_utf8(denied).expect("ascii");
    assert!(text.starts_with("HTTP/1.1 401"), "{text:?}");
    assert!(!text.contains("100 Continue"), "{text:?}");
    drop(seen);
}

/// Only the expectation this edge can meet is answered.
///
/// RFC 9110 §10.1.1 defines exactly one expectation and makes it a token,
/// so the match is case-insensitive and nothing else matches. An
/// expectation this edge does not know is left to the backend, which is the
/// party that might be able to meet it.
#[tokio::test]
async fn the_expectation_is_matched_by_name_and_not_by_presence() {
    for value in ["100-continue", "100-CONTINUE", "100-Continue"] {
        let backend = CountingBackend::new(OK_RESPONSE);
        let (_, seen) = exchange(&expecting(value), &supplied(), &backend).await;
        let text = String::from_utf8(seen).expect("ascii");
        assert!(text.starts_with("HTTP/1.1 100"), "{value}: {text:?}");
    }
    for value in ["200-ok", "something-else", ""] {
        let backend = CountingBackend::new(OK_RESPONSE);
        let (_, seen) = exchange(&expecting(value), &supplied(), &backend).await;
        let text = String::from_utf8(seen).expect("ascii");
        assert!(
            !text.contains("100 Continue"),
            "{value:?} is not an expectation this edge answers: {text:?}"
        );
    }
}

/// The backend's own interim response is still skipped, and now for a
/// better reason than before: the client has already had one.
#[tokio::test]
async fn a_backend_interim_response_is_not_relayed_on_top_of_the_edge_s() {
    let (outcome, seen) = against_keepalive(
        &expecting("100-continue"),
        "HTTP/1.1 100 Continue\r\n\r\nHTTP/1.1 200 OK\r\nContent-Length: 2\r\n\r\n{}",
    )
    .await;

    assert_eq!(outcome, Outcome::Forwarded);
    assert_eq!(
        seen.matches("100 Continue").count(),
        1,
        "exactly one interim response reaches the client: {seen:?}"
    );
    assert!(seen.contains("HTTP/1.1 200 OK"), "{seen:?}");
}

/// Which 502 the "would not take the connection" path writes.
///
/// `Outcome::BadGateway` is one variant across three causes on purpose, so
/// the outcome alone cannot pin this: only the body distinguishes a backend
/// that was never reached from one that answered unreadably, and getting it
/// backwards sends the reader to the component that is working.
#[tokio::test]
async fn a_backend_that_was_never_reached_says_so_in_the_body() {
    struct Refusing;
    impl Backend for Refusing {
        type Stream = DuplexStream;
        fn authority(&self) -> &'static str {
            "127.0.0.1:11434"
        }
        async fn connect(&self) -> std::io::Result<DuplexStream> {
            Err(std::io::Error::new(
                std::io::ErrorKind::ConnectionRefused,
                "nothing is listening",
            ))
        }
    }

    let (mut client, mut edge) = duplex(64 * 1024);
    client.write_all(&authed("GET")).await.unwrap();
    client.shutdown().await.unwrap();
    let (credential, _) = Credential::new(&supplied()).expect("a usable token");
    let outcome = serve_exchange(&mut edge, &credential, &Refusing, TEST_PEER)
        .await
        .unwrap();
    drop(edge);
    let mut seen = Vec::new();
    client.read_to_end(&mut seen).await.unwrap();
    let text = String::from_utf8(seen).expect("ascii");

    assert_eq!(outcome, Outcome::BadGateway);
    assert!(text.contains(r#""code":"backend_unreachable""#), "{text}");
    assert!(!text.contains(r#""code":"bad_gateway""#), "{text}");
}

/// The negative control for the test above: a backend that *was* reached and
/// answered unreadably keeps the other body.
///
/// Without this, always writing `backend_unreachable` would satisfy the test
/// above while destroying the distinction it exists to make.
#[tokio::test]
async fn a_backend_that_answered_unreadably_keeps_the_other_body() {
    let backend = CountingBackend::new(
        b"HTTP/1.1 200 OK\r\nContent-Length: 2\r\nTransfer-Encoding: chunked\r\n\r\n{}",
    );
    let (outcome, seen) = exchange(
        &get(Some(&format!("Bearer {TOKEN}"))),
        &supplied(),
        &backend,
    )
    .await;
    let text = String::from_utf8(seen).expect("ascii");

    assert_eq!(outcome, Outcome::BadGateway);
    assert_eq!(backend.connects(), 1, "the backend was reached");
    assert!(text.contains(r#""code":"bad_gateway""#), "{text}");
    assert!(!text.contains(r#""code":"backend_unreachable""#), "{text}");
}

/// `Expect` is a comma-separated list (RFC 9110 §10.1.1), so the continue
/// can arrive beside another expectation.
#[tokio::test]
async fn an_expectation_list_containing_continue_is_still_answered() {
    for value in ["100-continue, foo", "foo, 100-continue", "foo,100-CONTINUE"] {
        let backend = CountingBackend::new(OK_RESPONSE);
        let (_, seen) = exchange(&expecting(value), &supplied(), &backend).await;
        let text = String::from_utf8(seen).expect("ascii");
        assert!(
            text.starts_with("HTTP/1.1 100 Continue"),
            "{value:?} asks for the continue: {text:?}"
        );
    }
    // And the control: a list with no continue in it is still not answered.
    for value in ["foo", "foo, bar", "100-continues"] {
        let backend = CountingBackend::new(OK_RESPONSE);
        let (_, seen) = exchange(&expecting(value), &supplied(), &backend).await;
        let text = String::from_utf8(seen).expect("ascii");
        assert!(!text.contains("100 Continue"), "{value:?}: {text:?}");
    }
}

/// A request admitted by a named token tells the backend the name, in the
/// edge's words; one the primary admitted carries no such header, so a
/// backend with one client never learns that names exist. The client's own
/// claim, in either case, is gone.
#[tokio::test]
async fn the_backend_is_told_which_named_token_admitted_and_nothing_when_none_did() {
    const LAPTOP: &str = "sk-zzq-laptop-sentinel";
    let (credential, _) = Credential::new(&supplied()).expect("a usable policy");
    credential
        .add_named("laptop", LAPTOP.to_owned())
        .expect("a valid name and token");

    for (bearer, expected) in [(LAPTOP, Some("laptop")), (TOKEN, None)] {
        let backend = CountingBackend::new(OK_RESPONSE);
        let mut req = b"GET /v1/models HTTP/1.1\r\nHost: 127.0.0.1:8080\r\n".to_vec();
        req.extend_from_slice(format!("Authorization: Bearer {bearer}\r\n").as_bytes());
        req.extend_from_slice(b"X-Modelpipe-Device: forged\r\n\r\n");

        let (mut client, mut edge) = duplex(64 * 1024);
        client.write_all(&req).await.unwrap();
        client.shutdown().await.unwrap();
        let outcome = serve_exchange(&mut edge, &credential, &backend, TEST_PEER)
            .await
            .expect("no transport failure");
        drop(edge);
        assert_eq!(outcome, Outcome::Forwarded);

        let sent = String::from_utf8(backend.received().await).expect("ascii");
        let lower = sent.to_ascii_lowercase();
        assert!(
            !lower.contains("forged"),
            "the client's claim is gone: {sent}"
        );
        match expected {
            Some(name) => {
                assert_eq!(
                    lower.matches("\r\nx-modelpipe-device:").count(),
                    1,
                    "exactly one device marker: {sent}"
                );
                assert!(
                    sent.contains(&format!("X-Modelpipe-Device: {name}")),
                    "and it names the token: {sent}"
                );
            }
            None => assert!(
                !lower.contains("x-modelpipe-device"),
                "the primary admits with no device marker: {sent}"
            ),
        }
    }
}

/// With an upstream bearer set, the backend is handed the edge's
/// credential and never the device's — the property that lets every
/// device hold a different key while the backend keeps exactly one.
#[tokio::test]
async fn the_backend_is_handed_the_upstream_bearer_and_never_the_devices() {
    const LAPTOP: &str = "sk-zzq-laptop-sentinel";
    const UPSTREAM: &str = "sk-zzq-backend-sentinel";
    let (credential, _) = Credential::new(&TokenPolicy::Named).expect("a usable policy");
    credential
        .add_named("laptop", LAPTOP.to_owned())
        .expect("a valid name and token");
    assert!(credential.set_upstream(Some(UPSTREAM.to_owned())));

    let backend = CountingBackend::new(OK_RESPONSE);
    let mut req = b"GET /v1/models HTTP/1.1\r\nHost: 127.0.0.1:8080\r\n".to_vec();
    req.extend_from_slice(format!("Authorization: Bearer {LAPTOP}\r\n\r\n").as_bytes());

    let (mut client, mut edge) = duplex(64 * 1024);
    client.write_all(&req).await.unwrap();
    client.shutdown().await.unwrap();
    let outcome = serve_exchange(&mut edge, &credential, &backend, TEST_PEER)
        .await
        .expect("no transport failure");
    drop(edge);
    assert_eq!(outcome, Outcome::Forwarded);

    let sent = String::from_utf8(backend.received().await).expect("ascii");
    assert!(
        sent.contains(&format!("Authorization: Bearer {UPSTREAM}")),
        "the edge's bearer: {sent}"
    );
    assert!(
        !sent.contains(LAPTOP),
        "the device's key must never reach the backend: {sent}"
    );
    assert_eq!(
        sent.to_ascii_lowercase()
            .matches("\r\nauthorization:")
            .count(),
        1,
        "exactly one: {sent}"
    );
}