modelpipe 0.4.0

//! The pipe, end to end, over a real iroh connection.
//!
//! Everything below runs two endpoints on one machine and pairs them with a
//! real ticket. That is the point: every layer beneath has been tested in
//! isolation, and this is where the claim "the README's first code block is
//! true" is either demonstrated or not.
//!
//! These are also the only tests that can check the asymmetry the product
//! is built on — that restarting the listener rotates the ticket while
//! rotating the token leaves every pairing intact — because it is a
//! statement about two live sides, not about either one.

mod common;

use std::time::Duration;

use common::{MockBackend, Scratch, request, within};
use modelpipe::{
    CloseReason, ConnectOptions, NetworkMetrics, PipeStatus, ServeOptions, Ticket, TokenPolicy,
};
use tokio::io::{AsyncReadExt as _, AsyncWriteExt as _};

const OK_BODY: &str = r#"{"object":"list","data":[]}"#;

/// Bring up a listener over `backend`, and a connect side paired to it.
async fn paired(
    backend: &MockBackend,
    auth: TokenPolicy,
) -> (modelpipe::ServeHandle, modelpipe::ConnectHandle, String) {
    let mut serve_opts = ServeOptions::default();
    serve_opts.auth = auth;
    // Boxed: binding an iroh endpoint is a large future, and holding one
    // inline in a test that also holds the connect side pushes the whole
    // task's frame past what clippy's nursery is willing to see on a stack.
    let serving = within(
        "serve must bind",
        Box::pin(modelpipe::serve(&backend.url, serve_opts)),
    )
    .await
    .expect("serve");

    let ticket = serving.ticket();
    let connected = within(
        "connect must bind its local port",
        Box::pin(modelpipe::connect(&ticket, ConnectOptions::default())),
    )
    .await
    .expect("connect");
    // `connect` returns with the port bound and the dial still running, so
    // the pairing is not up yet. Every test below sends a request the
    // moment this returns, and a pipe with no connection behind it answers
    // 502 — which would make this helper the source of a failure belonging
    // to nothing it is testing.
    within("the pairing must form", carrying(&connected)).await;

    let url = connected.base_url();
    (serving, connected, url)
}

/// Wait until `ready` holds, checking it on a slow tick.
///
/// Deliberately without a deadline of its own: every caller wraps it in
/// [`within`], so a property that never arrives is named by the wait it
/// was waiting for rather than by a bare elapsed timer. Written once
/// because the two handles are separate types — a serve-side wait and a
/// connect-side wait cannot share a signature, and a closure is the only
/// thing they can share.
async fn until(mut ready: impl FnMut() -> bool) {
    while !ready() {
        tokio::time::sleep(Duration::from_millis(10)).await;
    }
}

/// Wait until the connect side has actually reached the peer.
///
/// `Idle` is the state a freshly returned handle is in, and the state it
/// stays in while the dial runs; anything else means a connection formed.
async fn carrying(handle: &modelpipe::ConnectHandle) {
    until(|| handle.status() != PipeStatus::Idle).await;
}

/// With discovery and port-mapping off on both sides, the ticket carries
/// every path its holder has — on one machine, that is enough. This is the
/// configuration an embedder that minted the ticket a moment ago and will
/// never need it to survive a change of network can run in, and it is the
/// one that contacts nothing but the relay.
#[tokio::test]
async fn a_pairing_still_forms_with_discovery_and_port_mapping_off() {
    let backend = MockBackend::json(200, OK_BODY).await;
    let mut serve_opts = ServeOptions::default();
    serve_opts.auth = TokenPolicy::Generate;
    serve_opts.port_mapping = false;
    serve_opts.discovery = false;
    let serving = within(
        "serve must bind without discovery",
        Box::pin(modelpipe::serve(&backend.url, serve_opts)),
    )
    .await
    .expect("serve");

    let mut connect_opts = ConnectOptions::default();
    connect_opts.port_mapping = false;
    connect_opts.discovery = false;
    let connected = within(
        "connect must bind on the ticket's own paths",
        Box::pin(modelpipe::connect(&serving.ticket(), connect_opts)),
    )
    .await
    .expect("connect");
    within("the pairing must form on those paths", carrying(&connected)).await;

    let response = within(
        "a request must cross the pipe",
        request(&connected.base_url(), "/v1/models", Some(&bearer(&serving))),
    )
    .await
    .expect("request");
    assert!(response.starts_with("HTTP/1.1 200 OK"), "got: {response}");

    connected.shutdown().await;
    serving.shutdown().await;
}

/// Wait until the connect side reports `wanted`.
///
/// Polled rather than driven by `status_changed`, and the difference is the
/// subject of the test below. That method snapshots at the moment it is
/// *polled* and then waits for a change, which is exactly right for a
/// watcher already parked on the handle and exactly wrong for a caller that
/// arrives after the transition: measured here, the connection's death was
/// noticed and published before `ServeHandle::shutdown` had even returned,
/// so a `status_changed` called afterwards waited for a second change that
/// was never coming.
async fn settles_on(handle: &modelpipe::ConnectHandle, wanted: PipeStatus) {
    until(|| handle.status() == wanted).await;
}

fn bearer(handle: &modelpipe::ServeHandle) -> String {
    format!("Bearer {}", handle.token().expect("a token is enforced"))
}

// ── Coming up ────────────────────────────────────────────────────────────

/// `connect` returns when the **local port** is bound, not when the peer
/// answers.
///
/// The dial is what takes the time: iroh spends about thirty seconds giving
/// up on a peer that is not there, and a caller blocked for it cannot even
/// be told which port it was given, let alone point a client at it. That
/// wait is the whole reason the dial moved off `connect`'s path, and this
/// is the test that would have to be deleted to move it back.
///
/// Written against a peer that is genuinely gone — a listener minted and
/// then shut down — rather than a fabricated address, because a bogus
/// endpoint id fails at `addr_from` without ever reaching the dial and
/// would pass just as happily with the old ordering.
#[tokio::test]
async fn connect_binds_its_port_without_waiting_for_a_peer_that_is_not_there() {
    let backend = MockBackend::json(200, OK_BODY).await;
    let serving = within(
        "serve",
        Box::pin(modelpipe::serve(&backend.url, ServeOptions::default())),
    )
    .await
    .expect("serve");
    let ticket = serving.ticket();
    serving.shutdown().await;
    drop(serving);

    // Five seconds is the assertion. The old ordering took about thirty,
    // and no amount of slow machine turns thirty into five.
    let connected = tokio::time::timeout(
        Duration::from_secs(5),
        Box::pin(modelpipe::connect(&ticket, ConnectOptions::default())),
    )
    .await
    .expect("connect must not wait on a dial that will not land")
    .expect("binding the local port is all it has to do");

    assert_eq!(
        connected.status(),
        PipeStatus::Idle,
        "nobody has been reached, and the handle is what says so"
    );

    // And the port is genuinely open, which is the point of returning
    // early: a client can be pointed at it now and gets a 502 rather than a
    // refused connection while this side keeps looking.
    let authority = connected.local_addr().to_string();
    within("the advertised port must accept", async {
        tokio::net::TcpStream::connect(&authority)
            .await
            .expect("the local listener is up");
    })
    .await;

    within(
        "shutdown must not wait on the dial either",
        connected.shutdown(),
    )
    .await;
}

// ── The first byte ───────────────────────────────────────────────────────

