hyperhive/hive-jobq/src/scheduler.rs

//! The settle loop — drives a [`Graph`] to completion over a resource pool the
//! scheduler owns directly.
//!
//! [`Scheduler::settle`] claims every currently-runnable pending node (its
//! [`Dep::Node`] edges satisfied *and* all its [`Dep::Resource`] units acquired
//! atomically), marks it `Running`, records the units it holds, and returns the
//! newly-started ids for the caller's runner to execute. The runner reports each
//! node's result back with [`Scheduler::complete`]; a running node may grow more
//! work first via [`Scheduler::append`]. Concurrency is emergent from resource
//! capacity — there is no separate active-node cap.
//!
//! Single-threaded by design: the scheduler is the only driver, holds the
//! [`ResourceTable`] as a plain owned field, mutating it through `&mut self` —
//! no interior mutability, no guard objects.
//!
//! Completion rolls up the parent tree: a node with children parks in
//! [`State::Finishing`] until they finish; a child is gated on its parent there.
//!
//! Resource holding follows the [`crate::Node::parent`] tree, not the deps.
//! For each resource a node needs, the scheduler walks its parent-ancestors:
//! no ancestor holds it → acquire fresh units (this node *owns* them, held for
//! its whole subtree); an ancestor owns it but its grant is lent to a different
//! branch → acquire an additional unit if one is free, else wait; an ancestor
//! owns it and the grant is free (or lent to a branch this node is inside) →
//! *borrow* it, no new unit. A grant is lent to one branch at a time; nodes
//! inside a branch are covered by its borrow (dep-sequenced, so no concurrent
//! work under a cap-1 lease). An owner's unit releases only once the owner and
//! its whole subtree are terminal. (Early release once no subtree node needs it
//! is a deferred optimization — unsafe under dynamically-appended subnodes.)

use std::collections::HashMap;
use std::hash::Hash;

use crate::builder::{BuildError, JobBuilder, NodeGuid};
use crate::resources::ResourceTable;
use crate::{Dep, Graph, GraphError, NodeId, State, TerminalState};

/// The result of a node's own execution, reported to [`Scheduler::complete`].
///
/// `Cancelled` is not an outcome a runner reports — it is scheduler-driven (an
/// `AfterOk` dependency failed), so a runner only ever says `Done` or `Failed`.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Outcome {
    /// The node's work succeeded.
    Done,
    /// The node's work failed, carrying the failure reason — recorded on
    /// [`crate::Node::error`] for the failed node itself (a node that rolls up
    /// `Failed` from a child, or is cancelled, carries no error of its own).
    Failed(String),
}

/// Drives a [`Graph`] over an owned resource pool: claim runnable nodes, record
/// the units each *owns*, track which node is currently *borrowing* each grant,
/// and release an owner's grant once its whole subtree is terminal.
pub struct Scheduler<N, R: Clone + Eq + Hash> {
    graph: Graph<N, R>,
    resources: ResourceTable<R>,
    /// Fresh units each owner node acquired: `owner → [(resource, count)]`.
    /// Recorded against the node that *acquired* the units (never a borrower);
    /// released back to the table once the owner and its whole [`Node::parent`]
    /// subtree are terminal.
    owned: HashMap<NodeId, Vec<(R, u32)>>,
    /// Which branch currently borrows a given owner's grant: `(owner, resource)
    /// → branch-root node`. A grant is lent to one branch at a time; nodes
    /// inside that branch are covered by it. Cleared when the branch leaves
    /// (its subtree terminal), freeing the grant for a waiting sibling.
    borrowed: HashMap<(NodeId, R), NodeId>,
}

impl<N, R: Clone + Eq + Hash> Scheduler<N, R> {
    /// A scheduler over `graph` with `resources` as the capacity pool.
    #[must_use]
    pub fn new(graph: Graph<N, R>, resources: ResourceTable<R>) -> Self {
        Self {
            graph,
            resources,
            owned: HashMap::new(),
            borrowed: HashMap::new(),
        }
    }

    /// The graph, for inspection (state, hierarchy, UI rendering).
    #[must_use]
    pub fn graph(&self) -> &Graph<N, R> {
        &self.graph
    }

    /// Append a node under `parent` — e.g. a running node growing more work into
    /// its own subtree. Delegates to [`Graph::insert`]; call [`Scheduler::settle`]
    /// afterwards to start it once it is runnable.
    ///
    /// # Errors
    /// Propagates [`GraphError`] for a dangling dependency or parent id.
    pub fn append(
        &mut self,
        payload: N,
        deps: Vec<Dep<R>>,
        parent: Option<NodeId>,
    ) -> Result<NodeId, GraphError> {
        self.graph.insert(payload, deps, parent)
    }

