refactor(#2591): fix wire-shape consumers (server await_dags, tests)

- server::await_dags: a DAG is settled when gone from the snapshot (fully
  Done) or present with all nodes terminal; pending only with a non-terminal
  node (DagView no longer carries a rolled-up state).
- DagView::rollup_state() added to hive-sh4re — the shared node-set roll-up
  derivation every Rust consumer uses.
- JobQueue::build_log_id_of(node_id) — the node_id -> build_logs lookup the
  query endpoint will use; tests assert log-id via it now.
- tests: derive roll-up state; drop the off-wire step/build_log_id wire asserts.

hive-c0re/src/job_queue/mod.rs

@@ -361,7 +361,6 @@ impl JobQueue {
         let mut inner = self.lock();
         let inner = &mut *inner;
         let started = inner.sched.settle();
-        let now = now_unix();
         let mut claims = Vec::with_capacity(started.len());
         for id in started {
             let Some(node) = inner.sched.graph().node(id) else {
@@ -511,6 +510,17 @@ impl JobQueue {
         true
     }
 
+    /// The `build_logs` row id linked to `node_id`, if any — the lookup behind
+    /// the `GET /api/build-log/<node_id>` query endpoint (the client fetches a
+    /// node's captured build output on demand rather than receiving it inline).
+    #[must_use]
+    pub fn build_log_id_of(&self, node_id: NodeId) -> Option<i64> {
+        self.lock()
+            .node_rt
+            .get(&node_id)
+            .and_then(|r| r.build_log_id)
+    }
+
     /// A DAG's terminal roll-up summary, computed on demand from its container.
     /// `None` if the DAG id is unknown. Test-only — production reads the summary
     /// `complete_node` returns when the container rolls up terminal.

hive-c0re/src/job_queue/tests.rs

@@ -45,11 +45,14 @@ fn claim_one(q: &JobQueue) -> Claim {
 }
 
 fn state_of(q: &JobQueue, dag_id: u64) -> State {
+    // A fully-`Done` DAG drops out of the snapshot (its nodes are all
+    // excluded) — absence is the completion signal, so map it to `Done`.
+    // Otherwise derive the roll-up from the node set, exactly as every wire
+    // consumer does.
     q.snapshot()
         .iter()
         .find(|d| d.id == dag_id)
-        .expect("dag present")
+        .map_or(State::Done, DagView::rollup_state)
-        .state
 }
 
 // ---- submit (dedup removed — every submit is a fresh DAG) ----
@@ -664,7 +667,7 @@ fn failed_node_cancels_downstream_but_afterany_reconcile_runs() {
     q.complete_node(id, reconcile.node_id, Ok(()));
     let snap = q.snapshot();
     let dag = snap.iter().find(|d| d.id == id).expect("dag");
-    assert_eq!(dag.state, State::Failed, "roll-up failed");
+    assert_eq!(dag.rollup_state(), State::Failed, "roll-up failed");
     let by_kind = |k: &str| {
         dag.nodes
             .iter()
@@ -872,10 +875,10 @@ fn set_step_only_on_running_and_signals_change() {
     );
     assert!(q.set_step(id, c.node_id, "next phase"));
     assert!(q.set_step_running(id, "via running lookup"));
+    // `step` is host-side only now (off the wire); completion clears it
+    // internally, but there's no wire field to observe — the return-value
+    // contract above (running-gating + change signalling) is the behaviour.
     q.complete_node(id, c.node_id, Ok(()));
-    let snap = q.snapshot();
-    let node = &snap.iter().find(|d| d.id == id).expect("dag").nodes[0];
-    assert_eq!(node.step, None, "step cleared on completion");
 }
 
 #[test]
@@ -890,9 +893,13 @@ fn set_build_log_id_links_running_node() {
     assert!(q.set_build_log_id(id, c.node_id, 42));
     assert!(q.set_build_log_id_running(id, 43));
     q.complete_node(id, c.node_id, Ok(()));
-    let snap = q.snapshot();
+    // The log id is fetched by node id (the `GET /api/build-log/<id>` lookup),
-    let node = &snap.iter().find(|d| d.id == id).expect("dag").nodes[0];
+    // not carried on the wire — it survives completion in the node runtime.
-    assert_eq!(node.build_log_id, Some(43), "log id survives completion");
+    assert_eq!(
+        q.build_log_id_of(c.node_id),
+        Some(43),
+        "log id survives completion"
+    );
 }
 
 #[test]

hive-c0re/src/server.rs

@@ -795,9 +795,14 @@ async fn await_dags(coord: &Arc<Coordinator>, ids: &[u64], timeout: std::time::D
     let deadline = std::time::Instant::now() + timeout;
     loop {
         let snap = coord.job_queue.snapshot();
-        let pending = ids
+        // A DAG has settled when it's either gone from the snapshot (fully
-            .iter()
+        // `Done` DAGs drop out) or still present but with every node terminal
-            .any(|id| snap.iter().any(|d| d.id == *id && !d.state.is_terminal()));
+        // (a `Failed`/`Cancelled` DAG lingers). It's pending only while it has
+        // a non-terminal node.
+        let pending = ids.iter().any(|id| {
+            snap.iter()
+                .any(|d| d.id == *id && d.nodes.iter().any(|n| !n.state.is_terminal()))
+        });
         if !pending {
             return;
         }

hive-sh4re/src/jobs.rs

@@ -142,3 +142,37 @@ pub struct DagView {
     /// lingers until aged out by the history cap.
     pub nodes: Vec<NodeView>,
 }
+
+impl DagView {
+    /// Roll-up state derived from the node set — the shared derivation every
+    /// Rust consumer (hivectl, the wait loops, tests) uses so the dashboard's
+    /// JS render and the host agree: `Failed` if any node failed, else
+    /// `Running` if any running, else `Queued` if any queued, else
+    /// `Cancelled` if any cancelled, else `Done`. `Done` nodes are excluded
+    /// from the wire, so a DAG that is *entirely* done isn't sent at all —
+    /// its absence from the snapshot is what signals completion.
+    #[must_use]
+    pub fn rollup_state(&self) -> State {
+        let mut any_running = false;
+        let mut any_queued = false;
+        let mut any_cancelled = false;
+        for n in &self.nodes {
+            match n.state {
+                State::Failed => return State::Failed,
+                State::Running => any_running = true,
+                State::Queued => any_queued = true,
+                State::Cancelled => any_cancelled = true,
+                State::Done => {}
+            }
+        }
+        if any_running {
+            State::Running
+        } else if any_queued {
+            State::Queued
+        } else if any_cancelled {
+            State::Cancelled
+        } else {
+            State::Done
+        }
+    }
+}