fix(#2802): a group whose children were all dropped rolls up Cancelled

The roll-up treated a cancelled child the same as a failed one, so a DAG
the operator cancelled before it started reported `Failed` — it claimed to
have failed at something when nothing under it ever ran.

`Failed` still outranks `Cancelled`: a group where one step broke and the
rest were dropped in response is a failure, and that is the fact worth
surfacing. Only a group with no failed child at all reports the cancel.

This also settles a disagreement. `DagView::rollup_state` on the wire has
always ranked failed over cancelled over the rest; the graph's own roll-up
had no `Cancelled` outcome to rank, so the two described the same DAG
differently depending on which one you asked.

hive-c0re/src/job_queue/mod.rs

@@ -104,8 +104,8 @@ struct DagMeta {
 /// container node** ([`NodeKind::Dag`], `parent = None`) whose subtree is the
 /// DAG's work — so the container's `NodeId` is the DAG id, its rolled-up state
 /// is the DAG state, and there are no grouping side-tables: membership + meta
-/// are graph queries ([`QueueInner::container`] / [`QueueInner::subtree`] /
-/// [`QueueInner::dag_meta`]). One shared crate [`Graph`] holds every DAG.
+/// are graph queries ([`QueueInner::container`] / [`QueueInner::dag_meta`] +
+/// the `hive_jobq::Graph` accessors). One shared crate [`Graph`] holds every DAG.
 struct QueueInner {
     sched: Scheduler<NodeKind, Resource>,
     /// Per-node runtime metadata (the build-log id) — mutable after
@@ -309,7 +309,7 @@ impl JobQueue {
             };
             let kind = node.payload.clone();
             let agent = node.payload.agent().to_owned();
-            let Some(container) = inner.dag_of(id) else {
+            let Some(container) = inner.sched.graph().root_of(id) else {
                 continue;
             };
             let Some(meta) = inner.dag_meta(container) else {
@@ -371,30 +371,9 @@ impl JobQueue {
         let Some(container) = inner.container(dag_id) else {
             return false;
         };
-        let work = inner.subtree(container);
-        let all_pending = work.iter().all(|&id| {
-            inner
-                .sched
-                .graph()
-                .node(id)
-                .is_some_and(|n| n.state == JobState::Pending)
-        });
-        if !all_pending {
+        if !inner.sched.cancel_node(container) {
             return false;
         }
-        for id in work {
-            if inner.observes_cancellation(id) {
-                continue;
-            }
-            inner.sched.cancel_node(id);
-        }
-        // Re-run the container's roll-up now that its children are `Cancelled`.
-        // With a spared tail still `Pending` this is a deliberate no-op — the
-        // container has a non-terminal child, so `settle_terminal` parks it back
-        // in `Finishing` and it rolls up for real once the tail finishes. With no
-        // tail (a power op) every child *is* terminal, so it settles synchronously
-        // here exactly as before.
-        inner.sched.complete(container, Outcome::Done);
         drop(inner);
         self.notify.notify_one();
         true
@@ -403,7 +382,9 @@ impl JobQueue {
     /// Link a `build_logs` row to a specific `Running` node.
     pub fn set_build_log_id(&self, dag_id: u64, node_id: NodeId, log_id: i64) -> bool {
         let mut inner = self.lock();
-        if inner.dag_of(node_id).map(NodeId::get) != Some(dag_id) || !inner.node_running(node_id) {
+        if inner.sched.graph().root_of(node_id).map(NodeId::get) != Some(dag_id)
+            || !inner.node_running(node_id)
+        {
             return false;
         }
         inner.node_rt.entry(node_id).or_default().build_log_id = Some(log_id);
@@ -438,7 +419,11 @@ impl JobQueue {
     pub fn first_error(&self, dag_id: u64) -> Option<String> {
         let inner = self.lock();
         let container = inner.container(dag_id)?;
-        inner.dag_first_error(container)
+        inner
+            .sched
+            .graph()
+            .first_error(container)
+            .map(ToOwned::to_owned)
     }
 
     /// The `(dag_id, agent, kind)` triples for every per-agent lease currently
@@ -456,7 +441,7 @@ impl JobQueue {
                 let Resource::Agent(agent) = res else {
                     return None;
                 };
-                let container = inner.dag_of(holder)?;
+                let container = inner.sched.graph().root_of(holder)?;
                 let kind = inner.dag_meta(container)?.transient?;
                 Some((container.get(), agent, kind))
             })
@@ -505,21 +490,6 @@ impl QueueInner {
         })
     }
 
-    /// The DAG container a node belongs to. A container is exactly a group root,
-    /// so this is the graph's own parent-chain walk.
-    fn dag_of(&self, id: NodeId) -> Option<NodeId> {
-        self.sched.graph().root_of(id)
-    }
-
-    /// The DAG's work nodes — its `container`'s subtree, excluding the container.
-    fn subtree(&self, container: NodeId) -> Vec<NodeId> {
-        self.sched
-            .graph()
-            .descendants(container)
-            .map(|n| n.id)
-            .collect()
-    }
-
     /// The container's carried domain metadata as an owned read-view. The data
     /// lives solely in the [`NodeKind::Dag`] payload — this is a derived read,
     /// not a stored side-table.
@@ -545,26 +515,6 @@ impl QueueInner {
         })
     }
 
-    /// Whether `id` has an edge that accepts a **dropped** dependency — i.e. the
-    /// node exists to report on work that may never run. Used by
-    /// [`JobQueue::cancel`] to decide what to spare, so the decision comes from
-    /// the node's own declared edges rather than a hardcoded list of kinds.
-    fn observes_cancellation(&self, id: NodeId) -> bool {
-        self.sched.graph().node(id).is_some_and(|n| {
-            n.deps.iter().any(|d| {
-                matches!(d, Dep::Node { when, .. } if when.accepts(hive_jobq::TerminalState::Cancelled))
-            })
-        })
-    }
-
-    /// First failed work node's error, for the dashboard's DAG-level error line.
-    fn dag_first_error(&self, container: NodeId) -> Option<String> {
-        self.sched
-            .graph()
-            .first_error(container)
-            .map(ToOwned::to_owned)
-    }
-
     /// Project a DAG into its wire [`DagView`]: a near-raw view of the
     /// container's work nodes, with `Done` nodes excluded. Lifecycle
     /// (`state` / `started_at` / `finished_at` / `error`) is read straight
@@ -587,10 +537,8 @@ impl QueueInner {
         // from a `Done`-filtered node set, so the host computes them here.
         let mut started: Vec<DateTime<Utc>> = Vec::new();
         let mut finished: Vec<DateTime<Utc>> = Vec::new();
-        for id in self.subtree(container) {
-            let Some(node) = self.sched.graph().node(id) else {
-                continue;
-            };
+        for node in self.sched.graph().descendants(container) {
+            let id = node.id;
             if let Some(s) = node.started_at {
                 started.push(s);
             }
@@ -668,9 +616,9 @@ impl QueueInner {
     /// subtree (read off the graph `Node`, as unix seconds), for the history
     /// cap ordering.
     fn dag_finished_at(&self, container: NodeId) -> i64 {
-        self.subtree(container)
-            .iter()
-            .filter_map(|id| self.sched.graph().node(*id))
+        self.sched
+            .graph()
+            .descendants(container)
             .filter_map(|n| n.finished_at)
             .map(|t| t.timestamp())
             .max()

hive-jobq/src/scheduler.rs

@@ -32,7 +32,7 @@ use std::collections::HashMap;
 use std::hash::Hash;
 
 use crate::resources::ResourceTable;
-use crate::{Dep, Graph, GraphError, NodeId, State};
+use crate::{Dep, Graph, GraphError, NodeId, State, TerminalState};
 
 /// The result of a node's own execution, reported to [`Scheduler::complete`].
 ///
@@ -233,15 +233,31 @@ impl<N, R: Clone + Eq + Hash> Scheduler<N, R> {
             .all(|n| n.parent != Some(id) || n.state.is_terminal())
     }
 
-    /// Whether any direct child of `id` ended `Failed`/`Cancelled` — the roll-up
-    /// failure condition for the parent.
-    /// `Skipped` children are **not** counted: being ruled out by an edge is the
-    /// expected fate of every branch not taken, so counting it would make any
-    /// group that branches on outcome roll up failed no matter how the run went.
-    fn any_child_failed(&self, id: NodeId) -> bool {
-        self.graph
-            .nodes()
-            .any(|n| n.parent == Some(id) && matches!(n.state, State::Failed | State::Cancelled))
+    /// What `id` rolls up to once all its children are terminal: `Failed` if any
+    /// child failed, else `Cancelled` if any was cancelled, else `Done`.
+    ///
+    /// `Failed` outranks `Cancelled` because the failure is the actionable fact —
+    /// a group where one step broke and the rest were dropped in response is a
+    /// failure, not a cancellation. A group whose children were *all* dropped
+    /// never failed at anything, and says so.
+    ///
+    /// `Skipped` children are ignored: being ruled out by an edge is the expected
+    /// fate of every branch not taken, so counting it would make any group that
+    /// branches on outcome roll up non-`Done` however the run went.
+    fn rolled_up_state(&self, id: NodeId) -> State {
+        let mut any_cancelled = false;
+        for child in self.graph.nodes().filter(|n| n.parent == Some(id)) {
+            match child.state {
+                State::Failed => return State::Failed,
+                State::Cancelled => any_cancelled = true,
+                _ => {}
+            }
+        }
+        if any_cancelled {
+            State::Cancelled
+        } else {
+            State::Done
+        }
     }
 
     /// Transition a node whose own logic just *succeeded* to its resulting state:
@@ -254,12 +270,10 @@ impl<N, R: Clone + Eq + Hash> Scheduler<N, R> {
     /// its dependency succeeds, and leaving it `Pending` would wedge the subtree
     /// non-terminal forever.
     fn settle_terminal(&mut self, id: NodeId) {
-        let state = if !self.all_children_terminal(id) {
-            State::Finishing
-        } else if self.any_child_failed(id) {
-            State::Failed
+        let state = if self.all_children_terminal(id) {
+            self.rolled_up_state(id)
         } else {
-            State::Done
+            State::Finishing
         };
         self.graph.set_state(id, state);
         if state.is_terminal() {
@@ -279,11 +293,7 @@ impl<N, R: Clone + Eq + Hash> Scheduler<N, R> {
             {
                 break;
             }
-            let state = if self.any_child_failed(a) {
-                State::Failed
-            } else {
-                State::Done
-            };
+            let state = self.rolled_up_state(a);
             self.graph.set_state(a, state);
             // Any terminal outcome can rule a dependent out — see `settle_terminal`.
             self.cascade_cancel(a);
@@ -291,25 +301,65 @@ impl<N, R: Clone + Eq + Hash> Scheduler<N, R> {
         }
     }
 
-    /// Cancel a still-*pending* node (and cascade to the dependents it rules out):
-    /// mark it [`State::Cancelled`] and report whether it was cancellable. A
-    /// node that has already started (`Running`) or finished is left untouched —
-    /// an in-flight node's work is not interruptible. A pending node holds no
-    /// resources, so nothing is released here; call [`Scheduler::settle`]
-    /// afterwards to let now-terminal dependents advance (e.g. a weak-edge
-    /// terminal node observing the cancellation).
+    /// Cancel the not-yet-started work at `id`: mark it [`State::Cancelled`]
+    /// and cascade to the dependents that rules out. Cancelling a node cancels
+    /// what hangs under it — a group is abandoned by abandoning its root.
+    /// Reports whether anything was cancelled.
+    ///
+    /// **All-or-nothing.** A subtree with any node already `Running` or terminal
+    /// is left completely untouched: an in-flight node's work is not
+    /// interruptible, and cancelling only the pending half would leave the group
+    /// half-executed with no way to finish it.
+    ///
+    /// **A node whose edge accepts [`TerminalState::Cancelled`] is spared.** Such
+    /// a node is asking to run precisely when the work it follows is dropped,
+    /// which is what lets a reporting tail settle whatever it reports to instead
+    /// of leaving it dangling forever. Nothing is special-cased by payload — the
+    /// node's own declared edges decide. Note `DepWhen::AFTER_ANY` deliberately
+    /// does *not* accept `Cancelled`, so an ordinary weak-edged step is cancelled
+    /// along with the rest; there is nothing to do when no node ever ran.
+    ///
+    /// Cancelled nodes were pending, so they hold no resources and none are
+    /// released here.
     pub fn cancel_node(&mut self, id: NodeId) -> bool {
-        if self
-            .graph
-            .node(id)
-            .is_some_and(|n| n.state == State::Pending)
-        {
-            self.graph.set_state(id, State::Cancelled);
-            self.cascade_cancel(id);
-            true
-        } else {
-            false
+        // The unit is the work *under* `id`; a node with no children is its own
+        // work. A group root's state is its subtree's roll-up rather than a step
+        // that ran, so the root itself is not part of the gate.
+        let under: Vec<NodeId> = self.graph.descendants(id).map(|n| n.id).collect();
+        let is_group = !under.is_empty();
+        let targets = if is_group { under } else { vec![id] };
+        if !targets.iter().all(|&n| {
+            self.graph
+                .node(n)
+                .is_some_and(|n| n.state == State::Pending)
+        }) {
+            return false;
         }
+        for n in targets {
+            if self.observes_cancellation(n) {
+                continue;
+            }
+            self.graph.set_state(n, State::Cancelled);
+            self.cascade_cancel(n);
+        }
+        if is_group {
+            // Re-run the root's roll-up now its children are terminal. With a
+            // spared node still pending this is a deliberate no-op: the root
+            // still has a non-terminal child, so it parks back in `Finishing`
+            // and rolls up for real once that node finishes.
+            self.complete(id, Outcome::Done);
+        }
+        true
+    }
+
+    /// Whether `id` has an edge that accepts a **dropped** dependency — i.e. it
+    /// exists to report on work that may never run.
+    fn observes_cancellation(&self, id: NodeId) -> bool {
+        self.graph.node(id).is_some_and(|n| {
+            n.deps.iter().any(
+                |d| matches!(d, Dep::Node { when, .. } if when.accepts(TerminalState::Cancelled)),
+            )
+        })
     }
 
     /// Snapshot the currently-held grants as `(resource, owner)` pairs — one
@@ -924,6 +974,82 @@ mod tests {
         assert_eq!(s.graph().node(c).unwrap().state, State::Running);
     }
 
+    /// Cancelling a group root abandons the work under it. The root's own state
+    /// is a roll-up (it parks in `Finishing`), never `Pending`, so gating on the
+    /// root instead of its children would refuse every group.
+    #[test]
+    fn cancel_node_cancels_the_work_under_a_group_root() {
+        let mut s: Scheduler<&str, String> = Scheduler::new(Graph::new(), ResourceTable::new());
+        let root = s.append("root", vec![], None).expect("root");
+        let a = s.append("a", vec![], Some(root)).expect("a");
+        let b = s.append("b", vec![after_ok(a)], Some(root)).expect("b");
+        // The root runs first and parks in `Finishing` while its children are
+        // outstanding — the state a group root is actually in when cancelled.
+        assert_eq!(s.settle(), vec![root]);
+        s.complete(root, Outcome::Done);
+        assert_eq!(s.graph().node(root).unwrap().state, State::Finishing);
+
+        assert!(s.cancel_node(root));
+        assert_eq!(s.graph().node(a).unwrap().state, State::Cancelled);
+        assert_eq!(s.graph().node(b).unwrap().state, State::Cancelled);
+        // With every child terminal the root leaves `Finishing` and rolls up —
+        // as `Cancelled`, not `Failed`: nothing under it failed at anything, the
+        // work was dropped.
+        assert_eq!(s.graph().node(root).unwrap().state, State::Cancelled);
+    }
+
+    /// All-or-nothing: an in-flight node's work is not interruptible, and
+    /// cancelling only the pending half would strand the group half-executed.
+    #[test]
+    fn cancel_node_refuses_a_group_with_anything_running() {
+        let mut s: Scheduler<&str, String> = Scheduler::new(Graph::new(), ResourceTable::new());
+        let root = s.append("root", vec![], None).expect("root");
+        let a = s.append("a", vec![], Some(root)).expect("a");
+        let b = s.append("b", vec![after_ok(a)], Some(root)).expect("b");
+        assert_eq!(s.settle(), vec![root]);
+        s.complete(root, Outcome::Done);
+        assert_eq!(s.settle(), vec![a], "a is claimed and running");
+
+        assert!(!s.cancel_node(root), "refused while a runs");
+        assert_eq!(s.graph().node(a).unwrap().state, State::Running);
+        assert_eq!(
+            s.graph().node(b).unwrap().state,
+            State::Pending,
+            "the pending half is left alone too — nothing partial"
+        );
+    }
+
+    /// A node whose edge accepts `Cancelled` asked to run when the work it
+    /// follows is dropped. Sparing it is what lets a reporting tail settle the
+    /// thing it reports to instead of leaving it dangling forever.
+    #[test]
+    fn cancel_node_spares_a_node_that_observes_cancellation() {
+        let mut s: Scheduler<&str, String> = Scheduler::new(Graph::new(), ResourceTable::new());
+        let root = s.append("root", vec![], None).expect("root");
+        let work = s.append("work", vec![], Some(root)).expect("work");
+        let tail = s
+            .append(
+                "tail",
+                vec![Dep::Node {
+                    id: work,
+                    when: DepWhen::of(&[TerminalState::Cancelled]),
+                }],
+                Some(root),
+            )
+            .expect("tail");
+        assert_eq!(s.settle(), vec![root]);
+        s.complete(root, Outcome::Done);
+
+        assert!(s.cancel_node(root));
+        assert_eq!(s.graph().node(work).unwrap().state, State::Cancelled);
+        assert_eq!(
+            s.graph().node(tail).unwrap().state,
+            State::Pending,
+            "spared, and now runnable since its dep is Cancelled"
+        );
+        assert_eq!(s.settle(), vec![tail], "the tail still gets to report");
+    }
+
     #[test]
     fn resource_state_reports_owners() {
         let mut s = scheduler_with_slots(1);