jobq graph: state filter on /api/jobq/graph + multi-select checkboxes

GET /api/jobq/graph gains a states query param (comma-separated
hive_jobq::State names): narrows the served root groups to the named
states, keeping a group whole (filtering by a root's own state, which
is already its subtree's rolled-up answer). Absent, empty, or fully
unrecognised is the identity filter, matching prior behaviour.

hive-jobq-graph.js gains a row of per-state checkboxes above the tree,
re-fetching the endpoint with the selection on toggle. Default
selection hides Done and Skipped.

Server-side filtering (not client-side hiding) so hive-jobq-graph-update's
node list, and everything downstream of it in builds.js (count pill,
live-log panel), only ever sees what's actually shown.

hive-c0re/src/dashboard/state_snapshot.rs

@@ -651,24 +651,57 @@ fn build_approval_views(approvals: Vec<Approval>) -> Vec<ApprovalView> {
     out
 }
 
+/// `/api/jobq/graph` query string. Today's only field is `states`: a
+/// comma-separated allow-list of `hive_jobq::State` names (`"Pending"`,
+/// `"Running"`, ...). Empty / absent ⇒ no filter (current behaviour, every
+/// visible root). Set ⇒ only **root** groups whose own state is named are
+/// served — a root's state is already its subtree's rolled-up answer (see
+/// `hive_jobq_wire`'s doc), so filtering the root filters the whole group.
+/// Unknown tokens are silently ignored (an unrecognised name matches
+/// nothing rather than erroring the whole request), mirroring
+/// `DashboardStreamQuery::kinds` above.
+#[derive(Deserialize, Default, IntoParams)]
+pub(super) struct JobqGraphQuery {
+    states: Option<String>,
+}
+
+/// Parses [`JobqGraphQuery::states`] into the list
+/// [`crate::job_queue::JobQueue::graph_snapshot`] wants. `None` when absent
+/// or when every token failed to parse — both mean "no filter" rather than
+/// "match nothing", so an empty/garbled query reads as the unfiltered call
+/// it replaces rather than an empty result set.
+fn parse_states(raw: Option<&str>) -> Option<Vec<hive_jobq::State>> {
+    let states: Vec<hive_jobq::State> = raw?
+        .split(',')
+        .map(str::trim)
+        .filter(|s| !s.is_empty())
+        .filter_map(|s| serde_json::from_value(serde_json::Value::String(s.to_owned())).ok())
+        .collect();
+    (!states.is_empty()).then_some(states)
+}
+
 #[utoipa::path(
     get,
     path = "/api/jobq/graph",
+    params(JobqGraphQuery),
     responses(
         (status = 200, description = "every node of every retained job group, \
             as generic `hive_jobq` graph nodes: identity, the parent tree, \
             dependency edges with their accepted-outcome sets, lifecycle, and \
             one opaque per-node payload. Group roots ride as ordinary nodes \
-            (`parent: null`) and `Done` nodes are not filtered — a consumer \
-            renders the graph without knowing what any node means.",
+            (`parent: null`) and `Done` nodes are not filtered by default — a \
+            consumer renders the graph without knowing what any node means. \
+            `?states=` narrows to root groups in the named states.",
             body = Vec<hive_jobq_wire::GraphNode>),
     ),
     tag = "state_snapshot"
 )]
 pub(super) async fn jobq_graph(
     State(state): State<AppState>,
+    axum::extract::Query(q): axum::extract::Query<JobqGraphQuery>,
 ) -> axum::Json<Vec<hive_jobq_wire::GraphNode>> {
-    axum::Json(state.coord.job_queue.graph_snapshot())
+    let states = parse_states(q.states.as_deref());
+    axum::Json(state.coord.job_queue.graph_snapshot(states.as_deref()))
 }
 
 #[utoipa::path(

hive-c0re/src/job_queue/mod.rs

