//! 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::{MANAGER_AGENT, Message};
use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
use tokio::net::{UnixListener, UnixStream};
use tokio::task::JoinHandle;

use crate::coordinator::Coordinator;

mod config_approvals;
mod lifecycle_handlers;
mod reminders;
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 config_approvals::{handle_request_init_config, handle_request_update_meta_inputs};
use lifecycle_handlers::{
    handle_kill, handle_list_descendants, handle_restart, handle_start, handle_update,
};
use reminders::{handle_remind, resolve_agent_state_target};
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::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::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 variants with socket-specific
/// semantics (e.g. `GetLooseEnds` / `CountPendingReminders` / `ReminderRollup`
/// where the manager can target other agents) or for manager-only variants.
///
/// 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::Ask {
            question,
            options,
            multi,
            ttl_seconds,
            to,
        } => crate::questions::handle_ask(
            coord,
            agent,
            question,
            options,
            *multi,
            *ttl_seconds,
            to.as_deref(),
        )
        .map_or_else(
            |message| hive_core_agent_sock::Response::Err { message },
            |id| hive_core_agent_sock::Response::QuestionQueued { id },
        ),
        hive_core_agent_sock::Request::Answer { id, answer } => {
            crate::questions::handle_answer(coord, agent, *id, answer).map_or_else(
                |message| hive_core_agent_sock::Response::Err { message },
                |()| hive_core_agent_sock::Response::Ok,
            )
        }
        hive_core_agent_sock::Request::Remind {
            message,
            timing,
            file_path,
        } => handle_remind(coord, agent, message, timing, file_path.as_deref()),
        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_deref()).await
        }
        hive_core_agent_sock::Request::CancelLooseEnd { kind, id } => {
            crate::questions::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::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::DeliveredMessage {
                        from: d.message.from,
                        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: from.to_owned(),
        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<&str>,
) -> hive_core_agent_sock::Response {
    let target = name.unwrap_or(agent);
    // `name` is agent-supplied and flows into filesystem reads below
    // (`read_agent_status_live`, `read_agent_matrix_identities` →
    // `agent_notes_dir(target)`), where a `../` component would traverse at
    // the OS level. Validate it before any path is built. The `None` default
    // (`target == agent`) is the caller's own authenticated name, already
    // valid — but validating unconditionally is simplest and harmless.
    let target_id = match hive_types::Ident::parse(target) {
        Ok(id) => id,
        Err(reason) => {
            return hive_core_agent_sock::Response::Err {
                message: format!("get_agent_meta: invalid agent name {target:?}: {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();
    hive_core_agent_sock::Response::AgentMeta {
        name: target.to_owned(),
        running,
        hyperhive_rev: crate::auto_update::current_flake_rev(&coord.hyperhive_flake),
        status_text,
        status_set_at,
        hive_name,
        swarm_name,
        // 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).
        matrix_accounts: read_agent_matrix_identities(&target_id),
    }
}

/// 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::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::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::OPERATOR_RECIPIENT.to_owned(),
        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: subtree-relational
/// verbs (lifecycle/config/logs) require the caller to be an ancestor of the
/// target (`is_descendant_of`, so the root covers all); hive-wide agent-state
/// queries require the `QueryAgentState` capability; hive-wide orchestration
/// verbs (schedules / meta-inputs) 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 {
        // Lifecycle + config: caller must be an ancestor of the target
        // (a parent owns its whole subtree; the root covers every agent).
        Request::Start { name } => handle_start(coord, agent, name).await,
        Request::Restart { name } => handle_restart(coord, agent, name).await,
        Request::Kill { name } => handle_kill(coord, agent, name).await,
        Request::Update { name } => handle_update(coord, agent, name),
        Request::ListDescendants => handle_list_descendants(agent).await,
        Request::RequestInitConfig { name, description } => {
            handle_request_init_config(coord, agent, name, description.clone())
        }
        // Agent-state queries: own subtree is free; other agents + the
        // hive-wide `"*"` sweep require `QueryAgentState`.
        Request::GetLooseEnds { agent: target } => {
            handle_get_loose_ends(coord, agent, target.as_deref())
        }
        Request::CountPendingReminders { agent: target } => {
            handle_count_pending_reminders(coord, agent, target.as_deref())
        }
        Request::ReminderRollup {
            since_secs,
            agent: target,
        } => handle_reminder_rollup(coord, agent, target.as_deref(), *since_secs),
        // Todos (loose-ends v2): in-container subsystems push/clear
        // their own; the agent lists / marks its own done. Scoped to the
        // calling agent (the socket identity) — no cross-agent access.
        Request::UpsertTodo {
            subsystem,
            key,
            summary,
            source,
        } => handle_upsert_todo(
            coord,
            agent,
            subsystem,
            key.as_deref(),
            summary,
            source.as_deref(),
        ),
        Request::ClearTodo {
            subsystem,
            key,
            all,
        } => handle_clear_todo(coord, agent, subsystem, key.as_deref(), *all),
        Request::ListTodos { subsystem } => handle_list_todos(coord, agent, subsystem.as_deref()),
        Request::MarkTodoDone { id } => handle_mark_todo_done(coord, agent, *id),
        // Orchestration / diagnostics verbs — gated per-verb on tool-group
        // membership or topology (see `dispatch_orchestration`).
        _ => dispatch_orchestration(req, agent, coord).await,
    }
}

/// Handle the hive-wide orchestration verbs (scheduling, meta-input updates)
/// plus container-log reads. No blanket socket gate: each verb gates on the
/// grantable capability that authorises it — the matching tool-group
/// (`scheduling` / `approvals`), or `is_descendant_of` for `get_logs`
/// (a parent reads its subtree's logs). 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::RequestUpdateMetaInputs {
            inputs,
            description,
        } => {
            if let Some(err) = require_group(agent, "approvals", "request update_meta_inputs") {
                return err;
            }
            handle_request_update_meta_inputs(coord, agent, inputs, description.as_deref())
        }
        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)
        }
        Request::FireScheduleNow { id } => {
            if let Some(err) = require_group(agent, "scheduling", "fire a schedule") {
                return err;
            }
            handle_fire_schedule_now(coord, agent, *id).await
        }
        Request::GetLogs {
            agent: target,
            lines,
        } => {
            if let Some(err) = require_descendant(agent, target, "read logs of") {
                return err;
            }
            handle_get_logs(target, *lines).await
        }
        // Host-admin-only / unknown variants: never valid on either socket.
        _ => Response::Err {
            message: "request not handled on this socket".to_owned(),
        },
    }
}

