//! Unix-socket request server for the per-agent sockets. The socket file's
//! existence on disk authenticates the caller: connecting to
//! `<.../agents/foo/mcp.sock>` means you are `foo`. There is no privilege
//! flag; [`serve`] / [`dispatch`] derive authority uniformly from the caller's
//! identity. An agent-targeting verb may name any agent.

use std::path::{Path, PathBuf};
use std::sync::Arc;

use anyhow::{Context, Result};
use hive_core_agent_sock::{Request, Response};
use hive_sh4re::inbox::Message;
use hive_sock_client::{LineDeadlines, RequestLine, read_request_line};
use tokio::io::{AsyncWriteExt, BufReader};
use tokio::net::{UnixListener, UnixStream};
use tokio::task::JoinHandle;

use crate::coordinator::Coordinator;

pub struct AgentSocket {
    pub path: PathBuf,
    pub handle: JoinHandle<()>,
}

pub fn start(agent: &str, socket_path: &Path, coord: Arc<Coordinator>) -> Result<AgentSocket> {
    use std::os::unix::fs::PermissionsExt as _;
    let agent = agent.to_owned();
    if let Some(parent) = socket_path.parent() {
        std::fs::create_dir_all(parent)
            .with_context(|| format!("create agent socket dir {}", parent.display()))?;
    }
    if socket_path.exists() {
        std::fs::remove_file(socket_path).context("remove stale agent socket")?;
    }
    let listener = UnixListener::bind(socket_path)
        .with_context(|| format!("bind agent socket {}", socket_path.display()))?;
    // The socket is bind-mounted into exactly one container as
    // `/run/hive/mcp.sock` (`lifecycle::set_nspawn_flags`); the
    // in-container harness connects as the per-agent unix user,
    // not root, so the default `tokio::net::UnixListener::bind`
    // perms (0755) lock it out. 0666 lets the agent user connect;
    // the bind source dir is per-agent on host so blast radius is
    // unchanged.
    std::fs::set_permissions(socket_path, std::fs::Permissions::from_mode(0o666))
        .with_context(|| format!("chmod agent socket {}", socket_path.display()))?;
    tracing::info!(%agent, socket = %socket_path.display(), "agent socket listening");

    let path = socket_path.to_path_buf();
    let handle = tokio::spawn(async move {
        let err = accept_until_fatal(
            &agent,
            || async { listener.accept().await.map(|(stream, _)| stream) },
            |stream| {
                let agent = agent.clone();
                let coord = coord.clone();
                tokio::spawn(async move {
                    if let Err(e) = serve(stream, agent, coord).await {
                        // A peer closing its end mid-read/write (the harness
                        // process exiting, a turn ending) is routine and the
                        // overwhelming majority of what reaches here —
                        // demoted so it doesn't bury a genuine failure.
                        if is_broken_pipe(&e) {
                            tracing::debug!(error = ?e, "agent connection closed by peer");
                        } else {
                            tracing::warn!(error = ?e, "agent connection failed");
                        }
                    }
                });
            },
        )
        .await;
        exit_on_dead_listener(&agent, &err);
    });
    Ok(AgentSocket { path, handle })
}

/// Pause after an accept error that is not about a single connection. EMFILE
/// and ENFILE leave the listener readable, so retrying at once would spin until
/// some fd is closed. Same policy as axum's `serve`, which the dashboard runs.
const ACCEPT_RETRY_DELAY: std::time::Duration = std::time::Duration::from_secs(1);

/// Accept errors that mean the listening fd itself is unusable: retrying it
/// returns the same error forever.
fn accept_error_is_fatal(e: &std::io::Error) -> bool {
    matches!(
        e.raw_os_error(),
        Some(libc::EBADF | libc::EFAULT | libc::EINVAL | libc::ENOTSOCK | libc::EOPNOTSUPP)
    )
}

