From 58e86a3adfc619585e124cd8cef41ddfa7c045cc Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?m=C3=BCde?= Date: Mon, 6 Jul 2026 21:05:52 +0200 Subject: [PATCH] refactor(hive-c0re): shrink dashboard mod root convert to dashboard/mod.rs; state snapshot, meta inputs, tombstones, and misc api handlers move to their own files --- hive-c0re/src/dashboard/meta_inputs.rs | 279 ++++ hive-c0re/src/dashboard/misc_api.rs | 220 +++ hive-c0re/src/dashboard/mod.rs | 1528 ++------------------- hive-c0re/src/dashboard/state_snapshot.rs | 757 ++++++++++ hive-c0re/src/dashboard/tombstones.rs | 159 +++ 5 files changed, 1522 insertions(+), 1421 deletions(-) create mode 100644 hive-c0re/src/dashboard/meta_inputs.rs create mode 100644 hive-c0re/src/dashboard/misc_api.rs create mode 100644 hive-c0re/src/dashboard/state_snapshot.rs create mode 100644 hive-c0re/src/dashboard/tombstones.rs diff --git a/hive-c0re/src/dashboard/meta_inputs.rs b/hive-c0re/src/dashboard/meta_inputs.rs new file mode 100644 index 00000000..b02fc82a --- /dev/null +++ b/hive-c0re/src/dashboard/meta_inputs.rs @@ -0,0 +1,279 @@ +//! META INPUTS panel backend: walks `meta/flake.lock` into +//! `MetaInputView` rows for the snapshot, emits the `MetaInputsChanged` +//! event after lock bumps, and handles `POST /api/meta-update` (bulk +//! flake-input update + rebuild ripple via the job queue). + +use axum::{ + extract::{Form, State}, + http::StatusCode, + response::{IntoResponse, Response}, +}; +use serde::{Deserialize, Serialize}; + +use crate::coordinator::Coordinator; + +use super::{AppState, error_response}; + +#[derive(Serialize, Clone, Debug)] +pub struct MetaInputView { + /// Input key in meta's `flake.nix` — `hyperhive`, `agent-`, etc. + pub name: String, + /// Full locked sha. Not displayed verbatim; the dashboard + /// truncates to the first 12 chars for the chip. + pub rev: String, + /// Unix seconds — `locked.lastModified`. Drives the relative + /// "2h ago" timestamp on each input row. + pub last_modified: i64, + /// `original.url` if available, for the tooltip / row meta text. + #[serde(skip_serializing_if = "Option::is_none")] + pub url: Option, +} + +/// Walk `flake.lock`'s `nodes` graph from `root` and emit one +/// `MetaInputView` per fetched input, at **every** depth. That +/// surfaces the direct meta inputs (`hyperhive`, `agent-`), the +/// agent flakes' own inputs (`agent-dmatrix/mcp-matrix`, +/// `hyperhive/nixpkgs`), and any deeper transitive inputs — so the +/// operator can bump any of them individually. Names are +/// slash-separated paths from root, the syntax `nix flake update` +/// accepts for transitive inputs. +/// +/// Filtering: +/// - Inputs that resolve via a `follows` chain (lock value is an +/// array) are skipped — they alias another node, not their own +/// fetched derivation, so updating them does nothing. +/// - A node is emitted only when it carries a `locked.rev`. +/// - Each fetched node is walked exactly once (a `visited` set): +/// the lock graph shares nodes (many flakes reference one +/// nixpkgs), so without this a shared subtree re-walks per parent +/// and a cycle would recurse forever. The result is a spanning +/// tree — every input shown once, at its shallowest path. +pub(super) fn read_meta_inputs() -> Vec { + let mut out = Vec::new(); + let Ok(raw) = std::fs::read_to_string("/var/lib/hyperhive/meta/flake.lock") else { + return out; + }; + let Ok(json) = serde_json::from_str::(&raw) else { + return out; + }; + let Some(nodes) = json.get("nodes").and_then(|v| v.as_object()) else { + return out; + }; + let Some(root_name) = json.get("root").and_then(|v| v.as_str()) else { + return out; + }; + let mut visited = std::collections::HashSet::new(); + visited.insert(root_name.to_owned()); + walk_meta_inputs(nodes, root_name, "", &mut visited, &mut out); + // hyperhive first, then alphabetical. String-sorting the + // slash-paths puts every node directly above its own children + // (`agent-foo`, `agent-foo/bar`, `agent-foo/bar/baz`), so the + // result is a pre-order traversal the tree renderer can consume. + out.sort_by(|a, b| match (a.name.as_str(), b.name.as_str()) { + ("hyperhive", _) => std::cmp::Ordering::Less, + (_, "hyperhive") => std::cmp::Ordering::Greater, + _ => a.name.cmp(&b.name), + }); + out +} + +fn walk_meta_inputs( + nodes: &serde_json::Map, + node_name: &str, + prefix: &str, + visited: &mut std::collections::HashSet, + out: &mut Vec, +) { + let Some(node) = nodes.get(node_name) else { + return; + }; + let Some(inputs_map) = node.get("inputs").and_then(|v| v.as_object()) else { + return; + }; + // Two passes: claim (and emit) every direct input of this node + // before descending into any of them. A shallow input that a + // deeper flake also references then keeps its shallow path + // rather than being captured first by the deep walk. + let mut to_recurse: Vec<(String, String)> = Vec::new(); + for (alias, target) in inputs_map { + // Inputs map value is either a string (node name) or an + // array (a `follows` chain). The latter just aliases another + // node — we can't `nix flake update` it directly, so skip. + let serde_json::Value::String(target_name) = target else { + continue; + }; + // Walk each fetched node once — guards shared subtrees and + // cycles, and keeps the panel free of duplicate rows. + if !visited.insert(target_name.clone()) { + continue; + } + let Some(target_node) = nodes.get(target_name) else { + continue; + }; + let path = if prefix.is_empty() { + alias.clone() + } else { + format!("{prefix}/{alias}") + }; + if let Some(rev) = target_node + .get("locked") + .and_then(|v| v.get("rev")) + .and_then(|v| v.as_str()) + { + let last_modified = target_node + .get("locked") + .and_then(|v| v.get("lastModified")) + .and_then(serde_json::Value::as_i64) + .unwrap_or(0); + let url = target_node + .get("original") + .and_then(|v| v.get("url")) + .and_then(|v| v.as_str()) + .map(str::to_owned); + out.push(MetaInputView { + name: path.clone(), + rev: rev.to_owned(), + last_modified, + url, + }); + } + to_recurse.push((target_name.clone(), path)); + } + // Recurse hyperhive's subtree before any agent's — without this, + // when meta's top-level `nixpkgs` is a `follows` alias the + // `String` check above skips it, and the alphabetical BTreeMap + // iteration descends into `agent-*` first. The agent walk then + // claims `nixpkgs` at `agent-X/nixpkgs` instead of + // `hyperhive/nixpkgs`, which is where the operator expects it. + // Sort by the same "hyperhive first, then alpha" + // priority `read_meta_inputs` uses for the final output. + to_recurse.sort_by(|(a, _), (b, _)| match (a.as_str(), b.as_str()) { + ("hyperhive", _) => std::cmp::Ordering::Less, + (_, "hyperhive") => std::cmp::Ordering::Greater, + _ => a.cmp(b), + }); + for (target_name, path) in to_recurse { + walk_meta_inputs(nodes, &target_name, &path, visited, out); + } +} + +/// Snapshot meta/flake.lock's root inputs + emit +/// `MetaInputsChanged`. Call after any mutation that bumps a lock +/// (`run_meta_update`, `auto_update::rebuild_agent`). +pub(crate) fn emit_meta_inputs_snapshot(coord: &Coordinator) { + let inputs = read_meta_inputs(); + coord.emit_dashboard_event(crate::dashboard_events::DashboardEvent::MetaInputsChanged { + seq: coord.next_seq(), + inputs, + }); +} + +/// Form for `POST /meta-update`. Inputs ride in as a comma-separated +/// list under the `inputs` field — the JS submitter joins the +/// checked boxes since axum's `Form` extractor doesn't natively +/// decode repeated keys without a helper. +#[derive(Deserialize)] +pub(super) struct MetaUpdateForm { + inputs: String, +} + +/// Bulk-update selected meta flake inputs, then rebuild the affected +/// agents in the background. Idempotent w.r.t. selection — choosing +/// an input that's already at the latest sha is a no-op (no commit, +/// no rebuild ripple). Returns immediately after queueing the work; +/// dashboard polls for progress via container `pending` spinners + +/// the meta-inputs row sha update. +pub(super) async fn post_meta_update( + State(state): State, + Form(form): Form, +) -> Response { + let inputs: Vec = form + .inputs + .split(',') + .map(|s| s.trim().to_owned()) + .filter(|s| !s.is_empty()) + .collect(); + if inputs.is_empty() { + return error_response("meta-update: no inputs selected"); + } + let inputs_label = inputs.join(", "); + // Cascade rebuild children fan out from the MetaLock node when the + // lock bump lands — appended by the scheduler so they build against + // the post-bump lock, and a failed bump simply fans out nothing. + crate::job_queue::submit::meta_update( + &state.coord, + inputs, + crate::job_queue::Source::Manual, + format!("meta-update via dashboard ({inputs_label})"), + ); + (StatusCode::OK, "ok").into_response() +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn walk_meta_inputs_keeps_nixpkgs_under_hyperhive_post_follows_refactor() { + // Reproduce the shape where meta has + // `nixpkgs.follows = "hyperhive/nixpkgs"` at the top level + // (rendered as an array — `["hyperhive" "nixpkgs"]` — which + // walk_meta_inputs skips because we can't `nix flake update` + // a follows alias). The remaining top-level inputs are + // `hyperhive` (string) and `agent-z` (string). Without the + // hyperhive-first recursion sort, the BTreeMap alphabetical + // order descends into `agent-z` first and claims + // `nixpkgs` at `agent-z/nixpkgs`. + let raw = r#"{ + "root": "root", + "version": 7, + "nodes": { + "root": { + "inputs": { + "hyperhive": "hyperhive", + "nixpkgs": ["hyperhive", "nixpkgs"], + "agent-z": "agent-z" + } + }, + "hyperhive": { + "inputs": { "nixpkgs": "nixpkgs" }, + "locked": {"rev": "hhrev", "lastModified": 1}, + "original": {"url": "git+file:///tmp/hyperhive"} + }, + "agent-z": { + "inputs": { "nixpkgs": "nixpkgs" }, + "locked": {"rev": "azrev", "lastModified": 2}, + "original": {"url": "git+file:///tmp/agent-z"} + }, + "nixpkgs": { + "locked": {"rev": "npkrev", "lastModified": 3}, + "original": {"url": "github:NixOS/nixpkgs/nixos-26.05"} + } + } + }"#; + let json: serde_json::Value = serde_json::from_str(raw).unwrap(); + let nodes = json.get("nodes").unwrap().as_object().unwrap(); + let root_name = json.get("root").unwrap().as_str().unwrap(); + let mut visited = std::collections::HashSet::new(); + visited.insert(root_name.to_owned()); + let mut out = Vec::new(); + walk_meta_inputs(nodes, root_name, "", &mut visited, &mut out); + + let nixpkgs = out + .iter() + .find(|v| v.rev == "npkrev") + .expect("nixpkgs node should be emitted exactly once"); + assert_eq!( + nixpkgs.name, "hyperhive/nixpkgs", + "nixpkgs should be claimed under hyperhive, not under agent-z. \ + got: {:?}", + nixpkgs.name + ); + // And the agent-z path should NOT also carry a nixpkgs entry — + // the spanning-tree visited set guarantees it's claimed once. + assert!( + !out.iter().any(|v| v.name == "agent-z/nixpkgs"), + "agent-z/nixpkgs should not be emitted (already claimed under hyperhive)" + ); + } +} diff --git a/hive-c0re/src/dashboard/misc_api.rs b/hive-c0re/src/dashboard/misc_api.rs new file mode 100644 index 00000000..d4c8d175 --- /dev/null +++ b/hive-c0re/src/dashboard/misc_api.rs @@ -0,0 +1,220 @@ +//! Remaining single-endpoint dashboard handlers: the operator inbox +//! (`Y3R C4LL`) + mark-all-read, operator compose (`op-send`), +//! spawn-request, hive-wide turn stats, container resources, and the +//! audit log. + +use axum::{ + extract::{Form, Path as AxumPath, State}, + http::StatusCode, + response::{IntoResponse, Response}, +}; +use serde::Deserialize; + +use super::{AppState, error_response, scan_validated_paths, validate_agent_name}; + +/// Unread operator-directed messages for the dashboard's Y3R C4LL inbox. +/// Returns messages addressed to `"operator"` that haven't been +/// acked yet (the operator clears them via the existing +/// `POST /api/agent/operator/mark-all-read`). Newest-first; path-shaped +/// tokens are validated so the client renders file links like the +/// terminal does. Shape: `{ "messages": [{ id, from, body, at, +/// in_reply_to, file_refs }] }`. +pub(super) async fn api_operator_inbox(State(state): State) -> Response { + const INBOX_LIMIT: u64 = 100; + match state + .coord + .broker + .unread_for_recipient("operator", INBOX_LIMIT) + { + Ok(messages) => { + let items: Vec = messages + .into_iter() + .filter_map(|m| { + let crate::broker::MessageEvent::Sent { + id, + from, + body, + at, + in_reply_to, + .. + } = m + else { + return None; + }; + let file_refs = scan_validated_paths(&body); + Some(serde_json::json!({ + "id": id, + "from": from, + "body": body, + "at": hive_sh4re::wire_time::from_secs(at), + "in_reply_to": in_reply_to, + "file_refs": file_refs, + })) + }) + .collect(); + axum::Json(serde_json::json!({ "messages": items })).into_response() + } + Err(e) => error_response(&format!("operator-inbox failed: {e:#}")), + } +} + +#[derive(Deserialize)] +pub(super) struct StatsHiveQuery { + window: Option, +} + +/// Hive-wide turn-stats rollup for the dashboard swarm-stats view. +/// Aggregates every agent's `hyperhive-turn-stats.sqlite` read-only +/// (skips missing/unreadable ones). Window defaults to `24h`. +pub(super) async fn api_stats_hive( + State(state): State, + axum::extract::Query(q): axum::extract::Query, +) -> Response { + let window = crate::hive_stats::Window::parse(q.window.as_deref().unwrap_or("24h")); + axum::Json(crate::hive_stats::hive_snapshot( + window, + &state.coord.model_prices, + )) + .into_response() +} + +/// Live per-agent-container CPU + memory load from cgroup v2. Samples +/// CPU over a short interval (~200 ms), so this call briefly awaits. +pub(super) async fn api_container_resources() -> Response { + axum::Json(crate::container_stats::gather().await).into_response() +} + +/// `GET /api/audit-log` — most-recent agent-initiated privileged-action +/// audit entries, newest first (server-clamped to 500). Backs the +/// operator dashboard's audit view. Returns +/// `{ "entries": [AuditEntry…], "total": N }` so the UI can show +/// "latest 500 of N" rather than silently capping. `ts_unix` is in +/// **seconds**. +pub(super) async fn api_audit_log(State(state): State) -> Response { + const LIMIT: usize = 500; + let entries = match state.coord.audit_log.list_recent(LIMIT) { + Ok(rows) => rows, + Err(e) => return error_response(&format!("audit-log: {e:#}")), + }; + let total = match state.coord.audit_log.count_total() { + Ok(n) => n, + Err(e) => return error_response(&format!("audit-log count: {e:#}")), + }; + axum::Json(serde_json::json!({ "entries": entries, "total": total })).into_response() +} + +/// Operator-driven "clear this agent's inbox" — backs the side-panel +/// "mark all read" button. Marks every message addressed to the +/// agent as acked (backfilling `delivered_at` for any still-pending +/// rows so vacuum can collect them). Returns `{ "marked": N }` so the +/// frontend can show "cleared N messages" feedback without an extra +/// fetch. +pub(super) async fn post_mark_all_read( + State(state): State, + AxumPath(name): AxumPath, +) -> Response { + if let Some(reason) = validate_agent_name(&name) { + return (StatusCode::BAD_REQUEST, format!("bad agent name: {reason}")).into_response(); + } + match state.coord.broker.mark_all_read(&name) { + Ok(n) => { + tracing::info!(%name, marked = n, "operator marked all messages read"); + axum::Json(serde_json::json!({ "marked": n })).into_response() + } + Err(e) => error_response(&format!("mark-all-read {name} failed: {e:#}")), + } +} + +/// Operator-side compose form on the dashboard terminal. Drops a +/// message into the broker as `{from: "operator", to, body}`. Same +/// shape that per-agent web UIs use via `OperatorMsg`, but here the +/// operator picks the recipient explicitly with `@name`. No +/// validation that `to` resolves to a known agent — broker accepts +/// arbitrary recipients (and the agent's inbox grows whether or not +/// they exist, which is fine for spawn-then-greet flows). +#[derive(Deserialize)] +pub(super) struct OpSendForm { + to: String, + body: String, +} + +pub(super) async fn post_op_send( + State(state): State, + Form(form): Form, +) -> Response { + let to = form.to.trim().to_owned(); + let body = form.body.trim().to_owned(); + if to.is_empty() { + return error_response("op-send: `to` required"); + } + if body.is_empty() { + return error_response("op-send: `body` required"); + } + if to == "*" { + let errors = state + .coord + .broadcast_send(hive_sh4re::OPERATOR_RECIPIENT, &body); + if !errors.is_empty() { + return error_response(&format!( + "op-send broadcast partial fail: {}", + errors.join("; ") + )); + } + } else if let Err(e) = state.coord.broker.send(&hive_sh4re::Message { + from: hive_sh4re::OPERATOR_RECIPIENT.to_owned(), + to: to.clone(), + body, + in_reply_to: None, + }) { + return error_response(&format!("op-send to {to} failed: {e:#}")); + } + // 200 instead of 303 → the client doesn't refetch /api/state. The + // broker `send` already emitted a `MessageEvent` which the + // dashboard channel forwarder mirrors as `DashboardEvent::Sent`, + // and the page's terminal + inbox derive from that stream — so the + // operator's send shows up the same way an agent's send does, with + // no full-state refresh in between. + (axum::http::StatusCode::OK, "ok").into_response() +} + +#[derive(Deserialize)] +pub(super) struct RequestSpawnForm { + name: String, +} + +pub(super) async fn post_request_spawn( + State(state): State, + Form(form): Form, +) -> Response { + let name = form.name.trim().to_owned(); + if name.is_empty() { + return error_response("spawn: `name` required"); + } + match state.coord.approvals.submit_kind( + &name, + hive_sh4re::ApprovalKind::Spawn, + "", + None, + "operator", + ) { + Ok(id) => { + tracing::info!(%id, %name, "operator: spawn approval queued via dashboard"); + // Phase 5b: notify the dashboard event channel so live + // subscribers can append the row without a snapshot + // refetch. Spawn approvals carry no diff/sha. + state + .coord + .emit_approval_added(crate::coordinator::ApprovalAdded { + id, + agent: &name, + approval_kind: "spawn", + sha_short: None, + diff: None, + description: None, + pr_number: None, + }); + (StatusCode::OK, "ok").into_response() + } + Err(e) => error_response(&format!("request-spawn {name} failed: {e:#}")), + } +} diff --git a/hive-c0re/src/dashboard/mod.rs b/hive-c0re/src/dashboard/mod.rs index e9e70ed8..18abd4e5 100644 --- a/hive-c0re/src/dashboard/mod.rs +++ b/hive-c0re/src/dashboard/mod.rs @@ -2,43 +2,34 @@ //! container's web UI), pending approvals (with unified diff vs the applied //! repo, plus approve/deny buttons), and the manager. -use std::convert::Infallible; use std::net::SocketAddr; -use std::path::Path; use std::sync::Arc; use anyhow::{Context, Result}; -use axum::extract::Form; use axum::{ Router, - extract::{Path as AxumPath, State}, - http::{HeaderMap, StatusCode}, - response::{ - IntoResponse, Response, - sse::{Event, KeepAlive, Sse}, - }, + http::StatusCode, + response::{IntoResponse, Response}, routing::{get, post}, }; -use hive_sh4re::Approval; -use serde::{Deserialize, Serialize}; -use tokio_stream::wrappers::BroadcastStream; -use tokio_stream::{Stream, StreamExt}; -use crate::container_view::{ContainerView, claude_has_session}; use crate::coordinator::Coordinator; use crate::lifecycle; -use chrono::{DateTime, Utc}; mod approvals; mod build_logs; mod journal; mod lifecycle_ops; mod matrix_accounts; +mod meta_inputs; +mod misc_api; pub(crate) mod permissions; mod questions; mod reminders; mod schedules; mod state_files; +mod state_snapshot; +mod tombstones; mod topology; mod webhook; @@ -47,10 +38,20 @@ mod webhook; // re-exported at the module root to preserve the `crate::dashboard::approval_diff` // path across the submodule split. pub(crate) use approvals::approval_diff; +// Run after lock bumps by the job queue (`job_queue/exec.rs`); the view +// type feeds `DashboardEvent::MetaInputsChanged` (`dashboard_events.rs`). +// Re-exported to preserve the `crate::dashboard::*` paths across the split. +pub use meta_inputs::MetaInputView; +pub(crate) use meta_inputs::emit_meta_inputs_snapshot; // Run at broker-message ingest by the coordinator + the operator-msg path // (`main.rs`); re-exported to preserve the `crate::dashboard::scan_validated_paths` // path across the split. pub use state_files::scan_validated_paths; +// Called after destroy/purge/spawn finalisation (`actions.rs`); the view +// type feeds `DashboardEvent::TombstonesChanged` (`dashboard_events.rs`). +// Re-exported to preserve the `crate::dashboard::*` paths across the split. +pub use tombstones::TombstoneView; +pub(crate) use tombstones::emit_tombstones_snapshot; #[derive(Clone)] struct AppState { @@ -68,7 +69,7 @@ pub async fn serve(port: u16, coord: Arc) -> Result<()> { // API-only: the gateway static-serves the dashboard dist and proxies // non-static requests here (see hive-gateway.nix). Unmatched paths 404. let app = Router::new() - .route("/api/state", get(api_state)) + .route("/api/state", get(state_snapshot::api_state)) .route("/api/journal/{name}", get(journal::get_journal)) .route("/api/journal-host", get(journal::get_journal_host)) .route("/api/approval-diff/{id}", get(approvals::get_approval_diff)) @@ -78,10 +79,13 @@ pub async fn serve(port: u16, coord: Arc) -> Result<()> { get(matrix_accounts::get_matrix_accounts), ) .route("/api/reminders", get(reminders::api_reminders)) - .route("/api/operator-inbox", get(api_operator_inbox)) - .route("/api/stats-hive", get(api_stats_hive)) - .route("/api/container-resources", get(api_container_resources)) - .route("/api/audit-log", get(api_audit_log)) + .route("/api/operator-inbox", get(misc_api::api_operator_inbox)) + .route("/api/stats-hive", get(misc_api::api_stats_hive)) + .route( + "/api/container-resources", + get(misc_api::api_container_resources), + ) + .route("/api/audit-log", get(misc_api::api_audit_log)) .route("/api/build-logs", get(build_logs::get_build_logs_all)) .route( "/api/build-logs/{agent}", @@ -99,7 +103,10 @@ pub async fn serve(port: u16, coord: Arc) -> Result<()> { "/api/build-logs/id/{id}/raw", get(build_logs::get_build_log_raw), ) - .route("/api/agent/{name}/mark-all-read", post(post_mark_all_read)) + .route( + "/api/agent/{name}/mark-all-read", + post(misc_api::post_mark_all_read), + ) .route("/api/topology/set-parent", post(topology::post_set_parent)) .route( "/api/topology/set-parent-bulk", @@ -173,7 +180,10 @@ pub async fn serve(port: u16, coord: Arc) -> Result<()> { "/api/cancel-question/{id}", post(questions::post_cancel_question), ) - .route("/api/purge-tombstone/{name}", post(post_purge_tombstone)) + .route( + "/api/purge-tombstone/{name}", + post(tombstones::post_purge_tombstone), + ) .route( "/api/matrix-account-login", post(matrix_accounts::post_matrix_account_login), @@ -186,11 +196,17 @@ pub async fn serve(port: u16, coord: Arc) -> Result<()> { "/api/retry-reminder/{id}", post(reminders::post_retry_reminder), ) - .route("/api/request-spawn", post(post_request_spawn)) - .route("/api/op-send", post(post_op_send)) - .route("/api/meta-update", post(post_meta_update)) - .route("/api/dashboard/stream", get(dashboard_stream)) - .route("/api/dashboard/history", get(dashboard_history)) + .route("/api/request-spawn", post(misc_api::post_request_spawn)) + .route("/api/op-send", post(misc_api::post_op_send)) + .route("/api/meta-update", post(meta_inputs::post_meta_update)) + .route( + "/api/dashboard/stream", + get(state_snapshot::dashboard_stream), + ) + .route( + "/api/dashboard/history", + get(state_snapshot::dashboard_history), + ) // No static fallback — the gateway owns the dist; unmatched paths 404. .with_state(AppState { coord }); // Binds loopback-only; external access via gateway. @@ -254,1183 +270,6 @@ fn try_bind(addr: SocketAddr) -> std::io::Result { sock.listen(1024) } -#[allow(clippy::struct_excessive_bools)] -#[derive(Serialize)] -struct StateSnapshot { - /// Broker seq at the moment this snapshot was assembled. Clients - /// dedupe their buffered SSE traffic against this value: any - /// `MessageEvent` with `seq <= snapshot.seq` is already reflected in - /// the snapshot (or pre-dates it); anything with `seq > snapshot.seq` - /// is post-snapshot and should be applied. Set to 0 in the - /// pre-emit case (no events ever fired) — clients treat that as - /// "apply everything you've buffered". - seq: u64, - hostname: String, - any_stale: bool, - containers: Vec, - transients: Vec, - approvals: Vec, - /// Last 30 resolved approvals (approved / denied / failed), newest- - /// first. Drives the "history" tab on the approvals section. - approval_history: Vec, - /// Pending operator-targeted questions (`target IS NULL`). Any - /// agent can `ask` the operator and `ask` returns immediately with - /// the id; on `/answer-question` we mark the row answered and - /// fire `HelperEvent::QuestionAnswered` back into the asker's - /// inbox. Peer-to-peer questions live in the same table but never - /// surface here (see `OperatorQuestions::pending`). - questions: Vec, - /// Last 20 answered questions, newest-first. - question_history: Vec, - /// State dirs (config history + claude creds + /state/ notes) that - /// survive after a destroy-without-purge. The operator can re-spawn - /// with the same name to resume, or PURG3 to wipe them. - tombstones: Vec, - /// Sub-agents whose FNV-1a hashed web UI port collides with at - /// least one other agent. Operator resolves by renaming. The - /// dashboard renders a banner at the top listing each cluster. - port_conflicts: Vec, - /// Inputs in `meta/flake.lock` the operator can selectively - /// `nix flake update`. Hyperhive first, then `agent-` rows. - meta_inputs: Vec, - /// True while a dashboard-triggered `meta-update` (flake lock bump + - /// agent rebuild ripple) is running in the background. Lets a - /// client that cold-loads mid-update render the META INPUTS panel's - /// disabled "updating…" state; live transitions arrive via the - /// `MetaUpdateRunning` event. - meta_update_running: bool, - /// Current state of the global job queue — pending + running DAGs - /// (rebuild / meta-update / spawn / power ops) with their per-node - /// breakdowns, plus the most recent few terminal DAGs the queue - /// retains for history. Live transitions arrive via the - /// `RebuildQueueChanged` event. See `job_queue/`. Field name kept - /// from the old flat queue for wire compatibility. - rebuild_queue: Vec, - /// Whether the hive-forge container is up. When true the dashboard - /// links each container's config + each approval's commit into the - /// forge's `agent-configs` repos. - forge_present: bool, - /// Whether the matrix GUI is reachable at `/matrix/`. Sourced from - /// `HIVE_MATRIX_GUI_ENABLED` env var (set by the c0re NixOS module - /// when `services.hyperhive.matrix.gui.enable` is on). The gateway - /// (hive-gateway.nix) does the actual `/matrix/` static serving; - /// this flag is just an availability signal for iris's dashboard - /// chrome so the `M4TR1X →` tab doesn't flash when the GUI is off. - matrix_gui_enabled: bool, - /// Whether `hive-gateway` is in front of this dashboard. Sourced - /// from the `HIVE_GATEWAY_ENABLED` env var, which the c0re NixOS - /// module now always sets (the gateway runs unconditionally - /// alongside hyperhive), so this is effectively always true: the - /// dashboard frontend builds same-origin `/agent//` links to - /// the per-agent web UI (the gateway routes them via the - /// runtime-generated `agents.conf` include file — see - /// `gateway_nginx.rs`). The `false` branch (direct - /// `http://:/` TCP links) is retained as a defensive - /// fallback for the env being unset. See `docs/gateway.md::Vhost map`. - gateway_enabled: bool, - /// Public URL of the forge vhost served by hive-gateway (e.g. - /// `"https://forge.pr1ma.darkest.space"`). Sourced from the - /// `HIVE_FORGE_PUBLIC_URL` env var, which the c0re NixOS module - /// sets when `forge.behindGateway = true`. `None` when absent — - /// the frontend falls back to `http://:3000`. - forge_public_url: Option, - /// Human name of this single-host hive instance (e.g. `"pr1ma"`). - /// Sourced from `HYPERHIVE_HIVE_NAME` env var, set by the c0re - /// NixOS module from `services.hyperhive.hiveName`. `None` when - /// the option is unset — chrome falls back to `hostname`. - hive_name: Option, - /// Human name of the wider swarm this hive belongs to (e.g. - /// `"constellat1on"`). Sourced from `HYPERHIVE_SWARM_NAME` env - /// var, set from `services.hyperhive.swarmName`. `None` when - /// unset — chrome omits the swarm segment of the breadcrumb. - swarm_name: Option, - /// Peer hives in the same swarm. Parsed from `HYPERHIVE_PEERS` - /// (JSON array of `{domain,cert_fingerprint}` objects, emitted by - /// the c0re NixOS module from `services.hyperhive.swarm.peers`). - /// Empty on single-hive deploys. Feeds the P33RS dashboard tab. - peer_hives: Vec, - /// Server-level warnings for the dashboard's top-of-page banner - /// (currently host disk-pressure; more producers can be added - /// backend-side). Empty when all clear. Built by - /// `host_stats::server_warnings`; the frontend renders this list - /// generically, so new warning kinds need no frontend change. - server_warnings: Vec, -} - -/// One peer hive for the P33RS dashboard tab. Derived from -/// `HYPERHIVE_PEERS` env; `url` is the peer's HTTPS dashboard root. -/// `cert_fingerprint` is `Some("sha256:")` when the peer uses a -/// self-signed cert and the operator pinned its fingerprint in -/// `services.hyperhive.swarm.peers`. -#[derive(Serialize)] -struct PeerHiveView { - name: String, - url: String, - cert_fingerprint: Option, -} - -/// `OpQuestion` + computed `question_refs` / `answer_refs`. Built -/// from the snapshot read; the live channel attaches the same -/// fields directly on `QuestionAdded` / `QuestionResolved`. -#[derive(Serialize)] -struct QuestionView { - #[serde(flatten)] - inner: crate::operator_questions::OpQuestion, - #[serde(skip_serializing_if = "Vec::is_empty")] - question_refs: Vec, - #[serde(skip_serializing_if = "Vec::is_empty")] - answer_refs: Vec, -} - -impl QuestionView { - fn from_question(q: crate::operator_questions::OpQuestion) -> Self { - let question_refs = scan_validated_paths(&q.question); - let answer_refs = q - .answer - .as_deref() - .map(scan_validated_paths) - .unwrap_or_default(); - Self { - inner: q, - question_refs, - answer_refs, - } - } -} - -#[derive(Serialize)] -struct PortConflict { - port: u16, - /// All agent names sharing this port (sorted, ≥2 entries). - agents: Vec, -} - -#[derive(Serialize, Clone, Debug)] -pub struct TombstoneView { - pub name: String, - /// Bytes used by the state dir tree. Cheap-ish to compute; let the - /// operator know how much they're holding onto. - pub state_bytes: u64, - /// Mtime (unix seconds) of the state dir; rough "last seen". - pub last_seen: i64, - pub has_creds: bool, -} - -#[derive(Serialize)] -struct TransientView { - name: String, - kind: &'static str, - secs: u64, -} - -#[derive(Serialize)] -struct ApprovalHistoryView { - id: i64, - agent: String, - kind: &'static str, - /// First 12 chars of the canonical sha (preferred) or - /// manager-supplied ref. None for resolved spawn approvals. - sha_short: Option, - /// `approved` / `denied` / `failed`. - status: &'static str, - /// RFC 3339 UTC. Renders as a relative time on the dashboard. - resolved_at: DateTime, - /// Operator-supplied deny reason (for `denied`) or build error - /// (for `failed`). None on `approved`. - #[serde(skip_serializing_if = "Option::is_none")] - note: Option, -} - -#[derive(Serialize)] -struct ApprovalView { - id: i64, - agent: String, - kind: &'static str, - /// First 12 chars of the `commit_ref`, for `ApplyCommit` only. - sha_short: Option, - /// Raw unified diff text, for `ApplyCommit` only. The client splits - /// on `\n` and per-line classifies (`+` / `-` / `@@` / `--- ` / `+++ ` - /// → diff-add / diff-del / diff-hunk / diff-file). Shipping raw - /// instead of pre-rendered HTML saves bytes on the wire (no - /// per-line `` markup) and removes the only HTML-escape - /// surface from the snapshot. - diff: Option, - /// Manager-supplied description shown on the approval card. - #[serde(skip_serializing_if = "Option::is_none")] - description: Option, - /// Forge PR number, for `MergeConfigPr` only. Lets the frontend - /// build a "review PR on forge" link - /// (`{forgeBase}/agent-configs/{agent}/pulls/{pr_number}`) the same - /// way it builds the `apply_commit` "commit on forge" link from the - /// sha. `None` for every other kind. - #[serde(skip_serializing_if = "Option::is_none")] - pr_number: Option, - /// Raw `commit_ref` payload for `UpdateMetaInputs` (JSON-encoded - /// `Vec` of input names; `"[]"` = all inputs) and - /// `SchedulePrompt` (JSON-encoded `SchedulePromptPayload`). The - /// frontend parses this to render a human-readable card body. - /// `None` for every other kind. - #[serde(skip_serializing_if = "Option::is_none")] - commit_ref: Option, - /// RFC 3339 UTC time the approval was queued. Rendered as a - /// relative time on the card so the operator can spot a stale - /// request. - requested_at: DateTime, -} - -/// Replace silent `.unwrap_or_default()` on the data sources behind -/// `/api/state` so that whichever query degrades surfaces in journald -/// instead of leaving the operator staring at an empty list. The -/// dashboard still degrades to a sensible default value; the warn -/// is just the diagnostic breadcrumb the old code swallowed. -fn log_default(what: &str, result: std::result::Result) -> T -where - T: Default, - E: std::fmt::Debug, -{ - match result { - Ok(v) => v, - Err(e) => { - tracing::warn!(target: "api_state", source = %what, error = ?e, "snapshot source failed; using default"); - T::default() - } - } -} - -/// Window over which container crashes count toward the `agents_crashing` -/// banner warning. Wide enough that a crash-looping container (restarted -/// by `Restart=on-failure` every few seconds) keeps the warning lit -/// between flaps, short enough that a single recovered crash clears within -/// minutes. -const CRASH_WARNING_WINDOW: std::time::Duration = std::time::Duration::from_mins(10); - -async fn api_state(headers: HeaderMap, State(state): State) -> axum::Json { - let host = headers - .get("host") - .and_then(|h| h.to_str().ok()) - .unwrap_or("localhost"); - let hostname = host.split(':').next().unwrap_or(host).to_owned(); - - // Capture the unified dashboard-channel seq *before* any read so the - // dedupe contract is "events with seq > snapshot.seq are - // post-snapshot, never missed." An event landing during snapshot - // construction may be doubly applied (snapshot caught the write + - // client also applies the SSE frame) — that's a renderer's problem - // to make idempotent, not ours to avoid here. - let seq = state.coord.current_seq(); - - // Refresh the coordinator's cached container snapshot before - // reading. Cold-load clients then see whatever the latest rescan - // produced; live clients converge via the matching - // `ContainerStateChanged` / `ContainerRemoved` events the rescan - // emits. - // - // Bound the rescan: it shells out (`nixos-container list` etc.), so a - // saturated/wedged build backend — e.g. hive-c0re mid-startup-sweep - // hammering slow `nixos-container update` subprocesses — can stall it - // long enough that `/api/state` hangs for the whole request (the - // ~minute-long /state reported in the field). On timeout we skip the - // fresh rescan and serve the last cached snapshot instead; live - // clients still converge via the SSE events a later successful rescan - // emits, and the next /state call retries the refresh. Introspection - // stays responsive regardless of the build backend's health. - if tokio::time::timeout( - std::time::Duration::from_secs(3), - state.coord.rescan_containers_and_emit(), - ) - .await - .is_err() - { - tracing::warn!( - "api_state: container rescan exceeded 3s (build backend likely saturated); \ - serving last cached snapshot" - ); - } - let containers = state.coord.containers_snapshot().await; - let any_stale = containers.iter().any(|c| c.needs_update); - let transient_snapshot = state.coord.transient_snapshot(); - let pending_approvals = approvals::gc_orphans( - &state.coord, - log_default("approvals.pending", state.coord.approvals.pending()), - ); - let transients = build_transient_views(&containers, &transient_snapshot); - let approvals = build_approval_views(pending_approvals).await; - let approval_history = log_default( - "approvals.recent_resolved", - state.coord.approvals.recent_resolved(30), - ) - .into_iter() - .map(history_view) - .collect(); - let tombstones = build_tombstone_views(&state.coord, &containers, &transient_snapshot); - let port_conflicts = build_port_conflicts(&containers); - - // Both operator-targeted and peer threads surface on the dashboard - // (the client filters by target). Each row is wrapped in QuestionView - // so the snapshot carries the same file_refs the live event variants - // attach. - let questions: Vec = - log_default("questions.pending_all", state.coord.questions.pending_all()) - .into_iter() - .map(QuestionView::from_question) - .collect(); - let question_history: Vec = log_default( - "questions.recent_answered_all", - state.coord.questions.recent_answered_all(20), - ) - .into_iter() - .map(QuestionView::from_question) - .collect(); - - // Banner warnings: host probes (disk) + agent-state (pending logins, - // crashing agents). Built before the response struct because the - // agent-state producer borrows `containers`, which moves in below. - let server_warnings = { - let mut w = crate::host_stats::server_warnings(); - w.extend(crate::host_stats::agent_state_warnings( - &containers, - &state.coord.recent_crash_counts(CRASH_WARNING_WINDOW), - )); - w - }; - - axum::Json(StateSnapshot { - seq, - hostname, - any_stale, - containers, - transients, - approvals, - approval_history, - meta_inputs: read_meta_inputs(), - meta_update_running: state.coord.meta_update_in_progress(), - questions, - question_history, - tombstones, - port_conflicts, - rebuild_queue: state.coord.job_queue.snapshot(), - forge_present: crate::forge::is_present().await, - matrix_gui_enabled: std::env::var_os("HIVE_MATRIX_GUI_ENABLED").is_some_and(|v| { - // Accept any truthy string ("1", "true", "yes") since the - // env var is set by NixOS module wiring with the literal - // "1"; defensive parse so manual overrides also work. - let s = v.to_string_lossy().to_ascii_lowercase(); - matches!(s.as_str(), "1" | "true" | "yes") - }), - gateway_enabled: std::env::var_os("HIVE_GATEWAY_ENABLED").is_some_and(|v| { - // Same truthy-string parse as `matrix_gui_enabled`; the - // env var is set by the c0re NixOS module to the literal - // "1" — the gateway always runs alongside hyperhive. - let s = v.to_string_lossy().to_ascii_lowercase(); - matches!(s.as_str(), "1" | "true" | "yes") - }), - forge_public_url: std::env::var("HIVE_FORGE_PUBLIC_URL") - .ok() - .filter(|s| !s.is_empty()), - hive_name: std::env::var("HYPERHIVE_HIVE_NAME") - .ok() - .filter(|s| !s.is_empty()), - swarm_name: std::env::var("HYPERHIVE_SWARM_NAME") - .ok() - .filter(|s| !s.is_empty()), - peer_hives: parse_peer_hives(), - server_warnings, - }) -} - -/// Parse `HYPERHIVE_PEERS` env var into dashboard-ready `PeerHiveView` -/// entries. The env var is a JSON array of `{domain, cert_fingerprint}` -/// objects emitted by the c0re NixOS module from -/// `services.hyperhive.swarm.peers`. Each entry becomes -/// `{ name: domain, url: "https://domain/" }` for the P33RS tab. -/// Returns empty vec when unset (single-hive deploy). -fn parse_peer_hives() -> Vec { - #[derive(serde::Deserialize)] - struct Raw { - domain: String, - cert_fingerprint: Option, - } - let Ok(json) = std::env::var("HYPERHIVE_PEERS") else { - return Vec::new(); - }; - let Ok(raw): Result, _> = serde_json::from_str(&json) else { - tracing::warn!("HYPERHIVE_PEERS is not valid JSON; ignoring"); - return Vec::new(); - }; - raw.into_iter() - .map(|r| { - let cert_fingerprint = r.cert_fingerprint.and_then(|fp| { - if validate_cert_fingerprint(&fp) { - Some(fp) - } else { - tracing::warn!( - domain = %r.domain, - fingerprint = %fp, - "HYPERHIVE_PEERS: invalid cert_fingerprint format \ - (expected `sha256:<64 hex chars>`); ignoring fingerprint" - ); - None - } - }); - PeerHiveView { - name: r.domain.clone(), - url: format!("https://{}/", r.domain), - cert_fingerprint, - } - }) - .collect() -} - -/// Validate a TLS certificate fingerprint string from `HYPERHIVE_PEERS`. -/// Accepts `sha256:<64 hex chars>` (upper or lower case). -fn validate_cert_fingerprint(fp: &str) -> bool { - let Some(hex) = fp.strip_prefix("sha256:") else { - return false; - }; - hex.len() == 64 && hex.chars().all(|c| c.is_ascii_hexdigit()) -} - -/// Group live containers by their assigned web UI port; clusters with -/// more than one member are port-hash collisions the operator needs -/// to resolve by renaming. Manager (fixed at 8000) and sub-agents -/// (8100..8999) can't collide with each other — collisions are -/// strictly between sub-agents. -fn build_port_conflicts(containers: &[ContainerView]) -> Vec { - let mut by_port: std::collections::BTreeMap> = - std::collections::BTreeMap::new(); - for c in containers { - by_port.entry(c.port).or_default().push(c.name.clone()); - } - by_port - .into_iter() - .filter(|(_, agents)| agents.len() > 1) - .map(|(port, mut agents)| { - agents.sort(); - PortConflict { port, agents } - }) - .collect() -} - -#[derive(Serialize, Clone, Debug)] -pub struct MetaInputView { - /// Input key in meta's `flake.nix` — `hyperhive`, `agent-`, etc. - pub name: String, - /// Full locked sha. Not displayed verbatim; the dashboard - /// truncates to the first 12 chars for the chip. - pub rev: String, - /// Unix seconds — `locked.lastModified`. Drives the relative - /// "2h ago" timestamp on each input row. - pub last_modified: i64, - /// `original.url` if available, for the tooltip / row meta text. - #[serde(skip_serializing_if = "Option::is_none")] - pub url: Option, -} - -/// Walk `flake.lock`'s `nodes` graph from `root` and emit one -/// `MetaInputView` per fetched input, at **every** depth. That -/// surfaces the direct meta inputs (`hyperhive`, `agent-`), the -/// agent flakes' own inputs (`agent-dmatrix/mcp-matrix`, -/// `hyperhive/nixpkgs`), and any deeper transitive inputs — so the -/// operator can bump any of them individually. Names are -/// slash-separated paths from root, the syntax `nix flake update` -/// accepts for transitive inputs. -/// -/// Filtering: -/// - Inputs that resolve via a `follows` chain (lock value is an -/// array) are skipped — they alias another node, not their own -/// fetched derivation, so updating them does nothing. -/// - A node is emitted only when it carries a `locked.rev`. -/// - Each fetched node is walked exactly once (a `visited` set): -/// the lock graph shares nodes (many flakes reference one -/// nixpkgs), so without this a shared subtree re-walks per parent -/// and a cycle would recurse forever. The result is a spanning -/// tree — every input shown once, at its shallowest path. -fn read_meta_inputs() -> Vec { - let mut out = Vec::new(); - let Ok(raw) = std::fs::read_to_string("/var/lib/hyperhive/meta/flake.lock") else { - return out; - }; - let Ok(json) = serde_json::from_str::(&raw) else { - return out; - }; - let Some(nodes) = json.get("nodes").and_then(|v| v.as_object()) else { - return out; - }; - let Some(root_name) = json.get("root").and_then(|v| v.as_str()) else { - return out; - }; - let mut visited = std::collections::HashSet::new(); - visited.insert(root_name.to_owned()); - walk_meta_inputs(nodes, root_name, "", &mut visited, &mut out); - // hyperhive first, then alphabetical. String-sorting the - // slash-paths puts every node directly above its own children - // (`agent-foo`, `agent-foo/bar`, `agent-foo/bar/baz`), so the - // result is a pre-order traversal the tree renderer can consume. - out.sort_by(|a, b| match (a.name.as_str(), b.name.as_str()) { - ("hyperhive", _) => std::cmp::Ordering::Less, - (_, "hyperhive") => std::cmp::Ordering::Greater, - _ => a.name.cmp(&b.name), - }); - out -} - -fn walk_meta_inputs( - nodes: &serde_json::Map, - node_name: &str, - prefix: &str, - visited: &mut std::collections::HashSet, - out: &mut Vec, -) { - let Some(node) = nodes.get(node_name) else { - return; - }; - let Some(inputs_map) = node.get("inputs").and_then(|v| v.as_object()) else { - return; - }; - // Two passes: claim (and emit) every direct input of this node - // before descending into any of them. A shallow input that a - // deeper flake also references then keeps its shallow path - // rather than being captured first by the deep walk. - let mut to_recurse: Vec<(String, String)> = Vec::new(); - for (alias, target) in inputs_map { - // Inputs map value is either a string (node name) or an - // array (a `follows` chain). The latter just aliases another - // node — we can't `nix flake update` it directly, so skip. - let serde_json::Value::String(target_name) = target else { - continue; - }; - // Walk each fetched node once — guards shared subtrees and - // cycles, and keeps the panel free of duplicate rows. - if !visited.insert(target_name.clone()) { - continue; - } - let Some(target_node) = nodes.get(target_name) else { - continue; - }; - let path = if prefix.is_empty() { - alias.clone() - } else { - format!("{prefix}/{alias}") - }; - if let Some(rev) = target_node - .get("locked") - .and_then(|v| v.get("rev")) - .and_then(|v| v.as_str()) - { - let last_modified = target_node - .get("locked") - .and_then(|v| v.get("lastModified")) - .and_then(serde_json::Value::as_i64) - .unwrap_or(0); - let url = target_node - .get("original") - .and_then(|v| v.get("url")) - .and_then(|v| v.as_str()) - .map(str::to_owned); - out.push(MetaInputView { - name: path.clone(), - rev: rev.to_owned(), - last_modified, - url, - }); - } - to_recurse.push((target_name.clone(), path)); - } - // Recurse hyperhive's subtree before any agent's — without this, - // when meta's top-level `nixpkgs` is a `follows` alias the - // `String` check above skips it, and the alphabetical BTreeMap - // iteration descends into `agent-*` first. The agent walk then - // claims `nixpkgs` at `agent-X/nixpkgs` instead of - // `hyperhive/nixpkgs`, which is where the operator expects it. - // Sort by the same "hyperhive first, then alpha" - // priority `read_meta_inputs` uses for the final output. - to_recurse.sort_by(|(a, _), (b, _)| match (a.as_str(), b.as_str()) { - ("hyperhive", _) => std::cmp::Ordering::Less, - (_, "hyperhive") => std::cmp::Ordering::Greater, - _ => a.cmp(b), - }); - for (target_name, path) in to_recurse { - walk_meta_inputs(nodes, &target_name, &path, visited, out); - } -} - -/// Transient state for agents whose container does NOT yet exist -/// (`Spawning`). Lifecycle ops on existing containers surface as -/// `ContainerView.pending` inline; this list only catches pre-creation. -fn build_transient_views( - containers: &[ContainerView], - transient_snapshot: &std::collections::HashMap, -) -> Vec { - transient_snapshot - .iter() - .filter(|(name, _)| !containers.iter().any(|c| &c.name == *name)) - .map(|(name, st)| TransientView { - name: name.clone(), - kind: transient_label(st.kind), - secs: st.since.elapsed().as_secs(), - }) - .collect() -} - -/// Render each pending approval into its dashboard view (short sha + -/// unified diff for `ApplyCommit`, just the name for `Spawn`). -/// Project a resolved sqlite row into the lean shape the dashboard -/// history tab consumes — no `diff_html` (rendering 30 of them -/// per /api/state poll would mean 30 git diffs per refresh). -fn history_view(a: Approval) -> ApprovalHistoryView { - let displayed = a.fetched_sha.as_deref().unwrap_or(&a.commit_ref); - let sha_short = if displayed.is_empty() { - None - } else { - Some(displayed[..displayed.len().min(12)].to_owned()) - }; - let status = match a.status { - hive_sh4re::ApprovalStatus::Approved => "approved", - hive_sh4re::ApprovalStatus::Denied => "denied", - hive_sh4re::ApprovalStatus::Failed => "failed", - hive_sh4re::ApprovalStatus::Cancelled => "cancelled", - // Pending shouldn't appear in recent_resolved, but be defensive. - hive_sh4re::ApprovalStatus::Pending => "pending", - }; - let kind = match a.kind { - hive_sh4re::ApprovalKind::ApplyCommit => "apply_commit", - hive_sh4re::ApprovalKind::Spawn => "spawn", - hive_sh4re::ApprovalKind::InitConfig => "init_config", - hive_sh4re::ApprovalKind::UpdateMetaInputs => "update_meta_inputs", - hive_sh4re::ApprovalKind::SchedulePrompt => "schedule_prompt", - hive_sh4re::ApprovalKind::MergeConfigPr => "merge_config_pr", - }; - ApprovalHistoryView { - id: a.id, - agent: a.agent, - kind, - sha_short, - status, - resolved_at: a.resolved_at.unwrap_or_default(), - note: a.note, - } -} - -async fn build_approval_views(approvals: Vec) -> Vec { - let mut out = Vec::with_capacity(approvals.len()); - for a in approvals { - out.push(match a.kind { - hive_sh4re::ApprovalKind::ApplyCommit => { - // Prefer the canonical fetched sha from applied; - // commit_ref is only the manager's claim and may be - // amended out from under us. - let displayed = a.fetched_sha.as_deref().unwrap_or(&a.commit_ref); - let sha = displayed[..displayed.len().min(12)].to_owned(); - let diff = approval_diff(&a.agent, a.id).await; - ApprovalView { - id: a.id, - agent: a.agent.clone(), - kind: "apply_commit", - sha_short: Some(sha), - diff: Some(diff), - description: a.description, - pr_number: None, - commit_ref: None, - requested_at: a.requested_at, - } - } - hive_sh4re::ApprovalKind::Spawn => ApprovalView { - id: a.id, - agent: a.agent, - kind: "spawn", - sha_short: None, - diff: None, - description: a.description, - pr_number: None, - commit_ref: None, - requested_at: a.requested_at, - }, - hive_sh4re::ApprovalKind::InitConfig => ApprovalView { - id: a.id, - agent: a.agent, - kind: "init_config", - sha_short: None, - diff: None, - description: a.description, - pr_number: None, - commit_ref: None, - requested_at: a.requested_at, - }, - hive_sh4re::ApprovalKind::UpdateMetaInputs => ApprovalView { - id: a.id, - agent: a.agent, - kind: "update_meta_inputs", - sha_short: None, - diff: None, - description: a.description, - pr_number: None, - commit_ref: Some(a.commit_ref), - requested_at: a.requested_at, - }, - hive_sh4re::ApprovalKind::SchedulePrompt => ApprovalView { - id: a.id, - agent: a.agent, - kind: "schedule_prompt", - sha_short: None, - diff: None, - description: a.description, - pr_number: None, - commit_ref: Some(a.commit_ref), - requested_at: a.requested_at, - }, - hive_sh4re::ApprovalKind::MergeConfigPr => { - // commit_ref = PR number; fetched_sha = the reviewed PR - // head. Show the head sha; the forge PR diff surface is - // a later phase of the PR-based config flow — None for now. - let sha = a - .fetched_sha - .as_deref() - .map(|s| s[..s.len().min(12)].to_owned()); - // Surface the PR number so the frontend can link to the - // PR on the forge. commit_ref holds the number as text. - let pr_number = a.commit_ref.parse::().ok(); - ApprovalView { - id: a.id, - agent: a.agent, - kind: "merge_config_pr", - sha_short: sha, - diff: None, - description: a.description, - pr_number, - commit_ref: None, - requested_at: a.requested_at, - } - } - }); - } - out -} - -/// State-dir names that don't appear in the live container list. Each -/// one surfaces in the dashboard as a row with R3V1V3 + PURG3 actions. -fn build_tombstone_views( - coord: &Coordinator, - containers: &[ContainerView], - transient_snapshot: &std::collections::HashMap, -) -> Vec { - let _ = coord; // kept_state_names is a free fn but takes &self by future plan - let live: std::collections::HashSet<&str> = containers - .iter() - .map(|c| c.name.as_str()) - .chain(transient_snapshot.keys().map(String::as_str)) - .collect(); - Coordinator::kept_state_names() - .into_iter() - .filter(|name| !live.contains(name.as_str())) - .map(|name| { - let root = Coordinator::agent_state_root(&name); - let state_bytes = dir_size_bytes(&root); - let last_seen = std::fs::metadata(&root) - .and_then(|m| m.modified()) - .ok() - .and_then(|t| t.duration_since(std::time::UNIX_EPOCH).ok()) - .and_then(|d| i64::try_from(d.as_secs()).ok()) - .unwrap_or(0); - let has_creds = claude_has_session(&Coordinator::agent_claude_dir(&name)); - TombstoneView { - name, - state_bytes, - last_seen, - has_creds, - } - }) - .collect() -} - -/// Sum the byte size of every regular file under `root`. Cheap to compute -/// for typical agent state (config repo + claude creds + notes file — -/// usually a few MB); fine to do inline on each /api/state. Returns 0 on -/// any error. -fn dir_size_bytes(root: &Path) -> u64 { - fn walk(p: &Path, acc: &mut u64) { - let Ok(rd) = std::fs::read_dir(p) else { return }; - for entry in rd.flatten() { - let Ok(ft) = entry.file_type() else { continue }; - if ft.is_dir() { - walk(&entry.path(), acc); - } else if ft.is_file() - && let Ok(meta) = entry.metadata() - { - *acc += meta.len(); - } - } - } - let mut total = 0u64; - walk(root, &mut total); - total -} - -async fn dashboard_history(State(state): State) -> Response { - // Backfill source for the dashboard terminal. Returns up to ~200 - // historical broker messages (no other event kinds are persisted) - // converted to `DashboardEvent::Sent` JSON so the client can replay - // through the same dispatch path as live frames. Wrapped in - // `{ seq, events }`: the seq is the dashboard channel's high-water - // mark at fetch time. Clients use it to dedupe their buffered live - // SSE traffic (drop anything with `seq <= history_seq`) so a frame - // that lands between SSE-subscribe and history-fetch isn't shown - // twice and isn't lost. Historical rows carry `seq = 0`; the - // boundary seq is what closes the dedupe window. - const HISTORY_LIMIT: u64 = 200; - let seq = state.coord.current_seq(); - match state.coord.broker.recent_all(HISTORY_LIMIT) { - Ok(mut messages) => { - messages.reverse(); - let events: Vec = messages - .into_iter() - .map(|m| match m { - crate::broker::MessageEvent::Sent { - id, - from, - to, - body, - at, - in_reply_to, - } => { - let file_refs = scan_validated_paths(&body); - crate::dashboard_events::DashboardEvent::Sent { - seq: 0, - id, - from, - to, - body, - at: hive_sh4re::wire_time::from_secs(at), - in_reply_to, - file_refs, - } - } - crate::broker::MessageEvent::Delivered { - id, - from, - to, - body, - at, - in_reply_to, - } => { - let file_refs = scan_validated_paths(&body); - crate::dashboard_events::DashboardEvent::Delivered { - seq: 0, - id, - from, - to, - body, - at: hive_sh4re::wire_time::from_secs(at), - in_reply_to, - file_refs, - } - } - }) - .collect(); - axum::Json(serde_json::json!({ "seq": seq, "events": events })).into_response() - } - Err(e) => error_response(&format!("dashboard/history failed: {e:#}")), - } -} - -/// `/dashboard/stream` query string. Today's only field is `kinds`: -/// a comma-separated allow-list of event-`kind` strings. -/// Empty / absent ⇒ no filter (current behaviour, all variants -/// forwarded). Set ⇒ only the named kinds reach the subscriber, -/// non-matches are skipped before the JSON serialise cost. -/// -/// Useful for narrow pages (e.g. `flow.js` only cares about `sent` -/// / `delivered` / `container_state_changed` / `container_removed`) -/// that want to drop the dispatch overhead on every unrelated mutation. -#[derive(Deserialize, Default)] -struct DashboardStreamQuery { - /// Comma-separated event kinds to forward. Each token is - /// trimmed; unknown kinds are silently ignored on lookup - /// (subscriber sees nothing instead of an error). - kinds: Option, -} - -async fn dashboard_stream( - State(state): State, - axum::extract::Query(q): axum::extract::Query, -) -> Sse>> { - let rx = state.coord.dashboard_subscribe(); - // Pre-parse the allow-list once at subscription time, so the - // per-event hot path is just a `HashSet::contains` on a - // `&'static str` — no string churn per frame. - let kind_filter: Option> = q.kinds.and_then(|raw| { - let set: std::collections::HashSet = raw - .split(',') - .map(str::trim) - .filter(|s| !s.is_empty()) - .map(str::to_owned) - .collect(); - if set.is_empty() { None } else { Some(set) } - }); - let stream = BroadcastStream::new(rx).filter_map(move |res| { - // Drop lagged frames. Browsers reconnect; the seq dedupe on - // reconnect skips any frame already reflected in the snapshot. - let event = res.ok()?; - if let Some(filter) = kind_filter.as_ref() - && !filter.contains(event.kind_tag()) - { - return None; - } - let json = serde_json::to_string(&event).ok()?; - Some(Ok(Event::default().data(json))) - }); - Sse::new(stream).keep_alive(KeepAlive::default()) -} - -#[derive(Deserialize)] -struct RequestSpawnForm { - name: String, -} - -#[cfg(test)] -mod tests { - use super::*; - - #[test] - fn problem_details_carry_rfc9457_status_and_detail() { - // Contract the frontend depends on: the problem_details crate - // serialises the RFC 9457 members we rely on — `status` (numeric) - // and `detail` (the caller message; the FE reads `.detail`). - let pd = problem_details::ProblemDetails::from_status_code(StatusCode::BAD_REQUEST) - .with_detail("bad input"); - let v = serde_json::to_value(&pd).expect("problem details serialise"); - assert_eq!(v["status"], 400); - assert_eq!(v["detail"], "bad input"); - // The 500 wrapper path carries the internal-error status. - let five = - problem_details::ProblemDetails::from_status_code(StatusCode::INTERNAL_SERVER_ERROR) - .with_detail("boom"); - let fv = serde_json::to_value(&five).expect("problem details serialise"); - assert_eq!(fv["status"], 500); - } - - #[test] - fn walk_meta_inputs_keeps_nixpkgs_under_hyperhive_post_follows_refactor() { - // Reproduce the shape where meta has - // `nixpkgs.follows = "hyperhive/nixpkgs"` at the top level - // (rendered as an array — `["hyperhive" "nixpkgs"]` — which - // walk_meta_inputs skips because we can't `nix flake update` - // a follows alias). The remaining top-level inputs are - // `hyperhive` (string) and `agent-z` (string). Without the - // hyperhive-first recursion sort, the BTreeMap alphabetical - // order descends into `agent-z` first and claims - // `nixpkgs` at `agent-z/nixpkgs`. - let raw = r#"{ - "root": "root", - "version": 7, - "nodes": { - "root": { - "inputs": { - "hyperhive": "hyperhive", - "nixpkgs": ["hyperhive", "nixpkgs"], - "agent-z": "agent-z" - } - }, - "hyperhive": { - "inputs": { "nixpkgs": "nixpkgs" }, - "locked": {"rev": "hhrev", "lastModified": 1}, - "original": {"url": "git+file:///tmp/hyperhive"} - }, - "agent-z": { - "inputs": { "nixpkgs": "nixpkgs" }, - "locked": {"rev": "azrev", "lastModified": 2}, - "original": {"url": "git+file:///tmp/agent-z"} - }, - "nixpkgs": { - "locked": {"rev": "npkrev", "lastModified": 3}, - "original": {"url": "github:NixOS/nixpkgs/nixos-26.05"} - } - } - }"#; - let json: serde_json::Value = serde_json::from_str(raw).unwrap(); - let nodes = json.get("nodes").unwrap().as_object().unwrap(); - let root_name = json.get("root").unwrap().as_str().unwrap(); - let mut visited = std::collections::HashSet::new(); - visited.insert(root_name.to_owned()); - let mut out = Vec::new(); - walk_meta_inputs(nodes, root_name, "", &mut visited, &mut out); - - let nixpkgs = out - .iter() - .find(|v| v.rev == "npkrev") - .expect("nixpkgs node should be emitted exactly once"); - assert_eq!( - nixpkgs.name, "hyperhive/nixpkgs", - "nixpkgs should be claimed under hyperhive, not under agent-z. \ - got: {:?}", - nixpkgs.name - ); - // And the agent-z path should NOT also carry a nixpkgs entry — - // the spanning-tree visited set guarantees it's claimed once. - assert!( - !out.iter().any(|v| v.name == "agent-z/nixpkgs"), - "agent-z/nixpkgs should not be emitted (already claimed under hyperhive)" - ); - } - - #[test] - fn validate_agent_name_accepts_canonical_shapes() { - assert!(validate_agent_name("damocles").is_none()); - assert!(validate_agent_name("hm1nd").is_none()); - assert!(validate_agent_name("agent-with-dashes").is_none()); - assert!(validate_agent_name("snake_case").is_none()); - assert!(validate_agent_name("mixed_2-3").is_none()); - let max = "a".repeat(63); - assert!( - validate_agent_name(&max).is_none(), - "63-char name should pass" - ); - } - - // The two-axis guard (`guard_agent_name`) wires `validate_agent_name` - // + an async coordinator lookup. The lookup needs a populated - // `Coordinator`, which needs sqlite + tokio runtime; rather than - // build that scaffolding for an integration-flavoured test we cover - // the format axis here (the existence axis is enforced by the - // shared `containers_snapshot` API, tested in `coordinator.rs`'s - // own suite). 9 cases below cover the boundary-length case and - // other expected rejects to make the contract explicit. - #[test] - fn validate_agent_name_rejects_bad_input() { - assert!(validate_agent_name("").is_some()); - let too_long = "a".repeat(64); - assert!(validate_agent_name(&too_long).is_some()); - // Path-traversal attempts. - assert!(validate_agent_name("../etc/passwd").is_some()); - assert!(validate_agent_name("alice/bob").is_some()); - // Uppercase rejected — canonical lowercase convention. - assert!(validate_agent_name("Alice").is_some()); - // No spaces, dots, special chars. - assert!(validate_agent_name("alice bob").is_some()); - assert!(validate_agent_name("alice.bob").is_some()); - assert!(validate_agent_name("alice;DROP TABLE messages").is_some()); - // Non-ASCII (incl. unicode homoglyphs of ASCII dash). - assert!(validate_agent_name("damóclès").is_some()); - assert!(validate_agent_name("alice\u{2013}bob").is_some()); // en-dash - } -} - -/// Snapshot the current tombstone list and emit a -/// `TombstonesChanged` event. Call after any mutation that could -/// add or remove a tombstone (`actions::destroy`, -/// `post_purge_tombstone`, spawn finalisation). Cheap — the list -/// is tiny. -pub(crate) async fn emit_tombstones_snapshot(coord: &Arc) { - let containers = coord.containers_snapshot().await; - let transient_snapshot = coord.transient_snapshot(); - let tombstones = build_tombstone_views(coord, &containers, &transient_snapshot); - coord.emit_dashboard_event(crate::dashboard_events::DashboardEvent::TombstonesChanged { - seq: coord.next_seq(), - tombstones, - }); -} - -/// Snapshot meta/flake.lock's root inputs + emit -/// `MetaInputsChanged`. Call after any mutation that bumps a lock -/// (`run_meta_update`, `auto_update::rebuild_agent`). -pub(crate) fn emit_meta_inputs_snapshot(coord: &Coordinator) { - let inputs = read_meta_inputs(); - coord.emit_dashboard_event(crate::dashboard_events::DashboardEvent::MetaInputsChanged { - seq: coord.next_seq(), - inputs, - }); -} - -/// Unread operator-directed messages for the dashboard's Y3R C4LL inbox. -/// Returns messages addressed to `"operator"` that haven't been -/// acked yet (the operator clears them via the existing -/// `POST /api/agent/operator/mark-all-read`). Newest-first; path-shaped -/// tokens are validated so the client renders file links like the -/// terminal does. Shape: `{ "messages": [{ id, from, body, at, -/// in_reply_to, file_refs }] }`. -async fn api_operator_inbox(State(state): State) -> Response { - const INBOX_LIMIT: u64 = 100; - match state - .coord - .broker - .unread_for_recipient("operator", INBOX_LIMIT) - { - Ok(messages) => { - let items: Vec = messages - .into_iter() - .filter_map(|m| { - let crate::broker::MessageEvent::Sent { - id, - from, - body, - at, - in_reply_to, - .. - } = m - else { - return None; - }; - let file_refs = scan_validated_paths(&body); - Some(serde_json::json!({ - "id": id, - "from": from, - "body": body, - "at": hive_sh4re::wire_time::from_secs(at), - "in_reply_to": in_reply_to, - "file_refs": file_refs, - })) - }) - .collect(); - axum::Json(serde_json::json!({ "messages": items })).into_response() - } - Err(e) => error_response(&format!("operator-inbox failed: {e:#}")), - } -} - -#[derive(Deserialize)] -struct StatsHiveQuery { - window: Option, -} - -/// Hive-wide turn-stats rollup for the dashboard swarm-stats view. -/// Aggregates every agent's `hyperhive-turn-stats.sqlite` read-only -/// (skips missing/unreadable ones). Window defaults to `24h`. -async fn api_stats_hive( - State(state): State, - axum::extract::Query(q): axum::extract::Query, -) -> Response { - let window = crate::hive_stats::Window::parse(q.window.as_deref().unwrap_or("24h")); - axum::Json(crate::hive_stats::hive_snapshot( - window, - &state.coord.model_prices, - )) - .into_response() -} - -/// Live per-agent-container CPU + memory load from cgroup v2. Samples -/// CPU over a short interval (~200 ms), so this call briefly awaits. -async fn api_container_resources() -> Response { - axum::Json(crate::container_stats::gather().await).into_response() -} - -/// `GET /api/audit-log` — most-recent agent-initiated privileged-action -/// audit entries, newest first (server-clamped to 500). Backs the -/// operator dashboard's audit view. Returns -/// `{ "entries": [AuditEntry…], "total": N }` so the UI can show -/// "latest 500 of N" rather than silently capping. `ts_unix` is in -/// **seconds**. -async fn api_audit_log(State(state): State) -> Response { - const LIMIT: usize = 500; - let entries = match state.coord.audit_log.list_recent(LIMIT) { - Ok(rows) => rows, - Err(e) => return error_response(&format!("audit-log: {e:#}")), - }; - let total = match state.coord.audit_log.count_total() { - Ok(n) => n, - Err(e) => return error_response(&format!("audit-log count: {e:#}")), - }; - axum::Json(serde_json::json!({ "entries": entries, "total": total })).into_response() -} - /// Validate that a path-param agent name conforms to the hyperhive /// naming whitelist: 1-63 chars of `[a-z0-9_-]`. Rejects empty, /// uppercase, slashes, dots, and any non-ASCII (incl. unicode @@ -1485,223 +324,6 @@ async fn guard_agent_name(state: &AppState, name: &str) -> Option { None } -/// Operator-driven "clear this agent's inbox" — backs the side-panel -/// "mark all read" button. Marks every message addressed to the -/// agent as acked (backfilling `delivered_at` for any still-pending -/// rows so vacuum can collect them). Returns `{ "marked": N }` so the -/// frontend can show "cleared N messages" feedback without an extra -/// fetch. -async fn post_mark_all_read( - State(state): State, - AxumPath(name): AxumPath, -) -> Response { - if let Some(reason) = validate_agent_name(&name) { - return (StatusCode::BAD_REQUEST, format!("bad agent name: {reason}")).into_response(); - } - match state.coord.broker.mark_all_read(&name) { - Ok(n) => { - tracing::info!(%name, marked = n, "operator marked all messages read"); - axum::Json(serde_json::json!({ "marked": n })).into_response() - } - Err(e) => error_response(&format!("mark-all-read {name} failed: {e:#}")), - } -} - -async fn post_purge_tombstone( - State(state): State, - AxumPath(name): AxumPath, -) -> Response { - // Format guard FIRST so a name like `..` can't traverse into the - // parent of `/var/lib/hyperhive/agents/{name}` and have - // `remove_dir_all` wipe `/var/lib/hyperhive/` itself. Existing - // manager + live-container checks below don't catch `..` — only - // the whitelist does. Existence check via - // `containers_snapshot()` is deliberately NOT used here: - // tombstoned agents are gone from the snapshot by design; that's - // the whole point of this endpoint. - if let Some(reason) = validate_agent_name(&name) { - return (StatusCode::BAD_REQUEST, format!("bad agent name: {reason}")).into_response(); - } - // Sanity: refuse to purge if a live container still exists with this - // name. The dashboard already filters tombstones to non-live names, - // but the operator could send a stale POST. - let live = lifecycle::list().await.unwrap_or_default(); - if live - .iter() - .any(|c| c == &format!("{}{name}", lifecycle::AGENT_PREFIX) || c == &name) - { - return error_response(&format!( - "refusing to purge {name}: container still exists — use DESTR0Y first" - )); - } - let mut errors = Vec::new(); - for dir in [ - Coordinator::agent_state_root(&name), - Coordinator::agent_applied_dir(&name), - ] { - if dir.exists() - && let Err(e) = std::fs::remove_dir_all(&dir) - { - errors.push(format!("{}: {e}", dir.display())); - } - } - let _ = state - .coord - .approvals - .fail_pending_for_agent(&name, "agent state purged"); - if errors.is_empty() { - tracing::info!(%name, "tombstone purged"); - // Fire the post-purge tombstones snapshot so dashboards - // drop the row live; matching form carries - // `data-no-refresh`. - emit_tombstones_snapshot(&state.coord).await; - (StatusCode::OK, "ok").into_response() - } else { - error_response(&format!("purge {name} partial: {}", errors.join(", "))) - } -} - -/// Operator-side compose form on the dashboard terminal. Drops a -/// message into the broker as `{from: "operator", to, body}`. Same -/// shape that per-agent web UIs use via `OperatorMsg`, but here the -/// operator picks the recipient explicitly with `@name`. No -/// validation that `to` resolves to a known agent — broker accepts -/// arbitrary recipients (and the agent's inbox grows whether or not -/// they exist, which is fine for spawn-then-greet flows). -#[derive(Deserialize)] -struct OpSendForm { - to: String, - body: String, -} - -/// Form for `POST /meta-update`. Inputs ride in as a comma-separated -/// list under the `inputs` field — the JS submitter joins the -/// checked boxes since axum's `Form` extractor doesn't natively -/// decode repeated keys without a helper. -#[derive(Deserialize)] -struct MetaUpdateForm { - inputs: String, -} - -/// Bulk-update selected meta flake inputs, then rebuild the affected -/// agents in the background. Idempotent w.r.t. selection — choosing -/// an input that's already at the latest sha is a no-op (no commit, -/// no rebuild ripple). Returns immediately after queueing the work; -/// dashboard polls for progress via container `pending` spinners + -/// the meta-inputs row sha update. -async fn post_meta_update( - State(state): State, - Form(form): Form, -) -> Response { - let inputs: Vec = form - .inputs - .split(',') - .map(|s| s.trim().to_owned()) - .filter(|s| !s.is_empty()) - .collect(); - if inputs.is_empty() { - return error_response("meta-update: no inputs selected"); - } - let inputs_label = inputs.join(", "); - // Cascade rebuild children fan out from the MetaLock node when the - // lock bump lands — appended by the scheduler so they build against - // the post-bump lock, and a failed bump simply fans out nothing. - crate::job_queue::submit::meta_update( - &state.coord, - inputs, - crate::job_queue::Source::Manual, - format!("meta-update via dashboard ({inputs_label})"), - ); - (StatusCode::OK, "ok").into_response() -} - -async fn post_op_send(State(state): State, Form(form): Form) -> Response { - let to = form.to.trim().to_owned(); - let body = form.body.trim().to_owned(); - if to.is_empty() { - return error_response("op-send: `to` required"); - } - if body.is_empty() { - return error_response("op-send: `body` required"); - } - if to == "*" { - let errors = state - .coord - .broadcast_send(hive_sh4re::OPERATOR_RECIPIENT, &body); - if !errors.is_empty() { - return error_response(&format!( - "op-send broadcast partial fail: {}", - errors.join("; ") - )); - } - } else if let Err(e) = state.coord.broker.send(&hive_sh4re::Message { - from: hive_sh4re::OPERATOR_RECIPIENT.to_owned(), - to: to.clone(), - body, - in_reply_to: None, - }) { - return error_response(&format!("op-send to {to} failed: {e:#}")); - } - // 200 instead of 303 → the client doesn't refetch /api/state. The - // broker `send` already emitted a `MessageEvent` which the - // dashboard channel forwarder mirrors as `DashboardEvent::Sent`, - // and the page's terminal + inbox derive from that stream — so the - // operator's send shows up the same way an agent's send does, with - // no full-state refresh in between. - (axum::http::StatusCode::OK, "ok").into_response() -} - -async fn post_request_spawn( - State(state): State, - Form(form): Form, -) -> Response { - let name = form.name.trim().to_owned(); - if name.is_empty() { - return error_response("spawn: `name` required"); - } - match state.coord.approvals.submit_kind( - &name, - hive_sh4re::ApprovalKind::Spawn, - "", - None, - "operator", - ) { - Ok(id) => { - tracing::info!(%id, %name, "operator: spawn approval queued via dashboard"); - // Phase 5b: notify the dashboard event channel so live - // subscribers can append the row without a snapshot - // refetch. Spawn approvals carry no diff/sha. - state - .coord - .emit_approval_added(crate::coordinator::ApprovalAdded { - id, - agent: &name, - approval_kind: "spawn", - sha_short: None, - diff: None, - description: None, - pr_number: None, - }); - (StatusCode::OK, "ok").into_response() - } - Err(e) => error_response(&format!("request-spawn {name} failed: {e:#}")), - } -} - -fn transient_label(k: crate::coordinator::TransientKind) -> &'static str { - use crate::coordinator::TransientKind::{ - Destroying, Rebuilding, Restarting, Spawning, Starting, Stopping, - }; - match k { - Spawning => "spawning", - Starting => "starting", - Stopping => "stopping", - Restarting => "restarting", - Rebuilding => "rebuilding", - Destroying => "destroying", - } -} - /// Convert either a logical name or a container name back to the logical /// name. Sub-agents are `h-foo` → `foo`; manager stays `root`. fn strip_container_prefix(name: &str) -> String { @@ -1730,3 +352,67 @@ fn error_problem(message: &str) -> problem_details::ProblemDetails { fn error_response(message: &str) -> Response { error_problem(message).into_response() } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn problem_details_carry_rfc9457_status_and_detail() { + // Contract the frontend depends on: the problem_details crate + // serialises the RFC 9457 members we rely on — `status` (numeric) + // and `detail` (the caller message; the FE reads `.detail`). + let pd = problem_details::ProblemDetails::from_status_code(StatusCode::BAD_REQUEST) + .with_detail("bad input"); + let v = serde_json::to_value(&pd).expect("problem details serialise"); + assert_eq!(v["status"], 400); + assert_eq!(v["detail"], "bad input"); + // The 500 wrapper path carries the internal-error status. + let five = + problem_details::ProblemDetails::from_status_code(StatusCode::INTERNAL_SERVER_ERROR) + .with_detail("boom"); + let fv = serde_json::to_value(&five).expect("problem details serialise"); + assert_eq!(fv["status"], 500); + } + + #[test] + fn validate_agent_name_accepts_canonical_shapes() { + assert!(validate_agent_name("damocles").is_none()); + assert!(validate_agent_name("hm1nd").is_none()); + assert!(validate_agent_name("agent-with-dashes").is_none()); + assert!(validate_agent_name("snake_case").is_none()); + assert!(validate_agent_name("mixed_2-3").is_none()); + let max = "a".repeat(63); + assert!( + validate_agent_name(&max).is_none(), + "63-char name should pass" + ); + } + + // The two-axis guard (`guard_agent_name`) wires `validate_agent_name` + // + an async coordinator lookup. The lookup needs a populated + // `Coordinator`, which needs sqlite + tokio runtime; rather than + // build that scaffolding for an integration-flavoured test we cover + // the format axis here (the existence axis is enforced by the + // shared `containers_snapshot` API, tested in `coordinator.rs`'s + // own suite). 9 cases below cover the boundary-length case and + // other expected rejects to make the contract explicit. + #[test] + fn validate_agent_name_rejects_bad_input() { + assert!(validate_agent_name("").is_some()); + let too_long = "a".repeat(64); + assert!(validate_agent_name(&too_long).is_some()); + // Path-traversal attempts. + assert!(validate_agent_name("../etc/passwd").is_some()); + assert!(validate_agent_name("alice/bob").is_some()); + // Uppercase rejected — canonical lowercase convention. + assert!(validate_agent_name("Alice").is_some()); + // No spaces, dots, special chars. + assert!(validate_agent_name("alice bob").is_some()); + assert!(validate_agent_name("alice.bob").is_some()); + assert!(validate_agent_name("alice;DROP TABLE messages").is_some()); + // Non-ASCII (incl. unicode homoglyphs of ASCII dash). + assert!(validate_agent_name("damóclès").is_some()); + assert!(validate_agent_name("alice\u{2013}bob").is_some()); // en-dash + } +} diff --git a/hive-c0re/src/dashboard/state_snapshot.rs b/hive-c0re/src/dashboard/state_snapshot.rs new file mode 100644 index 00000000..aac10c58 --- /dev/null +++ b/hive-c0re/src/dashboard/state_snapshot.rs @@ -0,0 +1,757 @@ +//! `/api/state` cold-load snapshot plus the dashboard's live read side: +//! the `StateSnapshot` shape and its view builders, the +//! `/api/dashboard/stream` SSE channel, and the `/api/dashboard/history` +//! backfill. SPA shape + SSE channels: docs/web-ui/shape.md. + +use std::convert::Infallible; + +use axum::{ + extract::State, + http::HeaderMap, + response::{ + IntoResponse, Response, + sse::{Event, KeepAlive, Sse}, + }, +}; +use chrono::{DateTime, Utc}; +use hive_sh4re::Approval; +use serde::{Deserialize, Serialize}; +use tokio_stream::wrappers::BroadcastStream; +use tokio_stream::{Stream, StreamExt}; + +use crate::container_view::ContainerView; + +use super::meta_inputs::{MetaInputView, read_meta_inputs}; +use super::tombstones::{TombstoneView, build_tombstone_views}; +use super::{AppState, approval_diff, approvals, error_response, scan_validated_paths}; + +#[allow(clippy::struct_excessive_bools)] +#[derive(Serialize)] +pub(super) struct StateSnapshot { + /// Broker seq at the moment this snapshot was assembled. Clients + /// dedupe their buffered SSE traffic against this value: any + /// `MessageEvent` with `seq <= snapshot.seq` is already reflected in + /// the snapshot (or pre-dates it); anything with `seq > snapshot.seq` + /// is post-snapshot and should be applied. Set to 0 in the + /// pre-emit case (no events ever fired) — clients treat that as + /// "apply everything you've buffered". + seq: u64, + hostname: String, + any_stale: bool, + containers: Vec, + transients: Vec, + approvals: Vec, + /// Last 30 resolved approvals (approved / denied / failed), newest- + /// first. Drives the "history" tab on the approvals section. + approval_history: Vec, + /// Pending operator-targeted questions (`target IS NULL`). Any + /// agent can `ask` the operator and `ask` returns immediately with + /// the id; on `/answer-question` we mark the row answered and + /// fire `HelperEvent::QuestionAnswered` back into the asker's + /// inbox. Peer-to-peer questions live in the same table but never + /// surface here (see `OperatorQuestions::pending`). + questions: Vec, + /// Last 20 answered questions, newest-first. + question_history: Vec, + /// State dirs (config history + claude creds + /state/ notes) that + /// survive after a destroy-without-purge. The operator can re-spawn + /// with the same name to resume, or PURG3 to wipe them. + tombstones: Vec, + /// Sub-agents whose FNV-1a hashed web UI port collides with at + /// least one other agent. Operator resolves by renaming. The + /// dashboard renders a banner at the top listing each cluster. + port_conflicts: Vec, + /// Inputs in `meta/flake.lock` the operator can selectively + /// `nix flake update`. Hyperhive first, then `agent-` rows. + meta_inputs: Vec, + /// True while a dashboard-triggered `meta-update` (flake lock bump + + /// agent rebuild ripple) is running in the background. Lets a + /// client that cold-loads mid-update render the META INPUTS panel's + /// disabled "updating…" state; live transitions arrive via the + /// `MetaUpdateRunning` event. + meta_update_running: bool, + /// Current state of the global job queue — pending + running DAGs + /// (rebuild / meta-update / spawn / power ops) with their per-node + /// breakdowns, plus the most recent few terminal DAGs the queue + /// retains for history. Live transitions arrive via the + /// `RebuildQueueChanged` event. See `job_queue/`. Field name kept + /// from the old flat queue for wire compatibility. + rebuild_queue: Vec, + /// Whether the hive-forge container is up. When true the dashboard + /// links each container's config + each approval's commit into the + /// forge's `agent-configs` repos. + forge_present: bool, + /// Whether the matrix GUI is reachable at `/matrix/`. Sourced from + /// `HIVE_MATRIX_GUI_ENABLED` env var (set by the c0re NixOS module + /// when `services.hyperhive.matrix.gui.enable` is on). The gateway + /// (hive-gateway.nix) does the actual `/matrix/` static serving; + /// this flag is just an availability signal for iris's dashboard + /// chrome so the `M4TR1X →` tab doesn't flash when the GUI is off. + matrix_gui_enabled: bool, + /// Whether `hive-gateway` is in front of this dashboard. Sourced + /// from the `HIVE_GATEWAY_ENABLED` env var, which the c0re NixOS + /// module now always sets (the gateway runs unconditionally + /// alongside hyperhive), so this is effectively always true: the + /// dashboard frontend builds same-origin `/agent//` links to + /// the per-agent web UI (the gateway routes them via the + /// runtime-generated `agents.conf` include file — see + /// `gateway_nginx.rs`). The `false` branch (direct + /// `http://:/` TCP links) is retained as a defensive + /// fallback for the env being unset. See `docs/gateway.md::Vhost map`. + gateway_enabled: bool, + /// Public URL of the forge vhost served by hive-gateway (e.g. + /// `"https://forge.pr1ma.darkest.space"`). Sourced from the + /// `HIVE_FORGE_PUBLIC_URL` env var, which the c0re NixOS module + /// sets when `forge.behindGateway = true`. `None` when absent — + /// the frontend falls back to `http://:3000`. + forge_public_url: Option, + /// Human name of this single-host hive instance (e.g. `"pr1ma"`). + /// Sourced from `HYPERHIVE_HIVE_NAME` env var, set by the c0re + /// NixOS module from `services.hyperhive.hiveName`. `None` when + /// the option is unset — chrome falls back to `hostname`. + hive_name: Option, + /// Human name of the wider swarm this hive belongs to (e.g. + /// `"constellat1on"`). Sourced from `HYPERHIVE_SWARM_NAME` env + /// var, set from `services.hyperhive.swarmName`. `None` when + /// unset — chrome omits the swarm segment of the breadcrumb. + swarm_name: Option, + /// Peer hives in the same swarm. Parsed from `HYPERHIVE_PEERS` + /// (JSON array of `{domain,cert_fingerprint}` objects, emitted by + /// the c0re NixOS module from `services.hyperhive.swarm.peers`). + /// Empty on single-hive deploys. Feeds the P33RS dashboard tab. + peer_hives: Vec, + /// Server-level warnings for the dashboard's top-of-page banner + /// (currently host disk-pressure; more producers can be added + /// backend-side). Empty when all clear. Built by + /// `host_stats::server_warnings`; the frontend renders this list + /// generically, so new warning kinds need no frontend change. + server_warnings: Vec, +} + +/// One peer hive for the P33RS dashboard tab. Derived from +/// `HYPERHIVE_PEERS` env; `url` is the peer's HTTPS dashboard root. +/// `cert_fingerprint` is `Some("sha256:")` when the peer uses a +/// self-signed cert and the operator pinned its fingerprint in +/// `services.hyperhive.swarm.peers`. +#[derive(Serialize)] +struct PeerHiveView { + name: String, + url: String, + cert_fingerprint: Option, +} + +/// `OpQuestion` + computed `question_refs` / `answer_refs`. Built +/// from the snapshot read; the live channel attaches the same +/// fields directly on `QuestionAdded` / `QuestionResolved`. +#[derive(Serialize)] +struct QuestionView { + #[serde(flatten)] + inner: crate::operator_questions::OpQuestion, + #[serde(skip_serializing_if = "Vec::is_empty")] + question_refs: Vec, + #[serde(skip_serializing_if = "Vec::is_empty")] + answer_refs: Vec, +} + +impl QuestionView { + fn from_question(q: crate::operator_questions::OpQuestion) -> Self { + let question_refs = scan_validated_paths(&q.question); + let answer_refs = q + .answer + .as_deref() + .map(scan_validated_paths) + .unwrap_or_default(); + Self { + inner: q, + question_refs, + answer_refs, + } + } +} + +#[derive(Serialize)] +struct PortConflict { + port: u16, + /// All agent names sharing this port (sorted, ≥2 entries). + agents: Vec, +} + +#[derive(Serialize)] +struct TransientView { + name: String, + kind: &'static str, + secs: u64, +} + +#[derive(Serialize)] +struct ApprovalHistoryView { + id: i64, + agent: String, + kind: &'static str, + /// First 12 chars of the canonical sha (preferred) or + /// manager-supplied ref. None for resolved spawn approvals. + sha_short: Option, + /// `approved` / `denied` / `failed`. + status: &'static str, + /// RFC 3339 UTC. Renders as a relative time on the dashboard. + resolved_at: DateTime, + /// Operator-supplied deny reason (for `denied`) or build error + /// (for `failed`). None on `approved`. + #[serde(skip_serializing_if = "Option::is_none")] + note: Option, +} + +#[derive(Serialize)] +struct ApprovalView { + id: i64, + agent: String, + kind: &'static str, + /// First 12 chars of the `commit_ref`, for `ApplyCommit` only. + sha_short: Option, + /// Raw unified diff text, for `ApplyCommit` only. The client splits + /// on `\n` and per-line classifies (`+` / `-` / `@@` / `--- ` / `+++ ` + /// → diff-add / diff-del / diff-hunk / diff-file). Shipping raw + /// instead of pre-rendered HTML saves bytes on the wire (no + /// per-line `` markup) and removes the only HTML-escape + /// surface from the snapshot. + diff: Option, + /// Manager-supplied description shown on the approval card. + #[serde(skip_serializing_if = "Option::is_none")] + description: Option, + /// Forge PR number, for `MergeConfigPr` only. Lets the frontend + /// build a "review PR on forge" link + /// (`{forgeBase}/agent-configs/{agent}/pulls/{pr_number}`) the same + /// way it builds the `apply_commit` "commit on forge" link from the + /// sha. `None` for every other kind. + #[serde(skip_serializing_if = "Option::is_none")] + pr_number: Option, + /// Raw `commit_ref` payload for `UpdateMetaInputs` (JSON-encoded + /// `Vec` of input names; `"[]"` = all inputs) and + /// `SchedulePrompt` (JSON-encoded `SchedulePromptPayload`). The + /// frontend parses this to render a human-readable card body. + /// `None` for every other kind. + #[serde(skip_serializing_if = "Option::is_none")] + commit_ref: Option, + /// RFC 3339 UTC time the approval was queued. Rendered as a + /// relative time on the card so the operator can spot a stale + /// request. + requested_at: DateTime, +} + +/// Replace silent `.unwrap_or_default()` on the data sources behind +/// `/api/state` so that whichever query degrades surfaces in journald +/// instead of leaving the operator staring at an empty list. The +/// dashboard still degrades to a sensible default value; the warn +/// is just the diagnostic breadcrumb the old code swallowed. +fn log_default(what: &str, result: std::result::Result) -> T +where + T: Default, + E: std::fmt::Debug, +{ + match result { + Ok(v) => v, + Err(e) => { + tracing::warn!(target: "api_state", source = %what, error = ?e, "snapshot source failed; using default"); + T::default() + } + } +} + +/// Window over which container crashes count toward the `agents_crashing` +/// banner warning. Wide enough that a crash-looping container (restarted +/// by `Restart=on-failure` every few seconds) keeps the warning lit +/// between flaps, short enough that a single recovered crash clears within +/// minutes. +const CRASH_WARNING_WINDOW: std::time::Duration = std::time::Duration::from_mins(10); + +pub(super) async fn api_state( + headers: HeaderMap, + State(state): State, +) -> axum::Json { + let host = headers + .get("host") + .and_then(|h| h.to_str().ok()) + .unwrap_or("localhost"); + let hostname = host.split(':').next().unwrap_or(host).to_owned(); + + // Capture the unified dashboard-channel seq *before* any read so the + // dedupe contract is "events with seq > snapshot.seq are + // post-snapshot, never missed." An event landing during snapshot + // construction may be doubly applied (snapshot caught the write + + // client also applies the SSE frame) — that's a renderer's problem + // to make idempotent, not ours to avoid here. + let seq = state.coord.current_seq(); + + // Refresh the coordinator's cached container snapshot before + // reading. Cold-load clients then see whatever the latest rescan + // produced; live clients converge via the matching + // `ContainerStateChanged` / `ContainerRemoved` events the rescan + // emits. + // + // Bound the rescan: it shells out (`nixos-container list` etc.), so a + // saturated/wedged build backend — e.g. hive-c0re mid-startup-sweep + // hammering slow `nixos-container update` subprocesses — can stall it + // long enough that `/api/state` hangs for the whole request (the + // ~minute-long /state reported in the field). On timeout we skip the + // fresh rescan and serve the last cached snapshot instead; live + // clients still converge via the SSE events a later successful rescan + // emits, and the next /state call retries the refresh. Introspection + // stays responsive regardless of the build backend's health. + if tokio::time::timeout( + std::time::Duration::from_secs(3), + state.coord.rescan_containers_and_emit(), + ) + .await + .is_err() + { + tracing::warn!( + "api_state: container rescan exceeded 3s (build backend likely saturated); \ + serving last cached snapshot" + ); + } + let containers = state.coord.containers_snapshot().await; + let any_stale = containers.iter().any(|c| c.needs_update); + let transient_snapshot = state.coord.transient_snapshot(); + let pending_approvals = approvals::gc_orphans( + &state.coord, + log_default("approvals.pending", state.coord.approvals.pending()), + ); + let transients = build_transient_views(&containers, &transient_snapshot); + let approvals = build_approval_views(pending_approvals).await; + let approval_history = log_default( + "approvals.recent_resolved", + state.coord.approvals.recent_resolved(30), + ) + .into_iter() + .map(history_view) + .collect(); + let tombstones = build_tombstone_views(&state.coord, &containers, &transient_snapshot); + let port_conflicts = build_port_conflicts(&containers); + + // Both operator-targeted and peer threads surface on the dashboard + // (the client filters by target). Each row is wrapped in QuestionView + // so the snapshot carries the same file_refs the live event variants + // attach. + let questions: Vec = + log_default("questions.pending_all", state.coord.questions.pending_all()) + .into_iter() + .map(QuestionView::from_question) + .collect(); + let question_history: Vec = log_default( + "questions.recent_answered_all", + state.coord.questions.recent_answered_all(20), + ) + .into_iter() + .map(QuestionView::from_question) + .collect(); + + // Banner warnings: host probes (disk) + agent-state (pending logins, + // crashing agents). Built before the response struct because the + // agent-state producer borrows `containers`, which moves in below. + let server_warnings = { + let mut w = crate::host_stats::server_warnings(); + w.extend(crate::host_stats::agent_state_warnings( + &containers, + &state.coord.recent_crash_counts(CRASH_WARNING_WINDOW), + )); + w + }; + + axum::Json(StateSnapshot { + seq, + hostname, + any_stale, + containers, + transients, + approvals, + approval_history, + meta_inputs: read_meta_inputs(), + meta_update_running: state.coord.meta_update_in_progress(), + questions, + question_history, + tombstones, + port_conflicts, + rebuild_queue: state.coord.job_queue.snapshot(), + forge_present: crate::forge::is_present().await, + matrix_gui_enabled: std::env::var_os("HIVE_MATRIX_GUI_ENABLED").is_some_and(|v| { + // Accept any truthy string ("1", "true", "yes") since the + // env var is set by NixOS module wiring with the literal + // "1"; defensive parse so manual overrides also work. + let s = v.to_string_lossy().to_ascii_lowercase(); + matches!(s.as_str(), "1" | "true" | "yes") + }), + gateway_enabled: std::env::var_os("HIVE_GATEWAY_ENABLED").is_some_and(|v| { + // Same truthy-string parse as `matrix_gui_enabled`; the + // env var is set by the c0re NixOS module to the literal + // "1" — the gateway always runs alongside hyperhive. + let s = v.to_string_lossy().to_ascii_lowercase(); + matches!(s.as_str(), "1" | "true" | "yes") + }), + forge_public_url: std::env::var("HIVE_FORGE_PUBLIC_URL") + .ok() + .filter(|s| !s.is_empty()), + hive_name: std::env::var("HYPERHIVE_HIVE_NAME") + .ok() + .filter(|s| !s.is_empty()), + swarm_name: std::env::var("HYPERHIVE_SWARM_NAME") + .ok() + .filter(|s| !s.is_empty()), + peer_hives: parse_peer_hives(), + server_warnings, + }) +} + +/// Parse `HYPERHIVE_PEERS` env var into dashboard-ready `PeerHiveView` +/// entries. The env var is a JSON array of `{domain, cert_fingerprint}` +/// objects emitted by the c0re NixOS module from +/// `services.hyperhive.swarm.peers`. Each entry becomes +/// `{ name: domain, url: "https://domain/" }` for the P33RS tab. +/// Returns empty vec when unset (single-hive deploy). +fn parse_peer_hives() -> Vec { + #[derive(serde::Deserialize)] + struct Raw { + domain: String, + cert_fingerprint: Option, + } + let Ok(json) = std::env::var("HYPERHIVE_PEERS") else { + return Vec::new(); + }; + let Ok(raw): Result, _> = serde_json::from_str(&json) else { + tracing::warn!("HYPERHIVE_PEERS is not valid JSON; ignoring"); + return Vec::new(); + }; + raw.into_iter() + .map(|r| { + let cert_fingerprint = r.cert_fingerprint.and_then(|fp| { + if validate_cert_fingerprint(&fp) { + Some(fp) + } else { + tracing::warn!( + domain = %r.domain, + fingerprint = %fp, + "HYPERHIVE_PEERS: invalid cert_fingerprint format \ + (expected `sha256:<64 hex chars>`); ignoring fingerprint" + ); + None + } + }); + PeerHiveView { + name: r.domain.clone(), + url: format!("https://{}/", r.domain), + cert_fingerprint, + } + }) + .collect() +} + +/// Validate a TLS certificate fingerprint string from `HYPERHIVE_PEERS`. +/// Accepts `sha256:<64 hex chars>` (upper or lower case). +fn validate_cert_fingerprint(fp: &str) -> bool { + let Some(hex) = fp.strip_prefix("sha256:") else { + return false; + }; + hex.len() == 64 && hex.chars().all(|c| c.is_ascii_hexdigit()) +} + +/// Group live containers by their assigned web UI port; clusters with +/// more than one member are port-hash collisions the operator needs +/// to resolve by renaming. Manager (fixed at 8000) and sub-agents +/// (8100..8999) can't collide with each other — collisions are +/// strictly between sub-agents. +fn build_port_conflicts(containers: &[ContainerView]) -> Vec { + let mut by_port: std::collections::BTreeMap> = + std::collections::BTreeMap::new(); + for c in containers { + by_port.entry(c.port).or_default().push(c.name.clone()); + } + by_port + .into_iter() + .filter(|(_, agents)| agents.len() > 1) + .map(|(port, mut agents)| { + agents.sort(); + PortConflict { port, agents } + }) + .collect() +} + +/// Transient state for agents whose container does NOT yet exist +/// (`Spawning`). Lifecycle ops on existing containers surface as +/// `ContainerView.pending` inline; this list only catches pre-creation. +fn build_transient_views( + containers: &[ContainerView], + transient_snapshot: &std::collections::HashMap, +) -> Vec { + transient_snapshot + .iter() + .filter(|(name, _)| !containers.iter().any(|c| &c.name == *name)) + .map(|(name, st)| TransientView { + name: name.clone(), + kind: transient_label(st.kind), + secs: st.since.elapsed().as_secs(), + }) + .collect() +} + +fn transient_label(k: crate::coordinator::TransientKind) -> &'static str { + use crate::coordinator::TransientKind::{ + Destroying, Rebuilding, Restarting, Spawning, Starting, Stopping, + }; + match k { + Spawning => "spawning", + Starting => "starting", + Stopping => "stopping", + Restarting => "restarting", + Rebuilding => "rebuilding", + Destroying => "destroying", + } +} + +/// Render each pending approval into its dashboard view (short sha + +/// unified diff for `ApplyCommit`, just the name for `Spawn`). +/// Project a resolved sqlite row into the lean shape the dashboard +/// history tab consumes — no `diff_html` (rendering 30 of them +/// per /api/state poll would mean 30 git diffs per refresh). +fn history_view(a: Approval) -> ApprovalHistoryView { + let displayed = a.fetched_sha.as_deref().unwrap_or(&a.commit_ref); + let sha_short = if displayed.is_empty() { + None + } else { + Some(displayed[..displayed.len().min(12)].to_owned()) + }; + let status = match a.status { + hive_sh4re::ApprovalStatus::Approved => "approved", + hive_sh4re::ApprovalStatus::Denied => "denied", + hive_sh4re::ApprovalStatus::Failed => "failed", + hive_sh4re::ApprovalStatus::Cancelled => "cancelled", + // Pending shouldn't appear in recent_resolved, but be defensive. + hive_sh4re::ApprovalStatus::Pending => "pending", + }; + let kind = match a.kind { + hive_sh4re::ApprovalKind::ApplyCommit => "apply_commit", + hive_sh4re::ApprovalKind::Spawn => "spawn", + hive_sh4re::ApprovalKind::InitConfig => "init_config", + hive_sh4re::ApprovalKind::UpdateMetaInputs => "update_meta_inputs", + hive_sh4re::ApprovalKind::SchedulePrompt => "schedule_prompt", + hive_sh4re::ApprovalKind::MergeConfigPr => "merge_config_pr", + }; + ApprovalHistoryView { + id: a.id, + agent: a.agent, + kind, + sha_short, + status, + resolved_at: a.resolved_at.unwrap_or_default(), + note: a.note, + } +} + +async fn build_approval_views(approvals: Vec) -> Vec { + let mut out = Vec::with_capacity(approvals.len()); + for a in approvals { + out.push(match a.kind { + hive_sh4re::ApprovalKind::ApplyCommit => { + // Prefer the canonical fetched sha from applied; + // commit_ref is only the manager's claim and may be + // amended out from under us. + let displayed = a.fetched_sha.as_deref().unwrap_or(&a.commit_ref); + let sha = displayed[..displayed.len().min(12)].to_owned(); + let diff = approval_diff(&a.agent, a.id).await; + ApprovalView { + id: a.id, + agent: a.agent.clone(), + kind: "apply_commit", + sha_short: Some(sha), + diff: Some(diff), + description: a.description, + pr_number: None, + commit_ref: None, + requested_at: a.requested_at, + } + } + hive_sh4re::ApprovalKind::Spawn => ApprovalView { + id: a.id, + agent: a.agent, + kind: "spawn", + sha_short: None, + diff: None, + description: a.description, + pr_number: None, + commit_ref: None, + requested_at: a.requested_at, + }, + hive_sh4re::ApprovalKind::InitConfig => ApprovalView { + id: a.id, + agent: a.agent, + kind: "init_config", + sha_short: None, + diff: None, + description: a.description, + pr_number: None, + commit_ref: None, + requested_at: a.requested_at, + }, + hive_sh4re::ApprovalKind::UpdateMetaInputs => ApprovalView { + id: a.id, + agent: a.agent, + kind: "update_meta_inputs", + sha_short: None, + diff: None, + description: a.description, + pr_number: None, + commit_ref: Some(a.commit_ref), + requested_at: a.requested_at, + }, + hive_sh4re::ApprovalKind::SchedulePrompt => ApprovalView { + id: a.id, + agent: a.agent, + kind: "schedule_prompt", + sha_short: None, + diff: None, + description: a.description, + pr_number: None, + commit_ref: Some(a.commit_ref), + requested_at: a.requested_at, + }, + hive_sh4re::ApprovalKind::MergeConfigPr => { + // commit_ref = PR number; fetched_sha = the reviewed PR + // head. Show the head sha; the forge PR diff surface is + // a later phase of the PR-based config flow — None for now. + let sha = a + .fetched_sha + .as_deref() + .map(|s| s[..s.len().min(12)].to_owned()); + // Surface the PR number so the frontend can link to the + // PR on the forge. commit_ref holds the number as text. + let pr_number = a.commit_ref.parse::().ok(); + ApprovalView { + id: a.id, + agent: a.agent, + kind: "merge_config_pr", + sha_short: sha, + diff: None, + description: a.description, + pr_number, + commit_ref: None, + requested_at: a.requested_at, + } + } + }); + } + out +} + +pub(super) async fn dashboard_history(State(state): State) -> Response { + // Backfill source for the dashboard terminal. Returns up to ~200 + // historical broker messages (no other event kinds are persisted) + // converted to `DashboardEvent::Sent` JSON so the client can replay + // through the same dispatch path as live frames. Wrapped in + // `{ seq, events }`: the seq is the dashboard channel's high-water + // mark at fetch time. Clients use it to dedupe their buffered live + // SSE traffic (drop anything with `seq <= history_seq`) so a frame + // that lands between SSE-subscribe and history-fetch isn't shown + // twice and isn't lost. Historical rows carry `seq = 0`; the + // boundary seq is what closes the dedupe window. + const HISTORY_LIMIT: u64 = 200; + let seq = state.coord.current_seq(); + match state.coord.broker.recent_all(HISTORY_LIMIT) { + Ok(mut messages) => { + messages.reverse(); + let events: Vec = messages + .into_iter() + .map(|m| match m { + crate::broker::MessageEvent::Sent { + id, + from, + to, + body, + at, + in_reply_to, + } => { + let file_refs = scan_validated_paths(&body); + crate::dashboard_events::DashboardEvent::Sent { + seq: 0, + id, + from, + to, + body, + at: hive_sh4re::wire_time::from_secs(at), + in_reply_to, + file_refs, + } + } + crate::broker::MessageEvent::Delivered { + id, + from, + to, + body, + at, + in_reply_to, + } => { + let file_refs = scan_validated_paths(&body); + crate::dashboard_events::DashboardEvent::Delivered { + seq: 0, + id, + from, + to, + body, + at: hive_sh4re::wire_time::from_secs(at), + in_reply_to, + file_refs, + } + } + }) + .collect(); + axum::Json(serde_json::json!({ "seq": seq, "events": events })).into_response() + } + Err(e) => error_response(&format!("dashboard/history failed: {e:#}")), + } +} + +/// `/dashboard/stream` query string. Today's only field is `kinds`: +/// a comma-separated allow-list of event-`kind` strings. +/// Empty / absent ⇒ no filter (current behaviour, all variants +/// forwarded). Set ⇒ only the named kinds reach the subscriber, +/// non-matches are skipped before the JSON serialise cost. +/// +/// Useful for narrow pages (e.g. `flow.js` only cares about `sent` +/// / `delivered` / `container_state_changed` / `container_removed`) +/// that want to drop the dispatch overhead on every unrelated mutation. +#[derive(Deserialize, Default)] +pub(super) struct DashboardStreamQuery { + /// Comma-separated event kinds to forward. Each token is + /// trimmed; unknown kinds are silently ignored on lookup + /// (subscriber sees nothing instead of an error). + kinds: Option, +} + +pub(super) async fn dashboard_stream( + State(state): State, + axum::extract::Query(q): axum::extract::Query, +) -> Sse>> { + let rx = state.coord.dashboard_subscribe(); + // Pre-parse the allow-list once at subscription time, so the + // per-event hot path is just a `HashSet::contains` on a + // `&'static str` — no string churn per frame. + let kind_filter: Option> = q.kinds.and_then(|raw| { + let set: std::collections::HashSet = raw + .split(',') + .map(str::trim) + .filter(|s| !s.is_empty()) + .map(str::to_owned) + .collect(); + if set.is_empty() { None } else { Some(set) } + }); + let stream = BroadcastStream::new(rx).filter_map(move |res| { + // Drop lagged frames. Browsers reconnect; the seq dedupe on + // reconnect skips any frame already reflected in the snapshot. + let event = res.ok()?; + if let Some(filter) = kind_filter.as_ref() + && !filter.contains(event.kind_tag()) + { + return None; + } + let json = serde_json::to_string(&event).ok()?; + Some(Ok(Event::default().data(json))) + }); + Sse::new(stream).keep_alive(KeepAlive::default()) +} diff --git a/hive-c0re/src/dashboard/tombstones.rs b/hive-c0re/src/dashboard/tombstones.rs new file mode 100644 index 00000000..8cd6c408 --- /dev/null +++ b/hive-c0re/src/dashboard/tombstones.rs @@ -0,0 +1,159 @@ +//! Tombstone rows for the dashboard: state dirs surviving a +//! destroy-without-purge. Builds `TombstoneView`s for the snapshot, +//! emits the `TombstonesChanged` event after mutations, and handles +//! `POST /api/purge-tombstone/{name}`. + +use std::path::Path; +use std::sync::Arc; + +use axum::{ + extract::{Path as AxumPath, State}, + http::StatusCode, + response::{IntoResponse, Response}, +}; +use serde::Serialize; + +use crate::container_view::{ContainerView, claude_has_session}; +use crate::coordinator::Coordinator; +use crate::lifecycle; + +use super::{AppState, error_response, validate_agent_name}; + +#[derive(Serialize, Clone, Debug)] +pub struct TombstoneView { + pub name: String, + /// Bytes used by the state dir tree. Cheap-ish to compute; let the + /// operator know how much they're holding onto. + pub state_bytes: u64, + /// Mtime (unix seconds) of the state dir; rough "last seen". + pub last_seen: i64, + pub has_creds: bool, +} + +/// State-dir names that don't appear in the live container list. Each +/// one surfaces in the dashboard as a row with R3V1V3 + PURG3 actions. +pub(super) fn build_tombstone_views( + coord: &Coordinator, + containers: &[ContainerView], + transient_snapshot: &std::collections::HashMap, +) -> Vec { + let _ = coord; // kept_state_names is a free fn but takes &self by future plan + let live: std::collections::HashSet<&str> = containers + .iter() + .map(|c| c.name.as_str()) + .chain(transient_snapshot.keys().map(String::as_str)) + .collect(); + Coordinator::kept_state_names() + .into_iter() + .filter(|name| !live.contains(name.as_str())) + .map(|name| { + let root = Coordinator::agent_state_root(&name); + let state_bytes = dir_size_bytes(&root); + let last_seen = std::fs::metadata(&root) + .and_then(|m| m.modified()) + .ok() + .and_then(|t| t.duration_since(std::time::UNIX_EPOCH).ok()) + .and_then(|d| i64::try_from(d.as_secs()).ok()) + .unwrap_or(0); + let has_creds = claude_has_session(&Coordinator::agent_claude_dir(&name)); + TombstoneView { + name, + state_bytes, + last_seen, + has_creds, + } + }) + .collect() +} + +/// Sum the byte size of every regular file under `root`. Cheap to compute +/// for typical agent state (config repo + claude creds + notes file — +/// usually a few MB); fine to do inline on each /api/state. Returns 0 on +/// any error. +fn dir_size_bytes(root: &Path) -> u64 { + fn walk(p: &Path, acc: &mut u64) { + let Ok(rd) = std::fs::read_dir(p) else { return }; + for entry in rd.flatten() { + let Ok(ft) = entry.file_type() else { continue }; + if ft.is_dir() { + walk(&entry.path(), acc); + } else if ft.is_file() + && let Ok(meta) = entry.metadata() + { + *acc += meta.len(); + } + } + } + let mut total = 0u64; + walk(root, &mut total); + total +} + +/// Snapshot the current tombstone list and emit a +/// `TombstonesChanged` event. Call after any mutation that could +/// add or remove a tombstone (`actions::destroy`, +/// `post_purge_tombstone`, spawn finalisation). Cheap — the list +/// is tiny. +pub(crate) async fn emit_tombstones_snapshot(coord: &Arc) { + let containers = coord.containers_snapshot().await; + let transient_snapshot = coord.transient_snapshot(); + let tombstones = build_tombstone_views(coord, &containers, &transient_snapshot); + coord.emit_dashboard_event(crate::dashboard_events::DashboardEvent::TombstonesChanged { + seq: coord.next_seq(), + tombstones, + }); +} + +pub(super) async fn post_purge_tombstone( + State(state): State, + AxumPath(name): AxumPath, +) -> Response { + // Format guard FIRST so a name like `..` can't traverse into the + // parent of `/var/lib/hyperhive/agents/{name}` and have + // `remove_dir_all` wipe `/var/lib/hyperhive/` itself. Existing + // manager + live-container checks below don't catch `..` — only + // the whitelist does. Existence check via + // `containers_snapshot()` is deliberately NOT used here: + // tombstoned agents are gone from the snapshot by design; that's + // the whole point of this endpoint. + if let Some(reason) = validate_agent_name(&name) { + return (StatusCode::BAD_REQUEST, format!("bad agent name: {reason}")).into_response(); + } + // Sanity: refuse to purge if a live container still exists with this + // name. The dashboard already filters tombstones to non-live names, + // but the operator could send a stale POST. + let live = lifecycle::list().await.unwrap_or_default(); + if live + .iter() + .any(|c| c == &format!("{}{name}", lifecycle::AGENT_PREFIX) || c == &name) + { + return error_response(&format!( + "refusing to purge {name}: container still exists — use DESTR0Y first" + )); + } + let mut errors = Vec::new(); + for dir in [ + Coordinator::agent_state_root(&name), + Coordinator::agent_applied_dir(&name), + ] { + if dir.exists() + && let Err(e) = std::fs::remove_dir_all(&dir) + { + errors.push(format!("{}: {e}", dir.display())); + } + } + let _ = state + .coord + .approvals + .fail_pending_for_agent(&name, "agent state purged"); + if errors.is_empty() { + tracing::info!(%name, "tombstone purged"); + // Fire the post-purge tombstones snapshot so dashboards + // drop the row live; matching form carries + // `data-no-refresh`. + emit_tombstones_snapshot(&state.coord).await; + (StatusCode::OK, "ok").into_response() + } else { + error_response(&format!("purge {name} partial: {}", errors.join(", "))) + } +}