//! 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, transient, } => handle_wake(coord, agent, from, body, *transient), 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. Transient wakes /// fire the broadcast channel only (no sqlite row, no redelivery on /// restart — used by bash-task completions); durable wakes persist /// through the broker like any other message. fn handle_wake( coord: &Arc, agent: &str, from: &str, body: &str, transient: bool, ) -> hive_sh4re::Response { let broker = &coord.broker; if transient { broker.ping(agent, from, body); hive_sh4re::Response::Ok } else { match 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).await, 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). async 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"); // If the hyperhive rev is stale, route through the rebuild queue so the // container runs current nix derivations before it starts. Same logic as // `run_start`; this covers the MCP `start` tool path. let current_rev = crate::auto_update::current_flake_rev(&coord.hyperhive_flake); if let Some(ref rev) = current_rev { let stored = std::fs::read_to_string(crate::auto_update::rev_marker_path(name)).ok(); if stored.as_deref() != Some(rev.as_str()) { tracing::info!(%agent, %name, "start: rev stale — enqueuing rebuild"); coord.rebuild_queue.enqueue( crate::rebuild_queue::QueueKind::Rebuild, name.to_owned(), crate::rebuild_queue::QueueSource::Manual, format!("start {name}: rev stale — rebuilding first"), None, ); coord.emit_rebuild_queue_snapshot(); return AgentResponse::Ok; } } match crate::lifecycle::start(name).await { Ok(()) => { coord.kick_agent(name, "container started"); AgentResponse::Ok } Err(e) => AgentResponse::Err { message: format!("{e:#}"), }, } } /// `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, "enqueue restart"); coord.rebuild_queue.enqueue( crate::rebuild_queue::QueueKind::Restart, name.to_owned(), crate::rebuild_queue::QueueSource::Manual, format!("agent `{agent}` restart tool"), None, ); coord.emit_rebuild_queue_snapshot(); AgentResponse::Ok } /// 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"); 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, "enqueue rebuild"); coord.rebuild_queue.enqueue( crate::rebuild_queue::QueueKind::Rebuild, name.to_owned(), crate::rebuild_queue::QueueSource::Manual, format!("agent `{agent}` update tool"), None, ); coord.emit_rebuild_queue_snapshot(); AgentResponse::Ok } /// `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()); } }