/// Hand each accepted connection to `on_conn`, retrying every accept error
/// except a fatal one, which is returned. `agent` labels the log lines.
async fn accept_until_fatal<A, F>(
    agent: &str,
    mut accept: A,
    mut on_conn: impl FnMut(UnixStream),
) -> std::io::Error
where
    A: FnMut() -> F,
    F: std::future::Future<Output = std::io::Result<UnixStream>>,
{
    loop {
        match accept().await {
            Ok(stream) => on_conn(stream),
            Err(e) if accept_error_is_fatal(&e) => return e,
            Err(e) => {
                tracing::warn!(%agent, error = ?e, "socket accept failed; retrying");
                if !matches!(
                    e.kind(),
                    std::io::ErrorKind::ConnectionAborted
                        | std::io::ErrorKind::ConnectionReset
                        | std::io::ErrorKind::ConnectionRefused
                ) {
                    tokio::time::sleep(ACCEPT_RETRY_DELAY).await;
                }
            }
        }
    }
}

/// Exit on a listener that can no longer accept. A running hive-c0re re-binds
/// an agent socket only on that agent's next start; on restart
/// (`Restart = "on-failure"` on the unit) `mcp_sockets::sync_on_start` re-binds
/// every one.
fn exit_on_dead_listener(agent: &str, e: &std::io::Error) -> ! {
    tracing::error!(
        %agent,
        error = ?e,
        "socket listener cannot accept; exiting so hive-c0re restarts and re-binds all sockets"
    );
    std::process::exit(1)
}

/// Whether `e`'s root cause is the peer closing the connection.
/// `downcast_ref` searches the whole error chain (anyhow's own
/// `object_downcast`, not just the outermost type), so this finds a
/// wrapped `std::io::Error` whether or not something added `.context()`
/// on top of it.
fn is_broken_pipe(e: &anyhow::Error) -> bool {
    e.downcast_ref::<std::io::Error>().is_some_and(|io_err| {
        matches!(
            io_err.kind(),
            std::io::ErrorKind::BrokenPipe | std::io::ErrorKind::ConnectionReset
        )
    })
}

/// Answer each request line on `stream` until EOF. An over-long (or timed-out) line gets a
/// `Response::Err` and then the connection is closed, because the rest of the
/// line is still unread.
async fn serve(stream: UnixStream, agent: String, coord: Arc<Coordinator>) -> Result<()> {
    let (read, mut write) = stream.into_split();
    let mut reader = BufReader::new(read);
    let mut line = Vec::new();
    loop {
        let outcome = read_request_line(&mut reader, &mut line, LineDeadlines::NONE).await?;
        if outcome == RequestLine::Eof {
            return Ok(());
        }
        let refusal = outcome.refusal();
        let (resp, too_long) = match refusal {
            Some(message) => (Response::Err { message }, true),
            None => match serde_json::from_slice::<Request>(&line) {
                Ok(req) => (dispatch(&req, &agent, &coord).await, false),
                Err(e) => (
                    Response::Err {
                        message: format!("parse error: {e}"),
                    },
                    false,
                ),
            },
        };
        let mut payload = serde_json::to_string(&resp)?;
        payload.push('\n');
        write.write_all(payload.as_bytes()).await?;
        write.flush().await?;
        if too_long {
            anyhow::bail!("`{agent}` sent an over-long or stalled request line; closed");
        }
    }
}

/// Max long-poll window the caller can ask for; values above the
/// cap are clamped. 180s keeps us under typical TCP/proxy idle
/// limits while still letting agents park their turn until a
/// message arrives. Omitting `wait_seconds` (or passing `0`) means
/// "peek, don't wait" — claude can call recv whenever it wants a
/// cheap "is there anything pending?" check without blocking the
/// turn for 30 seconds. To actually park, the caller passes a
/// positive `wait_seconds`.
pub(crate) const RECV_LONG_POLL_MAX: std::time::Duration = std::time::Duration::from_mins(3);