    /// Insert a whole job under `root_parent`, returning the id each handle's
    /// node was minted as.
    ///
    /// `declare` receives a fresh [`JobBuilder`], names the job's nodes on it,
    /// and returns the handles whose ids it wants back — they come back in
    /// that order. The builder never leaves this call. That is the whole
    /// insertion API — a caller cannot construct a builder, hold one, or
    /// insert one itself, so there is no way to end up with a job-shaped value
    /// being passed around as a spec.
    ///
    /// The one insertion entry point for a job. Nodes go straight into
    /// [`Graph::insert_unchecked`]: [`crate::builder::check_job_shape`] has
    /// already decided every rejection the graph could raise, so re-validating
    /// per node could only report a problem *after* the earlier nodes were
    /// inserted. Call [`Scheduler::settle`] afterwards to start whatever became
    /// runnable.
    ///
    /// **Atomic in the job's own shape.** A forward edge, a forward parent, or
    /// a request for a handle this job never declared is rejected *before* the
    /// first node is inserted, so a malformed job leaves the graph untouched
    /// rather than half-built.
    ///
    /// # Errors
    /// Propagates [`BuildError`] — a forward reference in the job's own
    /// declarations, a handle from a different job, or a graph rejection.
    pub fn insert_job(
        &mut self,
        root_parent: Option<NodeId>,
        declare: impl FnOnce(&JobBuilder<N, R>) -> Vec<NodeGuid>,
    ) -> Result<Vec<NodeId>, BuildError> {
        let job = JobBuilder::new();
        let wanted = declare(&job);
        let graph = &mut self.graph;
        job.insert_with(root_parent, &wanted, |payload, deps, parent| {
            graph.insert_unchecked(payload, deps, parent)
        })
    }

    /// Claim every currently-runnable pending node and start it: node-deps
    /// satisfied and all resource-deps acquired atomically (all-or-nothing).
    /// Each claimed node is marked `Running`, its acquired units recorded, and
    /// its id returned for the runner to execute. A single pass suffices — a
    /// node started here is `Running`, not terminal, so it cannot satisfy another
    /// node's dependency in the same pass; it only consumes resources.
    #[must_use]
    pub fn settle(&mut self) -> Vec<NodeId> {
        let pending: Vec<NodeId> = self
            .graph
            .nodes()
            .filter(|n| n.state == State::Pending)
            .map(|n| n.id)
            .collect();
        let mut started = Vec::new();
        for id in pending {
            if self.node_deps_satisfied(id) && self.try_start(id) {
                started.push(id);
            }
        }
        started
    }

    /// Try to start node `id`. For each resource it needs, decide per the parent
    /// tree (see the module docs): acquire fresh units (owner), acquire an extra
    /// unit (grant lent elsewhere), or borrow an ancestor's grant. The fresh set
    /// is taken atomically (all-or-nothing via [`ResourceTable::try_acquire_all`])
    /// and recorded as owned; borrows are recorded only once the fresh set is
    /// secured. If the fresh set doesn't fit, nothing is taken, no borrow is
    /// recorded, and the node stays pending. Returns whether it started.
    fn try_start(&mut self, id: NodeId) -> bool {
        let mut to_acquire: Vec<(R, u32)> = Vec::new();
        let mut to_borrow: Vec<(NodeId, R)> = Vec::new();
        for (name, count) in self.resource_reqs(id) {
            match self.parent_ancestor_owning(id, &name) {
                // Case 1: no ancestor holds it → this node acquires + owns it.
                None => to_acquire.push((name, count)),
                Some(owner) => match self.borrowed.get(&(owner, name.clone())).copied() {
                    // Case 3: the grant is free → borrow it, no new unit.
                    None => to_borrow.push((owner, name)),
                    // Covered: already lent to a branch this node is inside.
                    Some(branch) if self.parent_chain_contains(id, branch) => {}
                    // Case 2: lent to a different branch → take an extra unit.
                    Some(_) => to_acquire.push((name, count)),
                },
            }
        }
        if !to_acquire.is_empty() && !self.resources.try_acquire_all(&to_acquire) {
            return false;
        }
        if !to_acquire.is_empty() {
            self.owned.entry(id).or_default().extend(to_acquire);
        }
        for (owner, name) in to_borrow {
            self.borrowed.insert((owner, name), id);
        }
        self.graph.set_state(id, State::Running);
        true
    }

    /// The nearest [`Node::parent`] ancestor of `id` that *owns* (holds real
    /// units of) `name`, or `None` if none does (⇒ `id` must acquire it fresh).
    fn parent_ancestor_owning(&self, id: NodeId, name: &R) -> Option<NodeId> {
        let mut cur = self.graph.node(id).and_then(|n| n.parent);
        while let Some(p) = cur {
            if self.node_owns(p, name) {
                return Some(p);
            }
            cur = self.graph.node(p).and_then(|n| n.parent);
        }
        None
    }

    /// Whether `ancestor` lies on `id`'s [`Node::parent`] chain (i.e. `id` is in
    /// `ancestor`'s subtree). `id` itself does not count as its own ancestor.
    fn parent_chain_contains(&self, id: NodeId, ancestor: NodeId) -> bool {
        let mut cur = self.graph.node(id).and_then(|n| n.parent);
        while let Some(p) = cur {
            if p == ancestor {
                return true;
            }
            cur = self.graph.node(p).and_then(|n| n.parent);
        }
        false
    }