/// Topology guard for the subtree-relational lifecycle/config/log tools: the
/// `target` must be the caller itself or one of its topology descendants — a
/// parent owns its whole subtree, and the root (`ruth`) covers every agent as
/// a consequence, with no positional/hardcoded privilege. Returns `Some(Err)`
/// to short-circuit the dispatch arm when it isn't, `None` when authorised.
/// `action` is the verb phrase for the message (e.g. `"start"`).
fn require_descendant(agent: &str, target: &str, action: &str) -> Option<Response> {
    if crate::topology::is_descendant_of(target, agent) {
        None
    } else {
        Some(Response::Err {
            message: format!(
                "agent `{agent}` cannot {action} `{target}`: \
                 not in its subtree (topology)"
            ),
        })
    }
}

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

/// Topology guard for `request_init_config`, which may legitimately target a
/// child that does not exist *yet* (seeding a brand-new sub-agent's config
/// repo). The caller may act on a
/// `target` that is EITHER already its direct child (re-init / config
/// update of an existing child) OR brand-new (absent from the topology
/// tree — the requester becomes its parent). A name that already
/// belongs to a *different* parent (or is a root agent) is refused so
/// one agent can't hijack another's sub-tree.
///
/// Also re-runs the agent-name format check (a traversal / malformed name
/// could never be a descendant): a brand-new name now flows straight to
/// `submit_init_config`, which builds filesystem paths from it, so validate
/// before that.
fn require_new_child(agent: &str, target: &str, action: &str) -> Option<Response> {
    if let Err(reason) = hive_types::Ident::parse(target) {
        return Some(Response::Err {
            message: format!("agent `{agent}` cannot {action} `{target}`: {reason}"),
        });
    }
    // brand-new name (absent from topology) — requester becomes the parent on
    // approval; allowed for any caller.
    if !crate::topology::read().contains_key(target) {
        return None;
    }
    // existing agent — allowed only if it's in the caller's subtree
    // (re-init / config update of an agent the caller owns; the root owns
    // every existing agent). Refuses an agent outside the caller's subtree
    // so one agent can't hijack another's config.
    if crate::topology::is_descendant_of(target, agent) {
        None
    } else {
        Some(Response::Err {
            message: format!(
                "agent `{agent}` cannot {action} `{target}`: it already exists \
                 outside its subtree in the topology tree"
            ),
        })
    }
}

