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:
//! topology (`is_descendant_of`) for subtree-relational verbs, capabilities
//! for hive-wide queries, and tool-group membership for the orchestration
//! verbs. `ruth` reaches every agent only as a consequence of being the
//! topology root, 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::{AsyncBufReadExt, AsyncWriteExt, BufReader};
use tokio::net::{UnixListener, UnixStream};
use tokio::task::JoinHandle;

use crate::coordinator::Coordinator;

mod config_approvals;
mod schedules;

pub(crate) use config_approvals::submit_merge_config_pr;
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 {
        loop {
            match listener.accept().await {
                Ok((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");
                        }
                    });
                }
                Err(e) => {
                    tracing::warn!(error = ?e, "agent listener accept failed; exiting");
                    return;
                }
            }
        }
    });
    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 being the topology root
/// (`is_descendant_of` covers every agent) plus 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 {
        loop {
            match listener.accept().await {
                Ok((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");
                        }
                    });
                }
                Err(e) => {
                    tracing::warn!(error = ?e, "manager listener accept failed");
                    return;
                }
            }
        }
    });
    Ok(())
}

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 = String::new();
    loop {
        line.clear();
        let n = reader.read_line(&mut line).await?;
        if n == 0 {
            return Ok(());
        }
        let resp = match serde_json::from_str::<Request>(line.trim()) {
            Ok(req) => dispatch(&req, &agent, &coord).await,
            Err(e) => Response::Err {
                message: format!("parse error: {e}"),
            },
        };
        let mut payload = serde_json::to_string(&resp)?;
        payload.push('\n');
        write.write_all(payload.as_bytes()).await?;
        write.flush().await?;
    }
}

/// 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::CreateRepo { repo } => handle_create_repo(agent, repo).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::GetHostJournal {
            unit,
            container,
            lines,
            priority,
            grep,
            since,
            until,
        } => {
            dispatch_host_journal(
                agent,
                HostJournalArgs {
                    unit,
                    container,
                    lines,
                    priority,
                    grep,
                    since,
                    until,
                },
            )
            .await
        }
        // 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
}

/// Validate an agent-supplied repo name: a single safe slug segment, no
/// path traversal. Forgejo validates server-side too, but rejecting early
/// gives a clear message and avoids building odd API paths.
fn valid_repo_name(name: &str) -> bool {
    !name.is_empty()
        && name.len() <= 100
        && !name.starts_with(['-', '.'])
        && name
            .chars()
            .all(|c| c.is_ascii_alphanumeric() || matches!(c, '-' | '_' | '.'))
}

/// `CreateRepo` — create a repo for `agent` *through hive-c0re* in the
/// c0re-owned `agents` org with operator-team branch protection.
/// The sanctioned create path now that agents can't create repos directly.
async fn handle_create_repo(agent: &str, repo: &str) -> hive_core_agent_sock::Response {
    if !valid_repo_name(repo) {
        return hive_core_agent_sock::Response::Err {
            message: format!(
                "invalid repo name {repo:?} — single segment of letters, digits, '-', '_', '.' \
                 (no leading '-'/'.', max 100 chars)"
            ),
        };
    }
    let Some(core_token) = crate::forge::core_token() else {
        return hive_core_agent_sock::Response::Err {
            message: "forge unavailable (no core token) — cannot create repo".to_owned(),
        };
    };
    match crate::forge::create_agent_repo(agent, repo, &core_token).await {
        Ok(full_name) => hive_core_agent_sock::Response::RepoCreated {
            clone_url: format!("{}/{full_name}.git", crate::forge::forge_http_base()),
            full_name,
        },
        Err(e) => hive_core_agent_sock::Response::Err {
            message: format!("create repo {repo:?} failed: {e:#}"),
        },
    }
}

/// `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> {
    if crate::tool_groups::groups_for(agent)
        .iter()
        .any(|g| g == group)
    {
        None
    } else {
        Some(Response::Err {
            message: format!("agent `{agent}` cannot {action}: requires the `{group}` tool group"),
        })
    }
}

/// `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");
            let sha_short = approval
                .fetched_sha
                .as_deref()
                .map(|s| s[..s.len().min(12)].to_owned());
            coord.emit_approval_resolved(crate::coordinator::ApprovalResolved {
                id: approval.id,
                agent: approval.agent.as_str(),
                approval_kind: <&str>::from(approval.kind),
                sha_short,
                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)
        .iter()
        .any(|g| g == APPROVALS_GROUP)
    {
        Ok(())
    } else {
        Err(
            "cancel_loose_end: cancelling approval rows requires the `approvals` tool group"
                .to_owned(),
        )
    }
}

/// Field-named journal-query knobs for [`dispatch_host_journal`].
/// Borrows straight from the matched `GetHostJournal` request variant.
pub struct HostJournalArgs<'a> {
    pub unit: &'a Option<String>,
    pub container: &'a Option<String>,
    pub lines: &'a Option<u32>,
    pub priority: &'a Option<hive_sh4re::journal::JournalPriority>,
    pub grep: &'a Option<String>,
    pub since: &'a Option<String>,
    pub until: &'a Option<String>,
}