    /// Whether node `holder` holds real units of resource `name`.
    fn node_owns(&self, holder: NodeId, name: &R) -> bool {
        self.owned
            .get(&holder)
            .is_some_and(|units| units.iter().any(|(n, _)| n == name))
    }

    /// Whether `root` and every node in its [`Node::parent`] subtree are
    /// terminal — the condition for releasing `root`'s owned grants (and for
    /// giving back a borrow whose branch-root is `root`).
    fn subtree_terminal(&self, root: NodeId) -> bool {
        self.graph.node(root).is_none_or(|n| n.state.is_terminal())
            && !self
                .graph
                .nodes()
                .any(|n| !n.state.is_terminal() && self.parent_chain_contains(n.id, root))
    }

    /// Report a running node's own logic result. On success the node is *not*
    /// terminal until its sub-nodes ([`Node::parent`] children) all finish — it
    /// rests in [`State::Finishing`] until then, rolling up to [`State::Done`]
    /// (every child `Done`) or [`State::Failed`] (any child `Failed`/`Cancelled`).
    /// On failure it is `Failed` at once and its pending sub-nodes are cancelled
    /// (gated on a `Finishing` the parent never reached). Terminality then
    /// propagates up the parent chain. Call [`Scheduler::settle`] again afterwards
    /// to start newly-unblocked work.
    pub fn complete(&mut self, id: NodeId, outcome: Outcome) {
        match outcome {
            Outcome::Failed(error) => {
                // Record the reason before the terminal transition so it's set
                // by the time `set_state` stamps `finished_at`.
                self.graph.set_error(id, error);
                self.graph.set_state(id, State::Failed);
                self.cascade_cancel(id);
            }
            Outcome::Done => self.settle_terminal(id),
        }
        self.roll_up_ancestors(id);
        self.release_ready();
    }

    /// Whether every direct child of `id` is terminal.
    fn all_children_terminal(&self, id: NodeId) -> bool {
        self.graph
            .nodes()
            .all(|n| n.parent != Some(id) || n.state.is_terminal())
    }

