feat(#2500): validate NodeId references on insert and deserialize

Per mara's direction — validate ids as they enter the graph so internal
iteration can trust every id the graph holds; the generational route for
removal comes later. Adds GraphError; insert() now rejects a dangling
Dep::Node / parent id (it is fallible); validate() checks all internal id
references resolve and that next_id is past the largest existing id;
deserialization runs validate() via #[serde(try_from = "GraphData<N>")], so
a loaded graph can never carry a dangling reference. 5 new tests; serde_json
added as a dev-dependency for the round-trip cases.

Cargo.lock

@@ -1662,6 +1662,15 @@ dependencies = [
  "serde",
 ]
 
+[[package]]
+name = "hive-jobq"
+version = "0.1.0"
+dependencies = [
+ "serde",
+ "serde_json",
+ "thiserror 2.0.18",
+]
+
 [[package]]
 name = "hive-matrix-mcp"
 version = "0.1.0"

Cargo.toml

@@ -9,6 +9,7 @@ members = [
   "hive-claude",
   "hive-forge",
   "hive-host-sock",
+  "hive-jobq",
   "hive-matrix-mcp",
   "hive-metric",
   "hive-priv",

hive-jobq/Cargo.toml

@@ -0,0 +1,14 @@
+[package]
+name = "hive-jobq"
+edition.workspace = true
+version.workspace = true
+
+[lints]
+workspace = true
+
+[dependencies]
+serde = { workspace = true }
+thiserror = { workspace = true }
+
+[dev-dependencies]
+serde_json = { workspace = true }

hive-jobq/src/lib.rs

@@ -0,0 +1,508 @@
+//! `hive-jobq` — a persistent job-DAG scheduler, extracted from hive-c0re's
+//! in-tree `job_queue` as a domain-agnostic library.
+//!
+//! # Model (v2)
+//!
+//! One **persistent graph** for the whole system, not a DAG per job. Enqueuing
+//! inserts a self-contained **node group** and returns its id; the scheduler
+//! runs a continuous loop, starting every node whose [`Dep`]s are satisfied:
+//!
+//! - **Resource** deps are named counting semaphores ([`ResourceName`]):
+//!   `build-slot` (capacity N), `agent/<name>` (capacity 1), or any name
+//!   (capacity 1, created on use). A node acquires *all* its resource deps
+//!   atomically at start (all-or-nothing) — no hold-and-wait, so no deadlock
+//!   and no cycle detection needed.
+//! - **Node** deps wait on a node/group per [`DepWhen`]: `AfterOk` needs
+//!   success (a failed dep cancels the dependent), `AfterAny` only terminal.
+//!
+//! A **node group** is a self-contained sub-graph; things depend on it as a
+//! whole (done = every inner node terminal), never on an inner node. Groups
+//! nest; a running node may grow its own group but not reach outside it.
+//!
+//! A [`NodeId`] is opaque, stable, and monotonic — persisted, so it survives
+//! restarts. Group membership is a parent edge ([`Node::parent`]), *not* in the
+//! id; the `1/1/2` hierarchy is a derived UI label. The node payload is generic
+//! (`N`) so the library stays container-agnostic — the caller supplies its own
+//! node kind. Resources are held by the acquiring node and released on
+//! completion via guard objects, recursive within a group.
+//!
+//! The resource-acquisition machinery, the guards, and the scheduler loop are
+//! follow-ups; this is the data model they build on.
+
+/// Opaque, stable, monotonic node identifier.
+///
+/// Assigned by the [`Graph`] on insert and persisted, so it is stable across
+/// restarts. Group membership is a separate parent edge ([`Node::parent`]) — it
+/// is deliberately *not* encoded in the id, so the id never changes as the tree
+/// grows or collapses. The hierarchical `1/1/2` path used in the UI is derived
+/// from the parent tree at render time.
+///
+/// The inner field is crate-private: an id can only originate from the graph's
+/// monotonic counter (or deserialization of a persisted graph), never be
+/// fabricated by a caller — that is what makes it opaque.
+#[derive(
+    Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash, serde::Serialize, serde::Deserialize,
+)]
+pub struct NodeId(pub(crate) u64);
+
+/// A named counting semaphore.
+///
+/// Examples: `build-slot` (capacity configured to the number of build slots),
+/// `agent/<name>` (capacity 1 — the per-agent lifecycle lock), or any other
+/// name, which is assumed to have capacity 1 and is created on first use.
+#[derive(
+    Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash, serde::Serialize, serde::Deserialize,
+)]
+pub struct ResourceName(pub String);
+
+/// When a [`Dep::Node`] edge is satisfied — the strong/weak distinction the
+/// current queue carries as `DepWhen`, load-bearing for failure safety.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
+pub enum DepWhen {
+    /// The dependency must reach [`State::Done`]. This is the default chain
+    /// edge: if the dependency *fails*, the dependent must not run and is
+    /// cancelled ([`State::Cancelled`]) down the chain — e.g. a failed
+    /// `Prebuild` must not let `StopForUpdate` stop a healthy container.
+    AfterOk,
+    /// The dependency need only be terminal — success or failure both satisfy
+    /// it. For steps that must converge regardless, e.g. `Reconcile` running
+    /// even when the preceding `Swap` failed.
+    AfterAny,
+}
+
+impl DepWhen {
+    /// Whether a dependency in `dep_state` satisfies this edge.
+    #[must_use]
+    pub fn satisfied_by(self, dep_state: State) -> bool {
+        match self {
+            DepWhen::AfterOk => dep_state == State::Done,
+            DepWhen::AfterAny => dep_state.is_terminal(),
+        }
+    }
+}
+
+/// One dependency of a node. A node becomes runnable once every [`Dep::Node`]
+/// edge it names is satisfied (per its [`DepWhen`]) *and* every [`Dep::Resource`]
+/// it names can be acquired (all of them, atomically).
+#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
+pub enum Dep {
+    /// Depend on another node (or a group, by its group node's id). Whether a
+    /// *failed* dependency satisfies the edge is decided by `when`: `AfterOk`
+    /// requires success (and cancels this node if the dep fails), `AfterAny`
+    /// only requires the dep to be terminal.
+    Node {
+        /// The node (or group) depended on.
+        id: NodeId,
+        /// Strong (`AfterOk`) vs weak (`AfterAny`).
+        when: DepWhen,
+    },
+    /// Hold `count` units of a named resource for the duration of this node's
+    /// run. Acquired atomically with the node's other resource deps at start,
+    /// released when the node completes.
+    Resource {
+        /// The resource to acquire.
+        name: ResourceName,
+        /// How many units to hold (usually 1).
+        count: u32,
+    },
+}
+
+/// A node's lifecycle state.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
+pub enum State {
+    /// Waiting on dependencies (node or resource).
+    Pending,
+    /// Dependencies satisfied, resources held, currently executing.
+    Running,
+    /// Completed successfully.
+    Done,
+    /// Completed unsuccessfully.
+    Failed,
+    /// Never ran: an `AfterOk` dependency failed, so this node (and the rest of
+    /// its strong-dependent chain) is cancelled rather than run.
+    Cancelled,
+}
+
+impl State {
+    /// A node is *terminal* once it has finished — successfully, unsuccessfully,
+    /// or cancelled — which is when its resources are released and dependents
+    /// are re-evaluated.
+    #[must_use]
+    pub fn is_terminal(self) -> bool {
+        matches!(self, State::Done | State::Failed | State::Cancelled)
+    }
+}
+
+/// A node in the graph, carrying a caller-defined payload `N`.
+///
+/// The library schedules over `Node`s and resources without interpreting the
+/// payload; the caller supplies `N` (its own node kind) and a runner to execute
+/// a claimed node.
+#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
+pub struct Node<N> {
+    /// Stable identity, assigned on insert.
+    pub id: NodeId,
+    /// The group this node belongs to, if any. `None` for a top-level group
+    /// node. Group membership lives here, not in the id.
+    pub parent: Option<NodeId>,
+    /// Caller-defined payload (the node's kind / work description).
+    pub payload: N,
+    /// What must hold before this node runs (other nodes + resources).
+    pub deps: Vec<Dep>,
+    /// Lifecycle state.
+    pub state: State,
+}
+
+/// An error from inserting into or loading a [`Graph`] with a dangling id.
+///
+/// A [`NodeId`] is only meaningful against the graph that minted it, so both
+/// entry points — [`Graph::insert`] and deserialization — reject references to
+/// nodes the graph does not contain. That is what lets internal iteration trust
+/// every id the graph holds.
+#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
+pub enum GraphError {
+    /// A node's dependency named an id not present in the graph.
+    #[error("dependency references unknown node {0:?}")]
+    UnknownDep(NodeId),
+    /// A node's parent named an id not present in the graph.
+    #[error("parent references unknown node {0:?}")]
+    UnknownParent(NodeId),
+    /// A loaded graph's `next_id` counter is not past the largest existing id,
+    /// so the next minted id would collide with one already in the graph.
+    #[error("next_id {next_id} must exceed the largest existing node id {max_id}")]
+    NextIdTooSmall {
+        /// The persisted counter value.
+        next_id: u64,
+        /// The largest id already present.
+        max_id: u64,
+    },
+}
+
+/// The single persistent graph of all nodes.
+///
+/// New jobs are inserted as node groups; the scheduler (added in a follow-up)
+/// walks this graph filling open slots. Completed groups are retained (no
+/// pruning in v1).
+#[derive(Debug, serde::Serialize, serde::Deserialize)]
+#[serde(try_from = "GraphData<N>")]
+pub struct Graph<N> {
+    nodes: Vec<Node<N>>,
+    next_id: u64,
+}
+
+// Deserialization target: the raw fields, turned into a `Graph` by the `TryFrom`
+// below — which runs [`Graph::validate`], so a loaded graph can never carry a
+// dangling id reference (Serialize does not validate; Deserialize always does).
+#[derive(serde::Deserialize)]
+struct GraphData<N> {
+    nodes: Vec<Node<N>>,
+    next_id: u64,
+}
+
+impl<N> TryFrom<GraphData<N>> for Graph<N> {
+    type Error = GraphError;
+
+    fn try_from(data: GraphData<N>) -> Result<Self, Self::Error> {
+        let graph = Graph {
+            nodes: data.nodes,
+            next_id: data.next_id,
+        };
+        graph.validate()?;
+        Ok(graph)
+    }
+}
+
+// A `derive(Default)` would wrongly require `N: Default` (an empty graph holds
+// no payload); an empty `Vec<Node<N>>` needs no such bound, so impl it directly.
+impl<N> Default for Graph<N> {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+impl<N> Graph<N> {
+    /// An empty graph.
+    #[must_use]
+    pub fn new() -> Self {
+        Self {
+            nodes: Vec::new(),
+            next_id: 0,
+        }
+    }
+
+    /// Mint the next stable node id.
+    fn mint_id(&mut self) -> NodeId {
+        let id = NodeId(self.next_id);
+        self.next_id += 1;
+        id
+    }
+
+    /// Insert a node with the given payload, deps, and parent group, returning
+    /// its freshly-minted id. The node starts [`State::Pending`].
+    ///
+    /// Every [`Dep::Node`] id and the `parent` id (if any) must already resolve
+    /// to a node in the graph — an id is only meaningful against the graph that
+    /// minted it, so a dangling reference is rejected here rather than surfacing
+    /// as a broken edge later.
+    ///
+    /// # Errors
+    /// Returns [`GraphError::UnknownParent`] / [`GraphError::UnknownDep`] if the
+    /// parent or a dependency references a node not in the graph.
+    pub fn insert(
+        &mut self,
+        payload: N,
+        deps: Vec<Dep>,
+        parent: Option<NodeId>,
+    ) -> Result<NodeId, GraphError> {
+        if let Some(parent_id) = parent
+            && self.node(parent_id).is_none()
+        {
+            return Err(GraphError::UnknownParent(parent_id));
+        }
+        for dep in &deps {
+            if let Dep::Node { id, .. } = dep
+                && self.node(*id).is_none()
+            {
+                return Err(GraphError::UnknownDep(*id));
+            }
+        }
+        let id = self.mint_id();
+        self.nodes.push(Node {
+            id,
+            parent,
+            payload,
+            deps,
+            state: State::Pending,
+        });
+        Ok(id)
+    }
+
+    /// Borrow a node by id.
+    #[must_use]
+    pub fn node(&self, id: NodeId) -> Option<&Node<N>> {
+        self.nodes.iter().find(|n| n.id == id)
+    }
+
+    /// The direct children of a group node (nodes whose `parent` is `id`).
+    pub fn children(&self, id: NodeId) -> impl Iterator<Item = &Node<N>> {
+        self.nodes.iter().filter(move |n| n.parent == Some(id))
+    }
+
+    /// A group is terminal once the group node itself is terminal *and* every
+    /// node inside it (recursively) is terminal. The node's own state matters:
+    /// a group node still `Pending`/`Running` is not terminal even with no
+    /// children yet, since a running node may still append some. (Deciding when
+    /// to *settle* a group node to terminal once its children are all done is a
+    /// separate concern the scheduler owns.) An unknown id is not terminal.
+    #[must_use]
+    pub fn group_terminal(&self, id: NodeId) -> bool {
+        let Some(node) = self.node(id) else {
+            return false;
+        };
+        node.state.is_terminal() && self.children(id).all(|child| self.group_terminal(child.id))
+    }
+
+    /// Check that every id the graph holds resolves: each node's `parent` and
+    /// every [`Dep::Node`] id names a node present in the graph, and `next_id`
+    /// is past the largest existing id. Deserialization runs this, so a loaded
+    /// graph is internally consistent and internal iteration can trust its ids.
+    ///
+    /// # Errors
+    /// Returns [`GraphError`] on a dangling parent / dependency reference, or a
+    /// `next_id` that would remint an id already in the graph.
+    pub fn validate(&self) -> Result<(), GraphError> {
+        for node in &self.nodes {
+            if let Some(parent_id) = node.parent
+                && self.node(parent_id).is_none()
+            {
+                return Err(GraphError::UnknownParent(parent_id));
+            }
+            for dep in &node.deps {
+                if let Dep::Node { id, .. } = dep
+                    && self.node(*id).is_none()
+                {
+                    return Err(GraphError::UnknownDep(*id));
+                }
+            }
+        }
+        if let Some(max_id) = self.nodes.iter().map(|n| n.id.0).max()
+            && self.next_id <= max_id
+        {
+            return Err(GraphError::NextIdTooSmall {
+                next_id: self.next_id,
+                max_id,
+            });
+        }
+        Ok(())
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn insert_mints_stable_monotonic_ids() {
+        let mut g: Graph<&str> = Graph::new();
+        let a = g.insert("sweep", vec![], None).unwrap();
+        let b = g
+            .insert(
+                "update",
+                vec![Dep::Node {
+                    id: a,
+                    when: DepWhen::AfterOk,
+                }],
+                Some(a),
+            )
+            .unwrap();
+        assert_eq!(a, NodeId(0));
+        assert_eq!(b, NodeId(1));
+        // Membership is the parent edge, not the id.
+        assert_eq!(g.node(b).unwrap().parent, Some(a));
+        assert_eq!(g.node(a).unwrap().parent, None);
+    }
+
+    fn set_state<N>(g: &mut Graph<N>, id: NodeId, state: State) {
+        let idx = g.nodes.iter().position(|n| n.id == id).unwrap();
+        g.nodes[idx].state = state;
+    }
+
+    #[test]
+    fn group_terminal_requires_the_group_node_and_all_children_terminal() {
+        let mut g: Graph<&str> = Graph::new();
+        let group = g.insert("group", vec![], None).unwrap();
+        let child = g.insert("child", vec![], Some(group)).unwrap();
+        // Both pending → not terminal.
+        assert!(!g.group_terminal(group));
+        // Child done, but the group node itself is still pending → NOT terminal:
+        // the group node's own state is load-bearing, not just its children.
+        set_state(&mut g, child, State::Done);
+        assert!(!g.group_terminal(group));
+        // Group node terminal too → the whole group is terminal.
+        set_state(&mut g, group, State::Done);
+        assert!(g.group_terminal(group));
+    }
+
+    #[test]
+    fn empty_running_group_is_not_terminal() {
+        // A running node with no children yet may still append some, so it must
+        // not read as terminal just because its child set is currently empty.
+        let mut g: Graph<&str> = Graph::new();
+        let group = g.insert("group", vec![], None).unwrap();
+        set_state(&mut g, group, State::Running);
+        assert!(!g.group_terminal(group));
+        // Once it finishes (having grown no children), it is terminal.
+        set_state(&mut g, group, State::Done);
+        assert!(g.group_terminal(group));
+    }
+
+    #[test]
+    fn state_terminality() {
+        assert!(State::Done.is_terminal());
+        assert!(State::Failed.is_terminal());
+        assert!(State::Cancelled.is_terminal());
+        assert!(!State::Pending.is_terminal());
+        assert!(!State::Running.is_terminal());
+    }
+
+    #[test]
+    fn after_ok_needs_success_after_any_needs_terminal() {
+        // AfterOk: only Done satisfies; a Failed/Cancelled dep does NOT (the
+        // dependent must be cancelled, not run).
+        assert!(DepWhen::AfterOk.satisfied_by(State::Done));
+        assert!(!DepWhen::AfterOk.satisfied_by(State::Failed));
+        assert!(!DepWhen::AfterOk.satisfied_by(State::Cancelled));
+        assert!(!DepWhen::AfterOk.satisfied_by(State::Running));
+        // AfterAny: any terminal state satisfies.
+        assert!(DepWhen::AfterAny.satisfied_by(State::Done));
+        assert!(DepWhen::AfterAny.satisfied_by(State::Failed));
+        assert!(DepWhen::AfterAny.satisfied_by(State::Cancelled));
+        assert!(!DepWhen::AfterAny.satisfied_by(State::Pending));
+    }
+
+    #[test]
+    fn insert_rejects_unknown_parent() {
+        let mut g: Graph<&str> = Graph::new();
+        let bogus = NodeId(7);
+        assert_eq!(
+            g.insert("x", vec![], Some(bogus)).unwrap_err(),
+            GraphError::UnknownParent(bogus)
+        );
+    }
+
+    #[test]
+    fn insert_rejects_unknown_dep() {
+        let mut g: Graph<&str> = Graph::new();
+        let bogus = NodeId(42);
+        let deps = vec![Dep::Node {
+            id: bogus,
+            when: DepWhen::AfterOk,
+        }];
+        assert_eq!(
+            g.insert("x", deps, None).unwrap_err(),
+            GraphError::UnknownDep(bogus)
+        );
+    }
+
+    #[test]
+    fn valid_graph_round_trips_through_serde() {
+        let mut g: Graph<String> = Graph::new();
+        let a = g.insert("a".to_owned(), vec![], None).unwrap();
+        g.insert(
+            "b".to_owned(),
+            vec![Dep::Node {
+                id: a,
+                when: DepWhen::AfterAny,
+            }],
+            Some(a),
+        )
+        .unwrap();
+        let json = serde_json::to_string(&g).unwrap();
+        let back: Graph<String> = serde_json::from_str(&json).unwrap();
+        assert!(back.validate().is_ok());
+        assert_eq!(back.node(a).unwrap().payload, "a");
+    }
+
+    #[test]
+    fn deserialize_rejects_a_dangling_dependency() {
+        // Build a graph whose only node depends on a non-existent id, serialize
+        // it (Serialize does not validate), and confirm deserialize rejects it.
+        let bad = Graph::<String> {
+            nodes: vec![Node {
+                id: NodeId(0),
+                parent: None,
+                payload: "x".to_owned(),
+                deps: vec![Dep::Node {
+                    id: NodeId(99),
+                    when: DepWhen::AfterOk,
+                }],
+                state: State::Pending,
+            }],
+            next_id: 1,
+        };
+        let json = serde_json::to_string(&bad).unwrap();
+        let err = serde_json::from_str::<Graph<String>>(&json).unwrap_err();
+        assert!(err.to_string().contains("unknown node"));
+    }
+
+    #[test]
+    fn validate_rejects_next_id_that_would_remint() {
+        let bad = Graph::<&str> {
+            nodes: vec![Node {
+                id: NodeId(5),
+                parent: None,
+                payload: "x",
+                deps: vec![],
+                state: State::Pending,
+            }],
+            next_id: 3,
+        };
+        assert_eq!(
+            bad.validate().unwrap_err(),
+            GraphError::NextIdTooSmall {
+                next_id: 3,
+                max_id: 5,
+            }
+        );
+    }
+}