/// Server-side hard cap on `Recv.max` — canonical value lives in
/// `hive_sh4re::inbox::RECV_BATCH_MAX` so the harness's wake-prompt hint and
/// this enforcement site can't drift apart.
pub(crate) const RECV_BATCH_MAX: u32 = hive_sh4re::inbox::RECV_BATCH_MAX;

pub(crate) fn recv_timeout(wait_seconds: Option<u64>) -> std::time::Duration {
    match wait_seconds {
        Some(s) => std::time::Duration::from_secs(s).min(RECV_LONG_POLL_MAX),
        None => std::time::Duration::ZERO,
    }
}

/// Unified dispatch for every socket connection. There is no privilege bit;
/// authority derives uniformly from the caller's identity.
async fn dispatch(
    req: &hive_core_agent_sock::Request,
    agent: &str,
    coord: &Arc<Coordinator>,
) -> hive_core_agent_sock::Response {
    match req {
        hive_core_agent_sock::Request::Send {
            to,
            body,
            in_reply_to,
        } => handle_send(coord, agent, to, body, *in_reply_to),
        hive_core_agent_sock::Request::Recv { wait_seconds, max } => {
            handle_recv(coord, agent, *wait_seconds, *max).await
        }
        hive_core_agent_sock::Request::Status => handle_status(coord, agent),
        hive_core_agent_sock::Request::OperatorMsg { body } => {
            handle_operator_msg(coord, agent, body)
        }
        hive_core_agent_sock::Request::Wake { from, body } => handle_wake(coord, agent, from, body),
        hive_core_agent_sock::Request::Recent { limit } => handle_recent(coord, agent, *limit),
        hive_core_agent_sock::Request::SetStatus { text } => handle_set_status(coord, text),
        hive_core_agent_sock::Request::GetAgentMeta { name } => {
            handle_get_agent_meta(coord, agent, name.as_ref()).await
        }
        hive_core_agent_sock::Request::AckTurn => handle_ack_turn(coord, agent),
        hive_core_agent_sock::Request::AckUntil { up_to } => handle_ack_until(coord, agent, *up_to),
        hive_core_agent_sock::Request::RequeueInflight => handle_requeue_inflight(coord, agent),
        hive_core_agent_sock::Request::GracefulStopComplete => {
            // Harness drained + is exiting: clear the fence so the
            // `GracefulStop` orchestration (which polls this flag) proceeds
            // to stop the container without waiting out its timeout.
            coord.clear_graceful_stop(agent);
            hive_core_agent_sock::Response::Ok
        }
        hive_core_agent_sock::Request::PauseAcknowledged => {
            // Harness saw its own pause marker flip between turns: clear
            // the fence so the pause DAG's drain node (which polls this
            // flag) resolves without waiting out its timeout.
            coord.clear_pause_pending(agent);
            hive_core_agent_sock::Response::Ok
        }
        hive_core_agent_sock::Request::GetLooseEnds => handle_get_loose_ends(coord, agent),
    }
}