/// Handle `GetHostJournal` from both the agent and manager sockets.
/// Capability-gated: the calling agent must hold `read_host_journal` in
/// `meta/capabilities.json`. Runs `journalctl` host-side and returns
/// the output as a `HostJournal` response.
///
/// The manager is not exempt - grant `read_host_journal` in
/// `meta/capabilities.json` to enable it for any agent including the manager.
pub async fn dispatch_host_journal(agent: &str, args: HostJournalArgs<'_>) -> Response {
    let HostJournalArgs {
        unit,
        container,
        lines,
        priority,
        grep,
        since,
        until,
    } = args;
    if !crate::capabilities::has_cap(agent, hive_sh4re::permissions::Capability::ReadHostJournal) {
        return Response::Err {
            message: "agent does not have the read_host_journal capability".to_owned(),
        };
    }
    let n = lines.unwrap_or(30).min(100);

    // A container (`-M`) read enters the container namespace and needs
    // root, so it's delegated to hive-priv. A host read (no container)
    // the unprivileged hive-core user can do directly via its
    // systemd-journal group membership.
    if let Some(c) = container {
        tracing::info!(%agent, machine = %c, %n, "get_host_journal (container)");
        return match crate::priv_client::read_container_journal(
            c,
            hive_priv_sock::JournalQuery {
                lines: n,
                unit: unit.clone(),
                priority: priority.as_ref().map(|p| p.as_journald_str().to_owned()),
                grep: grep.clone(),
                since: since.clone(),
                until: until.clone(),
                ..Default::default()
            },
        )
        .await
        {
            Ok((stdout, stderr)) => {
                let content = if stdout.is_empty() { stderr } else { stdout };
                Response::HostJournal { content }
            }
            Err(e) => Response::Err {
                message: format!("journal read: {e:#}"),
            },
        };
    }

    let mut args: Vec<String> = vec![
        "--no-pager".to_owned(),
        "--output=short".to_owned(),
        "-n".to_owned(),
        n.to_string(),
    ];
    if let Some(u) = unit {
        args.push("-u".to_owned());
        args.push(u.clone());
    }
    if let Some(p) = priority {
        args.push("-p".to_owned());
        args.push(p.as_journald_str().to_owned());
    }
    if let Some(g) = grep {
        args.push(format!("--grep={g}"));
    }
    if let Some(s) = since {
        args.push(format!("--since={s}"));
    }
    if let Some(u) = until {
        args.push(format!("--until={u}"));
    }
    tracing::info!(%agent, ?args, "get_host_journal");
    match tokio::process::Command::new("journalctl")
        .args(&args)
        .output()
        .await
    {
        Ok(out) => {
            let content = if out.status.success() || !out.stdout.is_empty() {
                String::from_utf8_lossy(&out.stdout).into_owned()
            } else {
                let stderr = String::from_utf8_lossy(&out.stderr);
                format!("journalctl exited {}: {stderr}", out.status)
            };
            Response::HostJournal { content }
        }
        Err(e) => Response::Err {
            message: format!("journalctl spawn failed: {e:#}"),
        },
    }
}

/// Fan out one message to each recipient in `targets`. Skips the sender
/// itself. Returns a list of `"<agent>: <error>"` strings for any delivery
/// failures (empty = all good).
pub(crate) fn fan_out_send(
    coord: &Arc<Coordinator>,
    from: &str,
    body: &str,
    in_reply_to: Option<i64>,
    targets: &[String],
) -> Vec<String> {
    let mut errors = Vec::new();
    for target in targets {
        if target == from {
            continue;
        }
        if let Err(e) = coord.broker.send(&Message {
            from: hive_sh4re::manager::trusted_sender(from),
            to: target.clone(),
            body: body.to_owned(),
            in_reply_to,
        }) {
            errors.push(format!("{target}: {e}"));
        }
    }
    errors
}

/// Common Send handler shared between dispatch arms. Applies the
/// 4 KiB body cap, then routes broadcast (`to == "*"`) / children fan-out
/// (`to == "<children>"`) / 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(", ")),
            }
        };
    }
    // `<children>`: fan out to every direct descendant of the sender per
    // topology.json. Bypasses the allow-list check — structural fan-out
    // targets are never user-listed peers. No-op (returns Ok) for leaf
    // agents that have no children.
    if to == hive_sh4re::manager::CHILDREN_RECIPIENT {
        let children = crate::topology::children_of(agent);
        let errors = fan_out_send(coord, agent, body, in_reply_to, &children);
        return if errors.is_empty() {
            Response::Ok
        } else {
            Response::Err {
                message: format!("children fan-out failed for agents: {}", errors.join(", ")),
            }
        };
    }
    // Resolve magic-recipient sentinels (`<parent>`) against topology.json;
    // no-op for ordinary names. Lets agents address structural roles without
    // learning the label — runtime reparenting propagates for free. See
    // `docs/process/conventions.md::Recipient sentinels`.
    let resolved = crate::topology::resolve_recipient(agent, to);
    // Validate that the resolved 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 resolved.contains('@') {
        return Response::Err {
            message: format!(
                "send failed: cross-hive recipient `{resolved}` is not supported \
                 via the broker — use Matrix MCP tools for cross-hive messaging"
            ),
        };
    }
    if resolved != 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(&resolved)
            .is_ok_and(|id| crate::paths::agent_state_dir(&id).exists());
        if !exists {
            return Response::Err {
                message: format!(
                    "send failed: unknown recipient `{resolved}` \
                     (no agent with that name exists on this hive)"
                ),
            };
        }
    }
    match coord.broker.send(&Message {
        from: hive_sh4re::manager::trusted_sender(agent),
        to: resolved,
        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}");
    }
}