//! 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:
//! tool-group membership for the orchestration verbs. 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;

mod schedules;

pub(crate) use schedules::filter_ghost_schedule_targets;
pub use schedules::schedule_to_wire_public;

use schedules::{
    EditSchedulePatch, handle_cancel_schedule, handle_edit_schedule, handle_fire_schedule_now,
    handle_list_schedules, handle_request_schedule_prompt,
};

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,
    }
}

/// Handle the subset of `Request` variants that are identical on both
/// the agent socket and the manager socket. Returns `Some(response)` for
/// every variant it handles; returns `None` for the remaining variants —
/// `GetLooseEnds` and the orchestration verbs (schedules), which need
/// per-verb tool-group gating — or for host-admin / unknown requests
/// invalid on either socket.
///
/// The unified `dispatch` calls this first; the remaining arms (which gate
/// on topology / capabilities / tool-groups) are handled there.
pub(crate) async fn dispatch_shared(
    req: &hive_core_agent_sock::Request,
    agent: &str,
    coord: &Arc<Coordinator>,
) -> Option<hive_core_agent_sock::Response> {
    Some(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::CancelLooseEnd { kind, id } => {
            handle_cancel_loose_end(coord, agent, *kind, *id).map_or_else(
                |message| hive_core_agent_sock::Response::Err { message },
                |()| hive_core_agent_sock::Response::Ok,
            )
        }
        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
        }
        // Not a shared variant.
        _ => return None,
    })
}

/// `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:#}"),
        },
    }
}

/// 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: hive-wide
/// orchestration verbs (schedules) require the matching tool-group (the
/// grantable capability).
async fn dispatch(req: &Request, agent: &str, coord: &Arc<Coordinator>) -> Response {
    if let Some(resp) = dispatch_shared(req, agent, coord).await {
        return resp;
    }
    match req {
        Request::GetLooseEnds => handle_get_loose_ends(coord, agent),
        // Orchestration verbs — gated per-verb on tool-group membership
        // (see `dispatch_orchestration`).
        _ => dispatch_orchestration(req, agent, coord).await,
    }
}

/// Handle the hive-wide orchestration verbs (scheduling). No blanket socket
/// gate: each verb gates on the grantable capability that authorises it —
/// the matching tool-group (`scheduling`). Any other variant is a
/// host-admin / unknown request invalid on either socket.
async fn dispatch_orchestration(req: &Request, agent: &str, coord: &Arc<Coordinator>) -> Response {
    match req {
        Request::RequestSchedulePrompt(payload) => {
            if let Some(err) = require_group(agent, "scheduling", "schedule a prompt") {
                return err;
            }
            handle_request_schedule_prompt(coord, agent, payload)
        }
        Request::CancelSchedule { id, targets } => {
            if let Some(err) = require_group(agent, "scheduling", "cancel a schedule") {
                return err;
            }
            handle_cancel_schedule(coord, agent, *id, targets.as_deref())
        }
        Request::EditSchedule {
            id,
            body,
            description,
            interval_seconds,
            next_fire_at_unix,
            targets_add,
            targets_remove,
        } => {
            if let Some(err) = require_group(agent, "scheduling", "edit a schedule") {
                return err;
            }
            handle_edit_schedule(
                coord,
                agent,
                *id,
                EditSchedulePatch {
                    body: body.clone(),
                    description: description.clone(),
                    interval_seconds: *interval_seconds,
                    next_fire_at_unix: *next_fire_at_unix,
                    targets_add: targets_add.clone(),
                    targets_remove: targets_remove.clone(),
                },
            )
        }
        Request::ListSchedules => {
            if let Some(err) = require_group(agent, "scheduling", "list schedules") {
                return err;
            }
            handle_list_schedules(coord, agent)
        }
        Request::FireScheduleNow { id } => {
            if let Some(err) = require_group(agent, "scheduling", "fire a schedule") {
                return err;
            }
            handle_fire_schedule_now(coord, agent, *id).await
        }
        // Host-admin-only / unknown variants: never valid on either socket.
        _ => Response::Err {
            message: "request not handled on this socket".to_owned(),
        },
    }
}

