//! Per-agent socket listener. Each socket file's existence on disk
//! authenticates the caller: connecting to `<.../agents/foo/mcp.sock>` means
//! you are `foo`.

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

use anyhow::{Context, Result};
use hive_sh4re::{AgentRequest, AgentResponse, Message};
use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
use tokio::net::{UnixListener, UnixStream};
use tokio::task::JoinHandle;

use crate::coordinator::Coordinator;

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

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

    let path = socket_path.to_path_buf();
    let handle = tokio::spawn(async move {
        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 })
}

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::<AgentRequest>(line.trim()) {
            Ok(req) => dispatch(&req, &agent, &coord).await,
            Err(e) => AgentResponse::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`. Bounds the size of a single
/// round-trip so a confused caller can't drain the entire inbox in
/// one go and blow past wire-buffer sizes; everything above the cap
/// silently clamps. 32 is comfortably above the burst sizes we've
/// seen in practice (post-rebuild rescue, multi-agent reply storms)
/// and well under the per-message `MESSAGE_MAX_BYTES` * N envelope
/// budget.
pub(crate) const RECV_BATCH_MAX: u32 = 32;

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

#[allow(clippy::too_many_lines)]
/// 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.
///
/// Both `agent_server::dispatch` and `manager_server::dispatch` call this
/// first; each then handles its own remaining arms.
pub(crate) async fn dispatch_shared(
    req: &hive_sh4re::Request,
    agent: &str,
    coord: &Arc<Coordinator>,
) -> Option<hive_sh4re::Response> {
    let broker = &coord.broker;
    Some(match req {
        hive_sh4re::Request::Send {
            to,
            body,
            in_reply_to,
        } => handle_send(coord, agent, to, body, *in_reply_to),
        hive_sh4re::Request::Recv { wait_seconds, max } => {
            let cap = max.unwrap_or(1).min(RECV_BATCH_MAX) as usize;
            match broker
                .recv_blocking_batch(agent, recv_timeout(*wait_seconds), cap)
                .await
            {
                Ok(deliveries) => hive_sh4re::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(),
                },
                Err(e) => hive_sh4re::Response::Err {
                    message: format!("{e:#}"),
                },
            }
        }
        hive_sh4re::Request::Status => match broker.count_pending(agent) {
            Ok(unread) => hive_sh4re::Response::Status { unread },
            Err(e) => hive_sh4re::Response::Err {
                message: format!("{e:#}"),
            },
        },
        hive_sh4re::Request::OperatorMsg { body } => match broker.send(&Message {
            from: hive_sh4re::OPERATOR_RECIPIENT.to_owned(),
            to: agent.to_owned(),
            body: body.clone(),
            in_reply_to: None,
        }) {
            Ok(()) => hive_sh4re::Response::Ok,
            Err(e) => hive_sh4re::Response::Err {
                message: format!("{e:#}"),
            },
        },
        hive_sh4re::Request::Wake {
            from,
            body,
            transient,
        } => {
            if *transient {
                // Transient wakes bypass sqlite — they fire the broadcast
                // channel only. No redelivery on restart; no message history
                // entry. Used by bash task completions.
                broker.ping(agent, from, body);
                hive_sh4re::Response::Ok
            } else {
                match broker.send(&Message {
                    from: from.clone(),
                    to: agent.to_owned(),
                    body: body.clone(),
                    in_reply_to: None,
                }) {
                    Ok(()) => hive_sh4re::Response::Ok,
                    Err(e) => hive_sh4re::Response::Err {
                        message: format!("{e:#}"),
                    },
                }
            }
        }
        hive_sh4re::Request::Recent { limit } => match broker.recent_for(agent, *limit) {
            Ok(rows) => hive_sh4re::Response::Recent { rows },
            Err(e) => hive_sh4re::Response::Err {
                message: format!("{e:#}"),
            },
        },
        hive_sh4re::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_sh4re::Response::Err { message },
            |id| hive_sh4re::Response::QuestionQueued { id },
        ),
        hive_sh4re::Request::Answer { id, answer } => {
            crate::questions::handle_answer(coord, agent, *id, answer).map_or_else(
                |message| hive_sh4re::Response::Err { message },
                |()| hive_sh4re::Response::Ok,
            )
        }
        hive_sh4re::Request::Remind {
            message,
            timing,
            file_path,
        } => handle_remind(coord, agent, message, timing, file_path.as_deref()),
        hive_sh4re::Request::SetStatus { text } => {
            if let Err(message) = crate::limits::check_status_text(text) {
                return Some(hive_sh4re::Response::Err { message });
            }
            // The harness writes the status file to its own `state/` dir
            // before sending this request (it runs as the agent user, so
            // it has write access). We just trigger a dashboard rescan so
            // the new value is reflected immediately.
            let coord2 = Arc::clone(coord);
            tokio::spawn(async move { coord2.rescan_containers_and_emit().await });
            hive_sh4re::Response::Ok
        }
        hive_sh4re::Request::GetAgentMeta { name } => {
            let target = name.as_deref().unwrap_or(agent);
            let (status_text, status_set_at, running) =
                crate::container_view::read_agent_status_live(target).await;
            let (hive_name, swarm_name) = crate::container_view::hive_swarm_names();
            hive_sh4re::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,
            }
        }
        hive_sh4re::Request::CancelLooseEnd { kind, id } => {
            crate::questions::handle_cancel_loose_end(coord, agent, *kind, *id).map_or_else(
                |message| hive_sh4re::Response::Err { message },
                |()| hive_sh4re::Response::Ok,
            )
        }
        hive_sh4re::Request::AckTurn => match broker.ack_turn(agent) {
            Ok(_n) => hive_sh4re::Response::Ok,
            Err(e) => hive_sh4re::Response::Err {
                message: format!("{e:#}"),
            },
        },
        hive_sh4re::Request::RequeueInflight => match broker.requeue_inflight(agent) {
            Ok(n) => {
                if n > 0 {
                    tracing::info!(%agent, requeued = %n, "requeued in-flight messages");
                }
                hive_sh4re::Response::Ok
            }
            Err(e) => hive_sh4re::Response::Err {
                message: format!("{e:#}"),
            },
        },
        hive_sh4re::Request::GetHostJournal {
            unit,
            container,
            lines,
            priority,
            grep,
            since,
            until,
        } => {
            dispatch_host_journal(agent, unit, container, lines, priority, grep, since, until).await
        }
        // Not a shared variant.
        _ => return None,
    })
}

#[allow(clippy::too_many_lines)]
async fn dispatch(req: &AgentRequest, agent: &str, coord: &Arc<Coordinator>) -> AgentResponse {
    if let Some(resp) = dispatch_shared(req, agent, coord).await {
        return resp;
    }
    match req {
        AgentRequest::GetLooseEnds { agent: target } => {
            let name = resolve_agent_state_target(agent, target.as_deref());
            match name {
                Ok(name) => match crate::loose_ends::for_agent(coord, name) {
                    Ok(loose_ends) => AgentResponse::LooseEnds { loose_ends },
                    Err(e) => AgentResponse::Err {
                        message: format!("{e:#}"),
                    },
                },
                Err(message) => AgentResponse::Err { message },
            }
        }
        AgentRequest::CountPendingReminders { agent: target } => {
            let name = resolve_agent_state_target(agent, target.as_deref());
            match name {
                Ok(name) => match coord.broker.count_pending_reminders_for(name) {
                    Ok(count) => AgentResponse::PendingRemindersCount { count },
                    Err(e) => AgentResponse::Err {
                        message: format!("{e:#}"),
                    },
                },
                Err(message) => AgentResponse::Err { message },
            }
        }
        AgentRequest::ReminderRollup {
            since_secs,
            agent: target,
        } => {
            let name = resolve_agent_state_target(agent, target.as_deref());
            match name {
                Ok(name) => match coord.broker.reminder_rollup_for(name, *since_secs) {
                    Ok(stats) => AgentResponse::ReminderRollup(stats),
                    Err(e) => AgentResponse::Err {
                        message: format!("{e:#}"),
                    },
                },
                Err(message) => AgentResponse::Err { message },
            }
        }
        AgentRequest::Start { name } => {
            if !crate::topology::children_of(agent)
                .iter()
                .any(|c| c == name)
            {
                return AgentResponse::Err {
                    message: format!(
                        "agent `{agent}` cannot start `{name}`: \
                         not a direct child in the topology tree"
                    ),
                };
            }
            tracing::info!(%agent, %name, "agent: start child");
            match crate::lifecycle::start(name).await {
                Ok(()) => {
                    coord.kick_agent(name, "container started");
                    AgentResponse::Ok
                }
                Err(e) => AgentResponse::Err {
                    message: format!("{e:#}"),
                },
            }
        }
        AgentRequest::Restart { name } => {
            // Topology check: the caller must be the direct parent of the
            // target. This is the only authorisation criterion — no
            // capability flag needed; parenthood is sufficient privilege.
            if !crate::topology::children_of(agent)
                .iter()
                .any(|c| c == name)
            {
                return AgentResponse::Err {
                    message: format!(
                        "agent `{agent}` cannot restart `{name}`: \
                         not a direct child in the topology tree"
                    ),
                };
            }
            tracing::info!(%agent, %name, "agent: enqueue restart for child");
            coord.rebuild_queue.enqueue(
                crate::rebuild_queue::QueueKind::Restart,
                name.to_owned(),
                crate::rebuild_queue::QueueSource::Manual,
                format!("agent `{agent}` restart tool"),
                None,
            );
            coord.emit_rebuild_queue_snapshot();
            AgentResponse::Ok
        }
        AgentRequest::Kill { name } => {
            if !crate::topology::children_of(agent)
                .iter()
                .any(|c| c == name)
            {
                return AgentResponse::Err {
                    message: format!(
                        "agent `{agent}` cannot kill `{name}`: \
                         not a direct child in the topology tree"
                    ),
                };
            }
            tracing::info!(%agent, %name, "agent: kill child");
            let result: anyhow::Result<()> = async {
                crate::lifecycle::kill(name).await?;
                coord.unregister_agent(name);
                Ok(())
            }
            .await;
            match result {
                Ok(()) => {
                    coord.notify_manager(&hive_sh4re::HelperEvent::Killed {
                        agent: name.clone(),
                    });
                    AgentResponse::Ok
                }
                Err(e) => AgentResponse::Err {
                    message: format!("{e:#}"),
                },
            }
        }
        AgentRequest::Update { name } => {
            if !crate::topology::children_of(agent)
                .iter()
                .any(|c| c == name)
            {
                return AgentResponse::Err {
                    message: format!(
                        "agent `{agent}` cannot rebuild `{name}`: \
                         not a direct child in the topology tree"
                    ),
                };
            }
            tracing::info!(%agent, %name, "agent: enqueue rebuild for child");
            coord.rebuild_queue.enqueue(
                crate::rebuild_queue::QueueKind::Rebuild,
                name.to_owned(),
                crate::rebuild_queue::QueueSource::Manual,
                format!("agent `{agent}` update tool"),
                None,
            );
            coord.emit_rebuild_queue_snapshot();
            AgentResponse::Ok
        }
        AgentRequest::ListDescendants => {
            tracing::debug!(%agent, "agent: list descendants");
            // All containers known to nixos-container (running only).
            let running_set: std::collections::HashSet<String> =
                match crate::lifecycle::list().await {
                    Ok(names) => names
                        .into_iter()
                        .filter_map(|c| {
                            c.strip_prefix(crate::lifecycle::AGENT_PREFIX)
                                .map(str::to_owned)
                        })
                        .collect(),
                    Err(e) => {
                        return AgentResponse::Err {
                            message: format!("list containers failed: {e:#}"),
                        };
                    }
                };
            // Walk the full topology and collect every descendant.
            let topo = crate::topology::read();
            let mut names: Vec<String> = topo
                .keys()
                .filter(|name| crate::topology::is_descendant_of(name, agent))
                .cloned()
                .collect();
            // Parents before children, then alpha within each tier.
            crate::auto_update::topology_sort(&mut names, &topo);
            let containers = names
                .into_iter()
                .map(|name| {
                    let running = running_set.contains(&name);
                    hive_sh4re::ContainerInfo { name, running }
                })
                .collect();
            AgentResponse::Containers { containers }
        }
        AgentRequest::RequestInitConfig { name, description } => {
            if !crate::topology::children_of(agent)
                .iter()
                .any(|c| c == name)
            {
                return AgentResponse::Err {
                    message: format!(
                        "agent `{agent}` cannot request_init_config for `{name}`: \
                         not a direct child in the topology tree"
                    ),
                };
            }
            tracing::info!(%agent, %name, "agent: request_init_config for child");
            match crate::manager_server::submit_init_config(coord, name, description.clone()) {
                Ok(_id) => AgentResponse::Ok,
                Err(e) => AgentResponse::Err {
                    message: format!("{e:#}"),
                },
            }
        }
        AgentRequest::RequestApplyCommit {
            agent: target_agent,
            commit_ref,
            description,
        } => {
            if !crate::topology::children_of(agent)
                .iter()
                .any(|c| c == target_agent)
            {
                return AgentResponse::Err {
                    message: format!(
                        "agent `{agent}` cannot request_apply_commit for `{target_agent}`: \
                         not a direct child in the topology tree"
                    ),
                };
            }
            tracing::info!(%agent, %target_agent, %commit_ref, "agent: request_apply_commit for child");
            match crate::manager_server::submit_apply_commit(
                coord,
                target_agent,
                commit_ref,
                description.as_deref(),
            )
            .await
            {
                Ok((id, sha)) => {
                    tracing::info!(%id, %target_agent, %sha, "agent: apply_commit approval queued");
                    AgentResponse::Ok
                }
                Err(e) => AgentResponse::Err {
                    message: format!("{e:#}"),
                },
            }
        }
        // Manager-only variants are not valid on the agent socket.
        _ => AgentResponse::Err {
            message: "request not supported on agent socket".to_owned(),
        },
    }
}

/// 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.
#[allow(clippy::too_many_arguments)]
pub async fn dispatch_host_journal(
    agent: &str,
    unit: &Option<String>,
    container: &Option<String>,
    lines: &Option<u32>,
    priority: &Option<hive_sh4re::JournalPriority>,
    grep: &Option<String>,
    since: &Option<String>,
    until: &Option<String>,
) -> AgentResponse {
    if !crate::capabilities::has_cap(agent, hive_sh4re::Capability::ReadHostJournal) {
        return AgentResponse::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,
            n,
            false,
            hive_sh4re::priv_proto::JournalOutput::Short,
            unit.clone(),
            priority.as_ref().map(|p| p.as_str().to_owned()),
            grep.clone(),
            since.clone(),
            until.clone(),
        )
        .await
        {
            Ok((stdout, stderr)) => {
                let content = if stdout.is_empty() { stderr } else { stdout };
                AgentResponse::HostJournal { content }
            }
            Err(e) => AgentResponse::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)
            };
            AgentResponse::HostJournal { content }
        }
        Err(e) => AgentResponse::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` (and via it, `manager_server`) can use it.
pub(crate) fn handle_send(
    coord: &Arc<Coordinator>,
    agent: &str,
    to: &str,
    body: &str,
    in_reply_to: Option<i64>,
) -> AgentResponse {
    if let Err(message) = crate::limits::check_size("send", body) {
        return AgentResponse::Err { message };
    }
    if to == "*" {
        let errors = coord.broadcast_send(agent, body);
        return if errors.is_empty() {
            AgentResponse::Ok
        } else {
            AgentResponse::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() {
            AgentResponse::Ok
        } else {
            AgentResponse::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 (issue #1165).
    //
    // 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 AgentResponse::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 {
        let state_root = crate::coordinator::Coordinator::agent_state_root(&resolved);
        if !state_root.exists() {
            return AgentResponse::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(()) => AgentResponse::Ok,
        Err(e) => AgentResponse::Err {
            message: format!("{e:#}"),
        },
    }
}

fn handle_remind(
    coord: &Arc<Coordinator>,
    agent: &str,
    message: &str,
    timing: &hive_sh4re::ReminderTiming,
    file_path: Option<&str>,
) -> AgentResponse {
    match store_remind(coord, agent, message, timing, file_path) {
        Ok(()) => AgentResponse::Ok,
        Err(message) => AgentResponse::Err { message },
    }
}

/// Shared remind-storage path used by both the agent and the manager
/// dispatchers. Validates timing, applies the auto-file overflow
/// dance (see [`prepare_remind_storage`]), and writes the reminder
/// row. Returns `Ok(())` on success, or a caller-ready error string
/// the dispatcher wraps in `*Response::Err`.
/// Maximum pending (un-delivered) reminders per agent. Exceeding this
/// causes `store_remind` to return an error so the agent knows to back
/// off instead of silently dropping. Override via
/// `HIVE_REMIND_MAX_PENDING_PER_AGENT`; set to `0` to disable the cap
/// (not recommended — a runaway agent can still flood the scheduler).
const DEFAULT_REMIND_MAX_PENDING: u64 = 50;

fn remind_max_pending() -> u64 {
    std::env::var("HIVE_REMIND_MAX_PENDING_PER_AGENT")
        .ok()
        .and_then(|s| s.trim().parse::<u64>().ok())
        .unwrap_or(DEFAULT_REMIND_MAX_PENDING)
}

pub(crate) fn store_remind(
    coord: &Arc<Coordinator>,
    agent: &str,
    message: &str,
    timing: &hive_sh4re::ReminderTiming,
    file_path: Option<&str>,
) -> Result<(), String> {
    let max = remind_max_pending();
    if max > 0 {
        let pending = coord.broker.count_pending_reminders_for(agent).unwrap_or(0);
        if pending >= max {
            return Err(format!(
                "reminder rejected: agent `{agent}` already has {pending} pending \
                 reminders (cap {max}). Cancel some via `cancel_loose_end` or wait \
                 for them to fire before scheduling more. Override the cap with \
                 `HIVE_REMIND_MAX_PENDING_PER_AGENT`."
            ));
        }
    }
    let due_at = resolve_due_at(timing).map_err(|e| format!("invalid reminder timing: {e:#}"))?;
    let (stored_message, stored_path) = prepare_remind_storage(agent, message, file_path)?;
    let id = coord
        .broker
        .store_reminder(agent, &stored_message, stored_path.as_deref(), due_at)
        .map_err(|e| format!("failed to store reminder: {e:#}"))?;
    tracing::info!(%id, %agent, %due_at, "reminder scheduled");
    coord.emit_reminders_snapshot();
    Ok(())
}

/// Decide what we actually store in the reminders row, applying the
/// same byte cap as the rest of the wire protocol
/// ([`crate::limits::MESSAGE_MAX_BYTES`]). Three outcomes:
///
/// 1. Body within the cap → stored verbatim, with whatever `file_path`
///    the caller passed (None or Some). The scheduler honours
///    `file_path` at delivery time as before.
/// 2. Body over the cap, no caller `file_path` → auto-generate a path
///    under `/agents/<agent>/state/reminders/auto-<ts>.md`, write the
///    body to disk now, store a short pointer hint as the message and
///    clear `file_path` (so the scheduler doesn't re-write at
///    delivery and overwrite the body with the hint).
/// 3. Body over the cap, caller provided `file_path` → honour the
///    caller's path: write the body to it now, store the same hint
///    and clear `file_path` for the same reason as (2).
///
/// Returns `(stored_message, stored_file_path)` on success, or a
/// caller-ready error string on auto-save failure (which is the only
/// way a Remind request can be refused for size — the agent never has
/// to think about the cap).
fn prepare_remind_storage(
    agent: &str,
    message: &str,
    file_path: Option<&str>,
) -> Result<(String, Option<String>), String> {
    if message.len() <= crate::limits::MESSAGE_MAX_BYTES {
        return Ok((message.to_owned(), file_path.map(str::to_owned)));
    }
    let req_path = match file_path {
        Some(p) => p.to_owned(),
        None => auto_reminder_path(agent),
    };
    let host_path = crate::reminder_scheduler::resolve_host_path(agent, &req_path)
        .map_err(|reason| format!("auto-save path `{req_path}` rejected: {reason}"))?;
    crate::reminder_scheduler::write_payload(agent, &host_path, message).map_err(|reason| {
        format!("auto-save of large reminder body to `{req_path}` failed: {reason}")
    })?;
    let hint = format!(
        "[reminder body of {} bytes auto-saved to `{req_path}`; read with your filesystem tools]",
        message.len()
    );
    Ok((hint, None))
}

/// Generate a per-agent path for an auto-saved reminder body. Uses
/// `unix_nanos` plus the agent name to keep collisions infinitesimal
/// across the agent's own state subtree (we're not stamping a hostname
/// since hive-c0re is single-host).
fn auto_reminder_path(agent: &str) -> String {
    let ts_ns = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map_or(0, |d| d.as_nanos());
    format!("/agents/{agent}/state/reminders/auto-{ts_ns}.md")
}

/// Resolve the target agent name for `GetLooseEnds`, `CountPendingReminders`,
/// and `ReminderRollup` on the agent socket. Rules:
///
/// - `None` → caller's own threads (always allowed).
/// - `Some(caller)` → same as `None`.
/// - `Some("<child>")` where child is a direct descendant of caller per
///   `topology.json` → allowed without any extra capability.
/// - `Some("<other>")` where other is not a child → requires the
///   `query_agent_state` capability; returns an error otherwise.
/// - `Some("*")` → always rejected (hive-wide scans are manager-only).
fn resolve_agent_state_target<'a>(
    caller: &'a str,
    target: Option<&'a str>,
) -> Result<&'a str, String> {
    match target {
        None => Ok(caller),
        Some("*") => Err(
            "hive-wide query (agent=\"*\") is not available on the agent socket; \
             use the manager socket for swarm-wide scans"
                .to_owned(),
        ),
        Some(name) => {
            if name == caller {
                return Ok(caller);
            }
            // Direct children are visible to their parent without extra capability.
            if crate::topology::children_of(caller)
                .iter()
                .any(|c| c == name)
            {
                return Ok(name);
            }
            if crate::capabilities::has_cap(caller, hive_sh4re::Capability::QueryAgentState) {
                Ok(name)
            } else {
                Err(format!(
                    "agent `{caller}` cannot query `{name}`: not a direct child and \
                     `query_agent_state` capability is not granted"
                ))
            }
        }
    }
}

/// Resolve the `due_at` unix timestamp for a Remind request. Returns
/// distinct error messages for each failure mode (overflow on
/// `InSeconds`, pre-epoch clock, `i64` cast wrap) so the caller can tell
/// what went wrong without inspecting the chain.
fn resolve_due_at(timing: &hive_sh4re::ReminderTiming) -> anyhow::Result<i64> {
    use hive_sh4re::ReminderTiming;
    match timing {
        ReminderTiming::InSeconds { seconds } => {
            let now = std::time::SystemTime::now();
            let future = now
                .checked_add(std::time::Duration::from_secs(*seconds))
                .ok_or_else(|| {
                    anyhow::anyhow!("InSeconds overflow: {seconds}s exceeds system time range")
                })?;
            let duration = future
                .duration_since(std::time::UNIX_EPOCH)
                .map_err(|e| anyhow::anyhow!("system time before UNIX_EPOCH: {e}"))?;
            i64::try_from(duration.as_secs())
                .map_err(|e| anyhow::anyhow!("unix timestamp exceeds i64 range: {e}"))
        }
        ReminderTiming::At { unix_timestamp } => Ok(*unix_timestamp),
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn auto_reminder_path_format() {
        let p = auto_reminder_path("damocles");
        assert!(p.starts_with("/agents/damocles/state/reminders/auto-"));
        assert!(
            std::path::Path::new(&p)
                .extension()
                .is_some_and(|ext| ext.eq_ignore_ascii_case("md"))
        );
    }

    #[test]
    fn prepare_remind_storage_passthrough_under_cap() {
        let (msg, fp) = prepare_remind_storage("foo", "small body", None).unwrap();
        assert_eq!(msg, "small body");
        assert_eq!(fp, None);
    }

    #[test]
    fn prepare_remind_storage_passthrough_with_caller_file_path() {
        let (msg, fp) =
            prepare_remind_storage("foo", "small", Some("/agents/foo/state/x.md")).unwrap();
        assert_eq!(msg, "small");
        assert_eq!(fp.as_deref(), Some("/agents/foo/state/x.md"));
    }
}