diff --git a/hive-c0re/src/dashboard.rs b/hive-c0re/src/dashboard/mod.rs similarity index 100% rename from hive-c0re/src/dashboard.rs rename to hive-c0re/src/dashboard/mod.rs diff --git a/hive-c0re/src/lifecycle.rs b/hive-c0re/src/lifecycle/mod.rs similarity index 100% rename from hive-c0re/src/lifecycle.rs rename to hive-c0re/src/lifecycle/mod.rs diff --git a/hive-c0re/src/socket_server.rs b/hive-c0re/src/socket_server.rs deleted file mode 100644 index f49b0ea6..00000000 --- a/hive-c0re/src/socket_server.rs +++ /dev/null @@ -1,2195 +0,0 @@ -//! 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_sh4re::{AgentRequest, AgentResponse, MANAGER_AGENT, 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) -> Result { - 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) -> Result<()> { - use std::os::unix::fs::PermissionsExt as _; - let dir = Coordinator::agent_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) -> 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::(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` — 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) -> 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_sh4re::Request, - agent: &str, - coord: &Arc, -) -> Option { - 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 } => { - handle_recv(coord, agent, *wait_seconds, *max).await - } - hive_sh4re::Request::Status => handle_status(coord, agent), - hive_sh4re::Request::OperatorMsg { body } => handle_operator_msg(coord, agent, body), - hive_sh4re::Request::Wake { from, body } => handle_wake(coord, agent, from, body), - hive_sh4re::Request::Recent { limit } => handle_recent(coord, agent, *limit), - 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 } => handle_set_status(coord, text), - hive_sh4re::Request::GetAgentMeta { name } => { - handle_get_agent_meta(coord, agent, name.as_deref()).await - } - 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::CreateRepo { repo } => handle_create_repo(agent, repo).await, - hive_sh4re::Request::AckTurn => handle_ack_turn(coord, agent), - hive_sh4re::Request::AckUntil { up_to } => handle_ack_until(coord, agent, *up_to), - hive_sh4re::Request::RequeueInflight => handle_requeue_inflight(coord, agent), - hive_sh4re::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_sh4re::Response::Ok - } - hive_sh4re::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, - agent: &str, - wait_seconds: Option, - max: Option, -) -> hive_sh4re::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_sh4re::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) => 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:#}"), - }, - } -} - -/// `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, - agent: &str, - from: &str, - body: &str, -) -> hive_sh4re::Response { - match coord.broker.send(&Message { - from: from.to_owned(), - to: agent.to_owned(), - body: body.to_owned(), - in_reply_to: None, - }) { - Ok(()) => hive_sh4re::Response::Ok, - Err(e) => hive_sh4re::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, text: &str) -> hive_sh4re::Response { - if let Err(message) = crate::limits::check_status_text(text) { - return hive_sh4re::Response::Err { message }; - } - let coord2 = Arc::clone(coord); - tokio::spawn(async move { coord2.rescan_containers_and_emit().await }); - hive_sh4re::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_sh4re::Response { - if !valid_repo_name(repo) { - return hive_sh4re::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_sh4re::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_sh4re::Response::RepoCreated { - clone_url: format!("{}/{full_name}.git", crate::forge::FORGE_HTTP), - full_name, - }, - Err(e) => hive_sh4re::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, - agent: &str, - name: Option<&str>, -) -> hive_sh4re::Response { - let target = name.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, - matrix_accounts: read_agent_matrix_identities(target), - } -} - -/// 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: &str) -> Vec { - 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::>(&s).ok()) - .unwrap_or_default() -} - -/// `Status` — count of pending (unread) inbox messages for `agent`. -fn handle_status(coord: &Arc, agent: &str) -> hive_sh4re::Response { - match coord.broker.count_pending(agent) { - Ok(unread) => hive_sh4re::Response::Status { unread }, - Err(e) => hive_sh4re::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, agent: &str, body: &str) -> hive_sh4re::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_sh4re::Response::Ok, - Err(e) => hive_sh4re::Response::Err { - message: format!("{e:#}"), - }, - } -} - -/// `Recent` — the last `limit` inbox rows for `agent` (read-only, -/// doesn't consume). -fn handle_recent(coord: &Arc, agent: &str, limit: u64) -> hive_sh4re::Response { - match coord.broker.recent_for(agent, limit) { - Ok(rows) => hive_sh4re::Response::Recent { rows }, - Err(e) => hive_sh4re::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, agent: &str) -> hive_sh4re::Response { - match coord.broker.ack_turn(agent) { - Ok(_n) => hive_sh4re::Response::Ok, - Err(e) => hive_sh4re::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, agent: &str, up_to: i64) -> hive_sh4re::Response { - match coord.broker.ack_until(agent, up_to) { - Ok(count) => hive_sh4re::Response::Acked { count }, - Err(e) => hive_sh4re::Response::Err { - message: format!("{e:#}"), - }, - } -} - -/// `RequeueInflight` — resurface `agent`'s unacked in-flight messages -/// (crash recovery on harness boot). -fn handle_requeue_inflight(coord: &Arc, agent: &str) -> hive_sh4re::Response { - match coord.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:#}"), - }, - } -} - -/// 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: &AgentRequest, agent: &str, coord: &Arc) -> AgentResponse { - 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). - AgentRequest::Start { name } => handle_start(coord, agent, name), - AgentRequest::Restart { name } => handle_restart(coord, agent, name).await, - AgentRequest::Kill { name } => handle_kill(coord, agent, name).await, - AgentRequest::Update { name } => handle_update(coord, agent, name), - AgentRequest::ListDescendants => handle_list_descendants(agent).await, - AgentRequest::RequestInitConfig { name, description } => { - handle_request_init_config(coord, agent, name, description.clone()) - } - AgentRequest::RequestApplyCommit { - agent: target_agent, - commit_ref, - description, - } => { - handle_request_apply_commit( - coord, - agent, - target_agent, - commit_ref, - description.as_deref(), - ) - .await - } - // Agent-state queries: own subtree is free; other agents + the - // hive-wide `"*"` sweep require `QueryAgentState`. - AgentRequest::GetLooseEnds { agent: target } => { - handle_get_loose_ends(coord, agent, target.as_deref()) - } - AgentRequest::CountPendingReminders { agent: target } => { - handle_count_pending_reminders(coord, agent, target.as_deref()) - } - AgentRequest::ReminderRollup { - since_secs, - agent: target, - } => handle_reminder_rollup(coord, agent, target.as_deref(), *since_secs), - // 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: &AgentRequest, - agent: &str, - coord: &Arc, -) -> AgentResponse { - match req { - AgentRequest::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()) - } - AgentRequest::RequestSchedulePrompt(payload) => { - if let Some(err) = require_group(agent, "scheduling", "schedule a prompt") { - return err; - } - handle_request_schedule_prompt(coord, agent, payload) - } - AgentRequest::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()) - } - AgentRequest::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(), - }, - ) - } - AgentRequest::ListSchedules => { - if let Some(err) = require_group(agent, "scheduling", "list schedules") { - return err; - } - handle_list_schedules(coord) - } - AgentRequest::FireScheduleNow { id } => { - if let Some(err) = require_group(agent, "scheduling", "fire a schedule") { - return err; - } - handle_fire_schedule_now(coord, agent, *id).await - } - AgentRequest::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. - _ => AgentResponse::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 { - if crate::topology::is_descendant_of(target, agent) { - None - } else { - Some(AgentResponse::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 { - if crate::tool_groups::groups_for(agent) - .iter() - .any(|g| g == group) - { - None - } else { - Some(AgentResponse::Err { - message: format!("agent `{agent}` cannot {action}: requires the `{group}` tool group"), - }) - } -} - -/// Topology guard for `request_init_config` / `request_apply_commit`, -/// which may legitimately target a child that does not exist *yet* -/// (spawning a brand-new sub-agent). 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 { - if let Some(reason) = crate::dashboard::validate_agent_name(target) { - return Some(AgentResponse::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(AgentResponse::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, - agent: &str, - target: Option<&str>, -) -> AgentResponse { - let result = if target == Some("*") { - if !crate::capabilities::has_cap(agent, hive_sh4re::Capability::QueryAgentState) { - return AgentResponse::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 AgentResponse::Err { message }, - } - }; - match result { - Ok(loose_ends) => AgentResponse::LooseEnds { loose_ends }, - Err(e) => AgentResponse::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, - agent: &str, - target: Option<&str>, -) -> AgentResponse { - match resolve_agent_state_target(agent, target) { - 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 }, - } -} - -/// `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, - agent: &str, - target: Option<&str>, - since_secs: u64, -) -> AgentResponse { - match resolve_agent_state_target(agent, target) { - 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 }, - } -} - -/// `Start` — start a container, kicking its next turn. The caller must be an -/// ancestor of `name` in the topology (the root covers every agent). -fn handle_start(coord: &Arc, agent: &str, name: &str) -> AgentResponse { - if let Some(err) = require_descendant(agent, name, "start") { - return err; - } - tracing::info!(%agent, %name, "start container"); - // Persist `wanted = Up` and submit the Start DAG; the submit layer - // upgrades a stale-rev start to a full rebuild so the container - // runs current nix derivations before it starts. - crate::job_queue::submit::start( - coord, - name, - crate::job_queue::Source::Manual, - format!("agent `{agent}` start tool"), - ); - AgentResponse::Ok -} - -/// `Restart` — enqueue a restart for a container. The caller must be an -/// ancestor of `name` in the topology. The infra-container branch is -/// orthogonal: it is gated on the `infra_admin` capability and audited, so it -/// stays ahead of the topology guard. -async fn handle_restart(coord: &Arc, agent: &str, name: &str) -> AgentResponse { - // Infra-container restart: an agent holding the `infra_admin` - // capability can restart a hive infrastructure container (hive-ci / - // hive-gateway / hive-forge / hive-matrix) by passing its name to the - // same restart tool. The `InfraContainer` enum parse both recognises - // these (never agent children, so disjoint from the child path below) - // and yields the typed value the restart path needs. - if let Ok(container) = name.parse::() { - return handle_restart_infra(coord, agent, container).await; - } - if let Some(err) = require_descendant(agent, name, "restart") { - return err; - } - tracing::info!(%agent, %name, "submit restart"); - crate::job_queue::submit::restart( - coord, - name, - crate::job_queue::Source::Manual, - format!("agent `{agent}` restart tool"), - ); - AgentResponse::Ok -} - -/// Restart a hive infrastructure container on behalf of an agent that -/// holds the `infra_admin` capability. The `container` is already a valid -/// [`InfraContainer`] (the caller parsed it); this gates on the capability -/// and routes the systemctl restart through hive-priv. Direct, not -/// approval-gated. -async fn handle_restart_infra( - coord: &Arc, - agent: &str, - container: hive_sh4re::priv_proto::InfraContainer, -) -> AgentResponse { - let name = container.unit_name(); - // Record the attempt in the operator-visible privileged-action audit - // trail, then emit a live `AuditEntryAdded` so the dashboard audit view - // appends it off `/dashboard/stream`. Best-effort: `record` returns the - // canonical row (or `None` on a sqlite blip), and we stream exactly that - // row so the stored + streamed views can't drift. `action` is stable so - // the dashboard can group/filter. - let audit = |outcome: crate::audit_log::AuditOutcome, detail: Option<&str>| { - if let Some(entry) = coord - .audit_log - .record(agent, "restart_infra", name, outcome, detail) - { - coord.emit_audit_entry(entry); - } - }; - if !crate::capabilities::has_cap(agent, hive_sh4re::Capability::InfraAdmin) { - tracing::warn!(%agent, %name, "agent: infra restart denied (no infra_admin capability)"); - audit( - crate::audit_log::AuditOutcome::Err, - Some("denied: missing infra_admin capability"), - ); - return AgentResponse::Err { - message: format!( - "restarting infra container `{name}` requires the `infra_admin` capability" - ), - }; - } - tracing::info!(%agent, %name, "agent: restart infra container"); - match crate::priv_client::restart_infra_container(container).await { - Ok(()) => { - audit(crate::audit_log::AuditOutcome::Ok, None); - AgentResponse::Ok - } - Err(e) => { - let msg = format!("{e:#}"); - audit(crate::audit_log::AuditOutcome::Err, Some(&msg)); - AgentResponse::Err { message: msg } - } - } -} - -/// `Kill` — kill a container, unregister it, notify the manager. The caller -/// must be an ancestor of `name` in the topology. -async fn handle_kill(coord: &Arc, agent: &str, name: &str) -> AgentResponse { - if let Some(err) = require_descendant(agent, name, "kill") { - return err; - } - tracing::info!(%agent, %name, "kill container"); - // Persist the intent even if the kill fails — otherwise the next - // reconcile would restart the container. - if let Err(e) = coord.power.set(name, crate::power::Wanted::Offline) { - tracing::warn!(%name, error = ?e, "agent_power: set wanted=offline failed"); - } - 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.to_owned(), - }); - AgentResponse::Ok - } - Err(e) => AgentResponse::Err { - message: format!("{e:#}"), - }, - } -} - -/// `Update` — enqueue a rebuild for a container. The caller must be an -/// ancestor of `name` in the topology. -fn handle_update(coord: &Arc, agent: &str, name: &str) -> AgentResponse { - if let Some(err) = require_descendant(agent, name, "rebuild") { - return err; - } - tracing::info!(%agent, %name, "submit rebuild"); - crate::job_queue::submit::rebuild( - coord, - name, - crate::job_queue::Source::Manual, - format!("agent `{agent}` update tool"), - ); - AgentResponse::Ok -} - -/// `ListDescendants` — every topological descendant of `agent` with -/// its running/stopped state, parents before children. -async fn handle_list_descendants(agent: &str) -> AgentResponse { - tracing::debug!(%agent, "agent: list descendants"); - // All containers known to nixos-container (running only). - let running_set: std::collections::HashSet = 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 = 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 } -} - -/// `RequestInitConfig` — queue an `InitConfig` approval for an agent. The -/// `name` must be brand-new (absent from the topology) or already in the -/// caller's subtree; the requester is recorded as the new agent's parent (the -/// root requesting a new agent → a top-level agent, matching reconcile's -/// default). -fn handle_request_init_config( - coord: &Arc, - agent: &str, - name: &str, - description: Option, -) -> AgentResponse { - if let Some(err) = require_new_child(agent, name, "request_init_config for") { - return err; - } - tracing::info!(%agent, %name, "request_init_config"); - match submit_init_config(coord, name, Some(agent), description) { - Ok(_id) => AgentResponse::Ok, - Err(e) => AgentResponse::Err { - message: format!("{e:#}"), - }, - } -} - -/// `RequestApplyCommit` — queue an apply-commit approval for an agent. The -/// target must be in the caller's subtree (the root covers every agent). -async fn handle_request_apply_commit( - coord: &Arc, - agent: &str, - target_agent: &str, - commit_ref: &str, - description: Option<&str>, -) -> AgentResponse { - if let Some(err) = require_new_child(agent, target_agent, "request_apply_commit for") { - return err; - } - tracing::info!(%agent, %target_agent, %commit_ref, "request_apply_commit"); - match submit_apply_commit(coord, target_agent, commit_ref, description, agent).await { - Ok((id, sha)) => { - tracing::info!(%id, %target_agent, %sha, "apply_commit approval queued"); - AgentResponse::Ok - } - Err(e) => AgentResponse::Err { - message: format!("{e:#}"), - }, - } -} - -/// 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, - pub container: &'a Option, - pub lines: &'a Option, - pub priority: &'a Option, - pub grep: &'a Option, - pub since: &'a Option, - pub until: &'a Option, -} - -/// 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<'_>) -> AgentResponse { - let HostJournalArgs { - unit, - container, - lines, - priority, - grep, - since, - until, - } = args; - 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, - hive_sh4re::priv_proto::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 }; - AgentResponse::HostJournal { content } - } - Err(e) => AgentResponse::Err { - message: format!("journal read: {e:#}"), - }, - }; - } - - let mut args: Vec = 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 `": "` strings for any delivery -/// failures (empty = all good). -pub(crate) fn fan_out_send( - coord: &Arc, - from: &str, - body: &str, - in_reply_to: Option, - targets: &[String], -) -> Vec { - 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 == ""`) / 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, - agent: &str, - to: &str, - body: &str, - in_reply_to: Option, -) -> 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(", ")), - } - }; - } - // ``: 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 (``) 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 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, - 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::().ok()) - .unwrap_or(DEFAULT_REMIND_MAX_PENDING) -} - -pub(crate) fn store_remind( - coord: &Arc, - 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//state/reminders/auto-.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> { - 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 a *named* `GetLooseEnds` / -/// `CountPendingReminders` / `ReminderRollup` query. Rules: -/// -/// - `None` (or `Some(caller)`) → the caller's own threads (always allowed). -/// - `Some(descendant)` in the caller's subtree (the root's subtree is the whole hive) → allowed, no extra capability. -/// - `Some(other)` outside the subtree → requires the `query_agent_state` capability; error otherwise. -/// - `Some("*")` → rejected here; the hive-wide sweep is handled by `handle_get_loose_ends` under the same `query_agent_state` gate. -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 only valid for loose-ends; \ - not available for this query" - .to_owned(), - ), - Some(name) => { - // Own subtree (the root covers all) is visible without extra - // capability; `is_descendant_of` returns true for `name == caller`. - if crate::topology::is_descendant_of(name, caller) { - return Ok(name); - } - if crate::capabilities::has_cap(caller, hive_sh4re::Capability::QueryAgentState) { - Ok(name) - } else { - Err(format!( - "agent `{caller}` cannot query `{name}`: not in its subtree 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 { - 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), - } -} - -// --------------------------------------------------------------------------- -// Orchestration handlers + submit/schedule helpers. -// The schedule / meta-input handlers are reached via the tool-group-gated arms -// in `dispatch_orchestration`; `submit_init_config` / `submit_apply_commit` -// are re-used by the lifecycle handlers; `schedule_to_wire_public` / -// `filter_ghost_schedule_targets` are re-used by the dashboard. -// --------------------------------------------------------------------------- - -/// `RequestUpdateMetaInputs` — queue an `UpdateMetaInputs` approval -/// carrying the JSON-encoded input list in `commit_ref` (no git commit -/// is involved; the field is the payload the approval handler decodes). -fn handle_request_update_meta_inputs( - coord: &Arc, - requester: &str, - inputs: &[String], - description: Option<&str>, -) -> AgentResponse { - let label = if inputs.is_empty() { - "all inputs".to_string() - } else { - inputs.join(", ") - }; - tracing::info!(%requester, %label, "request_update_meta_inputs"); - let commit_ref = serde_json::to_string(inputs).unwrap_or_default(); - let id = match coord - .approvals - .submit_kind( - requester, - hive_sh4re::ApprovalKind::UpdateMetaInputs, - &commit_ref, - description, - requester, - ) - .map_err(|e| anyhow::anyhow!("{e:#}")) - { - Ok(id) => id, - Err(e) => { - return AgentResponse::Err { - message: format!("queue update_meta_inputs approval: {e:#}"), - }; - } - }; - tracing::info!(%id, %label, "update_meta_inputs approval queued"); - coord.emit_approval_added(crate::coordinator::ApprovalAdded { - id, - agent: requester, - approval_kind: "update_meta_inputs", - sha_short: None, - diff: None, - description: description.map(str::to_owned), - pr_number: None, - }); - AgentResponse::Ok -} - -/// `ListSchedules` — snapshot every scheduled prompt onto the wire. -fn handle_list_schedules(coord: &Arc) -> AgentResponse { - match coord.scheduled_prompts.list() { - Ok(schedules) => AgentResponse::Schedules { - schedules: schedules.into_iter().map(schedule_to_wire).collect(), - }, - Err(e) => AgentResponse::Err { - message: format!("list scheduled prompts: {e:#}"), - }, - } -} - -/// `GetLogs` — read a child container's journal via hive-priv (the -/// `-M` read needs root). `journalctl -M` wants the `h-` machine -/// name, which `container_name` derives. -async fn handle_get_logs(agent: &str, lines: Option) -> AgentResponse { - 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_sh4re::priv_proto::JournalQuery { - lines: n, - ..Default::default() - }, - ) - .await - { - Ok((stdout, stderr)) => { - let content = if stdout.is_empty() { stderr } else { stdout }; - AgentResponse::Logs { content } - } - Err(e) => AgentResponse::Err { - message: format!("get_logs: {e:#}"), - }, - } -} - -/// Submit a `RequestSchedulePrompt` payload as an `ApprovalKind::SchedulePrompt` -/// row. Encodes the payload into the approval's `commit_ref` so the -/// approve handler can re-parse it without a side table. Validates -/// inputs (non-empty targets, non-empty body, sane interval) at -/// submit time — the operator should never see a malformed schedule -/// pending approval. -fn handle_request_schedule_prompt( - coord: &Arc, - requester: &str, - payload: &hive_sh4re::SchedulePromptPayload, -) -> AgentResponse { - if payload.targets.is_empty() { - return AgentResponse::Err { - message: "schedule must have at least one target".into(), - }; - } - if payload.body.trim().is_empty() { - return AgentResponse::Err { - message: "schedule body must be non-empty".into(), - }; - } - if let Some(0) = payload.interval_seconds { - return AgentResponse::Err { - message: "interval_seconds must be > 0 (use None for one-shot)".into(), - }; - } - let commit_ref = match serde_json::to_string(payload) { - Ok(s) => s, - Err(e) => { - return AgentResponse::Err { - message: format!("encode SchedulePromptPayload: {e:#}"), - }; - } - }; - let id = match coord.approvals.submit_kind( - requester, - hive_sh4re::ApprovalKind::SchedulePrompt, - &commit_ref, - payload.description.as_deref(), - requester, - ) { - Ok(id) => id, - Err(e) => { - return AgentResponse::Err { - message: format!("queue schedule_prompt approval: {e:#}"), - }; - } - }; - tracing::info!( - %id, - requester, - targets = ?payload.targets, - first_fire_at = payload.first_fire_at_unix, - interval = ?payload.interval_seconds, - "schedule_prompt approval queued" - ); - coord.emit_approval_added(crate::coordinator::ApprovalAdded { - id, - agent: requester, - approval_kind: "schedule_prompt", - sha_short: None, - diff: None, - description: payload.description.clone(), - pr_number: None, - }); - AgentResponse::Ok -} - -/// Cancel a schedule (whole or per-target). Manager-surface -/// authorization: a manager can cancel its own schedules + any -/// schedule whose owner is one of its sub-agents (topology-walked). -/// The operator surface bypasses this and can cancel anything; -/// agents reaching this path through the manager get the -/// topology-scoped check. -fn handle_cancel_schedule( - coord: &Arc, - requester: &str, - schedule_id: i64, - targets: Option<&[String]>, -) -> AgentResponse { - let schedule = match coord.scheduled_prompts.get(schedule_id) { - Ok(Some(s)) => s, - Ok(None) => { - return AgentResponse::Err { - message: format!("schedule {schedule_id} not found"), - }; - } - Err(e) => { - return AgentResponse::Err { - message: format!("read schedule {schedule_id}: {e:#}"), - }; - } - }; - if !cancel_authorized(requester, &schedule.owner) { - return AgentResponse::Err { - message: format!( - "not authorized: {requester} cannot cancel schedule owned by {owner}", - owner = schedule.owner - ), - }; - } - let result = match targets { - Some(list) if !list.is_empty() => coord - .scheduled_prompts - .cancel_targets(schedule_id, list) - .map_err(|e| format!("cancel targets: {e:#}")), - _ => coord - .scheduled_prompts - .cancel_all(schedule_id) - .map_err(|e| format!("cancel all: {e:#}")), - }; - match result { - Ok(()) => { - coord.emit_schedules_snapshot(); - AgentResponse::Ok - } - Err(message) => AgentResponse::Err { message }, - } -} - -/// Authorize + dispatch a `FireScheduleNow` request from the -/// manager surface. Same ownership rules as `CancelSchedule`: -/// requester can fire its own schedules + any owned by an agent -/// in its subtree. The actual fan-out lives in -/// `scheduled_prompts_worker::fire_now`. -async fn handle_fire_schedule_now( - coord: &Arc, - requester: &str, - schedule_id: i64, -) -> AgentResponse { - let schedule = match coord.scheduled_prompts.get(schedule_id) { - Ok(Some(s)) => s, - Ok(None) => { - return AgentResponse::Err { - message: format!("schedule {schedule_id} not found"), - }; - } - Err(e) => { - return AgentResponse::Err { - message: format!("read schedule {schedule_id}: {e:#}"), - }; - } - }; - if !cancel_authorized(requester, &schedule.owner) { - return AgentResponse::Err { - message: format!( - "not authorized: {requester} cannot fire schedule owned by {owner}", - owner = schedule.owner - ), - }; - } - // MCP fire_schedule_now stays no-reset (cadence intact); the - // reset-timer option is a dashboard-dialog affordance. - match crate::scheduled_prompts_worker::fire_now(coord, schedule_id, false).await { - Ok(_report) => { - coord.emit_schedules_snapshot(); - AgentResponse::Ok - } - Err(e) => AgentResponse::Err { - message: format!("fire schedule {schedule_id} now: {e:#}"), - }, - } -} - -/// Field-named PATCH payload for [`handle_edit_schedule`]. Every -/// field is "leave alone" when `None`; the double-`Option` fields -/// additionally distinguish clear (`Some(None)`) from set -/// (`Some(Some(v))`). -#[allow( - clippy::option_option, - reason = "double-Option carries three-state PATCH semantics: outer None = \ - leave alone, Some(None) = clear, Some(Some(v)) = set" -)] -struct EditSchedulePatch { - body: Option, - description: Option>, - interval_seconds: Option>, - next_fire_at_unix: Option, - targets_add: Option>, - targets_remove: Option>, -} - -/// Authorize + dispatch a `EditSchedule` patch. Same ownership -/// rules as `CancelSchedule` — the manager can edit -/// schedules it owns + any owned by an agent in its subtree. -/// Forwards the partial payload to -/// `ScheduledPrompts::update` which enforces the cancelled-row / -/// zero-interval validation. Returns `Ok` on a clean update; -/// `Err` with the underlying message on any auth / validation -/// failure so the dashboard can surface it verbatim. -fn handle_edit_schedule( - coord: &Arc, - requester: &str, - schedule_id: i64, - patch: EditSchedulePatch, -) -> AgentResponse { - let EditSchedulePatch { - body, - description, - interval_seconds, - next_fire_at_unix, - targets_add, - targets_remove, - } = patch; - let schedule = match coord.scheduled_prompts.get(schedule_id) { - Ok(Some(s)) => s, - Ok(None) => { - return AgentResponse::Err { - message: format!("schedule {schedule_id} not found"), - }; - } - Err(e) => { - return AgentResponse::Err { - message: format!("read schedule {schedule_id}: {e:#}"), - }; - } - }; - if !cancel_authorized(requester, &schedule.owner) { - return AgentResponse::Err { - message: format!( - "not authorized: {requester} cannot edit schedule owned by {owner}", - owner = schedule.owner - ), - }; - } - let patch = crate::scheduled_prompts::UpdateSchedule { - body, - description, - interval_seconds, - next_fire_at_unix, - targets_add, - targets_remove, - }; - match coord.scheduled_prompts.update(schedule_id, patch) { - Ok(()) => { - coord.emit_schedules_snapshot(); - AgentResponse::Ok - } - Err(e) => AgentResponse::Err { - message: format!("edit schedule {schedule_id}: {e:#}"), - }, - } -} - -/// Permission check for `CancelSchedule` on the manager surface. -/// `requester` (always `ruth` here) can cancel its own schedules. -/// Sub-agent ownership is delegated to topology — see -/// `crate::topology::is_descendant_of`. Also reused by -/// `handle_fire_schedule_now` — fire-auth follows the same shape. -fn cancel_authorized(requester: &str, owner: &str) -> bool { - if requester == owner { - return true; - } - if requester == hive_sh4re::OPERATOR_RECIPIENT { - return true; - } - // Manager can cancel anything owned by an agent in its subtree. - // For the current single-manager topology that covers everything, - // but the check stays correct as the tree grows. - crate::topology::is_descendant_of(owner, requester) -} - -/// `request_apply_commit` takes a commit SHA only — not a branch or -/// tag name. A branch is mutable; pinning the proposal to a concrete -/// sha keeps "what the manager asked to deploy" unambiguous and means -/// the `proposal/` tag is a faithful record of the request. -/// Accepts a 7..=40 char hex string (short or full sha); the exact -/// commit is resolved + existence-checked against the proposed repo -/// later in `lifecycle::git_fetch_to_tag`. -pub(crate) fn validate_commit_ref(commit_ref: &str) -> Result<()> { - let n = commit_ref.len(); - let hex = commit_ref.chars().all(|c| c.is_ascii_hexdigit()); - if !(7..=40).contains(&n) || !hex { - anyhow::bail!( - "commit_ref '{commit_ref}' is not a commit sha — request_apply_commit \ - takes a 7-40 char hex sha, not a branch or tag name" - ); - } - Ok(()) -} - -/// Queue an `InitConfig` approval for a brand-new agent whose config repo -/// does not yet exist. Shared between the manager and agent sockets. -/// -/// `parent`, when `Some`, is the agent that will own the new child once -/// the operator approves: it is stashed in the approval's `commit_ref` -/// field (unused for `InitConfig` otherwise — same pattern -/// `UpdateMetaInputs` uses to carry its inputs JSON) and consumed in -/// `run_approval_init_config` to write the `child -> parent` topology -/// edge. Callers pass the requesting agent, so the requester becomes the -/// new agent's parent (the root requesting a new agent → a top-level agent, -/// matching `topology::reconcile`'s default). `None` writes no explicit -/// edge (reconcile-default placement) — retained for that fallback. -pub(crate) fn submit_init_config( - coord: &Arc, - name: &str, - parent: Option<&str>, - description: Option, -) -> anyhow::Result { - let proposed_dir = crate::coordinator::Coordinator::agent_proposed_dir(name); - if proposed_dir.join(".git").exists() { - anyhow::bail!( - "proposed config repo for '{name}' already exists at {} - \ - use request_apply_commit to update an existing agent's config", - proposed_dir.display() - ); - } - let id = coord - .approvals - .submit_kind( - name, - hive_sh4re::ApprovalKind::InitConfig, - parent.unwrap_or(""), - description.as_deref(), - // `parent` is the requesting agent (becomes the new child's - // parent); it's also the submitter the approval events route - // back to. No declared parent = operator-initiated path. - parent.unwrap_or("operator"), - ) - .map_err(|e| anyhow::anyhow!("queue approval row: {e:#}"))?; - tracing::info!(%id, %name, "init_config approval queued"); - coord.emit_approval_added(crate::coordinator::ApprovalAdded { - id, - agent: name, - approval_kind: "init_config", - sha_short: None, - diff: None, - description, - pr_number: None, - }); - Ok(id) -} - -/// Submit-time half of the apply flow: queue the approval row, then -/// fetch the manager's commit from the proposed repo into applied and -/// pin it as `refs/tags/proposal/`. From this point on the manager -/// repo is irrelevant for this approval — even if the manager amends -/// or force-pushes, the canonical sha hive-c0re will eventually -/// approve/deny lives in applied's object DB. -/// -/// If anything fails after the row is inserted (sha missing in -/// proposed, fs error, git plumbing crash) we mark the row failed and -/// surface the error to the manager. We don't try to roll the row -/// back — the failure is part of the audit trail. -pub(crate) async fn submit_apply_commit( - coord: &Arc, - agent: &str, - commit_ref: &str, - description: Option<&str>, - submitter: &str, -) -> anyhow::Result<(i64, String)> { - validate_commit_ref(commit_ref)?; - let proposed_dir = crate::coordinator::Coordinator::agent_proposed_dir(agent); - let applied_dir = crate::coordinator::Coordinator::agent_applied_dir(agent); - if !proposed_dir.exists() { - anyhow::bail!( - "proposed repo missing for agent '{agent}' (expected at {})", - proposed_dir.display() - ); - } - if !applied_dir.join(".git").exists() { - // First deploy: seed the applied repo from proposed so we can plant - // the proposal/ tag below. setup_applied seeds at the root - // (template) commit of proposed, not at main, so deployed/0 is the - // template baseline. This makes the diff mara sees on approval - // show the manager's actual changes rather than an empty diff. - crate::lifecycle::setup_applied(&applied_dir, Some(&proposed_dir), agent) - .await - .context("seed applied repo for first spawn")?; - } - let id = coord - .approvals - .submit_kind( - agent, - hive_sh4re::ApprovalKind::ApplyCommit, - commit_ref, - description, - submitter, - ) - .map_err(|e| anyhow::anyhow!("queue approval row: {e:#}"))?; - let tag = format!("proposal/{id}"); - let sha = - match crate::lifecycle::git_fetch_to_tag(&applied_dir, &proposed_dir, commit_ref, &tag) - .await - { - Ok(s) => s, - Err(e) => { - // Surface the failure on the approval row so the - // dashboard reflects it instead of leaving a phantom - // pending entry. The note doubles as the operator-visible - // explanation of why the approval can't be approved. - let note = format!("{e:#}"); - let _ = coord.approvals.mark_failed(id, ¬e); - coord.emit_approval_resolved(crate::coordinator::ApprovalResolved { - id, - agent, - approval_kind: "apply_commit", - sha_short: None, - status: "failed", - note: Some(note), - description: description.map(str::to_owned), - }); - return Err(anyhow::anyhow!("git_fetch_to_tag: {e:#}")); - } - }; - coord - .approvals - .set_fetched_sha(id, &sha) - .map_err(|e| anyhow::anyhow!("persist fetched_sha: {e:#}"))?; - // Pre-flight gates: both reject the apply before approval if - // the agent's flake state would inflate meta's lock with duplicates - // or lie about what nix will fetch. Both checks independently read - // `:flake.lock` via git — they don't share state. Order matters - // only for early-exit + messaging: sync first means a stale lock - // bails with the actionable "run `nix flake lock`" hint rather than - // a dedup pass on a lock nix would never produce. - // - // Runs after `set_fetched_sha` so the failed row carries the sha - // that broke. Both failure paths mark + emit, then bail. - let sha_short = sha[..sha.len().min(12)].to_owned(); - if let Err(e) = crate::flake_check::check_lock_in_sync(&applied_dir, &tag, id).await { - let note = format!("{e:#}"); - let _ = coord.approvals.mark_failed(id, ¬e); - coord.emit_approval_resolved(crate::coordinator::ApprovalResolved { - id, - agent, - approval_kind: "apply_commit", - sha_short: Some(sha_short.clone()), - status: "failed", - note: Some(note), - description: description.map(str::to_owned), - }); - return Err(anyhow::anyhow!("flake lock-sync check: {e:#}")); - } - if let Err(e) = crate::flake_check::check_no_duplicate_inputs(&applied_dir, &tag).await { - let note = format!("{e:#}"); - let _ = coord.approvals.mark_failed(id, ¬e); - coord.emit_approval_resolved(crate::coordinator::ApprovalResolved { - id, - agent, - approval_kind: "apply_commit", - sha_short: Some(sha_short.clone()), - status: "failed", - note: Some(note), - description: description.map(str::to_owned), - }); - return Err(anyhow::anyhow!("flake dedup check: {e:#}")); - } - // Mirror the freshly-planted proposal/ tag to the forge. - if let Err(e) = crate::forge::push_config(agent).await { - tracing::warn!(%agent, %id, error = ?e, "forge: push_config after submit failed"); - } - // Phase 5b: surface the new pending approval on the dashboard - // event channel. Compute the diff once here so live subscribers - // get a fully-formed row without a snapshot refetch. `sha_short` - // is reused from the dedup gate above. - let diff = crate::dashboard::approval_diff(agent, id).await; - coord.emit_approval_added(crate::coordinator::ApprovalAdded { - id, - agent, - approval_kind: "apply_commit", - sha_short: Some(sha_short), - diff: Some(diff), - description: description.map(str::to_owned), - pr_number: None, - }); - Ok((id, sha)) -} - -/// Map a `scheduled_prompts::Schedule` to its public wire shape. -/// Field-by-field copy — the two types are intentionally identical; -/// the separation keeps hive-sh4re free of hive-c0re-internal types. -/// Public alias `schedule_to_wire_public` re-exports for -/// `dashboard.rs::api_schedules` without crossing the module -/// boundary into the socket-server file. -pub fn schedule_to_wire_public(s: crate::scheduled_prompts::Schedule) -> hive_sh4re::WireSchedule { - schedule_to_wire(s) -} - -/// Drop schedule targets that point at agents which no longer exist, so -/// the dashboard's schedule table doesn't render ghost columns for -/// destroyed agents. `live` is the set of logical agent names from the -/// last `nixos-container list` scan (stopped agents included, destroyed -/// ones absent); the `operator` pseudo-target is always retained since -/// it isn't a container. Applied only to the dashboard wire paths -/// (`api_schedules` + the `SchedulesChanged` SSE emit) — the -/// manager-facing `list_schedules` stays unfiltered so agents can still -/// see and cancel stale targets. This is a view filter: the underlying -/// schedule rows keep every target, so a re-spawned agent's targets -/// reappear on their own. -pub(crate) fn filter_ghost_schedule_targets( - schedules: &mut [hive_sh4re::WireSchedule], - live: &std::collections::HashSet, -) { - for s in schedules.iter_mut() { - s.targets - .retain(|t| t.target == hive_sh4re::OPERATOR_RECIPIENT || live.contains(&t.target)); - } -} - -fn schedule_to_wire(s: crate::scheduled_prompts::Schedule) -> hive_sh4re::WireSchedule { - hive_sh4re::WireSchedule { - id: s.id, - owner: s.owner, - body: s.body, - interval_seconds: s.interval_seconds, - next_fire_at_unix: hive_sh4re::wire_time::from_secs(s.next_fire_at_unix), - created_at_unix: hive_sh4re::wire_time::from_secs(s.created_at_unix), - source: match s.source { - crate::scheduled_prompts::ScheduleSource::Operator => { - hive_sh4re::WireScheduleSource::Operator - } - crate::scheduled_prompts::ScheduleSource::Approval { id } => { - hive_sh4re::WireScheduleSource::Approval { id } - } - }, - cancelled_at_unix: s.cancelled_at_unix.map(hive_sh4re::wire_time::from_secs), - paused_at_unix: s.paused_at_unix.map(hive_sh4re::wire_time::from_secs), - description: s.description, - targets: s - .targets - .into_iter() - .map(|t| hive_sh4re::WireScheduleTarget { - target: t.target, - cancelled_at_unix: t.cancelled_at_unix.map(hive_sh4re::wire_time::from_secs), - last_fired_at_unix: t.last_fired_at_unix.map(hive_sh4re::wire_time::from_secs), - last_result: t.last_result, - }) - .collect(), - } -} - -/// 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, 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()); - } - }); -} - -#[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")); - } - - fn target(name: &str) -> hive_sh4re::WireScheduleTarget { - hive_sh4re::WireScheduleTarget { - target: name.to_owned(), - cancelled_at_unix: None, - last_fired_at_unix: None, - last_result: None, - } - } - - fn schedule(targets: &[&str]) -> hive_sh4re::WireSchedule { - hive_sh4re::WireSchedule { - id: 1, - owner: "operator".to_owned(), - body: "ping".to_owned(), - interval_seconds: None, - next_fire_at_unix: hive_sh4re::wire_time::from_secs(0), - created_at_unix: hive_sh4re::wire_time::from_secs(0), - source: hive_sh4re::WireScheduleSource::Operator, - cancelled_at_unix: None, - paused_at_unix: None, - description: None, - targets: targets.iter().map(|t| target(t)).collect(), - } - } - - #[test] - fn ghost_filter_drops_dead_agents_keeps_live_and_operator() { - let live: std::collections::HashSet = ["iris".to_owned(), "damocles".to_owned()] - .into_iter() - .collect(); - let mut schedules = vec![schedule(&["iris", "ghost", "operator", "damocles"])]; - filter_ghost_schedule_targets(&mut schedules, &live); - let kept: Vec<&str> = schedules[0] - .targets - .iter() - .map(|t| t.target.as_str()) - .collect(); - // `ghost` (destroyed) dropped; live agents + operator pseudo-target kept. - assert_eq!(kept, vec!["iris", "operator", "damocles"]); - } - - #[test] - fn ghost_filter_can_empty_targets_when_all_dead() { - let live: std::collections::HashSet = std::collections::HashSet::new(); - let mut schedules = vec![schedule(&["gone1", "gone2"])]; - filter_ghost_schedule_targets(&mut schedules, &live); - // operator is never in the live set but is always retained; here - // there's no operator target, so everything drops. - assert!(schedules[0].targets.is_empty()); - } - - #[test] - fn accepts_short_and_full_sha() { - assert!(validate_commit_ref("e194f78").is_ok()); - assert!(validate_commit_ref("e194f7812ab").is_ok()); - assert!(validate_commit_ref(&"a".repeat(40)).is_ok()); - // Uppercase hex resolves fine through `git rev-parse`. - assert!(validate_commit_ref("E194F78").is_ok()); - } - - #[test] - fn rejects_branch_and_tag_names() { - // The exact bug class this guard exists for. - assert!(validate_commit_ref("main").is_err()); - assert!(validate_commit_ref("HEAD").is_err()); - assert!(validate_commit_ref("deployed/0").is_err()); - assert!(validate_commit_ref("feature-branch").is_err()); - } - - #[test] - fn rejects_too_short_too_long_and_empty() { - assert!(validate_commit_ref("").is_err()); - assert!(validate_commit_ref("abc123").is_err()); // 6 chars - assert!(validate_commit_ref(&"a".repeat(41)).is_err()); - } - - #[test] - fn resolve_agent_state_target_self_and_default_are_free() { - // No topology/capability state needed for these: `None` and the - // caller's own name resolve to the caller (`is_descendant_of` short- - // circuits to true when candidate == ancestor); `"*"` is rejected - // (the hive-wide sweep is handled by the loose-ends caller instead). - assert_eq!(resolve_agent_state_target("iris", None), Ok("iris")); - assert_eq!(resolve_agent_state_target("iris", Some("iris")), Ok("iris")); - assert!(resolve_agent_state_target("iris", Some("*")).is_err()); - } -} diff --git a/hive-c0re/src/socket_server/config_approvals.rs b/hive-c0re/src/socket_server/config_approvals.rs new file mode 100644 index 00000000..cabfc994 --- /dev/null +++ b/hive-c0re/src/socket_server/config_approvals.rs @@ -0,0 +1,343 @@ +//! Config-approval request handlers: `RequestInitConfig` / +//! `RequestApplyCommit` / `RequestUpdateMetaInputs`, plus the shared +//! submit helpers (`submit_init_config` / `submit_apply_commit`) and the +//! commit-sha shape check (`validate_commit_ref`). + +use std::sync::Arc; + +use anyhow::{Context, Result}; +use hive_sh4re::AgentResponse; + +use super::require_new_child; +use crate::coordinator::Coordinator; + +/// `RequestInitConfig` — queue an `InitConfig` approval for an agent. The +/// `name` must be brand-new (absent from the topology) or already in the +/// caller's subtree; the requester is recorded as the new agent's parent (the +/// root requesting a new agent → a top-level agent, matching reconcile's +/// default). +pub(super) fn handle_request_init_config( + coord: &Arc, + agent: &str, + name: &str, + description: Option, +) -> AgentResponse { + if let Some(err) = require_new_child(agent, name, "request_init_config for") { + return err; + } + tracing::info!(%agent, %name, "request_init_config"); + match submit_init_config(coord, name, Some(agent), description) { + Ok(_id) => AgentResponse::Ok, + Err(e) => AgentResponse::Err { + message: format!("{e:#}"), + }, + } +} + +/// `RequestApplyCommit` — queue an apply-commit approval for an agent. The +/// target must be in the caller's subtree (the root covers every agent). +pub(super) async fn handle_request_apply_commit( + coord: &Arc, + agent: &str, + target_agent: &str, + commit_ref: &str, + description: Option<&str>, +) -> AgentResponse { + if let Some(err) = require_new_child(agent, target_agent, "request_apply_commit for") { + return err; + } + tracing::info!(%agent, %target_agent, %commit_ref, "request_apply_commit"); + match submit_apply_commit(coord, target_agent, commit_ref, description, agent).await { + Ok((id, sha)) => { + tracing::info!(%id, %target_agent, %sha, "apply_commit approval queued"); + AgentResponse::Ok + } + Err(e) => AgentResponse::Err { + message: format!("{e:#}"), + }, + } +} + +/// `RequestUpdateMetaInputs` — queue an `UpdateMetaInputs` approval +/// carrying the JSON-encoded input list in `commit_ref` (no git commit +/// is involved; the field is the payload the approval handler decodes). +pub(super) fn handle_request_update_meta_inputs( + coord: &Arc, + requester: &str, + inputs: &[String], + description: Option<&str>, +) -> AgentResponse { + let label = if inputs.is_empty() { + "all inputs".to_string() + } else { + inputs.join(", ") + }; + tracing::info!(%requester, %label, "request_update_meta_inputs"); + let commit_ref = serde_json::to_string(inputs).unwrap_or_default(); + let id = match coord + .approvals + .submit_kind( + requester, + hive_sh4re::ApprovalKind::UpdateMetaInputs, + &commit_ref, + description, + requester, + ) + .map_err(|e| anyhow::anyhow!("{e:#}")) + { + Ok(id) => id, + Err(e) => { + return AgentResponse::Err { + message: format!("queue update_meta_inputs approval: {e:#}"), + }; + } + }; + tracing::info!(%id, %label, "update_meta_inputs approval queued"); + coord.emit_approval_added(crate::coordinator::ApprovalAdded { + id, + agent: requester, + approval_kind: "update_meta_inputs", + sha_short: None, + diff: None, + description: description.map(str::to_owned), + pr_number: None, + }); + AgentResponse::Ok +} + +/// `request_apply_commit` takes a commit SHA only — not a branch or +/// tag name. A branch is mutable; pinning the proposal to a concrete +/// sha keeps "what the manager asked to deploy" unambiguous and means +/// the `proposal/` tag is a faithful record of the request. +/// Accepts a 7..=40 char hex string (short or full sha); the exact +/// commit is resolved + existence-checked against the proposed repo +/// later in `lifecycle::git_fetch_to_tag`. +pub(crate) fn validate_commit_ref(commit_ref: &str) -> Result<()> { + let n = commit_ref.len(); + let hex = commit_ref.chars().all(|c| c.is_ascii_hexdigit()); + if !(7..=40).contains(&n) || !hex { + anyhow::bail!( + "commit_ref '{commit_ref}' is not a commit sha — request_apply_commit \ + takes a 7-40 char hex sha, not a branch or tag name" + ); + } + Ok(()) +} + +/// Queue an `InitConfig` approval for a brand-new agent whose config repo +/// does not yet exist. Shared between the manager and agent sockets. +/// +/// `parent`, when `Some`, is the agent that will own the new child once +/// the operator approves: it is stashed in the approval's `commit_ref` +/// field (unused for `InitConfig` otherwise — same pattern +/// `UpdateMetaInputs` uses to carry its inputs JSON) and consumed in +/// `run_approval_init_config` to write the `child -> parent` topology +/// edge. Callers pass the requesting agent, so the requester becomes the +/// new agent's parent (the root requesting a new agent → a top-level agent, +/// matching `topology::reconcile`'s default). `None` writes no explicit +/// edge (reconcile-default placement) — retained for that fallback. +pub(crate) fn submit_init_config( + coord: &Arc, + name: &str, + parent: Option<&str>, + description: Option, +) -> anyhow::Result { + let proposed_dir = crate::coordinator::Coordinator::agent_proposed_dir(name); + if proposed_dir.join(".git").exists() { + anyhow::bail!( + "proposed config repo for '{name}' already exists at {} - \ + use request_apply_commit to update an existing agent's config", + proposed_dir.display() + ); + } + let id = coord + .approvals + .submit_kind( + name, + hive_sh4re::ApprovalKind::InitConfig, + parent.unwrap_or(""), + description.as_deref(), + // `parent` is the requesting agent (becomes the new child's + // parent); it's also the submitter the approval events route + // back to. No declared parent = operator-initiated path. + parent.unwrap_or("operator"), + ) + .map_err(|e| anyhow::anyhow!("queue approval row: {e:#}"))?; + tracing::info!(%id, %name, "init_config approval queued"); + coord.emit_approval_added(crate::coordinator::ApprovalAdded { + id, + agent: name, + approval_kind: "init_config", + sha_short: None, + diff: None, + description, + pr_number: None, + }); + Ok(id) +} + +/// Submit-time half of the apply flow: queue the approval row, then +/// fetch the manager's commit from the proposed repo into applied and +/// pin it as `refs/tags/proposal/`. From this point on the manager +/// repo is irrelevant for this approval — even if the manager amends +/// or force-pushes, the canonical sha hive-c0re will eventually +/// approve/deny lives in applied's object DB. +/// +/// If anything fails after the row is inserted (sha missing in +/// proposed, fs error, git plumbing crash) we mark the row failed and +/// surface the error to the manager. We don't try to roll the row +/// back — the failure is part of the audit trail. +pub(crate) async fn submit_apply_commit( + coord: &Arc, + agent: &str, + commit_ref: &str, + description: Option<&str>, + submitter: &str, +) -> anyhow::Result<(i64, String)> { + validate_commit_ref(commit_ref)?; + let proposed_dir = crate::coordinator::Coordinator::agent_proposed_dir(agent); + let applied_dir = crate::coordinator::Coordinator::agent_applied_dir(agent); + if !proposed_dir.exists() { + anyhow::bail!( + "proposed repo missing for agent '{agent}' (expected at {})", + proposed_dir.display() + ); + } + if !applied_dir.join(".git").exists() { + // First deploy: seed the applied repo from proposed so we can plant + // the proposal/ tag below. setup_applied seeds at the root + // (template) commit of proposed, not at main, so deployed/0 is the + // template baseline. This makes the diff mara sees on approval + // show the manager's actual changes rather than an empty diff. + crate::lifecycle::setup_applied(&applied_dir, Some(&proposed_dir), agent) + .await + .context("seed applied repo for first spawn")?; + } + let id = coord + .approvals + .submit_kind( + agent, + hive_sh4re::ApprovalKind::ApplyCommit, + commit_ref, + description, + submitter, + ) + .map_err(|e| anyhow::anyhow!("queue approval row: {e:#}"))?; + let tag = format!("proposal/{id}"); + let sha = + match crate::lifecycle::git_fetch_to_tag(&applied_dir, &proposed_dir, commit_ref, &tag) + .await + { + Ok(s) => s, + Err(e) => { + // Surface the failure on the approval row so the + // dashboard reflects it instead of leaving a phantom + // pending entry. The note doubles as the operator-visible + // explanation of why the approval can't be approved. + let note = format!("{e:#}"); + let _ = coord.approvals.mark_failed(id, ¬e); + coord.emit_approval_resolved(crate::coordinator::ApprovalResolved { + id, + agent, + approval_kind: "apply_commit", + sha_short: None, + status: "failed", + note: Some(note), + description: description.map(str::to_owned), + }); + return Err(anyhow::anyhow!("git_fetch_to_tag: {e:#}")); + } + }; + coord + .approvals + .set_fetched_sha(id, &sha) + .map_err(|e| anyhow::anyhow!("persist fetched_sha: {e:#}"))?; + // Pre-flight gates: both reject the apply before approval if + // the agent's flake state would inflate meta's lock with duplicates + // or lie about what nix will fetch. Both checks independently read + // `:flake.lock` via git — they don't share state. Order matters + // only for early-exit + messaging: sync first means a stale lock + // bails with the actionable "run `nix flake lock`" hint rather than + // a dedup pass on a lock nix would never produce. + // + // Runs after `set_fetched_sha` so the failed row carries the sha + // that broke. Both failure paths mark + emit, then bail. + let sha_short = sha[..sha.len().min(12)].to_owned(); + if let Err(e) = crate::flake_check::check_lock_in_sync(&applied_dir, &tag, id).await { + let note = format!("{e:#}"); + let _ = coord.approvals.mark_failed(id, ¬e); + coord.emit_approval_resolved(crate::coordinator::ApprovalResolved { + id, + agent, + approval_kind: "apply_commit", + sha_short: Some(sha_short.clone()), + status: "failed", + note: Some(note), + description: description.map(str::to_owned), + }); + return Err(anyhow::anyhow!("flake lock-sync check: {e:#}")); + } + if let Err(e) = crate::flake_check::check_no_duplicate_inputs(&applied_dir, &tag).await { + let note = format!("{e:#}"); + let _ = coord.approvals.mark_failed(id, ¬e); + coord.emit_approval_resolved(crate::coordinator::ApprovalResolved { + id, + agent, + approval_kind: "apply_commit", + sha_short: Some(sha_short.clone()), + status: "failed", + note: Some(note), + description: description.map(str::to_owned), + }); + return Err(anyhow::anyhow!("flake dedup check: {e:#}")); + } + // Mirror the freshly-planted proposal/ tag to the forge. + if let Err(e) = crate::forge::push_config(agent).await { + tracing::warn!(%agent, %id, error = ?e, "forge: push_config after submit failed"); + } + // Phase 5b: surface the new pending approval on the dashboard + // event channel. Compute the diff once here so live subscribers + // get a fully-formed row without a snapshot refetch. `sha_short` + // is reused from the dedup gate above. + let diff = crate::dashboard::approval_diff(agent, id).await; + coord.emit_approval_added(crate::coordinator::ApprovalAdded { + id, + agent, + approval_kind: "apply_commit", + sha_short: Some(sha_short), + diff: Some(diff), + description: description.map(str::to_owned), + pr_number: None, + }); + Ok((id, sha)) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn accepts_short_and_full_sha() { + assert!(validate_commit_ref("e194f78").is_ok()); + assert!(validate_commit_ref("e194f7812ab").is_ok()); + assert!(validate_commit_ref(&"a".repeat(40)).is_ok()); + // Uppercase hex resolves fine through `git rev-parse`. + assert!(validate_commit_ref("E194F78").is_ok()); + } + + #[test] + fn rejects_branch_and_tag_names() { + // The exact bug class this guard exists for. + assert!(validate_commit_ref("main").is_err()); + assert!(validate_commit_ref("HEAD").is_err()); + assert!(validate_commit_ref("deployed/0").is_err()); + assert!(validate_commit_ref("feature-branch").is_err()); + } + + #[test] + fn rejects_too_short_too_long_and_empty() { + assert!(validate_commit_ref("").is_err()); + assert!(validate_commit_ref("abc123").is_err()); // 6 chars + assert!(validate_commit_ref(&"a".repeat(41)).is_err()); + } +} diff --git a/hive-c0re/src/socket_server/lifecycle_handlers.rs b/hive-c0re/src/socket_server/lifecycle_handlers.rs new file mode 100644 index 00000000..f6640c72 --- /dev/null +++ b/hive-c0re/src/socket_server/lifecycle_handlers.rs @@ -0,0 +1,201 @@ +//! Container-lifecycle request handlers (`Start` / `Restart` / `Kill` / +//! `Update` / `ListDescendants`), including the capability-gated +//! infra-container restart path. All are topology-guarded via +//! `super::require_descendant`. + +use std::sync::Arc; + +use hive_sh4re::AgentResponse; + +use super::require_descendant; +use crate::coordinator::Coordinator; + +/// `Start` — start a container, kicking its next turn. The caller must be an +/// ancestor of `name` in the topology (the root covers every agent). +pub(super) fn handle_start(coord: &Arc, agent: &str, name: &str) -> AgentResponse { + if let Some(err) = require_descendant(agent, name, "start") { + return err; + } + tracing::info!(%agent, %name, "start container"); + // Persist `wanted = Up` and submit the Start DAG; the submit layer + // upgrades a stale-rev start to a full rebuild so the container + // runs current nix derivations before it starts. + crate::job_queue::submit::start( + coord, + name, + crate::job_queue::Source::Manual, + format!("agent `{agent}` start tool"), + ); + AgentResponse::Ok +} + +/// `Restart` — enqueue a restart for a container. The caller must be an +/// ancestor of `name` in the topology. The infra-container branch is +/// orthogonal: it is gated on the `infra_admin` capability and audited, so it +/// stays ahead of the topology guard. +pub(super) async fn handle_restart( + coord: &Arc, + agent: &str, + name: &str, +) -> AgentResponse { + // Infra-container restart: an agent holding the `infra_admin` + // capability can restart a hive infrastructure container (hive-ci / + // hive-gateway / hive-forge / hive-matrix) by passing its name to the + // same restart tool. The `InfraContainer` enum parse both recognises + // these (never agent children, so disjoint from the child path below) + // and yields the typed value the restart path needs. + if let Ok(container) = name.parse::() { + return handle_restart_infra(coord, agent, container).await; + } + if let Some(err) = require_descendant(agent, name, "restart") { + return err; + } + tracing::info!(%agent, %name, "submit restart"); + crate::job_queue::submit::restart( + coord, + name, + crate::job_queue::Source::Manual, + format!("agent `{agent}` restart tool"), + ); + AgentResponse::Ok +} + +/// Restart a hive infrastructure container on behalf of an agent that +/// holds the `infra_admin` capability. The `container` is already a valid +/// [`InfraContainer`] (the caller parsed it); this gates on the capability +/// and routes the systemctl restart through hive-priv. Direct, not +/// approval-gated. +async fn handle_restart_infra( + coord: &Arc, + agent: &str, + container: hive_sh4re::priv_proto::InfraContainer, +) -> AgentResponse { + let name = container.unit_name(); + // Record the attempt in the operator-visible privileged-action audit + // trail, then emit a live `AuditEntryAdded` so the dashboard audit view + // appends it off `/dashboard/stream`. Best-effort: `record` returns the + // canonical row (or `None` on a sqlite blip), and we stream exactly that + // row so the stored + streamed views can't drift. `action` is stable so + // the dashboard can group/filter. + let audit = |outcome: crate::audit_log::AuditOutcome, detail: Option<&str>| { + if let Some(entry) = coord + .audit_log + .record(agent, "restart_infra", name, outcome, detail) + { + coord.emit_audit_entry(entry); + } + }; + if !crate::capabilities::has_cap(agent, hive_sh4re::Capability::InfraAdmin) { + tracing::warn!(%agent, %name, "agent: infra restart denied (no infra_admin capability)"); + audit( + crate::audit_log::AuditOutcome::Err, + Some("denied: missing infra_admin capability"), + ); + return AgentResponse::Err { + message: format!( + "restarting infra container `{name}` requires the `infra_admin` capability" + ), + }; + } + tracing::info!(%agent, %name, "agent: restart infra container"); + match crate::priv_client::restart_infra_container(container).await { + Ok(()) => { + audit(crate::audit_log::AuditOutcome::Ok, None); + AgentResponse::Ok + } + Err(e) => { + let msg = format!("{e:#}"); + audit(crate::audit_log::AuditOutcome::Err, Some(&msg)); + AgentResponse::Err { message: msg } + } + } +} + +/// `Kill` — kill a container, unregister it, notify the manager. The caller +/// must be an ancestor of `name` in the topology. +pub(super) async fn handle_kill( + coord: &Arc, + agent: &str, + name: &str, +) -> AgentResponse { + if let Some(err) = require_descendant(agent, name, "kill") { + return err; + } + tracing::info!(%agent, %name, "kill container"); + // Persist the intent even if the kill fails — otherwise the next + // reconcile would restart the container. + if let Err(e) = coord.power.set(name, crate::power::Wanted::Offline) { + tracing::warn!(%name, error = ?e, "agent_power: set wanted=offline failed"); + } + 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.to_owned(), + }); + AgentResponse::Ok + } + Err(e) => AgentResponse::Err { + message: format!("{e:#}"), + }, + } +} + +/// `Update` — enqueue a rebuild for a container. The caller must be an +/// ancestor of `name` in the topology. +pub(super) fn handle_update(coord: &Arc, agent: &str, name: &str) -> AgentResponse { + if let Some(err) = require_descendant(agent, name, "rebuild") { + return err; + } + tracing::info!(%agent, %name, "submit rebuild"); + crate::job_queue::submit::rebuild( + coord, + name, + crate::job_queue::Source::Manual, + format!("agent `{agent}` update tool"), + ); + AgentResponse::Ok +} + +/// `ListDescendants` — every topological descendant of `agent` with +/// its running/stopped state, parents before children. +pub(super) async fn handle_list_descendants(agent: &str) -> AgentResponse { + tracing::debug!(%agent, "agent: list descendants"); + // All containers known to nixos-container (running only). + let running_set: std::collections::HashSet = 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 = 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 } +} diff --git a/hive-c0re/src/socket_server/mod.rs b/hive-c0re/src/socket_server/mod.rs new file mode 100644 index 00000000..641b3729 --- /dev/null +++ b/hive-c0re/src/socket_server/mod.rs @@ -0,0 +1,1092 @@ +//! 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_sh4re::{AgentRequest, AgentResponse, 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 schedules::filter_ghost_schedule_targets; +pub use schedules::schedule_to_wire_public; + +use config_approvals::{ + handle_request_apply_commit, 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) -> Result { + 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) -> Result<()> { + use std::os::unix::fs::PermissionsExt as _; + let dir = Coordinator::agent_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) -> 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::(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` — 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) -> 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_sh4re::Request, + agent: &str, + coord: &Arc, +) -> Option { + 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 } => { + handle_recv(coord, agent, *wait_seconds, *max).await + } + hive_sh4re::Request::Status => handle_status(coord, agent), + hive_sh4re::Request::OperatorMsg { body } => handle_operator_msg(coord, agent, body), + hive_sh4re::Request::Wake { from, body } => handle_wake(coord, agent, from, body), + hive_sh4re::Request::Recent { limit } => handle_recent(coord, agent, *limit), + 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 } => handle_set_status(coord, text), + hive_sh4re::Request::GetAgentMeta { name } => { + handle_get_agent_meta(coord, agent, name.as_deref()).await + } + 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::CreateRepo { repo } => handle_create_repo(agent, repo).await, + hive_sh4re::Request::AckTurn => handle_ack_turn(coord, agent), + hive_sh4re::Request::AckUntil { up_to } => handle_ack_until(coord, agent, *up_to), + hive_sh4re::Request::RequeueInflight => handle_requeue_inflight(coord, agent), + hive_sh4re::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_sh4re::Response::Ok + } + hive_sh4re::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, + agent: &str, + wait_seconds: Option, + max: Option, +) -> hive_sh4re::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_sh4re::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) => 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:#}"), + }, + } +} + +/// `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, + agent: &str, + from: &str, + body: &str, +) -> hive_sh4re::Response { + match coord.broker.send(&Message { + from: from.to_owned(), + to: agent.to_owned(), + body: body.to_owned(), + in_reply_to: None, + }) { + Ok(()) => hive_sh4re::Response::Ok, + Err(e) => hive_sh4re::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, text: &str) -> hive_sh4re::Response { + if let Err(message) = crate::limits::check_status_text(text) { + return hive_sh4re::Response::Err { message }; + } + let coord2 = Arc::clone(coord); + tokio::spawn(async move { coord2.rescan_containers_and_emit().await }); + hive_sh4re::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_sh4re::Response { + if !valid_repo_name(repo) { + return hive_sh4re::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_sh4re::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_sh4re::Response::RepoCreated { + clone_url: format!("{}/{full_name}.git", crate::forge::FORGE_HTTP), + full_name, + }, + Err(e) => hive_sh4re::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, + agent: &str, + name: Option<&str>, +) -> hive_sh4re::Response { + let target = name.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, + matrix_accounts: read_agent_matrix_identities(target), + } +} + +/// 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: &str) -> Vec { + 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::>(&s).ok()) + .unwrap_or_default() +} + +/// `Status` — count of pending (unread) inbox messages for `agent`. +fn handle_status(coord: &Arc, agent: &str) -> hive_sh4re::Response { + match coord.broker.count_pending(agent) { + Ok(unread) => hive_sh4re::Response::Status { unread }, + Err(e) => hive_sh4re::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, agent: &str, body: &str) -> hive_sh4re::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_sh4re::Response::Ok, + Err(e) => hive_sh4re::Response::Err { + message: format!("{e:#}"), + }, + } +} + +/// `Recent` — the last `limit` inbox rows for `agent` (read-only, +/// doesn't consume). +fn handle_recent(coord: &Arc, agent: &str, limit: u64) -> hive_sh4re::Response { + match coord.broker.recent_for(agent, limit) { + Ok(rows) => hive_sh4re::Response::Recent { rows }, + Err(e) => hive_sh4re::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, agent: &str) -> hive_sh4re::Response { + match coord.broker.ack_turn(agent) { + Ok(_n) => hive_sh4re::Response::Ok, + Err(e) => hive_sh4re::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, agent: &str, up_to: i64) -> hive_sh4re::Response { + match coord.broker.ack_until(agent, up_to) { + Ok(count) => hive_sh4re::Response::Acked { count }, + Err(e) => hive_sh4re::Response::Err { + message: format!("{e:#}"), + }, + } +} + +/// `RequeueInflight` — resurface `agent`'s unacked in-flight messages +/// (crash recovery on harness boot). +fn handle_requeue_inflight(coord: &Arc, agent: &str) -> hive_sh4re::Response { + match coord.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:#}"), + }, + } +} + +/// 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: &AgentRequest, agent: &str, coord: &Arc) -> AgentResponse { + 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). + AgentRequest::Start { name } => handle_start(coord, agent, name), + AgentRequest::Restart { name } => handle_restart(coord, agent, name).await, + AgentRequest::Kill { name } => handle_kill(coord, agent, name).await, + AgentRequest::Update { name } => handle_update(coord, agent, name), + AgentRequest::ListDescendants => handle_list_descendants(agent).await, + AgentRequest::RequestInitConfig { name, description } => { + handle_request_init_config(coord, agent, name, description.clone()) + } + AgentRequest::RequestApplyCommit { + agent: target_agent, + commit_ref, + description, + } => { + handle_request_apply_commit( + coord, + agent, + target_agent, + commit_ref, + description.as_deref(), + ) + .await + } + // Agent-state queries: own subtree is free; other agents + the + // hive-wide `"*"` sweep require `QueryAgentState`. + AgentRequest::GetLooseEnds { agent: target } => { + handle_get_loose_ends(coord, agent, target.as_deref()) + } + AgentRequest::CountPendingReminders { agent: target } => { + handle_count_pending_reminders(coord, agent, target.as_deref()) + } + AgentRequest::ReminderRollup { + since_secs, + agent: target, + } => handle_reminder_rollup(coord, agent, target.as_deref(), *since_secs), + // 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: &AgentRequest, + agent: &str, + coord: &Arc, +) -> AgentResponse { + match req { + AgentRequest::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()) + } + AgentRequest::RequestSchedulePrompt(payload) => { + if let Some(err) = require_group(agent, "scheduling", "schedule a prompt") { + return err; + } + handle_request_schedule_prompt(coord, agent, payload) + } + AgentRequest::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()) + } + AgentRequest::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(), + }, + ) + } + AgentRequest::ListSchedules => { + if let Some(err) = require_group(agent, "scheduling", "list schedules") { + return err; + } + handle_list_schedules(coord) + } + AgentRequest::FireScheduleNow { id } => { + if let Some(err) = require_group(agent, "scheduling", "fire a schedule") { + return err; + } + handle_fire_schedule_now(coord, agent, *id).await + } + AgentRequest::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. + _ => AgentResponse::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 { + if crate::topology::is_descendant_of(target, agent) { + None + } else { + Some(AgentResponse::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 { + if crate::tool_groups::groups_for(agent) + .iter() + .any(|g| g == group) + { + None + } else { + Some(AgentResponse::Err { + message: format!("agent `{agent}` cannot {action}: requires the `{group}` tool group"), + }) + } +} + +/// Topology guard for `request_init_config` / `request_apply_commit`, +/// which may legitimately target a child that does not exist *yet* +/// (spawning a brand-new sub-agent). 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 { + if let Some(reason) = crate::dashboard::validate_agent_name(target) { + return Some(AgentResponse::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(AgentResponse::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, + agent: &str, + target: Option<&str>, +) -> AgentResponse { + let result = if target == Some("*") { + if !crate::capabilities::has_cap(agent, hive_sh4re::Capability::QueryAgentState) { + return AgentResponse::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 AgentResponse::Err { message }, + } + }; + match result { + Ok(loose_ends) => AgentResponse::LooseEnds { loose_ends }, + Err(e) => AgentResponse::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, + agent: &str, + target: Option<&str>, +) -> AgentResponse { + match resolve_agent_state_target(agent, target) { + 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 }, + } +} + +/// `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, + agent: &str, + target: Option<&str>, + since_secs: u64, +) -> AgentResponse { + match resolve_agent_state_target(agent, target) { + 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 }, + } +} + +/// 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, + pub container: &'a Option, + pub lines: &'a Option, + pub priority: &'a Option, + pub grep: &'a Option, + pub since: &'a Option, + pub until: &'a Option, +} + +/// 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<'_>) -> AgentResponse { + let HostJournalArgs { + unit, + container, + lines, + priority, + grep, + since, + until, + } = args; + 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, + hive_sh4re::priv_proto::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 }; + AgentResponse::HostJournal { content } + } + Err(e) => AgentResponse::Err { + message: format!("journal read: {e:#}"), + }, + }; + } + + let mut args: Vec = 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 `": "` strings for any delivery +/// failures (empty = all good). +pub(crate) fn fan_out_send( + coord: &Arc, + from: &str, + body: &str, + in_reply_to: Option, + targets: &[String], +) -> Vec { + 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 == ""`) / 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, + agent: &str, + to: &str, + body: &str, + in_reply_to: Option, +) -> 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(", ")), + } + }; + } + // ``: 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 (``) 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 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:#}"), + }, + } +} + +/// `GetLogs` — read a child container's journal via hive-priv (the +/// `-M` read needs root). `journalctl -M` wants the `h-` machine +/// name, which `container_name` derives. +async fn handle_get_logs(agent: &str, lines: Option) -> AgentResponse { + 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_sh4re::priv_proto::JournalQuery { + lines: n, + ..Default::default() + }, + ) + .await + { + Ok((stdout, stderr)) => { + let content = if stdout.is_empty() { stderr } else { stdout }; + AgentResponse::Logs { content } + } + Err(e) => AgentResponse::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, 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()); + } + }); +} diff --git a/hive-c0re/src/socket_server/reminders.rs b/hive-c0re/src/socket_server/reminders.rs new file mode 100644 index 00000000..674ebc1a --- /dev/null +++ b/hive-c0re/src/socket_server/reminders.rs @@ -0,0 +1,229 @@ +//! Reminder request handling: the `Remind` handler, the shared +//! `store_remind` storage path with its pending-cap and large-body +//! auto-save dance, timing resolution, and the agent-state target +//! resolution shared by the loose-ends / reminder query handlers. + +use std::sync::Arc; + +use hive_sh4re::AgentResponse; + +use crate::coordinator::Coordinator; + +pub(super) fn handle_remind( + coord: &Arc, + 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::().ok()) + .unwrap_or(DEFAULT_REMIND_MAX_PENDING) +} + +pub(crate) fn store_remind( + coord: &Arc, + 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//state/reminders/auto-.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> { + 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 a *named* `GetLooseEnds` / +/// `CountPendingReminders` / `ReminderRollup` query. Rules: +/// +/// - `None` (or `Some(caller)`) → the caller's own threads (always allowed). +/// - `Some(descendant)` in the caller's subtree (the root's subtree is the whole hive) → allowed, no extra capability. +/// - `Some(other)` outside the subtree → requires the `query_agent_state` capability; error otherwise. +/// - `Some("*")` → rejected here; the hive-wide sweep is handled by `handle_get_loose_ends` under the same `query_agent_state` gate. +pub(super) 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 only valid for loose-ends; \ + not available for this query" + .to_owned(), + ), + Some(name) => { + // Own subtree (the root covers all) is visible without extra + // capability; `is_descendant_of` returns true for `name == caller`. + if crate::topology::is_descendant_of(name, caller) { + return Ok(name); + } + if crate::capabilities::has_cap(caller, hive_sh4re::Capability::QueryAgentState) { + Ok(name) + } else { + Err(format!( + "agent `{caller}` cannot query `{name}`: not in its subtree 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 { + 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")); + } + + #[test] + fn resolve_agent_state_target_self_and_default_are_free() { + // No topology/capability state needed for these: `None` and the + // caller's own name resolve to the caller (`is_descendant_of` short- + // circuits to true when candidate == ancestor); `"*"` is rejected + // (the hive-wide sweep is handled by the loose-ends caller instead). + assert_eq!(resolve_agent_state_target("iris", None), Ok("iris")); + assert_eq!(resolve_agent_state_target("iris", Some("iris")), Ok("iris")); + assert!(resolve_agent_state_target("iris", Some("*")).is_err()); + } +} diff --git a/hive-c0re/src/socket_server/schedules.rs b/hive-c0re/src/socket_server/schedules.rs new file mode 100644 index 00000000..a8aa0783 --- /dev/null +++ b/hive-c0re/src/socket_server/schedules.rs @@ -0,0 +1,404 @@ +//! Scheduled-prompt request handlers (`ListSchedules` / +//! `RequestSchedulePrompt` / `CancelSchedule` / `EditSchedule` / +//! `FireScheduleNow`), their shared ownership check, and the +//! schedule-to-wire mapping reused by the dashboard +//! (`schedule_to_wire_public` / `filter_ghost_schedule_targets`). + +use std::sync::Arc; + +use hive_sh4re::AgentResponse; + +use crate::coordinator::Coordinator; + +/// `ListSchedules` — snapshot every scheduled prompt onto the wire. +pub(super) fn handle_list_schedules(coord: &Arc) -> AgentResponse { + match coord.scheduled_prompts.list() { + Ok(schedules) => AgentResponse::Schedules { + schedules: schedules.into_iter().map(schedule_to_wire).collect(), + }, + Err(e) => AgentResponse::Err { + message: format!("list scheduled prompts: {e:#}"), + }, + } +} + +/// Submit a `RequestSchedulePrompt` payload as an `ApprovalKind::SchedulePrompt` +/// row. Encodes the payload into the approval's `commit_ref` so the +/// approve handler can re-parse it without a side table. Validates +/// inputs (non-empty targets, non-empty body, sane interval) at +/// submit time — the operator should never see a malformed schedule +/// pending approval. +pub(super) fn handle_request_schedule_prompt( + coord: &Arc, + requester: &str, + payload: &hive_sh4re::SchedulePromptPayload, +) -> AgentResponse { + if payload.targets.is_empty() { + return AgentResponse::Err { + message: "schedule must have at least one target".into(), + }; + } + if payload.body.trim().is_empty() { + return AgentResponse::Err { + message: "schedule body must be non-empty".into(), + }; + } + if let Some(0) = payload.interval_seconds { + return AgentResponse::Err { + message: "interval_seconds must be > 0 (use None for one-shot)".into(), + }; + } + let commit_ref = match serde_json::to_string(payload) { + Ok(s) => s, + Err(e) => { + return AgentResponse::Err { + message: format!("encode SchedulePromptPayload: {e:#}"), + }; + } + }; + let id = match coord.approvals.submit_kind( + requester, + hive_sh4re::ApprovalKind::SchedulePrompt, + &commit_ref, + payload.description.as_deref(), + requester, + ) { + Ok(id) => id, + Err(e) => { + return AgentResponse::Err { + message: format!("queue schedule_prompt approval: {e:#}"), + }; + } + }; + tracing::info!( + %id, + requester, + targets = ?payload.targets, + first_fire_at = payload.first_fire_at_unix, + interval = ?payload.interval_seconds, + "schedule_prompt approval queued" + ); + coord.emit_approval_added(crate::coordinator::ApprovalAdded { + id, + agent: requester, + approval_kind: "schedule_prompt", + sha_short: None, + diff: None, + description: payload.description.clone(), + pr_number: None, + }); + AgentResponse::Ok +} + +/// Cancel a schedule (whole or per-target). Manager-surface +/// authorization: a manager can cancel its own schedules + any +/// schedule whose owner is one of its sub-agents (topology-walked). +/// The operator surface bypasses this and can cancel anything; +/// agents reaching this path through the manager get the +/// topology-scoped check. +pub(super) fn handle_cancel_schedule( + coord: &Arc, + requester: &str, + schedule_id: i64, + targets: Option<&[String]>, +) -> AgentResponse { + let schedule = match coord.scheduled_prompts.get(schedule_id) { + Ok(Some(s)) => s, + Ok(None) => { + return AgentResponse::Err { + message: format!("schedule {schedule_id} not found"), + }; + } + Err(e) => { + return AgentResponse::Err { + message: format!("read schedule {schedule_id}: {e:#}"), + }; + } + }; + if !cancel_authorized(requester, &schedule.owner) { + return AgentResponse::Err { + message: format!( + "not authorized: {requester} cannot cancel schedule owned by {owner}", + owner = schedule.owner + ), + }; + } + let result = match targets { + Some(list) if !list.is_empty() => coord + .scheduled_prompts + .cancel_targets(schedule_id, list) + .map_err(|e| format!("cancel targets: {e:#}")), + _ => coord + .scheduled_prompts + .cancel_all(schedule_id) + .map_err(|e| format!("cancel all: {e:#}")), + }; + match result { + Ok(()) => { + coord.emit_schedules_snapshot(); + AgentResponse::Ok + } + Err(message) => AgentResponse::Err { message }, + } +} + +/// Authorize + dispatch a `FireScheduleNow` request from the +/// manager surface. Same ownership rules as `CancelSchedule`: +/// requester can fire its own schedules + any owned by an agent +/// in its subtree. The actual fan-out lives in +/// `scheduled_prompts_worker::fire_now`. +pub(super) async fn handle_fire_schedule_now( + coord: &Arc, + requester: &str, + schedule_id: i64, +) -> AgentResponse { + let schedule = match coord.scheduled_prompts.get(schedule_id) { + Ok(Some(s)) => s, + Ok(None) => { + return AgentResponse::Err { + message: format!("schedule {schedule_id} not found"), + }; + } + Err(e) => { + return AgentResponse::Err { + message: format!("read schedule {schedule_id}: {e:#}"), + }; + } + }; + if !cancel_authorized(requester, &schedule.owner) { + return AgentResponse::Err { + message: format!( + "not authorized: {requester} cannot fire schedule owned by {owner}", + owner = schedule.owner + ), + }; + } + // MCP fire_schedule_now stays no-reset (cadence intact); the + // reset-timer option is a dashboard-dialog affordance. + match crate::scheduled_prompts_worker::fire_now(coord, schedule_id, false).await { + Ok(_report) => { + coord.emit_schedules_snapshot(); + AgentResponse::Ok + } + Err(e) => AgentResponse::Err { + message: format!("fire schedule {schedule_id} now: {e:#}"), + }, + } +} + +/// Field-named PATCH payload for [`handle_edit_schedule`]. Every +/// field is "leave alone" when `None`; the double-`Option` fields +/// additionally distinguish clear (`Some(None)`) from set +/// (`Some(Some(v))`). +#[allow( + clippy::option_option, + reason = "double-Option carries three-state PATCH semantics: outer None = \ + leave alone, Some(None) = clear, Some(Some(v)) = set" +)] +pub(super) struct EditSchedulePatch { + pub(super) body: Option, + pub(super) description: Option>, + pub(super) interval_seconds: Option>, + pub(super) next_fire_at_unix: Option, + pub(super) targets_add: Option>, + pub(super) targets_remove: Option>, +} + +/// Authorize + dispatch a `EditSchedule` patch. Same ownership +/// rules as `CancelSchedule` — the manager can edit +/// schedules it owns + any owned by an agent in its subtree. +/// Forwards the partial payload to +/// `ScheduledPrompts::update` which enforces the cancelled-row / +/// zero-interval validation. Returns `Ok` on a clean update; +/// `Err` with the underlying message on any auth / validation +/// failure so the dashboard can surface it verbatim. +pub(super) fn handle_edit_schedule( + coord: &Arc, + requester: &str, + schedule_id: i64, + patch: EditSchedulePatch, +) -> AgentResponse { + let EditSchedulePatch { + body, + description, + interval_seconds, + next_fire_at_unix, + targets_add, + targets_remove, + } = patch; + let schedule = match coord.scheduled_prompts.get(schedule_id) { + Ok(Some(s)) => s, + Ok(None) => { + return AgentResponse::Err { + message: format!("schedule {schedule_id} not found"), + }; + } + Err(e) => { + return AgentResponse::Err { + message: format!("read schedule {schedule_id}: {e:#}"), + }; + } + }; + if !cancel_authorized(requester, &schedule.owner) { + return AgentResponse::Err { + message: format!( + "not authorized: {requester} cannot edit schedule owned by {owner}", + owner = schedule.owner + ), + }; + } + let patch = crate::scheduled_prompts::UpdateSchedule { + body, + description, + interval_seconds, + next_fire_at_unix, + targets_add, + targets_remove, + }; + match coord.scheduled_prompts.update(schedule_id, patch) { + Ok(()) => { + coord.emit_schedules_snapshot(); + AgentResponse::Ok + } + Err(e) => AgentResponse::Err { + message: format!("edit schedule {schedule_id}: {e:#}"), + }, + } +} + +/// Permission check for `CancelSchedule` on the manager surface. +/// `requester` (always `ruth` here) can cancel its own schedules. +/// Sub-agent ownership is delegated to topology — see +/// `crate::topology::is_descendant_of`. Also reused by +/// `handle_fire_schedule_now` — fire-auth follows the same shape. +fn cancel_authorized(requester: &str, owner: &str) -> bool { + if requester == owner { + return true; + } + if requester == hive_sh4re::OPERATOR_RECIPIENT { + return true; + } + // Manager can cancel anything owned by an agent in its subtree. + // For the current single-manager topology that covers everything, + // but the check stays correct as the tree grows. + crate::topology::is_descendant_of(owner, requester) +} + +/// Map a `scheduled_prompts::Schedule` to its public wire shape. +/// Field-by-field copy — the two types are intentionally identical; +/// the separation keeps hive-sh4re free of hive-c0re-internal types. +/// Public alias `schedule_to_wire_public` re-exports for +/// `dashboard.rs::api_schedules` without crossing the module +/// boundary into the socket-server file. +pub fn schedule_to_wire_public(s: crate::scheduled_prompts::Schedule) -> hive_sh4re::WireSchedule { + schedule_to_wire(s) +} + +/// Drop schedule targets that point at agents which no longer exist, so +/// the dashboard's schedule table doesn't render ghost columns for +/// destroyed agents. `live` is the set of logical agent names from the +/// last `nixos-container list` scan (stopped agents included, destroyed +/// ones absent); the `operator` pseudo-target is always retained since +/// it isn't a container. Applied only to the dashboard wire paths +/// (`api_schedules` + the `SchedulesChanged` SSE emit) — the +/// manager-facing `list_schedules` stays unfiltered so agents can still +/// see and cancel stale targets. This is a view filter: the underlying +/// schedule rows keep every target, so a re-spawned agent's targets +/// reappear on their own. +pub(crate) fn filter_ghost_schedule_targets( + schedules: &mut [hive_sh4re::WireSchedule], + live: &std::collections::HashSet, +) { + for s in schedules.iter_mut() { + s.targets + .retain(|t| t.target == hive_sh4re::OPERATOR_RECIPIENT || live.contains(&t.target)); + } +} + +fn schedule_to_wire(s: crate::scheduled_prompts::Schedule) -> hive_sh4re::WireSchedule { + hive_sh4re::WireSchedule { + id: s.id, + owner: s.owner, + body: s.body, + interval_seconds: s.interval_seconds, + next_fire_at_unix: hive_sh4re::wire_time::from_secs(s.next_fire_at_unix), + created_at_unix: hive_sh4re::wire_time::from_secs(s.created_at_unix), + source: match s.source { + crate::scheduled_prompts::ScheduleSource::Operator => { + hive_sh4re::WireScheduleSource::Operator + } + crate::scheduled_prompts::ScheduleSource::Approval { id } => { + hive_sh4re::WireScheduleSource::Approval { id } + } + }, + cancelled_at_unix: s.cancelled_at_unix.map(hive_sh4re::wire_time::from_secs), + paused_at_unix: s.paused_at_unix.map(hive_sh4re::wire_time::from_secs), + description: s.description, + targets: s + .targets + .into_iter() + .map(|t| hive_sh4re::WireScheduleTarget { + target: t.target, + cancelled_at_unix: t.cancelled_at_unix.map(hive_sh4re::wire_time::from_secs), + last_fired_at_unix: t.last_fired_at_unix.map(hive_sh4re::wire_time::from_secs), + last_result: t.last_result, + }) + .collect(), + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn target(name: &str) -> hive_sh4re::WireScheduleTarget { + hive_sh4re::WireScheduleTarget { + target: name.to_owned(), + cancelled_at_unix: None, + last_fired_at_unix: None, + last_result: None, + } + } + + fn schedule(targets: &[&str]) -> hive_sh4re::WireSchedule { + hive_sh4re::WireSchedule { + id: 1, + owner: "operator".to_owned(), + body: "ping".to_owned(), + interval_seconds: None, + next_fire_at_unix: hive_sh4re::wire_time::from_secs(0), + created_at_unix: hive_sh4re::wire_time::from_secs(0), + source: hive_sh4re::WireScheduleSource::Operator, + cancelled_at_unix: None, + paused_at_unix: None, + description: None, + targets: targets.iter().map(|t| target(t)).collect(), + } + } + + #[test] + fn ghost_filter_drops_dead_agents_keeps_live_and_operator() { + let live: std::collections::HashSet = ["iris".to_owned(), "damocles".to_owned()] + .into_iter() + .collect(); + let mut schedules = vec![schedule(&["iris", "ghost", "operator", "damocles"])]; + filter_ghost_schedule_targets(&mut schedules, &live); + let kept: Vec<&str> = schedules[0] + .targets + .iter() + .map(|t| t.target.as_str()) + .collect(); + // `ghost` (destroyed) dropped; live agents + operator pseudo-target kept. + assert_eq!(kept, vec!["iris", "operator", "damocles"]); + } + + #[test] + fn ghost_filter_can_empty_targets_when_all_dead() { + let live: std::collections::HashSet = std::collections::HashSet::new(); + let mut schedules = vec![schedule(&["gone1", "gone2"])]; + filter_ghost_schedule_targets(&mut schedules, &live); + // operator is never in the live set but is always retained; here + // there's no operator target, so everything drops. + assert!(schedules[0].targets.is_empty()); + } +}