jobq: a generic per-state roll-up, served beside the graph

A consumer that wants "how much is in flight" — a summary line, a badge,
a health check — had to fetch the whole graph and tally it client-side,
on every poll, in every consumer.

`hive_jobq_wire::state_rollup` counts `roots` and their subtrees by
state, straight off a `Graph<N, R>` with **no bound on either
parameter**. A node's state is a scheduler concept, so counting by state
needs to know nothing about what the payload or the resource are;
bounding it like the projection does would make a host implement two
display traits to be allowed to count, which is a requirement about
rendering imposed on arithmetic.

It takes the roots for the same reason `wire_snapshot` does — which
groups are in view is the host's policy, and nothing is ever removed
from a graph — so passing the same set makes the roll-up describe
exactly the graph beside it.

Each entry carries BOTH counts: `nodes` (the whole subtree) and `roots`
(just the group tops). One rebuild is ~7 nodes and 1 root, so a summary
meaning *operations* and one meaning *steps* are different numbers over
the same queue, and picking one here would make this crate decide what
counts as a job — the domain question it exists not to answer. It
reports both structural facts; the viewer chooses.

A pair, not a map: JSON object keys are strings, so a map would spell
the state twice and give the wire no ordering. Every state rides with
its zeros in a fixed order, so a consumer can index positionally and
never handles a missing bucket. Tallying positionally against
`ALL_STATES` means a new upstream `State` fails the exhaustive match in
`state_index` rather than silently landing in an existing bucket.

hive-c0re serves it at `GET /api/jobq/rollup`. The queue-side method is
a call site, not an implementation: it supplies the lock and the same
`visible_roots` as `graph_snapshot`, so the summary cannot describe a
different visible set than the graph it summarises.

hive-c0re/src/dashboard/mod.rs

@@ -206,6 +206,7 @@ pub async fn serve(
         .routes(routes!(state_snapshot::dashboard_stream))
         .routes(routes!(state_snapshot::dashboard_history))
         .routes(routes!(state_snapshot::jobq_graph))
+        .routes(routes!(state_snapshot::jobq_rollup))
         .split_for_parts();
     // Just the JSON, not the UI — Swagger UI itself is nginx-hosted from
     // the nix store (see the module doc comment above `ApiDoc`). `api`
@@ -450,6 +451,7 @@ mod router_build_probe {
             .routes(routes!(build_logs::get_build_log_stream))
             .routes(routes!(state_snapshot::dashboard_stream))
             .routes(routes!(state_snapshot::dashboard_history))
-            .routes(routes!(state_snapshot::jobq_graph));
+            .routes(routes!(state_snapshot::jobq_graph))
+            .routes(routes!(state_snapshot::jobq_rollup));
     }
 }

hive-c0re/src/dashboard/state_snapshot.rs

@@ -678,6 +678,28 @@ pub(super) async fn jobq_graph(
     axum::Json(state.coord.job_queue.graph_snapshot())
 }
 
+#[utoipa::path(
+    get,
+    path = "/api/jobq/rollup",
+    responses(
+        (status = 200, description = "counts by lifecycle state over the same \
+            groups `/api/jobq/graph` serves, as `(state, nodes, roots)` \
+            triples. Every state is present, zero counts included, in a fixed \
+            order — a consumer renders a summary (\"3 running · 2 queued\") \
+            without fetching the graph and without re-deriving the tally. \
+            `roots` counts groups, `nodes` counts every step at any depth: one \
+            rebuild is 1 root and ~7 nodes, so a summary meaning *operations* \
+            reads `roots` and one meaning *steps* reads `nodes`.",
+            body = Vec<hive_jobq_wire::StateCount>),
+    ),
+    tag = "state_snapshot"
+)]
+pub(super) async fn jobq_rollup(
+    State(state): State<AppState>,
+) -> axum::Json<Vec<hive_jobq_wire::StateCount>> {
+    axum::Json(state.coord.job_queue.state_rollup())
+}
+
 #[utoipa::path(
     get,
     path = "/api/dashboard/history",

hive-c0re/src/job_queue/mod.rs

@@ -368,6 +368,18 @@ impl JobQueue {
         inner.graph().wire_snapshot(visible_roots(&inner))
     }
 