/// The README's first code block, made true.
#[tokio::test]
async fn a_request_crosses_the_pipe_and_the_response_comes_back() {
    let backend = MockBackend::json(200, OK_BODY).await;
    let (serving, connected, url) = paired(&backend, TokenPolicy::Generate).await;

    let response = within(
        "a request must cross the pipe",
        request(&url, "/v1/models", Some(&bearer(&serving))),
    )
    .await
    .expect("request");

    assert!(response.starts_with("HTTP/1.1 200 OK"), "got: {response}");
    assert!(
        response.contains(OK_BODY),
        "the body must arrive: {response}"
    );
    assert_eq!(
        backend.accepts(),
        1,
        "and the backend served it exactly once"
    );

    let sent = backend.received().await;
    assert!(sent.contains("GET /v1/models"), "the path survives: {sent}");
    assert!(
        sent.contains(&format!(
            "Host: {}",
            backend.url.trim_start_matches("http://")
        )),
        "the Host names the backend: {sent}"
    );
    assert!(
        sent.contains("Via: 1.1 modelpipe"),
        "the backend is told the request came through the tunnel: {sent}"
    );
    let peer = sent
        .lines()
        .find_map(|line| line.strip_prefix("X-Modelpipe-Peer: "))
        .expect("the backend is told which peer");
    assert_eq!(peer.len(), 12, "a twelve-hex-character fingerprint: {peer}");
    assert!(peer.chars().all(|c| c.is_ascii_hexdigit()), "{peer}");

    connected.shutdown().await;
    serving.shutdown().await;
}

/// `base_url` is meant to be pasted into a client, so it must be a URL
/// pointing at something that answers.
#[tokio::test]
async fn the_base_url_is_something_a_client_can_actually_use() {
    let backend = MockBackend::json(200, OK_BODY).await;
    let (serving, connected, url) = paired(&backend, TokenPolicy::Generate).await;

    assert!(
        url.starts_with("http://127.0.0.1:"),
        "loopback by default: {url}"
    );
    assert!(url.ends_with("/v1"), "and the OpenAI base path: {url}");
    assert_eq!(
        connected.local_addr().to_string(),
        url.trim_start_matches("http://").trim_end_matches("/v1"),
        "the URL names the port actually bound"
    );

    connected.shutdown().await;
    serving.shutdown().await;
}

// ── Auth, at the far end of a real connection ────────────────────────────

/// The claim the crate is built on, checked across the whole pipe rather
/// than at the edge in isolation: a refused request never becomes a backend
/// connection.
#[tokio::test]
async fn an_unauthorized_request_never_reaches_the_backend() {
    let backend = MockBackend::json(200, OK_BODY).await;
    let (serving, connected, url) = paired(&backend, TokenPolicy::Generate).await;

    for auth in [None, Some("Bearer wrong"), Some("Basic whatever")] {
        let response = within("a refusal must arrive", request(&url, "/v1/models", auth))
            .await
            .expect("request");
        assert!(
            response.starts_with("HTTP/1.1 401"),
            "{auth:?} must be refused: {response}"
        );
    }
    assert_eq!(
        backend.accepts(),
        0,
        "after three refused requests the backend was never contacted"
    );

    connected.shutdown().await;
    serving.shutdown().await;
}

/// Serving open is a deliberate configuration, and the flag's name is the
/// warning rather than a second check.
#[tokio::test]
async fn serving_open_forwards_without_a_credential() {
    let backend = MockBackend::json(200, OK_BODY).await;
    let (serving, connected, url) = paired(&backend, TokenPolicy::InsecureNoAuth).await;

    assert_eq!(serving.token(), None, "there is no token to report");
    let response = within("must forward", request(&url, "/v1/models", None))
        .await
        .expect("request");
    assert!(response.starts_with("HTTP/1.1 200"), "got: {response}");

    connected.shutdown().await;
    serving.shutdown().await;
}

// ── The asymmetry ────────────────────────────────────────────────────────

/// Half of the product's rotation story, and the half only two live sides
/// can demonstrate: **the token rotates in place**. The ticket does not
/// change, the pairing stays up, and the next request needs the new value.
#[tokio::test]
async fn rotating_the_token_leaves_the_ticket_and_the_live_pairing_intact() {
    let backend = MockBackend::json(200, OK_BODY).await;
    let (serving, connected, url) = paired(&backend, TokenPolicy::Generate).await;

    let ticket_before = serving.ticket().to_string();
    let old = bearer(&serving);
    assert!(
        within("first request", request(&url, "/v1/models", Some(&old)))
            .await
            .expect("request")
            .starts_with("HTTP/1.1 200")
    );

    let fresh = serving.rotate_token();
    assert_eq!(
        serving.ticket().to_string(),
        ticket_before,
        "rotating a token must not disturb the ticket"
    );

    let refused = within("old credential", request(&url, "/v1/models", Some(&old)))
        .await
        .expect("request");
    assert!(
        refused.starts_with("HTTP/1.1 401"),
        "the old token dies immediately: {refused}"
    );

    let accepted = within(
        "new credential",
        request(&url, "/v1/models", Some(&format!("Bearer {fresh}"))),
    )
    .await
    .expect("request");
    assert!(
        accepted.starts_with("HTTP/1.1 200"),
        "and the same pairing carries the new one: {accepted}"
    );

    connected.shutdown().await;
    serving.shutdown().await;
}

/// `set_token` is how a `Supplied` embedder propagates a rotation of a key
/// its own backend already knows.
#[tokio::test]
async fn a_supplied_credential_can_be_replaced_in_place() {
    let backend = MockBackend::json(200, OK_BODY).await;
    let (serving, connected, url) =
        paired(&backend, TokenPolicy::Supplied("first-key".to_owned())).await;

    assert_eq!(serving.token().as_deref(), Some("first-key"));
    serving
        .set_token("second-key".to_owned())
        .expect("a usable token is installed");

    assert!(
        within("old", request(&url, "/v1/models", Some("Bearer first-key")))
            .await
            .expect("request")
            .starts_with("HTTP/1.1 401")
    );
    assert!(
        within(
            "new",
            request(&url, "/v1/models", Some("Bearer second-key"))
        )
        .await
        .expect("request")
        .starts_with("HTTP/1.1 200")
    );

    connected.shutdown().await;
    serving.shutdown().await;
}

/// `grant_once` is the pairing primitive: one request bearing the code gets
/// through the edge, the next one bearing it does not, and the token the
/// listener enforces is unaffected throughout.
#[tokio::test]
async fn a_grant_admits_one_request_through_a_live_pipe_and_then_none() {
    let backend = MockBackend::json(200, OK_BODY).await;
    let (serving, connected, url) =
        paired(&backend, TokenPolicy::Supplied("the-real-key".to_owned())).await;

    serving
        .grant_once("483920".to_owned(), Duration::from_mins(2))
        .expect("a presentable code is granted");

    let first = within(
        "the code admits once",
        request(&url, "/v1/models", Some("Bearer 483920")),
    )
    .await
    .expect("request");
    assert!(first.starts_with("HTTP/1.1 200"), "got: {first}");

    let second = within(
        "the spent code is a wrong token",
        request(&url, "/v1/models", Some("Bearer 483920")),
    )
    .await
    .expect("request");
    assert!(second.starts_with("HTTP/1.1 401"), "got: {second}");

    let token = within(
        "the real key still works",
        request(&url, "/v1/models", Some("Bearer the-real-key")),
    )
    .await
    .expect("request");
    assert!(token.starts_with("HTTP/1.1 200"), "got: {token}");
    assert_eq!(serving.token().as_deref(), Some("the-real-key"));
    assert_eq!(
        backend.accepts(),
        2,
        "the refusal never reached the backend"
    );

    connected.shutdown().await;
    serving.shutdown().await;
}

