feat(#2949): claim_next — the seam that cannot be half-used

The caller supplies how to run a node and spawns what it gets back; it
never touches claiming or completion. The returned future runs the node
*and completes it*, so "forgot to finish the node" stops being something
a caller can do — completion is inside the thing they spawn.

The `Option` is answered synchronously, before anything is awaited, so the
run loop learns whether there was work without waiting on the node it just
started. That is what lets it choose between claiming again immediately
and backing off; an id alone cannot express that choice.

Locking: taken twice, briefly, and never held across the await — once to
claim, once inside the future to complete. A guard alive across an await
point would make the future non-`Send` and unspawnable, which is also why
the node itself runs unlocked for however long it takes. `Arc` +
`std::sync::Mutex` keep this runtime-agnostic: no tokio in this crate.

`run` receives an owned payload rather than a borrow for the same reason a
`&Job` could not be threaded through the executors: a reference parameter
is live for the whole future, borrowing the graph across the await and
poisoning `Send`.

The output carries the insert result instead of swallowing it. This crate
has no logger by design, so a malformed grown job is reported to the
caller, who can log it. The node completes either way — its own work
already happened.

hive-jobq/src/scheduler.rs

@@ -29,7 +29,9 @@
 //! is a deferred optimization — unsafe under dynamically-appended subnodes.)
 
 use std::collections::HashMap;
+use std::future::Future;
 use std::hash::Hash;
+use std::sync::{Arc, Mutex};
 
 use crate::builder::{BuildError, JobBuilder, NodeGuid};
 use crate::resources::ResourceTable;
@@ -160,6 +162,60 @@ impl<N, R: Clone + Eq + Hash> Scheduler<N, R> {
             .find(|&id| self.node_deps_satisfied(id) && self.try_start(id))
     }
 
+    /// Claim one runnable node and return **the work that runs it**, or `None`
+    /// when nothing is runnable right now.
+    ///
+    /// This is the seam: the caller supplies how to execute a node and spawns
+    /// the returned future, but never touches claiming or completion. The
+    /// future runs the node **and completes it**, so "forgot to finish the
+    /// node" is not expressible — completion is inside the thing you spawn.
+    ///
+    /// The `Option` is answered *synchronously*, before anything is awaited, so
+    /// the caller can decide "claim again immediately" vs "back off" without
+    /// waiting on the node it just started.
+    ///
+    /// ## Locking
+    /// The lock is taken twice, briefly, and **never held across the await**:
+    /// once here to claim, once inside the future to complete. That is what
+    /// keeps the returned future `Send` — a guard alive across an await point
+    /// would poison it — and it is why the node itself runs unlocked, for
+    /// however many minutes it needs.
+    ///
+    /// ## Why the payload is cloned
+    /// `run` gets an owned `N` rather than a borrow: a `&N` parameter is live
+    /// for the whole future, which both borrows the graph across the await and
+    /// makes the future non-`Send`.
+    ///
+    /// The output carries the insert result rather than swallowing it — this
+    /// crate has no logger, so a malformed grown job is reported to the caller,
+    /// who is the one that can log it. The node is completed either way: its
+    /// own work already happened.
+    pub fn claim_next<F, Fut>(
+        sched: &Arc<Mutex<Self>>,
+        run: F,
+    ) -> Option<impl Future<Output = (NodeId, Result<(), BuildError>)> + use<F, Fut, N, R>>
+    where
+        N: Clone,
+        F: FnOnce(NodeId, N, JobBuilder<N, R>) -> Fut,
+        Fut: Future<Output = (JobBuilder<N, R>, Outcome)>,
+    {
+        let (id, payload) = {
+            let mut guard = sched.lock().expect("jobq scheduler mutex poisoned");
+            let id = guard.claim_one()?;
+            let payload = guard.graph.node(id)?.payload.clone();
+            (id, payload)
+        };
+        let sched = Arc::clone(sched);
+        Some(async move {
+            let (grown, outcome) = run(id, payload, JobBuilder::new()).await;
+            let grew = sched
+                .lock()
+                .expect("jobq scheduler mutex poisoned")
+                .complete_growing(id, outcome, grown);
+            (id, grew)
+        })
+    }
+
     /// Claim every currently-runnable pending node. Equivalent to calling
     /// [`Self::claim_one`] until it yields `None`: a node started by an earlier
     /// iteration is `Running`, not terminal, so it cannot satisfy another