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::resources::ResourceTable;
use crate::{Dep, DepWhen, Graph, GraphError, NodeId, State};

/// 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, Copy, PartialEq, Eq)]
pub enum Outcome {
    /// The node's work succeeded.
    Done,
    /// The node's work failed.
    Failed,
}

/// 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)
    }

    /// 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 => {
                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())
    }

    /// Whether any direct child of `id` ended `Failed`/`Cancelled` — the roll-up
    /// failure condition for the parent.
    fn any_child_failed(&self, id: NodeId) -> bool {
        self.graph
            .nodes()
            .any(|n| n.parent == Some(id) && matches!(n.state, State::Failed | State::Cancelled))
    }

    /// 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`.
    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
        } else {
            State::Done
        };
        self.graph.set_state(id, state);
        if state == State::Failed {
            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 = if self.any_child_failed(a) {
                State::Failed
            } else {
                State::Done
            };
            self.graph.set_state(a, state);
            if state == State::Failed {
                self.cascade_cancel(a);
            }
            cur = self.graph.node(a).and_then(|n| n.parent);
        }
    }

    /// Cancel a still-*pending* node (and cascade to its `AfterOk` dependents):
    /// 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).
    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
        }
    }

    /// 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 cancellation out from a just-failed/cancelled `origin`: every
    /// still-`Pending` node that can no longer run gets marked `Cancelled`,
    /// transitively. Two edges carry it: (a) an `AfterOk` dep on a cancelled node
    /// (a strong dependency failed), and (b) being a *child* of one (its parent
    /// will never reach `Finishing`, so it was gated from ever starting — and
    /// leaving it pending would wedge the subtree non-terminal). Cancelled nodes
    /// were `Pending`, so they hold no resources.
    fn cascade_cancel(&mut self, origin: NodeId) {
        let mut stack = vec![origin];
        while let Some(cur) = stack.pop() {
            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: DepWhen::AfterOk } if *id == cur)
                            }))
                })
                .map(|n| n.id)
                .collect();
            for d in doomed {
                self.graph.set_state(d, State::Cancelled);
                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::*;

    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::AfterOk,
        }
    }

    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 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);
        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::AfterAny,
                }],
                None,
            )
            .expect("weak");
        assert_eq!(s.settle(), vec![root]);
        s.complete(root, Outcome::Failed);
        assert_eq!(s.graph().node(strong1).unwrap().state, State::Cancelled);
        assert_eq!(s.graph().node(strong2).unwrap().state, State::Cancelled);
        assert_eq!(s.settle(), vec![weak]);
    }

    #[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);
        assert_eq!(s.graph().node(child).unwrap().state, State::Cancelled);
        assert_eq!(s.graph().node(grandchild).unwrap().state, State::Cancelled);
    }

    #[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));
        assert_eq!(s.graph().node(a).unwrap().state, State::Cancelled);
        assert_eq!(s.graph().node(b).unwrap().state, State::Cancelled);
        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);
    }

    #[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)));
    }
}