+    /// Per-state counts over the **same** groups [`Queue::graph_snapshot`]
+    /// serves.
+    ///
+    /// Supplies the same two things and nothing else: the lock, and
+    /// [`visible_roots`]. The counting is [`hive_jobq_wire::state_rollup`]'s and
+    /// is generic over the payload — this is a call site, not an implementation.
+    #[must_use]
+    pub fn state_rollup(&self) -> Vec<hive_jobq_wire::StateCount> {
+        let inner = self.lock();
+        hive_jobq_wire::state_rollup(inner.graph(), visible_roots(&inner))
+    }
+
     /// Snapshot every live + retained DAG for `/api/state` + `RebuildQueueChanged`.
     #[must_use]
     pub fn snapshot(&self) -> Vec<DagView> {

hive-jobq-wire/src/lib.rs

@@ -169,6 +169,49 @@ pub struct GraphNode {
     pub payload: NodePayload,
 }
 
+/// How much of the graph is in one [`State`], counted two ways.
+///
+/// A pair, not a map: JSON object keys are strings, so a map would spell the
+/// state twice — once as the key and once in whatever the consumer parses it
+/// back into — and give the wire no ordering. A list of pairs keeps the state a
+/// state.
+///
+/// **Both counts, because "how many things are running" has two honest answers
+/// and which one a viewer wants is its own business.** One rebuild is ~7
+/// `nodes` and 1 `roots`; a summary line that means *operations* wants the
+/// latter, a progress bar over steps wants the former. Reporting one would make
+/// this crate decide what counts as a job — the domain question it exists not
+/// to answer — and reporting both costs a `u64`.
+#[derive(Debug, Clone, Copy, Serialize, Deserialize, ToSchema)]
+pub struct StateCount {
+    #[schema(value_type = StateSchema)]
+    pub state: State,
+    /// Every node in this state, at any depth.
+    pub nodes: u64,
+    /// Only the roots the caller asked for, in this state.
+    ///
+    /// A root's own state is already its subtree's answer (a group root does
+    /// not reach `Done` before its children), so this is "how many of the
+    /// visible groups are in this state" without the caller re-deriving it.
+    pub roots: u64,
+}
+
+/// Every [`State`], in the order [`ALL_STATES`] declares, so a consumer can
+/// index the roll-up positionally and never has to handle a missing key.
+///
+/// Zero counts are **included** on purpose: a summary that renders "3 running"
+/// needs to know the other buckets are empty rather than absent, and the whole
+/// vector is seven entries regardless of graph size.
+const ALL_STATES: [State; 7] = [
+    State::Pending,
+    State::Running,
+    State::Finishing,
+    State::Done,
+    State::Failed,
+    State::Cancelled,
+    State::Skipped,
+];
+
 /// What a node *is*, in terms the graph layer does not interpret.
 #[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
 pub struct NodePayload {
@@ -250,6 +293,67 @@ impl<N: WireNode, R: WireResource> GraphWire for Graph<N, R> {
     }
 }
 
+/// Count `roots` and their subtrees by [`State`], straight off the graph.
+///
+/// For a consumer that wants "how much is in flight" without carrying the
+/// graph: a summary line, a badge, a health check. Every consumer re-deriving
+/// the same tally is the thing this replaces.
+///
+/// **Free of `WireNode`/`WireResource`, unlike [`GraphWire`]** — a node's state
+/// is a scheduler concept, so counting by state needs to know nothing about
+/// what the payload or the resource *are*. Bounding it like the projection
+/// would make a host implement two display traits to be allowed to count, which
+/// is a requirement about rendering imposed on arithmetic.
+///
+/// Takes the same `roots` as [`GraphWire::wire_snapshot`] and for the same
+/// reason — *which* groups are in view is the host's policy — so passing the
+/// same set makes the roll-up describe exactly the graph beside it. Shares the
+/// caveat too: overlapping roots double-count.
+#[must_use]
+pub fn state_rollup<N, R>(
+    graph: &Graph<N, R>,
+    roots: impl IntoIterator<Item = NodeId>,
+) -> Vec<StateCount> {
+    // Tallied positionally against `ALL_STATES` rather than into a map: the
+    // output order is then the declared one for free, and adding a `State`
+    // upstream fails the exhaustive match in `state_index` instead of silently
+    // dropping a bucket.
+    let mut nodes = [0u64; ALL_STATES.len()];
+    let mut roots_by_state = [0u64; ALL_STATES.len()];
+    for root in roots {
+        if let Some(node) = graph.node(root) {
+            roots_by_state[state_index(node.state)] += 1;
+        }
+        for node in graph.node(root).into_iter().chain(graph.descendants(root)) {
+            nodes[state_index(node.state)] += 1;
+        }
+    }
+    ALL_STATES
+        .iter()
+        .zip(nodes)
+        .zip(roots_by_state)
+        .map(|((&state, nodes), roots)| StateCount {
+            state,
+            nodes,
+            roots,
+        })
+        .collect()
+}
+
+/// Position of `state` in [`ALL_STATES`]. Exhaustive on purpose — a new
+/// upstream variant must not silently land in an existing bucket.
+fn state_index(state: State) -> usize {
+    match state {
+        State::Pending => 0,
+        State::Running => 1,
+        State::Finishing => 2,
+        State::Done => 3,
+        State::Failed => 4,
+        State::Cancelled => 5,
+        State::Skipped => 6,
+    }
+}
+
 fn wire_node<N: WireNode, R: WireResource>(node: &hive_jobq::Node<N, R>) -> GraphNode {
     let id = node.id.get();
     GraphNode {
@@ -295,8 +399,8 @@ mod tests {
     use hive_jobq::scheduler::Scheduler;
 
     use super::{
-        BitFlags, GraphDep, GraphNode, GraphWire, NodePayload, State, StateSchema, TerminalState,
+        ALL_STATES, BitFlags, GraphDep, GraphNode, GraphWire, NodePayload, State, StateSchema,
-        TerminalStateSchema, WireId, WireNode,
+        TerminalState, TerminalStateSchema, WireId, WireNode, state_rollup,
     };
 
     /// The mirrors document what the wire actually says — a mirror that
@@ -534,4 +638,56 @@ mod tests {
         assert!(!json.contains("\"data\""), "null data is skipped: {json}");
         assert!(json.contains("\"label\""), "the label always rides: {json}");
     }
+
+    /// The roll-up counts **the same visible set** the snapshot serialises,
+    /// both ways, and always emits every state.
+    ///
+    /// Four assertions in one because they are the same promise from four
+    /// sides: a consumer indexes the vector positionally (fixed order), never
+    /// handles a missing bucket (zeros present), gets a node tally that
+    /// includes the group root like any other node (3, not 1), and gets a root
+    /// tally that is the *group* count for the same states (1, not 3) — the two
+    /// numbers this endpoint exists to stop consumers from confusing.
+    #[test]
+    fn the_rollup_counts_nodes_and_roots_and_names_every_state() {
+        let mut sched = sched();
+        let shown = sched
+            .insert_job(None, |job| {
+                let root = job.node("dag");
+                let child = job.node("prebuild").part_of(root);
+                let _grandchild = job.node("swap").part_of(child);
+                vec![root.guid()]
+            })
+            .expect("job inserts");
+        // A second group deliberately left out of `roots` — the roll-up is
+        // bounded by the caller exactly as `wire_snapshot` is, so an unshown
+        // group must not leak into the counts.
+        let _hidden = sched
+            .insert_job(None, |job| vec![job.node("other-dag").guid()])
+            .expect("job inserts");
+
+        let rollup = state_rollup(sched.graph(), shown.iter().copied());
+
+        assert_eq!(
+            rollup.iter().map(|c| c.state).collect::<Vec<_>>(),
+            ALL_STATES.to_vec(),
+            "every state rides, in the declared order, so a consumer can index it"
+        );
+        assert_eq!(
+            rollup.iter().map(|c| c.nodes).sum::<u64>(),
+            3,
+            "root + child + grandchild — every node, and the hidden group is excluded"
+        );
+        assert_eq!(
+            rollup.iter().map(|c| c.roots).sum::<u64>(),
+            1,
+            "one group, counted once — the same subtree the node tally reads as 3"
+        );
+        let pending = rollup
+            .iter()
+            .find(|c| matches!(c.state, State::Pending))
+            .expect("Pending is always present");
+        assert_eq!(pending.nodes, 3, "nothing has run yet");
+        assert_eq!(pending.roots, 1, "and its group is pending with it");
+    }
 }