/// Capability guard for the hive-wide orchestration verbs: the caller must
/// hold the given tool-group. The tool-group (c0re-owned `tool_groups.json`,
/// read server-side via [`crate::tool_groups::groups_for`]) is the grantable
/// capability — granting it to an orchestrator (e.g. the root) authorises
/// these verbs without any positional/hardcoded privilege. `action` is the
/// verb phrase for the message.
fn require_group(agent: &str, group: &str, action: &str) -> Option<Response> {
    match crate::tool_groups::groups_for(agent) {
        Ok(groups) if groups.iter().any(|g| g == group) => None,
        Ok(_) => Some(Response::Err {
            message: format!("agent `{agent}` cannot {action}: requires the `{group}` tool group"),
        }),
        Err(e) => Some(Response::Err {
            message: format!("agent `{agent}` cannot {action}: {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:#}"),
        },
    }
}

/// Handle `CancelLooseEnd` from a per-agent socket. Dispatches by kind, each
/// with its own auth check: reminder cancels are handled entirely in-container
/// (this arm should be unreachable in practice, kept only so the match stays
/// exhaustive) and approval cancels require the `approvals` tool-group (the
/// grantable capability) AND ownership — the canceller must be the approval's
/// submitter — so no positional / hardcoded privilege and no cross-agent
/// cancellation. (The operator's cancel-anything path is a separate handler.)
fn handle_cancel_loose_end(
    coord: &Arc<Coordinator>,
    canceller: &str,
    kind: hive_sh4re::inbox::CancelLooseEndKind,
    id: i64,
) -> Result<(), String> {
    match kind {
        hive_sh4re::inbox::CancelLooseEndKind::Reminder => {
            // Reminders are now agent-local (in-container store) — the
            // agent-mcp `cancel_loose_end` tool branches on this kind and
            // dials the agent's own socket directly, never forwarding to
            // hive-c0re. This arm should be unreachable in practice; kept
            // only so the match stays exhaustive.
            Err(format!(
                "reminder {id}: reminders are handled locally by the agent, \
                 not by hive-c0re"
            ))
        }
        hive_sh4re::inbox::CancelLooseEndKind::Approval => {
            // Withdrawing an approval needs the grantable `approvals`
            // tool-group (held by any approval-submitting orchestrator)
            // AND ownership: only the agent that submitted the approval
            // may withdraw it. Without the ownership check, any
            // approvals-group agent could cancel any other's approval by
            // id. A NULL submitter (legacy row predating the column) is
            // treated as operator-initiated (no agent tracking predates
            // the column).
            check_can_cancel_approval(canceller)?;
            let submitter = coord
                .approvals
                .submitter_of(id)
                .map_err(|e| format!("{e:#}"))?
                .unwrap_or_else(|| "operator".to_owned());
            if submitter != canceller {
                return Err(format!(
                    "cancel_loose_end: approval {id} was submitted by {submitter}, \
                     not {canceller}; only the submitting agent can withdraw it"
                ));
            }
            let approval = coord
                .approvals
                .mark_cancelled(id, canceller)
                .map_err(|e| format!("{e:#}"))?;
            tracing::info!(%id, %canceller, agent = %approval.agent, "approval cancelled");
            coord.emit_approval_resolved(crate::coordinator::ApprovalResolved {
                id: approval.id,
                agent: approval.agent.as_str(),
                approval_kind: <&str>::from(approval.kind),
                status: "cancelled",
                note: approval.note,
                description: approval.description,
            });
            Ok(())
        }
    }
}

/// Capability guard on the `Approval` cancel arm: the caller must hold the
/// `approvals` tool-group (the grantable capability for approval-submitting
/// orchestrators), checked server-side via `tool_groups::groups_for`. Pulled
/// out so the auth check has its own focused unit test — exercising the full
/// `handle_cancel_loose_end` flow would need a `Coordinator` fixture (broker +
/// sqlite). Keys on a grantable capability, not a positional / hardcoded
/// privilege.
fn check_can_cancel_approval(canceller: &str) -> Result<(), String> {
    const APPROVALS_GROUP: &str = "approvals";
    if crate::tool_groups::groups_for(canceller)
        .map_err(|e| format!("cancel_loose_end: {e}"))?
        .iter()
        .any(|g| g == APPROVALS_GROUP)
    {
        Ok(())
    } else {
        Err(
            "cancel_loose_end: cancelling approval rows requires the `approvals` tool group"
                .to_owned(),
        )
    }
}

/// Common Send handler shared between dispatch arms. Applies the
/// 4 KiB body cap, then routes broadcast (`to == "*"`) / unicast through
/// their respective broker calls.
/// `pub(crate)` so `dispatch_shared` can use it across both socket paths.
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::*;

    #[test]
    fn approval_cancel_rejects_callers_without_the_approvals_group() {
        // A caller that doesn't hold the `approvals` tool-group must not be
        // able to cancel approval rows even if it invents an id. The guard is
        // server-side so client cooperation is irrelevant — and it keys on a
        // grantable capability (the tool-group), not on any agent name.
        // `groups_for` of a name with no tool_groups.json entry is empty.
        let err = check_can_cancel_approval("nobody-with-no-groups").unwrap_err();
        assert!(err.contains("approvals` tool group"), "{err}");
    }

    #[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) = schedules::tests::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));
    }
}