/// `Recv` — long-poll the broker for up to `max` messages (capped at
/// `RECV_BATCH_MAX`), mapping deliveries onto the wire response.
async fn handle_recv(
    coord: &Arc<Coordinator>,
    agent: &str,
    wait_seconds: Option<u64>,
    max: Option<u32>,
) -> hive_core_agent_sock::Response {
    // Graceful-stop fence: while a graceful stop is pending for this agent,
    // return `GracefulStop` instead of polling the broker. The harness runs
    // one stop-checkpoint turn then exits; new sends keep queueing in the
    // broker for the agent's next start. Checked before the (blocking) poll
    // so a flag set between polls is seen on the next Recv — the orchestration
    // also fires a transient wake to break an in-flight long-poll.
    if coord.is_graceful_stop_pending(agent) {
        return hive_core_agent_sock::Response::GracefulStop;
    }
    let cap = max.unwrap_or(1).min(RECV_BATCH_MAX) as usize;
    match coord
        .broker
        .recv_blocking_batch(agent, recv_timeout(wait_seconds), cap)
        .await
    {
        Ok(deliveries) => {
            // `recv_batch` stamps `delivered_at` on every popped row, so a
            // `count_pending` here excludes the just-popped batch and reports
            // exactly how many still-pending messages remain to drain. A count
            // error is non-fatal — fall back to 0 rather than fail the recv.
            let remaining = coord.broker.count_pending(agent).unwrap_or(0);
            hive_core_agent_sock::Response::Messages {
                messages: deliveries
                    .into_iter()
                    .map(|d| hive_sh4re::inbox::DeliveredMessage {
                        from: d.message.from.to_string(),
                        body: d.message.body,
                        id: d.id,
                        redelivered: d.redelivered,
                        in_reply_to: d.message.in_reply_to,
                    })
                    .collect(),
                remaining,
            }
        }
        Err(e) => hive_core_agent_sock::Response::Err {
            message: format!("{e:#}"),
        },
    }
}

/// `Wake` — inject a wake into `agent`'s own inbox. Persisted through
/// the sqlite broker like any other message so the agent can ack it
/// via `AckUntil` and it appears in message history for post-mortem.
fn handle_wake(
    coord: &Arc<Coordinator>,
    agent: &str,
    from: &str,
    body: &str,
) -> hive_core_agent_sock::Response {
    match coord.broker.send(&Message {
        from: hive_sh4re::manager::trusted_sender(from),
        to: agent.to_owned(),
        body: body.to_owned(),
        in_reply_to: None,
    }) {
        Ok(()) => hive_core_agent_sock::Response::Ok,
        Err(e) => hive_core_agent_sock::Response::Err {
            message: format!("{e:#}"),
        },
    }
}

/// `SetStatus` — validate the status text, then trigger a dashboard
/// rescan. The harness has already written the status file to its own
/// `state/` dir (it runs as the agent user), so this only refreshes the
/// dashboard's view.
fn handle_set_status(coord: &Arc<Coordinator>, text: &str) -> hive_core_agent_sock::Response {
    if let Err(message) = crate::limits::check_status_text(text) {
        return hive_core_agent_sock::Response::Err { message };
    }
    let coord2 = Arc::clone(coord);
    tokio::spawn(async move { coord2.rescan_containers_and_emit().await });
    hive_core_agent_sock::Response::Ok
}

/// `GetAgentMeta` — identity + live status for `name` (defaults to the
/// caller). Reads the live container-view status and the hive/swarm
/// display names.
async fn handle_get_agent_meta(
    coord: &Arc<Coordinator>,
    agent: &str,
    name: Option<&hive_types::Ident>,
) -> hive_core_agent_sock::Response {
    // `name` arrives pre-validated by serde (wire field is `Ident`). The
    // `None` default (target == caller) still needs a parse since `agent`
    // is a plain `&str` here — but it's the caller's own authenticated
    // name, already valid in practice.
    let target_id = match name {
        Some(id) => id.clone(),
        None => match hive_types::Ident::parse(agent) {
            Ok(id) => id,
            Err(reason) => {
                return hive_core_agent_sock::Response::Err {
                    message: format!("get_agent_meta: invalid agent name {agent:?}: {reason}"),
                };
            }
        },
    };
    let (status_text, status_set_at, running) =
        crate::container_view::read_agent_status_live(&target_id).await;
    let (hive_name, swarm_name) = crate::container_view::hive_swarm_names();
    // Matrix identities are public handles (`name` / `user_id`
    // `@user:server` / `homeserver`) — the access token lives separately
    // in the agent's `matrix-token` and is never part of this response.
    // Peer visibility is intentional: it lets an agent verify/contact
    // another on a public matrix instance. The `Ident::parse` gate
    // above is what closes the real vector here (path traversal via `../`
    // in an agent-supplied name).
    let matrix_accounts = read_agent_matrix_identities(&target_id);
    hive_core_agent_sock::Response::AgentMeta {
        name: target_id.into_string(),
        running,
        hyperhive_rev: crate::auto_update::current_flake_rev(&coord.hyperhive_flake),
        status_text,
        status_set_at,
        hive_name,
        swarm_name,
        matrix_accounts,
    }
}