@@ -326,19 +326,32 @@ impl JobQueue {
 
     /// Every node of every visible group, as generic graph nodes.
     ///
-    /// **Nothing is hidden.** Group roots ride as ordinary nodes (so a
-    /// consumer needs no special case for "the container" and reads the
-    /// root's own `state` as the group's answer), and `Done` nodes stay (so a
-    /// finished step is visible rather than vanishing from the payload, which
-    /// is what makes a fast rebuild render as a single node).
+    /// **Nothing is hidden**, other than an explicit `states` ask. Group
+    /// roots ride as ordinary nodes (so a consumer needs no special case for
+    /// "the container" and reads the root's own `state` as the group's
+    /// answer), and `Done` nodes stay by default (so a finished step is
+    /// visible rather than vanishing from the payload, which is what makes a
+    /// fast rebuild render as a single node).
+    ///
+    /// `states`, when given, keeps only the **root** groups whose own state
+    /// is named — a root's state is already its subtree's rolled-up answer
+    /// (see `hive_jobq_wire`'s doc), so filtering the root filters the whole
+    /// group without needing to also filter descendants. `None` (or the
+    /// full state set) is the unfiltered call, matching prior behaviour.
+    /// A slice rather than a set: `State` (`hive_jobq`'s) derives `Eq` but
+    /// not `Hash`, and the whole vocabulary is 7 variants — a linear
+    /// membership check per root costs nothing at that size, so there is no
+    /// reason to widen `hive_jobq::State`'s derive list just for this.
     ///
     /// The projection itself is [`hive_jobq_wire`]'s; all this layer supplies
     /// is *which* groups to show — see [`visible_roots`] for why the graph
     /// can't decide that for itself.
     #[must_use]
-    pub fn graph_snapshot(&self) -> Vec<GraphNode> {
+    pub fn graph_snapshot(&self, states: Option<&[State]>) -> Vec<GraphNode> {
         let inner = self.lock();
-        inner.graph().wire_snapshot(visible_roots(&inner))
+        let roots = visible_roots(&inner);
+        let roots = filter_roots_by_state(&inner, roots, states);
+        inner.graph().wire_snapshot(roots)
     }
 
     /// Per-state counts over the **same** groups [`Queue::graph_snapshot`]
@@ -393,6 +406,33 @@ fn find_node(sched: &Sched, id: u64) -> Option<NodeId> {
         .find_map(|n| (n.id.get() == id).then_some(n.id))
 }
 
+/// [`Queue::graph_snapshot`]'s `states` ask, applied to an already-bounded
+/// root set: keeps only the roots whose own `state` is named.
+///
+/// Applied *after* [`visible_roots`] rather than folded into it — the
+/// history cap bounds how much settled work is retained at all, which is a
+/// different question from which of the retained (and live) groups the
+/// caller wants shown right now. `None` (or an unrecognised/absent query)
+/// is the identity filter.
+fn filter_roots_by_state(
+    sched: &Sched,
+    roots: Vec<NodeId>,
+    states: Option<&[State]>,
+) -> Vec<NodeId> {
+    let Some(states) = states else {
+        return roots;
+    };
+    roots
+        .into_iter()
+        .filter(|&id| {
+            sched
+                .graph()
+                .node(id)
+                .is_some_and(|n| states.contains(&n.state))
+        })
+        .collect()
+}
+
 /// The visible **group** set for [`Queue::graph_snapshot`]: every live group
 /// root, plus the newest [`MAX_HISTORY_DAGS`] settled ones.
 ///

hive-c0re/src/job_queue/tests.rs

@@ -306,7 +306,7 @@ fn state_of(q: &JobQueue, dag_id: u64) -> State {
     // rolled-up outcome), so there is nothing to derive here any more.
     // A root aged out of the retained history reads `Done`: it settled, or
     // it would still be live.
-    q.graph_snapshot()
+    q.graph_snapshot(None)
         .iter()
         .find(|n| n.id == dag_id)
         .map_or(State::Done, |n| n.state)
@@ -318,7 +318,7 @@ fn state_of(q: &JobQueue, dag_id: u64) -> State {
 /// asking "how many DAGs" counts the parentless ones. Counting rows would
 /// count steps, which is a different number: one rebuild is ~7 nodes.
 fn dag_count(q: &JobQueue) -> usize {
-    q.graph_snapshot()
+    q.graph_snapshot(None)
         .iter()
         .filter(|n| n.parent.is_none())
         .count()
@@ -1471,6 +1471,55 @@ fn history_retains_live_dags_and_the_newest_terminals() {
     );
 }
 
+/// `graph_snapshot`'s `states` ask filters by **root** state — a whole group
+/// is kept or dropped together, never split mid-subtree. `None` is the
+/// identity filter (every visible root, current default behaviour).
+#[test]
+fn graph_snapshot_states_filters_whole_groups_by_root_state() {
+    let q = JobQueue::new(2);
+    let roots_a = insert_named(&q, |builder| restart_online(builder, &["agent-a"], false));
+    let roots_b = insert_named(&q, |builder| restart_online(builder, &["agent-b"], false));
+    let [head_a] = roots_a.as_slice() else {
+        panic!("a one-agent restart names one root, got {roots_a:?}")
+    };
+    let [head_b] = roots_b.as_slice() else {
+        panic!("a one-agent restart names one root, got {roots_b:?}")
+    };
+
+    assert!(q.cancel(*head_a), "queued dag cancels");
+    assert_eq!(state_of(&q, *head_a), State::Cancelled);
+    assert_eq!(state_of(&q, *head_b), State::Pending, "untouched sibling");
+
+    let roots_of = |states: Option<&[State]>| -> Vec<u64> {
+        q.graph_snapshot(states)
+            .into_iter()
+            .filter(|n| n.parent.is_none())
+            .map(|n| n.id)
+            .collect()
+    };
+
+    assert_eq!(
+        roots_of(Some(&[State::Cancelled])),
+        vec![*head_a],
+        "only the cancelled group's root rides"
+    );
+
+    assert_eq!(
+        roots_of(Some(&[State::Pending])),
+        vec![*head_b],
+        "only the pending group's root rides"
+    );
+
+    let mut both = roots_of(None);
+    both.sort_unstable();
+    let mut expected = vec![*head_a, *head_b];
+    expected.sort_unstable();
+    assert_eq!(
+        both, expected,
+        "no filter shows every visible root, as before"
+    );
+}
+
 #[test]
 fn error_truncation_cuts_on_a_char_boundary() {
     // `truncate_error` is a pure `&str -> String`. This used to submit a DAG,