/// The other half: **restarting the listener rotates the ticket**, and the
/// old one does not merely fail authentication — it reaches nobody, because
/// the endpoint key is ephemeral and the restarted process is a different
/// endpoint entirely.
#[tokio::test]
async fn restarting_the_listener_mints_a_ticket_the_old_one_cannot_impersonate() {
    let backend = MockBackend::json(200, OK_BODY).await;

    let first = within(
        "serve",
        Box::pin(modelpipe::serve(&backend.url, ServeOptions::default())),
    )
    .await
    .expect("serve");
    let old_ticket = first.ticket().to_string();
    first.shutdown().await;
    drop(first);

    let second = within(
        "serve again",
        Box::pin(modelpipe::serve(&backend.url, ServeOptions::default())),
    )
    .await
    .expect("serve");
    let new_ticket = second.ticket().to_string();

    assert_ne!(
        old_ticket, new_ticket,
        "a restart must mint a different ticket"
    );
    let old: Ticket = old_ticket.parse().expect("the old ticket still parses");
    let new: Ticket = new_ticket.parse().expect("parses");
    assert_ne!(
        old.fingerprint(),
        new.fingerprint(),
        "and a different identity, not merely different addresses"
    );

    second.shutdown().await;
}

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

/// The product is a token stream. A buffering pipe would return the same
/// bytes with the same status and pass every test above.
#[tokio::test]
async fn a_streaming_response_arrives_as_it_is_produced() {
    let backend =
        MockBackend::streaming(&["data: one\n\n", "data: two\n\n", "data: [DONE]\n\n"]).await;
    let (serving, connected, url) = paired(&backend, TokenPolicy::Generate).await;

    let started = std::time::Instant::now();
    let response = within(
        "the stream must complete",
        request(&url, "/v1/chat/completions", Some(&bearer(&serving))),
    )
    .await
    .expect("request");

    assert!(response.contains("data: one"), "got: {response}");
    assert!(response.contains("data: [DONE]"), "got: {response}");
    assert!(
        started.elapsed() >= Duration::from_millis(60),
        "the backend paused between frames, so a response that arrived \
         instantly would mean the frames were produced before being sent"
    );

    connected.shutdown().await;
    serving.shutdown().await;
}

// ── Status and teardown ──────────────────────────────────────────────────

#[tokio::test]
async fn a_shutdown_pipe_reports_closed_and_never_blocks_a_watcher() {
    let backend = MockBackend::json(200, OK_BODY).await;
    let (serving, connected, _url) = paired(&backend, TokenPolicy::Generate).await;

    serving.shutdown().await;
    assert_eq!(serving.status(), PipeStatus::Closed);
    assert_eq!(
        within(
            "a closed pipe must not block a watcher",
            serving.status_changed()
        )
        .await,
        PipeStatus::Closed
    );

    connected.shutdown().await;
    assert_eq!(connected.status(), PipeStatus::Closed);
}

/// The two questions a client has to be able to answer apart, over a real
/// pairing: **is this pipe still trying, and did it end because I asked?**
///
/// `status` alone answers neither. A live connect side reads `Idle` while
/// it looks for a peer that went away, and a dead one reads `Closed`
/// whether a caller ended it or the local listener died — so a client
/// rendering the status alone shows "not connected" for a success and for a
/// failure alike, which is the gap the reason exists to close.
#[tokio::test]
async fn a_connect_side_says_whether_it_is_still_trying_and_why_it_stopped() {
    let backend = MockBackend::json(200, OK_BODY).await;
    let (serving, connected, _url) = paired(&backend, TokenPolicy::Generate).await;

    assert_eq!(
        connected.close_reason(),
        None,
        "a live pipe has not closed, so there is nothing to explain"
    );

    // The serve side goes away. This is the case a client must NOT read as
    // a close: the connect side is looking for it and would pick it up
    // again, so the status drops to `Idle` and the reason stays `None`.
    serving.shutdown().await;
    settles_on(&connected, PipeStatus::Idle).await;
    assert_eq!(
        connected.close_reason(),
        None,
        "a peer that went away has not closed this side, and it is still trying"
    );

    connected.shutdown().await;
    assert_eq!(connected.status(), PipeStatus::Closed);
    assert_eq!(
        connected.close_reason(),
        Some(CloseReason::Shutdown),
        "and a close this caller asked for is named as theirs"
    );
    assert_eq!(
        connected.close_reason().map(CloseReason::as_str),
        Some("shutdown")
    );
}

/// `shutdown` completing must mean the port is free, not merely that the
/// status says `Closed` — otherwise a caller that rebinds immediately gets
/// `EADDRINUSE`.
#[tokio::test]
async fn a_completed_shutdown_releases_the_local_port() {
    let backend = MockBackend::json(200, OK_BODY).await;
    let (serving, connected, _url) = paired(&backend, TokenPolicy::Generate).await;
    let port = connected.local_addr();

    connected.shutdown().await;
    drop(connected);

    tokio::net::TcpListener::bind(port)
        .await
        .expect("the port must be free the moment shutdown returns");

    serving.shutdown().await;
}

#[tokio::test]
async fn shutting_down_twice_is_harmless() {
    let backend = MockBackend::json(200, OK_BODY).await;
    let (serving, connected, _url) = paired(&backend, TokenPolicy::Generate).await;

    serving.shutdown().await;
    within("the second call must not hang", serving.shutdown()).await;
    connected.shutdown().await;
    within("nor on the connect side", connected.shutdown()).await;
}

// ── Teardown, observed rather than announced ─────────────────────────────

/// `shutdown` drains rather than cuts, and the only way to see the
/// difference is to have something in flight while it runs.
///
/// Every teardown assertion before this one checked that the status became
/// `Closed` — which `lifecycle.close()` sets with no transport involved —
/// so reducing `listener::shutdown` to `close(); mark_torn_down();` left
/// the whole suite green. Measured before the order was corrected: the
/// client was cut at frame 5 of 200.
#[tokio::test]
async fn a_serve_shutdown_lets_an_admitted_request_finish() {
    let backend = MockBackend::streaming(&[
        "data: one\n\n",
        "data: two\n\n",
        "data: three\n\n",
        "data: [DONE]\n\n",
    ])
    .await;
    let (serving, connected, url) = paired(&backend, TokenPolicy::Generate).await;
    let auth = bearer(&serving);
    let authority = url
        .trim_start_matches("http://")
        .trim_end_matches("/v1")
        .to_owned();

    // Start the request and wait until the first frame has arrived, so the
    // exchange is provably admitted and provably unfinished.
    let reading = tokio::spawn(async move {
        use tokio::io::{AsyncReadExt, AsyncWriteExt};
        let mut socket = tokio::net::TcpStream::connect(&authority)
            .await
            .expect("connect");
        socket
            .write_all(
                format!(
                    "GET /v1/chat/completions HTTP/1.1\r\nHost: x\r\n\
                     Authorization: {auth}\r\n\r\n"
                )
                .as_bytes(),
            )
            .await
            .expect("write");
        let mut seen = Vec::new();
        socket.read_to_end(&mut seen).await.expect("read");
        String::from_utf8_lossy(&seen).into_owned()
    });
    tokio::time::sleep(Duration::from_millis(40)).await;

    within("the drain must not hang", serving.shutdown()).await;

    let body = within("the admitted request must complete", reading)
        .await
        .expect("reader");
    assert!(
        body.contains("data: [DONE]"),
        "shutdown promises the drain, so an admitted request runs to \
         completion; the client got: {body}"
    );

    connected.shutdown().await;
}