/// Read the target agent's matrix identities from the daemon's
/// `matrix-accounts.json` snapshot (under the agent's state dir).
/// Best-effort: an absent / unparseable snapshot (no matrix provisioning,
/// or the daemon not up yet) yields an empty list. The `MatrixIdentity`
/// serde shape matches the snapshot entries; the snapshot's `live` field is
/// ignored (only live accounts are written).
fn read_agent_matrix_identities(
    agent: &hive_types::Ident,
) -> Vec<hive_sh4re::container::MatrixIdentity> {
    let path = Coordinator::agent_notes_dir(agent).join("matrix-accounts.json");
    std::fs::read_to_string(&path)
        .ok()
        .and_then(|s| serde_json::from_str::<Vec<hive_sh4re::container::MatrixIdentity>>(&s).ok())
        .unwrap_or_default()
}

/// `Status` — count of pending (unread) inbox messages for `agent`.
fn handle_status(coord: &Arc<Coordinator>, agent: &str) -> hive_core_agent_sock::Response {
    match coord.broker.count_pending(agent) {
        Ok(unread) => hive_core_agent_sock::Response::Status { unread },
        Err(e) => hive_core_agent_sock::Response::Err {
            message: format!("{e:#}"),
        },
    }
}

/// `OperatorMsg` — deliver an operator-authored message into `agent`'s
/// inbox (sender `operator`).
fn handle_operator_msg(
    coord: &Arc<Coordinator>,
    agent: &str,
    body: &str,
) -> hive_core_agent_sock::Response {
    match coord.broker.send(&Message {
        from: hive_sh4re::manager::trusted_sender(hive_sh4re::manager::OPERATOR_RECIPIENT),
        to: agent.to_owned(),
        body: body.to_owned(),
        in_reply_to: None,
    }) {
        Ok(()) => hive_core_agent_sock::Response::Ok,
        Err(e) => hive_core_agent_sock::Response::Err {
            message: format!("{e:#}"),
        },
    }
}

/// `Recent` — the last `limit` inbox rows for `agent` (read-only,
/// doesn't consume).
fn handle_recent(
    coord: &Arc<Coordinator>,
    agent: &str,
    limit: u64,
) -> hive_core_agent_sock::Response {
    match coord.broker.recent_for(agent, limit) {
        Ok(rows) => hive_core_agent_sock::Response::Recent { rows },
        Err(e) => hive_core_agent_sock::Response::Err {
            message: format!("{e:#}"),
        },
    }
}

/// `AckTurn` — mark `agent`'s in-flight delivered messages acked so
/// they don't redeliver on the next turn.
fn handle_ack_turn(coord: &Arc<Coordinator>, agent: &str) -> hive_core_agent_sock::Response {
    match coord.broker.ack_turn(agent) {
        Ok(_n) => hive_core_agent_sock::Response::Ok,
        Err(e) => hive_core_agent_sock::Response::Err {
            message: format!("{e:#}"),
        },
    }
}

/// `AckUntil` — bulk-ack every message addressed to `agent` with row
/// id `<= up_to` (the agent-side backlog-triage escape hatch).
fn handle_ack_until(
    coord: &Arc<Coordinator>,
    agent: &str,
    up_to: i64,
) -> hive_core_agent_sock::Response {
    match coord.broker.ack_until(agent, up_to) {
        Ok(count) => hive_core_agent_sock::Response::Acked { count },
        Err(e) => hive_core_agent_sock::Response::Err {
            message: format!("{e:#}"),
        },
    }
}

