749 lines
30 KiB
Rust
749 lines
30 KiB
Rust
//! `/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, 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>,
|
|
/// Live running/stopped status for the four hive infra containers
|
|
/// (`hive-ci`, `hive-forge`, `hive-gateway`, `hive-matrix`). Feeds the
|
|
/// C0R3 page's 1NFR4 sub-tab so the operator can start/stop/restart
|
|
/// them without an `infra_admin` agent's `restart` tool.
|
|
infra_containers: Vec<InfraContainerView>,
|
|
}
|
|
|
|
/// One row for the C0R3 page's 1NFR4 sub-tab.
|
|
#[derive(Serialize)]
|
|
struct InfraContainerView {
|
|
/// Container / systemd-unit name (e.g. `"hive-ci"`).
|
|
name: &'static str,
|
|
running: bool,
|
|
}
|
|
|
|
/// Live running/stopped status for all four hive infra containers.
|
|
/// Extracted out of [`api_state`] to keep it under clippy's
|
|
/// `too_many_lines` limit.
|
|
async fn infra_container_views() -> Vec<InfraContainerView> {
|
|
let mut infra_containers = Vec::with_capacity(hive_priv_sock::InfraContainer::ALL.len());
|
|
for container in hive_priv_sock::InfraContainer::ALL {
|
|
infra_containers.push(InfraContainerView {
|
|
name: container.unit_name(),
|
|
running: crate::lifecycle::infra_is_running(container).await,
|
|
});
|
|
}
|
|
infra_containers
|
|
}
|
|
|
|
/// 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 reviewed PR head sha, for `MergeConfigPr`
|
|
/// only. Display-only (the short chip on the card).
|
|
sha_short: 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}`). `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);
|
|
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
|
|
};
|
|
|
|
let infra_containers = infra_container_views().await;
|
|
|
|
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,
|
|
infra_containers,
|
|
})
|
|
}
|
|
|
|
/// 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 for
|
|
/// `MergeConfigPr`, 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 = a.kind.as_str();
|
|
ApprovalHistoryView {
|
|
id: a.id,
|
|
agent: a.agent,
|
|
kind,
|
|
sha_short,
|
|
status,
|
|
resolved_at: a.resolved_at.unwrap_or_default(),
|
|
note: a.note,
|
|
}
|
|
}
|
|
|
|
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::Spawn => ApprovalView {
|
|
id: a.id,
|
|
agent: a.agent,
|
|
kind: "spawn",
|
|
sha_short: 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,
|
|
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,
|
|
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,
|
|
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 config diff surface lives
|
|
// on the forge PR itself.
|
|
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,
|
|
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())
|
|
}
|