/// `GetLooseEnds` — read the target's loose ends. `None` / own / a subtree
/// descendant resolve freely (a parent sees its subtree, the root sees all);
/// any other named agent needs `QueryAgentState`; `"*"` is a hive-wide sweep
/// gated on `QueryAgentState`.
fn handle_get_loose_ends(coord: &Arc<Coordinator>, agent: &str, target: Option<&str>) -> Response {
    let result = if target == Some("*") {
        if !crate::capabilities::has_cap(agent, hive_sh4re::Capability::QueryAgentState) {
            return Response::Err {
                message: "query_agent_state capability required for hive-wide loose ends"
                    .to_owned(),
            };
        }
        crate::loose_ends::hive_wide(coord)
    } else {
        match resolve_agent_state_target(agent, target) {
            Ok(name) => crate::loose_ends::for_agent(coord, name),
            Err(message) => return Response::Err { message },
        }
    };
    match result {
        Ok(loose_ends) => Response::LooseEnds { loose_ends },
        Err(e) => Response::Err {
            message: format!("{e:#}"),
        },
    }
}

/// `UpsertTodo` — a subsystem pushes/updates one of this agent's todos.
/// Coalesces a wake ONLY when the row is new or actually changed, so
/// re-pushing an identical keyed todo is a silent no-op.
fn handle_upsert_todo(
    coord: &Arc<Coordinator>,
    agent: &str,
    subsystem: &str,
    key: Option<&str>,
    summary: &str,
    source: Option<&str>,
) -> Response {
    match coord.todos.upsert(agent, subsystem, key, summary, source) {
        Ok((_, changed)) => {
            if changed {
                let _ = coord.broker.send(&Message {
                    from: "todo".to_owned(),
                    to: agent.to_owned(),
                    body: "you have todos — call get_loose_ends to see them".to_owned(),
                    in_reply_to: None,
                });
            }
            Response::Ok
        }
        Err(e) => Response::Err {
            message: format!("{e:#}"),
        },
    }
}

/// `ClearTodo` — a producer clears a resolved todo by `(subsystem, key)`,
/// or wipes its whole set when `all` (cancel-and-recreate on restart).
fn handle_clear_todo(
    coord: &Arc<Coordinator>,
    agent: &str,
    subsystem: &str,
    key: Option<&str>,
    all: bool,
) -> Response {
    let result = if all {
        coord.todos.clear_subsystem(agent, subsystem)
    } else {
        coord.todos.clear(agent, subsystem, key)
    };
    match result {
        Ok(count) => Response::Acked {
            count: u64::try_from(count).unwrap_or(0),
        },
        Err(e) => Response::Err {
            message: format!("{e:#}"),
        },
    }
}

/// `ListTodos` — enumerate this agent's todos (optionally one subsystem's)
/// as `LooseEnd::Todo` rows, so a producer can reconcile its own set.
fn handle_list_todos(coord: &Arc<Coordinator>, agent: &str, subsystem: Option<&str>) -> Response {
    match crate::loose_ends::todos_for(coord, agent, subsystem) {
        Ok(loose_ends) => Response::LooseEnds { loose_ends },
        Err(e) => Response::Err {
            message: format!("{e:#}"),
        },
    }
}

/// `MarkTodoDone` — the agent clears one of its own todos by id (scoped to
/// the agent, so it can't touch another agent's).
fn handle_mark_todo_done(coord: &Arc<Coordinator>, agent: &str, id: i64) -> Response {
    match coord.todos.mark_done(agent, id) {
        Ok(count) => Response::Acked {
            count: u64::try_from(count).unwrap_or(0),
        },
        Err(e) => Response::Err {
            message: format!("{e:#}"),
        },
    }
}