    /// 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:
    /// [`State::Finishing`] while any child is still non-terminal, else `Failed`
    /// if a child failed, else `Done`. A node with no children skips `Finishing`.
    ///
    /// Cascades on **any** terminal outcome, `Done` included. Since an edge names
    /// the set of outcomes it accepts, success can rule a dependent out just as
    /// failure can — a `{Failed}` compensation branch is unsatisfiable the moment
    /// 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) {
            self.rolled_up_state(id)
        } else {
            State::Finishing
        };
        self.graph.set_state(id, state);
        if state.is_terminal() {
            self.cascade_cancel(id);
        }
    }

    /// After `start` became terminal, roll up every ancestor that was parked in
    /// `Finishing` awaiting its children: once all of an ancestor's children are
    /// terminal it transitions (Done / Failed), which may let *its* parent roll
    /// up too, and so on up the [`Node::parent`] chain.
    fn roll_up_ancestors(&mut self, start: NodeId) {
        let mut cur = self.graph.node(start).and_then(|n| n.parent);
        while let Some(a) = cur {
            if self.graph.node(a).map(|n| n.state) != Some(State::Finishing)
                || !self.all_children_terminal(a)
            {
                break;
            }
            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);
            cur = self.graph.node(a).and_then(|n| n.parent);
        }
    }

    /// 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 {
        // 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
    /// entry per resource each owning node holds. Lets a caller render live
    /// ownership (which node holds a given resource) as a pull query, instead of
    /// threading release events out of the scheduler.
    #[must_use]
    pub fn resource_state(&self) -> Vec<(R, NodeId)> {
        self.owned
            .iter()
            .flat_map(|(&holder, units)| units.iter().map(move |(name, _)| (name.clone(), holder)))
            .collect()
    }

    /// Propagate elimination out from a just-terminal `origin`: every
    /// still-`Pending` node that can no longer run gets marked
    /// [`State::Skipped`], transitively. Skipped nodes were `Pending`, so they
    /// hold no resources.
    ///
    /// `Skipped`, not `Cancelled`: these nodes were *ruled out by their edges*,
    /// which is a normal outcome, not a dropped job. `Cancelled` is reserved for
    /// work the caller abandoned before it started ([`Scheduler::cancel_node`]),
    /// and the two are distinguished precisely so a parent's roll-up can ignore
    /// the former while still treating the latter as not-success.
    ///
    /// One rule decides it: **a node is doomed once any edge it names can never
    /// be satisfied** — the dep settled on an outcome that edge does not accept.
    /// `AFTER_OK` on a `Failed` dep dooms (a strong dependency failed);
    /// `AFTER_ANY` never dooms, which is what lets a tail node survive the
    /// cancellation of the work it reports on. That falls out of the edge's own
    /// set rather than being a special case for particular node kinds.
    ///
    /// Plus the structural edge: being a *child* of a doomed node. Its parent
    /// will never reach `Finishing`, so it was gated from ever starting, and
    /// leaving it pending would wedge the subtree non-terminal.
    fn cascade_cancel(&mut self, origin: NodeId) {
        let mut stack = vec![origin];
        while let Some(cur) = stack.pop() {
            let Some(outcome) = self.graph.node(cur).and_then(|n| n.state.terminal()) else {
                continue;
            };
            let doomed: Vec<NodeId> = self
                .graph
                .nodes()
                .filter(|n| {
                    n.state == State::Pending
                        && (n.parent == Some(cur)
                            || n.deps.iter().any(|d| {
                                matches!(d, Dep::Node { id, when } if *id == cur && !when.accepts(outcome))
                            }))
                })
                .map(|n| n.id)
                .collect();
            for d in doomed {
                self.graph.set_state(d, State::Skipped);
                stack.push(d);
            }
        }
    }

    /// Give back any borrow whose branch has fully left (freeing the grant for a
    /// waiting sibling), then release every owner's grant whose whole subtree is
    /// terminal (dropping the units back into the table).
    fn release_ready(&mut self) {
        // 1. Return borrows whose branch-root subtree is now terminal.
        let returned: Vec<(NodeId, R)> = self
            .borrowed
            .iter()
            .filter(|&(_, &branch)| self.subtree_terminal(branch))
            .map(|((owner, name), _)| (*owner, name.clone()))
            .collect();
        for key in returned {
            self.borrowed.remove(&key);
        }
        // 2. Release owner grants whose whole subtree is terminal.
        let owners: Vec<NodeId> = self.owned.keys().copied().collect();
        for owner in owners {
            if self.subtree_terminal(owner)
                && let Some(units) = self.owned.remove(&owner)
            {
                self.resources.release_all(&units);
            }
        }
    }

    /// Whether `id` is clear to start: its parent's own logic is done *and* every
    /// [`Dep::Node`] edge is satisfied. The parent gate (a sub-node runs only
    /// after its parent reaches [`State::Finishing`] — the parent can't be
    /// terminal while this child is pending, so `Finishing` is the exact "logic
    /// done" signal) supplies parent→child ordering; `Dep::Node` edges (which the
    /// graph restricts to the same parent group) supply sibling ordering.
    /// `Dep::Resource` edges are handled by the atomic acquire in
    /// [`Scheduler::settle`], not here.
    fn node_deps_satisfied(&self, id: NodeId) -> bool {
        let Some(node) = self.graph.node(id) else {
            return false;
        };
        if let Some(parent) = node.parent
            && self.graph.node(parent).map(|n| n.state) != Some(State::Finishing)
        {
            return false;
        }
        node.deps.iter().all(|dep| match dep {
            Dep::Resource { .. } => true,
            Dep::Node { id, when } => self
                .graph
                .node(*id)
                .is_some_and(|n| when.satisfied_by(n.state)),
        })
    }

    /// The `(name, count)` resource units `id` must hold to run.
    fn resource_reqs(&self, id: NodeId) -> Vec<(R, u32)> {
        self.graph.node(id).map_or_else(Vec::new, |node| {
            node.deps
                .iter()
                .filter_map(|dep| match dep {
                    Dep::Resource { name, count } => Some((name.clone(), *count)),
                    Dep::Node { .. } => None,
                })
                .collect()
        })
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::{DepWhen, TerminalState};

    fn res(name: &str) -> String {
        name.to_owned()
    }

    /// A graph + a resource table with `build-slot` set to `slots`.
    fn scheduler_with_slots(slots: u32) -> Scheduler<&'static str, String> {
        let mut table = ResourceTable::new();
        table.set_capacity(res("build-slot"), slots);
        Scheduler::new(Graph::new(), table)
    }

    /// A single-unit resource dep on `name`.
    fn res_dep(name: &str) -> Vec<Dep<String>> {
        vec![Dep::Resource {
            name: res(name),
            count: 1,
        }]
    }

    fn after_ok(on: NodeId) -> Dep<String> {
        Dep::Node {
            id: on,
            when: DepWhen::AFTER_OK,
        }
    }

    fn avail(s: &Scheduler<&str, String>, name: &str) -> u32 {
        s.resources.available(&res(name))
    }

    #[test]
    fn leaf_owner_goes_done_directly_and_releases() {
        let mut s = scheduler_with_slots(1);
        let n = s
            .append("build", res_dep("build-slot"), None)
            .expect("insert");
        assert_eq!(s.settle(), vec![n]);
        assert_eq!(s.graph().node(n).unwrap().state, State::Running);
        assert_eq!(avail(&s, "build-slot"), 0);
        // No children → completing it goes straight to Done (skips Finishing).
        s.complete(n, Outcome::Done);
        assert_eq!(s.graph().node(n).unwrap().state, State::Done);
        assert_eq!(avail(&s, "build-slot"), 1);
    }

    #[test]
    fn lifecycle_timestamps_and_error_are_stamped() {
        let mut s = scheduler_with_slots(2);
        let ok = s.append("ok", vec![], None).expect("insert");
        let bad = s.append("bad", vec![], None).expect("insert");
        let downstream = s.append("down", vec![after_ok(bad)], None).expect("insert");

        // Before running: no timestamps.
        assert!(s.graph().node(ok).unwrap().started_at.is_none());
        assert!(s.graph().node(ok).unwrap().finished_at.is_none());

        let started = s.settle();
        assert!(started.contains(&ok) && started.contains(&bad));
        // Running → started_at stamped, finished_at still none.
        assert!(s.graph().node(ok).unwrap().started_at.is_some());
        assert!(s.graph().node(ok).unwrap().finished_at.is_none());

        // Done → finished_at stamped, no error.
        s.complete(ok, Outcome::Done);
        let n = s.graph().node(ok).unwrap();
        assert!(n.finished_at.is_some());
        assert_eq!(n.error, None);

        // Failed → the reason rides `Outcome::Failed`, finished_at stamped.
        s.complete(bad, Outcome::Failed("boom".to_owned()));
        let n = s.graph().node(bad).unwrap();
        assert_eq!(n.state, State::Failed);
        assert_eq!(n.error.as_deref(), Some("boom"));
        assert!(n.finished_at.is_some());

        // The `AFTER_OK` dependent: ruled out by its edge, so finished_at is set,
        // but it never ran (no started_at) and carries no error of its own.
        let n = s.graph().node(downstream).unwrap();
        assert_eq!(n.state, State::Skipped);
        assert!(n.started_at.is_none());
        assert!(n.finished_at.is_some());
        assert_eq!(n.error, None);
    }

    #[test]
    fn build_slot_cap_limits_concurrency_and_release_unblocks() {
        let mut s = scheduler_with_slots(2);
        // Three independent (unparented) nodes each own a fresh slot unit.
        let a = s.append("a", res_dep("build-slot"), None).expect("a");
        let b = s.append("b", res_dep("build-slot"), None).expect("b");
        let c = s.append("c", res_dep("build-slot"), None).expect("c");
        // cap 2 → a + b start, c blocks on the exhausted slot.
        assert_eq!(s.settle(), vec![a, b]);
        assert_eq!(s.graph().node(c).unwrap().state, State::Pending);
        // a finishes → its slot frees → c can now start.
        s.complete(a, Outcome::Done);
        assert_eq!(s.settle(), vec![c]);
        assert_eq!(s.graph().node(c).unwrap().state, State::Running);
    }

    #[test]
    fn parent_parks_in_finishing_until_children_roll_up() {
        // `root` (a group node) runs, then its two sub-nodes run. `root` is not
        // terminal until both children are — it waits in `Finishing`.
        let mut s: Scheduler<&str, String> = Scheduler::new(Graph::new(), ResourceTable::new());
        let root = s.append("root", vec![], None).expect("root");
        let c1 = s.append("c1", vec![], Some(root)).expect("c1");
        let c2 = s.append("c2", vec![], Some(root)).expect("c2");
        assert_eq!(s.settle(), vec![root]);
        // Children can't start yet — parent still Running (logic not done).
        assert!(s.settle().is_empty(), "children gated on parent logic");
        s.complete(root, Outcome::Done);
        assert_eq!(
            s.graph().node(root).unwrap().state,
            State::Finishing,
            "logic done, children pending → Finishing"
        );
        let mut started = s.settle();
        started.sort();
        let mut expected = vec![c1, c2];
        expected.sort();
        assert_eq!(started, expected, "children run once parent is Finishing");
        s.complete(c1, Outcome::Done);
        assert_eq!(
            s.graph().node(root).unwrap().state,
            State::Finishing,
            "still Finishing while c2 runs"
        );
        s.complete(c2, Outcome::Done);
        assert_eq!(
            s.graph().node(root).unwrap().state,
            State::Done,
            "rolls up to Done once every child is Done"
        );
    }

    #[test]
    fn failed_child_rolls_parent_up_to_failed() {
        let mut s: Scheduler<&str, String> = Scheduler::new(Graph::new(), ResourceTable::new());
        let root = s.append("root", vec![], None).expect("root");
        let child = s.append("child", vec![], Some(root)).expect("child");
        assert_eq!(s.settle(), vec![root]);
        s.complete(root, Outcome::Done);
        assert_eq!(s.settle(), vec![child]);
        s.complete(child, Outcome::Failed(String::new()));
        assert_eq!(
            s.graph().node(root).unwrap().state,
            State::Failed,
            "a failed child rolls the parent up to Failed"
        );
    }

    #[test]
    fn owner_holds_grant_for_its_whole_subtree() {
        // Group root R owns the slot; c1 (its child) borrows it; c2 (its child,
        // needs no slot, ordered after c1) doesn't. The slot is held until R's
        // WHOLE subtree is terminal — not freed after the last needer (c1).
        let mut s = scheduler_with_slots(1);
        let r = s.append("R", res_dep("build-slot"), None).expect("R");
        let c1 = s.append("c1", res_dep("build-slot"), Some(r)).expect("c1");
        let c2 = s.append("c2", vec![after_ok(c1)], Some(r)).expect("c2");
        assert_eq!(s.settle(), vec![r]);
        s.complete(r, Outcome::Done); // → Finishing (children pending)
        assert_eq!(avail(&s, "build-slot"), 0, "held: subtree not terminal");
        assert_eq!(s.settle(), vec![c1], "c1 borrows R's slot");
        assert_eq!(avail(&s, "build-slot"), 0, "borrow reuses R's unit");
        s.complete(c1, Outcome::Done);
        assert_eq!(
            avail(&s, "build-slot"),
            0,
            "still held: c2 pending in subtree"
        );
        assert_eq!(s.settle(), vec![c2]);
        s.complete(c2, Outcome::Done);
        assert_eq!(
            avail(&s, "build-slot"),
            1,
            "released once whole subtree done"
        );
        assert_eq!(s.graph().node(r).unwrap().state, State::Done);
    }

    #[test]
    fn child_borrows_ancestor_grant_released_when_subtree_done() {
        // Lease-shaped resource (agent/foo, default cap 1): the group root owns
        // it, its sub-node borrows it, released only once the subtree is done.
        let mut s: Scheduler<&str, String> = Scheduler::new(Graph::new(), ResourceTable::new());
        let owner = s
            .append("owner", res_dep("agent/foo"), None)
            .expect("owner");
        assert_eq!(s.settle(), vec![owner]);
        assert_eq!(avail(&s, "agent/foo"), 0);
        let child = s
            .append("child", res_dep("agent/foo"), Some(owner))
            .expect("child");
        s.complete(owner, Outcome::Done); // → Finishing
        assert_eq!(avail(&s, "agent/foo"), 0, "held while a borrower pends");
        assert_eq!(s.settle(), vec![child]);
        assert_eq!(avail(&s, "agent/foo"), 0, "borrow reuses the one unit");
        s.complete(child, Outcome::Done);
        assert_eq!(avail(&s, "agent/foo"), 1);
    }

    #[test]
    fn nested_borrowers_never_deadlock() {
        // R (owns foo) → c1 → gc1 → ggc1, each the child of the previous, all
        // needing agent/foo (cap 1). c1 borrows R's grant; gc1 + ggc1 are inside
        // c1's borrow-branch so they are *covered* — a deep nest never deadlocks
        // on the single unit, and foo is held for the whole nest.
        let mut s: Scheduler<&str, String> = Scheduler::new(Graph::new(), ResourceTable::new());
        let r = s.append("R", res_dep("agent/foo"), None).expect("R");
        let child = s
            .append("child", res_dep("agent/foo"), Some(r))
            .expect("child");
        let grand = s
            .append("grand", res_dep("agent/foo"), Some(child))
            .expect("grand");
        let great = s
            .append("great", res_dep("agent/foo"), Some(grand))
            .expect("great");
        assert_eq!(s.settle(), vec![r]);
        s.complete(r, Outcome::Done);
        assert_eq!(s.settle(), vec![child], "child borrows R's grant");
        s.complete(child, Outcome::Done);
        assert_eq!(s.settle(), vec![grand], "grand covered, no deadlock");
        s.complete(grand, Outcome::Done);
        assert_eq!(s.settle(), vec![great], "great covered too");
        assert_eq!(avail(&s, "agent/foo"), 0, "held across the whole nest");
        s.complete(great, Outcome::Done);
        assert_eq!(s.graph().node(r).unwrap().state, State::Done, "R rolled up");
        assert_eq!(avail(&s, "agent/foo"), 1, "released once the nest is done");
    }

    #[test]
    fn unrelated_nodes_needing_the_same_resource_are_serialized() {
        // Two unparented nodes need agent/foo (cap 1); neither is in the other's
        // subtree, so the second can't borrow — it waits for the first's release.
        let mut s: Scheduler<&str, String> = Scheduler::new(Graph::new(), ResourceTable::new());
        let a = s.append("a", res_dep("agent/foo"), None).expect("a");
        let b = s.append("b", res_dep("agent/foo"), None).expect("b");
        assert_eq!(s.settle(), vec![a], "only a acquires; b can't borrow it");
        assert_eq!(s.graph().node(b).unwrap().state, State::Pending);
        s.complete(a, Outcome::Done);
        assert_eq!(s.settle(), vec![b]);
        assert_eq!(s.graph().node(b).unwrap().state, State::Running);
    }

    #[test]
    fn sibling_borrowers_of_a_cap1_grant_serialize() {
        // Two children of the owner both need agent/foo (cap 1): one borrows the
        // grant, the other (grant lent to a sibling branch, no free unit) waits
        // until the borrow is returned — mutual exclusion within the group.
        let mut s: Scheduler<&str, String> = Scheduler::new(Graph::new(), ResourceTable::new());
        let owner = s
            .append("owner", res_dep("agent/foo"), None)
            .expect("owner");
        assert_eq!(s.settle(), vec![owner]);
        let c1 = s
            .append("c1", res_dep("agent/foo"), Some(owner))
            .expect("c1");
        let c2 = s
            .append("c2", res_dep("agent/foo"), Some(owner))
            .expect("c2");
        s.complete(owner, Outcome::Done); // → Finishing
        assert_eq!(s.settle(), vec![c1], "c1 borrows; c2 can't (cap 1)");
        assert_eq!(s.graph().node(c2).unwrap().state, State::Pending);
        s.complete(c1, Outcome::Done);
        assert_eq!(s.settle(), vec![c2], "borrow returned → c2 borrows");
        assert_eq!(avail(&s, "agent/foo"), 0, "still just the owner's unit");
    }

    #[test]
    fn sibling_borrowers_run_concurrently_when_capacity_allows() {
        // build-slot cap 2: owner holds one unit; c1 borrows it, c2 (grant lent
        // to a sibling branch) takes the *second* unit — both run at once.
        let mut s = scheduler_with_slots(2);
        let owner = s
            .append("owner", res_dep("build-slot"), None)
            .expect("owner");
        assert_eq!(s.settle(), vec![owner]);
        assert_eq!(avail(&s, "build-slot"), 1, "owner took one of two");
        let c1 = s
            .append("c1", res_dep("build-slot"), Some(owner))
            .expect("c1");
        let c2 = s
            .append("c2", res_dep("build-slot"), Some(owner))
            .expect("c2");
        s.complete(owner, Outcome::Done); // → Finishing
        let mut started = s.settle();
        started.sort();
        let mut expected = vec![c1, c2];
        expected.sort();
        assert_eq!(started, expected, "c1 borrows, c2 takes the 2nd unit");
        assert_eq!(avail(&s, "build-slot"), 0);
    }

    #[test]
    fn failed_after_ok_dep_cancels_dependents_but_after_any_still_runs() {
        // A group of top-level siblings ordered by `AfterOk`; the failure of
        // `root` cancels its strong-dependent chain, an `AfterAny` still runs.
        let mut s: Scheduler<&str, String> = Scheduler::new(Graph::new(), ResourceTable::new());
        let root = s.append("root", vec![], None).expect("root");
        let strong1 = s
            .append("strong1", vec![after_ok(root)], None)
            .expect("strong1");
        let strong2 = s
            .append("strong2", vec![after_ok(strong1)], None)
            .expect("strong2");
        let weak = s
            .append(
                "weak",
                vec![Dep::Node {
                    id: root,
                    when: DepWhen::AFTER_ANY,
                }],
                None,
            )
            .expect("weak");
        assert_eq!(s.settle(), vec![root]);
        s.complete(root, Outcome::Failed(String::new()));
        assert_eq!(s.graph().node(strong1).unwrap().state, State::Skipped);
        assert_eq!(s.graph().node(strong2).unwrap().state, State::Skipped);
        assert_eq!(s.settle(), vec![weak]);
    }

    /// The direction only a *set* edge can express: a branch that runs solely on
    /// failure. Success has to rule it out, which means the cascade must fire on
    /// `Done` too — otherwise it sits `Pending` forever and wedges the graph.
    #[test]
    fn success_cancels_a_failure_only_branch() {
        let mut s: Scheduler<&str, String> = Scheduler::new(Graph::new(), ResourceTable::new());
        let root = s.append("root", vec![], None).expect("root");
        let on_fail = s
            .append(
                "compensate",
                vec![Dep::Node {
                    id: root,
                    when: DepWhen::of(&[TerminalState::Failed]),
                }],
                None,
            )
            .expect("compensate");
        assert_eq!(s.settle(), vec![root]);
        s.complete(root, Outcome::Done);
        assert_eq!(
            s.graph().node(on_fail).unwrap().state,
            State::Skipped,
            "a Failed-only branch is unsatisfiable once its dep succeeds — and it is \
             `Skipped`, not `Cancelled`, so the parent roll-up ignores it"
        );
        assert!(s.settle().is_empty(), "and nothing is left runnable");
    }

    /// The mirror: the same branch is exactly what *does* run on failure, while
    /// an `AFTER_OK` sibling is cancelled. One edge set, both directions.
    #[test]
    fn failure_runs_the_failure_only_branch_and_cancels_the_ok_one() {
        let mut s: Scheduler<&str, String> = Scheduler::new(Graph::new(), ResourceTable::new());
        let root = s.append("root", vec![], None).expect("root");
        let on_ok = s
            .append("on_ok", vec![after_ok(root)], None)
            .expect("on_ok");
        let on_fail = s
            .append(
                "on_fail",
                vec![Dep::Node {
                    id: root,
                    when: DepWhen::of(&[TerminalState::Failed]),
                }],
                None,
            )
            .expect("on_fail");
        assert_eq!(s.settle(), vec![root]);
        s.complete(root, Outcome::Failed("boom".to_owned()));
        assert_eq!(s.graph().node(on_ok).unwrap().state, State::Skipped);
        assert_eq!(s.settle(), vec![on_fail]);
    }

    /// A weak edge accepts a dependency that was *ruled out*, so a tail still
    /// runs when the work it reports on never happened — held by the edge itself
    /// rather than by any node-kind special case.
    #[test]
    fn eliminated_dep_still_satisfies_a_weak_edge() {
        let mut s: Scheduler<&str, String> = Scheduler::new(Graph::new(), ResourceTable::new());
        let root = s.append("root", vec![], None).expect("root");
        let mid = s.append("mid", vec![after_ok(root)], None).expect("mid");
        let tail = s
            .append(
                "tail",
                vec![Dep::Node {
                    id: mid,
                    when: DepWhen::AFTER_ANY,
                }],
                None,
            )
            .expect("tail");
        assert_eq!(s.settle(), vec![root]);
        s.complete(root, Outcome::Failed("boom".to_owned()));
        assert_eq!(s.graph().node(mid).unwrap().state, State::Skipped);
        assert_eq!(
            s.settle(),
            vec![tail],
            "the tail runs off a cancelled dependency"
        );
    }

    /// Edges are **conjunctive**, so "any of these several nodes failed" is not
    /// directly expressible — a `{Failed}` edge on each would mean *all* failed.
    /// The composition that does work: the success branch depends `AFTER_OK` on
    /// every node (so it runs only if all succeeded, and is ruled out the moment
    /// one doesn't), and the failure branch hangs off *it* with `{Skipped}` —
    /// "run when the success branch was ruled out". The success branch is the
    /// aggregator, and exactly one of the two runs.
    #[test]
    fn ok_branch_aggregates_and_failure_branch_hangs_off_its_elimination() {
        let build = || {
            let mut s: Scheduler<&str, String> = Scheduler::new(Graph::new(), ResourceTable::new());
            let a = s.append("a", vec![], None).expect("a");
            let b = s.append("b", vec![], None).expect("b");
            let on_ok = s
                .append("on_ok", vec![after_ok(a), after_ok(b)], None)
                .expect("on_ok");
            let on_fail = s
                .append(
                    "on_fail",
                    vec![Dep::Node {
                        id: on_ok,
                        when: DepWhen::of(&[TerminalState::Skipped]),
                    }],
                    None,
                )
                .expect("on_fail");
            (s, a, b, on_ok, on_fail)
        };

        // Everything succeeds: the ok branch runs, the failure branch is ruled out.
        let (mut s, a, b, on_ok, on_fail) = build();
        assert_eq!(s.settle(), vec![a, b], "both roots start; neither tail can");
        s.complete(a, Outcome::Done);
        s.complete(b, Outcome::Done);
        assert_eq!(s.settle(), vec![on_ok]);
        s.complete(on_ok, Outcome::Done);
        assert_eq!(s.graph().node(on_fail).unwrap().state, State::Skipped);
        assert!(s.settle().is_empty());

        // One of them fails: the ok branch is ruled out, which is precisely the
        // signal the failure branch waits on.
        let (mut s, a, b, on_ok, on_fail) = build();
        assert_eq!(s.settle(), vec![a, b]);
        s.complete(a, Outcome::Failed("boom".to_owned()));
        s.complete(b, Outcome::Done);
        assert_eq!(s.graph().node(on_ok).unwrap().state, State::Skipped);
        assert_eq!(s.settle(), vec![on_fail]);
    }

    #[test]
    fn failed_parent_cancels_its_pending_children() {
        // A failed group node cancels its sub-nodes (they were gated from ever
        // running on a `Finishing` the parent never reached).
        let mut s: Scheduler<&str, String> = Scheduler::new(Graph::new(), ResourceTable::new());
        let root = s.append("root", vec![], None).expect("root");
        let child = s.append("child", vec![], Some(root)).expect("child");
        let grandchild = s.append("gc", vec![], Some(child)).expect("gc");
        assert_eq!(s.settle(), vec![root]);
        s.complete(root, Outcome::Failed(String::new()));
        assert_eq!(s.graph().node(child).unwrap().state, State::Skipped);
        assert_eq!(s.graph().node(grandchild).unwrap().state, State::Skipped);
    }

    #[test]
    fn cancel_node_cancels_pending_and_cascades_but_not_running() {
        let mut s: Scheduler<&str, String> = Scheduler::new(Graph::new(), ResourceTable::new());
        let a = s.append("a", vec![], None).expect("a");
        let b = s.append("b", vec![after_ok(a)], None).expect("b");
        assert!(s.cancel_node(a));
        // `a` was dropped by the caller; `b` was merely ruled out by its edge.
        assert_eq!(s.graph().node(a).unwrap().state, State::Cancelled);
        assert_eq!(s.graph().node(b).unwrap().state, State::Skipped);
        let c = s.append("c", vec![], None).expect("c");
        assert_eq!(s.settle(), vec![c]);
        assert!(!s.cancel_node(c));
        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);
        let g = s.append("g", res_dep("agent/foo"), None).expect("g");
        let b = s.append("b", res_dep("build-slot"), None).expect("b");
        assert_eq!(s.settle().len(), 2);
        let state = s.resource_state();
        assert!(state.contains(&(res("agent/foo"), g)));
        assert!(state.contains(&(res("build-slot"), b)));
    }
}