Watch
0
0
Fork
You've already forked hyperhive
0

drop hive and agent response fields the frontend never reads

This commit is contained in:
atlas 2026-11-13 03:25:41 +01:00 • committed by mara
commit a437f9ccd5
28 changed files with 347 additions and 598 deletions

2
Cargo.lock generated
View file

@ -1939,8 +1939,6 @@ dependencies = [
name = "hive-jobq-wire" name = "hive-jobq-wire"
version = "0.1.0" version = "0.1.0"
dependencies = [ dependencies = [
"chrono",
"enumflags2",
"hive-jobq", "hive-jobq",
"serde", "serde",
"serde_json", "serde_json",

View file

@ -127,7 +127,7 @@ Shape:
- `needs_login` — set when a turn hits 401 (expired OAuth credentials); - `needs_login` — set when a turn hits 401 (expired OAuth credentials);
cleared by `"online"` status (re-auth completed). Drives the cleared by `"online"` status (re-auth completed). Drives the
`needs_login` flag alongside the `claude_has_session` check. `needs_login` flag alongside the `claude_has_session` check.
- `active_model` — the resolved Claude model for the dashboard badge. - `active_model` — the resolved Claude model.
The turn loop is the only writer today, but it still goes The turn loop is the only writer today, but it still goes
read-modify-write under a shared in-process lock and merges into the read-modify-write under a shared in-process lock and merges into the

View file

@ -256,7 +256,7 @@ the dashboard renders one recent-builds list and one number bounds it.
### Wire shape ### Wire shape
`RebuildQueueChanged { seq }` (event name kept) — **a bare trigger, no `RebuildQueueChanged` (event name kept) — **a bare trigger, no
payload.** It says _the queue changed_; a client that wants to know how payload.** It says _the queue changed_; a client that wants to know how
re-fetches `GET /api/jobq/graph`. re-fetches `GET /api/jobq/graph`.

View file

@ -73,9 +73,7 @@ match wins):
Once a turn completes, the window the model reports Once a turn completes, the window the model reports
(`modelUsage.*.contextWindow`) replaces this value. (`modelUsage.*.contextWindow`) replaces this value.
The effective window drives watermarks; the harness exposes it at The effective window drives watermarks; no web endpoint serves it.
runtime via `/api/state.context_window_tokens` so the UI can show a
percentage-of-window ctx badge.
**Session identity — a constant title.** Every turn keys on one fixed, **Session identity — a constant title.** Every turn keys on one fixed,
harness-owned session title (`turn::session_title()`, default harness-owned session title (`turn::session_title()`, default

View file

@ -99,7 +99,7 @@ at a given viewport width. Two columns:
Values throughout come from `GET /api/state`'s cold-load snapshot, Values throughout come from `GET /api/state`'s cold-load snapshot,
kept in sync afterwards by the SSE stream (see Live view below) — kept in sync afterwards by the SSE stream (see Live view below) —
`context_window_tokens` for the ctx badge tooltip, `qualified_label` `qualified_label`
(the hive-qualified `name@domain` form, used for the browser tab title (the hive-qualified `name@domain` form, used for the browser tab title
so two tabs from different hives are distinguishable; the header's own so two tabs from different hives are distinguishable; the header's own
`◆ <label> ◆` stays short). `◆ <label> ◆` stays short).
@ -288,15 +288,14 @@ credentials arrive from the swarm secret store.
- `GET /api/state` — cold-load snapshot (`StateSnapshot`) consumed by - `GET /api/state` — cold-load snapshot (`StateSnapshot`) consumed by
the frontend (`useAgentState.ts`) on page load. Includes `turn_state`, the frontend (`useAgentState.ts`) on page load. Includes `turn_state`,
`context_window_tokens`, `qualified_label`, `qualified_label`,
`model`, `model_source`, `effort`, `effort_source`, `links`, and `model`, `model_source`, `effort`, `effort_source`, `links`, and
other fields described inline throughout this document. All other fields described inline throughout this document. All
subsequent state updates arrive via SSE. subsequent state updates arrive via SSE.
- `GET /api/dashboard-state` — lean snapshot of agent-owned fields - `GET /api/dashboard-state` — lean snapshot of agent-owned fields
fetched once per running agent by the dashboard's container row to fetched once per running agent by the dashboard's container row to
get fresh values without relying on hive-c0re's periodic file-reads. get fresh values without relying on hive-c0re's periodic file-reads.
Returns `{ status_text?, status_set_at?, ctx_tokens?, context_window_tokens, Returns `{ status_text?, status_set_at?, rate_limited, links }`. Only called when the container is running;
rate_limited, links }`. Only called when the container is running;
skipped (muted badges) when stopped. Also accessible via the gateway skipped (muted badges) when stopped. Also accessible via the gateway
at `/agent/<name>/api/dashboard-state`. at `/agent/<name>/api/dashboard-state`.
- `GET /api/todos` — harness-local todos snapshot (loose-ends v2) - `GET /api/todos` — harness-local todos snapshot (loose-ends v2)

View file

@ -60,7 +60,7 @@ active; hash changes don't reload the page.
container, sorted alphabetically by agent name; every container the container, sorted alphabetically by agent name; every container the
rescan reports gets a row. Pulsing red banner at the top rescan reports gets a row. Pulsing red banner at the top
of this section if any two sub-agents hash to the same port of this section if any two sub-agents hash to the same port
(`port_conflicts` from `/api/state`): the operator must rename (derived client-side from the container ports): the operator must rename
one of them and rebuild. `lifecycle::{spawn,rebuild}` also one of them and rebuild. `lifecycle::{spawn,rebuild}` also
preflight this and refuse with a clear error message naming the preflight this and refuse with a clear error message naming the
conflicting agent. conflicting agent.
@ -117,7 +117,7 @@ pipeline) with an `onUpdate` callback to drive the two things below it
that the generic view doesn't show. The component owns fetching, cold that the generic view doesn't show. The component owns fetching, cold
and live: `GET /api/jobq/graph` on mount, and a refetch via the mount and live: `GET /api/jobq/graph` on mount, and a refetch via the mount
handle's `.refresh()` on every `rebuild_queue_changed` SSE tick (that handle's `.refresh()` on every `rebuild_queue_changed` SSE tick (that
event is a bare `{ seq }` trigger carrying no queue payload; both event is a bare trigger carrying no queue payload; both
dashboard pages treat the tick as a pure refetch trigger). dashboard pages treat the tick as a pure refetch trigger).
Each row is one root graph node (`parent: null`); a multi-step op's Each row is one root graph node (`parent: null`); a multi-step op's
@ -425,9 +425,8 @@ below — some endpoints aren't in it yet.
- `POST /api/purge-tombstone/{name}` — wipe a tombstone's state dirs. - `POST /api/purge-tombstone/{name}` — wipe a tombstone's state dirs.
- `POST /api/rebuild-queue/{id}/cancel` — drop a `Queued` entry. - `POST /api/rebuild-queue/{id}/cancel` — drop a `Queued` entry.
Refuses `Running` / terminal-state entries (in-flight Refuses `Running` / terminal-state entries (in-flight
rebuilds can't be safely interrupted). Always 200; body is rebuilds can't be safely interrupted). Always 200 with an empty body, whether or not
`{"cancelled": true}` on a successful flip or the entry was in `Queued` state.
`{"cancelled": false}` when the entry wasn't in `Queued` state.
- `POST /api/agent/{name}/mark-all-read` — ack all pending broker - `POST /api/agent/{name}/mark-all-read` — ack all pending broker
messages for `{name}`. Backfills `delivered_at` for rows messages for `{name}`. Backfills `delivered_at` for rows
not yet delivered and sets `acked_at = now`. Returns not yet delivered and sets `acked_at = now`. Returns
@ -436,8 +435,7 @@ below — some endpoints aren't in it yet.
- `GET /api/dashboard/stream` — unified live event channel: - `GET /api/dashboard/stream` — unified live event channel:
broker `sent` / `delivered`, plus the mutation events listed broker `sent` / `delivered`, plus the mutation events listed
below. Each frame carries `seq`. below. <!-- vale write-good.Passive = YES -->
<!-- vale write-good.Passive = YES -->
## Dashboard event channel ## Dashboard event channel
@ -463,12 +461,12 @@ payload):
`crash_watch` poll. Client upserts/removes by name and `crash_watch` poll. Client upserts/removes by name and
reads the pending overlay from `transientsState` since the reads the pending overlay from `transientsState` since the
payload doesn't carry it. payload doesn't carry it.
- `rebuild_queue_changed` (seq) — **payload-free trigger**, fired on - `rebuild_queue_changed` — **payload-free trigger**, fired on
every queue mutation (enqueue, state transition, dedup collapse, every queue mutation (enqueue, state transition, dedup collapse,
terminal-history trim). Unlike the snapshot events below it ships terminal-history trim). Unlike the snapshot events below it ships
no state at all: the client re-fetches `GET /api/jobq/graph`, which no state at all: the client re-fetches `GET /api/jobq/graph`, which
is where it reads the queue from cold too. is where it reads the queue from cold too.
- `tombstones_changed` (seq, tombstones: `Vec<TombstoneView>`) — - `tombstones_changed` (tombstones: `Vec<TombstoneView>`) —
full snapshot of destroyed-but-state-kept agents. Emitted on full snapshot of destroyed-but-state-kept agents. Emitted on
destroy, purge, and revive. The K3PT ST4T3 tab subscribes. destroy, purge, and revive. The K3PT ST4T3 tab subscribes.

View file

@ -39,15 +39,12 @@
XSS-safe by construction the way the text-node `linkify` path is. XSS-safe by construction the way the text-node `linkify` path is.
Rendered `<a>`s are rewritten to `target="_blank"`. Rendered `<a>`s are rewritten to `target="_blank"`.
- `GET /api/state` → JSON snapshot the JS app renders into the - `GET /api/state` → JSON snapshot the JS app renders into the
DOM. Includes a top-level `seq` (the dashboard event channel's DOM.
high-water mark at the moment the server assembled the snapshot);
clients use it to dedupe their buffered SSE traffic against
the snapshot (drop frames with `seq <= snapshot.seq`).
- `GET /api/dashboard/stream` (dashboard) / `GET /events/stream` - `GET /api/dashboard/stream` (dashboard) / `GET /events/stream`
(per-agent) → `text/event-stream` SSE for live updates. The (per-agent) → `text/event-stream` SSE for live updates. The
dashboard stream carries broker `Sent` / `Delivered` (mirrored dashboard stream carries broker `Sent` / `Delivered` (mirrored
by a forwarder task from the broker's intra-process channel) by a forwarder task from the broker's intra-process channel)
plus mutation events (`transient_set` / `transient_cleared`). Each frame carries a `seq`. The plus mutation events (`transient_set` / `transient_cleared`). The
agent's matching backfill endpoint is `GET /events/history` (last 2000 agent's matching backfill endpoint is `GET /events/history` (last 2000
`LiveEvent`s wrapped in `{ seq, events }`). `LiveEvent`s wrapped in `{ seq, events }`).
**One unified channel**: browsers cap concurrent SSE **One unified channel**: browsers cap concurrent SSE

View file

@ -98,8 +98,8 @@ The broker stores the recipient exactly as the agent passed it.
JSON line-delimited over unix sockets in both directions (host admin JSON line-delimited over unix sockets in both directions (host admin
/ manager / agent). SSE streams (`/dashboard/stream` on hive-c0re, / manager / agent). SSE streams (`/dashboard/stream` on hive-c0re,
`/events/stream` on the per-agent web UIs) are `text/event-stream`; `/events/stream` on the per-agent web UIs) are `text/event-stream`;
each frame carries a `seq` field for the snapshot-dedupe dance the per-agent `/events/stream` frames carry a `seq` field for the
(see `docs/4-implementation/web-ui/shape.md`). Request/response types live in `hive-sh4re` snapshot-dedupe dance (see `docs/4-implementation/web-ui/shape.md`). Request/response types live in `hive-sh4re`
— change them in one place. The dashboard event vocabulary lives — change them in one place. The dashboard event vocabulary lives
in `hive-c0re::dashboard_events::DashboardEvent`. in `hive-c0re::dashboard_events::DashboardEvent`.

View file

@ -8,7 +8,6 @@
export type InferenceSource = "swarm" | "agent_nix" | "local_file" | "default"; export type InferenceSource = "swarm" | "agent_nix" | "local_file" | "default";
export interface AgentState { export interface AgentState {
seq: number;
label: string; label: string;
qualified_label: string; qualified_label: string;
dashboard_port: number; dashboard_port: number;
@ -17,7 +16,6 @@ export interface AgentState {
turn_state_since: number; turn_state_since: number;
model: string; model: string;
resolved_model: string | null; resolved_model: string | null;
context_window_tokens: number;
ctx_usage: TokenUsage | null; ctx_usage: TokenUsage | null;
cost_usage: TokenUsage | null; cost_usage: TokenUsage | null;
hive_name: string | null; hive_name: string | null;

View file

@ -92,7 +92,7 @@ export function syncTransientsFromSnapshot(s) {
// Snapshot ships `secs` (server-computed); reconstruct an // Snapshot ships `secs` (server-computed); reconstruct an
// approximate since_unix so the live ticker keeps progressing // approximate since_unix so the live ticker keeps progressing
// without surprising jumps when the next snapshot lands. // without surprising jumps when the next snapshot lands.
byKind.set(t.kind, t.since_unix ?? nowUnix - (t.secs || 0)); byKind.set(t.kind, nowUnix - (t.secs || 0));
} }
} }
export function applyTransientSet(ev) { export function applyTransientSet(ev) {
@ -193,9 +193,8 @@ function buildAgentMenu(c) {
// ─── port conflicts ────────────────────────────────────────────────────────── // ─── port conflicts ──────────────────────────────────────────────────────────
// Re-derive port conflicts from the live containers map. Mirrors the // Derive port conflicts from the live containers map, so the banner
// server-side `build_port_conflicts` so the banner reacts to event // reacts to event updates without a /api/state refetch.
// updates instead of waiting for a /api/state refetch.
function derivePortConflicts(containers) { function derivePortConflicts(containers) {
const byPort = new Map(); const byPort = new Map();
for (const c of containers) { for (const c of containers) {

View file

@ -44,12 +44,10 @@ type NodeState =
| "Cancelled" | "Cancelled"
| "Skipped"; | "Skipped";
type TerminalState = "Done" | "Failed" | "Cancelled" | "Skipped";
// Mirrors `hive_jobq_wire::GraphDep` — externally tagged on `kind`, // Mirrors `hive_jobq_wire::GraphDep` — externally tagged on `kind`,
// values are the Rust variant names verbatim. // values are the Rust variant names verbatim.
type GraphDep = type GraphDep =
| { kind: "Node"; id: number; accepts: TerminalState[] } | { kind: "Node"; id: number }
| { kind: "Resource"; name: string; count: number }; | { kind: "Resource"; name: string; count: number };
interface NodePayload { interface NodePayload {

View file

@ -109,8 +109,6 @@ fn adaptive_bucket_secs(span_secs: i64) -> i64 {
pub struct Snapshot { pub struct Snapshot {
pub window: &'static str, pub window: &'static str,
pub bucket_seconds: i64, pub bucket_seconds: i64,
pub now: i64,
pub from: i64,
/// Total turns in the window. /// Total turns in the window.
pub turn_count: u64, pub turn_count: u64,
/// Time-bucketed trend series, oldest first. Always covers the /// Time-bucketed trend series, oldest first. Always covers the
@ -231,8 +229,6 @@ fn empty_snapshot(window: Window) -> Snapshot {
Snapshot { Snapshot {
window: window.label(), window: window.label(),
bucket_seconds: window.bucket_secs(), bucket_seconds: window.bucket_secs(),
now,
from,
turn_count: 0, turn_count: 0,
buckets, buckets,
tool_breakdown: Vec::new(), tool_breakdown: Vec::new(),
@ -360,8 +356,6 @@ fn snapshot(path: &Path, window: Window) -> Result<Snapshot> {
Ok(Snapshot { Ok(Snapshot {
window: window.label(), window: window.label(),
bucket_seconds: bucket_secs, bucket_seconds: bucket_secs,
now,
from,
turn_count: acc.turn_count, turn_count: acc.turn_count,
buckets, buckets,
tool_breakdown: top_n(acc.tool_totals, 10), tool_breakdown: top_n(acc.tool_totals, 10),
@ -873,4 +867,32 @@ mod tests {
Some("cargo") Some("cargo")
); );
} }
/// `GET /api/stats` carries exactly the fields the stats page reads.
#[test]
fn the_stats_snapshot_serialises_exactly_its_read_fields() {
let value = serde_json::to_value(empty_snapshot(Window::Hour)).unwrap();
let mut keys: Vec<&str> = value
.as_object()
.unwrap()
.keys()
.map(String::as_str)
.collect();
keys.sort_unstable();
assert_eq!(
keys,
[
"bash_breakdown",
"bucket_seconds",
"buckets",
"duration_summary",
"models",
"result_mix",
"tool_breakdown",
"turn_count",
"wake_mix",
"window",
]
);
}
} }

View file

@ -6,7 +6,7 @@ use axum::{
http::StatusCode, http::StatusCode,
response::{IntoResponse, Response}, response::{IntoResponse, Response},
}; };
use serde::{Deserialize, Serialize}; use serde::Deserialize;
use std::future::Future; use std::future::Future;
use super::{AppState, SOCKET_FETCH_TIMEOUT, error_response}; use super::{AppState, SOCKET_FETCH_TIMEOUT, error_response};
@ -53,9 +53,8 @@ pub(super) struct MarkTodosDoneForm {
/// `hive-agent-sock`: the todos list is small (single-digit rows most of the /// `hive-agent-sock`: the todos list is small (single-digit rows most of the
/// time), so N same-host socket round-trips isn't a real cost, and it keeps /// time), so N same-host socket round-trips isn't a real cost, and it keeps
/// the wire protocol's `Request` enum — already used by the `cancel_loose_end` /// the wire protocol's `Request` enum — already used by the `cancel_loose_end`
/// MCP tool — unchanged. Unknown/already-acked ids just don't add to the /// MCP tool — unchanged. Unknown/already-acked ids succeed too (acking twice is
/// `acked` count (same "acking twice is not a new action" semantics as the /// not a new action); a request with no ids or where every id fails to parse
/// single-id path); a request with no ids or where every id fails to parse
/// is rejected as a client error rather than silently acking nothing. An id /// is rejected as a client error rather than silently acking nothing. An id
/// whose store call errors or gets no answer fails the request with a 500 /// whose store call errors or gets no answer fails the request with a 500
/// naming how many failed. More than [`MAX_MARK_DONE_IDS`] ids is a client /// naming how many failed. More than [`MAX_MARK_DONE_IDS`] ids is a client
@ -102,15 +101,14 @@ where
); );
} }
match settle_mark_done(answers) { match settle_mark_done(answers) {
Ok(acked) => axum::Json(MarkTodosDoneBody { acked }).into_response(), Ok(()) => StatusCode::OK.into_response(),
Err(message) => error_response(StatusCode::INTERNAL_SERVER_ERROR, &message), Err(message) => error_response(StatusCode::INTERNAL_SERVER_ERROR, &message),
} }
} }
/// Folds the per-id answers into the acked total, or the operator-facing /// Folds the per-id answers into success, or the operator-facing failure. `None` is a dial that got no answer; `Response::Err` and any
/// failure. `None` is a dial that got no answer; `Response::Err` and any
/// other variant are failures. `Acked { count: 0 }` is a success. /// other variant are failures. `Acked { count: 0 }` is a success.
fn settle_mark_done(answers: Vec<Option<hive_agent_sock::Response>>) -> Result<u64, String> { fn settle_mark_done(answers: Vec<Option<hive_agent_sock::Response>>) -> Result<(), String> {
let total = answers.len(); let total = answers.len();
let mut acked = 0u64; let mut acked = 0u64;
let mut failed = 0usize; let mut failed = 0usize;
@ -133,18 +131,13 @@ fn settle_mark_done(answers: Vec<Option<hive_agent_sock::Response>>) -> Result<u
} }
} }
match first_error { match first_error {
None => Ok(acked), None => Ok(()),
Some(error) => Err(format!( Some(error) => Err(format!(
"mark-done: {failed} of {total} failed ({acked} marked done): {error}" "mark-done: {failed} of {total} failed ({acked} marked done): {error}"
)), )),
} }
} }
#[derive(Serialize)]
struct MarkTodosDoneBody {
acked: u64,
}
#[cfg(test)] #[cfg(test)]
mod tests { mod tests {
use super::{mark_ids_done, settle_mark_done}; use super::{mark_ids_done, settle_mark_done};
@ -187,13 +180,13 @@ mod tests {
} }
#[tokio::test] #[tokio::test]
async fn mark_done_handler_answers_200_with_the_acked_total() { async fn mark_done_handler_answers_200_with_an_empty_body() {
let response = mark_ids_done(vec![1, 2], std::time::Duration::from_secs(3), |_| async { let response = mark_ids_done(vec![1, 2], std::time::Duration::from_secs(3), |_| async {
Some(hive_agent_sock::Response::Acked { count: 1 }) Some(hive_agent_sock::Response::Acked { count: 1 })
}) })
.await; .await;
assert_eq!(response.status(), axum::http::StatusCode::OK); assert_eq!(response.status(), axum::http::StatusCode::OK);
assert_eq!(body_text(response).await, r#"{"acked":2}"#); assert_eq!(body_text(response).await, "");
} }
#[tokio::test(start_paused = true)] #[tokio::test(start_paused = true)]
@ -256,6 +249,6 @@ mod tests {
Some(hive_agent_sock::Response::Acked { count: 0 }), Some(hive_agent_sock::Response::Acked { count: 0 }),
Some(hive_agent_sock::Response::Acked { count: 2 }), Some(hive_agent_sock::Response::Acked { count: 2 }),
]; ];
assert_eq!(settle_mark_done(answers).unwrap(), 2); assert!(settle_mark_done(answers).is_ok());
} }
} }

View file

@ -9,9 +9,6 @@ use crate::login::LoginState;
use super::AppState; use super::AppState;
pub(super) async fn api_state(State(state): State<AppState>) -> axum::Json<StateSnapshot> { pub(super) async fn api_state(State(state): State<AppState>) -> axum::Json<StateSnapshot> {
// Capture seq *before* any reads so the dedupe contract is
// "events with seq > snapshot.seq are post-snapshot, never missed."
let seq = state.bus.current_seq();
let login = *state.login.lock().unwrap(); let login = *state.login.lock().unwrap();
let status = status_of(login, state.bus.is_rate_limited()); let status = status_of(login, state.bus.is_rate_limited());
let dashboard_port = std::env::var("HIVE_DASHBOARD_PORT") let dashboard_port = std::env::var("HIVE_DASHBOARD_PORT")
@ -20,7 +17,6 @@ pub(super) async fn api_state(State(state): State<AppState>) -> axum::Json<State
.unwrap_or(7000); .unwrap_or(7000);
let inbox = recent_inbox(&state.socket).await; let inbox = recent_inbox(&state.socket).await;
let (turn_state, turn_state_since) = state.bus.state_snapshot(); let (turn_state, turn_state_since) = state.bus.state_snapshot();
let model = state.bus.model();
// The alias the operator selected (e.g. "sonnet") vs. what the last // The alias the operator selected (e.g. "sonnet") vs. what the last
// turn actually ran on (e.g. "claude-sonnet-4-5-20260805") — mara: // turn actually ran on (e.g. "claude-sonnet-4-5-20260805") — mara:
// "i cannot see if sonnet became claude-sonnet-5 or something else". // "i cannot see if sonnet became claude-sonnet-5 or something else".
@ -28,12 +24,10 @@ pub(super) async fn api_state(State(state): State<AppState>) -> axum::Json<State
// resolve from yet); same source `serve_common.rs`'s turn-stats // resolve from yet); same source `serve_common.rs`'s turn-stats
// rollup already prefers over the requested alias. // rollup already prefers over the requested alias.
let resolved_model = state.bus.last_resolved_model(); let resolved_model = state.bus.last_resolved_model();
let context_window_tokens = state.bus.effective_context_window(&model);
let ctx_usage = state.bus.last_ctx_usage(); let ctx_usage = state.bus.last_ctx_usage();
let cost_usage = state.bus.last_cost_usage(); let cost_usage = state.bus.last_cost_usage();
let shown = shown(&state.bus); let shown = shown(&state.bus);
axum::Json(StateSnapshot { axum::Json(StateSnapshot {
seq,
label: state.label.clone(), label: state.label.clone(),
qualified_label: crate::identity::qualify(&state.label), qualified_label: crate::identity::qualify(&state.label),
dashboard_port, dashboard_port,
@ -44,7 +38,6 @@ pub(super) async fn api_state(State(state): State<AppState>) -> axum::Json<State
model: shown.model, model: shown.model,
model_source: state.bus.model_source(), model_source: state.bus.model_source(),
resolved_model, resolved_model,
context_window_tokens,
ctx_usage, ctx_usage,
cost_usage, cost_usage,
links: agent_links(state.gui_vnc_port.is_some()), links: agent_links(state.gui_vnc_port.is_some()),
@ -70,19 +63,9 @@ pub(super) async fn api_dashboard_state(
) -> axum::Json<DashboardState> { ) -> axum::Json<DashboardState> {
let (status_text, status_set_at) = read_own_status(); let (status_text, status_set_at) = read_own_status();
let rate_limited = state.bus.is_rate_limited(); let rate_limited = state.bus.is_rate_limited();
let model = state.bus.model();
let context_window_tokens = state.bus.effective_context_window(&model);
// Full context-window size = input + cache-read + cache-creation. Using
// raw `input_tokens` here reported only the *uncached* sliver, which is
// ~0 once prompt caching kicks in — so every card showed `ctx·0k`. Match
// the agent page (which ships the whole `ctx_usage` and sums it) and the
// cache-TTL logic in turn.rs, both of which use `context_tokens()`.
let ctx_tokens = state.bus.last_ctx_usage().map(|u| u.context_tokens());
axum::Json(DashboardState { axum::Json(DashboardState {
status_text, status_text,
status_set_at, status_set_at,
ctx_tokens,
context_window_tokens,
rate_limited, rate_limited,
links: agent_links(state.gui_vnc_port.is_some()), links: agent_links(state.gui_vnc_port.is_some()),
}) })
@ -90,12 +73,6 @@ pub(super) async fn api_dashboard_state(
#[derive(Serialize)] #[derive(Serialize)]
pub(super) struct StateSnapshot { pub(super) struct StateSnapshot {
/// Bus seq at the moment this snapshot was assembled. Clients dedupe
/// their buffered SSE traffic against this value: events with
/// `seq <= snapshot.seq` are already reflected (or pre-date the
/// snapshot); `seq > snapshot.seq` is post-snapshot. Reset to 0 on
/// harness restart — clients treat reconnect as a fresh world.
seq: u64,
label: String, label: String,
/// Hive-qualified long name (`${label}@${hyperhive.domain}`) when /// Hive-qualified long name (`${label}@${hyperhive.domain}`) when
/// the host has been configured for a multi-hive swarm; falls back /// the host has been configured for a multi-hive swarm; falls back
@ -126,12 +103,6 @@ pub(super) struct StateSnapshot {
/// operator picked, which claude itself maps to a specific dated /// operator picked, which claude itself maps to a specific dated
/// snapshot. `None` before any turn has completed. /// snapshot. `None` before any turn has completed.
resolved_model: Option<String>, resolved_model: Option<String>,
/// Effective context-window token budget for the current model.
/// Primary source: API-reported `modelUsage.*.contextWindow` from
/// the last result event (authoritative per-inference active window).
/// Falls back to `harness_state::context_window_tokens(model)`.
/// Consumers (e.g. dashboard badge) use this to render ctx-usage %.
context_window_tokens: u64,
/// Last-inference token usage from the most recent completed /// Last-inference token usage from the most recent completed
/// turn — represents the current context-window size at turn-end. /// turn — represents the current context-window size at turn-end.
/// `null` until the first turn finishes. /// `null` until the first turn finishes.
@ -214,14 +185,6 @@ pub(super) struct DashboardState {
/// `None` when no status is set. /// `None` when no status is set.
#[serde(skip_serializing_if = "Option::is_none")] #[serde(skip_serializing_if = "Option::is_none")]
status_set_at: Option<i64>, status_set_at: Option<i64>,
/// Full context-window size from the most recent completed turn
/// (`ctx_usage.context_tokens()` = input + cache-read + cache-creation).
/// `None` until the first turn finishes. Drives the `ctx·Nk` card badge.
#[serde(skip_serializing_if = "Option::is_none")]
ctx_tokens: Option<u64>,
/// Effective context-window budget for the current model. Same
/// derivation as `StateSnapshot::context_window_tokens`.
context_window_tokens: u64,
/// True while the harness is parked after a rate-limit response. /// True while the harness is parked after a rate-limit response.
rate_limited: bool, rate_limited: bool,
/// Navigation links for the dashboard card's icon strip. This is /// Navigation links for the dashboard card's icon strip. This is
@ -487,4 +450,76 @@ mod tests {
shown("haiku", None) shown("haiku", None)
); );
} }
fn sorted_keys(v: &serde_json::Value) -> Vec<&str> {
let mut keys: Vec<&str> = v.as_object().unwrap().keys().map(String::as_str).collect();
keys.sort_unstable();
keys
}
/// `GET /api/state` carries exactly the fields the agent page and stats
/// page read; a field added back needs a reader.
#[test]
fn the_state_snapshot_serialises_exactly_its_read_fields() {
let snap = super::StateSnapshot {
label: "a".to_owned(),
qualified_label: "a".to_owned(),
dashboard_port: 7000,
status: "online",
inbox: Vec::new(),
turn_state: crate::events::TurnState::Idle,
turn_state_since: 0,
model: "haiku".to_owned(),
model_source: swarm_wire::agent_inference::Source::Default,
resolved_model: None,
ctx_usage: None,
cost_usage: None,
links: Vec::new(),
hive_name: None,
swarm_name: None,
effort: None,
effort_source: swarm_wire::agent_inference::Source::Default,
paused: false,
};
let value = serde_json::to_value(&snap).unwrap();
assert_eq!(
sorted_keys(&value),
[
"cost_usage",
"ctx_usage",
"dashboard_port",
"effort",
"effort_source",
"hive_name",
"inbox",
"label",
"links",
"model",
"model_source",
"paused",
"qualified_label",
"resolved_model",
"status",
"swarm_name",
"turn_state",
"turn_state_since",
]
);
}
/// `GET /api/dashboard-state` carries exactly what the dashboard card reads.
#[test]
fn the_dashboard_state_serialises_exactly_its_read_fields() {
let state = super::DashboardState {
status_text: Some("busy".to_owned()),
status_set_at: Some(1),
rate_limited: false,
links: Vec::new(),
};
let value = serde_json::to_value(&state).unwrap();
assert_eq!(
sorted_keys(&value),
["links", "rate_limited", "status_set_at", "status_text"]
);
}
} }

View file

@ -7,7 +7,6 @@
use std::path::Path; use std::path::Path;
use chrono::{DateTime, Utc};
use serde::Serialize; use serde::Serialize;
use crate::coordinator::Coordinator; use crate::coordinator::Coordinator;
@ -43,24 +42,6 @@ pub struct ContainerView {
#[serde(default)] #[serde(default)]
pub failed: bool, pub failed: bool,
pub needs_login: bool, pub needs_login: bool,
/// The Claude model the agent's harness is currently using, read from
/// `state/hyperhive-harness.json["active_model"]`. `None` when the
/// agent has never started a turn or the field is absent. Only
/// meaningful when `running` is true; the dashboard skips the badge
/// for stopped containers.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub active_model: Option<String>,
/// The agent's current free-text status (`set_status`). `None` when
/// unset, or when the container isn't running — see
/// [`read_agent_status_live`] for why a stopped agent's on-disk
/// status is treated as stale rather than surfaced.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub status_text: Option<String>,
/// When `status_text` was last set, RFC 3339 UTC on the wire (see
/// `hive_sh4re::wire_time`). `None` exactly when `status_text` is
/// `None`.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub status_set_at: Option<DateTime<Utc>>,
/// The agent's turn loop is parked: the container is up and still /// The agent's turn loop is parked: the container is up and still
/// serving its web UI / MCP daemons, but it drives no turns and its /// serving its web UI / MCP daemons, but it drives no turns and its
/// inbox messages queue unacked until it resumes. Sourced from the /// inbox messages queue unacked until it resumes. Sourced from the
@ -95,9 +76,6 @@ impl From<ContainerView> for hive_sh4re::container::AgentStatusRow {
needs_login: v.needs_login, needs_login: v.needs_login,
pending_reminders: 0, pending_reminders: 0,
paused: v.paused, paused: v.paused,
active_model: v.active_model,
status_text: v.status_text,
status_set_at: v.status_set_at,
} }
} }
} }
@ -173,25 +151,6 @@ pub async fn build_all() -> anyhow::Result<Vec<ContainerView>> {
let needs_login = running let needs_login = running
&& ((!is_api_key && !claude_has_session(&Coordinator::agent_claude_dir(&logical))) && ((!is_api_key && !claude_has_session(&Coordinator::agent_claude_dir(&logical)))
|| needs_login_sentinel); || needs_login_sentinel);
// Read the active model from the harness state file. Only surfaced
// when the container is running — stale model info from a stopped
// agent is misleading (the model may change on next boot).
let active_model = if running {
read_active_model(&logical)
} else {
None
};
// Same running-gate as `active_model` just above, and the same
// rule `read_agent_status_live` applies for its other callers —
// duplicated inline rather than calling that helper because it
// would re-derive `running` with its own `lifecycle::is_running`
// shell-out when this loop already has it for free.
let (status_text, status_set_at) = if running {
let (text, set_at) = read_agent_status(&logical);
(text, set_at.map(hive_sh4re::wire_time::from_secs))
} else {
(None, None)
};
let paused = Coordinator::is_paused(&logical); let paused = Coordinator::is_paused(&logical);
out.push(ContainerView { out.push(ContainerView {
port: lifecycle::agent_web_port(logical.as_str()), port: lifecycle::agent_web_port(logical.as_str()),
@ -200,9 +159,6 @@ pub async fn build_all() -> anyhow::Result<Vec<ContainerView>> {
container: c.clone(), container: c.clone(),
name: logical.into_string(), name: logical.into_string(),
needs_login, needs_login,
active_model,
status_text,
status_set_at,
paused, paused,
}); });
} }
@ -329,21 +285,6 @@ fn status_if_running(
(text, set_at, true) (text, set_at, true)
} }
/// Read the active Claude model from `hyperhive-harness.json` (the
/// consolidated harness state file in the agent's state dir). Written by
/// the harness on startup and on every `apply_model` / `emit_status` call,
/// so it always reflects the model the agent runs. Returns `None` when the
/// field is absent or the harness has not yet started a turn.
fn read_active_model(name: &hive_types::Ident) -> Option<String> {
let path = Coordinator::agent_notes_dir(name).join("hyperhive-harness.json");
let raw = std::fs::read_to_string(path).ok()?;
let v: serde_json::Value = serde_json::from_str(&raw).ok()?;
v.get("active_model")
.and_then(serde_json::Value::as_str)
.filter(|s| !s.is_empty())
.map(str::to_owned)
}
/// Host-side hive + swarm display names, read from the c0re service's /// Host-side hive + swarm display names, read from the c0re service's
/// own process env. The hyperhive NixOS module sets these from /// own process env. The hyperhive NixOS module sets these from
/// `services.hyperhive.deploy.hive-controller.hiveName` + `services.hyperhive.swarm.name` /// `services.hyperhive.deploy.hive-controller.hiveName` + `services.hyperhive.swarm.name`
@ -366,6 +307,34 @@ pub fn hive_swarm_names() -> (Option<String>, Option<String>) {
#[cfg(test)] #[cfg(test)]
mod tests { mod tests {
/// The serialised row has exactly the keys the dashboard reads; `container`
/// is internal and skipped.
#[test]
fn the_container_view_serialises_exactly_its_read_fields() {
use super::ContainerView;
let view = ContainerView {
name: "alice".to_owned(),
container: "h-alice".to_owned(),
port: 7000,
running: true,
failed: false,
needs_login: false,
paused: false,
};
let value = serde_json::to_value(&view).unwrap();
let mut keys: Vec<&str> = value
.as_object()
.unwrap()
.keys()
.map(String::as_str)
.collect();
keys.sort_unstable();
assert_eq!(
keys,
["failed", "name", "needs_login", "paused", "port", "running"]
);
}
/// The row says nothing of an update being pending or of a deployed commit: /// The row says nothing of an update being pending or of a deployed commit:
/// the hive applies a changed declaration on its own, so there is no state /// the hive applies a changed declaration on its own, so there is no state
/// for an operator to act on. /// for an operator to act on.
@ -379,9 +348,6 @@ mod tests {
running: true, running: true,
failed: false, failed: false,
needs_login: false, needs_login: false,
active_model: None,
status_text: None,
status_set_at: None,
paused: false, paused: false,
}; };
let value = serde_json::to_value(&view).unwrap(); let value = serde_json::to_value(&view).unwrap();
@ -403,7 +369,6 @@ mod tests {
#[test] #[test]
fn agent_status_row_carries_every_field_the_row_has_and_drops_the_rest() { fn agent_status_row_carries_every_field_the_row_has_and_drops_the_rest() {
use super::ContainerView; use super::ContainerView;
use chrono::{TimeZone, Utc};
use hive_sh4re::container::AgentStatusRow; use hive_sh4re::container::AgentStatusRow;
let view = ContainerView { let view = ContainerView {
@ -413,9 +378,6 @@ mod tests {
running: true, running: true,
failed: false, failed: false,
needs_login: false, needs_login: false,
active_model: Some("claude-opus".to_owned()),
status_text: Some("shipping".to_owned()),
status_set_at: Some(Utc.timestamp_opt(1_700_000_000, 0).unwrap()),
paused: true, paused: true,
}; };
@ -425,12 +387,6 @@ mod tests {
assert!(!row.failed); assert!(!row.failed);
assert!(!row.needs_login); assert!(!row.needs_login);
assert!(row.paused); assert!(row.paused);
assert_eq!(row.active_model.as_deref(), Some("claude-opus"));
assert_eq!(row.status_text.as_deref(), Some("shipping"));
assert_eq!(
row.status_set_at,
Some(Utc.timestamp_opt(1_700_000_000, 0).unwrap())
);
// No source for reminders on this side (see the `From` impl's doc // No source for reminders on this side (see the `From` impl's doc
// comment) — always the stub, never left uninitialised. // comment) — always the stub, never left uninitialised.
assert_eq!(row.pending_reminders, 0); assert_eq!(row.pending_reminders, 0);

View file

@ -4,7 +4,6 @@
use std::collections::{HashMap, HashSet}; use std::collections::{HashMap, HashSet};
use std::path::{Path, PathBuf}; use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, Mutex}; use std::sync::{Arc, Mutex};
use anyhow::{Context, Result}; use anyhow::{Context, Result};
@ -17,8 +16,8 @@ use crate::job_queue::RunningTransient;
use crate::socket_server::{self, AgentSocket}; use crate::socket_server::{self, AgentSocket};
/// Capacity of the dashboard event channel. Slow browser subscribers /// Capacity of the dashboard event channel. Slow browser subscribers
/// (idle tab, throttled connection) drop frames past this — that's /// (idle tab, throttled connection) drop frames past this; the
/// fine, the seq dedupe makes a reconnect resync safe. /// dashboard refetches `/api/state` on reconnect.
const DASHBOARD_CHANNEL: usize = 256; const DASHBOARD_CHANNEL: usize = 256;
pub struct Coordinator { pub struct Coordinator {
@ -114,11 +113,7 @@ pub struct Coordinator {
/// stream. Carries broker messages (mirrored from `broker.subscribe` /// stream. Carries broker messages (mirrored from `broker.subscribe`
/// by the forwarder task in `main.rs`) and dashboard-only mutation /// by the forwarder task in `main.rs`) and dashboard-only mutation
/// events (container state, rebuild queue, etc.). /// events (container state, rebuild queue, etc.).
/// Snapshot endpoints capture `event_seq` before reading state so
/// the client can dedupe its buffered live traffic against the
/// snapshot.
dashboard_events: broadcast::Sender<DashboardEvent>, dashboard_events: broadcast::Sender<DashboardEvent>,
event_seq: AtomicU64,
/// Last container snapshot seen by `rescan_containers_and_emit`, /// Last container snapshot seen by `rescan_containers_and_emit`,
/// keyed by `ContainerView.name`. The rescan diffs a fresh /// keyed by `ContainerView.name`. The rescan diffs a fresh
@ -377,7 +372,6 @@ impl Coordinator {
updating: Mutex::new(HashSet::new()), updating: Mutex::new(HashSet::new()),
updating_changed: tokio::sync::Notify::new(), updating_changed: tokio::sync::Notify::new(),
dashboard_events, dashboard_events,
event_seq: AtomicU64::new(0),
last_containers: tokio::sync::Mutex::new(HashMap::new()), last_containers: tokio::sync::Mutex::new(HashMap::new()),
job_queue: Arc::new(crate::job_queue::JobQueue::new(build_slots)), job_queue: Arc::new(crate::job_queue::JobQueue::new(build_slots)),
@ -432,9 +426,7 @@ impl Coordinator {
/// keeps `snapshot` because *that* is still what a client ends up with; /// keeps `snapshot` because *that* is still what a client ends up with;
/// what changed is who serves it. /// what changed is who serves it.
pub fn emit_rebuild_queue_snapshot(self: &Arc<Self>) { pub fn emit_rebuild_queue_snapshot(self: &Arc<Self>) {
self.emit_dashboard_event(DashboardEvent::RebuildQueueChanged { self.emit_dashboard_event(DashboardEvent::RebuildQueueChanged);
seq: self.next_seq(),
});
} }
/// Subscribe to the shutdown watch channel. Background tasks call /// Subscribe to the shutdown watch channel. Background tasks call
@ -461,24 +453,7 @@ impl Coordinator {
self.dashboard_events.subscribe() self.dashboard_events.subscribe()
} }
/// Stamp the next sequence number. Each emission of a /// Broadcast a freshly-built `DashboardEvent`. Returns silently when there are no
/// `DashboardEvent` should fill its `seq` with `next_seq()` so the
/// frame the wire carries is the one the client uses to dedupe.
pub fn next_seq(&self) -> u64 {
self.event_seq.fetch_add(1, Ordering::SeqCst) + 1
}
/// Current high-water seq. Snapshot endpoints read this *before*
/// gathering state so the (snapshot.seq, snapshot) pair satisfies:
/// any frame with `seq > snapshot.seq` is post-snapshot. The seq
/// captured here may grow during snapshot construction — clients
/// may double-apply such events, which renderers must tolerate.
pub fn current_seq(&self) -> u64 {
self.event_seq.load(Ordering::SeqCst)
}
/// Broadcast a freshly-built `DashboardEvent` (caller fills `seq`
/// via `next_seq()`). Returns silently when there are no
/// subscribers — the dashboard channel is best-effort presentation /// subscribers — the dashboard channel is best-effort presentation
/// plumbing, not a delivery guarantee. /// plumbing, not a delivery guarantee.
pub fn emit_dashboard_event(&self, event: DashboardEvent) { pub fn emit_dashboard_event(&self, event: DashboardEvent) {
@ -533,16 +508,10 @@ impl Coordinator {
} }
drop(last); drop(last);
for c in changed_or_new { for c in changed_or_new {
self.emit_dashboard_event(DashboardEvent::ContainerStateChanged { self.emit_dashboard_event(DashboardEvent::ContainerStateChanged { container: c });
seq: self.next_seq(),
container: c,
});
} }
for name in removed { for name in removed {
self.emit_dashboard_event(DashboardEvent::ContainerRemoved { self.emit_dashboard_event(DashboardEvent::ContainerRemoved { name });
seq: self.next_seq(),
name,
});
} }
} }
@ -594,7 +563,6 @@ impl Coordinator {
.and_then(|d| i64::try_from(d.as_secs()).ok()) .and_then(|d| i64::try_from(d.as_secs()).ok())
.unwrap_or(0); .unwrap_or(0);
self.emit_dashboard_event(DashboardEvent::TransientSet { self.emit_dashboard_event(DashboardEvent::TransientSet {
seq: self.next_seq(),
name: name.to_owned(), name: name.to_owned(),
transient_kind: label, transient_kind: label,
since_unix, since_unix,
@ -623,7 +591,6 @@ impl Coordinator {
(deliberate_stop, std::time::Instant::now()), (deliberate_stop, std::time::Instant::now()),
); );
self.emit_dashboard_event(DashboardEvent::TransientCleared { self.emit_dashboard_event(DashboardEvent::TransientCleared {
seq: self.next_seq(),
name: name.to_owned(), name: name.to_owned(),
transient_kind: label.to_owned(), transient_kind: label.to_owned(),
}); });

View file

@ -2,16 +2,11 @@
use axum::{ use axum::{
extract::{Path as AxumPath, State}, extract::{Path as AxumPath, State},
response::{IntoResponse, Response}, http::StatusCode,
}; };
use super::AppState; use super::AppState;
#[derive(serde::Serialize, utoipa::ToSchema)]
pub(super) struct CancelResultBody {
cancelled: bool,
}
/// Drop still-queued work from the job /// Drop still-queued work from the job
/// queue. /// queue.
/// ///
@ -23,27 +18,25 @@ pub(super) struct CancelResultBody {
/// Refuses `Running` / terminal nodes: an /// Refuses `Running` / terminal nodes: an
/// in-flight node owns the agent's nix store + nixos-container update /// in-flight node owns the agent's nix store + nixos-container update
/// lock and can't be safely interrupted from the queue side. Always /// lock and can't be safely interrupted from the queue side. Always
/// returns 200; the body is `{"cancelled": true}` on a successful /// returns 200 with an empty body, whether or not anything was cancelled
/// flip to Cancelled, `{"cancelled": false}` when the DAG was /// (Running / terminal / gone nodes are left alone). On a successful flip to
/// Running / terminal / gone. On success a fresh `RebuildQueueChanged` /// Cancelled a fresh `RebuildQueueChanged` snapshot fires so the state flip
/// snapshot fires so the state flip surfaces live. /// surfaces live.
#[utoipa::path( #[utoipa::path(
post, post,
path = "/api/rebuild-queue/{id}/cancel", path = "/api/rebuild-queue/{id}/cancel",
params(("id" = u64, Path, description = "job-queue node id (a DAG's root cancels the group)")), params(("id" = u64, Path, description = "job-queue node id (a DAG's root cancels the group)")),
responses((status = 200, description = "whether the DAG was cancelled", body = CancelResultBody)), responses((status = 200, description = "request handled; an empty body, whether or not the node was cancelled")),
tag = "rebuild_queue" tag = "rebuild_queue"
)] )]
pub(super) async fn post_rebuild_queue_cancel( pub(super) async fn post_rebuild_queue_cancel(
State(state): State<AppState>, State(state): State<AppState>,
AxumPath(id): AxumPath<u64>, AxumPath(id): AxumPath<u64>,
) -> Response { ) -> StatusCode {
if state.coord.job_queue.cancel(id) { if state.coord.job_queue.cancel(id) {
// Any terminal side effect is the DAG's own spared tail node, which the // Any terminal side effect is the DAG's own spared tail node, which the
// scheduler picks up on its next pass — nothing to fire from here. // scheduler picks up on its next pass — nothing to fire from here.
state.coord.emit_rebuild_queue_snapshot(); state.coord.emit_rebuild_queue_snapshot();
axum::Json(CancelResultBody { cancelled: true }).into_response()
} else {
axum::Json(CancelResultBody { cancelled: false }).into_response()
} }
StatusCode::OK
} }

View file

@ -24,14 +24,6 @@ use super::tombstones::{TombstoneView, build_tombstone_views};
#[allow(clippy::struct_excessive_bools)] #[allow(clippy::struct_excessive_bools)]
#[derive(Serialize)] #[derive(Serialize)]
pub(super) struct StateSnapshot { 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, hostname: String,
/// Current rev of the pinned `hyperhive` flake input, resolved the /// Current rev of the pinned `hyperhive` flake input, resolved the
/// same way `get_agent_meta`'s per-agent `hyperhive_rev` is (see /// same way `get_agent_meta`'s per-agent `hyperhive_rev` is (see
@ -46,15 +38,6 @@ pub(super) struct StateSnapshot {
/// survive after a destroy-without-purge. The operator can re-spawn /// survive after a destroy-without-purge. The operator can re-spawn
/// with the same name to resume, or PURG3 to wipe them. /// with the same name to resume, or PURG3 to wipe them.
tombstones: Vec<TombstoneView>, tombstones: Vec<TombstoneView>,
/// Agents whose FNV-1a hashed web UI port collides with at least
/// one other. Operator resolves by renaming. The dashboard renders
/// a banner at the top listing each cluster.
port_conflicts: Vec<PortConflict>,
/// Whether the hive-forge container is up. When true the dashboard
/// links each container's config into the forge's `agent-configs`
/// repos.
forge_present: bool,
/// Whether `hive-gateway` is in front of this dashboard. Sourced /// Whether `hive-gateway` is in front of this dashboard. Sourced
/// from the `HIVE_GATEWAY_ENABLED` env var, which the c0re NixOS /// from the `HIVE_GATEWAY_ENABLED` env var, which the c0re NixOS
/// module now always sets (the gateway runs unconditionally /// module now always sets (the gateway runs unconditionally
@ -89,13 +72,6 @@ pub(super) struct StateSnapshot {
swarm_name: Option<String>, swarm_name: Option<String>,
} }
#[derive(Serialize)]
struct PortConflict {
port: u16,
/// All agent names sharing this port (sorted, ≥2 entries).
agents: Vec<String>,
}
#[derive(Serialize)] #[derive(Serialize)]
struct TransientView { struct TransientView {
name: String, name: String,
@ -107,8 +83,7 @@ struct TransientView {
/// Cold-load snapshot of the whole dashboard. /// Cold-load snapshot of the whole dashboard.
/// ///
/// Includes the roster, tombstones, job queue, and more. Live /// Includes the roster, tombstones, job queue, and more. Live
/// clients then follow `/api/dashboard/stream` (SSE) for incremental updates /// clients then follow `/api/dashboard/stream` (SSE) for incremental updates.
/// keyed off `seq`.
// `StateSnapshot` is a large tree of nested view types (`ContainerView`, // `StateSnapshot` is a large tree of nested view types (`ContainerView`,
// `TransientView`, ...) with no `ToSchema` anywhere in that // `TransientView`, ...) with no `ToSchema` anywhere in that
// graph; wiring it up is a schema-modelling project of its own, well past // graph; wiring it up is a schema-modelling project of its own, well past
@ -130,14 +105,6 @@ pub(super) async fn api_state(
.unwrap_or("localhost"); .unwrap_or("localhost");
let hostname = host.split(':').next().unwrap_or(host).to_owned(); 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 // Refresh the coordinator's cached container snapshot before
// reading. Cold-load clients then see whatever the latest rescan // reading. Cold-load clients then see whatever the latest rescan
// produced; live clients converge via the matching // produced; live clients converge via the matching
@ -169,18 +136,14 @@ pub(super) async fn api_state(
let transient_snapshot = state.coord.transient_snapshot(); let transient_snapshot = state.coord.transient_snapshot();
let transients = build_transient_views(&containers, &transient_snapshot); let transients = build_transient_views(&containers, &transient_snapshot);
let tombstones = build_tombstone_views(&containers); let tombstones = build_tombstone_views(&containers);
let port_conflicts = build_port_conflicts(&containers);
axum::Json(StateSnapshot { axum::Json(StateSnapshot {
seq,
hostname, hostname,
hyperhive_rev: crate::auto_update::current_flake_rev(&state.coord.hyperhive_flake), hyperhive_rev: crate::auto_update::current_flake_rev(&state.coord.hyperhive_flake),
containers, containers,
transients, transients,
tombstones, tombstones,
port_conflicts,
forge_present: crate::forge::is_present().await,
gateway_enabled: std::env::var_os("HIVE_GATEWAY_ENABLED").is_some_and(|v| { gateway_enabled: std::env::var_os("HIVE_GATEWAY_ENABLED").is_some_and(|v| {
// Accept any truthy string ("1", "true", "yes"): the c0re // Accept any truthy string ("1", "true", "yes"): the c0re
// NixOS module sets the literal "1" (the gateway always runs // NixOS module sets the literal "1" (the gateway always runs
@ -203,29 +166,6 @@ pub(super) async fn api_state(
}) })
} }
/// Group containers by their assigned web UI port; clusters with more
/// than one member are port-hash collisions the operator resolves by
/// renaming an agent.
///
/// Every container's port comes from [`crate::lifecycle::agent_web_port`],
/// which hashes the name into one range for all of them — so the manager
/// is a candidate for a collision exactly like any sub-agent.
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 /// Transient state for agents whose container does NOT yet exist
/// (`Spawning`). Lifecycle ops on existing containers surface as /// (`Spawning`). Lifecycle ops on existing containers surface as
/// `ContainerView.pending` inline; this list only catches pre-creation. /// `ContainerView.pending` inline; this list only catches pre-creation.
@ -253,8 +193,7 @@ fn build_transient_views(
.collect(); .collect();
// The caller rebuilds this map on every poll, so its iteration order // The caller rebuilds this map on every poll, so its iteration order
// differs between renders and the rows would shuffle under an operator // differs between renders and the rows would shuffle under an operator
// watching an operation run. Sorted for the same reason // watching an operation run. Sorted so the order is stable.
// `build_port_conflicts` sorts its clusters.
out.sort_by(|a, b| { out.sort_by(|a, b| {
a.name a.name
.cmp(&b.name) .cmp(&b.name)
@ -355,7 +294,7 @@ pub(super) struct DashboardStreamQuery {
params(DashboardStreamQuery), params(DashboardStreamQuery),
responses( responses(
(status = 200, description = "server-sent event stream; each event's \ (status = 200, description = "server-sent event stream; each event's \
`data` is a JSON-serialised `DashboardEvent` (seq-tagged)", `data` is a JSON-serialised `DashboardEvent`",
body = String, content_type = "text/event-stream"), body = String, content_type = "text/event-stream"),
), ),
tag = "state_snapshot" tag = "state_snapshot"
@ -378,8 +317,7 @@ pub(super) async fn dashboard_stream(
if set.is_empty() { None } else { Some(set) } if set.is_empty() { None } else { Some(set) }
}); });
let stream = BroadcastStream::new(rx).filter_map(move |res| { let stream = BroadcastStream::new(rx).filter_map(move |res| {
// Drop lagged frames. Browsers reconnect; the seq dedupe on // Drop lagged frames. Browsers reconnect and refetch the snapshot.
// reconnect skips any frame already reflected in the snapshot.
let event = res.ok()?; let event = res.ok()?;
if let Some(filter) = kind_filter.as_ref() if let Some(filter) = kind_filter.as_ref()
&& !filter.contains(event.kind_tag()) && !filter.contains(event.kind_tag())
@ -398,7 +336,7 @@ mod tests {
use chrono::{Duration, Utc}; use chrono::{Duration, Utc};
use super::{ContainerView, build_port_conflicts, build_transient_views}; use super::{ContainerView, build_transient_views};
use crate::job_queue::RunningTransient; use crate::job_queue::RunningTransient;
fn rt(agent: &str, label: &str, age_secs: i64) -> RunningTransient { fn rt(agent: &str, label: &str, age_secs: i64) -> RunningTransient {
@ -418,6 +356,18 @@ mod tests {
m m
} }
fn cv(name: &str, port: u16, running: bool) -> ContainerView {
ContainerView {
name: name.to_owned(),
container: format!("h-{name}"),
port,
running,
failed: false,
needs_login: false,
paused: false,
}
}
#[test] #[test]
fn an_agent_whose_container_exists_is_not_transient() { fn an_agent_whose_container_exists_is_not_transient() {
let snap = snapshot(vec![rt("alice", "spawn", 5)]); let snap = snapshot(vec![rt("alice", "spawn", 5)]);
@ -471,92 +421,44 @@ mod tests {
assert!(build_transient_views(&[], &snapshot(vec![])).is_empty()); assert!(build_transient_views(&[], &snapshot(vec![])).is_empty());
} }
fn cv(name: &str, port: u16, running: bool) -> ContainerView { /// The JSON object a cold-loading dashboard receives carries exactly these
ContainerView { /// keys; a field added back must be added here, with a reader.
name: name.to_owned(),
container: format!("h-{name}"),
port,
running,
failed: false,
needs_login: false,
active_model: None,
status_text: None,
status_set_at: None,
paused: false,
}
}
#[test] #[test]
fn distinct_ports_produce_no_conflicts() { fn the_state_snapshot_serialises_exactly_its_read_fields() {
let cs = [cv("alice", 8101, true), cv("bob", 8102, true)]; let snap = super::StateSnapshot {
assert!(build_port_conflicts(&cs).is_empty()); hostname: "h".to_owned(),
} hyperhive_rev: None,
containers: Vec::new(),
#[test] transients: Vec::new(),
fn a_lone_container_does_not_conflict_with_itself() { tombstones: Vec::new(),
assert!(build_port_conflicts(&[cv("alice", 8101, true)]).is_empty()); gateway_enabled: true,
} swarm_public_url: None,
authelia_public_url: None,
#[test] hive_name: None,
fn two_agents_on_one_port_are_reported_together() { swarm_name: None,
let cs = [cv("alice", 8101, true), cv("bob", 8101, true)]; };
let got = build_port_conflicts(&cs); let value = serde_json::to_value(&snap).unwrap();
assert_eq!(got.len(), 1); let mut keys: Vec<&str> = value
assert_eq!(got[0].port, 8101); .as_object()
assert_eq!(got[0].agents, ["alice", "bob"]); .unwrap()
} .keys()
.map(String::as_str)
/// The manager takes its port from the same hash as everyone else, so .collect();
/// it collides like any other name. A doc comment here used to claim it keys.sort_unstable();
/// sat at a fixed 8000 and could not — no such special case exists. assert_eq!(
#[test] keys,
fn the_manager_can_collide_with_a_sub_agent() { [
let cs = [cv("manager", 8101, true), cv("alice", 8101, true)]; "authelia_public_url",
let got = build_port_conflicts(&cs); "containers",
assert_eq!(got.len(), 1); "gateway_enabled",
assert_eq!(got[0].agents, ["alice", "manager"]); "hive_name",
} "hostname",
"hyperhive_rev",
/// A stopped agent keeps its hashed port, so the clash is real the "swarm_name",
/// moment it starts. Reporting it while it is down is the point. "swarm_public_url",
#[test] "tombstones",
fn a_stopped_container_still_claims_its_port() { "transients",
let cs = [cv("alice", 8101, true), cv("bob", 8101, false)]; ]
assert_eq!(build_port_conflicts(&cs).len(), 1); );
}
#[test]
fn names_are_sorted_within_a_conflict() {
let cs = [cv("zoe", 8101, true), cv("adam", 8101, true)];
assert_eq!(build_port_conflicts(&cs)[0].agents, ["adam", "zoe"]);
}
#[test]
fn more_than_two_agents_land_in_one_cluster() {
let cs = [
cv("c", 8101, true),
cv("a", 8101, true),
cv("b", 8101, true),
];
let got = build_port_conflicts(&cs);
assert_eq!(got.len(), 1, "one cluster, not one row per pair");
assert_eq!(got[0].agents, ["a", "b", "c"]);
}
#[test]
fn separate_clusters_are_ordered_by_port() {
let cs = [
cv("d", 8300, true),
cv("c", 8300, true),
cv("b", 8200, true),
cv("a", 8200, true),
];
let ports: Vec<u16> = build_port_conflicts(&cs).iter().map(|p| p.port).collect();
assert_eq!(ports, [8200, 8300]);
}
#[test]
fn no_containers_is_not_a_conflict() {
assert!(build_port_conflicts(&[]).is_empty());
} }
} }

View file

@ -106,7 +106,6 @@ pub(crate) async fn emit_tombstones_snapshot(coord: &Arc<Coordinator>) {
let containers = coord.containers_snapshot().await; let containers = coord.containers_snapshot().await;
let tombstones = build_tombstone_views(&containers); let tombstones = build_tombstone_views(&containers);
coord.emit_dashboard_event(crate::dashboard_events::DashboardEvent::TombstonesChanged { coord.emit_dashboard_event(crate::dashboard_events::DashboardEvent::TombstonesChanged {
seq: coord.next_seq(),
tombstones, tombstones,
}); });
} }

View file

@ -1,7 +1,6 @@
//! Unified dashboard event channel — all near-real-time browser events //! Unified dashboard event channel — all near-real-time browser events
//! flow through `Coordinator.dashboard_events`. Each event carries a //! flow through `Coordinator.dashboard_events`. Design rationale
//! monotonic `seq` for client-side dedupe against `/api/state` snapshots. //! (single channel, broker forwarder isolation):
//! Design rationale (single channel, broker forwarder isolation):
//! `docs/4-implementation/web-ui/shape.md::One unified channel`. //! `docs/4-implementation/web-ui/shape.md::One unified channel`.
use serde::Serialize; use serde::Serialize;
@ -15,7 +14,6 @@ use chrono::{DateTime, Utc};
pub enum DashboardEvent { pub enum DashboardEvent {
/// Broker `Sent` event mirrored onto the dashboard channel. /// Broker `Sent` event mirrored onto the dashboard channel.
Sent { Sent {
seq: u64,
/// Broker row id. Allows the dashboard to track reply threads. /// Broker row id. Allows the dashboard to track reply threads.
id: i64, id: i64,
from: String, from: String,
@ -27,7 +25,6 @@ pub enum DashboardEvent {
}, },
/// Broker `Delivered` event mirrored onto the dashboard channel. /// Broker `Delivered` event mirrored onto the dashboard channel.
Delivered { Delivered {
seq: u64,
/// Broker row id. Allows the dashboard to track reply threads. /// Broker row id. Allows the dashboard to track reply threads.
id: i64, id: i64,
from: String, from: String,
@ -43,7 +40,6 @@ pub enum DashboardEvent {
/// locally from `since_unix` so a slow rebuild's elapsed time /// locally from `since_unix` so a slow rebuild's elapsed time
/// ticks without polling. /// ticks without polling.
TransientSet { TransientSet {
seq: u64,
name: String, name: String,
/// What the pill renders. For queue-driven work this is the running /// What the pill renders. For queue-driven work this is the running
/// node's own wire tag (`"swap"`, `"create"`, `"stop_for_update"`, …) — /// node's own wire tag (`"swap"`, `"create"`, `"stop_for_update"`, …) —
@ -66,7 +62,6 @@ pub enum DashboardEvent {
/// holding two open pills for one agent cannot tell which one cleared, and /// holding two open pills for one agent cannot tell which one cleared, and
/// has to drop both or guess. /// has to drop both or guess.
TransientCleared { TransientCleared {
seq: u64,
name: String, name: String,
transient_kind: String, transient_kind: String,
}, },
@ -81,21 +76,18 @@ pub enum DashboardEvent {
/// last one cached on the coordinator. Mutation sites (lifecycle /// last one cached on the coordinator. Mutation sites (lifecycle
/// endpoints, `actions::destroy`, `crash_watch`'s poll loop) /// endpoints, `actions::destroy`, `crash_watch`'s poll loop)
/// call the rescan after their work lands. /// call the rescan after their work lands.
ContainerStateChanged { seq: u64, container: ContainerView }, ContainerStateChanged { container: ContainerView },
/// A container that was in the previous snapshot is gone. Clients /// A container that was in the previous snapshot is gone. Clients
/// drop the row by name. Fired alongside any /// drop the row by name. Fired alongside any
/// `nixos-container destroy` (operator-driven or otherwise) on the /// `nixos-container destroy` (operator-driven or otherwise) on the
/// next rescan. /// next rescan.
ContainerRemoved { seq: u64, name: String }, ContainerRemoved { name: String },
/// Full snapshot of the tombstones list. Emitted by destroy (with /// Full snapshot of the tombstones list. Emitted by destroy (with
/// or without purge) and purge-tombstone. Snapshot /// or without purge) and purge-tombstone. Snapshot
/// shape (not diff) because the list is tiny (single-digit /// shape (not diff) because the list is tiny (single-digit
/// typical) and recomputing avoids the add/remove races a /// typical) and recomputing avoids the add/remove races a
/// per-row event would have. /// per-row event would have.
TombstonesChanged { TombstonesChanged { tombstones: Vec<TombstoneView> },
seq: u64,
tombstones: Vec<TombstoneView>,
},
/// The job queue changed — **a bare trigger, no payload.** /// The job queue changed — **a bare trigger, no payload.**
/// ///
@ -110,7 +102,7 @@ pub enum DashboardEvent {
/// push-on-change is what every other live surface here does /// push-on-change is what every other live surface here does
/// (transients, container state), and a poll loop would be /// (transients, container state), and a poll loop would be
/// slower for the same information. /// slower for the same information.
RebuildQueueChanged { seq: u64 }, RebuildQueueChanged,
} }
impl DashboardEvent { impl DashboardEvent {
@ -135,7 +127,7 @@ impl DashboardEvent {
DashboardEvent::ContainerRemoved { .. } => "container_removed", DashboardEvent::ContainerRemoved { .. } => "container_removed",
DashboardEvent::TombstonesChanged { .. } => "tombstones_changed", DashboardEvent::TombstonesChanged { .. } => "tombstones_changed",
DashboardEvent::RebuildQueueChanged { .. } => "rebuild_queue_changed", DashboardEvent::RebuildQueueChanged => "rebuild_queue_changed",
} }
} }
} }
@ -163,7 +155,6 @@ mod tests {
fn kind_tag_matches_serde_kind_field() { fn kind_tag_matches_serde_kind_field() {
let samples: Vec<DashboardEvent> = vec![ let samples: Vec<DashboardEvent> = vec![
DashboardEvent::Sent { DashboardEvent::Sent {
seq: 1,
id: 1, id: 1,
from: "a".into(), from: "a".into(),
to: "b".into(), to: "b".into(),
@ -172,7 +163,6 @@ mod tests {
in_reply_to: None, in_reply_to: None,
}, },
DashboardEvent::Delivered { DashboardEvent::Delivered {
seq: 1,
id: 1, id: 1,
from: "a".into(), from: "a".into(),
to: "b".into(), to: "b".into(),
@ -181,25 +171,19 @@ mod tests {
in_reply_to: None, in_reply_to: None,
}, },
DashboardEvent::TransientSet { DashboardEvent::TransientSet {
seq: 1,
name: "x".into(), name: "x".into(),
transient_kind: "rebuilding".into(), transient_kind: "rebuilding".into(),
since_unix: 0, since_unix: 0,
}, },
DashboardEvent::TransientCleared { DashboardEvent::TransientCleared {
seq: 1,
name: "x".into(), name: "x".into(),
transient_kind: "rebuilding".into(), transient_kind: "rebuilding".into(),
}, },
DashboardEvent::ContainerRemoved { DashboardEvent::ContainerRemoved { name: "x".into() },
seq: 1,
name: "x".into(),
},
DashboardEvent::TombstonesChanged { DashboardEvent::TombstonesChanged {
seq: 1,
tombstones: Vec::new(), tombstones: Vec::new(),
}, },
DashboardEvent::RebuildQueueChanged { seq: 1 }, DashboardEvent::RebuildQueueChanged,
]; ];
for ev in samples { for ev in samples {
let v: serde_json::Value = serde_json::to_value(&ev).expect("serialise"); let v: serde_json::Value = serde_json::to_value(&ev).expect("serialise");
@ -210,4 +194,40 @@ mod tests {
assert_eq!(ev.kind_tag(), serde_kind, "kind_tag() drift on {ev:?}"); assert_eq!(ev.kind_tag(), serde_kind, "kind_tag() drift on {ev:?}");
} }
} }
/// Every frame is its `kind` tag plus the variant's own fields, nothing else.
#[test]
fn frames_carry_no_sequence_number() {
let removed = serde_json::to_value(DashboardEvent::ContainerRemoved { name: "x".into() })
.expect("serialise");
let mut keys: Vec<&str> = removed
.as_object()
.unwrap()
.keys()
.map(String::as_str)
.collect();
keys.sort_unstable();
assert_eq!(keys, ["kind", "name"]);
let transient = serde_json::to_value(DashboardEvent::TransientSet {
name: "x".into(),
transient_kind: "rebuilding".into(),
since_unix: 0,
})
.expect("serialise");
let mut keys: Vec<&str> = transient
.as_object()
.unwrap()
.keys()
.map(String::as_str)
.collect();
keys.sort_unstable();
assert_eq!(keys, ["kind", "name", "since_unix", "transient_kind"]);
let trigger = serde_json::to_value(DashboardEvent::RebuildQueueChanged).expect("serialise");
assert_eq!(
trigger,
serde_json::json!({ "kind": "rebuild_queue_changed" })
);
}
} }

Some files were not shown because too many files have changed in this diff Show more