/// One accepted-but-silent TCP connection must not hold the drain open.
///
/// This is what every `OpenAI` SDK does on its first call — open the socket,
/// then think — and what any health probe does deliberately. The in-flight
/// guard used to be taken at accept, and `copy_bidirectional` never returns
/// for a socket that says nothing, so a single one wedged `shutdown`
/// permanently. In the CLI that is unrecoverable: tokio keeps the SIGINT
/// handler installed, so the second Ctrl-C is swallowed too.
#[tokio::test]
async fn an_idle_local_connection_does_not_wedge_the_connect_side_drain() {
    let backend = MockBackend::json(200, OK_BODY).await;
    let (serving, connected, url) = paired(&backend, TokenPolicy::Generate).await;
    let authority = url
        .trim_start_matches("http://")
        .trim_end_matches("/v1")
        .to_owned();

    let _idle = tokio::net::TcpStream::connect(&authority)
        .await
        .expect("an SDK preconnect");
    tokio::time::sleep(Duration::from_millis(50)).await;

    within(
        "one silent connection must not hold the drain open",
        connected.shutdown(),
    )
    .await;

    serving.shutdown().await;
}

/// `shutdown_timeout` returning must mean the port is free, exactly as
/// `shutdown` does — and it must leave a later `shutdown` able to say the
/// same. It used to set the teardown latch itself while the accept loop
/// still owned the listener, so it returned `true` with the port bound and
/// poisoned the latch for every call after it.
#[tokio::test]
async fn a_connect_shutdown_timeout_releases_the_port_and_leaves_the_latch_honest() {
    let backend = MockBackend::json(200, OK_BODY).await;
    let (serving, connected, _url) = paired(&backend, TokenPolicy::Generate).await;
    let port = connected.local_addr();

    let drained = within(
        "nothing is in flight, so the drain must succeed",
        connected.shutdown_timeout(Duration::from_secs(5)),
    )
    .await;
    assert!(drained, "there was nothing to wait for");
    tokio::net::TcpListener::bind(port)
        .await
        .expect("the port must be free the moment shutdown_timeout returns");

    // And the promise survives: a later `shutdown` must not resolve against
    // a latch someone else already set.
    within("a second call must not hang", connected.shutdown()).await;
    serving.shutdown().await;
}

/// A connect-side `shutdown` must *tell* the serve side, not leave it to
/// time out.
///
/// The bound is the whole assertion, and it is the only place in this file
/// that puts one on the departure.
/// `a_live_pairing_reports_a_transport_path_on_both_sides` also waits for
/// the set to empty, but under `within`'s twenty seconds — which QUIC's
/// idle timeout fits comfortably inside, so a pipe that told the far side
/// nothing at all would pass it.
///
/// This is the property, not the mechanism, and the honest limit is worth
/// stating: the mechanism it was written for — an endpoint dropped rather
/// than closed, aborting the driver before the `CONNECTION_CLOSE` frame
/// escapes — cannot be reproduced in one process, because both endpoints
/// share a live runtime here and the queued frame goes out regardless.
/// `peer_tests.rs` asserts the mechanism on the socket itself; this asserts
/// what an operator on the other machine actually sees.
#[tokio::test]
async fn a_connect_shutdown_is_announced_rather_than_left_to_the_idle_timeout() {
    let backend = MockBackend::json(200, OK_BODY).await;
    let (serving, connected, url) = paired(&backend, TokenPolicy::Generate).await;

    // A request first, so the peer is provably registered before the
    // teardown that has to unregister it — the registration happens on the
    // serve side's accept task, which `carrying` does not wait for.
    within(
        "a request must cross the pipe",
        request(&url, "/v1/models", Some(&bearer(&serving))),
    )
    .await
    .expect("request");
    assert_eq!(serving.peers().len(), 1, "the peer is registered");

    connected.shutdown().await;

    // Two seconds against an idle timeout of fifteen at the very least: the
    // close frame either escaped or it did not, and no slow machine turns
    // fifteen into two.
    let noticed = tokio::time::timeout(Duration::from_secs(2), async {
        while !serving.peers().is_empty() {
            tokio::time::sleep(Duration::from_millis(10)).await;
        }
    })
    .await;
    assert!(
        noticed.is_ok(),
        "the serve side was never told, and still lists {:?}",
        serving.peers()
    );

    serving.shutdown().await;
}

/// A live pairing reports the path it is actually using, on both sides.
///
/// The connect side published no status at all: `Direct` and `Relayed` were
/// unreachable there, so `status()` said `Idle` on a working pipe and
/// `status_changed()` never fired. Deleting the serve side's peer
/// registration — the crate's only other producer — also left the suite
/// green, because every other status assertion checks only `Closed`.
#[tokio::test]
async fn a_live_pairing_reports_a_transport_path_on_both_sides() {
    let backend = MockBackend::json(200, OK_BODY).await;
    let (serving, connected, url) = paired(&backend, TokenPolicy::Generate).await;

    within(
        "a request must cross the pipe",
        request(&url, "/v1/models", Some(&bearer(&serving))),
    )
    .await
    .expect("request");

    for (side, status) in [("serve", serving.status()), ("connect", connected.status())] {
        assert!(
            matches!(status, PipeStatus::Direct | PipeStatus::Relayed),
            "the {side} side is carrying traffic and reports {status:?}"
        );
    }

    // The per-peer view names the one peer and agrees with the aggregate.
    let peers = serving.peers();
    assert_eq!(peers.len(), 1, "one connect side is paired: {peers:?}");
    assert_eq!(
        peers[0].path,
        serving.status(),
        "one peer: the aggregate is it"
    );
    assert_eq!(peers[0].fingerprint.len(), 12);
    assert!(peers[0].fingerprint.chars().all(|c| c.is_ascii_hexdigit()));

    connected.shutdown().await;
    // The peer's departure is noticed asynchronously; wait for the set to
    // say so rather than asserting a race.
    within("the peer leaves the set", async {
        while !serving.peers().is_empty() {
            tokio::time::sleep(Duration::from_millis(10)).await;
        }
    })
    .await;
    serving.shutdown().await;
}

// ── Cancellation ─────────────────────────────────────────────────────────

/// The failure that is invisible to every other test in this file.
///
/// A client that hangs up mid-generation must take the backend's work with
/// it. If it does not, the model keeps producing tokens for a request
/// nobody is waiting for — and every functional assertion still passes,
/// because the request "worked". The only way to see it is to count what
/// the backend produced after the client left.
#[tokio::test]
async fn a_client_that_disconnects_mid_stream_stops_the_backend() {
    let (backend, frames_written) = MockBackend::endless_stream().await;
    let (serving, connected, url) = paired(&backend, TokenPolicy::Generate).await;

    let authority = url
        .trim_start_matches("http://")
        .trim_end_matches("/v1")
        .to_owned();
    let auth = bearer(&serving);

    // Open a request, read enough to know the stream is flowing, then hang
    // up without reading the rest.
    {
        let mut socket = tokio::net::TcpStream::connect(&authority)
            .await
            .expect("connect");
        let request = format!(
            "GET /v1/chat/completions HTTP/1.1\r\nHost: {authority}\r\n\
             Authorization: {auth}\r\n\r\n"
        );
        tokio::io::AsyncWriteExt::write_all(&mut socket, request.as_bytes())
            .await
            .expect("write");

        let mut seen = vec![0u8; 64];
        within(
            "the stream must start",
            tokio::io::AsyncReadExt::read(&mut socket, &mut seen),
        )
        .await
        .expect("read");
        // Dropped here: the client is gone mid-generation.
    }

    // Let the news travel, then see whether the backend is still producing.
    tokio::time::sleep(Duration::from_millis(300)).await;
    let after_disconnect = frames_written.load(std::sync::atomic::Ordering::SeqCst);
    tokio::time::sleep(Duration::from_millis(300)).await;
    let later = frames_written.load(std::sync::atomic::Ordering::SeqCst);

    assert_eq!(
        later,
        after_disconnect,
        "the backend produced {} more frames after the client left; a \
         cancelled request must not leave a generation running",
        later - after_disconnect
    );

    connected.shutdown().await;
    serving.shutdown().await;
}

