hyperhive/hive-c0re/src/socket_server/mod.rs

//! Unix-socket request server, shared by the per-agent sockets and the
//! (pure-transport) manager socket. The socket file's existence on disk
//! authenticates the caller: connecting to `<.../agents/foo/mcp.sock>` means
//! you are `foo`; the manager socket simply serves as `ruth`. There is no
//! privilege flag — both transports run the same [`serve`] / [`dispatch`]
//! code, and authority derives uniformly from the caller's identity. An
//! agent-targeting verb may name any agent, so `ruth` reaches every agent
//! exactly the way every other agent does, not via any hardcoded name match.

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_sh4re::manager::MANAGER_AGENT;
use tokio::io::{AsyncBufRead, AsyncBufReadExt, AsyncReadExt, 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 {
                        tracing::warn!(error = ?e, "agent connection failed");
                    }
                });
            },
        )
        .await;
        exit_on_dead_listener(&agent, &err);
    });
    Ok(AgentSocket { path, handle })
}

/// Bind + serve the manager socket. This is now **pure transport**: it grants
/// no authority of its own — it just serves requests as `agent = MANAGER_AGENT`
/// ("ruth"), and ruth's reach comes entirely from the capabilities /
/// tool-groups it holds, identical to connecting on a per-agent socket — ruth
/// uses the standard per-agent runtime dir + socket, with no dedicated helpers.
pub fn start_manager(coord: Arc<Coordinator>) -> Result<()> {
    use std::os::unix::fs::PermissionsExt as _;
    let dir = crate::paths::agent_runtime_dir(crate::lifecycle::MANAGER_NAME);
    std::fs::create_dir_all(&dir)
        .with_context(|| format!("create manager dir {}", dir.display()))?;
    let socket = Coordinator::socket_path(crate::lifecycle::MANAGER_NAME);
    if socket.exists() {
        std::fs::remove_file(&socket).context("remove stale manager socket")?;
    }
    let listener = UnixListener::bind(&socket)
        .with_context(|| format!("bind manager socket {}", socket.display()))?;
    // 0666 so the in-container root user (non-root) can connect; the bind
    // source dir is manager-only on host (see the per-agent socket above).
    std::fs::set_permissions(&socket, std::fs::Permissions::from_mode(0o666))
        .with_context(|| format!("chmod manager socket {}", socket.display()))?;
    tracing::info!(socket = %socket.display(), "manager socket listening");

    tokio::spawn(async move {
        let err = accept_until_fatal(
            MANAGER_AGENT,
            || async { listener.accept().await.map(|(stream, _)| stream) },
            |stream| {
                let coord = coord.clone();
                tokio::spawn(async move {
                    // Pure transport: serve as `ruth`, no privilege grant.
                    if let Err(e) = serve(stream, MANAGER_AGENT.to_owned(), coord).await {
                        tracing::warn!(error = ?e, "manager connection failed");
                    }
                });
            },
        )
        .await;
        exit_on_dead_listener(MANAGER_AGENT, &err);
    });
    Ok(())
}

/// 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, and the manager socket
/// never; on restart (`Restart = "on-failure"` on the unit) `start_manager` and
/// `mcp_sockets::sync_on_start` re-bind 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)
}

/// Longest request line accepted on the agent and manager sockets, newline
/// included. The largest request a client sends is an `OperatorMsg` from the
/// agent web UI's `/send` (`hive-agent/src/web_ui/actions.rs:21`), whose body
/// axum's default limit caps at 2 MiB; JSON escaping can roughly double that to
/// 4 MiB, and this is 4× that.
const REQUEST_LINE_MAX_BYTES: usize = 16 * 1024 * 1024;

enum RequestLine {
    Eof,
    Complete,
    TooLong,
}

/// Read one request line into `buf`, consuming at most
/// [`REQUEST_LINE_MAX_BYTES`] from `reader`.
async fn read_request_line<R: AsyncBufRead + Unpin>(
    reader: &mut R,
    buf: &mut Vec<u8>,
) -> std::io::Result<RequestLine> {
    buf.clear();
    let n = (&mut *reader)
        .take(REQUEST_LINE_MAX_BYTES as u64)
        .read_until(b'\n', buf)
        .await?;
    Ok(if n == 0 {
        RequestLine::Eof
    } else if n == REQUEST_LINE_MAX_BYTES && !buf.ends_with(b"\n") {
        RequestLine::TooLong
    } else {
        RequestLine::Complete
    })
}

/// Answer each request line on `stream` until EOF. An over-long 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 (resp, too_long) = match read_request_line(&mut reader, &mut line).await? {
            RequestLine::Eof => return Ok(()),
            RequestLine::TooLong => (
                Response::Err {
                    message: format!(
                        "request too long: no newline within {REQUEST_LINE_MAX_BYTES} bytes; \
                         connection closed"
                    ),
                },
                true,
            ),
            RequestLine::Complete => 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 a request line over {REQUEST_LINE_MAX_BYTES} bytes; 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 — per-agent sockets and the
/// (now pure-transport) manager socket alike. 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 (from the `operator` recipient).
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 or the
    // special "operator" recipient. Without this check a typo in `to`
    // silently queues a message nobody will ever read.
    //
    // Cross-hive messaging (`name@hive` qualified names) is not routed
    // through the broker — use the Matrix MCP tools for that instead.
    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 {
        // 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 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))
    }

    #[tokio::test]
    async fn over_long_line_is_rejected_after_reading_only_the_bound() {
        let total = 2 * REQUEST_LINE_MAX_BYTES as u64;
        let mut reader = BufReader::new(tokio::io::repeat(b'x').take(total));
        let mut buf = Vec::new();
        let outcome = read_request_line(&mut reader, &mut buf)
            .await
            .expect("read");
        // At most one `BufReader` fill past the bound left the source.
        let unread = reader.into_inner().limit();
        let consumed = total - unread;
        assert!(
            consumed <= (REQUEST_LINE_MAX_BYTES + 8 * 1024) as u64,
            "consumed {consumed} of {total} bytes"
        );
        assert_eq!(buf.len(), REQUEST_LINE_MAX_BYTES);
        assert!(matches!(outcome, RequestLine::TooLong));
    }

    #[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));
    }
}