job_queue: four more template tests read the graph

perm_change and the graceful rebuild chain walked their whole DAG to
collect node kinds in order; both now assert declared_shape. The
graceful one gets a sharper claim out of it -- signal and drain go
between the build and the stop, and nothing else changes -- which is
what distinguishes it from the non-graceful chain.

reparent_bulk needed the node's payload rather than its wiring, so
payload_of() reads it off the graph. The assertion is unchanged: one
node carries every move, because bulk atomicity is why a single node was
chosen.

resubmit_while_running_is_new_dag no longer claims a node to stage the
"while running" part. submit appends a container and inserts the
declared group; it never consults the state of any existing node, so a
running earlier DAG cannot change the outcome. The property is no dedup,
covered by identical_resubmit_is_a_distinct_dag -- this one keeps the
named scenario because a config bump mid-build is what people actually
worry about.

hive-c0re/src/job_queue/tests.rs

@@ -239,6 +239,14 @@ fn node_of(q: &JobQueue, dag: u64, kind: &str) -> hive_jobq::NodeId {
     found.pop().expect("checked above")
 }
 
+/// The payload of the one node of `kind` under `dag`, for assertions about what
+/// a node *carries* rather than how it is wired.
+fn payload_of(q: &JobQueue, dag: u64, kind: &str) -> NodeKind {
+    let id = node_of(q, dag, kind);
+    let sched = q.sched().lock().expect("job_queue mutex poisoned");
+    sched.graph().node(id).expect("node exists").payload.clone()
+}
+
 /// 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.
@@ -379,11 +387,17 @@ fn distinct_submits_never_collapse() {
 
 #[test]
 fn resubmit_while_running_is_new_dag() {
+    // The "while running" is not load-bearing and used to be staged by claiming
+    // a node first. `submit` appends a container and inserts the declared
+    // group; it never consults the state of any existing node, so whether an
+    // earlier DAG is running cannot change the outcome. What is actually being
+    // asserted — no dedup, ever — is `identical_resubmit_is_a_distinct_dag`.
+    //
+    // Kept as the *named* case because "a config bump mid-build must not be
+    // swallowed" is the scenario people worry about, and a reader looking for
+    // it should find it.
     let q = JobQueue::new(1);
     let a = submit(&q, rebuild("agent-a", "first"));
-    let claim = claim_one(&q); // Prebuild running
-    assert_eq!(claim.dag_id, a);
-    // While the original runs, re-submit is legitimate new work.
     let again = submit(&q, rebuild("agent-a", "config bumped during build"));
     assert_ne!(a, again);
     assert_eq!(q.snapshot().len(), 2);
@@ -484,26 +498,25 @@ fn graceful_rebuild_chain_drains_before_stopping() {
             }),
         },
     );
-    for expected in [
+    assert_eq!(
-        "meta_sync",
+        declared_shape(&q, id)
-        "prebuild",
+            .iter()
-        "signal",
+            .map(|d| d.kind)
-        "drain",
+            .collect::<Vec<_>>(),
-        "stop_for_update",
+        vec![
-        "swap",
+            "meta_sync",
-        "post_swap",
+            "prebuild",
-        "reconcile",
+            // The graceful window goes between the build and the stop: the
-    ] {
+            // agent gets its turn to finish before the container goes down.
-        let c = claim_one(&q);
+            "signal",
-        assert_eq!(c.dag_id, id);
+            "drain",
-        assert_eq!(c.kind.as_str(), expected);
+            "stop_for_update",
-        assert!(
+            "swap",
-            q.claim_ready().is_empty(),
+            "post_swap",
-            "chain must serialize: nothing ready while {expected} runs"
+            "reconcile",
-        );
+        ],
-        q.complete_node(c.node_id, Ok(()));
+        "graceful inserts signal + drain ahead of the stop, and nothing else"
-    }
+    );
-    assert_eq!(state_of(&q, id), State::Done);
 }
 
 /// The non-graceful shape is the default everywhere except the boot sweep:
@@ -1648,21 +1661,25 @@ fn perm_change_shape_prefixes_rebuild_chain() {
             },
         ),
     );
-    for expected in [
+    assert_eq!(
-        "write_perm_file",
+        declared_shape(&q, id)
-        "meta_sync",
+            .iter()
-        "prebuild",
+            .map(|d| d.kind)
-        "stop_for_update",
+            .collect::<Vec<_>>(),
-        "swap",
+        vec![
-        "post_swap",
+            "write_perm_file",
-        "reconcile",
+            "meta_sync",
-    ] {
+            "prebuild",
-        let c = claim_one(&q);
+            "stop_for_update",
-        assert_eq!(c.kind.as_str(), expected);
+            "swap",
-        q.complete_node(c.node_id, Ok(()));
+            "post_swap",
-    }
+            "reconcile",
-    settle_rebuild_tail(&q, "agent-a", true);
+            // the ok / !ok tail pair
-    assert_eq!(state_of(&q, id), State::Done);
+            "emit_rebuilt",
+            "emit_rebuilt",
+        ],
+        "the perm write prefixes an otherwise ordinary rebuild chain"
+    );
 }
 
 #[test]
@@ -1705,12 +1722,13 @@ fn reparent_bulk_shape_carries_every_move_on_one_node() {
         &q,
         templates::reparent(moves.clone(), Source::Manual, "set-parent-bulk".to_owned()),
     );
-    let c = claim_one(&q);
+    assert_eq!(
-    assert_eq!(c.kind.as_str(), "reparent");
+        declared_shape(&q, id),
-    let NodeKind::Reparent { moves: got } = &c.kind else {
+        vec![row("reparent", None, &[])],
-        panic!("expected a Reparent node, got {:?}", c.kind);
+        "one node for the whole request, not one per move"
+    );
+    let NodeKind::Reparent { moves: got } = payload_of(&q, id, "reparent") else {
+        panic!("expected a Reparent node");
     };
-    assert_eq!(got, &moves);
+    assert_eq!(got, moves, "every move rides the single node");
-    q.complete_node(c.node_id, Ok(()));
-    assert_eq!(state_of(&q, id), State::Done);
 }