// ── Connection reuse ─────────────────────────────────────────────────────

/// One bi-stream carries one exchange, so a client must not put a second
/// request on the same local connection — it would go down a stream the
/// serve side has finished with, and hang until the client's timeout.
///
/// Real `OpenAI` clients pool connections by default, so this is not an edge
/// case: it is what the first SDK to point at modelpipe would do. Telling
/// the client is the whole mechanism, and it is one header.
#[tokio::test]
async fn a_response_tells_the_client_not_to_reuse_the_connection() {
    let backend = MockBackend::json(200, OK_BODY).await;
    let (serving, connected, url) = paired(&backend, TokenPolicy::Generate).await;

    let response = within(
        "a request must cross the pipe",
        request(&url, "/v1/models", Some(&bearer(&serving))),
    )
    .await
    .expect("request");

    assert!(
        response.to_ascii_lowercase().contains("connection: close"),
        "a pooling client will otherwise send its next request down a \
         stream nobody is reading: {response}"
    );

    connected.shutdown().await;
    serving.shutdown().await;
}

/// A listener restarted with a stored identity keeps the ticket it had.
///
/// The other half of the asymmetry, and the one that was not previously
/// available at any price. `restarting_the_listener_mints_a_ticket_the_old
/// _one_cannot_impersonate` above pins the default — a fresh key per
/// process, so a restart re-pairs every device — and this pins the opt-out.
///
/// What is compared is the fingerprint, which is the identity and nothing
/// else — the addresses beside it in a ticket are hints for avoiding the
/// relay, and a restarted process holds a different UDP port regardless.
/// It is a prefix rather than the whole key because that is what the public
/// surface offers, and it is the value a person compares by eye for exactly
/// this question; the full-key form of the claim is
/// `the_same_key_binds_to_the_same_endpoint_and_a_different_one_does_not`
/// in `transport_tests.rs`, where the bytes are reachable.
///
/// Reaching the restarted listener's *new port* with the old ticket is then
/// iroh's discovery doing its job, over a network this suite deliberately
/// does not require. The claim owned here is the one this crate can be
/// wrong about: that the key comes back, and the ticket still names this
/// listener.
#[tokio::test]
async fn a_listener_restarted_with_a_stored_identity_keeps_its_ticket() {
    let backend = MockBackend::json(200, OK_BODY).await;
    let scratch = Scratch::new("identity");
    let key = scratch.join("key");

    let mut first = ServeOptions::default();
    first.identity = Some(key.clone());
    let before = within(
        "serve must bind",
        Box::pin(modelpipe::serve(&backend.url, first)),
    )
    .await
    .expect("serve");
    let ticket_before = before.ticket();
    before.shutdown().await;

    let mut second = ServeOptions::default();
    second.identity = Some(key.clone());
    let after = within(
        "the restarted listener must bind",
        Box::pin(modelpipe::serve(&backend.url, second)),
    )
    .await
    .expect("serve");
    let ticket_after = after.ticket();

    assert_eq!(
        ticket_before.fingerprint(),
        ticket_after.fingerprint(),
        "a stored identity is what makes a ticket outlive the process"
    );

    after.shutdown().await;
}

/// The control, and the promise that the default has not quietly changed:
/// without a stored identity the restarted listener is a different peer, as
/// it has always been.
#[tokio::test]
async fn a_listener_restarted_without_one_is_a_different_peer_as_before() {
    let backend = MockBackend::json(200, OK_BODY).await;

    let before = within(
        "serve must bind",
        Box::pin(modelpipe::serve(&backend.url, ServeOptions::default())),
    )
    .await
    .expect("serve");
    let ticket_before = before.ticket();
    before.shutdown().await;

    let after = within(
        "serve must bind again",
        Box::pin(modelpipe::serve(&backend.url, ServeOptions::default())),
    )
    .await
    .expect("serve");

    assert_ne!(
        ticket_before.fingerprint(),
        after.ticket().fingerprint(),
        "the default stays ephemeral, which is the revocation the README sells"
    );

    after.shutdown().await;
}

/// An identity file the operator cannot use stops the listener before it
/// starts, rather than after — which would mean finding out as a ticket
/// that is not the one they expected, on a listener already accepting.
#[tokio::test]
async fn an_unusable_identity_refuses_to_serve_at_all() {
    let backend = MockBackend::json(200, OK_BODY).await;
    let scratch = Scratch::new("bad-identity");
    let key = scratch.join("key");
    std::fs::write(&key, "not a key\n").expect("write");

    let mut opts = ServeOptions::default();
    opts.identity = Some(key);
    let refused = within(
        "serve must refuse rather than hang",
        Box::pin(modelpipe::serve(&backend.url, opts)),
    )
    .await;

    let Err(refused) = refused else {
        panic!("an unusable identity must not start a listener");
    };
    assert!(!refused.is_retryable(), "the operator named this path");
    assert_eq!(backend.accepts(), 0, "and nothing was served");
}