/// `RequeueInflight` — resurface `agent`'s unacked in-flight messages
/// (crash recovery on harness boot).
fn handle_requeue_inflight(
    coord: &Arc<Coordinator>,
    agent: &str,
) -> hive_core_agent_sock::Response {
    match coord.broker.requeue_inflight(agent) {
        Ok(n) => {
            if n > 0 {
                tracing::info!(%agent, requeued = %n, "requeued in-flight messages");
            }
            hive_core_agent_sock::Response::Ok
        }
        Err(e) => hive_core_agent_sock::Response::Err {
            message: format!("{e:#}"),
        },
    }
}

/// `GetLooseEnds` — always the caller's own loose ends.
fn handle_get_loose_ends(coord: &Arc<Coordinator>, agent: &str) -> Response {
    match crate::loose_ends::for_agent(coord, agent) {
        Ok(loose_ends) => Response::LooseEnds { loose_ends },
        Err(e) => Response::Err {
            message: format!("{e:#}"),
        },
    }
}

/// `Send` handler. Applies the
/// 4 KiB body cap, then routes broadcast (`to == "*"`) / unicast through
/// their respective broker calls.
pub(crate) fn handle_send(
    coord: &Arc<Coordinator>,
    agent: &str,
    to: &str,
    body: &str,
    in_reply_to: Option<i64>,
) -> Response {
    if let Err(message) = crate::limits::check_size("send", body) {
        return Response::Err { message };
    }
    if to == "*" {
        let errors = coord.broadcast_send(agent, body);
        return if errors.is_empty() {
            Response::Ok
        } else {
            Response::Err {
                message: format!("broadcast failed for agents: {}", errors.join(", ")),
            }
        };
    }
    // Validate that the recipient is a known local agent. Without this
    // check a typo in `to` silently queues a message nobody will ever read.
    //
    // Cross-hive messaging (`name@hive` qualified names) and the operator
    // are not reached through the broker — the Matrix MCP tools are.
    if to.contains('@') {
        return Response::Err {
            message: format!(
                "send failed: cross-hive recipient `{to}` is not supported \
                 via the broker — use Matrix MCP tools for cross-hive messaging"
            ),
        };
    }
    if to == hive_sh4re::manager::OPERATOR_RECIPIENT {
        return Response::Err {
            message: hive_sh4re::manager::OPERATOR_SEND_REFUSAL.to_owned(),
        };
    }
    // A name that doesn't parse as an Ident can't be a local agent, so
    // it collapses into the same "unknown recipient" error as a valid
    // name with no state dir.
    let exists =
        hive_types::Ident::parse(to).is_ok_and(|id| crate::paths::agent_state_dir(&id).exists());
    if !exists {
        return Response::Err {
            message: format!(
                "send failed: unknown recipient `{to}` \
                 (no agent with that name exists on this hive)"
            ),
        };
    }
    match coord.broker.send(&Message {
        from: hive_sh4re::manager::trusted_sender(agent),
        to: to.to_owned(),
        body: body.to_owned(),
        in_reply_to,
    }) {
        Ok(()) => Response::Ok,
        Err(e) => Response::Err {
            message: format!("{e:#}"),
        },
    }
}

#[cfg(test)]
mod tests {
    use hive_sock_client::REQUEST_LINE_MAX_BYTES;
    use tokio::io::AsyncBufReadExt;

    use super::*;

    /// A real `Coordinator` over a throwaway sqlite dir. `serve` takes an
    /// `Arc<Coordinator>`, so there is no lighter way in; `open` touches
    /// nothing outside the db path it is handed.
    fn coordinator() -> (tempfile::TempDir, Arc<Coordinator>) {
        let dir = tempfile::tempdir().expect("tempdir");
        let coord = Coordinator::open(
            &dir.path().join("broker.sqlite"),
            crate::coordinator::HiveEnv::default(),
            crate::hive_stats::PriceTable::new(),
            1,
        )
        .expect("open coordinator");
        (dir, Arc::new(coord))
    }

