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
This commit is contained in:
müde
parent
3ee87d394c
commit
58e86a3adf
5 changed files with 1522 additions and 1421 deletions
279
hive-c0re/src/dashboard/meta_inputs.rs
Normal file
279
hive-c0re/src/dashboard/meta_inputs.rs
Normal file
|
|
@ -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-<n>`, 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<String>,
|
||||
}
|
||||
|
||||
/// 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-<n>`), 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<MetaInputView> {
|
||||
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::<serde_json::Value>(&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<String, serde_json::Value>,
|
||||
node_name: &str,
|
||||
prefix: &str,
|
||||
visited: &mut std::collections::HashSet<String>,
|
||||
out: &mut Vec<MetaInputView>,
|
||||
) {
|
||||
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<AppState>,
|
||||
Form(form): Form<MetaUpdateForm>,
|
||||
) -> Response {
|
||||
let inputs: Vec<String> = 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)"
|
||||
);
|
||||
}
|
||||
}
|
||||
220
hive-c0re/src/dashboard/misc_api.rs
Normal file
220
hive-c0re/src/dashboard/misc_api.rs
Normal file
|
|
@ -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<AppState>) -> Response {
|
||||
const INBOX_LIMIT: u64 = 100;
|
||||
match state
|
||||
.coord
|
||||
.broker
|
||||
.unread_for_recipient("operator", INBOX_LIMIT)
|
||||
{
|
||||
Ok(messages) => {
|
||||
let items: Vec<serde_json::Value> = 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<String>,
|
||||
}
|
||||
|
||||
/// 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<AppState>,
|
||||
axum::extract::Query(q): axum::extract::Query<StatsHiveQuery>,
|
||||
) -> 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<AppState>) -> 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<AppState>,
|
||||
AxumPath(name): AxumPath<String>,
|
||||
) -> 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<AppState>,
|
||||
Form(form): Form<OpSendForm>,
|
||||
) -> 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<AppState>,
|
||||
Form(form): Form<RequestSpawnForm>,
|
||||
) -> 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:#}")),
|
||||
}
|
||||
}
|
||||
File diff suppressed because it is too large
Load diff
757
hive-c0re/src/dashboard/state_snapshot.rs
Normal file
757
hive-c0re/src/dashboard/state_snapshot.rs
Normal file
|
|
@ -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<ContainerView>,
|
||||
transients: Vec<TransientView>,
|
||||
approvals: Vec<ApprovalView>,
|
||||
/// Last 30 resolved approvals (approved / denied / failed), newest-
|
||||
/// first. Drives the "history" tab on the approvals section.
|
||||
approval_history: Vec<ApprovalHistoryView>,
|
||||
/// 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<QuestionView>,
|
||||
/// Last 20 answered questions, newest-first.
|
||||
question_history: Vec<QuestionView>,
|
||||
/// 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<TombstoneView>,
|
||||
/// 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<PortConflict>,
|
||||
/// Inputs in `meta/flake.lock` the operator can selectively
|
||||
/// `nix flake update`. Hyperhive first, then `agent-<n>` rows.
|
||||
meta_inputs: Vec<MetaInputView>,
|
||||
/// 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<crate::job_queue::DagView>,
|
||||
/// 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/<name>/` 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://<hostname>:<port>/` 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://<hostname>:3000`.
|
||||
forge_public_url: Option<String>,
|
||||
/// 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<String>,
|
||||
/// 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<String>,
|
||||
/// 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<PeerHiveView>,
|
||||
/// 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<crate::host_stats::ServerWarning>,
|
||||
}
|
||||
|
||||
/// 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:<hex64>")` 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<String>,
|
||||
}
|
||||
|
||||
/// `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<String>,
|
||||
#[serde(skip_serializing_if = "Vec::is_empty")]
|
||||
answer_refs: Vec<String>,
|
||||
}
|
||||
|
||||
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<String>,
|
||||
}
|
||||
|
||||
#[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<String>,
|
||||
/// `approved` / `denied` / `failed`.
|
||||
status: &'static str,
|
||||
/// RFC 3339 UTC. Renders as a relative time on the dashboard.
|
||||
resolved_at: DateTime<Utc>,
|
||||
/// Operator-supplied deny reason (for `denied`) or build error
|
||||
/// (for `failed`). None on `approved`.
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
note: Option<String>,
|
||||
}
|
||||
|
||||
#[derive(Serialize)]
|
||||
struct ApprovalView {
|
||||
id: i64,
|
||||
agent: String,
|
||||
kind: &'static str,
|
||||
/// First 12 chars of the `commit_ref`, for `ApplyCommit` only.
|
||||
sha_short: Option<String>,
|
||||
/// 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 `<span>` markup) and removes the only HTML-escape
|
||||
/// surface from the snapshot.
|
||||
diff: Option<String>,
|
||||
/// Manager-supplied description shown on the approval card.
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
description: Option<String>,
|
||||
/// 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<u64>,
|
||||
/// Raw `commit_ref` payload for `UpdateMetaInputs` (JSON-encoded
|
||||
/// `Vec<String>` 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<String>,
|
||||
/// 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<Utc>,
|
||||
}
|
||||
|
||||
/// 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<T, E>(what: &str, result: std::result::Result<T, E>) -> 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<AppState>,
|
||||
) -> axum::Json<StateSnapshot> {
|
||||
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<QuestionView> =
|
||||
log_default("questions.pending_all", state.coord.questions.pending_all())
|
||||
.into_iter()
|
||||
.map(QuestionView::from_question)
|
||||
.collect();
|
||||
let question_history: Vec<QuestionView> = 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<PeerHiveView> {
|
||||
#[derive(serde::Deserialize)]
|
||||
struct Raw {
|
||||
domain: String,
|
||||
cert_fingerprint: Option<String>,
|
||||
}
|
||||
let Ok(json) = std::env::var("HYPERHIVE_PEERS") else {
|
||||
return Vec::new();
|
||||
};
|
||||
let Ok(raw): Result<Vec<Raw>, _> = 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<PortConflict> {
|
||||
let mut by_port: std::collections::BTreeMap<u16, Vec<String>> =
|
||||
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<String, crate::coordinator::TransientState>,
|
||||
) -> Vec<TransientView> {
|
||||
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<Approval>) -> Vec<ApprovalView> {
|
||||
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::<u64>().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<AppState>) -> 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<crate::dashboard_events::DashboardEvent> = 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<String>,
|
||||
}
|
||||
|
||||
pub(super) async fn dashboard_stream(
|
||||
State(state): State<AppState>,
|
||||
axum::extract::Query(q): axum::extract::Query<DashboardStreamQuery>,
|
||||
) -> Sse<impl Stream<Item = Result<Event, Infallible>>> {
|
||||
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<std::collections::HashSet<String>> = q.kinds.and_then(|raw| {
|
||||
let set: std::collections::HashSet<String> = 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())
|
||||
}
|
||||
159
hive-c0re/src/dashboard/tombstones.rs
Normal file
159
hive-c0re/src/dashboard/tombstones.rs
Normal file
|
|
@ -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<String, crate::coordinator::TransientState>,
|
||||
) -> Vec<TombstoneView> {
|
||||
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<Coordinator>) {
|
||||
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<AppState>,
|
||||
AxumPath(name): AxumPath<String>,
|
||||
) -> 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(", ")))
|
||||
}
|
||||
}
|
||||
Loading…
Reference in a new issue