job_queue: three more tests read the graph instead of running it

non_graceful_rebuild_has_no_signal_or_drain claimed and completed six
nodes to collect their kinds. Mutation-checked rather than trusted: with
the template's graceful flag flipped, it fails with signal and drain in
the list. An absence assertion is the easiest kind to make vacuous, so
it is the one that most needs the check.

reparent_shape claimed a node only to get an id for its resource
assertion. node_of() finds it by kind and asserts there is exactly one,
so the test cannot quietly start being about a different node.

multi_agent_restart asserted "both subgraphs start concurrently" by
claiming and seeing two. That is not one fact but two declared ones plus
a jobq guarantee: both heads are group roots with no deps, so nothing
orders them; and each declares only its own agent lease, so nothing
makes them contend. Whether a scheduler then runs independent,
resource-disjoint roots at once belongs to hive-jobq and is tested
there. This asserts the two declarations.

hive-c0re/src/job_queue/tests.rs

@@ -217,6 +217,28 @@ fn declared_shape(q: &JobQueue, dag: u64) -> Vec<Declared> {
         .collect()
 }
 
+/// The id of the one node of `kind` under `dag`, for the resource assertions.
+///
+/// Panics unless there is exactly one — every caller is about a shape where the
+/// kind is unique, so two would mean the assertion had quietly stopped being
+/// about the node the test names.
+fn node_of(q: &JobQueue, dag: u64, kind: &str) -> hive_jobq::NodeId {
+    let sched = q.sched().lock().expect("job_queue mutex poisoned");
+    let graph = sched.graph();
+    let root = graph.resolve_id(dag).expect("dag id is a real node id");
+    let mut found: Vec<_> = graph
+        .nodes()
+        .filter(|n| n.id != root && graph.root_of(n.id) == Some(root) && n.payload.as_str() == kind)
+        .map(|n| n.id)
+        .collect();
+    assert_eq!(
+        found.len(),
+        1,
+        "expected exactly one {kind} node in the dag"
+    );
+    found.pop().expect("checked above")
+}
+
 /// Kinds of every node under `dag` still `Pending` — the nodes that could yet
 /// run. Stronger than asking the scheduler what is *ready right now*: a node
 /// blocked on a dep is not ready but is very much still alive.
@@ -511,14 +533,11 @@ fn non_graceful_rebuild_has_no_signal_or_drain() {
             }),
         },
     );
-    let mut kinds = Vec::new();
-    for _ in 0..6 {
-        let c = claim_one(&q);
-        kinds.push(c.kind.as_str().to_owned());
-        q.complete_node(c.node_id, Ok(()));
-    }
     assert_eq!(
-        kinds,
+        declared_shape(&q, id)
+            .iter()
+            .map(|d| d.kind)
+            .collect::<Vec<_>>(),
         vec![
             "meta_sync",
             "prebuild",
@@ -526,10 +545,9 @@ fn non_graceful_rebuild_has_no_signal_or_drain() {
             "swap",
             "post_swap",
             "reconcile"
-        ]
+        ],
+        "exactly six nodes, and neither of them is signal or drain"
     );
-    // Settled after exactly those six — nothing else was declared.
-    assert_eq!(state_of(&q, id), State::Done);
 }
 
 /// A cleanly-finished DAG leaves the snapshot even though its not-taken
@@ -751,23 +769,45 @@ fn multi_agent_restart_is_one_dag_with_concurrent_per_agent_subgraphs() {
     );
     // A hive-wide restart is ONE DAG, not one-per-agent.
     assert_eq!(q.snapshot().len(), 1);
-    // Each agent's subgraph head (StopForUpdate, since both are running) is
-    // a root, so both are claimable at once — each takes its OWN agent's
-    // lease (no contention across distinct agents), all inside the single DAG.
-    let claims = q.claim_ready();
-    assert!(claims.iter().all(|c| c.dag_id == id));
-    let mut heads: Vec<(&str, &str)> = claims
-        .iter()
-        .map(|c| (c.agent.as_str(), c.kind.as_str()))
+    // Each agent's subgraph head (StopForUpdate, since both are running) is a
+    // group root with no deps, so nothing orders them against each other; and
+    // each declares only its OWN agent's lease, so nothing makes them contend.
+    // Those two declared facts are what "they run concurrently" *means* here —
+    // that a scheduler then does run independent, resource-disjoint roots at
+    // once is hive_jobq's property, tested there.
+    let heads: Vec<_> = declared_shape(&q, id)
+        .into_iter()
+        .filter(|d| d.kind == "stop_for_update")
         .collect();
-    heads.sort_unstable();
     assert_eq!(
         heads,
         vec![
-            ("agent-a", "stop_for_update"),
-            ("agent-b", "stop_for_update"),
+            row("stop_for_update", None, &[]),
+            row("stop_for_update", None, &[]),
         ],
-        "both per-agent subgraphs start concurrently, each acquiring its own lease"
+        "both per-agent heads are independent group roots"
+    );
+    let sched = q.sched().lock().expect("job_queue mutex poisoned");
+    let graph = sched.graph();
+    let root = graph.resolve_id(id).expect("dag id");
+    let mut leases: Vec<String> = graph
+        .nodes()
+        .filter(|n| graph.root_of(n.id) == Some(root) && n.payload.as_str() == "stop_for_update")
+        .flat_map(|n| {
+            n.deps.iter().filter_map(|dep| match dep {
+                hive_jobq::Dep::Resource { name, .. } => Some(format!("{name:?}")),
+                hive_jobq::Dep::Node { .. } => None,
+            })
+        })
+        .collect();
+    leases.sort();
+    assert_eq!(
+        leases,
+        vec![
+            format!("{:?}", Resource::Agent("agent-a".to_owned())),
+            format!("{:?}", Resource::Agent("agent-b".to_owned())),
+        ],
+        "each head declares only its own agent's lease — disjoint, so no contention"
     );
 }
 
@@ -1930,17 +1970,18 @@ fn reparent_shape_is_a_lone_agentless_meta_window_node() {
             "set-parent".to_owned(),
         ),
     );
-    let c = claim_one(&q);
-    assert_eq!(c.kind.as_str(), "reparent");
-    assert_eq!(c.agent, "", "Reparent is agentless — no per-agent lease");
     assert_eq!(
-        declared_resources(&q, c.node_id),
+        declared_shape(&q, id),
+        vec![row("reparent", None, &[])],
+        "one node, no rebuild subgraph"
+    );
+    let node = node_of(&q, id, "reparent");
+    assert_eq!(
+        declared_resources(&q, node),
         vec![Resource::MetaWindow],
         "a topology commit must declare the same MetaWindow as WritePermFile, \
          and nothing else — no lease (agentless), no build slot (no nix work)"
     );
-    q.complete_node(c.node_id, Ok(()));
-    assert_eq!(state_of(&q, id), State::Done);
 }
 
 #[test]