    #[test]
    fn send_to_operator_is_refused_naming_matrix_and_the_no_matrix_fallback() {
        let (_dir, coord) = coordinator();
        let resp = handle_send(
            &coord,
            "iris",
            hive_sh4re::manager::OPERATOR_RECIPIENT,
            "hi",
            None,
        );
        assert!(
            matches!(&resp, Response::Err { message }
                if message.contains("mcp__matrix__") && message.contains("set_status")),
            "{resp:?}"
        );
    }

    #[tokio::test]
    async fn serve_answers_an_over_long_line_with_an_error_and_closes() {
        let (_dir, coord) = coordinator();
        let (client, server) = UnixStream::pair().expect("socketpair");
        let handler = tokio::spawn(serve(server, "iris".to_owned(), coord));
        let (read, mut write) = client.into_split();
        let writer = tokio::spawn(async move {
            let chunk = vec![b'x'; 64 * 1024];
            for _ in 0..2 * REQUEST_LINE_MAX_BYTES / chunk.len() {
                if write.write_all(&chunk).await.is_err() {
                    return;
                }
            }
            let _ = write.shutdown().await;
        });
        let mut reply = String::new();
        BufReader::new(read)
            .read_line(&mut reply)
            .await
            .expect("reply");
        let resp: Response = serde_json::from_str(&reply).expect("response json");
        assert!(
            matches!(&resp, Response::Err { message } if message.starts_with("request too long")),
            "{reply}"
        );
        assert!(
            handler.await.expect("join").is_err(),
            "connection not closed"
        );
        writer.await.expect("join");
    }

    #[tokio::test]
    async fn accept_loop_survives_a_transient_error() {
        let (client, _peer) = UnixStream::pair().expect("socketpair");
        let mut script = std::collections::VecDeque::from([
            Err(std::io::Error::from_raw_os_error(libc::EMFILE)),
            Ok(client),
            Err(std::io::Error::from_raw_os_error(libc::EBADF)),
        ]);
        let mut served = 0;
        let err = accept_until_fatal(
            "iris",
            || std::future::ready(script.pop_front().expect("accept after a fatal error")),
            |_stream| served += 1,
        )
        .await;
        assert_eq!(served, 1, "connection after EMFILE not served");
        assert_eq!(err.raw_os_error(), Some(libc::EBADF));
    }

    /// `is_broken_pipe` must recognise both peer-closed kinds whether the
    /// `io::Error` reaches it bare (how `serve`'s own `?`s produce it) or
    /// wrapped in `.context()` (anyhow's `downcast_ref` walks the whole
    /// chain, so wrapping doesn't hide it — confirmed here rather than
    /// assumed), and must not misclassify an unrelated `io::Error` or a
    /// non-io `anyhow::Error`.
    #[test]
    fn is_broken_pipe_matches_peer_closed_kinds_only() {
        let broken_pipe = std::io::Error::from(std::io::ErrorKind::BrokenPipe);
        let conn_reset = std::io::Error::from(std::io::ErrorKind::ConnectionReset);
        let permission_denied = std::io::Error::from(std::io::ErrorKind::PermissionDenied);

        assert!(is_broken_pipe(&anyhow::Error::new(broken_pipe)));
        assert!(is_broken_pipe(&anyhow::Error::new(conn_reset)));
        assert!(!is_broken_pipe(&anyhow::Error::new(permission_denied)));

        let contextual = anyhow::Error::new(std::io::Error::from(std::io::ErrorKind::BrokenPipe))
            .context("write response");
        assert!(
            is_broken_pipe(&contextual),
            "anyhow's downcast_ref searches the whole chain, so a \
             .context()-wrapped io::Error must still be found"
        );

        assert!(!is_broken_pipe(&anyhow::anyhow!("not an io error at all")));
    }
}