job_queue: filter jobq graph snapshot by per-node state

graph_snapshot previously filtered which whole roots got projected
based on the root node's own state, so a group root that was still
Running but had already-Done internal steps couldn't be filtered
down to just its live nodes, and a filtered-out root hid its entire
subtree even when a descendant still matched.

Apply the states filter after GraphWire::wire_snapshot instead, over
every node in the flattened tree, not just roots. The jobq-graph
client already handles an orphaned node (parent filtered out) by
promoting it to a rendered root, so this is safe on the client side
with no changes needed there.

Fixes hyperhive#3210

hive-c0re/src/job_queue/mod.rs

@@ -333,25 +333,32 @@ impl JobQueue {
     /// 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.
+    /// `states`, when given, keeps every **individual node** (root or
+    /// descendant) whose own state is named — not just whole root groups.
+    /// A still-live group (root not yet terminal) can otherwise hold any
+    /// number of already-finished steps inside it; filtering only at the
+    /// root leaves every one of those visible regardless of the ask, which
+    /// is exactly the clutter a state filter exists to remove. `None` (or
+    /// the full state set) is the unfiltered call, matching prior
+    /// behaviour. A slice rather than a set: `State` derives `Eq` but not
+    /// `Hash`, and the vocabulary is 7 variants — a linear check per node
+    /// costs nothing at that size.
+    ///
+    /// A node whose *parent* got filtered out still rides with its original
+    /// `parent` id — `<hive-jobq-graph>` (the one consumer) already treats
+    /// an unresolvable parent as a new root (`buildTree`'s fallback), so a
+    /// filtered-out ancestor surfaces a still-matching descendant one level
+    /// higher rather than hiding or orphaning it.
     ///
     /// 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.
+    /// is *which* nodes to show — see [`visible_roots`] for why the graph
+    /// can't decide the root-visibility half of that for itself.
     #[must_use]
     pub fn graph_snapshot(&self, states: Option<&[State]>) -> Vec<GraphNode> {
         let inner = self.lock();
         let roots = visible_roots(&inner);
-        let roots = filter_roots_by_state(&inner, roots, states);
-        inner.graph().wire_snapshot(roots)
+        let nodes = inner.graph().wire_snapshot(roots);
+        filter_nodes_by_state(nodes, states)
     }
 
     /// Per-state counts over the **same** groups [`Queue::graph_snapshot`]
@@ -406,30 +413,23 @@ 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.
+/// [`Queue::graph_snapshot`]'s `states` ask, applied to the already-
+/// projected node list: keeps every node — root or descendant — 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> {
+/// Applied *after* [`GraphWire::wire_snapshot`] rather than as a root
+/// pre-filter — narrowing which roots are visible at all is
+/// [`visible_roots`]'s job (a different question: how much settled work is
+/// retained, full stop); this is "of what's retained and live, which
+/// individual nodes does the caller want shown right now." `None` (or an
+/// unrecognised/absent query) is the identity filter.
+fn filter_nodes_by_state(nodes: Vec<GraphNode>, states: Option<&[State]>) -> Vec<GraphNode> {
     let Some(states) = states else {
-        return roots;
+        return nodes;
     };
-    roots
+    nodes
         .into_iter()
-        .filter(|&id| {
-            sched
-                .graph()
-                .node(id)
-                .is_some_and(|n| states.contains(&n.state))
-        })
+        .filter(|n| states.contains(&n.state))
         .collect()
 }
 

hive-c0re/src/job_queue/tests.rs

@@ -1471,11 +1471,18 @@ 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).
+/// `graph_snapshot`'s `states` ask, exercised through the real scheduler:
+/// a queued (never-run) DAG cancels as one unit, so every node in it shares
+/// one state and this only proves root-level inclusion/exclusion — the
+/// per-node case (a live group holding a mix of finished and unfinished
+/// steps) isn't expressible without driving the scheduler, which this test
+/// module deliberately can't do (see the module doc comment). See
+/// `filter_nodes_by_state_keeps_matching_nodes_from_a_mixed_state_tree`
+/// below for that half, exercised directly against hand-built `GraphNode`s.
+/// `None` is the identity filter (every visible node, current default
+/// behaviour).
 #[test]
-fn graph_snapshot_states_filters_whole_groups_by_root_state() {
+fn graph_snapshot_states_filters_by_node_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));
@@ -1520,6 +1527,60 @@ fn graph_snapshot_states_filters_whole_groups_by_root_state() {
     );
 }
 
+/// One hand-built node, bypassing the scheduler entirely — `filter_nodes_by_state`
+/// is a pure `Vec<GraphNode> -> Vec<GraphNode>` transform, so this exercises it
+/// directly rather than trying (and failing, per this module's own rule) to
+/// drive a live group into a genuinely mixed state through the real queue.
+fn node(id: u64, parent: Option<u64>, state: State) -> GraphNode {
+    GraphNode {
+        id,
+        parent,
+        state,
+        deps: Vec::new(),
+        created_at: Utc::now(),
+        started_at: None,
+        finished_at: None,
+        error: None,
+        payload: hive_jobq_wire::NodePayload {
+            label: id.to_string(),
+            data: serde_json::Value::Null,
+        },
+    }
+}
+
+/// The case `graph_snapshot_states_filters_by_node_state` above can't reach:
+/// a still-live group (root `Running`) holding a mix of already-`Done` and
+/// still-`Pending` steps. Filtering out `Done` must drop exactly the `Done`
+/// node and nothing else — the root and the `Pending` sibling both stay,
+/// even though the root itself isn't in the requested state set.
+#[test]
+fn filter_nodes_by_state_keeps_matching_nodes_from_a_mixed_state_tree() {
+    let nodes = vec![
+        node(1, None, State::Running),    // root: whole group still live
+        node(2, Some(1), State::Done),    // finished step, should be hidden
+        node(3, Some(1), State::Pending), // not-yet-run step, should stay
+    ];
+
+    let mut kept: Vec<u64> =
+        filter_nodes_by_state(nodes.clone(), Some(&[State::Running, State::Pending]))
+            .into_iter()
+            .map(|n| n.id)
+            .collect();
+    kept.sort_unstable();
+    assert_eq!(
+        kept,
+        vec![1, 3],
+        "Done is filtered out even though its still-live parent (root) isn't"
+    );
+
+    let mut unfiltered: Vec<u64> = filter_nodes_by_state(nodes, None)
+        .into_iter()
+        .map(|n| n.id)
+        .collect();
+    unfiltered.sort_unstable();
+    assert_eq!(unfiltered, vec![1, 2, 3], "None is the identity filter");
+}
+
 #[test]
 fn error_truncation_cuts_on_a_char_boundary() {
     // `truncate_error` is a pure `&str -> String`. This used to submit a DAG,