/// `CountPendingReminders` — resolve the target (own / subtree free, else
/// `QueryAgentState`) then count its pending reminders.
fn handle_count_pending_reminders(
    coord: &Arc<Coordinator>,
    agent: &str,
    target: Option<&str>,
) -> Response {
    match resolve_agent_state_target(agent, target) {
        Ok(name) => match coord.broker.count_pending_reminders_for(name) {
            Ok(count) => Response::PendingRemindersCount { count },
            Err(e) => Response::Err {
                message: format!("{e:#}"),
            },
        },
        Err(message) => Response::Err { message },
    }
}

/// `ReminderRollup` — resolve the target (own / subtree free, else
/// `QueryAgentState`) then roll up its reminders fired in the last
/// `since_secs`.
fn handle_reminder_rollup(
    coord: &Arc<Coordinator>,
    agent: &str,
    target: Option<&str>,
    since_secs: u64,
) -> Response {
    match resolve_agent_state_target(agent, target) {
        Ok(name) => match coord.broker.reminder_rollup_for(name, since_secs) {
            Ok(stats) => Response::ReminderRollup(stats),
            Err(e) => Response::Err {
                message: format!("{e:#}"),
            },
        },
        Err(message) => Response::Err { message },
    }
}

/// 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::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::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_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_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: from.to_owned(),
            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::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/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::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: agent.to_owned(),
        to: resolved,
        body: body.to_owned(),
        in_reply_to,
    }) {
        Ok(()) => Response::Ok,
        Err(e) => Response::Err {
            message: format!("{e:#}"),
        },
    }
}

/// `GetLogs` — read a child container's journal via hive-priv (the
/// `-M` read needs root). `journalctl -M` wants the `h-<name>` machine
/// name, which `container_name` derives.
async fn handle_get_logs(agent: &str, lines: Option<u32>) -> Response {
    let n = lines.unwrap_or(50);
    let machine = crate::lifecycle::container_name(agent);
    tracing::info!(%agent, %machine, %n, "manager: get_logs");
    match crate::priv_client::read_container_journal(
        &machine,
        hive_priv_sock::JournalQuery {
            lines: n,
            ..Default::default()
        },
    )
    .await
    {
        Ok((stdout, stderr)) => {
            let content = if stdout.is_empty() { stderr } else { stdout };
            Response::Logs { content }
        }
        Err(e) => Response::Err {
            message: format!("get_logs: {e:#}"),
        },
    }
}

/// On `Ask { ttl_seconds: Some(n) }`, sleep n seconds and then try to
/// resolve the question with `[expired]`. If the operator (or any
/// other path) already answered it, `answer()` returns Err and we
/// no-op silently. Otherwise fire a `QuestionAnswered` helper event
/// with `answerer = "ttl-watchdog"` so the asker can distinguish a
/// real answer from a deadline trip without parsing the answer text.
const TTL_SENTINEL: &str = "[expired]";
/// Synthetic `answerer` label used when the ttl watchdog resolves a
/// question instead of a real human / agent. Lives in a distinct
/// namespace from agent names + the operator so the asker can pattern
/// match `event.answerer == "ttl-watchdog"`.
const TTL_ANSWERER: &str = "ttl-watchdog";

pub fn spawn_question_watchdog(coord: &Arc<Coordinator>, id: i64, ttl_secs: u64) {
    let coord = coord.clone();
    tokio::spawn(async move {
        tokio::time::sleep(std::time::Duration::from_secs(ttl_secs)).await;
        // Watchdog has its own answerer label so the authorisation
        // check in `answer()` permits it for any target. We bypass
        // the public `answer()` path by calling it with the operator
        // identity, since the operator is always permitted; the
        // event we fire carries the real watchdog label for observers.
        if let Ok((question, asker, target)) =
            coord
                .questions
                .answer(id, TTL_SENTINEL, hive_sh4re::OPERATOR_RECIPIENT)
        {
            tracing::info!(%id, %asker, "question expired (ttl)");
            coord.notify_agent(
                &asker,
                &hive_sh4re::HelperEvent::QuestionAnswered {
                    id,
                    question,
                    answer: TTL_SENTINEL.to_owned(),
                    answerer: TTL_ANSWERER.to_owned(),
                },
            );
            coord.emit_question_resolved(id, TTL_SENTINEL, TTL_ANSWERER, false, target.as_deref());
        }
    });
}