/// A connect side whose peer has gone says so, and goes on looking.
///
/// Before this it did neither. `dial` opened one connection and held it for
/// life, so a serve side that went away left the client machine answering
/// 502 to every request while `status()` still read `direct` — the one
/// place a user could have found out, saying the opposite of the truth.
/// Measured: the serve process killed, the connect process left running,
/// still `direct` and still 502ing with no reconnection ever attempted.
///
/// `Idle` is what `ConnectHandle`'s own documentation has always promised
/// for this and no code could reach. Note what is *not* asserted: that the
/// pipe comes back. It cannot here — the endpoint key is minted per
/// process, so a restarted listener is a different peer that this ticket
/// has no relation to. Surviving a restart is a re-pairing, and needs an
/// identity that outlives the process.
#[tokio::test]
async fn a_connect_side_whose_peer_goes_away_reports_idle_rather_than_pretending() {
    let backend = MockBackend::json(200, OK_BODY).await;
    let (serving, connected, url) = paired(&backend, TokenPolicy::Generate).await;

    // A working pipe first, so the status this starts from is a real one.
    assert!(
        matches!(connected.status(), PipeStatus::Direct | PipeStatus::Relayed),
        "a live pairing reports the path it is using: {:?}",
        connected.status()
    );

    serving.shutdown().await;

    within(
        "the connect side must notice its peer has gone",
        settles_on(&connected, PipeStatus::Idle),
    )
    .await;
    assert_ne!(
        connected.status(),
        PipeStatus::Closed,
        "and this side is still up and still looking, not gone"
    );

    // The client is told, rather than left holding a socket that never
    // answers.
    let refused = within(
        "a request with no peer must still be answered",
        request(&url, "/v1/models", Some("Bearer whatever")),
    )
    .await
    .expect("request");
    assert!(refused.starts_with("HTTP/1.1 502"), "got: {refused}");
    // *Which* 502, which the status line cannot say. The three share a
    // status on purpose — a client's recovery is the same in each case —
    // so the body is the only thing that tells a person whether to look at
    // their model server or at the machine it runs on. This one is written
    // on the client's own machine about a peer that is not there, and
    // borrowing either sentence about a backend would name a component
    // that is not in the picture.
    assert!(
        refused.contains(r#""code":"tunnel_unavailable""#),
        "the connect side must say the tunnel is down, not blame a backend: {refused}"
    );
    assert!(
        !refused.contains("backend"),
        "there is no backend in this failure: {refused}"
    );
    connected.shutdown().await;
}

/// A rotation that cannot be presented is refused *and reported*, with the
/// credential already in force left exactly where it was.
///
/// The silent version of this is the dangerous one, and it is the one that
/// shipped: a rotation reads its replacement from somewhere — a config
/// file, a secrets fetch, an environment variable — and when that somewhere
/// comes back blank an embedder who is told nothing believes the old key is
/// dead and retires it everywhere else, while this listener goes on
/// accepting it. A credential the operator thinks is revoked and is not.
/// `serve` has always refused the same value loudly.
///
/// Its negative control is `a_supplied_credential_can_be_replaced_in_place`
/// above: that one proves a usable token really does displace the old one,
/// so this cannot pass by `set_token` having stopped working at all.
#[tokio::test]
async fn a_refused_rotation_reports_it_and_leaves_the_previous_credential_in_force() {
    let backend = MockBackend::json(200, OK_BODY).await;
    let (serving, connected, url) =
        paired(&backend, TokenPolicy::Supplied("the-only-key".to_owned())).await;

    for blank in ["", "   ", "\t\n"] {
        assert!(
            serving.set_token(blank.to_owned()).is_err(),
            "{blank:?} is a credential no conforming client could ever send"
        );
    }

    assert_eq!(
        serving.token().as_deref(),
        Some("the-only-key"),
        "the handle still reports what it is actually enforcing"
    );
    assert!(
        within(
            "the key the operator may now believe is dead",
            request(&url, "/v1/models", Some("Bearer the-only-key")),
        )
        .await
        .expect("request")
        .starts_with("HTTP/1.1 200"),
        "and it is still the key that works"
    );

    connected.shutdown().await;
    serving.shutdown().await;
}

/// An upload that stops mid-body must not hold the drain open.
///
/// The unit tests pin the answer the edge gives; this pins the consequence
/// that made it a release blocker. A wedged exchange never releases its
/// in-flight guard, and `shutdown` waits on precisely that — so one aborted
/// upload made the first Ctrl-C on `modelpipe serve` hang while the second
/// cut the pipe, taking every other request with it.
///
/// Measured, before the two halves were told apart: still running at twenty
/// seconds, against one second for the same shutdown with only ordinary
/// traffic in flight. Its negative control is
/// `a_serve_shutdown_lets_an_admitted_request_finish` above — that one
/// proves the drain still waits for work genuinely in progress, so this one
/// cannot pass by `shutdown` having been reduced to a cut.
#[tokio::test]
async fn an_aborted_upload_does_not_wedge_the_serve_side_drain() {
    let backend = MockBackend::reads_whole_body(
        "HTTP/1.1 200 OK\r\nContent-Length: 2\r\nConnection: close\r\n\r\n{}",
    )
    .await;
    let (serving, connected, url) = paired(&backend, TokenPolicy::Generate).await;
    let authority = url
        .trim_start_matches("http://")
        .trim_end_matches("/v1")
        .to_owned();

    let mut socket = tokio::net::TcpStream::connect(&authority)
        .await
        .expect("a client");
    let request = format!(
        "POST /v1/chat/completions HTTP/1.1\r\nHost: {authority}\r\n\
         Authorization: {}\r\nContent-Length: 1000\r\n\r\n{{\"model\":\"",
        bearer(&serving)
    );
    socket
        .write_all(request.as_bytes())
        .await
        .expect("the head and a tenth of the body");
    socket.flush().await.expect("flush");
    // Half-close, not a full one: the upload is over and the client is
    // still listening, which is what an interrupted `curl -d @file` leaves
    // behind.
    socket.shutdown().await.expect("half-close");

    let reader = tokio::spawn(async move {
        let mut seen = Vec::new();
        let _ = socket.read_to_end(&mut seen).await;
        String::from_utf8_lossy(&seen).into_owned()
    });
    // Long enough for the exchange to be admitted and registered in flight,
    // which is what makes this a test of the drain rather than of an empty
    // one.
    tokio::time::sleep(Duration::from_millis(50)).await;

    within(
        "an aborted upload must not hold the serve-side drain open",
        serving.shutdown(),
    )
    .await;

    let seen = reader.await.expect("the reader task");
    assert!(
        seen.starts_with("HTTP/1.1 400"),
        "and the client is told, rather than left with an empty stream: {seen}"
    );
    connected.shutdown().await;
}

// ── What an embedder can ask of a live pipe ──────────────────────────────

/// The transition `status()` + `status_changed()` cannot see, seen.
///
/// This is the whole reason `status_changed_since` exists, and it is the
/// one claim about it that needs two live sides: the window it closes is
/// between a caller reading the status and going back to waiting, and only
/// a real pipe moves on its own inside that window.
///
/// The sequence below arranges the window deliberately rather than racing
/// for it — the peer is taken away and the transition is *waited out* — so
/// what is being asserted is a property of the two methods rather than the
/// timing of the machine running them. `status_changed` snapshots inside
/// itself, so by the time it is called there is nothing left to report and
/// it parks; `status_changed_since` is handed the value that was rendered,
/// so it answers at once. Both halves are asserted, because either alone
/// would pass against a method that simply always returned immediately.
#[tokio::test]
async fn a_transition_that_lands_before_the_next_wait_is_reported_rather_than_lost() {
    let backend = MockBackend::json(200, OK_BODY).await;
    let (serving, connected, _url) = paired(&backend, TokenPolicy::Generate).await;

    // What a caller would have rendered: a live path, read once.
    let rendered = connected.status();
    assert!(
        matches!(rendered, PipeStatus::Direct | PipeStatus::Relayed),
        "a live pairing reports the path it is using: {rendered:?}"
    );

    // Now the pipe moves, and the move completes while nobody is waiting.
    serving.shutdown().await;
    within(
        "the connect side must notice its peer has gone",
        settles_on(&connected, PipeStatus::Idle),
    )
    .await;

    // The coalescing form has nothing left to say and waits for a further
    // change that is not coming — a peer that is gone stays gone, and iroh
    // spends about thirty seconds on each dial after it.
    assert!(
        tokio::time::timeout(Duration::from_millis(200), connected.status_changed())
            .await
            .is_err(),
        "status_changed snapshots at the call, so the transition is already behind it"
    );

    // The same fact, asked with the value that was rendered.
    let seen = within(
        "a caller holding its own snapshot must be told what it missed",
        connected.status_changed_since(rendered),
    )
    .await;
    assert_eq!(
        seen,
        Some(PipeStatus::Idle),
        "the transition happened, and this is the form that reports it"
    );

    connected.shutdown().await;
}

/// The same window, on the other side of the pipe.
///
/// Mirrored rather than shared, and not for symmetry's sake. The two
/// handles are separate types carrying separate bodies of one contract,
/// so the property that separates `status_changed_since` from
/// `status_changed` — that the snapshot comes from the *caller* — is a
/// claim about each body on its own. Measured: with the serve side's
/// method rewritten to snapshot inside itself, exactly as the coalescing
/// form does, the whole suite still passed. That is the entire defect this
/// pair of methods exists to close, and it survived on one side.
///
/// `a_watcher_carrying_its_last_value_forward_ends_when_the_pipe_does`
/// does watch both sides, and cannot see this: a watcher whose own held
/// value has reached `Closed` still ends under the coalescing body, so it
/// passes either way.
///
/// The sequence is the connect-side test's, run the other way round — the
/// *peer* is taken away rather than the listener, because a listener that
/// shut itself down is closed and has no `Idle` to report.
#[tokio::test]
async fn the_serve_side_reports_a_transition_that_lands_before_its_next_wait_too() {
    let backend = MockBackend::json(200, OK_BODY).await;
    let (serving, connected, _url) = paired(&backend, TokenPolicy::Generate).await;

    // `paired` waits on the *connect* side's view of the pairing. Each side
    // publishes its own, and what this test renders has to be there first.
    within(
        "the serve side must see the peer it is carrying",
        until(|| serving.status() != PipeStatus::Idle),
    )
    .await;

    // What a caller would have rendered: a live path, read once.
    let rendered = serving.status();
    assert!(
        matches!(rendered, PipeStatus::Direct | PipeStatus::Relayed),
        "a live pairing reports the path it is carrying: {rendered:?}"
    );

    // Now the pipe moves, and the move completes while nobody is waiting.
    connected.shutdown().await;
    within(
        "the serve side must notice its last peer has gone",
        until(|| serving.status() == PipeStatus::Idle),
    )
    .await;

    // The coalescing form has nothing left to say: the peer that left is
    // not coming back, and no second change is on its way.
    assert!(
        tokio::time::timeout(Duration::from_millis(200), serving.status_changed())
            .await
            .is_err(),
        "status_changed snapshots at the call, so the transition is already behind it"
    );

    // The same fact, asked with the value that was rendered.
    let seen = within(
        "a caller holding its own snapshot must be told what it missed",
        serving.status_changed_since(rendered),
    )
    .await;
    assert_eq!(
        seen,
        Some(PipeStatus::Idle),
        "the peer left while nobody waited, and this is the form that reports it"
    );

    serving.shutdown().await;
}

/// The loop a language binding writes, run over a real pipe until it ends.
///
/// `Closed` is terminal, so a watcher that carries its last value forward
/// would be handed `Closed` again for ever, immediately, with no await
/// anywhere in the path — a torn-down pipe costing a core until the app is
/// killed.
///
/// **Bounded rather than timed, and the bound is the assertion.** A loop
/// with no await in it starves the runtime it is on, timers included, so a
/// timeout around this hangs the suite instead of failing it — measured,
/// not assumed. Counting the turns is what turns that defect into a
/// message.
///
/// Both sides, because they are two implementations of one contract and
/// the prose describing that contract has drifted between them before.
#[tokio::test]
async fn a_watcher_carrying_its_last_value_forward_ends_when_the_pipe_does() {
    let backend = MockBackend::json(200, OK_BODY).await;
    let (serving, connected, _url) = paired(&backend, TokenPolicy::Generate).await;
    // Shared rather than moved: the watchers have to outlive the call that
    // ends the pipe, and both handles take `&self` for teardown precisely
    // so an embedder can hold them like this.
    let serving = std::sync::Arc::new(serving);
    let connected = std::sync::Arc::new(connected);

    /// Carry the last value forward until the sequence ends, and report
    /// where it ended.
    macro_rules! watch {
        ($handle:expr) => {{
            let handle = $handle.clone();
            tokio::spawn(async move {
                // Idle, a path, and the close is three; ten leaves room for
                // a pipe that flaps once and still names a repeat for what
                // it is.
                let mut turns = 0;
                let mut held = handle.status();
                let mut last = held;
                while let Some(next) = handle.status_changed_since(held).await {
                    last = next;
                    held = next;
                    turns += 1;
                    assert!(turns <= 10, "the sequence has to end rather than repeat");
                }
                last
            })
        }};
    }
    let serve_watch = watch!(serving);
    let connect_watch = watch!(connected);

    // Both are parked on live pipes; ending the pipes is what has to end
    // them, and nothing else in this test will.
    tokio::time::sleep(Duration::from_millis(50)).await;
    connected.shutdown().await;
    serving.shutdown().await;

    assert_eq!(
        within(
            "the connect side's watcher must end with its pipe",
            connect_watch
        )
        .await
        .expect("the watcher must not panic"),
        PipeStatus::Closed,
        "the last value a watcher sees is the terminal one"
    );
    assert_eq!(
        within(
            "the serve side's watcher must end with its pipe",
            serve_watch
        )
        .await
        .expect("the watcher must not panic"),
        PipeStatus::Closed
    );
}

/// Telling a live pipe that the network moved does not disturb it.
///
/// The notice itself has no effect this machine can observe — iroh re-reads
/// the interface state and returns early when nothing actually changed, and
/// nothing here changes one — so what is asserted is the property an
/// embedder relies on when it wires this into a resume handler it will call
/// on every foreground: that calling it is free. A request crossing the
/// pipe afterwards is the evidence.
///
/// Both handles, and both before and after, because "free" is a claim about
/// each side's endpoint separately.
#[tokio::test]
async fn telling_both_sides_the_network_moved_leaves_the_pipe_carrying() {
    let backend = MockBackend::json(200, OK_BODY).await;
    let (serving, connected, url) = paired(&backend, TokenPolicy::Generate).await;

    within(
        "the serve side must accept a network-change notice",
        serving.notify_network_change(),
    )
    .await;
    within(
        "and so must the connect side",
        connected.notify_network_change(),
    )
    .await;

    let response = within(
        "a request must still cross the pipe afterwards",
        request(&url, "/v1/models", Some(&bearer(&serving))),
    )
    .await
    .expect("request");
    assert!(response.starts_with("HTTP/1.1 200 OK"), "got: {response}");
    assert!(
        !matches!(connected.status(), PipeStatus::Closed),
        "and the pipe is still up, not closed under the notice"
    );

    connected.shutdown().await;
    serving.shutdown().await;
}

/// The counters are readable from both handles while the pipe is live, and
/// they describe two separate endpoints.
///
/// The rate-limit field is what this accessor exists for and is the one no
/// test can produce — a relay has to decide to throttle, and nothing in
/// this process can make it. So what is pinned here is everything around
/// it: the numbers are readable without a runtime trick, they are each
/// side's own, and nothing that happens to a healthy pipe counts as a
/// throttle. A pipe that reported a rate limit it had not been given would
/// send an operator looking at the wrong machine.
#[tokio::test]
async fn both_sides_report_their_own_transport_counters() {
    let backend = MockBackend::json(200, OK_BODY).await;
    let (serving, connected, url) = paired(&backend, TokenPolicy::Generate).await;

    let response = within(
        "a request must cross the pipe first, so the counters describe a used pipe",
        request(&url, "/v1/models", Some(&bearer(&serving))),
    )
    .await
    .expect("request");
    assert!(response.starts_with("HTTP/1.1 200 OK"), "got: {response}");

    for (side, metrics) in [
        ("serve", serving.network_metrics()),
        ("connect", connected.network_metrics()),
    ] {
        assert_eq!(
            metrics.relay_connections_ratelimited, 0,
            "nothing throttled the {side} side, and it must not say otherwise: {metrics:?}"
        );
        // `Copy`, which is what the doc means by holding one in a UI's own
        // state costing nothing — and what lets two readings be compared.
        let held = metrics;
        assert_eq!(held, metrics);
    }

    connected.shutdown().await;
    serving.shutdown().await;
}

/// Each side counts the relay connection its own endpoint could not make.
///
/// **The reading the test above cannot make.** A healthy pipe leaves every
/// field at zero, so `both_sides_report_their_own_transport_counters` is
/// satisfied by a `network_metrics` that returns `NetworkMetrics::default()`
/// and never looks at an endpoint at all — measured, on both facade methods
/// at once, with the whole suite still passing. Only a counter that has
/// actually moved separates the two, and `metrics_of` being pinned beside
/// itself in `network_tests` does not pin the two public methods that call
/// it.
///
/// **A relay that answers and is not one is what moves it**, and the
/// distinction is iroh's rather than this test's. A URL nothing is behind
/// is never dialled at all: the probe that decides whether a relay is
/// reachable fails first, no relay connection is ever attempted, and the
/// counters stay at zero — which is what `network_tests` observes of its
/// `never` endpoint and why an unreachable relay cannot be used here. A
/// plain HTTP server answering `200` passes that probe and then fails the
/// handshake, so each endpoint counts a relay connection it could not make,
/// retries, and counts another.
///
/// Hermetic, unlike the tests around it: the fake relay is a socket on
/// loopback, the pairing forms over the ticket's direct addresses — the
/// configuration `a_pairing_still_forms_with_discovery_and_port_mapping_off`
/// covers — and nothing here waits on a route to the internet.
///
/// The third listener is what keeps each reading *that handle's*. It knows
/// only a relay nothing is behind, so it never dials one and reports
/// nothing at all, while the two sides beside it count failure after
/// failure — a facade wired to a process-wide number rather than to an
/// endpoint is one this still catches.
#[tokio::test]
async fn each_side_counts_the_relay_connection_its_own_endpoint_could_not_make() {
    let backend = MockBackend::json(200, OK_BODY).await;
    // Answers every request with `200 OK`, which is exactly what a relay
    // must not answer the handshake with: the upgrade to the relay protocol
    // needs `101 Switching Protocols`.
    let fake_relay = MockBackend::json(200, OK_BODY).await;

    let mut serve_opts = ServeOptions::default();
    serve_opts.auth = TokenPolicy::Generate;
    serve_opts.relay = Some(fake_relay.url.clone());
    serve_opts.port_mapping = false;
    serve_opts.discovery = false;
    let serving = within(
        "serve must bind against a relay that is not one",
        Box::pin(modelpipe::serve(&backend.url, serve_opts)),
    )
    .await
    .expect("a relay that does not behave like one is not a startup error");

    let mut connect_opts = ConnectOptions::default();
    connect_opts.relay = Some(fake_relay.url.clone());
    connect_opts.port_mapping = false;
    connect_opts.discovery = false;
    let connected = within(
        "connect must bind against the same one",
        Box::pin(modelpipe::connect(&serving.ticket(), connect_opts)),
    )
    .await
    .expect("connect");
    within(
        "the pairing must still form, on the ticket's direct addresses",
        carrying(&connected),
    )
    .await;

    // Each side, read from its own handle. A default snapshot never arrives
    // at this, because a default is zero for ever.
    within(
        "the serve side must count the relay connection its endpoint could not make",
        until(|| serving.network_metrics().relay_connections_failed > 0),
    )
    .await;
    within(
        "and the connect side must count its own",
        until(|| connected.network_metrics().relay_connections_failed > 0),
    )
    .await;

    // The other half of the same claim: a count of failures is evidence
    // only if the successes stayed where they belong. Nothing on either side
    // reached a relay, and neither may say it did.
    for (side, metrics) in [
        ("serve", serving.network_metrics()),
        ("connect", connected.network_metrics()),
    ] {
        assert_eq!(
            metrics.relay_connections, 0,
            "the {side} side reached no relay, and must not count one: {metrics:?}"
        );
        assert_eq!(
            metrics.relay_connections_ratelimited, 0,
            "a relay that never completed a handshake cannot have throttled \
             the {side} side: {metrics:?}"
        );
    }

    // A listener that knows only a relay nothing is behind — loopback on a
    // port nothing listens on, as `network_tests` names it, so nothing here
    // waits on a resolver. Never dialled and never paired.
    let mut nowhere = ServeOptions::default();
    nowhere.relay = Some("https://127.0.0.1:1/".to_owned());
    nowhere.port_mapping = false;
    nowhere.discovery = false;
    let elsewhere = within(
        "a listener must bind against a relay that is not there",
        Box::pin(modelpipe::serve(&backend.url, nowhere)),
    )
    .await
    .expect("a relay that does not answer is not a startup error");
    assert_eq!(
        elsewhere.network_metrics(),
        NetworkMetrics::default(),
        "an endpoint that dialled nothing has nothing to report, whatever the \
         pipe beside it has been counting: {:?}",
        elsewhere.network_metrics()
    );

    elsewhere.shutdown().await;
    connected.shutdown().await;
    serving.shutdown().await;
}

/// A ticket an embedder is about to print can be asked what it carries.
///
/// The failure this closes is documented at `ServeHandle::ticket` and had
/// no accessor to check it with: the relay is the half that arrives last,
/// so a ticket read the instant `serve` returns can name direct addresses
/// and nothing else — and a machine that cannot be hole-punched to is then
/// unreachable through it. `wait_online` is the switch that waits; this is
/// how an embedder finds out whether it worked.
///
/// **This needs a route to a relay**, and so it must: what is being checked
/// is that the accessor sees one on a ticket a live listener minted, which
/// no ticket built in a test can stand in for. `network_tests` and
/// `transport_tests` carry the same dependency, for the same kind of reason.
/// The unit tests beside `relay_urls` cover its behaviour against the
/// normative vectors and need nothing at all.
#[tokio::test]
async fn a_live_ticket_says_which_paths_it_carries() {
    let backend = MockBackend::json(200, OK_BODY).await;
    let mut serve_opts = ServeOptions::default();
    serve_opts.auth = TokenPolicy::Generate;
    serve_opts.wait_online = Some(Duration::from_secs(20));
    let serving = within(
        "serve must bind",
        Box::pin(modelpipe::serve(&backend.url, serve_opts)),
    )
    .await
    .expect("serve");

    let ticket = serving.ticket();
    assert!(
        !ticket.relay_urls().is_empty(),
        "a listener that waited to come online carries the relay it reached: {ticket:?}"
    );
    for url in ticket.relay_urls() {
        assert!(
            url.starts_with("http"),
            "a relay body is a URL, handed back as written: {url}"
        );
    }
    // The direct addresses are the machine's own interfaces, and this test
    // runs on a machine that has some.
    assert!(
        !ticket.direct_addrs().is_empty(),
        "and the local paths beside it: {ticket:?}"
    );
    // Round-tripping through the printed form is the journey a real ticket
    // makes, and the accessors have to survive it.
    let printed: Ticket = ticket.to_string().parse().expect("its own string parses");
    assert_eq!(printed.relay_urls(), ticket.relay_urls());
    assert_eq!(printed.direct_addrs(), ticket.direct_addrs());

    serving.shutdown().await;
}